Scheduling Validation DAGs with Apache Airflow

This page builds the concrete cadence-driven pipeline that the drill scheduling and orchestration guide describes in the abstract: a complete Apache Airflow DAG that runs a nightly backup-validation and restore-drill campaign end to end. The DAG resolves the newest backup artifact with a sensor, hands it to sandbox provisioning, runs a checksum validation pipeline that gates strictly on the runner’s POSIX exit code, tears the sandbox down whether or not validation passed, and finally writes the run into your immutable audit trails. Every task is thin and delegates real work to a runner that returns an exit code, so Airflow does what it is good at — sequencing, retrying, and recording — and nothing it is not.

Architecture and Execution Model

Figure. The five tasks of the nightly campaign, from artifact resolution through exit-code-gated validation to guaranteed teardown and audit emission.

The flow is linear in the happy path but branches on failure semantics that the diagram cannot show: teardown and audit run on ALL_DONE rather than all_success, so a validation failure still reclaims the sandbox and still records the run. That asymmetry — most tasks gate on success, but cleanup and audit gate only on completion — is the single design decision that makes the DAG safe to leave running unattended.

Prerequisites

• Python 3.9+ on the scheduler and worker hosts.

• Apache Airflow 2.9+, installed with the constraint file the project publishes:

  pip install "apache-airflow>=2.9"
  

• Drill runner scripts on PATH at /opt/dr/bin — resolve_newest.py, provision_sandbox.py, validate_restore.py, teardown_sandbox.py, and emit_audit.py — each exiting 0 on success, 1 on a hard failure, and 2 on a configuration error.

• An executor that supports sensor reschedule mode (Celery, Kubernetes, or Local with enough slots) so the artifact sensor releases its worker slot between pokes.

• Append-only audit storage reachable from the worker, so emit_audit.py can write a record that cannot be edited after the fact.

Production Implementation

The full DAG is one file. It uses the TaskFlow API so data — the resolved artifact id — flows between tasks as return values rather than manual XCom plumbing, and it keeps each task’s failure semantics explicit.

"""Nightly backup-validation + restore-drill campaign (Apache Airflow).

Resolves the newest backup artifact, provisions an isolated sandbox, runs a
validation that gates on the runner's POSIX exit code, tears the sandbox down
regardless of outcome, and emits an immutable audit record.

    pip install "apache-airflow>=2.9"
"""
from __future__ import annotations

import subprocess

import pendulum
from airflow.decorators import dag, task
from airflow.exceptions import AirflowException
from airflow.sensors.base import PokeReturnValue
from airflow.utils.trigger_rule import TriggerRule

RUNNER = "/opt/dr/bin"


def _exit_code(argv: list[str]) -> int:
    """Run a drill step and return its POSIX exit code without raising."""
    return subprocess.run(argv, text=True).returncode


@dag(
    dag_id="nightly_validation_campaign",
    schedule="0 2 * * *",             # 02:00 daily — the calendar cadence
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,                    # skip missed windows; never backfill
    max_active_runs=1,                # one campaign at a time protects infra
    default_args={
        "retries": 2,
        "retry_delay": pendulum.duration(minutes=5),
    },
    tags=["dr", "validation", "nightly"],
)
def nightly_validation_campaign():

    @task.sensor(poke_interval=60, timeout=1800, mode="reschedule")
    def resolve_newest_artifact() -> PokeReturnValue:
        """Poke the artifact store until a fresh backup is present."""
        proc = subprocess.run(
            [f"{RUNNER}/resolve_newest.py"], capture_output=True, text=True)
        if proc.returncode == 0 and proc.stdout.strip():
            return PokeReturnValue(is_done=True, xcom_value=proc.stdout.strip())
        return PokeReturnValue(is_done=False)

    @task(retries=3, retry_delay=pendulum.duration(minutes=2))
    def provision_sandbox(artifact_id: str) -> str:
        code = _exit_code([f"{RUNNER}/provision_sandbox.py", artifact_id])
        if code != 0:
            raise AirflowException(
                f"provision failed for {artifact_id}: exit {code}")
        return artifact_id

    @task(sla=pendulum.duration(hours=1))
    def run_validation(artifact_id: str) -> int:
        code = _exit_code([f"{RUNNER}/validate_restore.py", artifact_id])
        if code == 2:
            raise AirflowException(f"config error validating {artifact_id}")
        if code == 1:
            raise AirflowException(
                f"validation FAILED for {artifact_id}: quarantine")
        return code  # 0 == restored inside the recovery envelope

    @task(trigger_rule=TriggerRule.ALL_DONE)
    def teardown_sandbox(artifact_id: str) -> None:
        """Always reclaim the sandbox, even after a failed validation."""
        _exit_code([f"{RUNNER}/teardown_sandbox.py", artifact_id])

    @task(trigger_rule=TriggerRule.ALL_DONE)
    def emit_audit(artifact_id: str) -> None:
        """Write an append-only audit record whether the run passed or failed."""
        subprocess.run([f"{RUNNER}/emit_audit.py", artifact_id], text=True)

    artifact = resolve_newest_artifact()
    provisioned = provision_sandbox(artifact)
    validated = run_validation(provisioned)
    teardown = teardown_sandbox(provisioned)
    teardown.set_upstream(validated)
    audit = emit_audit(artifact)
    audit.set_upstream(teardown)


