Merge branch 'main' into duncan-harvey/azure-eventhubs-integration

Author: duncanpharvey

.github/workflows/test_lib_injection.yml

@@ -30,7 +30,7 @@ jobs:
           - "3.11"
           - "3.12"
           - "3.13"
-          - "3.14.0rc3"
+          - "3.14"
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
         with:

.gitlab/benchmarks/bp-runner.microbenchmarks.fail-on-breach.yml

@@ -1067,7 +1067,7 @@ experiments:
               - max_rss_usage < 34.00 MB
           - name: span-start-finish-traceid128
             thresholds:
-              - execution_time < 55.00 ms
+              - execution_time < 56.00 ms
               - max_rss_usage < 34.00 MB
           - name: span-start-traceid128
             thresholds:

ddtrace/_trace/tracer.py

@@ -53,12 +53,14 @@
 from ddtrace.internal.runtime import get_runtime_id
 from ddtrace.internal.schema.processor import BaseServiceProcessor
 from ddtrace.internal.utils import _get_metas_to_propagate
+from ddtrace.internal.utils.deprecations import DDTraceDeprecationWarning
 from ddtrace.internal.utils.formats import format_trace_id
 from ddtrace.internal.writer import AgentWriterInterface
 from ddtrace.internal.writer import HTTPWriter
 from ddtrace.settings._config import config
 from ddtrace.settings.asm import config as asm_config
 from ddtrace.settings.peer_service import _ps_config
+from ddtrace.vendor.debtcollector.removals import remove
 from ddtrace.version import get_version
 
 
@@ -197,6 +199,11 @@ def _atexit(self) -> None:
         )
         self.shutdown(timeout=self.SHUTDOWN_TIMEOUT)
 
+    @remove(
+        message="on_start_span is being removed with no replacement",
+        removal_version="4.0.0",
+        category=DDTraceDeprecationWarning,
+    )
     def on_start_span(self, func: Callable[[Span], None]) -> Callable[[Span], None]:
         """Register a function to execute when a span start.
 
@@ -208,6 +215,11 @@ def on_start_span(self, func: Callable[[Span], None]) -> Callable[[Span], None]:
         core.on("trace.span_start", callback=func)
         return func
 
+    @remove(
+        message="deregister_on_start_span is being removed with no replacement",
+        removal_version="4.0.0",
+        category=DDTraceDeprecationWarning,
+    )
     def deregister_on_start_span(self, func: Callable[[Span], None]) -> Callable[[Span], None]:
         """Unregister a function registered to execute when a span starts.
 

ddtrace/appsec/_asm_request_context.py

@@ -2,6 +2,7 @@
 import json
 import re
 from types import TracebackType
+from typing import TYPE_CHECKING
 from typing import Any
 from typing import Callable
 from typing import Dict
@@ -17,8 +18,6 @@
 from ddtrace.appsec._constants import APPSEC
 from ddtrace.appsec._constants import SPAN_DATA_NAMES
 from ddtrace.appsec._constants import Constant_Class
-from ddtrace.appsec._utils import DDWaf_info
-from ddtrace.appsec._utils import DDWaf_result
 from ddtrace.appsec._utils import Telemetry_result
 from ddtrace.appsec._utils import get_triggers
 from ddtrace.contrib.internal.trace_utils_base import _normalize_tag_name
@@ -29,6 +28,10 @@
 from ddtrace.settings.asm import config as asm_config
 
 
+if TYPE_CHECKING:
+    from ddtrace.appsec._utils import DDWaf_info
+    from ddtrace.appsec._utils import DDWaf_result
+
 logger = ddlogger.get_logger(__name__)
 
 
@@ -95,7 +98,7 @@ def __init__(self, span: Optional[Span] = None, rc_products: str = ""):
         else:
             self.framework = self.span.name
         self.framework = self.framework.lower().replace(" ", "_")
-        self.waf_info: Optional[Callable[[], DDWaf_info]] = None
+        self.waf_info: Optional[Callable[[], "DDWaf_info"]] = None
         self.waf_addresses: Dict[str, Any] = {}
         self.callbacks: Dict[str, Any] = {_CONTEXT_CALL: []}
         self.telemetry: Telemetry_result = Telemetry_result()
@@ -375,15 +378,15 @@ def set_waf_callback(value) -> None:
     set_value(_CALLBACKS, _WAF_CALL, value)
 
 
-def set_waf_info(info: Callable[[], DDWaf_info]) -> None:
+def set_waf_info(info: Callable[[], "DDWaf_info"]) -> None:
     env = _get_asm_context()
     if env is None:
         logger.warning(WARNING_TAGS.SET_WAF_INFO_NO_ASM_CONTEXT, extra=log_extra, stack_info=True)
         return
     env.waf_info = info
 
 
