managed-airflow-dag-troubleshooting

Apache Airflow 托管服务（原名 Cloud Composer）DAG 故障排查指南

本技能介绍如何排查 Managed Airflow DAG（DAG 运行和任务实例），并使用 gcloud composer、gcloud logging 和 gcloud storage 命令获取远程日志和代码。

一般规则

1. 提供有关如何排查失败作业的建议。只提供用户实际可以采取的步骤。所有故障排查建议都必须以直接发现为依据。

2. 排查故障时，请遵循以下做法，以始终提供确定性诊断：

   • 获取相关日志：始终使用 gcloud logging read 获取所调查任务的日志；检查日志中是否存在特定错误模式：Python 回溯、API 错误代码（如 400、403、404、500）或 Airflow 信号（如 AirflowTaskTimeout）。

   • 获取任务元数据：排查任务时，使用以下命令获取任务状态和元数据（执行状态、尝试次数、时间戳和执行详情）：

     gcloud composer environments run {env_name} \
         --location {location} \
         tasks states-for-dag-run -- -d {dag_id} -r {run_id}
     

     或者，获取单个任务实例的信息：

     gcloud composer environments run {env_name} \
         --location {location} \
         tasks state -- {dag_id} {task_id} {execution_date}
     

   • 检索并比对 DAG 源代码：使用 gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py . 下载远程 DAG 源代码（通过 gcloud composer environments describe {env_name} --location {location} --format="value(config.dagGcsPrefix)" 查找环境存储桶）。将代码中的参数（如表 ID、磁盘大小、URI 路径）与任务日志中的错误消息进行比对。

   • 说明代码错误和可能的修复方法：说明代码中实际可见的错误（如有）；建议可能有效的修复方法（如确实很可能有帮助）；必要时讨论源代码是否可用。如果某些源代码不可用（例如，从主源代码文件以外的文件导入），请说明这一点（可以提及软件包名称）；在这种情况下，如果触发规则未知，请考虑最可能的触发规则。

   • 检查环境级错误：使用 gcloud logging read 查询 Cloud Logging，查看是否存在与故障相关的高层环境问题或已知平台错误（请参阅下方的已知问题）。必须返回所有发现的问题。

   • 识别 DAG 运行中的失败任务：排查失败的 DAG 运行时，指出导致故障的任务（使用 tasks states-for-dag-run 或 Cloud Logging 识别失败任务）。提供任务实例名称。如果有多个任务失败，请指出其中的关键任务（即 DAG 运行成功所必需的任务；请检查任务依赖关系和触发规则），并重点分析该任务。

   • 验证代码中的服务配置：如果日志表明特定服务（如 BigQuery、Dataform、Compute Engine）存在问题，请根据日志详情验证 DAG 源代码中的配置。

   • 将日志与代码关联起来：例如，如果 BigQuery 返回 404，请验证 DAG 源代码中的数据集 ID 或表 ID 是否与实际情况相符。

   • 优先排查已知平台问题：对照下方的已知问题进行检查。如果 Cloud Logging 查询返回匹配的平台错误信号，应优先考虑这一诊断结果。

3. 提供证据进行总结（确定性响应）： 你的回答必须具体。避免使用“检查权限”或“检查日志”之类的泛泛建议。应改为指出“服务账号缺少 X 权限”。

   • 问题： 指出具体的根本原因和准确的任务实例 ID。明确说明问题属于代码逻辑错误、配置不匹配还是环境超时。
   • 证据： 必填。 提供日志中的原文（textPayload）或导致失败的 DAG 中的具体代码行。不要概述证据，直接展示数据。
   • 建议： 提供可执行的修复方案。如果是代码错误，请提供修正后的 Python 代码片段。如果是资源问题，请明确指出需要进行的具体配置更改。