dag_object = nightly_validation_campaign()

Scheduling Parameters Explained

Three DAG-level parameters carry most of the operational weight. schedule="0 2 * * *" fixes the cadence at 02:00 daily; Airflow computes the next run from the data interval, so you never manage a cron entry separately. catchup=False is the parameter teams most often get wrong — with the default True, restarting a scheduler that was down for a week would immediately queue seven missed campaigns, stampeding the shared restore infrastructure the moment it comes back. For drills you almost always want to skip missed windows and run only the current one. max_active_runs=1 guarantees the shared sandbox environment is never contended by two campaigns at once; if a run overruns into the next window, the next run waits rather than doubling the load.

The task-level parameters refine that further. The sensor’s mode="reschedule" frees its worker slot between pokes so a slow artifact store does not pin a worker for thirty minutes. The retries on provision_sandbox absorb transient cloud-API flakiness. The sla on run_validation records an SLA-miss event if the validation runs but exceeds its one-hour recovery-time budget — a passing-but-late drill is still a finding you want surfaced against your RTO.

Two of these are easy to set carelessly. The sensor timeout of 1800 seconds bounds how long the campaign waits for a fresh backup before giving up; set it too low and a slightly delayed backup fails the whole drill, set it too high and a genuinely missing backup wastes half an hour before it surfaces. Tie the value to the tail of your backup-completion distribution, not to a round number. The poke_interval trades responsiveness against load on the artifact store — sixty seconds is comfortable for an object store, but a metadata API with tight rate limits may need a longer interval or an exponential poke. Neither parameter is decorative; both directly shape how the campaign behaves on the nights something upstream is late.

How data flows between tasks

The TaskFlow API turns return values into XComs automatically, which is why the DAG never touches ti.xcom_push or ti.xcom_pull. The sensor’s PokeReturnValue carries an xcom_value — the resolved artifact id — that becomes the return of resolve_newest_artifact, and passing that return into provision_sandbox(artifact) establishes both the data dependency and the execution edge in one call. Because the id threads through every task, the audit record at the end is provably about the same artifact the sensor resolved at the start, with no room for a mismatch introduced by a separate lookup. Keep what flows through XCom small — an id, a path, a status — never a payload; XCom is backed by the metadata database and is the wrong place for anything larger than a reference.

Task-Level Failure Semantics

The DAG encodes three distinct failure behaviors deliberately. provision_sandbox and run_validation raise on any non-zero exit, which marks them failed and — under the default all_success trigger rule — skips nothing downstream that should not run, because the only tasks after them are the ALL_DONE cleanup pair. run_validation further distinguishes exit 2 (a configuration fault) from exit 1 (a genuine data divergence); both raise, but they carry different messages so error categorization downstream can escalate them as separate event classes. teardown_sandbox and emit_audit use TriggerRule.ALL_DONE, meaning they run once their upstreams reach any terminal state — success, failure, or skip — so the sandbox is always reclaimed and the run is always recorded even when validation fails hard.

