From cba419628b8e06eb21fbb768c6e8800e1840e31b Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Mon, 29 Jun 2026 15:20:16 -0700 Subject: [PATCH 1/2] Mark task worker span as failed when the task fails run() converts every task failure into a returned state plus error rather than re-raising, so the worker. OTel span never observed the failure through exception propagation and was left with an unset status. Derive the span status from run()'s result so a failed task shows as an error in traces, consistently and regardless of trace detail level. --- .../airflow/sdk/execution_time/task_runner.py | 9 +++ .../execution_time/test_task_runner.py | 76 +++++++++++++++++++ 2 files changed, 85 insertions(+) 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 b6616dc30cfd7..94363b984b7f6 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -2343,6 +2343,15 @@ def main(): ): state, _, error = run(ti, context, log) context["exception"] = error + # run() funnels every failure path into `error` rather than + # re-raising, so the worker span never sees the exception via + # propagation. Mark it here, where the worker span is in scope + # regardless of trace detail level. + if error is not None: + span.record_exception(error) + span.set_status( + Status(StatusCode.ERROR, description=f"Exception: {type(error).__name__}") + ) finalize(ti, state, context, log, error) # If run() couldn't deliver a FAILED / UP_FOR_RETRY terminal # state to the supervisor, fail closed now — finalize() has diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py index 1aad1a02d5eab..fadcbb4e8ce15 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py +++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py @@ -465,6 +465,82 @@ def test_main_sends_reschedule_task_when_startup_reschedules( ] +@pytest.mark.parametrize( + ("state", "error", "expect_error_status"), + [ + (TaskInstanceState.FAILED, RuntimeError("boom"), True), + (TaskInstanceState.SUCCESS, None, False), + ], +) +@mock.patch("airflow.sdk.execution_time.task_runner.finalize") +@mock.patch("airflow.sdk.execution_time.task_runner.run") +@mock.patch("airflow.sdk.execution_time.task_runner.BundleVersionLock") +@mock.patch("airflow.sdk.execution_time.task_runner.startup") +@mock.patch("airflow.sdk.execution_time.task_runner.get_startup_details") +@mock.patch("airflow.sdk.execution_time.task_runner.CommsDecoder") +def test_main_marks_worker_span_error_on_failure( + mock_comms_decoder_cls, + mock_get_startup_details, + mock_startup, + mock_bundle_lock, + mock_run, + mock_finalize, + make_ti_context, + state, + error, + expect_error_status, +): + """main() marks the worker span ERROR when run() reports a failure, and leaves it unset otherwise. + + run() funnels every failure into its returned ``error`` rather than re-raising, so the worker + span never sees the exception via propagation; main() must set the status from that return value. + """ + mock_comms_instance = mock.Mock() + mock_comms_instance.socket = None + mock_comms_decoder_cls.__getitem__.return_value.return_value = mock_comms_instance + + what = StartupDetails( + ti=TaskInstance( + id=uuid7(), + task_id="my_task", + dag_id="test_dag", + run_id="test_run", + try_number=1, + dag_version_id=uuid7(), + queue="default", + context_carrier={}, + ), + dag_rel_path="", + bundle_info=BundleInfo(name="my-bundle", version=None), + ti_context=make_ti_context(), + start_date=timezone.utcnow(), + sentry_integration="", + ) + mock_get_startup_details.return_value = what + + ti = mock.Mock() + ti.bundle_instance.name = "my-bundle" + ti.bundle_instance.version = None + ti._terminal_state_send_failed = False + mock_startup.return_value = (ti, {}, mock.Mock()) + mock_run.return_value = (state, mock.Mock(), error) + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + t = provider.get_tracer("test") + + with mock.patch("airflow.sdk.execution_time.task_runner.tracer", t): + task_runner.main() + + worker = {s.name: s for s in exporter.get_finished_spans()}["worker.my_task"] + if expect_error_status: + assert worker.status.status_code == trace.StatusCode.ERROR + assert any(e.name == "exception" for e in worker.events) + else: + assert worker.status.status_code != trace.StatusCode.ERROR + + def test_run_swallows_supervisor_terminal_send_failure(create_runtime_ti, mock_supervisor_comms): """ When the supervisor rejects the terminal-state send (e.g. the server From 0229bf69d2adc4308846ee5e8f03ed6ab0e590e7 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:59:18 -0700 Subject: [PATCH 2/2] Lower sleep duration in execution timeout test If the SIGALRM-based kill ever fails to fire, the test would otherwise block for the full sleep duration before failing with a not-raised error; 200ms bounds that worst case while staying well above the 10ms timeout. --- task-sdk/tests/task_sdk/execution_time/test_task_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py index fadcbb4e8ce15..9d15ac40c6967 100644 --- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py +++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py @@ -5728,7 +5728,7 @@ def test_applies_execution_timeout(self): task = self._make_task(execution_timeout=timedelta(milliseconds=10)) def execute(context): - time.sleep(2) + time.sleep(0.2) with pytest.raises(AirflowTaskTimeout): _run_execute_callable(context={}, execute=execute, task=task)