chore(debugger): implement Phase 1 guardrail foundations

ddtrace/debugging/_probe/model.py

@@ -139,6 +139,8 @@ class ProbeConditionMixin(AbstractProbeMixIn):
     condition: Optional[DDExpression]
     condition_error_rate: float = field(compare=False)
     condition_error_limiter: RateLimiter = field(init=False, repr=False, compare=False)
+    # monotonic timestamp before which evaluation should be skipped at probe entry (RFC: probe-entry skip)
+    _error_throttled_until: float = field(default=0.0, init=False, repr=False, compare=False)
 
     def __post_init__(self) -> None:
         super().__post_init__()

ddtrace/debugging/_signal/collector.py

@@ -8,6 +8,7 @@
 from ddtrace.debugging._signal.model import Signal
 from ddtrace.debugging._signal.model import SignalState
 from ddtrace.debugging._signal.model import SignalTrack
+from ddtrace.debugging._signal.snapshot import Snapshot
 from ddtrace.internal._encoding import BufferFull
 from ddtrace.internal.compat import ExcInfoType
 from ddtrace.internal.logger import get_logger
@@ -42,22 +43,69 @@ def _enqueue(self, log_signal: LogSignal) -> None:
             self._tracks[log_signal.__track__].put(log_signal)
         except BufferFull:
             log.debug("Encoder buffer full")
-            meter.increment("encoder.buffer.full")
+            meter.increment(
+                "dynamic_instrumentation.guardrails.events.dropped",
+                tags={"reason": "queueFull", "event_type": "snapshot"},
+            )
         except KeyError:
             log.error("No encoder for signal track %s", log_signal.__track__)
 
     def push(self, signal: Signal) -> None:
         if signal.state is SignalState.SKIP_COND:
-            meter.increment("skip", tags={"cause": "cond", "probe_id": signal.probe.probe_id})
-        elif signal.state in {SignalState.SKIP_COND_ERROR, SignalState.COND_ERROR}:
-            meter.increment("skip", tags={"cause": "cond_error", "probe_id": signal.probe.probe_id})
-        elif signal.state is SignalState.SKIP_RATE:
-            meter.increment("skip", tags={"cause": "rate", "probe_id": signal.probe.probe_id})
+            # Condition evaluated to False — not a guardrail event, no metric
+            pass
+        elif signal.state is SignalState.SKIP_COND_ERROR:
+            meter.increment(
+                "dynamic_instrumentation.guardrails.events.skipped",
+                tags={"reason": "evaluationErrorThrottled", "probe_type": type(signal.probe).__name__},
+            )
+        elif signal.state is SignalState.COND_ERROR:
+            meter.increment(
+                "dynamic_instrumentation.guardrails.evaluation.errors",
+                tags={"probe_type": type(signal.probe).__name__, "error_kind": "condition"},
+            )
+        elif signal.state is SignalState.SKIP_RATE_GLOBAL:
+            meter.increment(
+                "dynamic_instrumentation.guardrails.events.skipped",
+                tags={"reason": "rateLimitGlobal", "probe_type": type(signal.probe).__name__},
+            )
+        elif signal.state is SignalState.SKIP_RATE_PROBE:
+            meter.increment(
+                "dynamic_instrumentation.guardrails.events.skipped",
+                tags={"reason": "rateLimitProbe", "probe_type": type(signal.probe).__name__},
+            )
         elif signal.state is SignalState.SKIP_BUDGET:
-            meter.increment("skip", tags={"cause": "budget", "probe_id": signal.probe.probe_id})
+            meter.increment(
+                "dynamic_instrumentation.guardrails.events.skipped",
+                tags={"reason": "budgetExceededInvocation", "probe_type": type(signal.probe).__name__},
+            )
         elif signal.state is SignalState.DONE:
             meter.increment("signal", tags={"probe_id": signal.probe.probe_id})
 
