Lesson 2 - Reading Logs

Welcome to Reading Logs#

Module 2 Lesson 4 toured the Logs tab for a task that already succeeded — useful for orientation, but it skips the moment logs actually matter most: something broke, and you need to find out what, fast. This lesson triggers a task that genuinely fails and reads its real log, structure and all.

By the end of this lesson, you will be able to:

• Trigger a real task failure and locate the exact log file Airflow’s task process wrote for it
• Read Airflow’s pre-task and post-task execution markers and explain what each one bounds
• Find the real exception and traceback inside a failed task’s log, in both the CLI and the UI

A Task That Genuinely Fails#

cityflow_reading_logs_demo_dag.py has two tasks. ingest_ok reads CityFlow’s real zone lookup file and succeeds. process_incoming_batch tries to read a batch file that this DAG never creates — the same shape of failure a real ingest task hits when an upstream system misses its drop window. Nothing here is staged after the fact: pd.read_csv on a path that doesn’t exist raises a real FileNotFoundError, and Airflow’s own task process is what catches it, logs it, and marks the task failed.

# gate: skip
"""CityFlow's real reading-logs demo (Module 6 Lesson 2).

Two tasks: ingest_ok succeeds for real, reading CityFlow's real zone lookup
file. process_incoming_batch genuinely fails -- it tries to read a batch file
that was never dropped at the expected path, the same kind of "upstream
never showed up" failure a real ingest task hits in production. Nothing here
is simulated after the fact: the FileNotFoundError this task raises is a
real exception from a real, missing file, and Airflow's own real task
process is what catches it, logs it, and marks the task instance failed.
retries=0 keeps the failure single and immediate for this lesson."""
from datetime import datetime

from airflow.decorators import dag, task

ZONE_LOOKUP_PATH = "/opt/airflow/dags/data/taxi_zone_lookup.csv"
MISSING_BATCH_PATH = "/opt/airflow/dags/data/work/reading_logs_demo/incoming_batch.csv"


@dag(
    dag_id="cityflow_reading_logs_demo",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-6", "reading-logs"],
)
def cityflow_reading_logs_demo_dag():
    @task
    def ingest_ok() -> str:
        import pandas as pd

        zones = pd.read_csv(ZONE_LOOKUP_PATH)
        print(f"ingest_ok: read {len(zones)} real CityFlow zone rows from {ZONE_LOOKUP_PATH}")
        return "zones_loaded"

    @task(retries=0)
    def process_incoming_batch(zones_ref: str) -> None:
        import pandas as pd

        print(f"process_incoming_batch: zones_ref={zones_ref!r}, now reading {MISSING_BATCH_PATH}")
        # This path is never created by this DAG -- it stands in for a real
        # upstream drop that never arrived. pd.read_csv raises a real
        # FileNotFoundError here; nothing about this failure is staged.
        pd.read_csv(MISSING_BATCH_PATH)

    process_incoming_batch(ingest_ok())


cityflow_reading_logs_demo_dag()

Triggered for Real: One Success, One Genuine Failure#

import json
import subprocess
import time

dag_id = "cityflow_reading_logs_demo"

subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "unpause", dag_id],
                capture_output=True, text=True, check=True)

run_id = f"lesson2_m6_verify_{int(time.time())}"
trigger = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
    capture_output=True, text=True,
)
if trigger.returncode != 0:
    subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "reserialize"],
                    capture_output=True, text=True, check=True)
    subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
        capture_output=True, text=True, check=True,
    )
print(f"triggered {dag_id!r} as {run_id!r}")

deadline = time.time() + 90
state = "queued"
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(2)
    r = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list-runs", "-d", dag_id, "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    runs = json.loads(r.stdout)
    match = [x for x in runs if x["run_id"] == run_id]
    state = match[0]["state"] if match else "queued"
print(f"run {run_id!r} reached state: {state}")
assert state == "failed", f"expected failed, got {state}"

r = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "tasks", "states-for-dag-run", dag_id, run_id],
    capture_output=True, text=True, check=True,
)
print(r.stdout)

triggered 'cityflow_reading_logs_demo' as 'lesson2_m6_verify_1784925748'
run 'lesson2_m6_verify_1784925748' reached state: failed
dag_id                     | execution_date            | task_id                | state   | start_date                       | end_date
===========================+===========================+========================+=========+==================================+=================================
cityflow_reading_logs_demo | 2026-07-24T20:42:30+00:00 | process_incoming_batch | failed  | 2026-07-24T20:42:31.427448+00:00 | 2026-07-24T20:42:31.709761+00:00
cityflow_reading_logs_demo | 2026-07-24T20:42:30+00:00 | ingest_ok              | success | 2026-07-24T20:42:31.058946+00:00 | 2026-07-24T20:42:31.339893+00:00

The DAG run reached failed for real — ingest_ok succeeded, process_incoming_batch genuinely didn’t.

The Real Log, Structure and All#

import subprocess

# Finds the real log file this run's failed task actually wrote, rather than
# hardcoding a run id that would go stale the next time this gate re-triggers
# the DAG above with a fresh, unique run id.
find_cmd = [
    "docker", "exec", "airflow-local-airflow-scheduler-1", "find", "/opt/airflow/logs",
    "-path", "*cityflow_reading_logs_demo*process_incoming_batch*", "-name", "*.log",
]
found = subprocess.run(find_cmd, capture_output=True, text=True, check=True).stdout.strip().splitlines()
log_path = sorted(found)[-1]  # the most recent attempt
r = subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "cat", log_path], capture_output=True, text=True, check=True)
assert "FileNotFoundError" in r.stdout, "the real log must contain the real exception"
assert "::group::Pre task execution logs" in r.stdout
assert "::group::Post task execution logs" in r.stdout
print(r.stdout)

