Airflow UI标记K8s Pod Operator任务失败但容器仍运行问题排查

我在DAG中通过循环创建多个KubernetesPodOperator任务并行运行，K8s节点资源足够支撑。容器共享挂载为PVC的EFS卷并执行读写操作。

现象细节 #

• 任务在Airflow UI初始显示绿色（运行中），约2小时后部分任务被标记为失败（红色），但对应的K8s容器仍处于运行状态（可通过kubectl get pods确认）。
• 这些被标记失败的任务随后会恢复为绿色（运行状态）并最终成功完成，但因曾短暂标记失败，下游设置了trigger_rule=all_success的任务被标记为skipped，即便上游最终全部正常完成。
• 代码日志无异常。

问题疑问 #

1. 该现象成因是什么？
2. 能否在DAG内部实现下游任务在上游正常完成后重新运行（无需CLI等外部命令）？
3. 是否与EFS有关？
4. 应从哪些方向排查任务短暂标记失败后又恢复的问题？

附DAG代码 #

my_task_spec = k8s.V1Pod(
    metadata="my_task",
    spec=k8s.V1PodSpec(containers=[
        k8s.V1Container(
            name="base",
            image="#{image}#",
            volume_mounts=[volume_mount],
            resources=k8s.V1ResourceRequirements(
                requests={
                    "cpu":"1000m",
                    "memory":"2Gi",
                },
                limits={
                    "cpu":"2000m",
                    "memory":"4Gi",
                }
            )
        )
    ], ))

for i in range(num_pods):
    #skipping details
    KubernetesPodOperator(
        task_id=task_id,
        full_pod_spec=my_task_spec,
        volumes=[volume],
        name=task_id,
        get_logs=True,
        is_delete_operator_pod=True,
        in_cluster=False,
        config_file=kube_config_path,
        startup_timeout_seconds=1200,
        trigger_rule='all_success',
        dag=dag)

for task in dag.tasks:
    if task.task_id.startswith('my_task'):
        task.set_downstream(some_other_task) # this is being marked as skipped once any task above fails

1. 现象成因 #

核心是Airflow调度器/worker与K8s Pod的状态同步出现临时中断或超时：

• KubernetesPodOperator会周期性拉取Pod状态，若某个周期内无法获取到Pod状态（比如K8s apiserver响应延迟、API请求超时），Airflow会误判Pod已失败，将任务标记为失败。
• 后续当Airflow重新成功获取到Pod的运行状态时，会修正任务状态为运行中，待Pod执行完成后标记为成功。
• 下游任务因all_success触发规则，只要上游任一任务曾进入失败状态，Airflow会立即触发skipped逻辑，且上游状态修正后不会回溯触发下游重新运行。

2. DAG内实现下游任务自动重跑 #

可以通过以下方式在DAG内部实现，无需外部CLI：

• 修改触发规则：将下游任务的trigger_rule改为all_done，确保所有上游任务完成（无论中间状态如何）后下游都会运行；若需严格校验上游最终成功，可搭配BranchPythonOperator检查所有上游任务的最终状态，再决定是否触发下游。
• 添加成功回调：给上游任务配置on_success_callback，当所有上游任务最终成功时，通过Airflow内部API（如TaskInstance.set_state）手动触发下游任务的重新运行。
• 配置下游重试：给下游任务设置retries参数，结合自定义的重试条件，确保仅当上游全部成功时触发重试。

3. 是否与EFS有关 #

大概率无直接关联，但存在间接影响的可能：

• 若EFS出现IO阻塞或延迟，可能导致Pod内进程响应变慢，但不会直接触发Airflow误判Pod状态。
• 除非EFS挂载问题导致Pod日志采集异常（因设置了get_logs=True），Airflow在拉取日志时超时，进而误判Pod状态。可尝试关闭get_logs=True测试是否仍出现该问题。

4. 排查方向 #

• Airflow组件日志：重点查看任务标记失败时刻的调度器/worker日志，是否存在K8s API请求超时、权限错误或连接中断的报错（如kubernetes.client.exceptions.ApiException）。
• K8s apiserver日志：检查对应时间段内是否有请求延迟、限流或异常，确认Airflow组件的请求是否被拒绝或超时。
• 网络连通性测试：验证Airflow调度器/worker到K8s apiserver的网络稳定性，排查是否存在间歇性丢包或延迟过高的情况。
• 调整Operator参数：
  • 增大get_logs_timeout（默认10秒），避免因日志采集超时误判状态；
  • 开启log_events_on_failure=True，获取更详细的Pod事件日志辅助排查；
• Pod状态监控：任务运行期间通过kubectl describe pod <pod-name>查看Pod事件记录，排查是否存在节点资源波动、网络中断等异常。

内容的提问来源于stack exchange，提问作者Naxi