Feat: Hatchet OTel Instrumentor

.github/workflows/e2e.yaml

@@ -28,7 +28,7 @@ jobs:
           virtualenvs-create: true
           virtualenvs-in-project: true
       - name: Install dependencies
-        run: poetry install --no-interaction
+        run: poetry install --no-interaction --all-extras
 
       - name: Generate Env File
         run: |

.github/workflows/lint.yaml

@@ -26,7 +26,7 @@ jobs:
           virtualenvs-in-project: true
 
       - name: Install linting tools
-        run: poetry install
+        run: poetry install --all-extras
 
       - name: Run Black
         run: poetry run black . --check --verbose --diff --color

examples/opentelemetry_instrumentation/client.py

@@ -0,0 +1,7 @@
+from dotenv import load_dotenv
+
+from hatchet_sdk import Hatchet
+
+load_dotenv()
+
+hatchet = Hatchet(debug=True)

examples/opentelemetry_instrumentation/test_otel_instrumentation.py

@@ -0,0 +1,60 @@
+import json
+
+import pytest
+from opentelemetry.trace import NoOpTracerProvider
+
+from hatchet_sdk import Hatchet, Worker
+from hatchet_sdk.clients.admin import TriggerWorkflowOptions
+from hatchet_sdk.clients.events import PushEventOptions
+from hatchet_sdk.opentelemetry.instrumentor import HatchetInstrumentor
+
+trace_provider = NoOpTracerProvider()
+
+instrumentor = HatchetInstrumentor(tracer_provider=trace_provider)
+instrumentor.instrument()
+
+tracer = trace_provider.get_tracer(__name__)
+
+
+def create_additional_metadata() -> dict[str, str]:
+    return instrumentor.inject_traceparent_into_metadata(
+        {"hello": "world"}, instrumentor.create_traceparent()
+    )
+
+
+def create_push_options() -> PushEventOptions:
+    return {"additional_metadata": create_additional_metadata()}
+
+
+@pytest.mark.parametrize("worker", ["otel"], indirect=True)
+def test_push_event(hatchet: Hatchet, worker: Worker) -> None:
+    key = "otel:event"
+    payload = {"test": "test"}
+
+    with tracer.start_as_current_span("push_event"):
+        event = hatchet.event.push(
+            event_key=key,
+            payload=payload,
+            options=create_push_options(),
+        )
+
+        """Assert on `endswith` to ignore namespacing"""
+        assert event.key.endswith(key)
+        assert event.payload == json.dumps(payload)
+
+
+@pytest.mark.skip("Failing in CI for unknown reason")
+@pytest.mark.asyncio()
+@pytest.mark.parametrize("worker", ["otel"], indirect=True)
+async def test_run_workflow(aiohatchet: Hatchet, worker: Worker) -> None:
+    with tracer.start_as_current_span("run_workflow") as span:
+        workflow = aiohatchet.admin.run_workflow(
+            "OTelWorkflow",
+            {"test": "test"},
+            options=TriggerWorkflowOptions(
+                additional_metadata=create_additional_metadata()
+            ),
+        )
+
+        with pytest.raises(Exception, match="Workflow Errors"):
+            await workflow.result()

examples/opentelemetry_instrumentation/tracer.py

@@ -0,0 +1,45 @@
+import os
+from typing import cast
+
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.trace import NoOpTracerProvider
+
+trace_provider: TracerProvider | NoOpTracerProvider
+
+if os.getenv("CI", "false") == "true":
+    trace_provider = NoOpTracerProvider()
+else:
+    resource = Resource(
+        attributes={
+            SERVICE_NAME: os.getenv("HATCHET_CLIENT_OTEL_SERVICE_NAME", "test-service")
+        }
+    )
+
+    headers = dict(
+        [
+            cast(
+                tuple[str, str],
+                tuple(
+                    os.getenv(
+                        "HATCHET_CLIENT_OTEL_EXPORTER_OTLP_HEADERS", "foo=bar"
+                    ).split("=")
+                ),
+            )
+        ]
+    )
+
+    processor = BatchSpanProcessor(
+        OTLPSpanExporter(
+            endpoint=os.getenv(
+                "HATCHET_CLIENT_OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"
+            ),
+            headers=headers,
+        ),
+    )
+
+    trace_provider = TracerProvider(resource=resource)
+
+    trace_provider.add_span_processor(processor)

examples/opentelemetry_instrumentation/triggers.py

