chore(internal): add native based event system

Author: brettlangdon

.gitlab/native.yml

@@ -21,3 +21,8 @@ include:
       echo -e "\e[0Ksection_start:`date +%s`:cargo_test[collapsed=true]\r\e[0Kcargo test"
       cargo test --no-fail-fast --locked
       echo -e "\e[0Ksection_end:`date +%s`:cargo_test\r\e[0K"
+    - |
+      echo -e "\e[0Ksection_start:`date +%s`:check_events_pyi[collapsed=true]\r\e[0Kcheck events.pyi"
+      cargo build
+      git diff --exit-code ../../ddtrace/internal/events.pyi || (echo "Error: events.pyi is out of date. Run 'cargo build' in src/native/ and commit the changes." && exit 1)
+      echo -e "\e[0Ksection_end:`date +%s`:check_events_pyi\r\e[0K"

ddtrace/_trace/events.py

@@ -0,0 +1,17 @@
+"""
+Trace-specific events.
+
+This module re-exports trace-related events from the native event system
+with domain-specific type annotations.
+"""
+
+from ddtrace.internal import events
+
+
+tracer_span_started = events.tracer_span_started
+tracer_span_finished = events.tracer_span_finished
+
+__all__ = [
+    "tracer_span_started",
+    "tracer_span_finished",
+]

ddtrace/_trace/events.pyi

@@ -0,0 +1,10 @@
+"""
+Type stubs for trace-specific events with concrete type annotations.
+"""
+
+from .span import Span
+from ddtrace.internal.events import Event
+
+# Trace events with concrete types
+tracer_span_started: Event[[Span]]
+tracer_span_finished: Event[[Span]]

ddtrace/_trace/tracer.py

@@ -16,6 +16,8 @@
 from typing import cast
 
 from ddtrace._trace.context import Context
+from ddtrace._trace.events import tracer_span_finished
+from ddtrace._trace.events import tracer_span_started
 from ddtrace._trace.processor import SpanAggregator
 from ddtrace._trace.processor import SpanProcessor
 from ddtrace._trace.processor import TopLevelSpanProcessor
@@ -30,7 +32,6 @@
 from ddtrace.constants import PID
 from ddtrace.constants import VERSION_KEY
 from ddtrace.internal import atexit
-from ddtrace.internal import core
 from ddtrace.internal import debug
 from ddtrace.internal import forksafe
 from ddtrace.internal import hostname
@@ -205,7 +206,8 @@ def on_start_span(self, func: Callable[[Span], None]) -> Callable[[Span], None]:
         :param func: The function to call when starting a span.
                      The started span will be passed as argument.
         """
-        core.on("trace.span_start", callback=func)
+        handler = tracer_span_started.on(func)
+        setattr(func, "_tracer_span_started_handler", handler)
         return func
 
     def deregister_on_start_span(self, func: Callable[[Span], None]) -> Callable[[Span], None]:
@@ -215,7 +217,9 @@ def deregister_on_start_span(self, func: Callable[[Span], None]) -> Callable[[Sp
 
         :param func: The function to stop calling when starting a span.
         """
-        core.reset_listeners("trace.span_start", callback=func)
+        handler = getattr(func, "_tracer_span_started_handler", None)
+        if handler:
+            tracer_span_started.remove(handler)
         return func
 
     def sample(self, span):
