Add OpenTelemetry middleware

.github/workflows/ci.yaml

@@ -67,7 +67,7 @@ jobs:
           go-version-file: protoc-gen-connect-python/go.mod
           cache-dependency-path: "**/go.mod"
 
-      - run: uv sync
+      - run: uv sync --all-packages
 
       - name: run lints
         if: startsWith(matrix.os, 'ubuntu-')
@@ -85,6 +85,10 @@ jobs:
         run: uv run pytest ${{ matrix.coverage == 'cov' && '--cov=connectrpc --cov-report=xml' || '' }}
         working-directory: conformance
 
+      - name: run OTel tests
+        run: uv run pytest ${{ matrix.coverage == 'cov' && '--cov=connectrpc --cov-report=xml' || '' }}
+        working-directory: connectrpc-otel
+
       - name: run Go tests
         run: go test ./...
         working-directory: protoc-gen-connect-python

connect-python.code-workspace

@@ -8,6 +8,10 @@
       "name": "conformance",
       "path": "./conformance",
     },
+    {
+      "name": "connectrpc-otel",
+      "path": "./connectrpc-otel",
+    },
     {
       "name": "example",
       "path": "./example",

connectrpc-otel/README.md

@@ -0,0 +1,24 @@
+# connectrpc-otel
+
+OpenTelemetry middleware for connect-python to generate server and client spans
+for ConnectRPC requests.
+
+Auto-instrumentation is currently not supported.
+
+## Example
+
+```python
+
+from connectrpc_otel import OpenTelemetryInterceptor
+
+from eliza_connect import ElizaServiceWSGIApplication, ElizaServiceClientSync
+
+from ._service import MyElizaService
+
+app = ElizaServiceWSGIApplication(MyElizaService(), interceptors=[OpenTelemetryInterceptor()])
+
+def make_request():
+    client = ElizaServiceClientSync("http://localhost:8080", interceptors=[OpenTelemetryInterceptor(client=True)])
+    resp = client.Say(SayRequest(sentence="Hello!"))
+    print(resp)
+```

connectrpc-otel/connectrpc_otel/__init__.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+__all__ = ["OpenTelemetryInterceptor"]
+
+from ._interceptor import OpenTelemetryInterceptor

connectrpc-otel/connectrpc_otel/_interceptor.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from contextlib import AbstractContextManager, contextmanager
+from typing import TYPE_CHECKING, TypeAlias, TypeVar, cast
+
+from opentelemetry.propagate import get_global_textmap
+from opentelemetry.propagators.textmap import Setter, TextMapPropagator, default_setter
+from opentelemetry.trace import (
+    Span,
+    SpanKind,
+    TracerProvider,
+    get_current_span,
+    get_tracer_provider,
+)
+
+from connectrpc.errors import ConnectError
+
+from ._semconv import (
+    CLIENT_ADDRESS,
+    CLIENT_PORT,
+    ERROR_TYPE,
+    RPC_METHOD,
+    RPC_RESPONSE_STATUS_CODE,
+    RPC_SYSTEM_NAME,
+    SERVER_ADDRESS,
+    SERVER_PORT,
+    RpcSystemNameValues,
+)
+from ._version import __version__
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator, MutableMapping
+
+    from opentelemetry.util.types import AttributeValue
+
+    from connectrpc.request import RequestContext
+
+REQ = TypeVar("REQ")
+RES = TypeVar("RES")
+
+Token: TypeAlias = tuple[AbstractContextManager, Span]
+
+# Workaround bad typing
+_DEFAULT_TEXTMAP_SETTER = cast("Setter[MutableMapping[str, str]]", default_setter)
+
+
+class OpenTelemetryInterceptor:
+    """Interceptor to generate telemetry for RPC server and client requests."""
+
+    def __init__(
+        self,
+        *,
+        propagator: TextMapPropagator | None = None,
+        tracer_provider: TracerProvider | None = None,
+        client: bool = False,
+    ) -> None:
+        """Creates a new OpenTelemetry interceptor.
+
+        Args:
+            propagator: The OpenTelemetry TextMapPropagator to use. If not
+                provided, the global default will be used.
+            tracer_provider: The OpenTelemetry TracerProvider to use. If not
+                provided, the global default will be used.
+            client: Whether this interceptor is for a client or server.
+        """
+        self._client = client
+        tracer_provider = tracer_provider or get_tracer_provider()
+        self._tracer = tracer_provider.get_tracer("connectrpc-otel", __version__)
+        self._propagator = propagator or get_global_textmap()
+
+    async def on_start(self, ctx: RequestContext) -> Token:
+        return self.on_start_sync(ctx)
+
+    def on_start_sync(self, ctx: RequestContext) -> Token:
+        cm = self._start_span(ctx)
+        span = cm.__enter__()
+        return cm, span
+
+    async def on_end(
+        self, token: Token, ctx: RequestContext, error: Exception | None
+    ) -> None:
+        self.on_end_sync(token, ctx, error)
+
+    def on_end_sync(
+        self, token: Token, ctx: RequestContext, error: Exception | None
+    ) -> None:
+        cm, span = token
+        self._finish_span(span, error)
+        if error:
+            cm.__exit__(type(error), error, error.__traceback__)
+        else:
+            cm.__exit__(None, None, None)
+
+    @contextmanager
+    def _start_span(self, ctx: RequestContext) -> Iterator[Span]:
+        parent_otel_ctx = None
+        if self._client:
+            span_kind = SpanKind.CLIENT
+            carrier = ctx.request_headers()
+            self._propagator.inject(carrier, setter=_DEFAULT_TEXTMAP_SETTER)
+        else:
+            span_kind = SpanKind.SERVER
+            parent_span = get_current_span()
+            if not parent_span.get_span_context().is_valid:
+                carrier = ctx.request_headers()
+                parent_otel_ctx = self._propagator.extract(carrier)
+
+        rpc_method = f"{ctx.method().service_name}/{ctx.method().name}"
+
+        attrs: MutableMapping[str, AttributeValue] = {
+            RPC_SYSTEM_NAME: RpcSystemNameValues.CONNECTRPC.value,
+            RPC_METHOD: rpc_method,
+        }
+        if sa := ctx.server_address():
+            addr, port = sa.rsplit(":", 1)
+            attrs[SERVER_ADDRESS] = addr
+            attrs[SERVER_PORT] = int(port)
+        if ca := ctx.client_address():
+            addr, port = ca.rsplit(":", 1)
+            attrs[CLIENT_ADDRESS] = addr
+            attrs[CLIENT_PORT] = int(port)
+
+        with self._tracer.start_as_current_span(
+            rpc_method, kind=span_kind, attributes=attrs, context=parent_otel_ctx
+        ) as span:
+            yield span
+
+    def _finish_span(self, span: Span, error: Exception | None) -> None:
+        if error:
+            if isinstance(error, ConnectError):
+                span.set_attribute(RPC_RESPONSE_STATUS_CODE, error.code.value)
+            else:
+                span.set_attribute(RPC_RESPONSE_STATUS_CODE, "unknown")
+                span.set_attribute(ERROR_TYPE, type(error).__qualname__)

connectrpc-otel/connectrpc_otel/_semconv.py

@@ -0,0 +1,22 @@
+# Vendored in OpenTelemetry semantic conventions for connect-python to avoid
+# unstable imports. We don't copy docstrings since for us they are implementation
+# details and should be obvious enough.
+
+# https://github.com/open-telemetry/opentelemetry-python/blob/main/opentelemetry-semantic-conventions/src/opentelemetry/semconv/_incubating/attributes/rpc_attributes.py
+from __future__ import annotations
+
+from enum import Enum
+from typing import Final
+
+CLIENT_ADDRESS: Final = "client.address"
+CLIENT_PORT: Final = "client.port"
+ERROR_TYPE: Final = "error.type"
+RPC_METHOD: Final = "rpc.method"
+RPC_RESPONSE_STATUS_CODE: Final = "rpc.response.status_code"
+RPC_SYSTEM_NAME: Final = "rpc.system.name"
+SERVER_ADDRESS: Final = "server.address"
+SERVER_PORT: Final = "server.port"
+
+
+class RpcSystemNameValues(Enum):
+    CONNECTRPC = "connectrpc"

connectrpc-otel/connectrpc_otel/_version.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from importlib.metadata import version
+
+__version__ = version("connectrpc-otel")

connectrpc-otel/pyproject.toml

@@ -0,0 +1,61 @@
+[project]
+name = "connectrpc-otel"
+version = "0.1.0"
+description = "OpenTelemetry instrumentation for connectrpc"
+maintainers = [
+    { name = "Anuraag Agrawal", email = "anuraaga@gmail.com" },
+    { name = "Spencer Nelson", email = "spencer@firetiger.com" },
+    { name = "Stefan VanBuren", email = "svanburen@buf.build" },
+    { name = "Yasushi Itoh", email = "i2y.may.roku@gmail.com" },
+]
+requires-python = ">= 3.10"
+readme = "README.md"
+license = "Apache-2.0"
+keywords = [
+    "opentelemetry",
+    "otel",
+    "connectrpc",
+    "connect-python",
+    "middleware",
+    "tracing",
+    "observability",
+]
+classifiers = [
+    "Development Status :: 4 - Beta",
+    "Environment :: Console",
+    "Intended Audience :: Developers",
+    "License :: OSI Approved :: Apache Software License",
+    "Operating System :: OS Independent",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
+    "Topic :: System :: Networking",
+]
+dependencies = ["connect-python>=0.8.0", "opentelemetry-api>=1.39.1"]
+
+[dependency-groups]
+dev = [
+    "opentelemetry-sdk==1.39.1",
+    "opentelemetry-instrumentation-asgi==0.60b1",
+    "opentelemetry-instrumentation-wsgi==0.60b1",
+
+    "connect-python-example",
+    "pytest",
+]
+
+[project.urls]
+Homepage = "https://github.com/connectrpc/connect-python"
+Repository = "https://github.com/connectrpc/connect-python"
+Issues = "https://github.com/connectrpc/connect-python/issues"
+
+[build-system]
+requires = ["uv_build>=0.10.0,<0.11.0"]
+build-backend = "uv_build"
+
+[tool.uv.build-backend]
+module-name = "connectrpc_otel"
+module-root = ""