feat(openfeature): emit server-side EVP flagevaluation

benchmarks/openfeature_flagevaluation/config.yaml

@@ -0,0 +1,50 @@
+defaults: &defaults
+  num_flags: 100
+  num_users: 50
+  num_context_fields: 10
+
+# Eval hot path the caller pays: finally_after hook (cheap capture + non-blocking enqueue).
+hook-enqueue-typical:
+  <<: *defaults
+  mode: hook_enqueue
+
+hook-enqueue-stress:
+  <<: *defaults
+  mode: hook_enqueue
+  num_flags: 10
+  num_users: 1000
+  num_context_fields: 250
+
+hook-enqueue-scale:
+  <<: *defaults
+  mode: hook_enqueue
+  num_flags: 2500
+  num_users: 500
+  num_context_fields: 20
+
+# Background worker hot path: canonical key + two-tier aggregation.
+aggregate-typical:
+  <<: *defaults
+  mode: aggregate
+
+aggregate-stress:
+  <<: *defaults
+  mode: aggregate
+  num_flags: 10
+  num_users: 1000
+  num_context_fields: 250
+
+aggregate-scale:
+  <<: *defaults
+  mode: aggregate
+  num_flags: 2500
+  num_users: 500
+  num_context_fields: 20
+
+# End-to-end per-flush shape: N enqueues followed by drain + aggregate.
+hook-plus-drain-scale:
+  <<: *defaults
+  mode: hook_plus_drain
+  num_flags: 2500
+  num_users: 500
+  num_context_fields: 20

benchmarks/openfeature_flagevaluation/requirements_scenario.txt

@@ -0,0 +1,5 @@
+# openfeature-sdk provides EvaluationContext / FlagEvaluationDetails / HookContext,
+# which the scenario imports directly. It is a test-time dependency, not a ddtrace
+# runtime dependency, so the benchmark base venv does not install it otherwise.
+# Mirrors the openfeature venv pin in riotfile.py.
+openfeature-sdk~=0.8.0

benchmarks/openfeature_flagevaluation/scenario.py

@@ -0,0 +1,146 @@
+"""Hot-path microbenchmark for the EVP ``flagevaluation`` aggregation pipeline.
+
+The profiles mirror the Go/Ruby OpenFeature EVP benchmark suite:
+
+* typical/100flags_50users_10fields
+* stress/10flags_1000users_250fields
+* scale/2500flags_500users_20fields
+
+The scenario drives ``FlagEvalEVPHook`` and ``FlagEvaluationWriter`` directly so the
+benchmark isolates in-process EVP work: no network, no Remote Config backend, and no
+native flag evaluation cost.
+"""
+
+import bm
+from openfeature.evaluation_context import EvaluationContext
+from openfeature.flag_evaluation import FlagEvaluationDetails
+from openfeature.flag_evaluation import FlagType
+from openfeature.flag_evaluation import Reason
+from openfeature.hook import HookContext
+
+
+def _make_hook_context(flag_key, targeting_key, attrs):
+    ctx = EvaluationContext(targeting_key=targeting_key, attributes=attrs)
+    return HookContext(
+        flag_key=flag_key,
+        flag_type=FlagType.BOOLEAN,
+        default_value=False,
+        evaluation_context=ctx,
+    )
+
+
+def _make_details(flag_key, variant, allocation_key):
+    return FlagEvaluationDetails(
+        flag_key=flag_key,
+        value=True,
+        variant=variant,
+        reason=Reason.TARGETING_MATCH,
+        flag_metadata={"allocation_key": allocation_key},
+    )
+
+
+class OpenFeatureFlagEvaluation(bm.Scenario):
+    # Which segment of the pipeline to exercise.
+    mode: str
+    # Number of distinct flags cycled through.
+    num_flags: int
+    # Number of distinct targeting keys cycled through.
+    num_users: int
+    # Number of evaluation-context attributes per evaluation.
+    num_context_fields: int
+
+    def run(self):
+        from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook
+        from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter
+
+        mode = self.mode
+        num_flags = max(1, self.num_flags)
+        num_users = max(1, self.num_users)
+        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)}
+        flag_keys = ["flag-{}".format(i) for i in range(num_flags)]
+        targeting_keys = ["user-{}".format(i) for i in range(num_users)]
+        hook_contexts = [
+            _make_hook_context(
+                flag_key=flag_keys[i % num_flags],
+                targeting_key=targeting_keys[i % num_users],
+                attrs=attrs,
+            )
+            for i in range(cycle_count)
+        ]
+        details_list = [
+            _make_details(
+                flag_key=flag_keys[i % num_flags],
+                variant="variant-{}".format(i % 4),
+                allocation_key="alloc-{}".format(i % num_flags),
+            )
+            for i in range(cycle_count)
+        ]
+
+        writer = FlagEvaluationWriter(interval=3600.0)
+        hook = FlagEvalEVPHook(writer)
+
+        if mode == "hook_enqueue":
+
+            def _(loops):
+                for i in range(loops):
+                    idx = i % cycle_count
+                    hook.finally_after(hook_contexts[idx], details_list[idx], {})
+                    # Keep the queue from saturating so this measures steady-state
+                    # enqueue, not queue overflow accounting.
+                    if writer._queue.full():
+                        writer._drain_queue()
+
+            yield _
+
+        elif mode == "aggregate":
+            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)
+            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),
+                    runtime_default=False,
+                    error_message="",
+                    eval_time_ms=1_760_000_000_000 + i,
+                )
+                for i in range(cycle_count)
+            ]
+
+            def _(loops):
+                for i in range(loops):
+                    writer._aggregate(events[i % cycle_count])
+                    # Reset periodically so the measured loop is not dominated by
+                    # ever-growing maps after enough unique buckets have been seen.
+                    if i > 0 and (i % 10000) == 0:
+                        writer._full.clear()
+                        writer._degraded.clear()
+                        writer._per_flag_count.clear()
+                        writer._global_count = 0
+
+            yield _
+
+        elif mode == "hook_plus_drain":
+
+            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
+
+            yield _
+
+        else:
+            raise ValueError("unknown mode: {}".format(mode))

