feat(openfeature): parse observeFullEvaluationData and hash targeting_key in flagevaluations events

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

@@ -537,6 +537,32 @@ experiments:
             thresholds:
               - execution_time < 38.50 ms
 
+          # openfeatureflagevaluation
+          - name: openfeatureflagevaluation-aggregate-scale
+            thresholds:
+              - execution_time < 0.064 ms
+          - name: openfeatureflagevaluation-aggregate-stress
+            thresholds:
+              - execution_time < 0.550 ms
+          - name: openfeatureflagevaluation-aggregate-typical
+            thresholds:
+              - execution_time < 0.029 ms
+          - name: openfeatureflagevaluation-hook-enqueue-bounded-adversarial
+            thresholds:
+              - execution_time < 0.380 ms
+          - name: openfeatureflagevaluation-hook-enqueue-scale
+            thresholds:
+              - execution_time < 0.043 ms
+          - name: openfeatureflagevaluation-hook-enqueue-stress
+            thresholds:
+              - execution_time < 0.300 ms
+          - name: openfeatureflagevaluation-hook-enqueue-typical
+            thresholds:
+              - execution_time < 0.033 ms
+          - name: openfeatureflagevaluation-hook-plus-drain-scale
+            thresholds:
+              - execution_time < 0.120 ms
+
           # otelsdkspan
           - name: otelsdkspan-add-event
             thresholds:

benchmarks/openfeature_flagevaluation/config.yaml

@@ -3,7 +3,7 @@ defaults: &defaults
   num_users: 50
   num_context_fields: 10
 
-# Eval hot path the caller pays: finally_after hook (cheap capture + non-blocking enqueue).
+# Eval hot path the caller pays: bounded immutable snapshot + non-blocking enqueue.
 hook-enqueue-typical:
   <<: *defaults
   mode: hook_enqueue
@@ -22,6 +22,12 @@ hook-enqueue-scale:
   num_users: 500
   num_context_fields: 20
 
+# Worst-case bounded walk: every list value is inspected then discarded, plus lookahead.
+hook-enqueue-bounded-adversarial:
+  <<: *defaults
+  mode: hook_enqueue_adversarial
+  num_context_fields: 257
+
 # Background worker hot path: canonical key + two-tier aggregation.
 aggregate-typical:
   <<: *defaults

benchmarks/openfeature_flagevaluation/scenario.py

@@ -45,7 +45,10 @@ def _make_details(flag_key, variant, allocation_key):
         value=True,
         variant=variant,
         reason=Reason.TARGETING_MATCH,
-        flag_metadata={"allocation_key": allocation_key},
+        flag_metadata={
+            "allocation_key": allocation_key,
+            "__dd_observe_full_evaluation_data": True,
+        },
     )
 
 
@@ -69,7 +72,10 @@ def run(self):
         num_fields = max(0, self.num_context_fields)
         cycle_count = max(num_flags, num_users)
 
