Airflow

Investigate DAG failures and extract execution context from Apache Airflow.

Overview

When a DAG failure alert fires, OpenSRE queries your Airflow REST API for the failing DAG run, task instances, and logs — then folds that evidence into the RCA pipeline alongside metrics and logs from your observability integrations.

It supports:

• DAG run inspection
• Task instance retrieval
• Failure detection
• Evidence collection for RCA generation

This integration is designed for incident-driven workflows, where an alert referencing a DAG triggers an investigation.

Prerequisites

• A reachable Airflow REST API (Airflow 2.x /api/v1)
• Network access from the OpenSRE environment
• Auth: token (AIRFLOW_AUTH_TOKEN) or basic auth (AIRFLOW_USERNAME / AIRFLOW_PASSWORD)

Setup

There is no dedicated opensre integrations setup airflow wizard today. Configure via environment variables or the persistent store.

Option 1: Environment variables

AIRFLOW_BASE_URL=http://localhost:8080

# Authentication (choose one)

# Basic Auth
AIRFLOW_USERNAME=your_username
AIRFLOW_PASSWORD=your_password

# Token-based (if supported)
AIRFLOW_AUTH_TOKEN=your_token

# Optional
AIRFLOW_TIMEOUT_SECONDS=15
AIRFLOW_VERIFY_SSL=true
AIRFLOW_MAX_RESULTS=50
AIRFLOW_DAG_ID=test_fail_dag   # optional default DAG for tools

Variable	Default	Description
AIRFLOW_BASE_URL	http://localhost:8080/api/v1	Airflow API base URL
AIRFLOW_USERNAME	—	Basic auth username (with password)
AIRFLOW_PASSWORD	—	Basic auth password
AIRFLOW_AUTH_TOKEN	—	Bearer / token auth (alternative to basic)
AIRFLOW_TIMEOUT_SECONDS	15	Request timeout
AIRFLOW_VERIFY_SSL	true	Verify TLS certificates
AIRFLOW_MAX_RESULTS	50	Result cap
AIRFLOW_DAG_ID	—	Optional default DAG id for tools

Option 2: Persistent store

{
  "version": 1,
  "integrations": [
    {
      "id": "airflow-prod",
      "service": "airflow",
      "status": "active",
      "credentials": {
        "base_url": "http://localhost:8080",
        "username": "your_username",
        "password": "your_password"
      }
    }
  ]
}

Local smoke setup

Start Airflow locally:

docker run -p 8080:8080 apache/airflow:2.8.1 standalone

Create a failing DAG:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def fail_task():
    raise Exception("Intentional failure")

with DAG(
    dag_id="test_fail_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:
    PythonOperator(
        task_id="fail_task",
        python_callable=fail_task,
    )

Trigger the DAG:

airflow dags trigger test_fail_dag

Then investigate with an alert that references the DAG:

uv run opensre investigate

Example alert payload:

{
  "source": "airflow",
  "message": "Airflow DAG test_fail_dag failed",
  "metadata": {
    "dag_id": "test_fail_dag"
  }
}

Credentials

Provide either AIRFLOW_AUTH_TOKEN or AIRFLOW_USERNAME + AIRFLOW_PASSWORD. Prefer a dedicated read-only Airflow user for investigations.

Investigation tools

Tool	What it does
get_recent_airflow_failures	Recent failed / retrying task evidence for a DAG
get_airflow_dag_runs	DAG run execution history
get_airflow_task_instances	Task-level status for a run

These tools use the Airflow REST API via integrations/airflow. When source = airflow, the planner seeds Airflow tools into the action space; exact tool selection remains LLM-driven.

Related (Tracer, not Airflow API): get_airflow_metrics pulls orchestration metrics from Tracer when a trace_id is available — see Tracer.

Behavior notes

• Per-run failures are isolated — one failing request does not break the loop
• Network/API errors are handled defensively; partial evidence is preserved when possible
• Planner routing is probabilistic (LLM-based); there is no hard-coded Airflow bypass

Verify

There is no dedicated opensre integrations verify airflow target today. Confirm auth with a live investigation or by calling the Airflow API (GET /dags) from the same environment.

Tests

uv run pytest tests/e2e/airflow/test_orchestrator.py -v

Expected:

test_airflow_investigation_e2e PASSED

Routing coverage (if present in your checkout):

uv run pytest tests/nodes/plan_actions/test_airflow_routing.py -v

Troubleshooting

Symptom	Fix
Auth required / 401	Set AIRFLOW_AUTH_TOKEN or AIRFLOW_USERNAME + AIRFLOW_PASSWORD
Connection refused	Confirm Airflow is up and AIRFLOW_BASE_URL matches the REST API
Empty DAG runs	Check dag_id in the alert metadata / AIRFLOW_DAG_ID
SSL errors	Set AIRFLOW_VERIFY_SSL=false only for trusted lab endpoints

Limitations

• Requires a reachable Airflow instance
• No CI-backed Airflow instance by default (local validation required)
• No setup/verify CLI wiring yet

Security

• Prefer a dedicated read-only Airflow account over admin credentials
• Enable TLS verification in production (AIRFLOW_VERIFY_SSL=true)
• Store tokens/passwords in .env or the integration store — not in source control