• 主页
• airflow.providers.cncf.kubernetes
• airflow.providers.cncf.kubernetes.utils
• airflow.providers.cncf.kubernetes.utils.pod_manager

airflow.providers.cncf.kubernetes.utils.pod_manager

启动 POD。

模块内容

类

PodPhase	可能的 Pod 阶段。
PodOperatorHookProtocol	协议，用于定义 KubernetesPodOperator 依赖的方法。
PodLoggingStatus	从 fetch_container_logs 退出时返回 Pod 的状态和上次日志时间。
PodManager	创建、监控以及以其他方式与 Kubernetes Pod 交互，以便与 KubernetesPodOperator 一起使用。
OnFinishAction	Pod 完成时采取的动作。

函数

should_retry_start_pod(exception)	检查异常是否指示瞬态错误并值得重试。
get_container_status(pod, container_name)	检索容器状态。
container_is_running(pod, container_name)	检查 V1Pod pod 以确定 container_name 是否正在运行。
container_is_completed(pod, container_name)	检查 V1Pod pod 以确定 container_name 是否已完成。
container_is_succeeded(pod, container_name)	检查 V1Pod pod 以确定 container_name 是否已完成并成功。
container_is_wait(pod, container_name)	检查 V1Pod pod 以确定 container_name 是否正在等待。
container_is_terminated(pod, container_name)	检查 V1Pod pod 以确定 container_name 是否已终止。
get_container_termination_message(pod, container_name)
check_exception_is_kubernetes_api_unauthorized(exc)
is_log_group_marker(line)	检查该行是否为日志组标记，例如 ::group:: 或 ::endgroup::。

exception airflow.providers.cncf.kubernetes.utils.pod_manager.PodLaunchFailedException

基类：airflow.exceptions.AirflowException

当 pod 在 KubernetesPodOperator 中启动失败时。

airflow.providers.cncf.kubernetes.utils.pod_manager.should_retry_start_pod(exception)

检查异常是否指示瞬态错误并值得重试。

class airflow.providers.cncf.kubernetes.utils.pod_manager.PodPhase

可能的 Pod 阶段。

请参阅 https://kubernetes.ac.cn/docs/concepts/workloads/pods/pod-lifecycle/#pod-phase。

PENDING = 'Pending'

RUNNING = 'Running'

FAILED = 'Failed'

SUCCEEDED = 'Succeeded'

terminal_states

class airflow.providers.cncf.kubernetes.utils.pod_manager.PodOperatorHookProtocol

基类：Protocol

协议，用于定义 KubernetesPodOperator 依赖的方法。

KubernetesPodOperator 的子类（例如 GKEStartPodOperator）可能会使用不扩展 KubernetesHook 的 hook。 我们使用此协议来记录 KPO 使用的方法，并确保这些方法存在于其他此类 hook 中。

property core_v1_client: kubernetes.client.CoreV1Api

获取经过身份验证的客户端对象。

property is_in_cluster: bool

公开 hook 是否配置了 load_incluster_config。

get_pod(name, namespace)

从 Kubernetes API 读取 pod 对象。

get_namespace()

返回连接中定义的命名空间。

get_xcom_sidecar_container_image()

返回连接中定义的 xcom sidecar 镜像。

get_xcom_sidecar_container_resources()

返回连接中定义的 xcom sidecar 资源。

airflow.providers.cncf.kubernetes.utils.pod_manager.get_container_status(pod, container_name)

检索容器状态。

airflow.providers.cncf.kubernetes.utils.pod_manager.container_is_running(pod, container_name)

检查 V1Pod pod 以确定 container_name 是否正在运行。

如果该容器存在且正在运行，则返回 True。否则返回 False。

airflow.providers.cncf.kubernetes.utils.pod_manager.container_is_completed(pod, container_name)

检查 V1Pod pod 以确定 container_name 是否已完成。

如果该容器存在且已完成，则返回 True。否则返回 False。

airflow.providers.cncf.kubernetes.utils.pod_manager.container_is_succeeded(pod, container_name)

检查 V1Pod pod 以确定 container_name 是否已完成并成功。

如果该容器存在、已完成且成功，则返回 True。否则返回 False。

airflow.providers.cncf.kubernetes.utils.pod_manager.container_is_wait(pod, container_name)

检查 V1Pod pod 以确定 container_name 是否正在等待。