-        attrs = {"attr_{}".format(i): "value_{}".format(i) for i in range(num_fields)}
+        if mode == "hook_enqueue_adversarial":
+            attrs = {"discarded": ["x" * 257 for _ in range(num_fields)]}
+        else:
+            attrs = {"attr_{}".format(i): "value_{}".format(i) for i in range(num_fields)}
         flag_keys = ["flag-{}".format(i) for i in range(num_flags)]
         targeting_keys = ["user-{}".format(i) for i in range(num_users)]
         hook_contexts = [
@@ -92,7 +98,7 @@ def run(self):
         writer = FlagEvaluationWriter(interval=3600.0)
         hook = FlagEvalEVPHook(writer)
 
-        if mode == "hook_enqueue":
+        if mode in ("hook_enqueue", "hook_enqueue_adversarial"):
             # Preserve Queue.put_nowait locking and notification overhead without
             # periodically draining and aggregating events in this isolated mode.
             writer._queue = _DiscardingQueue(maxsize=writer._queue.maxsize)
@@ -108,17 +114,19 @@ def _(loops):
             from ddtrace.internal.openfeature._flagevaluation_writer import _EvalEvent
             from ddtrace.internal.openfeature._flagevaluation_writer import flatten_and_prune_context
 
-            bounded_attrs = flatten_and_prune_context(attrs)
+            bounded_attrs, truncation_reasons = flatten_and_prune_context(attrs)
+            assert not truncation_reasons
             events = [
                 _EvalEvent(
                     flag_key=flag_keys[i % num_flags],
                     variant="variant-{}".format(i % 4),
                     allocation_key="alloc-{}".format(i % num_flags),
                     targeting_key=targeting_keys[i % num_users],
-                    attrs=dict(bounded_attrs),
+                    attrs=bounded_attrs,
                     runtime_default=False,
                     error_message="",
                     eval_time_ms=1_760_000_000_000 + i,
+                    observe_full_evaluation_data=True,
                 )
                 for i in range(cycle_count)
             ]
@@ -137,17 +145,28 @@ def _(loops):
             yield _
 
         elif mode == "hook_plus_drain":
+            # One complete scale cycle must fit without using queue state to decide
+            # when the benchmark drains. enqueue() already checks backpressure.
+            assert cycle_count <= writer._queue.maxsize
+
+            def _drain_and_reset() -> None:
+                writer._drain_queue()
+                writer._full.clear()
+                writer._degraded.clear()
+                writer._per_flag_count.clear()
+                writer._global_count = 0
 
             def _(loops):
                 for i in range(loops):
                     idx = i % cycle_count
                     hook.finally_after(hook_contexts[idx], details_list[idx], {})
-                    if writer._queue.full() or (i % cycle_count) == (cycle_count - 1):
-                        writer._drain_queue()
-                        writer._full.clear()
-                        writer._degraded.clear()
-                        writer._per_flag_count.clear()
-                        writer._global_count = 0
+                    if (i + 1) % cycle_count == 0:
+                        _drain_and_reset()
+
+                # pyperf can calibrate a loop count that is not a whole scale cycle.
+                # Drain the remainder so every measured enqueue is also aggregated.
+                if loops % cycle_count:
+                    _drain_and_reset()
 
             yield _
 

benchmarks/suitespec.yml

@@ -386,3 +386,11 @@ suites:
       - benchmarks/suitespec.yml
     cpus_per_run: 1
     type: 'microbenchmark'
+  openfeature_flagevaluation:
+    paths:
+      - ddtrace/internal/openfeature/_flag_eval_evp_hook.py
+      - ddtrace/internal/openfeature/_flagevaluation_writer.py
+      - benchmarks/openfeature_flagevaluation/*
+      - benchmarks/suitespec.yml
+    cpus_per_run: 1
+    type: 'microbenchmark'

ddtrace/internal/openfeature/_config.py

@@ -1,10 +1,26 @@
+"""
+Module-level storage for the parsed FFE configuration and its consent value.
+
+The two values are bundled in one NamedTuple with a single module-level
+reference so a reader always sees a consistent (config, consent) pair. This
+closes the consent-lifecycle race described in the Java pilot's review
+(concern:bind-consent-to-evaluated-config).
+"""
+
 from typing import Any
+from typing import NamedTuple
 from typing import Optional
+from typing import Union
 
 from ddtrace.internal.native._native import ffe
 
 
-FFE_CONFIG: Optional[ffe.Configuration] = None
+class _FfeSnapshot(NamedTuple):
+    """Atomic bundle of native config and the consent value read off the UFC."""
+
+    config: ffe.Configuration
+    observe_full_evaluation_data: bool
+
 
 # Registry of provider instances (ddtrace.internal.openfeature._provider.DataDogProvider)
 # that need to be notified when new FFE configuration arrives. Kept here rather than in
@@ -15,16 +31,42 @@
 _provider_instances: list[Any] = []
 
 
-def _get_ffe_config():
-    """Retrieve the current FFE configuration."""
-    return FFE_CONFIG
+# Module-level global. Reads and writes are done through the accessors below so
+# callers only ever observe a consistent snapshot.
+_FFE_SNAPSHOT: Optional[_FfeSnapshot] = None
+
+
+# AIDEV-NOTE: Preserved for existing callers only. New evaluation paths must
+# use _get_ffe_snapshot() so consent is observed atomically with the exact
+# configuration being evaluated. Reading only the config and then consulting
+# live consent downstream reintroduces the consent-lifecycle race.
+def _get_ffe_config() -> Optional[ffe.Configuration]:
+    """Retrieve just the native FFE configuration. Preserved for compatibility."""
+    snap = _FFE_SNAPSHOT
+    return snap.config if snap is not None else None
+
+
+def _get_ffe_snapshot() -> Optional[_FfeSnapshot]:
+    """Retrieve the full snapshot (config + consent)."""
+    return _FFE_SNAPSHOT
+
+
+def _set_ffe_config(value: Union[None, ffe.Configuration, _FfeSnapshot]) -> None:
+    """Set the FFE snapshot and notify registered providers.
 
+    Accepts either a bare native Configuration (existing test callers) or a
+    _FfeSnapshot. A bare Configuration is stored as consent-off; None clears.
+    """
+    global _FFE_SNAPSHOT
+    if value is None:
+        _FFE_SNAPSHOT = None
+    elif isinstance(value, _FfeSnapshot):
+        _FFE_SNAPSHOT = value
+    else:
+        # Legacy path: a raw Configuration means consent-off (fail closed).
+        _FFE_SNAPSHOT = _FfeSnapshot(config=value, observe_full_evaluation_data=False)
 
-def _set_ffe_config(config):
-    """Set the FFE configuration and notify registered providers."""
-    global FFE_CONFIG
-    FFE_CONFIG = config
-    if config is not None:
+    if _FFE_SNAPSHOT is not None:
         _notify_providers_config_received()
 
 

ddtrace/internal/openfeature/_flag_eval_evp_hook.py

@@ -2,7 +2,7 @@
 FlagEvalEVPHook — OpenFeature `finally_after` hook for EVP flagevaluation emission.
 
 Hook design:
-- Cheap capture only in finally_after (no aggregation, no serialization, no I/O).
+- One bounded context snapshot in finally_after (no aggregation or I/O).
 - Non-blocking enqueue to FlagEvaluationWriter.
 - The finally_after stage covers success, error, and default eval paths.
 - Does NOT replace or modify the OTel FlagEvalMetricsHook in _flageval_metrics.py (the existing
@@ -20,6 +20,7 @@
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_TIMESTAMP_METADATA_KEY
 from ddtrace.internal.openfeature._flagevaluation_writer import METADATA_ALLOCATION_KEY
+from ddtrace.internal.openfeature._flagevaluation_writer import METADATA_OBSERVE_FULL_EVALUATION_DATA
 from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter
 from ddtrace.internal.openfeature._flagevaluation_writer import _EvalEvent
 
@@ -29,11 +30,11 @@
 
 class FlagEvalEVPHook(Hook):
     """
-    OpenFeature Hook that enqueues cheap evaluation snapshots for EVP aggregation.
+    OpenFeature Hook that enqueues bounded evaluation snapshots for EVP aggregation.
 
     Implements `finally_after` (covers the success, error, and default eval paths).
-    Does NO aggregation, serialization, or I/O on the hook thread.  All heavy work
-    is deferred to FlagEvaluationWriter's background periodic worker.
+    Context bounding happens synchronously before queue insertion; aggregation and I/O
+    remain deferred to FlagEvaluationWriter's background periodic worker.
     """
 
     def __init__(self, writer: FlagEvaluationWriter) -> None:
@@ -46,25 +47,27 @@ def finally_after(
         hints: HookHints,
     ) -> None:
         """
-        Cheap capture + non-blocking enqueue.
+        Bounded capture + non-blocking enqueue.
 
-        Extracts only scalar fields (no allocation, no serialization) and hands
-        a _EvalEvent snapshot to FlagEvaluationWriter.enqueue() which is
-        non-blocking (queue.Queue.put_nowait — drops on queue.Full).
+        Extracts scalar fields and hands the caller context to
+        FlagEvaluationWriter.enqueue(), which performs an O(1) queue-full precheck and
+        then creates the bounded immutable snapshot before queue.Queue.put_nowait.
 
         Eval-time: uses details.flag_metadata["dd.eval.timestamp_ms"] when present
         (stamped by the provider at eval entry); falls back to hook-fire time.
 
         Runtime-default: True when the variant is absent (details.variant is None).
 
-        Attrs: shallow copy of the evaluation context attributes dict so the hook
-        returns immediately and the worker can safely iterate attrs off-path.
+        Attrs: borrowed only for the synchronous enqueue call. The queue never retains
+        this caller-owned mapping or any mutable context leaf.
         """
         try:
             flag_key: str = hook_context.flag_key or ""
 
             # Extract allocation_key from flag_metadata (same key as METADATA_ALLOCATION_KEY).
-            metadata: typing.Mapping[str, typing.Any] = details.flag_metadata or {}
+            metadata: typing.Mapping[str, typing.Any] = (
+                details.flag_metadata if details.flag_metadata is not None else {}
+            )
             allocation_key: str = ""
             ak = metadata.get(METADATA_ALLOCATION_KEY)
             if isinstance(ak, str) and ak:
@@ -83,21 +86,37 @@ def finally_after(
                 variant = details.variant
             runtime_default = variant == ""
 
+            # Consent for this evaluation, read only from metadata the evaluator
+            # stamped. Anything not exactly True is treated as consent-off:
+            # a missing key, a non-bool, or None -- so a broken upstream cannot
+            # silently opt in.
+            observe_full_evaluation_data = metadata.get(METADATA_OBSERVE_FULL_EVALUATION_DATA) is True
+
             # Targeting key and attributes from the evaluation context.
             eval_ctx = hook_context.evaluation_context
-            targeting_key = eval_ctx.targeting_key or ""
-            # Shallow copy so we don't hold a reference to the caller's live dict.
-            attrs: dict[str, typing.Any] = dict(eval_ctx.attributes or {})
+            # Preserve None versus an explicit empty string. The writer omits
+            # missing or malformed keys but serializes an empty evaluated key.
+            targeting_key = eval_ctx.targeting_key
+            # Consent-off contexts never enter the queue. With consent, enqueue()
+            # snapshots synchronously and never retains this borrowed mapping.
+            attrs: typing.Mapping[str, typing.Any] = (
+                eval_ctx.attributes if observe_full_evaluation_data and eval_ctx.attributes is not None else {}
+            )
 
-            # Error message (best-effort; absent on success paths).
-            error_message = ""
-            if details.error_message:
-                error_message = str(details.error_message)
-            elif details.error_code:
-                error_message = (
+            error_code = ""
+            if details.error_code:
+                error_code = (
                     str(details.error_code.value) if hasattr(details.error_code, "value") else str(details.error_code)
                 )
 
+            # AIDEV-NOTE: Error messages can echo raw evaluation-context values.
+            # Under consent-off, substitute the stable error code before enqueue
+            # so raw text never enters the aggregation key, queue, or wire event.
+            if observe_full_evaluation_data and details.error_message:
+                error_message = str(details.error_message)
+            else:
+                error_message = error_code
+
             event = _EvalEvent(
                 flag_key=flag_key,
                 variant=variant,
@@ -107,6 +126,7 @@ def finally_after(
                 runtime_default=runtime_default,
                 error_message=error_message,
                 eval_time_ms=eval_time_ms,
+                observe_full_evaluation_data=observe_full_evaluation_data,
             )
 
             self._writer.enqueue(event)