@@ -566,7 +570,7 @@ def _start_span(
             for p in chain(self._span_processors, SpanProcessor.__processors__, [self._span_aggregator]):
                 if p:
                     p.on_span_start(span)
-        core.dispatch("trace.span_start", (span,))
+        tracer_span_started.dispatch(span)
         return span
 
     start_span = _start_span
@@ -584,7 +588,7 @@ def _on_span_finish(self, span: Span) -> None:
                 if p:
                     p.on_span_finish(span)
 
-        core.dispatch("trace.span_finish", (span,))
+        tracer_span_finished.dispatch(span)
         log.debug("finishing span - %r (enabled:%s)", span, self.enabled)
 
     def _log_compat(self, level, msg):

ddtrace/internal/events.py

@@ -0,0 +1,36 @@
+"""
+Event Hub - High-performance event system for ddtrace
+
+This module provides a clean Python interface to the native event system,
+allowing both Python and Rust code to register listeners and fire events
+with minimal overhead.
+
+Usage:
+    from ddtrace.internal.events import trace_span_started
+
+    # Register a listener
+    def my_handler(span):
+        print(f"Span started: {span}")
+
+    handle = trace_span_started.on(my_handler)
+
+    # Fire an event
+    trace_span_started.dispatch(some_span)
+
+    # Remove listener
+    trace_span_started.remove(handle)
+
+    # Check if any listeners exist
+    if trace_span_started.has_listeners():
+        trace_span_started.dispatch(span)
+"""
+
+from ddtrace.internal.native._native import events as _events
+
+
+# Re-export everything from the native events submodule
+__all__ = _events.__all__
+
+# Dynamically add all event objects to this module's namespace
+for name in _events.__all__:
+    locals()[name] = getattr(_events, name)

ddtrace/internal/events.pyi

@@ -0,0 +1,39 @@
+# This file is automatically generated during compilation of src/native/
+# Do not edit manually - changes will be overwritten
+# Generated by build.rs from define_event! macro calls
+# Any modifications should be committed to the repository after regeneration
+# mypy: disable-error-code="name-defined"
+
+from typing import Callable, ParamSpec, Protocol
+
+P = ParamSpec('P')
+
+class ListenerHandle:
+    """Handle for registered event listeners"""
+    id: int
+    event_id: str
+    def __str__(self) -> str: ...
+    def __repr__(self) -> str: ...
+
+class Event(Protocol[P]):
+    """Generic event protocol with strongly typed dispatch and listeners"""
+    def dispatch(self, *args: P.args, **kwargs: P.kwargs) -> None:
+        """Fire the event to all registered listeners"""
+        ...
+
+    def on(self, callback: Callable[P, None]) -> ListenerHandle:
+        """Register a listener callback for this event"""
+        ...
+
+    def remove(self, handle: ListenerHandle) -> bool:
+        """Remove a specific listener by handle"""
+        ...
+
+    def has_listeners(self) -> bool:
+        """Check if any listeners are registered for this event"""
+        ...
+
+# Event instances
+
+tracer_span_started: Event[["ddtrace._trace.span.Span"]]  # noqa: F821
+tracer_span_finished: Event[["ddtrace._trace.span.Span"]]  # noqa: F821

src/native/Cargo.lock

@@ -16,6 +16,7 @@ profiling = ["dep:datadog-profiling-ffi"]
 
 [dependencies]
 anyhow = { version = "1.0", optional = true }
+ctor = "0.2"
 datadog-crashtracker = { git = "https://github.com/DataDog/libdatadog", rev = "v21.0.0", optional = true }
 datadog-ddsketch = { git = "https://github.com/DataDog/libdatadog", rev = "v21.0.0" }
 datadog-library-config = { git = "https://github.com/DataDog/libdatadog", rev = "v21.0.0" }
@@ -25,14 +26,18 @@ datadog-profiling-ffi = { git = "https://github.com/DataDog/libdatadog", rev = "
     "cbindgen",
 ] }
 ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "v21.0.0" }
+paste = "1.0"
 pyo3 = { version = "0.25", features = ["extension-module", "anyhow"] }
 tracing = { version = "0.1", default-features = false }
 
 [build-dependencies]
+glob = "0.3"
+proc-macro2 = "1.0"
 pyo3-build-config = "0.25"
 build_common = { git = "https://github.com/DataDog/libdatadog", rev = "v21.0.0", features = [
     "cbindgen",
 ] }
+syn = { version = "2.0", features = ["full", "extra-traits", "visit"] }
 
 [lib]
 name = "_native"