This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1a0efe7d456 Include TI UUID in scheduler, DAG processor, triggerer,
and worker logs (#65458)
1a0efe7d456 is described below
commit 1a0efe7d4564afb8b38e10d939b70bdba25fc2ee
Author: Kaxil Naik <[email protected]>
AuthorDate: Sun Apr 19 01:44:07 2026 +0100
Include TI UUID in scheduler, DAG processor, triggerer, and worker logs
(#65458)
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.
---
.../src/airflow/dag_processing/processor.py | 24 +++++++++++++++++++---
.../src/airflow/jobs/scheduler_job_runner.py | 12 +++++------
.../src/airflow/jobs/triggerer_job_runner.py | 11 +++++++++-
airflow-core/src/airflow/models/taskinstance.py | 2 +-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 8 +++++---
.../src/airflow/sdk/execution_time/task_runner.py | 14 +++++++++++--
6 files changed, 55 insertions(+), 16 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index 758472598d2..6f8c293fe5e 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -301,7 +301,14 @@ def _execute_callbacks(
dagbag: DagBag, callback_requests: list[CallbackRequest], log:
FilteringBoundLogger
) -> None:
for request in callback_requests:
- log.debug("Processing Callback Request", request=request.to_json())
+ if isinstance(request, (TaskCallbackRequest, EmailRequest)):
+ log.debug(
+ "Processing Callback Request",
+ request=request.to_json(),
+ ti_id=str(request.ti.id),
+ )
+ else:
+ log.debug("Processing Callback Request", request=request.to_json())
with BundleVersionLock(
bundle_name=request.bundle_name,
bundle_version=request.bundle_version,
@@ -366,6 +373,7 @@ def _execute_task_callbacks(dagbag: DagBag, request:
TaskCallbackRequest, log: F
dag_id=request.ti.dag_id,
task_id=request.ti.task_id,
run_id=request.ti.run_id,
+ ti_id=str(request.ti.id),
)
return
@@ -415,11 +423,21 @@ def _execute_task_callbacks(dagbag: DagBag, request:
TaskCallbackRequest, log: F
for idx, callback in enumerate(callbacks):
callback_repr = get_callback_representation(callback)
- log.info("Executing Task callback at index %d: %s", idx, callback_repr)
+ log.info(
+ "Executing Task callback at index %d: %s (ti_id=%s)",
+ idx,
+ callback_repr,
+ request.ti.id,
+ )
try:
callback(context)
except Exception:
- log.exception("Error in callback at index %d: %s", idx,
callback_repr)
+ log.exception(
+ "Error in callback at index %d: %s (ti_id=%s)",
+ idx,
+ callback_repr,
+ request.ti.id,
+ )
def _execute_email_callbacks(dagbag: DagBag, request: EmailRequest, log:
FilteringBoundLogger) -> None:
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index d949012d4f5..e9db7625c38 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -971,9 +971,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
continue
self.log.debug(
- "Queueing workload for TI: %s id=%s try_number=%d state=%s
scheduler_job_id=%s executor=%s",
+ "Queueing workload for TI: %s try_number=%d state=%s
scheduler_job_id=%s executor=%s",
ti,
- ti.id,
ti.try_number,
ti.state,
self.job.id,
@@ -1250,12 +1249,11 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if ti.try_number != try_number:
cls.logger().warning(
"TI try_number mismatch: db_try_number=%d
event_try_number=%d "
- "ti=%s ti_id=%s state=%s job_id=%s. "
+ "ti=%s state=%s job_id=%s. "
"Another scheduler may have already modified this TI.",
ti.try_number,
try_number,
ti,
- ti.id,
ti.state,
job_id,
)
@@ -1267,7 +1265,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
continue
msg = (
- "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s,
map_index=%s, "
+ "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s,
map_index=%s, ti_id=%s, "
"run_start_date=%s, run_end_date=%s, "
"run_duration=%s, state=%s, executor=%s, executor_state=%s,
try_number=%s, max_tries=%s, "
"pool=%s, queue=%s, priority_weight=%d, operator=%s,
queued_dttm=%s, scheduled_dttm=%s,"
@@ -1279,6 +1277,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
ti.task_id,
ti.run_id,
ti.map_index,
+ ti.id,
ti.start_date,
ti.end_date,
ti.duration,
@@ -1387,7 +1386,8 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
# Handle cleared tasks that were successfully terminated by
executor
if ti.state == TaskInstanceState.RESTARTING and state ==
TaskInstanceState.SUCCESS:
cls.logger().info(
- "Task %s was cleared and successfully terminated.
Setting to scheduled for retry.", ti
+ "Task %s was cleared and successfully terminated.
Setting to scheduled for retry.",
+ ti,
)
# Adjust max_tries to allow retry beyond normal limits
(like clearing does)
ti.max_tries = ti.try_number + ti.task.retries
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index f44037dc3e6..74b83a3851a 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -1305,7 +1305,16 @@ class TriggerRunner:
await greenback.ensure_portal()
- bind_log_contextvars(trigger_id=trigger_id)
+ ti = trigger.task_instance
+ bind_log_contextvars(
+ trigger_id=trigger_id,
+ ti_id=str(ti.id) if ti else None,
+ dag_id=ti.dag_id if ti else None,
+ task_id=ti.task_id if ti else None,
+ run_id=ti.run_id if ti else None,
+ try_number=ti.try_number if ti else None,
+ map_index=ti.map_index if ti else None,
+ )
name = self.triggers[trigger_id]["name"]
self.log.info("trigger %s starting", name)
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 6fb2b38696c..8dba8a17188 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -1097,7 +1097,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
prefix = f"<TaskInstance: {self.dag_id}.{self.task_id} {self.run_id} "
if self.map_index != -1:
prefix += f"map_index={self.map_index} "
- return prefix + f"[{self.state}]>"
+ return prefix + f"[{self.state}] ti_id={self.id}>"
def next_retry_datetime(self):
"""
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index b0f72b91db3..84694b47c69 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -692,7 +692,7 @@ class TestSchedulerJob:
bundle_name="dag_maker",
bundle_version=None,
msg=f"Executor {executor} reported that the task instance "
- "<TaskInstance:
test_process_executor_events_with_callback.dummy_task test [queued]> "
+ f"<TaskInstance:
test_process_executor_events_with_callback.dummy_task test [queued]
ti_id={ti1.id}> "
"finished with state failed, but the task instance's state
attribute is queued. "
"Learn more:
https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally";,
context_from_server=mock.ANY,
@@ -1904,11 +1904,13 @@ class TestSchedulerJob:
some_pool = Pool(pool="some_pool", slots=2, description="my pool",
include_deferred=False)
session.add(some_pool)
session.commit()
+ cannot_run_ti_id = next(t for t in dr.task_instances if t.task_id ==
"cannot_run").id
with caplog.at_level(logging.WARNING):
self.job_runner._executable_task_instances_to_queued(max_tis=32,
session=session)
assert (
- "Not executing <TaskInstance: "
- "SchedulerJobTest.test_test_not_enough_pool_slots.cannot_run
test [scheduled]>. "
+ f"Not executing <TaskInstance: "
+ f"SchedulerJobTest.test_test_not_enough_pool_slots.cannot_run
test [scheduled] "
+ f"ti_id={cannot_run_ti_id}>. "
"Requested pool slots (4) are greater than total pool slots:
'2' for pool: some_pool"
in caplog.text
)
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 73e8fd0cbf6..a4d8f6dbbd9 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -38,6 +38,7 @@ import structlog
from opentelemetry import trace
from opentelemetry.trace.propagation.tracecontext import
TraceContextTextMapPropagator
from pydantic import AwareDatetime, ConfigDict, Field, JsonValue, TypeAdapter
+from structlog.contextvars import bind_contextvars
from airflow.dag_processing.bundles.base import BaseDagBundle,
BundleVersionLock
from airflow.dag_processing.bundles.manager import DagBundlesManager
@@ -924,6 +925,17 @@ def startup(msg: StartupDetails) ->
tuple[RuntimeTaskInstance, Context, Logger]:
setproctitle(f"airflow worker -- {msg.ti.id}")
+ # Bind TI identifiers so every subsequent log line from this worker process
+ # carries ti_id, enabling single-grep lifecycle reconstruction across
components.
+ bind_contextvars(
+ ti_id=str(msg.ti.id),
+ dag_id=msg.ti.dag_id,
+ task_id=msg.ti.task_id,
+ run_id=msg.ti.run_id,
+ try_number=msg.ti.try_number,
+ map_index=msg.ti.map_index,
+ )
+
try:
get_listener_manager().hook.on_starting(component=TaskRunnerMarker())
except Exception:
@@ -1170,8 +1182,6 @@ def _validate_task_inlets_and_outlets(*, ti:
RuntimeTaskInstance, log: Logger) -
def _defer_task(
defer: TaskDeferred, ti: RuntimeTaskInstance, log: Logger
) -> tuple[ToSupervisor, TaskInstanceState]:
- # TODO: Should we use structlog.bind_contextvars here for dag_id, task_id
& run_id?
-
log.info("Pausing task as DEFERRED. ", dag_id=ti.dag_id,
task_id=ti.task_id, run_id=ti.run_id)
classpath, trigger_kwargs = defer.trigger.serialize()
queue: str | None = None