如果该容器存在且正在等待，则返回 True。否则返回 False。

airflow.providers.cncf.kubernetes.utils.pod_manager.container_is_terminated(pod, container_name)

检查 V1Pod pod 以确定 container_name 是否已终止。

如果该容器存在且已终止，则返回 True。否则返回 False。

airflow.providers.cncf.kubernetes.utils.pod_manager.get_container_termination_message(pod, container_name)

airflow.providers.cncf.kubernetes.utils.pod_manager.check_exception_is_kubernetes_api_unauthorized(exc)

exception airflow.providers.cncf.kubernetes.utils.pod_manager.PodLaunchTimeoutException

基类：airflow.exceptions.AirflowException

当 Pod 在指定的超时时间内未离开 Pending 阶段时。

exception airflow.providers.cncf.kubernetes.utils.pod_manager.PodNotFoundException

基类：airflow.exceptions.AirflowException

预期的 Pod 在 kube-api 中不存在。

class airflow.providers.cncf.kubernetes.utils.pod_manager.PodLoggingStatus

从 fetch_container_logs 退出时返回 Pod 的状态和上次日志时间。

running: bool

last_log_time: pendulum.DateTime | None

class airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager(kube_client, callbacks=None)

基类： airflow.utils.log.logging_mixin.LoggingMixin

创建、监控以及以其他方式与 Kubernetes Pod 交互，以便与 KubernetesPodOperator 一起使用。

run_pod_async(pod, **kwargs)

异步运行 POD。

delete_pod(pod)

删除 POD。

create_pod(pod)

异步启动 Pod。

await_pod_start(pod, startup_timeout=120, startup_check_interval=1)

等待 Pod 达到 Pending 以外的阶段。

参数

• pod (kubernetes.client.models.v1_pod.V1Pod) –

• startup_timeout (int) – Pod 启动的超时时间（以秒为单位）（如果 Pod 处于 Pending 状态时间过长，则任务失败）

• startup_check_interval (int) – 检查之间的间隔（以秒为单位）

返回

返回类型

None

await_container_completion(pod, container_name)

等待给定 Pod 中的给定容器完成。

参数

• pod (kubernetes.client.models.v1_pod.V1Pod) – 将被监控的 Pod 规范

• container_name (str) – Pod 中要监控的容器的名称

await_pod_completion(pod, istio_enabled=False, container_name='base')

监控 Pod 并返回最终状态。

参数

• istio_enabled (bool) – 命名空间中是否启用了 istio

• pod (kubernetes.client.models.v1_pod.V1Pod) – 将被监控的 Pod 规范

• container_name (str) – pod 内容器的名称

返回

tuple[State, str | None]

返回类型

kubernetes.client.models.v1_pod.V1Pod

parse_log_line(line)

解析 K8s 日志行并返回最终状态。

参数

line (str) – k8s 日志行

返回

时间戳和日志消息

返回类型

tuple[pendulum.DateTime | None, str]

container_is_running(pod, container_name)

读取 pod 并检查容器是否正在运行。

container_is_terminated(pod, container_name)

读取 pod 并检查容器是否已终止。

read_pod_logs(pod, container_name, tail_lines=None, timestamps=False, since_seconds=None, follow=True, post_termination_timeout=120, **kwargs)

从 POD 读取日志。

read_pod_events(pod)

从 POD 读取事件。

read_pod(pod)

读取 POD 信息。

await_xcom_sidecar_container_start(pod, timeout=900, log_interval=30)

在执行 do_xcom_push 之前，检查 sidecar 容器是否已达到“正在运行”状态。

extract_xcom(pod)

检索 XCom 值并终止 xcom sidecar 容器。

extract_xcom_json(pod)

检索 XCom 值并检查 xcom json 是否有效。

extract_xcom_kill(pod)

终止 xcom sidecar 容器。

class airflow.providers.cncf.kubernetes.utils.pod_manager.OnFinishAction

基类: str, enum.Enum

Pod 完成时采取的动作。

KEEP_POD = 'keep_pod'

DELETE_POD = 'delete_pod'

DELETE_SUCCEEDED_POD = 'delete_succeeded_pod'

airflow.providers.cncf.kubernetes.utils.pod_manager.is_log_group_marker(line)

检查该行是否为日志组标记，例如 ::group:: 或 ::endgroup::。

上一个 下一个

此条目是否有帮助？