Airflow Monitoring and Logging

Monitoring tells you what is happening in your pipelines right now. Logging records what happened so you can investigate problems after the fact. Both are essential for keeping production Airflow systems healthy.

Airflow's Built-In Monitoring Tools

The DAGs Dashboard

The main DAGs page shows a quick health summary for every workflow. Green circles in the "Recent Tasks" column confirm pipelines are running as expected. Red circles signal failures that need attention.

Grid View as a Health Check

Grid View is the most powerful monitoring view. Each column is one DAG run. Each row is one task. Color tells you the state at a glance.

Healthy Pipeline:
Tasks          │ Mon  │ Tue  │ Wed  │ Thu  │ Fri  │
───────────────┼──────┼──────┼──────┼──────┼──────┤
extract        │  ✅  │  ✅  │  ✅  │  ✅  │  ✅  │
transform      │  ✅  │  ✅  │  ✅  │  ✅  │  ✅  │
load           │  ✅  │  ✅  │  ✅  │  ✅  │  ✅  │

Problem Spotted:
Tasks          │ Mon  │ Tue  │ Wed  │ Thu  │ Fri  │
───────────────┼──────┼──────┼──────┼──────┼──────┤
extract        │  ✅  │  ✅  │  ❌  │  ✅  │  ✅  │
transform      │  ✅  │  ✅  │  ⬜  │  ✅  │  ✅  │
load           │  ✅  │  ✅  │  ⬜  │  ✅  │  ✅  │

Wednesday's extract failed. Transform and load never ran.
Fix the extract task and clear it to re-run.

Task Logs: Your Primary Debugging Tool

Every task writes its output and errors to a log file. Access logs in three ways:

From the UI

Click any task square in Grid View → click Log. You see the full console output — every print statement, warning, error, and stack trace.

From the CLI

airflow tasks logs my_dag_id my_task_id 2024-01-15

From the Filesystem

~/airflow/logs/
└── dag_id=my_dag/
    └── run_id=scheduled__2024-01-15T00:00:00+00:00/
        └── task_id=my_task/
            └── attempt=1.log

Log files persist on disk (or in your configured remote storage) even after a DAG run finishes.

Writing Useful Log Messages in Task Code

Use Python's logging module instead of print(). Airflow captures both, but proper log levels help you filter messages by severity:

import logging

logger = logging.getLogger(__name__)

def process_orders():
    logger.info("Starting order processing")

    orders = fetch_orders()
    logger.info(f"Fetched {len(orders)} orders from API")

    if len(orders) == 0:
        logger.warning("No orders found — check if the API returned empty data")
        return

    for order in orders:
        try:
            process_single_order(order)
        except Exception as e:
            logger.error(f"Failed to process order {order['id']}: {e}")
            raise

    logger.info("Order processing complete")

In the UI logs, you can filter by level (INFO, WARNING, ERROR) to focus on what matters.

Remote Log Storage

Local disk logs disappear when a server is replaced or a Kubernetes pod is deleted. Remote log storage persists logs permanently in cloud storage.

Log Flow With Remote Storage:
Task Runs → Writes log → Uploads to S3 / GCS / Azure Blob
                              │
                              ▼
              Airflow UI fetches log from remote storage
              (even after the server is gone)

Configure S3 remote logging in airflow.cfg:

[logging]
remote_logging = True
remote_log_conn_id = aws_default
remote_base_log_folder = s3://my-airflow-logs/logs

Airflow Metrics With StatsD

Airflow emits operational metrics (task durations, success rates, queue lengths) in StatsD format. Connect these to Prometheus + Grafana to build dashboards.

[metrics]
statsd_on = True
statsd_host = localhost
statsd_port = 8125
statsd_prefix = airflow

Key metrics Airflow emits:

The Airflow REST API for Monitoring

Query Airflow's status programmatically using its built-in REST API. Useful for building custom dashboards or triggering actions from external systems:

# List all DAG runs for a specific DAG
curl -X GET "http://localhost:8080/api/v1/dags/my_dag/dagRuns" \
  -u "admin:admin"

# Get the status of a specific task instance
curl -X GET \
  "http://localhost:8080/api/v1/dags/my_dag/dagRuns/run_001/taskInstances/my_task" \
  -u "admin:admin"

Alerting: Getting Notified When Things Break

Email Alerts

Configure SMTP in airflow.cfg and set email_on_failure=True in default_args:

[smtp]
smtp_host = smtp.gmail.com
smtp_starttls = True
smtp_ssl = False
smtp_user = airflow@company.com
smtp_password = your_app_password
smtp_port = 587
smtp_mail_from = airflow@company.com

default_args = {
    "email": ["data-team@company.com"],
    "email_on_failure": True,
    "email_on_retry": False,
}

Slack Alerts With a Callback

from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator

def slack_failure_alert(context):
    task_id = context["task_instance"].task_id
    dag_id  = context["dag"].dag_id
    SlackWebhookOperator(
        task_id="slack_alert",
        slack_webhook_conn_id="slack_webhook",
        message=f":red_circle: FAILED: {dag_id} / {task_id}",
        dag=context["dag"],
    ).execute(context)

Health Check Endpoint

Airflow exposes a /health endpoint you can ping from any uptime monitoring tool:

curl http://localhost:8080/health

Response:
{
  "metadatabase": {"status": "healthy"},
  "scheduler":    {"status": "healthy", "latest_scheduler_heartbeat": "2024-01-15T10:05:00"}
}

If scheduler.status is not "healthy", the Scheduler stopped running and no tasks will execute until it restarts.

Monitoring Checklist for Production