4. 由编排流水线生成的 DAG： 有些 DAG 可能由编排流水线生成。针对这类 DAG 的特殊要求是，需要根据流水线 YAML 中定义的逻辑操作来解释失败。

   • 确定 DAG 是否由 Orchestration Pipelines 生成： 由专用工具部署的 Orchestration Pipeline DAG 会在其 DAG Run 元数据中设置 bundle_name、version_id 和 pipeline_name （DagRun.note 中包含 JSON 元数据）。所有这些 DAG（即由专用工具部署和手动创建的 Orchestration Pipeline DAG）都设置了 op:orchestration_pipeline 标签（可在 DAG 源代码中验证 DAG 属性（包括标签），也可通过 gcloud composer environments run {env_name} --location {location} dags list 验证）。
   • 由专用工具部署的 Orchestration Pipeline DAG 还设置了以下标签（这些标签中的信息应与上述 DAG Run 属性中的数据一致）：
     • 流水线名称 - 标签 op:pipeline，例如 op:pipeline:xyz 表示名称为 xyz
     • Bundle 名称 - 标签 op:bundle
     • 版本 ID - 标签 op:version
   • 从环境存储桶中获取解析后的流水线 YAML 定义：
     • 确定 YAML 文件的位置：
       1. 使用 gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .（或 gcloud storage cat gs://{bucket_name}/dags/{dag_file}.py）从环境存储桶中获取 DAG 源代码。
       2. 检查源代码中是否调用了 generate 或 generate_dags 函数：
          • 场景 1：找到 generate 调用。第一个参数是 YAML 文件的 路径，相对于环境存储桶中的 dags 文件夹。
          • 场景 2：找到 generate_dags 调用。
            • 提取第一个参数，即数据文件夹。如果它以 /home/airflow/gcs/ 开头，则移除此此前缀，得到相对于环境存储桶根目录的路径。
            • 提取 bundle_name、version_id 和 pipeline_name （如上所述）。
            • 构造路径： {data_directory}/{bundle_name}/versions/{version_id}/{pipeline_name}.yml （或 .yaml）。
          • 场景 3：如果未找到这两种调用，则默认使用环境存储桶中的路径： data/{bundle_name}/versions/{version_id}/{pipeline_name}.yml （或 .yaml）。
       3. 使用 gcloud storage cp gs://{bucket_name}/{yaml_path} .（或 gcloud storage cat gs://{bucket_name}/{yaml_path}）下载 YAML 文件。
   • 使用任务实例元数据/备注（例如任务 note 中的 op_action_name），将失败的 Airflow 任务映射回逻辑操作名称。
   • 如果故障涉及用户资产（例如 Python 脚本），请检查操作定义中的路径。如果这些资产位于环境存储桶中，请下载并读取它们以进行调试（gcloud storage cp gs://{bucket_name}/{asset_path} .）。如果它们位于自定义制品存储桶中（请查看日志/配置中的 GCS URI），请说明无法直接读取这些资产的限制，但可根据现有日志进行分析。

5. 你可以假定默认设置的环境变量（可在 DAG 代码中使用，但在自定义环境配置中不可见）正确，例如 GCS_BUCKET；用户无法更改这些变量。

6. GCP API 返回的“未找到”（404）错误可能会造成误导。即使资源实际存在，如果调用方没有访问或查看该资源的权限，也可能返回“未找到”错误。如果预期资源应该存在，建议检查是否具备适当的权限。

重要约束和说明

• 只读优先：不要立即尝试修复代码。必须先通过日志和远程代码证明根本原因。
• 禁止猜测：如果日志为空或无法找到代码，请明确说明。始终按原样引用错误消息。
• 安全性：注意保护机密信息。如果日志中包含敏感信息（例如密码），请在分析中将其删去。

应用修复——仅在明确请求时

完成根因分析（RCA）并准备好修复方案后：

1. 检查代码库：如果当前工作区看起来并非 Managed Airflow 环境的事实来源：
   • 请用户打开正确的代码库。
   • 或询问用户是否希望将远程 DAG 下载到当前工作区以应用修复（并提醒用户可能会覆盖现有文件）。

相关 gcloud 命令

环境和 DAG 发现

• 列出 composer 环境：

  gcloud composer environments list \
      --locations=us-central1 \
      --format="table(name,location,state)"
  

• 描述环境（获取 DAG 存储桶和配置）：

  gcloud composer environments describe {env_name} \
      --location {region} \
      --format="value(config.dagGcsPrefix)"
  

• 列出 composer DAG：

  gcloud composer environments run {env_name} \
      --location {region} \
      dags list
  

• 列出 composer DAG 运行记录：

  gcloud composer environments run {env_name} \
      --location {region} \
      dags list-runs -- -d {dag_id} --no-backfill
  

