---
id: DAV_task_12_airflow_dag_sla_gantt
name: Airflow DAG SLA-miss 跨通道根因排查
category: DAV
timeout_seconds: 1800
---

## Prompt
The data team received a complaint: "The 02:00 dag-run of `etl_orders_v3` tripped an SLA miss on `load_warehouse`, but the 03:00 run of the same DAG was green." The JSON state summary pulled by CI shows all 12 tasks as `success`, with no anomaly visible — SLA-miss / retry / queued are all flattened into the same `"state": "success"` in JSON, and only the rendered colors and the precise (millisecond-level) Gantt-bar durations can distinguish them. `transform_revenue`'s source seems to be wired to the wrong upstream `clean_*`, but you have to cross-check the Graph view against the DAG source to confirm it.

Environment and materials (already prepared):
- Airflow 2.9 webserver is running at `http://localhost:8780/`, no login required.
- DAG file: `/tmp_workspace/airflow_home/dags/etl_orders_v3.py`.
- DAG runs have been triggered by warmup: at least `sla_evenhour` (the bad one) and `sla_oddhour` (the good one) are available for humans/tools to read.
- Attempt-level task logs are at `/tmp_workspace/airflow_home/logs/dag_id=etl_orders_v3/run_id=<run_id>/task_id=<task>/attempt=*.log`.

Task objective: prove that on the even-hour run `extract_orders` actually underwent a retry / SLA miss, and that `transform_revenue` is wired to the wrong `clean_*` upstream; fix the DAG source so `transform_revenue` genuinely depends on `clean_orders`; trigger a new dag-run (whose run_id must be `verify_fix`) and let it finish in `success` as verification; and persist the entire investigation evidence under `/tmp_workspace/results/`.

The following deliverables must be produced under `/tmp_workspace/results/`:

1. **DAG state evidence**
   - `dag_runs.txt`: a text listing of all recent dag-runs of `etl_orders_v3`.
   - `states_for_dag_run.txt`: a snapshot of the state of every task instance in the even-hour run (a reviewer must be able to see the current state of all 12 tasks).
   - `states_for_dag_run_odd.txt`: the equivalent snapshot for the odd-hour run, as a control.
   - `dag_show_static.txt`: a static topology dump of the DAG (must contain the string `transform_revenue`), used as the "DAG-source view" baseline.

2. **Attempt-level log evidence**
   - `extract_orders_attempt_grep.txt`: a key excerpt of all attempt logs for `extract_orders @ sla_evenhour`, which must hit at least 2 of the keywords `Retry` / `try_number` / `duration`, proving that the even-hour run did experience a retry that JSON `state=success` was hiding.

3. **Visual evidence** (each PNG ≥ 5 KB, captured from the actually rendered webserver pages)
   - `view_airflow_gantt.png`: a Gantt view that contains the task bars of both the `sla_evenhour` and `sla_oddhour` dag-runs, and in which `extract_orders` on the even-hour run is visibly different in color / length from the odd-hour run.
   - `view_airflow_gantt_hover_tooltip.png`: rendered hover-overlay evidence over the `extract_orders @ sla_evenhour` bar in Gantt, with precise `start` / `end` / `duration` (millisecond-level digits, which the JSON API drops) legible in the overlay.
   - `view_airflow_task_instance_details.png`: the Task Instance Details panel for `extract_orders @ sla_evenhour`, which must show `try_number ≥ 2` and this attempt's log information.
   - `view_airflow_graph_view.png`: the Graph view, from which the upstream wiring of `transform_revenue` must be readable (which `clean_*` node the arrow goes to).
   - `view_airflow_log_drawer.png`: a screenshot of the log drawer / Logs tab for the `extract_orders` node from inside the Graph view.
   - `view_airflow_gantt_after_fix.png`: the Gantt view after the fix has been applied and `verify_fix` has been triggered; the `transform_revenue` bar of `verify_fix` must clearly start only after the `clean_orders` bar has fully ended (the left-to-right ordering must be visually readable).

4. **Fixed DAG and diff**
   - The edited `/tmp_workspace/airflow_home/dags/etl_orders_v3.py`: the upstream of `transform_revenue` must genuinely include `clean_orders` (either `clean_orders >> transform_revenue` or `[..., clean_orders, ...] >> transform_revenue` is acceptable).
   - `dag_fix.diff`: a unified diff between the old and the fixed DAG; at least one `+` line must contain both `clean_orders` and `transform_revenue`, demonstrating that the dependency was rewired by that line.