+        # Emit evaluation duration if measured
+        if signal._eval_duration_ms is not None:
+            meter.distribution(
+                "dynamic_instrumentation.guardrails.evaluation.duration",
+                signal._eval_duration_ms,
+                tags={"probe_type": type(signal.probe).__name__, "evaluation_kind": "condition"},
+            )
+
+        # Emit capture and template evaluation durations for snapshots
+        if isinstance(signal, Snapshot):
+            if signal._template_eval_duration_ms is not None:
+                meter.distribution(
+                    "dynamic_instrumentation.guardrails.evaluation.duration",
+                    signal._template_eval_duration_ms,
+                    tags={"probe_type": type(signal.probe).__name__, "evaluation_kind": "template"},
+                )
+            if signal._capture_duration_ms is not None:
+                truncated = "true" if signal.errors else "false"
+                meter.distribution(
+                    "dynamic_instrumentation.guardrails.capture.duration",
+                    signal._capture_duration_ms,
+                    tags={"probe_type": type(signal.probe).__name__, "truncated": truncated},
+                )
+
         if (
             isinstance(signal, LogSignal)
             and signal.state in {SignalState.DONE, SignalState.COND_ERROR}

ddtrace/debugging/_signal/model.py

@@ -28,6 +28,8 @@
 from ddtrace.internal.metrics import Metrics
 from ddtrace.internal.rate_limiter import BudgetRateLimiterWithJitter as RateLimiter
 from ddtrace.internal.rate_limiter import RateLimitExceeded
+from ddtrace.internal.settings.dynamic_instrumentation import config as di_config
+from ddtrace.internal.utils.time import Time
 from ddtrace.trace import Context
 from ddtrace.trace import Span
 
@@ -42,7 +44,8 @@ class SignalState(str, Enum):
     NONE = "NONE"
     SKIP_COND = "SKIP_COND"
     SKIP_COND_ERROR = "SKIP_COND_ERROR"
-    SKIP_RATE = "SKIP_RATE"
+    SKIP_RATE_GLOBAL = "SKIP_RATE_GLOBAL"
+    SKIP_RATE_PROBE = "SKIP_RATE_PROBE"
     SKIP_BUDGET = "SKIP_BUDGET"
     COND_ERROR = "COND_ERROR"
     DONE = "DONE"
@@ -86,6 +89,7 @@ class Signal(abc.ABC):
     errors: list[EvaluationError] = field(default_factory=list)
     timestamp: float = field(default_factory=time.time)
     uuid: str = field(default_factory=lambda: str(uuid4()), init=False)
+    _eval_duration_ms: Optional[float] = field(default=None, init=False, repr=False)
 
     def __post_init__(self) -> None:
         probe = self.probe
@@ -106,19 +110,47 @@ def _eval_condition(self, scope: Mapping[str, Any]) -> bool:
         if condition is None:
             return True
 
+        # RFC: skip at probe entry when error rate has been exceeded previously
+        if Time.monotonic() < probe._error_throttled_until:
+            self.state = SignalState.SKIP_COND_ERROR
+            return False
+
+        t_start = Time.monotonic()
         try:
-            if bool(condition.eval(scope)):
-                return True
+            result = bool(condition.eval(scope))
         except DDExpressionEvaluationError as e:
             self.errors.append(EvaluationError(expr=e.dsl, message=e.error))
-            self.state = (
-                SignalState.SKIP_COND_ERROR
-                if probe.condition_error_limiter.limit() is RateLimitExceeded
-                else SignalState.COND_ERROR
+            if probe.condition_error_limiter.limit() is RateLimitExceeded:
+                probe._error_throttled_until = Time.monotonic() + probe.condition_error_limiter.tau
+                self.state = SignalState.SKIP_COND_ERROR
+            else:
+                self.state = SignalState.COND_ERROR
+            return False
+        finally:
+            self._eval_duration_ms = (Time.monotonic() - t_start) * 1000
+
+        # RFC: treat evaluation overruns as guardrail input — feed into the error throttle
+        eval_timeout_ms = di_config.evaluation_timeout_ms
+        if self._eval_duration_ms > eval_timeout_ms:
+            self.errors.append(
+                EvaluationError(
+                    expr=condition.dsl,
+                    message=(
+                        f"Condition evaluation exceeded budget: {self._eval_duration_ms:.1f}ms > {eval_timeout_ms}ms"
+                    ),
+                )
             )
-        else:
-            self.state = SignalState.SKIP_COND
+            if probe.condition_error_limiter.limit() is RateLimitExceeded:
+                probe._error_throttled_until = Time.monotonic() + probe.condition_error_limiter.tau
+                self.state = SignalState.SKIP_COND_ERROR
+                return False
+            self.state = SignalState.COND_ERROR
+            return False
+
+        if result:
+            return True
 
+        self.state = SignalState.SKIP_COND
         return False
 
     def _rate_limit_exceeded(self) -> bool:
@@ -130,7 +162,7 @@ def _rate_limit_exceeded(self) -> bool:
 
         exceeded = self.session is None and probe.limiter.limit() is RateLimitExceeded
         if exceeded:
-            self.state = SignalState.SKIP_RATE
+            self.state = SignalState.SKIP_RATE_PROBE
 
         return exceeded
 
@@ -239,7 +271,7 @@ def do_line(self, global_limiter: Optional[RateLimiter] = None) -> None:
             return
 
         if global_limiter is not None and global_limiter.limit() is RateLimitExceeded:
-            self.state = SignalState.SKIP_RATE
+            self.state = SignalState.SKIP_RATE_GLOBAL
             return
 
         if self._rate_limit_exceeded():

ddtrace/debugging/_signal/snapshot.py

@@ -36,10 +36,9 @@
 from ddtrace.internal.compat import NO_EXCEPTION
 from ddtrace.internal.compat import ExcInfoType
 from ddtrace.internal.metrics import Metrics
+from ddtrace.internal.settings.dynamic_instrumentation import config as di_config
 from ddtrace.internal.utils.time import HourGlass
-
-
-CAPTURE_TIME_BUDGET = 0.2  # seconds
+from ddtrace.internal.utils.time import Time
 
 
 _NOTSET = object()
@@ -54,7 +53,7 @@ def _capture_context(
     retval: Any = _NOTSET,
     limits: CaptureLimits = DEFAULT_CAPTURE_LIMITS,
 ) -> dict[str, Any]:
-    with HourGlass(duration=CAPTURE_TIME_BUDGET) as hg:
+    with HourGlass(duration=di_config.capture_timeout_ms / 1000) as hg:
 
         def timeout(_: Any) -> bool:
             return not hg.trickling()
@@ -93,7 +92,7 @@ def _capture_expressions(
     exprs: list[CaptureExpression],
     scope: Mapping[str, Any],
 ) -> dict[str, Any]:
-    with HourGlass(duration=CAPTURE_TIME_BUDGET) as hg:
+    with HourGlass(duration=di_config.capture_timeout_ms / 1000) as hg:
 
         def timeout(_: Any) -> bool:
             return not hg.trickling()
@@ -131,6 +130,8 @@ class Snapshot(LogSignal):
     _stack: Optional[list[dict[str, Any]]] = field(default=None)
     _message: Optional[str] = field(default=None)
     duration: Optional[int] = field(default=None)  # nanoseconds
+    _capture_duration_ms: Optional[float] = field(default=None, init=False, repr=False)
+    _template_eval_duration_ms: Optional[float] = field(default=None, init=False, repr=False)
 
     def _eval_segment(self, segment: TemplateSegment, _locals: Mapping[str, Any]) -> str:
         probe = cast(LogProbeMixin, self.probe)
@@ -151,7 +152,21 @@ def _eval_segment(self, segment: TemplateSegment, _locals: Mapping[str, Any]) ->
 
     def _eval_message(self, _locals: Mapping[str, Any]) -> None:
         probe = cast(LogProbeMixin, self.probe)
+        t_start = Time.monotonic()
         self._message = "".join([self._eval_segment(s, _locals) for s in probe.segments])
+        self._template_eval_duration_ms = (Time.monotonic() - t_start) * 1000
+        # RFC: treat overruns as guardrail input — record as evaluation error
+        eval_timeout_ms = di_config.evaluation_timeout_ms
+        if self._template_eval_duration_ms > eval_timeout_ms:
+            self.errors.append(
+                EvaluationError(
+                    expr=probe.template,
+                    message=(
+                        f"Template evaluation exceeded budget: "
+                        f"{self._template_eval_duration_ms:.1f}ms > {eval_timeout_ms}ms"
+                    ),
+                )
+            )
 
     def _do(self, retval: Any, exc_info: ExcInfoType, scope: Mapping[str, Any]) -> Optional[dict[str, Any]]:
         probe = cast(LogProbeMixin, self.probe)
@@ -162,10 +177,16 @@ def _do(self, retval: Any, exc_info: ExcInfoType, scope: Mapping[str, Any]) -> O
         self._stack = utils.capture_stack(self.frame)
 
         if probe.take_snapshot:
-            return _capture_context(frame, exc_info, retval=retval, limits=probe.limits)
+            t_start = Time.monotonic()
+            result = _capture_context(frame, exc_info, retval=retval, limits=probe.limits)
+            self._capture_duration_ms = (Time.monotonic() - t_start) * 1000
+            return result
 
         if probe.capture_expressions:
-            return _capture_expressions(probe.capture_expressions, scope)
+            t_start = Time.monotonic()
+            result = _capture_expressions(probe.capture_expressions, scope)
+            self._capture_duration_ms = (Time.monotonic() - t_start) * 1000
+            return result
 
         return None
 

ddtrace/internal/settings/_supported_configurations.py

@@ -197,8 +197,10 @@
         "DD_DOWNLOAD_MAX_RETRIES",
         "DD_DRAMATIQ_SERVICE",
         "DD_DURABLE_CROSS_INVOCATION_TRACING_ENABLED",
+        "DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS",
         "DD_DYNAMIC_INSTRUMENTATION_DIAGNOSTICS_INTERVAL",
         "DD_DYNAMIC_INSTRUMENTATION_ENABLED",
+        "DD_DYNAMIC_INSTRUMENTATION_EVALUATION_TIMEOUT_MS",
         "DD_DYNAMIC_INSTRUMENTATION_MAX_PAYLOAD_SIZE",
         "DD_DYNAMIC_INSTRUMENTATION_METRICS_ENABLED",
         "DD_DYNAMIC_INSTRUMENTATION_PROBE_FILE",

ddtrace/internal/settings/dynamic_instrumentation.py

@@ -142,6 +142,22 @@ class DynamicInstrumentationConfig(DDConfig):
         help="List of identifiers to exclude from redaction",
     )
 
+    capture_timeout_ms = DDConfig.v(
+        int,
+        "capture.timeout_ms",
+        default=150,
+        help_type="Integer",
+        help="Maximum wall-time in milliseconds for snapshot capture and serialization per probe invocation",
+    )
+
+    evaluation_timeout_ms = DDConfig.v(
+        int,
+        "evaluation.timeout_ms",
+        default=50,
+        help_type="Integer",
+        help="Maximum wall-time in milliseconds for condition and template expression evaluation per probe invocation",
+    )
+
     probe_file = DDConfig.v(
         t.Optional[Path],
         "probe_file",

supported-configurations.json

@@ -1468,6 +1468,13 @@
         ]
       }
     ],
+    "DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS": [
+      {
+        "implementation": "C",
+        "type": "int",
+        "default": "150"
+      }
+    ],
     "DD_DYNAMIC_INSTRUMENTATION_DIAGNOSTICS_INTERVAL": [
       {
         "implementation": "A",
@@ -1482,6 +1489,13 @@
         "default": "false"
       }
     ],
+    "DD_DYNAMIC_INSTRUMENTATION_EVALUATION_TIMEOUT_MS": [
+      {
+        "implementation": "A",
+        "type": "int",
+        "default": "50"
+      }
+    ],
     "DD_DYNAMIC_INSTRUMENTATION_MAX_PAYLOAD_SIZE": [
       {
         "implementation": "B",