Add temporalActivityAttempt attribute to RunActivity spans

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.

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:

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)

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(

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