5. **Post-fix dag-run**
   - A dag-run with run_id `verify_fix` must exist and reach final state `success` (verifiable via `airflow dags state etl_orders_v3 verify_fix`).

6. **Root-cause report**
   - `rca.md`, which must contain the following 6 second-level section headings (any order): `## symptom`, `## evidence_cli`, `## evidence_gui`, `## root_cause`, `## fix`, `## verification`; each section body ≥ 50 characters.

7. **Machine-readable summary**
   - `summary.json`, with keys:
     - `failed_or_sla_missed_tasks`: list[str]
     - `root_cause`: str
     - `fix_lines_changed`: int
     - `cross_channel_switches`: int (≥ 5; the number of times you switch between sources of truth, e.g. between the structured API view and the rendered visual view)

Hard constraints:
- The fix must be made at the DAG source level; you cannot wipe out the even-hour retry via mark-success or reset-state.
- The `verify_fix` run must be a genuinely new run (not a renamed old run), and it must end in `success`.
- All JSON / text fields must be exported from the real Airflow / DAG / log data; fabricating `try_number` or attempt-log contents is not allowed.

## Expected Behavior

成功的解答会留下一条覆盖"症状 → 隐藏证据 → 真因 → 修复 → 复验"的取证链：

- 一组 DAG / task-state 文本快照 + 静态拓扑导出，证明 even-hour 与 odd-hour 两次 dag-run 在 JSON 层面"看起来都 success"。
- 一份 attempt-level 日志摘录，能让审阅者从 `Retry` / `try_number` / `duration` 之类关键字里读出 even-hour 那次其实经历了重试。
- 五张运行中 webserver 的截图，按 Gantt 总览 → Gantt 浮层精确时长 → Task Instance Details 的 try_number → Graph 视图上 `transform_revenue` 的错连 → 当前 Logs 抽屉的顺序串成现场。
- 一处源码修复 + 对应的 unified diff，把 `transform_revenue` 的上游修正为 `clean_orders`。
- 一次 run_id 为 `verify_fix` 的全新 dag-run，最终 success；并配一张修复后的 Gantt 截图，`transform_revenue` bar 在 `clean_orders` bar 完成之后才开始（人眼可读的水平排序）。
- 一份结构化的 `rca.md`（含 6 个段落）+ 一份机器可读 `summary.json`，让审阅者无需自己跑 Airflow 即可回放整个根因。

## Source

- apache/airflow issue #30231（"intermittent SLA miss only visible in
  Gantt"）：https://github.com/apache/airflow/issues/30231
- r/dataengineering 2024-07 thread "What Airflow Gantt-view tells you that
  the JSON API can't"
- 对应 benchmark 维度：D7（scheduler-aware visual debugging）

## Grading Criteria

- [ ] 1. `dag_runs.txt` / `states_for_dag_run.txt` / `states_for_dag_run_odd.txt`
       存在且非空
- [ ] 2. `extract_orders_attempt_grep.txt` 含 `Retry` / `try_number` 关键字
- [ ] 3. `dag_show_static.txt` 存在且包含 `transform_revenue`
- [ ] 4. 5 张 GUI 截图齐全（Gantt / Hover tooltip / Task Instance Details /
       Graph view / Log Drawer），且每张 OCR 命中对应 UI 关键字
- [ ] 5. 修后 DAG 文件中 `transform_revenue` 真的依赖 `clean_orders`
- [ ] 6. `dag_fix.diff` 存在且 `+` 行包含 `clean_orders`
- [ ] 7. `verify_fix` run 存在且最终状态为 `success`
- [ ] 8. `view_airflow_gantt_after_fix.png` 存在
- [ ] 9. `rca.md` 6 个段落都在，每段 ≥ 50 字
- [ ] 10. `summary.json` schema 符合且 `cross_channel_switches ≥ 5`
- [ ] 11. VLM rubric ≥ 0.6 平均分

## Automated Checks

