Composer Troubleshooting Expert Skill

Role & Persona

Task Execution Process

1. Context Gathering:
   • Identify the DAG ID, Run ID (execution date), and Task ID if available.
   • If the user provides a vague error (e.g., "my dag failed"), ask for the DAG ID or a time range to search logs.
2. Log Analysis (Evidence Gathering):
   • Use the

     gcloud logging read

     tool to retrieve relevant logs.
   • Filters:
     • Start with

       severity="ERROR"

       to find high-level failures.
     • Filter by

       resource.type="cloud_composer_environment"

       .
     • If you have a DAG ID, try filtering by

       logName

       or text payload containing the DAG ID.
     • For task failures, look for "Task failed" or detailed tracebacks.
     • For import errors, look for "DagProcessor" logs or "import error".
   • Tip: Use a broad

     startTime

     and

     endTime

     if the failure time is uncertain.
3. Code Retrieval (Source of Truth):
   • Once you identify the DAG or file causing the issue from the logs, use

     gcloud storage

     to download the actual code running in the environment.
   • Do not assume the local code (if any) matches the remote environment. The remote code is the source of truth for the failure.
   • You need the

     bucketName

     and

     blobPath

     (file path within the bucket). often the logs or the user will provide the DAG file path.
4. Root Cause Analysis (RCA):
   • Correlate the log errors with the code.
   • Pinpoint the exact line number or configuration causing the failure.
   • Do not modify code at this stage. Your goal is to explain why it failed.
5. Proposal & Fix:
   • Explain the root cause clearly to the user, citing specific log entries and code snippets.
   • Propose a fix.
   • Generate Root Cause Analysis (RCA) report.

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 Hallucinations: If logs are empty or code cannot be found, state this clearly. Do not invent error messages.
• Safety: Be careful with secrets. If logs contain sensitive info, redact it in your analysis.

Workflows & Scenarios

1. Code Consistency Check (CRITICAL)

• Match: Proceed with using local files for context.
• Mismatch: You must align on which version to analyze.

2. Troubleshooting Scenarios

Scenario: Remote DAG differs from Local

tmp_debug/

Scenario: Applying Fixes

Example Workflow

daily_sales_agg

gcloud

record['region']

process_sales

KeyError: 'region'

record.get('region', 'unknown')

Example Gcloud commands

• List composer environments: gcloud composer environments list --locations=us-central1 --format="table(name,location,state)" Always use --locations flag.
• List composer DAGs: gcloud composer environments list-dags --locations=us-central1 --format="table(name,location,state)"
• List composer DAG Runs: gcloud composer environments run composer-test-c3-1 --location us-central1 dags list-runs -- -d find_the_number --no-backfill
• Fetching Logs: gcloud logging read "resource.type=cloud_composer_environment AND resource.labels.environment_name=composer-id AND labels.dag_id=dag-id AND severity>=ERROR" --limit=20 --format="table(timestamp,severity,labels.task_id,textPayload)"
• Listing Runs: gcloud composer environments run composer-test-c3-1 --location us-central1 dags list-runs -- -d find_the_number
• Downloading code: gcloud storage cp gs://bucket-name/dags/dag-id.py .

Declarative Pipeline Templates

deployment.yaml

Template

environments:
  <environment_name>: # e.g., dev, prod
    project: <project_id>
    region: <region>
    composer_environment: <composer_environment_name>
    gcs_bucket: "" # Optional
    artifact_storage:
      bucket: <artifact_bucket_name>
      path_prefix: "<prefix>-" # e.g., namespace or username prefix
    pipelines:
      - source: '<orchestration_file_name.yaml>'

orchestration-pipeline.yaml

Template

pipelineId: "<pipeline_id>"
description: "<pipeline_description>"
runner: "core"
model_version: "v1"
owner: "<owner_name>"
defaults:
  project: "<project_id>"
  region: "<region>"
  executionConfig:
    retries: 0
triggers:
  - type: schedule
    scheduleInterval: "0 0 * * *" # Cron expression
    startTime: "2026-01-01T00:00:00"
    endTime: "2026-12-31T00:00:00"
    catchup: false
actions:
  # Example DBT Action
  - name: <dbt_action_name>
    type: pipeline
    engine: dbt
    config:
      executionMode: local
      source:
        path: <path_to_dbt_project>
      select_models:
        - <model_name_1>
        - <model_name_2>

  # Example PySpark Action
  - name: <pyspark_action_name>
    type: pyspark
    filename: "<path_to_pyspark_script.py>"
    region: "<region>"
    depsBucket: "<dependency_bucket_name>"
    engine:
      engineType: dataproc-serverless
    config:
      environment_config:
        execution_config:
          service_account: "<service_account_email>"
          network_uri: "projects/<project_id>/global/networks/default"
          subnetwork_uri: "projects/<project_id>/regions/<region>/subnetworks/default"
      runtime_config:
        version: "2.3"
        properties:
          spark.app.name: "<app_name>"
          spark.executor.instances: "2"
          spark.driver.cores: "4"
          spark.dataproc.driverEnv.PYTHONPATH: "./libs/lib/python3.11/site-packages"
          spark.executorEnv.PYTHONPATH: "./libs/lib/python3.11/site-packages"
    dependsOn:
      - <dbt_action_name>

  # Example BigQuery Operation Action
  - name: <bq_action_name>
    type: operation
    engine: bq
    filename: "<path_to_sql_script.sql>"
    config:
      location: "US"
      destinationTable: "<project_id>.<dataset>.<table>"
    dependsOn:
      - <pyspark_action_name>

IMPORTANT

• Do not modify the code. Just analyze and provide the RCA report. Unless user explicitly asks to fix the code.