There is a deliberate reason provision_sandbox raises on any non-zero code while run_validation branches on the specific value. Provisioning has no meaningful “the data is bad” outcome — either the sandbox came up or it did not — so any failure is treated uniformly and, thanks to its retries, retried a few times before the campaign gives up. Validation, by contrast, produces two failures with opposite handling: an exit 2 means the check never actually ran and the artifact’s true state is unknown, whereas an exit 1 means the check ran and the artifact is provably corrupt. Collapsing those into one exception would let a configuration fault masquerade as a caught corruption in the audit record, which is exactly the confusion the exit-code contract exists to prevent. The messages differ so the downstream categorization layer, and the human reading the failed run, can tell “we could not test this” from “we tested this and it failed.”

The emit_audit task deserves a closer look because it is the one task that must never itself fail silently. It runs on ALL_DONE after teardown, so it fires on every run regardless of outcome, and it writes to append-only storage. If audit emission can fail, wrap its runner to retry and alert, because an unrecorded drill is worse than a failed one: a failed drill is a visible finding, while a drill with no audit record is indistinguishable from a night the campaign never ran at all.

Step-by-Step Execution Walkthrough

1. Drop the file into the DAGs folder. Place the module under $AIRFLOW_HOME/dags/; the scheduler parses it and the DAG appears as nightly_validation_campaign.
2. Unpause the DAG. Toggle it on in the UI or run airflow dags unpause nightly_validation_campaign. It will fire at the next 02:00 boundary.
3. Trigger a manual test run. airflow dags trigger nightly_validation_campaign starts a run immediately without waiting for the schedule.
4. Watch the sensor. resolve_newest_artifact pokes every 60 seconds up to a 30-minute timeout; in reschedule mode it releases its slot between pokes and resumes when the store reports a fresh artifact.
5. Follow the gate. Once provisioning succeeds, run_validation runs; a non-zero exit marks it failed and the campaign stops promoting the artifact.
6. Confirm cleanup and audit. Regardless of the validation result, teardown_sandbox and emit_audit run to completion, so no sandbox is orphaned and every run leaves a record.

Verification and Expected Output

A clean nightly run produces task logs resembling the following, with the DAG run marked success:

[resolve_newest_artifact] poke 1: no fresh artifact yet, rescheduling
[resolve_newest_artifact] poke 2: resolved backup-2026-07-18T01Z -> done
[provision_sandbox] sandbox up for backup-2026-07-18T01Z (exit 0)
[run_validation] restore + checksum passed for backup-2026-07-18T01Z (exit 0)
[teardown_sandbox] sandbox reclaimed (exit 0)
[emit_audit] audit record written: run=nightly_validation_campaign VALID
Marking run <DagRun nightly_validation_campaign @ 2026-07-18T02:00> success

Success means the sensor resolved exactly one artifact, provisioning and validation both returned 0, and the cleanup and audit tasks completed. A failed validation instead shows run_validation in the failed state with a validation FAILED ... quarantine message, yet teardown_sandbox and emit_audit still report success — the run is marked failed overall, the sandbox is gone, and the audit record captures the failure.

Failure Modes and Troubleshooting

Integration Notes

The DAG is deliberately self-contained but designed to compose. The artifact id the sensor resolves is the join key across every downstream system: pass it into the immutable audit trails store so the audit record ties the run to a specific backup, and emit the per-task exit codes into your metrics backend so the campaign’s health is observable across generations rather than only in the UI’s retention window. When the program grows beyond nightly cadence into event-driven per-artifact validation, this DAG becomes the scheduled half of a hybrid alongside a Celery graph, both sharing the same exit-code contract so a failure means the same thing regardless of which engine caught it.

Frequently Asked Questions

Related

• Drill scheduling and orchestration — the parent guide covering the full scheduling loop this DAG implements.
• Airflow vs Celery for DR Drill Orchestration — the decision framework for choosing the engine behind this DAG.
• Sandbox provisioning automation — the provisioning stage the DAG’s second task invokes.
• Immutable audit trails — where the DAG’s final task writes each run’s evidence.

This DAG is one component of the broader drill scheduling and orchestration guide. For authoritative behavior consult the Apache Airflow documentation and the Python subprocess reference.