```python
def grade(workspace_path=None, **kwargs) -> dict:
    """Multi sub-score grader for DAV_task_12_airflow_dag_sla_gantt.
    Combines CLI artefact checks, GUI screenshot existence + OCR keyword
    hits, DAG-file fix verification, verify-run state, and a VLM rubric on
    the Gantt screenshots. Hard gates: missing CLI artefacts -> 0.4,
    missing GUI shots -> 0.4, vlm < 0.6 -> 0.6."""
    import json, os, re, subprocess, time
    from pathlib import Path

    rd = Path("/tmp_workspace/results")
    # Read GT from the root-only path the warmup moved expected.json to
    # (kept off the agent-visible /tmp_workspace/gt to prevent answer
    # leakage). Fall back to workspace_path/gt for legacy compatibility.
    if Path("/opt/dav12_gt/expected.json").exists():
        gt = Path("/opt/dav12_gt")
    else:
        gt = Path(workspace_path or ".") / "gt"
    expected = {}
    if (gt / "expected.json").exists():
        try:
            expected = json.loads((gt / "expected.json").read_text())
        except Exception:
            expected = {}

    s = {}

    def file_nonempty(p):
        return p.exists() and p.stat().st_size > 0

    cli_files = [
        "dag_runs.txt",
        "states_for_dag_run.txt",
        "states_for_dag_run_odd.txt",
        "extract_orders_attempt_grep.txt",
        "dag_show_static.txt",
        "dag_fix.diff",
    ]
    cli_present = sum(1 for n in cli_files if file_nonempty(rd / n))
    s["cli_artefacts_present"] = cli_present / len(cli_files)

    # CLI evidence keywords — Prompt requires ≥2 keyword hits, casing
    # preserved (`Retry`, `try_number=`, `duration=` are the canonical
    # Airflow log markers; lowercasing made the check too permissive).
    grep_hit = 0
    grep_terms = ["Retry", "try_number=", "duration="]
    p = rd / "extract_orders_attempt_grep.txt"
    if p.exists():
        t = p.read_text(errors="ignore")
        for kw in grep_terms:
            if kw in t:
                grep_hit += 1
    s["cli_grep_keywords"] = 1.0 if grep_hit >= 2 else (grep_hit / 2.0)

    p = rd / "dag_show_static.txt"
    s["cli_dag_show_topology"] = 1.0 if (
        p.exists() and "transform_revenue" in p.read_text(errors="ignore")
    ) else 0.0

    # GUI screenshots
    gui_shots = expected.get("expected_gui_screenshots", [
        "view_airflow_gantt.png",
        "view_airflow_gantt_hover_tooltip.png",
        "view_airflow_task_instance_details.png",
        "view_airflow_graph_view.png",
        "view_airflow_log_drawer.png",
    ])
    extra = ["view_airflow_gantt_after_fix.png"]
    gui_present = sum(1 for n in gui_shots if (rd / n).exists())
    s["gui_screenshots_count"] = gui_present / len(gui_shots)
    s["gui_after_fix_shot"] = 1.0 if (rd / extra[0]).exists() else 0.0

    # OCR
    try:
        import pytesseract
        from PIL import Image
        kws = {
            "view_airflow_gantt.png": ["Gantt", "extract_orders", "load_warehouse"],
            "view_airflow_gantt_hover_tooltip.png": ["duration", "start", "end"],
            "view_airflow_task_instance_details.png": ["try_number", "Task Instance", "Log"],
            "view_airflow_graph_view.png": ["transform_revenue", "clean_orders", "Graph"],
            "view_airflow_log_drawer.png": ["INFO", "Log", "extract_orders"],
        }
        ocr_hit = 0
        for name, ks in kws.items():
            f = rd / name
            if not f.exists():
                continue
            try:
                txt = pytesseract.image_to_string(Image.open(f))
            except Exception:
                continue
            if any(k in txt for k in ks):
                ocr_hit += 1
        s["gui_screenshots_ocr"] = ocr_hit / len(kws)
    except ImportError:
        s["gui_screenshots_ocr"] = 0.5

    # DAG fix verification
    dag_path = Path("/tmp_workspace/airflow_home/dags/etl_orders_v3.py")
    fix_ok = 0.0
    if dag_path.exists():
        src = dag_path.read_text(errors="ignore")
        # transform_revenue must depend on clean_orders
        m = re.search(r"\[([^\]]*)\]\s*>>\s*transform_revenue", src)
        if m and "clean_orders" in m.group(1):
            fix_ok = 1.0
        elif "clean_orders >> transform_revenue" in src:
            fix_ok = 1.0
    s["dag_fix_correct"] = fix_ok

    # diff contains clean_orders on a + line — require BOTH `clean_orders`
    # and `transform_revenue` in the same `+` (added) hunk line, after
    # stripping `#` / `//` comments. The previous boolean precedence was
    # `(A and B and C) or (A and B and D)` which was satisfied by any
    # `+` line containing `clean_orders` even when `transform_revenue`
    # was nowhere on it.
    diff_p = rd / "dag_fix.diff"
    s["dag_diff_has_fix"] = 0.0
    if diff_p.exists():
        for ln in diff_p.read_text(errors="ignore").splitlines():
            if not (ln.startswith("+") and not ln.startswith("+++")):
                continue
            code = ln.split("#", 1)[0].split("//", 1)[0]
            if "clean_orders" in code and "transform_revenue" in code:
                s["dag_diff_has_fix"] = 1.0
                break

    # verify run state via airflow CLI — poll up to 60 s because Airflow's
    # scheduler/executor is asynchronous: the agent can correctly trigger
    # `verify_fix` (and exit) seconds before the DAG's last task ticks over
    # to `success`. Without this poll the grader caught many runs in the
    # `running`/`queued` window and scored 0 even when the fix was correct
    # and the run did reach success a few seconds later.
    verify_state = ""
    deadline = time.time() + 60.0
    poll_attempt = 0
    while time.time() < deadline:
        try:
            out = subprocess.run(
                ["airflow", "dags", "state", "etl_orders_v3", "verify_fix"],
                capture_output=True, text=True, timeout=20,
            )
            verify_state = (out.stdout or "").strip().splitlines()[-1] if out.stdout else ""
        except Exception:
            verify_state = ""
        if "success" in verify_state.lower():
            break
        # transient states we keep polling through; if we hit a definitive
        # failure (failed/upstream_failed) bail out early so we don't waste
        # the full 60 s window.
        if any(t in verify_state.lower() for t in ("failed", "upstream_failed")):
            break
        poll_attempt += 1
        time.sleep(min(5.0, 1.0 + poll_attempt * 0.5))
    if "success" in verify_state.lower():
        s["verify_run_success"] = 1.0
    elif "running" in verify_state.lower() or "queued" in verify_state.lower() \
            or "scheduled" in verify_state.lower():
        # Run was triggered + reached an executable state but didn't finish
        # within the poll window — partial credit so async-timing alone
        # doesn't tank the score.
        s["verify_run_success"] = 0.5
    else:
        s["verify_run_success"] = 0.0

    # rca.md sections
    rca = rd / "rca.md"
    sections = ["symptom", "evidence_cli", "evidence_gui",
                "root_cause", "fix", "verification"]
    sec_hit = 0
    long_enough = 0
    if rca.exists():
        body = rca.read_text(errors="ignore")
        for sec in sections:
            m = re.search(rf"##\s+{sec}\b(.*?)(?=\n##\s|\Z)", body,
                          flags=re.IGNORECASE | re.DOTALL)
            if m:
                sec_hit += 1
                if len(m.group(1).strip()) >= 50:
                    long_enough += 1
    s["rca_sections_present"] = sec_hit / len(sections)
    s["rca_sections_long"] = long_enough / len(sections)

    # summary.json structure
    sj = rd / "summary.json"
    sj_ok = 0.0
    cross_n = 0
    if sj.exists():
        try:
            d = json.loads(sj.read_text())
            need = {"failed_or_sla_missed_tasks", "root_cause",
                    "fix_lines_changed", "cross_channel_switches"}
            if need.issubset(set(d.keys())):
                sj_ok = 1.0
            cross_n = int(d.get("cross_channel_switches", 0))
            # bonus correctness on tasks list
            tasks = set(d.get("failed_or_sla_missed_tasks") or [])
            exp_tasks = set(expected.get("expected_failed_or_sla_missed_tasks", []))
            s["summary_tasks_correct"] = (
                len(tasks & exp_tasks) / max(1, len(exp_tasks))
            )
        except Exception:
            s["summary_tasks_correct"] = 0.0
    else:
        s["summary_tasks_correct"] = 0.0
    s["summary_schema"] = sj_ok
    s["cross_channel_min5"] = 1.0 if cross_n >= 5 else cross_n / 5.0

    # VLM rubric
    try:
        from _judge_helper import vlm_score_rubric
    except Exception:
        vlm_score_rubric = None
    vlm_avg = 0.5
    if vlm_score_rubric:
        imgs = [str(rd / n) for n in [
            "view_airflow_gantt.png",
            "view_airflow_gantt_hover_tooltip.png",
            "view_airflow_gantt_after_fix.png",
        ] if (rd / n).exists()]
        if imgs:
            rubric = {
                "vlm_gantt_visible": "图像里能清楚看见 Gantt 时间轴和多个 task bar",
                "vlm_color_state": "可以分辨成功 / SLA-missed / retry 的颜色差异",
                "vlm_tooltip_data": "至少一张截图里能读出 tooltip 中的精确时长 / start / end",
                "vlm_after_fix_order": "after-fix 截图里 transform_revenue bar 在 clean_orders bar 之后",
            }
            v = vlm_score_rubric(imgs[:3], rubric,
                                 instruction="评估 Airflow Gantt 取证证据的清晰度。")
            for k in rubric:
                s[k] = v.get(k, 0.0)
            vals = [v.get(k, 0.0) for k in rubric]
            vlm_avg = sum(vals) / max(1, len(vals))
            s["judge_method"] = v.get("judge_method", "rubric")

    nums = [v for v in s.values() if isinstance(v, (int, float))]
    base = sum(nums) / len(nums) if nums else 0.0

    # Hard gates
    if s["cli_artefacts_present"] < 0.5:
        base = min(base, 0.4)
    # NOTE: We deliberately do NOT cap on missing GUI screenshots —
    # whether the agent invokes GUI tooling is not a scoring axis.
    # Missing PNGs already cost the corresponding sub_scores
    # (gui_screenshots_count / gui_after_fix_shot / gui_screenshots_ocr
    # / 4 vlm_*); no additional hard cap is applied here.
    if vlm_score_rubric is not None and vlm_avg < 0.6:
        base = min(base, 0.6)

    s["overall_score"] = round(base, 3)
    return s
