managed-airflow-dag-troubleshooting

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Managed Service for Apache Airflow (formerly Cloud Composer) DAG troubleshooting guide

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

This skill provides instructions for troubleshooting Managed Airflow DAGs (DAG runs and task instances), utilizing

gcloud composer

gcloud logging

and

gcloud storage

commands to fetch remote logs and code.

本技能提供托管式Airflow DAG（DAG运行和任务实例）的故障排查说明，可借助

gcloud composer

、

gcloud logging

和

gcloud storage

命令获取远程日志与代码。

General rules

通用规则

Provide suggestions on how to troubleshoot the failed jobs. Provide only the steps that the user can actually take. Ground all troubleshooting advice in direct findings.
When troubleshooting a failure, follow the following practices to always provide a deterministic diagnosis:
- Fetch relevant logs: Always fetch the logs for a task under investigation using
```
gcloud logging read
```
  ; check the logs for specific error patterns: Python tracebacks, API error codes (e.g., 400, 403, 404, 500), or Airflow signals (e.g.,
```
AirflowTaskTimeout
```
  ).
- Fetch task metadata: When troubleshooting a task, fetch the task state and metadata (execution state, try number, timestamps, and execution details) using:
  bash
```
gcloud composer environments run {env_name} \
    --location {location} \
    tasks states-for-dag-run -- -d {dag_id} -r {run_id}
```
  or for an individual task instance:
  bash
```
gcloud composer environments run {env_name} \
    --location {location} \
    tasks state -- {dag_id} {task_id} {execution_date}
```
- Retrieve and compare DAG source code: Download the remote DAG source code using
```
gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .
```
  (find the environment bucket via
```
gcloud composer environments describe {env_name} --location {location} --format="value(config.dagGcsPrefix)"
```
  ). Compare the parameters in the code (e.g., table IDs, disk sizes, URI paths) against the error messages found in the task logs.
- Explain code mistakes and potential fixes: Explain mistakes in the code (if any are actually visible); suggest potential fixes (if they are very likely to be meaningful); discuss source code availability if needed - if some source code is unavailable (e.g. imported from a file other than the main source code file), mention this (you can mention the package name) - in such a case take into account most likely trigger rules if they are unknown.
- Check for environment-level errors: Query Cloud Logging with
```
gcloud logging read
```
  to see if there are high-level environment issues or known platform errors correlating with the failure (see Known issues below). You MUST return ALL found issues.
- Identify failing tasks in a DAG run: When troubleshooting a failed DAG run, mention the task that caused a failure (use
```
tasks states-for-dag-run
```
  or Cloud Logging to identify failed tasks). Provide a task instance name. If many tasks failed, mention which task was critical (mandatory for successful DAG run execution - look into task dependencies and trigger rules) and focus on this one.
- Verify service configurations in code: If logs suggest an issue with a specific service (e.g., BigQuery, Dataform, Compute Engine), use the log details to verify the configuration in the DAG source code.
- Correlate logs with code: E.g., if BigQuery returns a 404, verify the dataset ID or table ID in the DAG source code matches reality.
- Prioritize known platform issues: Check against Known issues below. If Cloud Logging queries return matching platform error signals, prioritize that diagnosis.
Summarize with Evidence (Deterministic Response): Your response must be specific. Avoid general advice like 'check your permissions.' or 'check the logs.' Instead, say 'The service account is missing X permission.'
- Problem: State the specific root cause and the exact task instance ID. Identify if it is a code logic error, a configuration mismatch, or an environment timeout.
- Evidence: Mandatory. Provide the verbatim text from the log (
```
textPayload
```
  ) or the specific line of code from the DAG that caused the failure. Do not summarize the evidence; show the data.
- Recommendation: Provide an actionable fix. If it is a code error, provide the corrected Python snippet. If it is a resource issue, specify the exact configuration change needed.
DAGs Generated by Orchestration Pipelines: Some DAGs may be generated by Orchestration Pipelines. A special requirement related to those DAGs is the need to explain the failure in terms of the logical actions defined in the pipeline YAML.
- Determine if a DAG is generated by Orchestration Pipelines: Orchestration Pipeline DAGs deployed by dedicated tools have
```
bundle_name
```
  ,
