arazzo: 1.0.1 info: title: Apache Airflow Diagnose a Failed DAG Run summary: Isolate the failed task in a DAG run, pull its logs for the last attempt, and record the diagnosis as a note. description: >- When a pipeline fails, the first job is always the same: find which task failed, read its log, and write down what happened. This workflow filters the run's task instances to the failed and upstream_failed states, reads the failing task instance to learn its try_number, pulls the full log for that attempt, and then annotates both the task instance and the DAG run so the next person to look sees the diagnosis in the Airflow UI. Every step spells out its request inline so the flow can be read and executed without opening the underlying OpenAPI description. version: 1.0.0 sourceDescriptions: - name: airflowApi url: ../openapi/apache-airflow-openapi.yaml type: openapi workflows: - workflowId: diagnose-failed-dag-run summary: Find the failed task in a DAG run, retrieve its log, and annotate the finding. description: >- Confirms the run failed, lists failed task instances, reads the target task to resolve its latest try number, fetches the full log for that attempt, and writes a note onto the task instance and the DAG run. inputs: type: object required: - dagId - dagRunId - taskId properties: dagId: type: string description: The DAG identifier the failed run belongs to. dagRunId: type: string description: The run identifier to diagnose. taskId: type: string description: >- The task instance to pull logs for, normally one of the task ids returned by the listFailedTasks step. diagnosis: type: string description: >- The human written finding to record against the task instance and the DAG run once the log has been read. steps: - stepId: confirmRunFailed description: >- Read the DAG run and require the failed state so the workflow is not run against a healthy pipeline. operationId: get_dag_run parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId successCriteria: - condition: $statusCode == 200 - context: $response.body condition: $.state == "failed" type: jsonpath outputs: state: $response.body#/state startDate: $response.body#/start_date endDate: $response.body#/end_date conf: $response.body#/conf - stepId: listFailedTasks description: >- Filter the run's task instances to the failed and upstream_failed states. The state filter repeats to express an OR condition, so both the task that actually broke and the tasks blocked behind it come back in one call. operationId: get_task_instances parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId - name: state in: query value: - failed - upstream_failed - name: limit in: query value: 100 successCriteria: - condition: $statusCode == 200 - context: $response.body condition: $.total_entries > 0 type: jsonpath outputs: failedTasks: $response.body#/task_instances failedCount: $response.body#/total_entries firstFailedTaskId: $response.body#/task_instances/0/task_id - stepId: readFailedTaskInstance description: >- Read the target task instance to resolve the try_number of its most recent attempt, which the log endpoint requires as a path parameter. operationId: get_task_instance parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId - name: task_id in: path value: $inputs.taskId successCriteria: - condition: $statusCode == 200 outputs: tryNumber: $response.body#/try_number maxTries: $response.body#/max_tries state: $response.body#/state operator: $response.body#/operator hostname: $response.body#/hostname pool: $response.body#/pool - stepId: fetchTaskLog description: >- Pull the complete log for the failing attempt. full_content returns the whole log body rather than a paginated slice, which is what a diagnosis needs. operationId: get_log parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId - name: task_id in: path value: $inputs.taskId - name: task_try_number in: path value: $steps.readFailedTaskInstance.outputs.tryNumber - name: full_content in: query value: true successCriteria: - condition: $statusCode == 200 outputs: logContent: $response.body#/content continuationToken: $response.body#/continuation_token - stepId: annotateTaskInstance description: >- Record the diagnosis on the task instance so it surfaces in the Airflow UI next to the failure itself. operationId: set_task_instance_note parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId - name: task_id in: path value: $inputs.taskId requestBody: contentType: application/json payload: note: $inputs.diagnosis successCriteria: - condition: $statusCode == 200 outputs: note: $response.body#/note - stepId: annotateDagRun description: >- Record the same diagnosis on the DAG run so anyone scanning the run list sees why this run failed without drilling into a task. operationId: set_dag_run_note parameters: - name: dag_id in: path value: $inputs.dagId - name: dag_run_id in: path value: $inputs.dagRunId requestBody: contentType: application/json payload: note: $inputs.diagnosis successCriteria: - condition: $statusCode == 200 outputs: note: $response.body#/note outputs: failedTasks: $steps.listFailedTasks.outputs.failedTasks failedCount: $steps.listFailedTasks.outputs.failedCount tryNumber: $steps.readFailedTaskInstance.outputs.tryNumber logContent: $steps.fetchTaskLog.outputs.logContent