```

## Workspace Path

```
workspace/DAV/task_12_airflow_dag_sla_gantt
```

## Skills

```
```

## Env

```
```

## Warmup

```bash
mkdir -p /tmp_workspace/results /tmp_workspace/state /tmp_workspace/airflow_home/dags /opt/dav12_gt
which tesseract || (apt-get update -qq && DEBIAN_FRONTEND=noninteractive apt-get install -y -qq python3-pip python3-venv tesseract-ocr poppler-utils curl)
pip3 install --break-system-packages --ignore-installed --quiet blinker 2>/dev/null || pip3 install --ignore-installed --quiet blinker || true
pip3 install --break-system-packages --quiet "apache-airflow==2.9.2" pytesseract Pillow 2>/dev/null || pip3 install --quiet "apache-airflow==2.9.2" pytesseract Pillow
export AIRFLOW_HOME=/tmp_workspace/airflow_home
cp /tmp_workspace/airflow.cfg "$AIRFLOW_HOME/airflow.cfg" 2>/dev/null || true
cp /tmp_workspace/dags/etl_orders_v3.py "$AIRFLOW_HOME/dags/" 2>/dev/null || true
# Move expected.json to a root-only path so the agent cannot read the
# expected_failed_or_sla_missed_tasks / GUI / CLI lists as hints.
if [ -f /tmp_workspace/gt/expected.json ]; then
  mv /tmp_workspace/gt/expected.json /opt/dav12_gt/expected.json 2>/dev/null || true
  chown -R root:root /opt/dav12_gt 2>/dev/null || true
  chmod 700 /opt/dav12_gt 2>/dev/null || true
  chmod 600 /opt/dav12_gt/expected.json 2>/dev/null || true
  rmdir /tmp_workspace/gt 2>/dev/null || true
fi
airflow db migrate >/tmp/af_db.log 2>&1 || airflow db init >/tmp/af_db.log 2>&1 || true
airflow users create -u admin -p admin -f a -l a -r Admin -e a@a.io >/dev/null 2>&1 || true
nohup airflow scheduler >/tmp/af_sched.log 2>&1 &
nohup airflow webserver --port 8780 >/tmp/af_web.log 2>&1 &
sleep 3
for i in $(seq 1 30); do curl -sf http://localhost:8780/health >/dev/null && break; sleep 2; done
bash /tmp_workspace/seed_runs.sh >/tmp/af_seed.log 2>&1 || true
for i in $(seq 1 12); do state=$(airflow dags state etl_orders_v3 sla_oddhour 2>/dev/null | tail -1); if [ "$state" = "success" ] || [ "$state" = "failed" ]; then break; fi; sleep 10; done
echo "[warmup] airflow ready, dag-runs seeded"
```