```
version_id
```
  , and
```
pipeline_name
```
  set in their DAG Run metadata (
```
DagRun.note
```
  that contains JSON metadata). All of them (i.e. Orchestration Pipeline DAGs deployed by dedicated tools and created manually) have an
```
op:orchestration_pipeline
```
  tag set (DAG properties, including tags, can be verified in the DAG source code or via
```
gcloud composer environments run {env_name} --location {location} dags list
```
  ).
- Orchestration Pipeline DAGs deployed by dedicated tools have additionally the following tags (information in those tags should be consistent with data in DAG Run attributes mentioned above):
  - pipeline name - tag
```
op:pipeline
```
    , e.g.
```
op:pipeline:xyz
```
    indicates a name
```
xyz
```
  - bundle name - tag
```
op:bundle
```
  - version id - tag
```
op:version
```
- Retrieve the resolved pipeline YAML definition from the environment bucket:
  - Determine the YAML file location:
    1. Retrieve the DAG source code from the environment bucket using
      gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .
      (or
      gcloud storage cat gs://{bucket_name}/dags/{dag_file}.py
      ).
    2. Inspect the source code for
      generate
      or
      generate_dags
      function calls:
      - Scenario 1:
        generate
        call found. The first argument is the path to the YAML file - relative to the
        dags
        folder in environment's bucket.
      - Scenario 2:
        generate_dags
        call found.
        
        Extract the first argument - this is the data folder. If it starts with
        /home/airflow/gcs/
        , remove this prefix to get a path relative to the root of environment's bucket.
        
        Extract
        bundle_name
        ,
        version_id
        , and
        pipeline_name
        (as explained above).
        
        Construct the path:
        {data_directory}/{bundle_name}/versions/{version_id}/{pipeline_name}.yml
        (or
        .yaml
        ).
      - Scenario 3: If neither call is found, default to the path:
        data/{bundle_name}/versions/{version_id}/{pipeline_name}.yml
        (or
        .yaml
        ) in an environment's bucket.
    3. Download the YAML file using
      gcloud storage cp gs://{bucket_name}/{yaml_path} .
      (or
      gcloud storage cat gs://{bucket_name}/{yaml_path}
      ).
- Map the failed Airflow task back to the logical action name using task instance metadata/notes (e.g.
```
op_action_name
```
  in task
```
note
```
  ).
- If the failure involves user assets (like Python scripts), check their path in the action definition. If they are in the environment bucket, download and read them to debug (
```
gcloud storage cp gs://{bucket_name}/{asset_path} .
```
  ). If they are in a custom artifact bucket (see GCS URIs in logs/config), note the limitation that they cannot be read directly but analyze based on available logs.
You can assume that environment variables set by default (they can be used in DAG code, but are not visible in custom environment configuration), e.g.
```
GCS_BUCKET
```
, are correct - users cannot change them.
"Not found" (404) errors from GCP APIs can be misleading. A "not found" error might be returned when a resource actually exists, but the caller does not have permissions to access or view it. If a resource is expected to exist, suggest verifying proper permissions.

提供失败任务的排查建议。仅提供用户实际可执行的步骤。所有排查建议都需基于直接发现的信息。
排查故障时，遵循以下操作规范以确保诊断结果明确：
- 获取相关日志：始终使用
```
gcloud logging read
```
  获取待排查任务的日志；检查日志中是否存在特定错误模式：Python回溯信息、API错误代码（如400、403、404、500）或Airflow信号（如
```
AirflowTaskTimeout
```
  ）。
- 获取任务元数据：排查任务时，使用以下命令获取任务状态和元数据（执行状态、重试次数、时间戳及执行详情）：
  bash
```
gcloud composer environments run {env_name} \
    --location {location} \
    tasks states-for-dag-run -- -d {dag_id} -r {run_id}
```
  或者针对单个任务实例：
  bash
