managed-airflow-dag-troubleshooting

Managed Service for 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 のソースコードを取得して比較する：次のコマンドを使用して、リモートの DAG ソースコードをダウンロードします。gcloud storage cp gs://{bucket_name}/dags/{dag_file}.py .（次のコマンドで環境バケットを確認します：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 の メタデータ（JSON メタデータを含む DagRun.note）に bundle_name、 version_id、pipeline_name が設定されています。すべての 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 であることを示します
     • バンドル名 - タグ 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 Job 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 による強制終了／エビクション）、またはメタデータデータベースに過負荷がかかっている場合に発生することがあります。ワーカーのメトリクスを確認し、ワーカーの CPU／メモリを増やすことを検討してください。

6. WORKER_OUT_OF_POD_STORAGE

• 問題の概要： ワーカーのポッドストレージが不足しているため、タスクインスタンスが失敗しました（エフェメラルディスクの空き容量がなくなったか、ストレージ制限によりポッドがエビクションされました）。

• 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
  

• 対処方法： 保存するデータ量に応じてワーカーのストレージ設定を更新するか、タスク実行中に作成された一時ファイルを削除してください。