@@ -0,0 +1,161 @@
+import asyncio
+
+from examples.opentelemetry_instrumentation.client import hatchet
+from examples.opentelemetry_instrumentation.tracer import trace_provider
+from hatchet_sdk.clients.admin import TriggerWorkflowOptions
+from hatchet_sdk.clients.events import PushEventOptions
+from hatchet_sdk.opentelemetry.instrumentor import HatchetInstrumentor
+
+instrumentor = HatchetInstrumentor(tracer_provider=trace_provider)
+tracer = trace_provider.get_tracer(__name__)
+
+
+def create_additional_metadata() -> dict[str, str]:
+    return instrumentor.inject_traceparent_into_metadata(
+        {"hello": "world"}, instrumentor.create_traceparent()
+    )
+
+
+def create_push_options() -> PushEventOptions:
+    return {"additional_metadata": create_additional_metadata()}
+
+
+def push_event() -> None:
+    print("\npush_event")
+    with tracer.start_as_current_span("push_event") as span:
+        hatchet.event.push(
+            "otel:event",
+            {"test": "test"},
+            options=create_push_options(),
+        )
+
+
+async def async_push_event() -> None:
+    print("\nasync_push_event")
+    with tracer.start_as_current_span("async_push_event") as span:
+        await hatchet.event.async_push(
+            "otel:event", {"test": "test"}, options=create_push_options()
+        )
+
+
+def bulk_push_event() -> None:
+    print("\nbulk_push_event")
+    with tracer.start_as_current_span("bulk_push_event") as span:
+        hatchet.event.bulk_push(
+            [
+                {
+                    "additional_metadata": create_additional_metadata(),
+                    "key": "otel:event",
+                    "payload": {"test": "test 1"},
+                },
+                {
+                    "additional_metadata": create_additional_metadata(),
+                    "key": "otel:event",
+                    "payload": {"test": "test 2"},
+                },
+            ],
+        )
+
+
+async def async_bulk_push_event() -> None:
+    print("\nasync_bulk_push_event")
+    with tracer.start_as_current_span("bulk_push_event") as span:
+        await hatchet.event.async_bulk_push(
+            [
+                {
+                    "additional_metadata": create_additional_metadata(),
+                    "key": "otel:event",
+                    "payload": {"test": "test 1"},
+                },
+                {
+                    "additional_metadata": create_additional_metadata(),
+                    "key": "otel:event",
+                    "payload": {"test": "test 2"},
+                },
+            ],
+        )
+
+
+def run_workflow() -> None:
+    print("\nrun_workflow")
+    with tracer.start_as_current_span("run_workflow") as span:
+        hatchet.admin.run_workflow(
+            "OTelWorkflow",
+            {"test": "test"},
+            options=TriggerWorkflowOptions(
+                additional_metadata=create_additional_metadata()
+            ),
+        )
+
+
+async def async_run_workflow() -> None:
+    print("\nasync_run_workflow")
+    with tracer.start_as_current_span("async_run_workflow") as span:
+        await hatchet.admin.aio.run_workflow(
+            "OTelWorkflow",
+            {"test": "test"},
+            options=TriggerWorkflowOptions(
+                additional_metadata=create_additional_metadata()
+            ),
+        )
+
+
+def run_workflows() -> None:
+    print("\nrun_workflows")
+    with tracer.start_as_current_span("run_workflows") as span:
+        hatchet.admin.run_workflows(
+            [
+                {
+                    "workflow_name": "OTelWorkflow",
+                    "input": {"test": "test"},
+                    "options": TriggerWorkflowOptions(
+                        additional_metadata=create_additional_metadata()
+                    ),
+                },
+                {
+                    "workflow_name": "OTelWorkflow",
+                    "input": {"test": "test 2"},
+                    "options": TriggerWorkflowOptions(
+                        additional_metadata=create_additional_metadata()
+                    ),
+                },
+            ],
+        )
+
+
+async def async_run_workflows() -> None:
+    print("\nasync_run_workflows")
+    with tracer.start_as_current_span("async_run_workflows") as span:
+        await hatchet.admin.aio.run_workflows(
+            [
+                {
+                    "workflow_name": "OTelWorkflow",
+                    "input": {"test": "test"},
+                    "options": TriggerWorkflowOptions(
+                        additional_metadata=create_additional_metadata()
+                    ),
+                },
+                {
+                    "workflow_name": "OTelWorkflow",
+                    "input": {"test": "test 2"},
+                    "options": TriggerWorkflowOptions(
+                        additional_metadata=create_additional_metadata()
+                    ),
+                },
+            ],
+        )
+
+
+async def main() -> None:
+    push_event()
+    await async_push_event()
+    bulk_push_event()
+    await async_bulk_push_event()
+    run_workflow()
+    # await async_run_workflow()
+    run_workflows()
+    # await async_run_workflows()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/opentelemetry_instrumentation/worker.py

@@ -0,0 +1,47 @@
+from examples.opentelemetry_instrumentation.client import hatchet
+from examples.opentelemetry_instrumentation.tracer import trace_provider
+from hatchet_sdk import Context
+from hatchet_sdk.opentelemetry.instrumentor import HatchetInstrumentor
+
+HatchetInstrumentor(
+    tracer_provider=trace_provider,
+).instrument()
+
+
+@hatchet.workflow(on_events=["otel:event"])
+class OTelWorkflow:
+    @hatchet.step()
+    def your_spans_are_children_of_hatchet_span(
+        self, context: Context
+    ) -> dict[str, str]:
+        with trace_provider.get_tracer(__name__).start_as_current_span("step1"):
+            print("executed step")
+            return {
+                "foo": "bar",
+            }
+
+    @hatchet.step()
+    def your_spans_are_still_children_of_hatchet_span(self, context: Context) -> None:
+        with trace_provider.get_tracer(__name__).start_as_current_span("step2"):
+            raise Exception("Manually instrumented step failed failed")
+
+    @hatchet.step()
+    def this_step_is_still_instrumented(self, context: Context) -> dict[str, str]:
+        print("executed still-instrumented step")
+        return {
+            "still": "instrumented",
+        }
+
+    @hatchet.step()
+    def this_step_is_also_still_instrumented(self, context: Context) -> None:
+        raise Exception("Still-instrumented step failed")
+
+
+def main() -> None:
+    worker = hatchet.worker("otel-example-worker", max_runs=1)
+    worker.register_workflow(OTelWorkflow())
+    worker.start()
+
+
+if __name__ == "__main__":
+    main()