From e7b43efd6dd0fa4f6cc9dac888323df93cc87da0 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 11 Sep 2026 00:21:30 -0500 Subject: [PATCH] Add temporalActivityAttempt attribute to RunActivity spans --- CHANGELOG.md | 6 ++ .../contrib/opentelemetry/_interceptor.py | 7 +- .../opentelemetry/_otel_interceptor.py | 14 ++-- .../opentelemetry/test_opentelemetry.py | 9 +++ .../test_opentelemetry_plugin.py | 66 +++++++++++++++++++ 5 files changed, 96 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 14a225277..486e7706f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,12 @@ to include examples, links to docs, or any other relevant information. ### Added +- `temporalio.contrib.opentelemetry`: `RunActivity` spans now carry a + `temporalActivityAttempt` attribute with the Activity attempt number, in both + `OpenTelemetryPlugin` and `TracingInterceptor`. The SDK already creates one + `RunActivity` span per attempt; the attribute lets tracing backends tell + retries apart. + - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`. - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome. - New properties and methods in ActivityExecution and ActivityExecutionDescription. diff --git a/temporalio/contrib/opentelemetry/_interceptor.py b/temporalio/contrib/opentelemetry/_interceptor.py index 6dca4596e..2989c2d83 100644 --- a/temporalio/contrib/opentelemetry/_interceptor.py +++ b/temporalio/contrib/opentelemetry/_interceptor.py @@ -370,7 +370,12 @@ async def execute_activity( self, input: temporalio.worker.ExecuteActivityInput ) -> Any: info = temporalio.activity.info() - attributes: dict[str, str] = {"temporalActivityID": info.activity_id} + attributes: dict[str, opentelemetry.util.types.AttributeValue] = { + "temporalActivityID": info.activity_id, + # One RunActivity span is created per attempt; the attempt number + # lets tracing backends tell retries apart. + "temporalActivityAttempt": info.attempt, + } if info.workflow_id: attributes["temporalWorkflowID"] = info.workflow_id if info.workflow_run_id: diff --git a/temporalio/contrib/opentelemetry/_otel_interceptor.py b/temporalio/contrib/opentelemetry/_otel_interceptor.py index ff07f0f1d..928c611b4 100644 --- a/temporalio/contrib/opentelemetry/_otel_interceptor.py +++ b/temporalio/contrib/opentelemetry/_otel_interceptor.py @@ -337,15 +337,19 @@ async def execute_activity( token = opentelemetry.context.attach(context) try: info = temporalio.activity.info() + attributes: dict[str, opentelemetry.util.types.AttributeValue] = { + "temporalWorkflowID": info.workflow_id or "", + "temporalRunID": info.workflow_run_id or "", + "temporalActivityID": info.activity_id, + # One RunActivity span is created per attempt; the attempt + # number lets tracing backends tell retries apart. + "temporalActivityAttempt": info.attempt, + } with _maybe_span( get_tracer(__name__), f"RunActivity:{info.activity_type}", add_temporal_spans=self._add_temporal_spans, - attributes={ - "temporalWorkflowID": info.workflow_id or "", - "temporalRunID": info.workflow_run_id or "", - "temporalActivityID": info.activity_id, - }, + attributes=attributes, kind=opentelemetry.trace.SpanKind.SERVER, ): return await super().execute_activity(input) diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 0ea9530e6..58e485ae6 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -425,6 +425,15 @@ async def test_opentelemetry_tracing(client: Client, env: WorkflowEnvironment): "SignalWorkflow:signal", "StartWorkflowUpdate:update", ] + # Each RunActivity span records its attempt number; the first activity + # failed once, so its two spans are attempts 1 and 2. + attempts = [ + (span.attributes or {})["temporalActivityAttempt"] + for span in exporter.get_finished_spans() + if span.name == "RunActivity:tracing_activity" + ] + assert attempts[:2] == [1, 2] + assert all(attempt == 1 for attempt in attempts[2:]) async def test_opentelemetry_tracing_update_with_start( diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 022645fee..3dad42570 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -14,12 +14,14 @@ ) from opentelemetry.sdk.trace.id_generator import RandomIdGenerator from opentelemetry.trace import ( + StatusCode, get_tracer, ) import temporalio.contrib.opentelemetry.workflow from temporalio import activity, nexus, workflow from temporalio.client import Client, WorkflowFailureError +from temporalio.common import RetryPolicy from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider from temporalio.contrib.opentelemetry._id_generator import TemporalIdGenerator from temporalio.exceptions import ApplicationError @@ -549,6 +551,70 @@ async def test_otel_tracing_with_added_spans( ) +@activity.defn +async def fail_first_attempt_activity() -> str: + if activity.info().attempt == 1: + raise ApplicationError("intentional failure on attempt 1") + return "done" + + +@workflow.defn +class RetryingActivityWorkflow: + @workflow.run + async def run(self) -> str: + return await workflow.execute_activity( + fail_first_attempt_activity, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=10), maximum_attempts=2 + ), + ) + + +async def test_otel_tracing_activity_attempts( + client: Client, + reset_otel_tracer_provider: Any, # type: ignore[reportUnusedParameter] +): + exporter = InMemorySpanExporter() + provider = create_tracer_provider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + opentelemetry.trace.set_tracer_provider(provider) + + new_config = client.config() + new_config["plugins"] = [OpenTelemetryPlugin(add_temporal_spans=True)] + new_client = Client(**new_config) + + async with new_worker( + new_client, + RetryingActivityWorkflow, + activities=[fail_first_attempt_activity], + max_cached_workflows=0, + ) as worker: + with get_tracer(__name__).start_as_current_span("Retry test"): + await new_client.execute_workflow( + RetryingActivityWorkflow.run, + id=f"retry-workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + + spans = exporter.get_finished_spans() + assert dump_spans(spans, with_attributes=False) == [ + "Retry test", + " StartWorkflow:RetryingActivityWorkflow", + " RunWorkflow:RetryingActivityWorkflow", + " StartActivity:fail_first_attempt_activity", + " RunActivity:fail_first_attempt_activity", + " RunActivity:fail_first_attempt_activity", + ] + # One RunActivity span per attempt, each carrying its attempt number. + attempts = [s for s in spans if s.name == "RunActivity:fail_first_attempt_activity"] + assert [(s.attributes or {})["temporalActivityAttempt"] for s in attempts] == [1, 2] + assert attempts[0].status.status_code == StatusCode.ERROR + assert attempts[1].status.status_code != StatusCode.ERROR + assert len([s for s in spans if s.name.startswith("StartActivity:")]) == 1 + + task_fail_once_workflow_has_failed = False