OpenTelemetry sample

README.md

@@ -52,11 +52,12 @@ Some examples require extra dependencies. See each sample's directory for specif
   * [hello_search_attributes](hello/hello_search_attributes.py) - Start workflow with search attributes then change
     while running.
   * [hello_signal](hello/hello_signal.py) - Send signals to a workflow.
+<!-- Keep this list in alphabetical order -->
 * [activity_worker](activity_worker) - Use Python activities from a workflow in another language.
 * [custom_converter](custom_converter) - Use a custom payload converter to handle custom types.
 * [custom_decorator](custom_decorator) - Custom decorator to auto-heartbeat a long-running activity.
 * [encryption](encryption) - Apply end-to-end encryption for all input/output.
-
+* [open_telemetry](open_telemetry) - Trace workflows with OpenTelemetry.
 
 ## Test
 
@@ -66,4 +67,4 @@ Running the tests requires `poe` to be installed.
 
 Once you have `poe` installed you can run:
 
-    poe test
+    poe test

open_telemetry/README.md

@@ -0,0 +1,60 @@
+# OpenTelemetry Sample
+
+This sample shows how to configure OpenTelemetry to capture workflow traces and SDK metrics.
+
+For this sample, the optional `open_telemetry` dependency group must be included. To include, run:
+
+    poetry install --with open_telemetry
+
+To run, first see [README.md](../README.md) for prerequisites. Then run the following to start a Jaeger container to
+view the trace results:
+
+    docker run -d --name jaeger \
+      -p 16686:16686 \
+      -p 6831:6831/udp \
+      jaegertracing/all-in-one:latest
+
+Since that is running in the background (`-d`), you can also run the metrics collector in the foreground:
+
+    docker run -p 4317:4317 \
+      -v /path/to/samples-python/open_telemetry/otel-metrics-collector-config.yaml:/etc/otel-collector-config.yaml \
+      otel/opentelemetry-collector:latest \
+      --config=/etc/otel-collector-config.yaml
+
+Replace `/path/to/samples-python` with the absolute path to the cloned samples repo.
+
+Now, from this directory, start the worker in its own terminal:
+
+    poetry run python worker.py
+
+This will start the worker. Then, in another terminal, run the following to execute the workflow:
+
+    poetry run python starter.py
+
+The workflow should complete with the hello result. The workflow trace can now be viewed in Jaeger at
+http://localhost:16686/. Under service, select `my-service` and "Find Traces". The workflow should appear and when
+clicked, may look something like:
+
+![Jaeger Screenshot](jaeger-screenshot.png)
+
+Note, in-workflow spans do not have a time associated with them. This is by intention since due to replay. In
+OpenTelemetry, only the process that started the span may end it. But in Temporal a span may cross workers/processes.
+Therefore we intentionally start-then-end in-workflow spans immediately. So while the start time and hierarchy is
+accurate, the duration is not.
+
+The metrics should have been dumped out in the terminal where the OpenTelemetry collector container is running.
+
+## OTLP gRPC
+
+Currently for tracing this example uses the `opentelemetry-exporter-jaeger-thrift` exporter because the common OTLP gRPC
+exporter `opentelemetry-exporter-otlp-proto-grpc` uses an older, incompatible `protobuf` library. See
+[this issue](https://github.com/open-telemetry/opentelemetry-python/issues/2880) for more information.
+
+Once OTel supports latest protobuf, the exporter can be changed and Jaeger could be run with:
+
+    docker run -d --name jaeger \
+      -e COLLECTOR_OTLP_ENABLED=true \
+      -p 16686:16686 \
+      -p 4317:4317 \
+      -p 4318:4318 \
+      jaegertracing/all-in-one:latest

open_telemetry/otel-metrics-collector-config.yaml

@@ -0,0 +1,15 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+exporters:
+  logging:
+    loglevel: debug
+processors:
+  batch:
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      exporters: [logging]
+      processors: [batch]

open_telemetry/starter.py

@@ -0,0 +1,30 @@
+import asyncio
+
+from temporalio.client import Client
+from temporalio.contrib.opentelemetry import TracingInterceptor
+
+from open_telemetry.worker import GreetingWorkflow, init_opentelemetry
+
+
+async def main():
+    init_opentelemetry()
+
+    # Connect client
+    client = await Client.connect(
+        "localhost:7233",
+        # Use OpenTelemetry interceptor
+        interceptors=[TracingInterceptor()],
+    )
+
+    # Run workflow
+    result = await client.execute_workflow(
+        GreetingWorkflow.run,
+        "Temporal",
+        id=f"open_telemetry-workflow-id",
+        task_queue="open_telemetry-task-queue",
+    )
+    print(f"Workflow result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

open_telemetry/worker.py

@@ -0,0 +1,82 @@
+import asyncio
+from datetime import timedelta
+
+from opentelemetry import trace
+
+# See note in README about why Thrift
+from opentelemetry.exporter.jaeger.thrift import JaegerExporter
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from temporalio import activity, workflow
+from temporalio.bridge import telemetry
+from temporalio.client import Client
+from temporalio.contrib.opentelemetry import TracingInterceptor
+from temporalio.worker import Worker
+
+
+@workflow.defn
+class GreetingWorkflow:
+    @workflow.run
+    async def run(self, name: str) -> str:
+        return await workflow.execute_activity(
+            compose_greeting,
+            name,
+            start_to_close_timeout=timedelta(seconds=10),
+        )
+
+
+@activity.defn
+async def compose_greeting(name: str) -> str:
+    return f"Hello, {name}!"
+
+
+interrupt_event = asyncio.Event()
+
+
+def init_opentelemetry() -> None:
+    # Setup global tracer for workflow traces
+    provider = TracerProvider(resource=Resource.create({SERVICE_NAME: "my-service"}))
+    provider.add_span_processor(BatchSpanProcessor(JaegerExporter()))
+    trace.set_tracer_provider(provider)
+
+    # Setup SDK metrics to OTel endpoint
+    telemetry.init_telemetry(
+        telemetry.TelemetryConfig(
+            otel_metrics=telemetry.OtelCollectorConfig(
+                url="http://localhost:4317", headers={}
+            )
+        )
+    )
+
+
+async def main():
+    init_opentelemetry()
+
+    # Connect client
+    client = await Client.connect(
+        "localhost:7233",
+        # Use OpenTelemetry interceptor
+        interceptors=[TracingInterceptor()],
+    )
+
+    # Run a worker for the workflow
+    async with Worker(
+        client,
+        task_queue="open_telemetry-task-queue",
+        workflows=[GreetingWorkflow],
+        activities=[compose_greeting],
+    ):
+        # Wait until interrupted
+        print("Worker started, ctrl+c to exit")
+        await interrupt_event.wait()
+        print("Shutting down")
+
+
+if __name__ == "__main__":
+    loop = asyncio.new_event_loop()
+    try:
+        loop.run_until_complete(main())
+    except KeyboardInterrupt:
+        interrupt_event.set()
+        loop.run_until_complete(loop.shutdown_asyncgens())