You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dstandish (via GitHub)" <gi...@apache.org> on 2026/05/28 21:07:40 UTC
[PR] Emit OpenTelemetry spans around task callback invocations [airflow]
dstandish opened a new pull request, #67659:
URL: https://github.com/apache/airflow/pull/67659
Follow-up to #67347 (which added `listener.<hook_name>` spans). This wraps each task callback and pre/post-execute hook invocation in a `callback.<kind>` span, gated on task span detail level > 1.
Covered invocations (all routed through `create_executable_runner(...).run(...)` in `task-sdk/.../task_runner.py`):
- the five state-change callbacks: `on_execute_callback`, `on_success_callback`, `on_failure_callback`, `on_retry_callback`, `on_skipped_callback` (centralized in `_run_task_state_change_callbacks`)
- the `pre_execute` / `post_execute` hooks in `_execute_task`
The callback function name and index are recorded as span attributes; a raising state-change callback records `ERROR` status on its span and is still swallowed/logged as before.
Note: the span is added at the genuine call sites in `task_runner.py`, **not** pushed down into `create_executable_runner`/`callback_runner.py`, because that runner is also reused to execute the actual task body (`python_callable`) in the standard `PythonOperator` — wrapping there would mislabel the main task execution.
Tests: new `TestCallbackSpans` unit tests in `test_task_runner.py` (span presence/attributes at level 2, none at level 1, error recording for a raising callback, pre/post-execute gating) and the otel integration test now asserts `callback.on_execute_callback` → `_execute_task` and `callback.on_success_callback` → `finalize`.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 4.7)
Generated-by: Claude Code (Opus 4.7) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #67659:
URL: https://github.com/apache/airflow/pull/67659#discussion_r3639152552
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -5813,6 +5814,155 @@ def test_no_task_execute_span_at_detail_level_1(self):
assert "task.execute" not in names
+class TestCallbackSpans:
+ """Tests for spans emitted around task callback / pre-post-execute invocations."""
+
+ @pytest.fixture(autouse=True)
+ def _sampled_carrier_provider(self):
+ """Make new_dagrun_trace_carrier produce a SAMPLED carrier (see TestDetailSpan)."""
+ provider = TracerProvider()
+ with mock.patch(
+ "airflow._shared.observability.traces.trace.get_tracer_provider",
+ return_value=provider,
+ ):
+ yield
Review Comment:
i consolidated the fixture
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #67659:
URL: https://github.com/apache/airflow/pull/67659#discussion_r3635658581
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -5813,6 +5814,155 @@ def test_no_task_execute_span_at_detail_level_1(self):
assert "task.execute" not in names
+class TestCallbackSpans:
+ """Tests for spans emitted around task callback / pre-post-execute invocations."""
+
+ @pytest.fixture(autouse=True)
+ def _sampled_carrier_provider(self):
+ """Make new_dagrun_trace_carrier produce a SAMPLED carrier (see TestDetailSpan)."""
+ provider = TracerProvider()
+ with mock.patch(
+ "airflow._shared.observability.traces.trace.get_tracer_provider",
+ return_value=provider,
+ ):
+ yield
Review Comment:
Seems to be a repetition / copy of `_sampled_carrier_provider` in `TestDetailSpan` (docstring says so too!). BTW, I think we should re-export `get_tracer_provider` from `airflow._shared.observability.traces.trace.get_tracer_provider` into `task-sdk/src/airflow/sdk/observability` and use instead of mocking from an internal shared module.
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2131,10 +2140,17 @@ def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
outlet_events = context_get_outlet_events(context)
+ def _run_hook(span_kind: str, hook, *run_args):
+ with detail_span(
+ f"callback.{span_kind}",
+ attributes={"airflow.callback.name": getattr(hook, "__name__", repr(hook))},
+ ):
+ create_executable_runner(hook, outlet_events, logger=log).run(*run_args)
+
if (pre_execute_hook := task._pre_execute_hook) is not None:
- create_executable_runner(pre_execute_hook, outlet_events, logger=log).run(context)
+ _run_hook("pre_execute", pre_execute_hook, context)
if getattr(pre_execute_hook := task.pre_execute, "__func__", None) is not BaseOperator.pre_execute:
- create_executable_runner(pre_execute_hook, outlet_events, logger=log).run(context)
+ _run_hook("pre_execute", pre_execute_hook, context)
Review Comment:
There are two independent checks here: `task._pre_execute_hook` (instance level hook) and `task.pre_execute` (a method override) and both if present get wrapped in a span named identically: "callback.pre_execute". Same for the two post_execute variants. If both fire on the same task we would get two sibling spans with the exact same name under `_execute_task`, it's a legitimate observation if you want it FYI.
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2131,10 +2140,17 @@ def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
outlet_events = context_get_outlet_events(context)
+ def _run_hook(span_kind: str, hook, *run_args):
Review Comment:
Typing for hook and / or run_args pls?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #67659:
URL: https://github.com/apache/airflow/pull/67659#discussion_r3639544439
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2131,10 +2140,17 @@ def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
outlet_events = context_get_outlet_events(context)
+ def _run_hook(span_kind: str, hook, *run_args):
Review Comment:
fixed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #67659:
URL: https://github.com/apache/airflow/pull/67659#discussion_r3639097956
##########
task-sdk/tests/task_sdk/execution_time/test_task_runner.py:
##########
@@ -5813,6 +5814,155 @@ def test_no_task_execute_span_at_detail_level_1(self):
assert "task.execute" not in names
+class TestCallbackSpans:
+ """Tests for spans emitted around task callback / pre-post-execute invocations."""
+
+ @pytest.fixture(autouse=True)
+ def _sampled_carrier_provider(self):
+ """Make new_dagrun_trace_carrier produce a SAMPLED carrier (see TestDetailSpan)."""
+ provider = TracerProvider()
+ with mock.patch(
+ "airflow._shared.observability.traces.trace.get_tracer_provider",
+ return_value=provider,
+ ):
+ yield
Review Comment:
the point of it being in _shared is so that it does not become public interface
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #67659:
URL: https://github.com/apache/airflow/pull/67659#discussion_r3639543347
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2131,10 +2140,17 @@ def _execute_task(context: Context, ti: RuntimeTaskInstance, log: Logger):
outlet_events = context_get_outlet_events(context)
+ def _run_hook(span_kind: str, hook, *run_args):
+ with detail_span(
+ f"callback.{span_kind}",
+ attributes={"airflow.callback.name": getattr(hook, "__name__", repr(hook))},
+ ):
+ create_executable_runner(hook, outlet_events, logger=log).run(*run_args)
+
if (pre_execute_hook := task._pre_execute_hook) is not None:
- create_executable_runner(pre_execute_hook, outlet_events, logger=log).run(context)
+ _run_hook("pre_execute", pre_execute_hook, context)
if getattr(pre_execute_hook := task.pre_execute, "__func__", None) is not BaseOperator.pre_execute:
- create_executable_runner(pre_execute_hook, outlet_events, logger=log).run(context)
+ _run_hook("pre_execute", pre_execute_hook, context)
Review Comment:
fixed
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Emit OpenTelemetry spans around task callback invocations [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #67659:
URL: https://github.com/apache/airflow/pull/67659#issuecomment-4986688643
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org