-def call_waf_callback(custom_data: Optional[Dict[str, Any]] = None, **kwargs) -> Optional[DDWaf_result]:
+def call_waf_callback(custom_data: Optional[Dict[str, Any]] = None, **kwargs) -> Optional["DDWaf_result"]:
     if not asm_config._asm_enabled:
         return None
     callback = get_value(_CALLBACKS, _WAF_CALL)
@@ -480,7 +483,7 @@ def asm_request_context_set(
 def set_waf_telemetry_results(
     rules_version: str,
     is_blocked: bool,
-    waf_results: DDWaf_result,
+    waf_results: "DDWaf_result",
     rule_type: Optional[str],
     is_sampled: bool,
 ) -> None:

ddtrace/appsec/_utils.py

@@ -10,7 +10,6 @@
 from typing import List
 from typing import Optional
 from typing import Union
-import uuid
 
 from ddtrace.appsec._constants import API_SECURITY
 from ddtrace.appsec._constants import APPSEC
@@ -236,6 +235,10 @@ def _safe_userid(user_id):
         return user_id
     except ValueError:
         try:
+            # Import uuid lazily because this also imports threading via the
+            # platform module
+            import uuid
+
             _ = uuid.UUID(user_id)
             return user_id
         except ValueError:

ddtrace/bootstrap/sitecustomize.py

@@ -55,6 +55,11 @@ def drop(module_name):
     if not asbool(do_cleanup):
         return
 
+    # We need to import these modules to make sure they grab references to the
+    # right modules before we start unloading stuff.
+    import ddtrace.internal.http  # noqa
+    import ddtrace.internal.uds  # noqa
+
     # Unload all the modules that we have imported, except for the ddtrace one.
     # NB: this means that every `import threading` anywhere in `ddtrace/` code
     # uses a copy of that module that is distinct from the copy that user code
@@ -94,6 +99,10 @@ def drop(module_name):
             # submodule makes use of threading so it is critical to unload when
             # gevent is used.
             "concurrent.futures",
+            # We unload the threading module in case it was imported by
+            # CPython on boot.
+            "threading",
+            "_thread",
         ]
     )
     for u in UNLOAD_MODULES:

ddtrace/internal/coverage/code.py

@@ -14,7 +14,6 @@
 from ddtrace.internal.coverage.util import collapse_ranges
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.module import ModuleWatchdog
-from ddtrace.internal.packages import is_user_code
 from ddtrace.internal.packages import platlib_path
 from ddtrace.internal.packages import platstdlib_path
 from ddtrace.internal.packages import purelib_path
@@ -336,9 +335,6 @@ def transform(self, code: CodeType, _module: ModuleType) -> CodeType:
             # Don't instrument code from standard library/site packages/etc.
             return code
 
-        if not is_user_code(code_path):
-            return code
-
         retval = self.instrument_code(code, _module.__package__ if _module is not None else "")
 
         if self._collect_import_coverage:

ddtrace/internal/ffande/__init__.py

@@ -0,0 +1,6 @@
+"""
+Feature Flagging and Experimentation (FFAndE) product module.
+
+This module handles Feature Flag configuration rules from Remote Configuration
+and forwards the raw bytes to the native FFAndE processor.
+"""

ddtrace/internal/ffande/_native.py

@@ -0,0 +1,51 @@
+"""
+Native interface for FFAndE (Feature Flagging and Experimentation) processing.
+
+This module provides the interface to the PyO3 native function that processes
+feature flag configuration rules.
+"""
+from typing import Optional
+
+from ddtrace.internal.logger import get_logger
+
+
+log = get_logger(__name__)
+
+is_available = True
+
+try:
+    from ddtrace.internal.native._native import ffande_process_config
+except ImportError:
+    is_available = False
+    log.debug("FFAndE native module not available, feature flag processing disabled")
+
+    # Provide a no-op fallback
+    def ffande_process_config(config_bytes: bytes) -> Optional[bool]:
+        """Fallback implementation when native module is not available."""
+        log.warning("FFAndE native module not available, ignoring configuration")
+        return None
+
+
+def process_ffe_configuration(config_bytes: bytes) -> bool:
+    """
+    Process feature flag configuration by forwarding raw bytes to native function.
+
+    Args:
+        config_bytes: Raw bytes from Remote Configuration payload
+
+    Returns:
+        True if processing was successful, False otherwise
+    """
+    if not is_available:
+        log.debug("FFAndE native module not available, skipping configuration")
+        return False
+
+    try:
+        result = ffande_process_config(config_bytes)
+        if result is None:
+            log.debug("FFAndE native processing returned None")
+            return False
+        return result
+    except Exception as e:
+        log.error("Error processing FFE configuration: %s", e, exc_info=True)
+        return False

