kaxil opened a new pull request, #65458: URL: https://github.com/apache/airflow/pull/65458
Engineers currently cannot grep a single `ti_id` and reconstruct a task's full lifecycle timeline. The Execution API binds `ti_id` to every log line via `bind_contextvars`, but scheduler, DAG processor, triggerer, and worker each emit it in some places and not others. This PR closes that gap. ## What changed | Component | Change | Rationale | |---|---|---| | **Worker** (`task-sdk/.../task_runner.py`) | `bind_contextvars(ti_id=str(msg.ti.id), ...)` at the top of `startup()` | One worker process = one TI. No cross-task leak risk, so context binding is strictly correct and instruments every subsequent log line (including ones added in future changes). | | **Triggerer** (`jobs/triggerer_job_runner.py:1308`) | Extend the existing `bind_log_contextvars(trigger_id=...)` to also bind `ti_id` + composite TI keys when `trigger.task_instance` is present | `run_trigger` runs inside `asyncio.create_task`, which snapshots the context at creation, so the binding stays scoped to that coroutine. | | **Scheduler** (`jobs/scheduler_job_runner.py`) | Add `ti_id=%s` as a positional arg to eight TI-touching log calls across `_enqueue_task_instances_with_queued_state`, `process_executor_events`, and `_maybe_requeue_stuck_ti` | Explicit positional args (no `bind_contextvars`) because the scheduler is a long-running process and any unhandled exception mid-loop would leave a stale `ti_id` bound for the rest of the process lifetime, corrupting all subsequent log lines. Per-iteration `with bound_contextvars(...)` would be correct but re-indents ~200 lines of hot-path code. Positional args are both safer and less invasive. | | **DAG processor** (`dag_processing/processor.py`) | Add `ti_id` to `_execute_callbacks` and `_execute_task_callbacks` log calls (kwarg where the call is pure-structlog style, `%-format` positional where the call is stdlib style) | Same reasoning as scheduler. | ## Design notes - **Why not `bind_contextvars` everywhere?** The earlier draft of this PR did exactly that. But we found a contextvar leak: if any exception propagates out of the per-TI loop body, the post-loop `unbind_contextvars` is skipped and the last TI's `ti_id` stays bound for the remainder of the scheduler process. Because `merge_contextvars` is wired into the stdlib logging `foreign_pre_chain`, this taints every subsequent log line -- including ones that have nothing to do with that TI. Inverts the goal of the PR. Per-iteration `bound_contextvars` context manager is correct but requires re-indenting long loop bodies. For the scheduler and DAG processor, explicit positional `ti_id=%s` args are both safer and less invasive. - **Why is it OK to bind in the worker?** A worker is a fresh process that runs exactly one TI. There is no cross-iteration or cross-TI state to leak into. The bind is strictly correct and instruments every log line for free. - **Why is it OK to bind in the triggerer?** `run_trigger` is launched via `asyncio.create_task`, which copies the current `contextvars.Context` at task creation. Any contextvar bound inside the coroutine is scoped to that task and cannot leak into another trigger's coroutine or the triggerer's supervisor loop. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