• 列出 DAG 运行中各任务实例的状态：

  gcloud composer environments run {env_name} \
      --location {region} \
      tasks states-for-dag-run -- -d {dag_id} -r {run_id}
  

• 获取特定任务实例的状态：

  gcloud composer environments run {env_name} \
      --location {region} \
      tasks state -- {dag_id} {task_id} {execution_date}
  

获取日志

• 获取 DAG / 任务的错误日志：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND labels.dag_id="{dag_id}" AND severity>=ERROR' \
      --limit=25 \
      --format="table(timestamp,severity,labels.task_id,textPayload)"
  

• 获取环境故障的调度器日志：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND log_id("airflow-scheduler") AND severity>=ERROR' \
      --limit=25 \
      --format="table(timestamp,severity,textPayload)"
  

获取代码和资源

• 从 GCS 下载 DAG 代码：

  gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .
  

• 从 GCS 下载流水线 YAML 定义或脚本：

  gcloud storage cp gs://{bucket_name}/{path_to_file} .
  

与 DAG 运行和任务实例相关的已知问题

使用下方查询运行 gcloud logging read，以识别特定的已知平台故障模式：

1. DAG_RUN_TIMEOUT

• **问题摘要：**任务实例执行因超出 DAG 超时限制而中断。未完成的任务会被标记为“SKIPPED”或失败。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND log_id("airflow-scheduler") AND textPayload=~"Run .* of .* has-timed-out"' --limit=10
  

2. TASK_QUEUED_TIMEOUT

• **问题摘要：**任务失败，因为它在队列中的停留时间超过了允许的最长时间。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND log_id("airflow-scheduler") AND textPayload=~"Task requeue attempts exceeded max; marking failed"' --limit=10
  

• **修复方法：**考虑增加工作线程资源（CPU、内存、工作线程数量），或调整 [celery]worker_concurrency。

3. TASK_STUCK_IN_QUEUE

• **问题摘要：**由于任务在队列中停留过久，达到 DAG 运行超时时间。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND log_id("airflow-scheduler") AND textPayload=~"Task stuck in queued; will try to requeue"' --limit=10
  

• **修复方法：**考虑增加超时时间，或降低环境负载。

4. BIGQUERY_JOB_FAILED

• **问题摘要：**BigQuery 运算符内部的 BigQuery 作业失败，导致任务失败。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND (log_id("airflow-worker") OR log_id("airflow-k8s-worker")) AND textPayload:"airflow/providers/google/cloud/operators/bigquery.py" AND textPayload:"Task failed with exception" AND severity=ERROR' --limit=10
  

• **修复方法：**检查工作线程日志中的 BigQuery 作业 ID（Job ID: ...），以诊断底层查询错误或权限问题。

5. DETECTED_ZOMBIE

• **问题摘要：**由于缺少心跳信号，执行器撤销了任务实例。任务实例会定期发送心跳（默认每 5 秒发送一次，由 job_heartbeat_sec 控制）；如果在 scheduler_zombie_task_threshold 指定的时间内（默认 300 秒）未收到心跳，则该任务会被视为僵尸任务，并标记为失败或等待重试。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND log_id("airflow-scheduler") AND (textPayload:"Detected zombie job:" OR textPayload:"Detected a task instance without a heartbeat:")' --limit=10
  

• **修复方法：**工作线程过载（CPU 或内存不足）而无法及时发送心跳、工作线程在任务未完成时被终止（OOM kill/驱逐），或元数据数据库过载，都可能导致此问题。检查工作线程指标，并考虑增加工作线程的 CPU 或内存。

6. WORKER_OUT_OF_POD_STORAGE

• **问题摘要：**工作线程的 Pod 存储空间不足（临时磁盘空间耗尽，或 Pod 因存储限制而被驱逐），导致任务实例失败。

• Cloud Logging 查询：

  gcloud logging read 'resource.type="cloud_composer_environment" AND resource.labels.environment_name="{env_name}" AND (log_id("airflow-worker") OR log_id("airflow-k8s-worker")) AND textPayload:"Pod ephemeral local storage usage exceeds the total limit of containers"' --limit=10
  

• **修复方法：**根据存储的数据量更新工作线程存储配置，或清理任务执行期间创建的临时文件。