ENG-9166 feat(otel): built-in OpenTelemetry trace points, metrics and the reflex-otel instrumentor (1/3)

.github/scripts/dispatch_release/detect.sh

@@ -18,9 +18,10 @@ declare -A MAP=(
   [reflex_components_sonner]=reflex-components-sonner
   [reflex_docgen]=reflex-docgen
   [reflex_hosting_cli]=reflex-hosting-cli
+  [reflex_otel]=reflex-otel
   [reflex_release]=reflex-release
 )
-ORDER=(hatch_reflex_pyi reflex_base reflex_components_code reflex_components_core reflex_components_dataeditor reflex_components_gridjs reflex_components_lucide reflex_components_markdown reflex_components_moment reflex_components_plotly reflex_components_radix reflex_components_react_player reflex_components_recharts reflex_components_sonner reflex_docgen reflex_hosting_cli reflex_release)
+ORDER=(hatch_reflex_pyi reflex_base reflex_components_code reflex_components_core reflex_components_dataeditor reflex_components_gridjs reflex_components_lucide reflex_components_markdown reflex_components_moment reflex_components_plotly reflex_components_radix reflex_components_react_player reflex_components_recharts reflex_components_sonner reflex_docgen reflex_hosting_cli reflex_otel reflex_release)
 
 PACKAGES=()
 for key in "${ORDER[@]}"; do

.github/workflows/dispatch_release.yml

@@ -129,6 +129,10 @@ on:
         description: "reflex-hosting-cli"
         type: boolean
         default: false
+      reflex_otel:
+        description: "reflex-otel"
+        type: boolean
+        default: false
       reflex_release:
         description: "reflex-release"
         type: boolean
@@ -171,6 +175,7 @@ jobs:
           reflex_docgen: ${{ inputs.reflex_docgen }}
           reflex_hosting_cli: ${{ inputs.reflex_hosting_cli }}
           reflex_release: ${{ inputs.reflex_release }}
+          reflex_otel: ${{ inputs.reflex_otel }}
         run: bash .github/scripts/dispatch_release/detect.sh
 
   materialize:

.github/workflows/publish.yml

@@ -68,6 +68,7 @@ on:
           - reflex-components-sonner
           - reflex-docgen
           - reflex-hosting-cli
+          - reflex-otel
           - reflex-release
           - reflex-site-shared
       version:

news/6227.feature.md

@@ -0,0 +1 @@
+Propagate a frontend `traceparent` into event spans, count websocket connections and message sizes, and wrap the ASGI app when the `reflex-otel` instrumentor is active.

packages/reflex-base/news/6227.feature.md

@@ -0,0 +1 @@
+Add inert OpenTelemetry trace points and metrics around event handler execution, state acquisition and socket messages (`reflex_base.otel`); they cost one boolean check until the `reflex-otel` package enables them.

packages/reflex-base/pyproject.toml

@@ -8,6 +8,7 @@ authors = [{ name = "Khaleel Al-Adhami", email = "khaleel@reflex.dev" }]
 maintainers = [{ name = "Khaleel Al-Adhami", email = "khaleel@reflex.dev" }]
 requires-python = ">=3.10"
 dependencies = [
+  "opentelemetry-api >=1.30.0,<2.0",
   "packaging >=24.2,<27",
   "rich >=13,<16",
   "typing_extensions >=4.13.0",

packages/reflex-base/src/reflex_base/event/context.py

@@ -8,10 +8,13 @@
 from collections.abc import Callable, Mapping
 from typing import TYPE_CHECKING, Any, Protocol
 
+from reflex_base import otel
 from reflex_base.context.base import BaseContext
 from reflex_base.utils.format import to_snake_case
 
 if TYPE_CHECKING:
+    from opentelemetry.context import Context
+
     from reflex.istate.manager import StateManager
     from reflex_base.event import Event
 
@@ -98,6 +101,8 @@ class EventContext(BaseContext):
     cached_states: dict[type, Any] = dataclasses.field(
         default_factory=dict, init=False, repr=False
     )
+    # OpenTelemetry context active when this event was enqueued (None when tracing is off).
+    otel_context: Context | None = dataclasses.field(default=None, repr=False)
 
     def fork(self, token: str | None = None) -> EventContext:
         """Return a new EventContext with the specified fields replaced.
@@ -115,6 +120,7 @@ def fork(self, token: str | None = None) -> EventContext:
             enqueue_impl=self.enqueue_impl,
             emit_delta_impl=self.emit_delta_impl,
             emit_event_impl=self.emit_event_impl,
+            otel_context=otel.capture_context(),
         )
 
     async def emit_delta(self, delta: Mapping[str, Mapping[str, Any]]) -> None:

packages/reflex-base/src/reflex_base/event/processor/base_state_processor.py

@@ -9,12 +9,14 @@
 from collections.abc import Mapping, Sequence
 from enum import Enum
 from importlib.util import find_spec
+from time import perf_counter
 from typing import TYPE_CHECKING, Any
 
 from reflex.istate.data import RouterData
 from reflex.istate.manager.token import BaseStateToken
 from reflex.istate.proxy import StateProxy
 from reflex.utils import console, types
+from reflex_base import otel
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor.event_processor import EventProcessor, EventQueueEntry
 from reflex_base.registry import RegisteredEventHandler
@@ -340,6 +342,7 @@ async def _execute_event(
         ctx = entry.ctx
         event = entry.event
         router_data = event.router_data or {}
+        acquire_start = perf_counter() if otel.enabled else 0.0
         # Get the state for the session exclusively.
         async with ctx.state_manager.modify_state_with_links(
             BaseStateToken(
@@ -348,6 +351,8 @@ async def _execute_event(
             ),
             event=entry.event,
         ) as state:
+            if otel.enabled:
+                otel.record_state_acquired(acquire_start, event)
             # Compatibility hack rehydrate the state before processing this event.
             needs_to_rehydrate = bool(
                 not state.router_data and event.name != _hydrate_event_name()
@@ -411,6 +416,10 @@ async def _handle_backend_exception(
             if ev_ctx is not None:
                 # Ensure the event context is set for the exception handler.
                 EventContext.set(ev_ctx)
+                if otel.enabled:
+                    # Chain the handler's events under the failed event's parent context
+                    # (its remote or enqueuing span); parent_txid links them to it.
+                    otel.attach_context(ev_ctx.otel_context)
             if events := self.backend_exception_handler(ex):
                 await chain_updates(
                     events=events,

packages/reflex-base/src/reflex_base/event/processor/event_processor.py

@@ -20,6 +20,7 @@
 from reflex.app_mixins.middleware import MiddlewareMixin
 from reflex.istate.manager import StateManager
 from reflex.utils import console
+from reflex_base import otel
 from reflex_base.event.context import EventContext
 from reflex_base.event.processor.future import EventFuture
 from reflex_base.event.processor.timeout import DrainTimeoutManager
@@ -576,7 +577,15 @@ async def _process_event_queue_entry(
         """
         # Set up the event context for this task.
         EventContext.set(entry.ctx)
-        await self._execute_event(entry=entry, registered_handler=registered_handler)
+        if not otel.enabled:
+            await self._execute_event(
+                entry=entry, registered_handler=registered_handler
+            )
+            return
+        with otel.event_span(entry.event, entry.ctx, registered_handler):
+            await self._execute_event(
+                entry=entry, registered_handler=registered_handler
+            )
 
     def _create_event_task(
         self,