Surface worker DAG parse duration in task log

[kaxil]

Customers and support engineers cannot tell whether a task's apparent slow startup (opaque gap between start_date and the first task log line) is due to their DAG file complexity or to Airflow-side bundle operations (git fetch, etc.).

Background

The worker's DAG parse happens after the TI state transitions to RUNNING:

1. Supervisor calls client.task_instances.start(...) which hits the ti_run endpoint -- this sets TI.state=RUNNING and writes TI.start_date.
2. Supervisor then sends StartupDetails to the worker subprocess.
3. Worker calls parse() -- bundle fetch + DAG file import happen here.
4. Worker runs the task.

So the existing queued_duration metric captures queued_dttm → RUNNING (pre-parse: pod boot, worker fork, supervisor handshake, API roundtrip). The parse phase itself is invisible in the UI today and shows up only as a gap between start_date and the first task log line.

What this PR does

Times the two phases inside parse() and emits a single structured log event just before return:

log.info(
    "Worker startup parse complete",
    bundle_name=...,
    bundle_version=...,
    dag_file=...,
    dag_id=...,
    bundle_prepare_ms=...,   # bundle fetch/verify (Airflow overhead)
    dag_file_parse_ms=...,   # DAG file import (user code)
)

The event lands in the customer's task log file (structlog kwargs, JSON-serialized by the supervisor's log channel). Support engineers can now answer "is it them or us?" from the first log line.

parse()'s public signature is unchanged (RuntimeTaskInstance); no test or pytest-fixture updates needed. Single file, ~18 lines added.

Design rationale

Why a log line and not a Stats.timing metric? Stats goes to StatsD/Prometheus dashboards, which most customers don't own. The task log file is what customers read when debugging slow tasks -- that's where the answer needs to be. Adding Stats.timing also requires registering metrics in shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml and following the _duration naming convention -- deferred as a separate PR if ops wants dashboard integration.

Why split bundle_prepare vs dag_file_parse? Different failure modes, different owners. bundle_prepare_ms is bimodal by nature: near-zero for warm GitDagBundle (cached worktree), non-trivial on first task after a bundle version change. dag_file_parse_ms attributes to user DAG code (heavy imports, top-level time.sleep, etc.).

Why emit from inside parse() instead of from startup()? Keeping parse()'s signature stable avoids breaking downstream callers (pytest fixtures, tests, any unknown use). Under run_as_user impersonation, parse() runs twice (pre-sudo and post-sudo), so this event fires twice for those tasks. The two entries expose impersonation re-parse overhead, which is informative rather than a defect.

Gotchas / known limitations

• Pre-parse gap is not measured. Worker process fork, Python import, supervisor handshake, and StartupDetails fetch can dominate on cold K8sExecutor pods. queued_duration already measures that full pre-RUNNING gap end-to-end; this PR adds the post-RUNNING phase.
• Warm GitDagBundle shows near-zero bundle_prepare_ms. Expected -- the cached worktree check short-circuits. Only cold pulls (first task after a version change) emit non-trivial values, and that latency is real.
• run_as_user impersonation emits the event twice per task. Once in the pre-sudo process, once in the sudo'd process. Customers reading the task log see both -- the second one is the measurement that matters for their task's real startup.