```
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查询返回匹配的平台错误信号，优先进行该诊断。
基于证据总结（确定性结论）：回复必须具体。避免给出一般性建议，如“检查权限”或“查看日志”。应改为“服务账号缺少X权限”这类表述。
- 问题：明确说明具体的根本原因和准确的任务实例ID。判断是代码逻辑错误、配置不匹配还是环境超时问题。
- 证据：必填项。提供日志中的原文文本（
```
textPayload
```
  ）或导致失败的DAG代码具体行。不得总结证据，需展示原始数据。
- 建议：提供可执行的修复方案。若为代码错误，提供修正后的Python代码片段。若为资源问题，明确说明所需的具体配置变更。
编排管道生成的DAG：部分DAG可能由编排管道生成。针对这类DAG的特殊要求是，需根据管道YAML中定义的逻辑操作解释故障原因。
- 判断DAG是否由编排管道生成：由专用工具部署的编排管道DAG在其DAG运行元数据（包含JSON元数据的
```
DagRun.note
```
  ）中设置了
```
bundle_name
```
  、
```
version_id
```
  和
```
pipeline_name
```
  。所有这类DAG（包括由专用工具部署和手动创建的编排管道DAG）都设置了
```
op:orchestration_pipeline
```
  标签（可在DAG源代码中或通过
```
gcloud composer environments run {env_name} --location {location} dags list
```
  验证DAG属性，包括标签）。
- 由专用工具部署的编排管道DAG还具有以下标签（这些标签中的信息应与上述DAG运行属性中的数据一致）：
  - 管道名称 - 标签
```
op:pipeline
```
    ，例如
```
op:pipeline:xyz
```
    表示名称为
```
xyz
```
  - 包名称 - 标签
```
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），需说明无法直接读取的限制，但可基于可用日志进行分析。
可默认认为默认设置的环境变量（可在DAG代码中使用，但在自定义环境配置中不可见）是正确的，例如
```
GCS_BUCKET
```
——用户无法修改这些变量。
GCP API返回的“未找到”（404）错误可能具有误导性。当资源实际存在但调用者没有访问或查看权限时，也可能返回“未找到”错误。若资源应存在，建议验证是否具备正确权限。

Important constraints & instructions

重要约束与说明

Read-Only First: Do NOT attempt to fix the code immediately. You must first prove the root cause using logs and remote code.
No Speculation: If logs are empty or code cannot be found, state this clearly. Always reference error messages as the are.
Safety: Be careful with secrets. If logs contain sensitive information (e.g. passwords), redact it in your analysis.

先只读操作：不要立即尝试修复代码。必须首先使用日志和远程代码证明根本原因。
禁止推测：若日志为空或无法找到代码，需明确说明。始终如实引用错误消息。
安全性：注意保密信息。若日志包含敏感信息（如密码），在分析时需进行脱敏处理。

Applying Fixes - only if explicitly requested

应用修复方案——仅在明确请求时执行

When the RCA is complete and a fix is ready:

Repository Check: If the current workspace does not seem to be the source of truth for the Managed Airflow environment:
- Ask the user to open the correct repository.
- OR ask if they want to download the remote DAG to the current workspace to apply the fix (warning them about potential overwrites).

当根本原因分析完成且修复方案就绪时：

仓库检查：若当前工作区似乎不是托管式Airflow环境的数据源：
- 请求用户打开正确的仓库。
- 或询问用户是否希望将远程DAG下载到当前工作区以应用修复方案（提醒用户可能存在覆盖风险）。

Relevant gcloud commands

Known issues related to DAG runs and task instances

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

Use

gcloud logging read

with the queries below to identify specific known platform failure modes:

使用

gcloud logging read

结合以下查询语句识别特定的已知平台故障模式：

1. DAG_RUN_TIMEOUT

Issue summary: The task instance execution was interrupted because a timeout for a DAG was exceeded. Unfinished tasks were marked as 'SKIPPED' or failed.

Cloud Logging Query:

bash

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

问题概述：由于DAG超时，任务实例执行被中断。未完成的任务被标记为“SKIPPED”或失败。

Cloud Logging查询语句：

bash

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

Issue summary: Task failed because it remained queued longer than the maximum allowed queue time.

Cloud Logging Query:

bash

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

Remediation: Consider increasing worker resources (CPU, memory, worker count) or adjusting
```
[celery]worker_concurrency
```
.