benchmarks/suitespec.yml

@@ -50,6 +50,10 @@ components:
     - ddtrace/appsec/iast/*
     - ddtrace/appsec/*
     - ddtrace/internal/settings/asm.py
+  openfeature_benchmark:
+    - ddtrace/openfeature/*
+    - ddtrace/internal/openfeature/*
+    - ddtrace/internal/settings/openfeature.py
   core:
     - ddtrace/internal/__init__.py
     - ddtrace/internal/_exceptions.py
@@ -222,6 +226,16 @@ suites:
       - benchmarks/suitespec.yml
     cpus_per_run: 1
     type: 'microbenchmark'
+  openfeature_flagevaluation:
+    paths:
+      - '@bootstrap'
+      - '@core'
+      - '@openfeature_benchmark'
+      - '@vendor'
+      - benchmarks/openfeature_flagevaluation/*
+      - benchmarks/suitespec.yml
+    cpus_per_run: 1
+    type: 'microbenchmark'
   appsec_iast_aspects:
     paths:
       - '@bootstrap'

ddtrace/internal/evp_proxy/constants.py

@@ -5,6 +5,7 @@
 EVP_SUBDOMAIN_HEADER_API_VALUE = "api"
 EVP_SUBDOMAIN_HEADER_COVERAGE_VALUE = "citestcov-intake"
 EVP_SUBDOMAIN_HEADER_EVENT_VALUE = "citestcycle-intake"
+EVP_SUBDOMAIN_HEADER_EVENT_PLATFORM_VALUE = "event-platform-intake"
 EVP_SUBDOMAIN_HEADER_NAME = "X-Datadog-EVP-Subdomain"
 DEFAULT_EVP_PAYLOAD_SIZE_LIMIT = 5 << 20  # 5 MiB (actual server limit is 5.1 MB)
 DEFAULT_EVP_EVENT_SIZE_LIMIT = 5_000_000  # 5 MB, LLM Obs event size limit

ddtrace/internal/openfeature/_flag_eval_evp_hook.py

@@ -0,0 +1,119 @@
+"""
+FlagEvalEVPHook — OpenFeature `finally_after` hook for EVP flagevaluation emission.
+
+Hook design:
+- Cheap capture only in finally_after (no aggregation, no serialization, no 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
+  feature_flag.evaluations OTel path is preserved unchanged).
+"""
+
+import time
+import typing
+
+from openfeature.flag_evaluation import FlagEvaluationDetails
+from openfeature.hook import Hook
+from openfeature.hook import HookContext
+from openfeature.hook import HookHints
+
+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 FlagEvaluationWriter
+from ddtrace.internal.openfeature._flagevaluation_writer import _EvalEvent
+
+
+logger = get_logger(__name__)
+
+
+class FlagEvalEVPHook(Hook):
+    """
+    OpenFeature Hook that enqueues cheap 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.
+    """
+
+    def __init__(self, writer: FlagEvaluationWriter) -> None:
+        self._writer = writer
+
+    def finally_after(
+        self,
+        hook_context: HookContext,
+        details: FlagEvaluationDetails[typing.Any],
+        hints: HookHints,
+    ) -> None:
+        """
+        Cheap 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).
+
+        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.
+        """
+        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 {}
+            allocation_key: str = ""
+            ak = metadata.get(METADATA_ALLOCATION_KEY)
+            if isinstance(ak, str) and ak:
+                allocation_key = ak
+
+            # Eval-time from provider-stamped metadata; fall back to hook-fire time.
+            eval_time_ms_raw = metadata.get(EVAL_TIMESTAMP_METADATA_KEY)
+            if isinstance(eval_time_ms_raw, (int, float)) and eval_time_ms_raw > 0:
+                eval_time_ms = int(eval_time_ms_raw)
+            else:
+                eval_time_ms = int(time.time() * 1000)
+
+            # Variant: absent variant signals a runtime default.
+            variant = ""
+            if details.variant:
+                variant = details.variant
+            runtime_default = variant == ""
+
+            # 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 {})
+
+            # 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 = (
+                    str(details.error_code.value) if hasattr(details.error_code, "value") else str(details.error_code)
+                )
+
+            event = _EvalEvent(
+                flag_key=flag_key,
+                variant=variant,
+                allocation_key=allocation_key,
+                targeting_key=targeting_key,
+                attrs=attrs,
+                runtime_default=runtime_default,
+                error_message=error_message,
+                eval_time_ms=eval_time_ms,
+            )
+
+            self._writer.enqueue(event)
+
+        except Exception:
+            # Never propagate hook exceptions — best-effort telemetry.
+            logger.debug(
+                "FlagEvalEVPHook.finally_after: failed to enqueue eval snapshot",
+                exc_info=True,
+            )

ddtrace/internal/openfeature/_flageval_metrics.py

@@ -136,7 +136,7 @@ def shutdown(self) -> None:
         log.debug("Flag evaluation metrics shutdown")
 
 
-class FlagEvalHook(Hook):
+class FlagEvalMetricsHook(Hook):
     """
     OpenFeature Hook that tracks flag evaluation metrics.