Add OTel tracing for standalone activities (#1471)

* feat(otel): add otel tracing for standalone activities for both legacy and new interceptors.

* docs: update README section on testing to reflect current implementation.

* style: remove unnecessary comments in tests.

Author: cconstable

.gitignore

@@ -2,6 +2,7 @@
 __pycache__
 /build
 /dist
+temporalio/bridge/libtemporal_sdk_bridge.dylib.dSYM/
 temporalio/bridge/target/
 temporalio/bridge/temporal_sdk_bridge*
 /tests/helpers/golangserver/golangserver

README.md

@@ -2059,28 +2059,33 @@ The environment is now ready to develop in.
 
 #### Testing
 
-To execute tests:
+To execute tests (in parallel if possible):
 
 ```bash
 poe test
 ```
 
-`poe test` spreads tests across multiple worker processes by default. If you
-need a serial run for debugging, invoke pytest directly:
+To execute tests serially:
 
 ```bash
 uv run pytest
 ```
 
-This runs against [Temporalite](https://github.com/temporalio/temporalite). To run against the time-skipping test
-server, pass `--workflow-environment time-skipping`. To run against the `default` namespace of an already-running
-server, pass the `host:port` to `--workflow-environment`. Can also use regular pytest arguments. For example, here's how
-to run a single test with debug logs on the console:
+To execute a single test:
 
 ```bash
 poe test -s --log-cli-level=DEBUG -k test_sync_activity_thread_cancel_caught
 ```
 
+**Temporal Server**
+
+- Tests that use the workflow test environment run against the [Temporal CLI dev server](https://docs.temporal.io/cli#start-dev-server).
+- By default, workflow-environment tests automatically start a local dev server.
+- On first run, the dev server binary may be downloaded so network access is required if no server is currently running.
+- To run workflow-environment tests against the time-skipping test server, pass `--workflow-environment time-skipping`. 
+- To run workflow-environment tests against the `default` namespace of an already-running server, pass the `host:port` to `--workflow-environment`.
+- Unit tests that do not use the workflow environment do not start a dev server.
+
 #### Proto Generation and Testing
 
 If you have docker available, run

temporalio/contrib/opentelemetry/_interceptor.py

@@ -341,6 +341,60 @@ async def start_update_with_start_workflow(
 
             return await super().start_update_with_start_workflow(input)
 
+    async def start_activity(
+        self, input: temporalio.client.StartActivityInput
+    ) -> temporalio.client.ActivityHandle[Any]:
+        with self.root._start_as_current_span(
+            f"StartActivity:{input.activity_type}",
+            attributes={
+                "temporalActivityID": input.id,
+                "temporalActivityType": input.activity_type,
+            },
+            input_with_headers=input,
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().start_activity(input)
+
+    async def cancel_activity(
+        self, input: temporalio.client.CancelActivityInput
+    ) -> None:
+        with self.root._start_as_current_span(
+            "CancelActivity",
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().cancel_activity(input)
+
+    async def terminate_activity(
+        self, input: temporalio.client.TerminateActivityInput
+    ) -> None:
+        with self.root._start_as_current_span(
+            "TerminateActivity",
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().terminate_activity(input)
+
+    async def describe_activity(
+        self, input: temporalio.client.DescribeActivityInput
+    ) -> temporalio.client.ActivityExecutionDescription:
+        with self.root._start_as_current_span(
+            "DescribeActivity",
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().describe_activity(input)
+
+    async def count_activities(
+        self, input: temporalio.client.CountActivitiesInput
+    ) -> temporalio.client.ActivityExecutionCount:
+        with self.root._start_as_current_span(
+            "CountActivities",
+            attributes={},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().count_activities(input)
+
 
 class _TracingActivityInboundInterceptor(temporalio.worker.ActivityInboundInterceptor):
     def __init__(

temporalio/contrib/opentelemetry/_otel_interceptor.py

@@ -303,6 +303,70 @@ async def start_update_with_start_workflow(
             )
             return await super().start_update_with_start_workflow(input)
 
+    async def start_activity(
+        self, input: temporalio.client.StartActivityInput
+    ) -> temporalio.client.ActivityHandle[Any]:
+        with _maybe_span(
+            get_tracer(__name__),
+            f"StartActivity:{input.activity_type}",
+            add_temporal_spans=self._add_temporal_spans,
+            attributes={
+                "temporalActivityID": input.id,
+                "temporalActivityType": input.activity_type,
+            },
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            input.headers = _context_to_headers(input.headers)
+            return await super().start_activity(input)
+
+    async def cancel_activity(
+        self, input: temporalio.client.CancelActivityInput
+    ) -> None:
+        with _maybe_span(
+            get_tracer(__name__),
+            "CancelActivity",
+            add_temporal_spans=self._add_temporal_spans,
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().cancel_activity(input)
+
+    async def terminate_activity(
+        self, input: temporalio.client.TerminateActivityInput
+    ) -> None:
+        with _maybe_span(
+            get_tracer(__name__),
+            "TerminateActivity",
+            add_temporal_spans=self._add_temporal_spans,
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().terminate_activity(input)
+
+    async def describe_activity(
+        self, input: temporalio.client.DescribeActivityInput
+    ) -> temporalio.client.ActivityExecutionDescription:
+        with _maybe_span(
+            get_tracer(__name__),
+            "DescribeActivity",
+            add_temporal_spans=self._add_temporal_spans,
+            attributes={"temporalActivityID": input.activity_id},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().describe_activity(input)
+
+    async def count_activities(
+        self, input: temporalio.client.CountActivitiesInput
+    ) -> temporalio.client.ActivityExecutionCount:
+        with _maybe_span(
+            get_tracer(__name__),
+            "CountActivities",
+            add_temporal_spans=self._add_temporal_spans,
+            attributes={},
+            kind=opentelemetry.trace.SpanKind.CLIENT,
+        ):
+            return await super().count_activities(input)
+
 
 class _TracingActivityInboundInterceptor(temporalio.worker.ActivityInboundInterceptor):
     def __init__(

tests/contrib/opentelemetry/test_opentelemetry.py

@@ -944,6 +944,70 @@ async def test_opentelemetry_interceptor_works_if_no_context(
 # * signal failure and wft failure from signal
 
 
+async def test_opentelemetry_standalone_activity_tracing(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/2741"
+        )
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    tracer = get_tracer(__name__, tracer_provider=provider)
+    client_config = client.config()
+    client_config["interceptors"] = [TracingInterceptor(tracer)]
+    client = Client(**client_config)
+
+    task_queue = f"task_queue_{uuid.uuid4()}"
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        activities=[tracing_activity],
+    ):
+        handle = await client.start_activity(
+            tracing_activity,
+            TracingActivityParam(heartbeat=False),
+            id=f"activity_{uuid.uuid4()}",
+            task_queue=task_queue,
+            schedule_to_close_timeout=timedelta(seconds=10),
+        )
+        await handle.result()
+
+    # Use a queue with no worker so activities stay in SCHEDULED state,
+    # allowing describe/cancel/terminate to be called without a race.
+    no_worker_queue = f"task_queue_{uuid.uuid4()}"
+
+    cancel_handle = await client.start_activity(
+        tracing_activity,
+        TracingActivityParam(heartbeat=False),
+        id=f"activity_{uuid.uuid4()}",
+        task_queue=no_worker_queue,
+        schedule_to_close_timeout=timedelta(seconds=30),
+    )
+    await cancel_handle.describe()
+    await cancel_handle.cancel()
+
+    terminate_handle = await client.start_activity(
+        tracing_activity,
+        TracingActivityParam(heartbeat=False),
+        id=f"activity_{uuid.uuid4()}",
+        task_queue=no_worker_queue,
+        schedule_to_close_timeout=timedelta(seconds=30),
+    )
+    await terminate_handle.terminate()
+
+    assert dump_spans(exporter.get_finished_spans(), with_attributes=False) == [
+        "StartActivity:tracing_activity",
+        "  RunActivity:tracing_activity",
+        "StartActivity:tracing_activity",
+        "DescribeActivity",
+        "CancelActivity",
+        "StartActivity:tracing_activity",
+        "TerminateActivity",
+    ]
+
+
 def test_opentelemetry_safe_detach():
     class _fake_self:
         def _load_workflow_context_carrier(*_args):

tests/contrib/opentelemetry/test_opentelemetry_plugin.py

@@ -567,6 +567,69 @@ async def test_otel_tracing_workflow_failure(
     ), f"Span hierarchy mismatch.\nExpected:\n{expected_hierarchy}\nActual:\n{actual_hierarchy}"
 
 
+async def test_otel_standalone_activity_tracing(
+    client: Client,
+    env: WorkflowEnvironment,
+    reset_otel_tracer_provider: Any,  # type: ignore[reportUnusedParameter]
+):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Java test server: https://github.com/temporalio/sdk-java/issues/2741"
+        )
+    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,
+        activities=[simple_no_context_activity],
+    ) as worker:
+        handle = await new_client.start_activity(
+            simple_no_context_activity,
+            id=f"activity_{uuid.uuid4()}",
+            task_queue=worker.task_queue,
+            schedule_to_close_timeout=timedelta(seconds=10),
+        )
+        await handle.result()
+
+    # Use a queue with no worker so activities stay in SCHEDULED state,
+    # allowing describe/cancel/terminate to be called without a race.
+    no_worker_queue = f"task_queue_{uuid.uuid4()}"
+
+    cancel_handle = await new_client.start_activity(
+        simple_no_context_activity,
+        id=f"activity_{uuid.uuid4()}",
+        task_queue=no_worker_queue,
+        schedule_to_close_timeout=timedelta(seconds=30),
+    )
+    await cancel_handle.describe()
+    await cancel_handle.cancel()
+
+    terminate_handle = await new_client.start_activity(
+        simple_no_context_activity,
+        id=f"activity_{uuid.uuid4()}",
+        task_queue=no_worker_queue,
+        schedule_to_close_timeout=timedelta(seconds=30),
+    )
+    await terminate_handle.terminate()
+
+    assert dump_spans(exporter.get_finished_spans(), with_attributes=False) == [
+        "StartActivity:simple_no_context_activity",
+        "  RunActivity:simple_no_context_activity",
+        "    Activity",
+        "StartActivity:simple_no_context_activity",
+        "DescribeActivity",
+        "CancelActivity",
+        "StartActivity:simple_no_context_activity",
+        "TerminateActivity",
+    ]
+
+
 def test_replay_safe_span_delegates_extra_attributes():
     """Test that _ReplaySafeSpan delegates attribute access to the underlying span.