feat: Add opt-in OpenTelemetry trace export

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## Unreleased
+
+### Features
+
+* add opt-in OpenTelemetry trace export (HTTP, upstream polling, httpx)
+  ([#194](https://github.com/Flagsmith/edge-proxy/issues/194))
+
 ## [2.23.0](https://github.com/Flagsmith/edge-proxy/compare/v2.22.0...v2.23.0) (2026-05-15)
 
 

README.md

@@ -87,7 +87,41 @@ docker run \
     -e CONFIG_PATH=/var/foo.json \
     -v /<path-to-config>/config.json:/var/foo.json \
     flagsmith/edge-proxy:latest
-``` 
+```
+
+## OpenTelemetry
+
+Edge Proxy supports opt-in distributed trace export over OTLP, aligned
+with the [Flagsmith OpenTelemetry
+docs](https://docs.flagsmith.com/deployment-self-hosting/observability/opentelemetry).
+
+Set `OTEL_EXPORTER_OTLP_ENDPOINT` to enable export. When unset, no
+OpenTelemetry code is loaded and there is no runtime overhead.
+
+```bash
+OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318   # enables OTel
+OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf          # or grpc
+OTEL_SERVICE_NAME=flagsmith-edge-proxy             # default
+OTEL_RESOURCE_ATTRIBUTES=deployment.environment=prod
+OTEL_TRACES_EXPORTER=otlp                          # or none to disable
+OTEL_SDK_DISABLED=true                             # hard disable
+OTEL_TRACING_EXCLUDED_URL_PATHS=proxy/health/liveness,proxy/health/readiness
+```
+
+Example Docker run:
+
+```sh
+docker run \
+  -e OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 \
+  -e OTEL_EXPORTER_OTLP_PROTOCOL=grpc \
+  -v ./config.json:/app/config.json \
+  flagsmith/edge-proxy:latest
+```
+
+When enabled, traces cover incoming HTTP requests (FastAPI
+auto-instrumentation), upstream environment-document polling, and
+outbound httpx calls. Structured logs include `trace_id` and `span_id`
+when an active span exists.
 
 ## Load Testing
 

pyproject.toml

@@ -15,6 +15,12 @@ dependencies = [
     "structlog",
     "uvicorn",
     "pydantic-settings>=2.2.1",
+    "opentelemetry-api>=1.25,<2",
+    "opentelemetry-sdk>=1.25,<2",
+    "opentelemetry-exporter-otlp-proto-http>=1.25,<2",
+    "opentelemetry-exporter-otlp-proto-grpc>=1.25,<2",
+    "opentelemetry-instrumentation-fastapi>=0.46b0,<1",
+    "opentelemetry-instrumentation-httpx>=0.46b0,<1",
 ]
 requires-python = ">= 3.12"
 

requirements.lock

@@ -47,6 +47,62 @@ jsonpath-rfc9535==0.2.0
     # via flagsmith-flag-engine
 marshmallow==3.21.1
     # via edge-proxy
+googleapis-common-protos==1.75.0
+    # via opentelemetry-exporter-otlp-proto-grpc
+    # via opentelemetry-exporter-otlp-proto-http
+grpcio==1.82.0
+    # via opentelemetry-exporter-otlp-proto-grpc
+opentelemetry-api==1.43.0
+    # via edge-proxy
+    # via opentelemetry-exporter-otlp-proto-grpc
+    # via opentelemetry-exporter-otlp-proto-http
+    # via opentelemetry-instrumentation
+    # via opentelemetry-instrumentation-asgi
+    # via opentelemetry-instrumentation-fastapi
+    # via opentelemetry-instrumentation-httpx
+    # via opentelemetry-sdk
+    # via opentelemetry-semantic-conventions
+opentelemetry-exporter-otlp-proto-common==1.43.0
+    # via opentelemetry-exporter-otlp-proto-grpc
+    # via opentelemetry-exporter-otlp-proto-http
+opentelemetry-exporter-otlp-proto-grpc==1.43.0
+    # via edge-proxy
+opentelemetry-exporter-otlp-proto-http==1.43.0
+    # via edge-proxy
+opentelemetry-instrumentation==0.64b0
+    # via opentelemetry-instrumentation-asgi
+    # via opentelemetry-instrumentation-fastapi
+    # via opentelemetry-instrumentation-httpx
+opentelemetry-instrumentation-asgi==0.64b0
+    # via opentelemetry-instrumentation-fastapi
+opentelemetry-instrumentation-fastapi==0.64b0
+    # via edge-proxy
+opentelemetry-instrumentation-httpx==0.64b0
+    # via edge-proxy
+opentelemetry-proto==1.43.0
+    # via opentelemetry-exporter-otlp-proto-common
+    # via opentelemetry-exporter-otlp-proto-grpc
+    # via opentelemetry-exporter-otlp-proto-http
+opentelemetry-sdk==1.43.0
+    # via edge-proxy
+    # via opentelemetry-exporter-otlp-proto-grpc
+    # via opentelemetry-exporter-otlp-proto-http
+opentelemetry-semantic-conventions==0.64b0
+    # via opentelemetry-instrumentation
+    # via opentelemetry-instrumentation-asgi
+    # via opentelemetry-instrumentation-fastapi
+    # via opentelemetry-instrumentation-httpx
+    # via opentelemetry-sdk
+opentelemetry-util-http==0.64b0
+    # via opentelemetry-instrumentation-asgi
+    # via opentelemetry-instrumentation-fastapi
+    # via opentelemetry-instrumentation-httpx
+protobuf==7.35.1
+    # via googleapis-common-protos
+    # via opentelemetry-proto
+wrapt==2.2.2
+    # via opentelemetry-instrumentation
+    # via opentelemetry-instrumentation-httpx
 orjson==3.10.0
     # via edge-proxy
 packaging==24.0

src/edge_proxy/environments.py

@@ -28,6 +28,7 @@
 )
 from edge_proxy.models import IdentityWithTraits
 from edge_proxy.settings import AppSettings, EnvironmentKeyPair
+from edge_proxy.telemetry import get_tracer
 
 logger = structlog.get_logger(__name__)
 
@@ -64,22 +65,30 @@ def __init__(
                 )(self.get_identity_response_data)
 
     async def refresh_environment_caches(self):
-        received_error = False
-        for key_pair in self.settings.environment_key_pairs:
-            try:
-                environment_document = await self._fetch_document(key_pair)
-                if self.cache.put_environment(
-                    environment_api_key=key_pair.client_side_key,
-                    environment_document=environment_document,
-                ):
-                    await self._clear_endpoint_caches()
-            except (httpx.HTTPError, orjson.JSONDecodeError):
-                logger.exception(
-                    "error_fetching_document", client_side_key=key_pair.client_side_key
-                )
-                received_error = True
-        if not received_error:
-            self.last_updated_at = datetime.now()
+        with get_tracer().start_as_current_span(
+            "poll.refresh_environment_caches",
+        ) as span:
+            span.set_attribute(
+                "environments.count",
+                len(self.settings.environment_key_pairs),
+            )
+            received_error = False
+            for key_pair in self.settings.environment_key_pairs:
+                try:
+                    environment_document = await self._fetch_document(key_pair)
+                    if self.cache.put_environment(
+                        environment_api_key=key_pair.client_side_key,
+                        environment_document=environment_document,
+                    ):
+                        await self._clear_endpoint_caches()
+                except (httpx.HTTPError, orjson.JSONDecodeError):
+                    logger.exception(
+                        "error_fetching_document",
+                        client_side_key=key_pair.client_side_key,
+                    )
+                    received_error = True
+            if not received_error:
+                self.last_updated_at = datetime.now()
 
     def get_flags_response_data(
         self, environment_key: str, feature: str | None = None
@@ -95,7 +104,13 @@ def get_flags_response_data(
         hide_disabled_flags = _get_hide_disabled_flags(environment_document)
 
         context = map_environment_document_to_context(environment_document)
-        evaluation_result = get_evaluation_result(context)
+        with get_tracer().start_as_current_span("evaluate.flags") as span:
+            span.set_attribute("flagsmith.single_feature", feature is not None)
+            evaluation_result = get_evaluation_result(context)
+            span.set_attribute(
+                "flagsmith.feature_count",
+                len(evaluation_result["flags"]),
+            )
 
         feature_types = self.cache.get_feature_types(environment_key)
         if feature_types is None:
@@ -146,7 +161,12 @@ def get_identity_response_data(
             identifier=input_data.identifier,
             traits=convert_traits_to_dict(input_data.traits),
         )
-        evaluation_result = get_evaluation_result(context)
+        with get_tracer().start_as_current_span("evaluate.identity") as span:
+            evaluation_result = get_evaluation_result(context)
+            span.set_attribute(
+                "flagsmith.feature_count",
+                len(evaluation_result["flags"]),
+            )
 
         feature_types = self.cache.get_feature_types(environment_key)
         if feature_types is None:
@@ -179,40 +199,56 @@ def get_environment(
         raise FlagsmithUnknownKeyError(environment_key)
 
     async def _fetch_document(self, key_pair: EnvironmentKeyPair) -> dict[str, Any]:
-        headers = {
-            "X-Environment-Key": key_pair.server_side_key,
-        }
-        environment_document = self.cache.get_environment(
-            environment_api_key=key_pair.client_side_key
-        )
-        if environment_document:
-            updated_at: str = environment_document.get("updated_at")
-            if updated_at:
-                try:
-                    epoch_seconds = datetime.fromisoformat(updated_at).timestamp()
-                    # Same implementation as https://docs.djangoproject.com/en/4.2/ref/utils/#django.utils.http.http_date
-                    headers["If-Modified-Since"] = formatdate(
-                        epoch_seconds, usegmt=True
-                    )
-                except ValueError:
+        with get_tracer().start_as_current_span("poll.fetch_document") as span:
+            headers = {
+                "X-Environment-Key": key_pair.server_side_key,
+            }
+            environment_document = self.cache.get_environment(
+                environment_api_key=key_pair.client_side_key
+            )
+            if environment_document:
+                updated_at: str = environment_document.get("updated_at")
+                if updated_at:
+                    try:
+                        epoch_seconds = datetime.fromisoformat(
+                            updated_at,
+                        ).timestamp()
+                        # Same implementation as
+                        # https://docs.djangoproject.com/en/4.2/ref/utils/
+                        # #django.utils.http.http_date
+                        headers["If-Modified-Since"] = formatdate(
+                            epoch_seconds, usegmt=True
+                        )
+                    except ValueError:
+                        logger.warning(
+                            "failed to parse updated_at, "
+                            f"environment={key_pair.client_side_key} "
+                            f"updated_at={updated_at}"
+                        )
+                else:
                     logger.warning(
-                        f"failed to parse updated_at, environment={key_pair.client_side_key} updated_at={updated_at}"
+                        "received environment with no updated_at: "
+                        f"{key_pair.client_side_key}"
                     )
-            else:
-                logger.warning(
-                    f"received environment with no updated_at: {key_pair.client_side_key}"
+            try:
+                response = await self._client.get(
+                    url=f"{self.settings.api_url}/environment-document/",
+                    headers=headers,
                 )
-        response = await self._client.get(
-            url=f"{self.settings.api_url}/environment-document/",
-            headers=headers,
-        )
-        if response.status_code == starlette.status.HTTP_304_NOT_MODIFIED:
-            assert environment_document, (
-                f"GET /environment-document returned 304 without a cached document. environment={key_pair.client_side_key}"
-            )
-            return environment_document
-        response.raise_for_status()
-        return orjson.loads(response.text)
+                span.set_attribute("http.status_code", response.status_code)
+                if response.status_code == starlette.status.HTTP_304_NOT_MODIFIED:
+                    span.set_attribute("flagsmith.cache_hit", True)
+                    assert environment_document, (
+                        "GET /environment-document returned 304 without a "
+                        f"cached document. environment={key_pair.client_side_key}"
+                    )
+                    return environment_document
+                span.set_attribute("flagsmith.cache_hit", False)
+                response.raise_for_status()
+                return orjson.loads(response.text)
+            except (httpx.HTTPError, orjson.JSONDecodeError) as exc:
+                span.record_exception(exc)
+                raise
 
     async def _clear_endpoint_caches(self):
         for func in (self.get_identity_response_data, self.get_flags_response_data):

src/edge_proxy/logging.py

@@ -5,6 +5,7 @@
 import structlog
 
 from edge_proxy.settings import LogFormat, LoggingSettings
+from edge_proxy.telemetry import add_otel_trace_context
 
 
 def _extract_gunicorn_access_log_event(
@@ -49,13 +50,17 @@ def _drop_color_message(
 ]
 
 
-def setup_logging(settings: LoggingSettings) -> None:
-    processors = [
-        *COMMON_PROCESSORS,
-        structlog.processors.EventRenamer(settings.log_event_field_name),
-        structlog.dev.set_exc_info,
-        structlog.processors.format_exc_info,
-    ]
+def setup_logging(settings: LoggingSettings, *, otel_enabled: bool = False) -> None:
+    processors: list[structlog.types.Processor] = [*COMMON_PROCESSORS]
+    if otel_enabled:
+        processors.append(add_otel_trace_context)
+    processors.extend(
+        [
+            structlog.processors.EventRenamer(settings.log_event_field_name),
+            structlog.dev.set_exc_info,
+            structlog.processors.format_exc_info,
+        ],
+    )
 
     structlog.configure(
         processors=processors

src/edge_proxy/server.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timedelta
 import asyncio
+import os
 
 import httpx
 from contextlib import asynccontextmanager
@@ -16,15 +17,24 @@
 from edge_proxy.logging import setup_logging
 from edge_proxy.models import IdentityWithTraits
 from edge_proxy.settings import get_settings
+from edge_proxy.telemetry import (
+    instrument_fastapi,
+    instrument_httpx,
+    setup_telemetry,
+)
 
 settings = get_settings()
-setup_logging(settings.logging)
+telemetry = setup_telemetry()
+setup_logging(settings.logging, otel_enabled=telemetry.enabled)
+http_client = httpx.AsyncClient(
+    timeout=settings.api_poll_timeout_seconds,
+    follow_redirects=True,
+)
+if telemetry.enabled:
+    instrument_httpx()
 environment_service = EnvironmentService(
     LocalMemEnvironmentsCache(),
-    httpx.AsyncClient(
-        timeout=settings.api_poll_timeout_seconds,
-        follow_redirects=True,
-    ),
+    http_client,
     settings,
 )
 
@@ -147,3 +157,9 @@ async def environment_document(
     allow_headers=["*"],
 )
 app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+if telemetry.enabled:
+    instrument_fastapi(
+        app,
+        excluded_urls=os.environ.get("OTEL_TRACING_EXCLUDED_URL_PATHS"),
+    )