问题概述：任务失败原因是排队时间超过了最大允许的排队时长。

Cloud Logging查询语句：

bash

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

修复建议：考虑增加Worker资源（CPU、内存、Worker数量）或调整
```
[celery]worker_concurrency
```
配置。

3. TASK_STUCK_IN_QUEUE

Issue summary: Task reached DAG run timeout because task was stuck in queue for too long.

Cloud Logging Query:

bash

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

Remediation: Consider increasing the timeout or reducing the load on the environment.

问题概述：任务因在队列中停留时间过长而触发DAG运行超时。

Cloud Logging查询语句：

bash

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

Issue summary: Task failed because of a BigQuery job failure inside a BigQuery operator.

Cloud Logging Query:

bash

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

Remediation: Inspect the worker logs for the BigQuery Job ID (
```
Job ID: ...
```
) to diagnose the underlying query error or permissions issue.

问题概述：任务失败原因是BigQuery算子内部的BigQuery作业失败。

Cloud Logging查询语句：

bash

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

修复建议：查看Worker日志中的BigQuery作业ID（
```
Job ID: ...
```
）以诊断底层查询错误或权限问题。

5. DETECTED_ZOMBIE

Issue summary: The task instance was revoked by the executor due to missing heartbeats. Task instances send heartbeats periodically (every
```
job_heartbeat_sec
```
, 5 seconds by default) and if heartbeats are missing for
```
scheduler_zombie_task_threshold
```
(300 seconds by default), the task is considered a zombie and marked as failed or up for retry.

Cloud Logging Query:

bash

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

Remediation: This can happen when a worker is overloaded (CPU/memory starvation) and unable to send heartbeats on time, a worker was terminated with unfinished tasks (OOM kill/eviction), or the metadata database is overloaded. Check worker metrics and consider scaling worker CPU/memory.

问题概述：由于缺少心跳，任务实例被执行器撤销。任务实例会定期发送心跳（默认每
```
job_heartbeat_sec
```
即5秒一次），若在
```
scheduler_zombie_task_threshold
```
（默认300秒）内未收到心跳，任务会被视为僵尸任务并标记为失败或待重试。

Cloud Logging查询语句：

bash

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

修复建议：这种情况可能发生在Worker过载（CPU/内存不足）无法按时发送心跳、Worker被终止但任务未完成（OOM终止/驱逐）或元数据库过载时。检查Worker指标并考虑扩展Worker的CPU/内存。

6. WORKER_OUT_OF_POD_STORAGE

Issue summary: Task instance failed because a worker is running out of pod storage (ephemeral disk space reached or pod evicted due to storage limits).

Cloud Logging Query:

bash

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

Remediation: Update the worker storage configuration according to the amount of data being stored or clean up temporary files created during task execution.

问题概述：任务实例失败原因是Worker的Pod存储不足（临时磁盘空间耗尽或因存储限制导致Pod被驱逐）。

Cloud Logging查询语句：

bash

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

修复建议：根据存储的数据量更新Worker存储配置，或清理任务执行过程中创建的临时文件。

managed-airflow-dag-troubleshooting

Original

Translation

Managed Service for Apache Airflow (formerly Cloud Composer) DAG troubleshooting guide

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

General rules

通用规则

Important constraints & instructions

重要约束与说明

Applying Fixes - only if explicitly requested

应用修复方案——仅在明确请求时执行

Relevant gcloud commands

相关gcloud命令

Environment & DAG Discovery

环境与DAG发现

Log Retrieval

日志获取

Code & Asset Retrieval

代码与资产获取

Known issues related to DAG runs and task instances

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

1. DAG_RUN_TIMEOUT

1. DAG_RUN_TIMEOUT

2. TASK_QUEUED_TIMEOUT

2. TASK_QUEUED_TIMEOUT

3. TASK_STUCK_IN_QUEUE

3. TASK_STUCK_IN_QUEUE

4. BIGQUERY_JOB_FAILED

4. BIGQUERY_JOB_FAILED

5. DETECTED_ZOMBIE

5. DETECTED_ZOMBIE

6. WORKER_OUT_OF_POD_STORAGE

6. WORKER_OUT_OF_POD_STORAGE