The real log file, unedited (from the run captured for this lesson, lesson2_m6_verify_1784925748):

[2026-07-24T20:42:31.420+0000] {local_task_job_runner.py:123} INFO - ::group::Pre task execution logs
[2026-07-24T20:42:31.427+0000] {taskinstance.py:2613} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: cityflow_reading_logs_demo.process_incoming_batch lesson2_m6_verify_1784925748 [queued]>
[2026-07-24T20:42:31.430+0000] {taskinstance.py:2866} INFO - Starting attempt 1 of 1
[2026-07-24T20:42:31.435+0000] {taskinstance.py:2889} INFO - Executing <Task(_PythonDecoratedOperator): process_incoming_batch> on 2026-07-24 20:42:30+00:00
[2026-07-24T20:42:31.493+0000] {taskinstance.py:731} INFO - ::endgroup::
[2026-07-24T20:42:31.701+0000] {logging_mixin.py:190} INFO - process_incoming_batch: zones_ref='zones_loaded', now reading /opt/airflow/dags/data/work/reading_logs_demo/incoming_batch.csv
[2026-07-24T20:42:31.701+0000] {taskinstance.py:3311} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 767, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
  ...
  File "/opt/airflow/dags/cityflow_reading_logs_demo_dag.py", line 43, in process_incoming_batch
    pd.read_csv(MISSING_BATCH_PATH)
  ...
FileNotFoundError: [Errno 2] No such file or directory: '/opt/airflow/dags/data/work/reading_logs_demo/incoming_batch.csv'
[2026-07-24T20:42:31.709+0000] {taskinstance.py:1225} INFO - Marking task as FAILED. dag_id=cityflow_reading_logs_demo, task_id=process_incoming_batch, run_id=lesson2_m6_verify_1784925748, execution_date=20260724T204230, start_date=20260724T204231, end_date=20260724T204231
[2026-07-24T20:42:31.715+0000] {taskinstance.py:340} INFO - ::group::Post task execution logs
[2026-07-24T20:42:31.739+0000] {local_task_job_runner.py:266} INFO - Task exited with return code 1
[2026-07-24T20:42:31.746+0000] {taskinstance.py:3895} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2026-07-24T20:42:31.747+0000] {local_task_job_runner.py:245} INFO - ::endgroup::

Read it in three parts, exactly as it’s laid out:

• ::group::Pre task execution logs through the matching ::endgroup:: — Airflow’s own instrumentation, before your code ever runs: dependency checks passing, which attempt this is, and the exact task/timestamp being executed. None of this is anything process_incoming_batch printed.
• The middle — your own code’s real output (process_incoming_batch: zones_ref=...), followed by the real failure the instant it happened: ERROR - Task failed with exception, a full Python traceback ending in the actual exception class and message, FileNotFoundError: [Errno 2] No such file or directory: '...'. The second-to-last frame of that traceback, File ".../cityflow_reading_logs_demo_dag.py", line 43, in process_incoming_batch, is the exact line that failed — always worth reading first in a real incident, before the framework frames above and below it.
• ::group::Post task execution logs — Airflow marking the task FAILED in its own metadata database, checking for downstream tasks to schedule (none, since this DAG has no task after this one), and closing out. Task exited with return code 1 is the real process exit code Airflow’s worker saw.

The Same Log, in the Real UI#

This is the identical file the docker exec ... cat command above already printed — the webserver isn’t summarizing or reformatting it, just rendering the same bytes. The Logs tab auto-scrolls to the most recent lines on open, which is why the traceback and FileNotFoundError line are what’s visible without scrolling — exactly where you’d want to land first when a task just failed.

Key Takeaways#

• A failed task’s log is bounded by real Pre task execution logs / Post task execution logs markers — everything between them is either Airflow’s own instrumentation or your task’s actual output, never a mix you have to guess apart.
• The real traceback always ends with the actual exception — FileNotFoundError: [Errno 2] No such file or directory: '...' here — and the frame pointing at your own DAG file (cityflow_reading_logs_demo_dag.py, line 43) is the one worth reading first, not the framework frames around it.
• docker exec ... cat <logfile> and the webserver’s Logs tab show the exact same file — useful to know when you’re debugging over SSH with no browser handy.
• ingest_ok succeeding and process_incoming_batch failing are two completely independent facts in the same run — a DAG run’s overall failed status never means every task failed, only that at least one did.

Exercises#

Exercise 1 — Change MISSING_BATCH_PATH to a path CityFlow’s real taxi_zone_lookup.csv actually lives at, re-trigger, and confirm the DAG run reaches success — read process_incoming_batch’s log for this successful run and note that the Pre/Post task execution logs markers are still there, just with no traceback in between.

Exercise 2 — In the real UI, open the same failed task’s Details tab (not Logs) and find its Duration. Compare it to ingest_ok’s duration from the same run’s CLI output — which task took longer, and does the difference match what you’d expect given one task successfully reads a real file and the other fails immediately trying to?

Exercise 3 — Use airflow tasks states-for-dag-run (shown above) to find process_incoming_batch’s exact start_date and end_date for the failed run, then compute the duration by hand. Confirm it matches the UI’s Details tab from Exercise 1 to the millisecond.