[v3-2-test] Include TI UUID in scheduler, DAG processor, triggerer, and worker logs (#65458)

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com

[github-actions]

Support engineers could not reconstruct a task's full lifecycle from logs
because only the Execution API emitted the TaskInstance UUID consistently.
Adding ti_id to log lines across the other components makes 'grep ti_id=X'
surface every log touching that task, from scheduling through completion.

• Worker: bind ti_id to structlog context at startup(). Fresh process per
  TI means no cross-task leak risk.
• Triggerer: extend existing bind_log_contextvars at trigger start. The
  asyncio.create_task context copy scopes the binding per coroutine.
• Scheduler: add ti_id=%s to eight TI-touching log calls across
  _enqueue_task_instances_with_queued_state, process_executor_events,
  and _maybe_requeue_stuck_ti. Explicit positional args avoid the
  contextvar leak a bind+unbind pattern would introduce on exception paths.
• DAG processor: add ti_id to callback-processing log lines in
  _execute_callbacks and _execute_task_callbacks.

• Move ti_id into TaskInstance.repr; revert redundant log-line additions

Addresses review feedback from @jedcunningham on #65458: instead of
sprinkling ti_id=%s onto individual scheduler log lines, put the UUID in
TaskInstance.repr once and let every %s-formatted TI log line inherit
it for free. Strictly better: covers log lines this PR didn't touch and
lines added by future PRs without further plumbing.

Net diff vs main goes from +61/-16 to +51/-14.

Changes:

• TaskInstance.repr now appends ti_id={self.id} before the closing
  bracket (matches the existing TaskInstanceNote repr precedent).
• Reverted 10 log-line ti_id additions in scheduler_job_runner.py where
  the existing %s format arg was a TaskInstance; repr now supplies ti_id.
• Kept the explicit ti_id=%s in the "TaskInstance Finished" msg: it
  formats individual fields (dag_id, task_id, etc.), not %s on the TI,
  so the repr shortcut does not apply.
• Kept DAG processor structlog-kwargs ti_id additions: those go through
  structlog's kwargs path, not repr.
• Updated one test assertion in test_scheduler_job.py that hardcoded the
  exact TaskInstance repr string.

• Update test_not_enough_pool_slots for new TaskInstance repr

After adding ti_id to TaskInstance.repr, test_not_enough_pool_slots
needs to include ti_id in the expected log substring. Same fix pattern
as test_process_executor_events_with_callback at line 695.

• Fix test_not_enough_pool_slots ordering assumption on MySQL

dr.task_instances[0] can return can_run first on MySQL (alphabetical
default ordering) instead of cannot_run, so the expected ti_id used
in the "Not executing" assertion grabbed the wrong task's UUID and
the substring check failed on MySQL CI even though it passed on SQLite.

Look up the TI by task_id instead to make the assertion order-independent.
(cherry picked from commit 1a0efe7)

Co-authored-by: Kaxil Naik kaxilnaik@gmail.com