ddtrace/internal/ffande/product.py

@@ -0,0 +1,145 @@
+"""
+FFAndE (Feature Flagging and Experimentation) product implementation.
+
+This product receives feature flag configuration rules from Remote Configuration
+and processes them through the native FFAndE processor.
+"""
+import enum
+import json
+import typing as t
+
+from ddtrace import config
+from ddtrace.internal.ffande._native import process_ffe_configuration
+from ddtrace.internal.logger import get_logger
+from ddtrace.internal.remoteconfig import Payload
+from ddtrace.internal.remoteconfig._connectors import PublisherSubscriberConnector
+from ddtrace.internal.remoteconfig._publishers import RemoteConfigPublisher
+from ddtrace.internal.remoteconfig._pubsub import PubSub
+from ddtrace.internal.remoteconfig._subscribers import RemoteConfigSubscriber
+
+
+requires = ["remote-configuration"]
+
+log = get_logger(__name__)
+
+# FFAndE product name constant
+FFE_FLAGS_PRODUCT = "FFE_FLAGS"
+
+
+class FFAndECapabilities(enum.IntFlag):
+    """FFAndE Remote Configuration capabilities."""
+
+    FFE_FLAG_CONFIGURATION_RULES = 1 << 46
+
+
+class FFAndEAdapter(PubSub):
+    """
+    FFAndE Remote Configuration adapter.
+
+    Receives feature flag configuration rules and forwards raw bytes to native processor.
+    """
+
+    __publisher_class__ = RemoteConfigPublisher
+    __subscriber_class__ = RemoteConfigSubscriber
+    __shared_data__ = PublisherSubscriberConnector()
+
+    def __init__(self):
+        self._publisher = self.__publisher_class__(self.__shared_data__)
+        self._subscriber = self.__subscriber_class__(self.__shared_data__, self.rc_callback, FFE_FLAGS_PRODUCT)
+
+        # Register with Remote Configuration poller
+        if config._remote_config_enabled:
+            from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
+
+            remoteconfig_poller.register(
+                FFE_FLAGS_PRODUCT,
+                self,
+                restart_on_fork=True,
+                capabilities=[FFAndECapabilities.FFE_FLAG_CONFIGURATION_RULES],
+            )
+
+    def rc_callback(self, payloads: t.Sequence[Payload]) -> None:
+        """
+        Process FFE configuration payloads from Remote Configuration.
+
+        Args:
+            payloads: Sequence of configuration payloads
+        """
+        for payload in payloads:
+            if payload.metadata is None:
+                log.debug("Ignoring invalid FFE payload with no metadata, path: %s", payload.path)
+                continue
+
+            log.debug("Received FFE config payload: %s", payload.metadata.id)
+
+            if payload.content is None:
+                log.debug(
+                    "Received FFE config deletion, product: %s, path: %s",
+                    payload.metadata.product_name,
+                    payload.path,
+                )
+                # Handle deletion/removal of configuration
+                continue
+
+            try:
+                # Serialize payload content to bytes for native processing
+                # The native function expects raw bytes, so we convert the dict to JSON
+                config_json = json.dumps(payload.content, ensure_ascii=False)
+                config_bytes = config_json.encode("utf-8")
+
+                log.debug("Processing FFE config ID: %s, size: %d bytes", payload.metadata.id, len(config_bytes))
+
+                success = process_ffe_configuration(config_bytes)
+                if success:
+                    log.debug("Successfully processed FFE config ID: %s", payload.metadata.id)
+                else:
+                    log.warning("Failed to process FFE config ID: %s", payload.metadata.id)
+
+            except Exception as e:
+                log.error("Error processing FFE config payload: %s", e, exc_info=True)
+
+
+_adapter: t.Optional[FFAndEAdapter] = None
+
+
+def post_preload():
+    """Called during preload phase."""
+    pass
+
+
+def start():
+    """Start the FFAndE product and register with Remote Configuration."""
+    global _adapter
+
+    if not config._remote_config_enabled:
+        log.debug("Remote configuration disabled, FFAndE not started")
+        return
+
+    _adapter = FFAndEAdapter()
+
+    log.debug("FFAndE product registered with Remote Configuration")
+
+
+def restart(join=False):
+    """Restart the FFAndE product."""
+    pass
+
+
+def stop(join=False):
+    """Stop the FFAndE product."""
+    global _adapter
+
+    if not config._remote_config_enabled:
+        return
+
+    from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
+
+    remoteconfig_poller.unregister(FFE_FLAGS_PRODUCT)
+    _adapter = None
+
+    log.debug("FFAndE product unregistered from Remote Configuration")
+
+
+def at_exit(join=False):
+    """Called at exit."""
+    stop(join=join)