排查托管 Airflow（Cloud…

当托管 Airflow 环境中的 DAG 运行或任务实例失败，需要基于日志和远程代码给出确定性诊断时使用。

适用场景

解决思路

1. 始终用 gcloud logging read 拉取被排查任务的日志，检查具体错误模式：Python traceback、API 错误码（400/403/404/500）、Airflow 信号（如 AirflowTaskTimeout）。
2. 拉取任务元数据：用 `gcloud composer environments run {env_name} --location {location} tasks states-for-dag-run -- -d {dag_id} -r {run_id}` 查看整个 DAG 运行中各任务状态；用 `gcloud composer environments run {env_name} --location {location} tasks state -- {dag_id} {task_id} {execution_date}` 查看单个任务实例状态。
3. 获取并比对 DAG 源码：先用 `gcloud composer environments describe {env_name} --location {location} --format="value(config.dagGcsPrefix)"` 找到环境 bucket，再用 `gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .` 下载远程 DAG 源码，把代码中的参数（表 ID、磁盘大小、URI 路径等）与任务日志中的错误信息逐一比对。
4. 解释代码错误并给出可能的修复：若代码中确实存在错误则说明；若修复很可能有意义则建议；若部分源码不可用（如从其他文件导入），说明这一情况（可提及包名），并在触发规则未知时按最可能的触发规则推断。
5. 检查环境级错误：用 gcloud logging read 查询是否存在与失败相关的高层环境问题或已知平台错误（见下方 Known issues），必须返回所有发现的问题。
6. 定位 DAG 运行中失败的任务：用 tasks states-for-dag-run 或 Cloud Logging 找出失败任务，给出任务实例名；若多个任务失败，指出对 DAG 成功执行关键（必选）的那个任务（结合任务依赖和触发规则），并聚焦于它。
7. 在代码中验证服务配置：若日志指向某个具体服务（如 BigQuery、Dataform、Compute Engine）的问题，用日志细节去核对 DAG 源码中的相应配置。
8. 把日志与代码关联起来：例如 BigQuery 返回 404 时，核对 DAG 源码中的 dataset ID 或 table ID 是否与实际一致。
9. 优先考虑已知平台问题：对照 Known issues，若 Cloud Logging 查询返回匹配的平台错误信号，优先按该诊断处理。
10. 输出必须具体且带证据（确定性回答）：避免“检查权限”“检查日志”这类泛泛建议，而应说“服务账号缺少 X 权限”。回答结构为：Problem（具体根因和确切的任务实例 ID，判断是代码逻辑错误、配置不匹配还是环境超时）、Evidence（必填，给出日志中的原文 textPayload 或导致失败的 DAG 具体代码行，不要转述）、Recommendation（可执行的修复，代码错误给出修正后的 Python 片段，资源问题给出确切配置变更）。
11. 对由 Orchestration Pipelines 生成的 DAG：判断方式为 DAG Run 元数据（DagRun.note 中的 JSON）含 bundle_name、version_id、pipeline_name，且 DAG 带 `op:orchestration_pipeline` 标签（可在源码或 `gcloud composer environments run {env_name} --location {location} dags list` 中查看）；专用工具部署的还带 `op:pipeline`、`op:bundle`、`op:version` 标签。
12. 对 Orchestration Pipeline DAG，从环境 bucket 取回解析后的 pipeline YAML：先下载 DAG 源码（`gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .` 或 `gcloud storage cat ...`），在源码中查找 generate 或 generate_dags 调用。场景 1：找到 generate，其第一个参数即 YAML 路径（相对环境 bucket 的 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）。最后用 `gcloud storage cp gs://{bucket_name}/{yaml_path} .` 或 `gcloud storage cat ...` 下载。
13. 用任务实例元数据/notes（如任务 note 中的 op_action_name）把失败的 Airflow 任务映射回 pipeline YAML 中的逻辑 action 名。
14. 若失败涉及用户资产（如 Python 脚本），检查其在 action 定义中的路径；若在环境 bucket 中则下载阅读调试（`gcloud storage cp gs://{bucket_name}/{asset_path} .`）；若在自定义 artifact bucket（见日志/配置中的 GCS URI），说明无法直接读取这一限制，并基于可用日志分析。
15. 可假定默认设置的环境变量（如 GCS_BUCKET，可在 DAG 代码中使用但不出现在自定义环境配置中）是正确的，用户无法修改它们。
16. GCP API 的 404 “not found” 可能具有误导性：资源实际存在但调用方无权限访问/查看时也会返回 404；若预期资源存在，建议核实权限。
17. 遵循约束：先只读排查，不要立即改代码，必须先用日志和远程代码证明根因；不要臆测，日志为空或找不到代码时明确说明，并原样引用错误信息；注意密钥安全，日志含敏感信息（如密码）时在分析中脱敏。
18. 仅在明确要求时才应用修复：若当前工作区不是该托管 Airflow 环境的真实来源，请用户打开正确的仓库，或询问是否要把远程 DAG 下载到当前工作区再修复（提醒可能覆盖文件）。
19. 常用命令——环境与 DAG 发现：`gcloud composer environments list --locations=us-central1 --format="table(name,location,state)"`；`gcloud composer environments describe {env_name} --location {region} --format="value(config.dagGcsPrefix)"`；`gcloud composer environments run {env_name} --location {region} dags list`；`gcloud composer environments run {env_name} --location {region} dags list-runs -- -d {dag_id} --no-backfill`。
20. 常用命令——日志检索：`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)"`。
21. 已知问题 1 DAG_RUN_TIMEOUT：DAG 超时导致任务实例执行被中断，未完成任务被标记为 SKIPPED 或失败。查询：`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`。
22. 已知问题 2 TASK_QUEUED_TIMEOUT：任务排队时间超过最大允许值而失败。查询：`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`。
23. 已知问题 3 TASK_STUCK_IN_QUEUE：任务在队列中卡住过久导致达到 DAG 运行超时。查询：`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`。修复：考虑增大超时或降低环境负载。
24. 已知问题 4 BIGQUERY_JOB_FAILED：BigQuery operator 内的 BigQuery 作业失败导致任务失败。查询：`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 Job ID（`Job ID: ...`）以诊断底层查询错误或权限问题。
25. 已知问题 5 DETECTED_ZOMBIE：任务实例因缺少心跳被 executor 撤销。任务实例周期性发送心跳（默认 job_heartbeat_sec 为 5 秒），若缺失超过 scheduler_zombie_task_threshold（默认 300 秒）则被视为 zombie 并标记失败或重试。查询：`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 kill/驱逐）或元数据库过载；检查 worker 指标并考虑扩容 CPU/内存。
26. 已知问题 6 WORKER_OUT_OF_POD_STORAGE：worker pod 存储耗尽（临时磁盘达到上限或 pod 因存储限制被驱逐）导致任务失败。查询：`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 存储配置，或清理任务执行期间产生的临时文件。

适用边界 · 注意事项

• 不要用于寻找托管 Airflow 环境整体性能优化建议（本技能只针对 DAG 运行/任务实例失败排查）。
• 不要在未用日志和远程代码证明根因前就直接修改代码。
• 不要在日志为空或找不到代码时臆测原因，应明确说明并原样引用错误信息。