diff --git a/benchmarks/openfeature_flagevaluation/config.yaml b/benchmarks/openfeature_flagevaluation/config.yaml new file mode 100644 index 00000000000..b70f81471ab --- /dev/null +++ b/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 diff --git a/benchmarks/openfeature_flagevaluation/requirements_scenario.txt b/benchmarks/openfeature_flagevaluation/requirements_scenario.txt new file mode 100644 index 00000000000..ade024d993d --- /dev/null +++ b/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 diff --git a/benchmarks/openfeature_flagevaluation/scenario.py b/benchmarks/openfeature_flagevaluation/scenario.py new file mode 100644 index 00000000000..4cd43eef071 --- /dev/null +++ b/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)) diff --git a/ddtrace/internal/evp_proxy/constants.py b/ddtrace/internal/evp_proxy/constants.py index 0e40a5f3e4c..055dc6aa57b 100644 --- a/ddtrace/internal/evp_proxy/constants.py +++ b/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 diff --git a/ddtrace/internal/openfeature/_flag_eval_evp_hook.py b/ddtrace/internal/openfeature/_flag_eval_evp_hook.py new file mode 100644 index 00000000000..0d61a007f88 --- /dev/null +++ b/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, + ) diff --git a/ddtrace/internal/openfeature/_flageval_metrics.py b/ddtrace/internal/openfeature/_flageval_metrics.py index c4cd18e8532..2c54eccf93c 100644 --- a/ddtrace/internal/openfeature/_flageval_metrics.py +++ b/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. diff --git a/ddtrace/internal/openfeature/_flagevaluation_writer.py b/ddtrace/internal/openfeature/_flagevaluation_writer.py new file mode 100644 index 00000000000..8b9cdf3b379 --- /dev/null +++ b/ddtrace/internal/openfeature/_flagevaluation_writer.py @@ -0,0 +1,830 @@ +""" +FlagEvaluationWriter — SDK-native EVP `flagevaluation` writer for dd-trace-py. + +Implements a two-tier aggregation design (full → degraded → drop-counted). Uses the same +PeriodicService + get_connection() transport path as the exposure writer in writer.py. + +Key design properties: +- Async, best-effort recording: the finally_after hook does cheap capture + non-blocking + enqueue. The writer bounds context before queueing; aggregate/flush work happens in the + background worker. +- Two-tier aggregation (full → degraded → drop-counted). +- Canonical context key: sorted, type-tagged, length-delimited — NOT a hash, so distinct + contexts always produce distinct keys with no collisions. +- Context pruning: ≤256 fields, string values ≤256 chars. +- Caps: GLOBAL_CAP=131_072 (full-tier), PER_FLAG_CAP=10_000 (per-flag full-tier), + DEGRADED_CAP=32_768 (degraded-tier). Beyond the degraded cap: drop-and-count. +- Eval-time from metadata key "dd.eval.timestamp_ms"; fallback to enqueue-time. +- First/last evaluation: min/max under lock. +- runtime_default_used: True when variant is None/absent. +- Killswitch: DD_FLAGGING_EVALUATION_COUNTS_ENABLED (default on); gates EVP path only. +- Non-blocking enqueue: queue.Queue(QUEUE_SIZE); drops + counts on queue.Full. +""" + +from collections.abc import Mapping +from collections.abc import Sequence +import http.client as httplib +import json +import queue +import struct +import threading +import time +import typing + +from ddtrace import config as ddconfig +from ddtrace.internal.evp_proxy.constants import DEFAULT_EVP_PAYLOAD_SIZE_LIMIT +from ddtrace.internal.evp_proxy.constants import EVP_PROXY_AGENT_BASE_PATH +from ddtrace.internal.evp_proxy.constants import EVP_SUBDOMAIN_HEADER_EVENT_PLATFORM_VALUE +from ddtrace.internal.evp_proxy.constants import EVP_SUBDOMAIN_HEADER_NAME +from ddtrace.internal.logger import get_logger +from ddtrace.internal.openfeature._flageval_metrics import METADATA_ALLOCATION_KEY as METADATA_ALLOCATION_KEY +from ddtrace.internal.periodic import PeriodicService +from ddtrace.internal.settings._agent import config as agent_config +from ddtrace.internal.telemetry import telemetry_writer +from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE +from ddtrace.internal.threads import PeriodicThread +from ddtrace.internal.utils.http import get_connection + + +logger = get_logger(__name__) + +# EVP endpoint for flag evaluation events. +FLAGEVALUATIONS_ENDPOINT = f"{EVP_PROXY_AGENT_BASE_PATH}/api/v2/flagevaluation" +EVP_SUBDOMAIN_VALUE = EVP_SUBDOMAIN_HEADER_EVENT_PLATFORM_VALUE +FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT = DEFAULT_EVP_PAYLOAD_SIZE_LIMIT +_JSON_SEPARATORS = (",", ":") + +# Context pruning limits — mirror worker.ts MAX_EVALUATION_CONTEXT_FIELDS / MAX_FIELD_LENGTH. +MAX_CONTEXT_FIELDS = 256 +MAX_FIELD_LENGTH = 256 +DEDICATED_TARGETING_KEY_CONTEXT_FIELDS = frozenset(("targetingKey", "targeting_key")) + +# Aggregation caps (sized for a >=2,500-flag scale target). +EVAL_SCALE_TARGET_FLAGS = 2_500 +EVAL_SCALE_FULL_BUCKETS_PER_FLAG = 50 +EVAL_SCALE_USERS_PER_FLAG = 1_000 +EVAL_SCALE_PER_FLAG_HEADROOM_MULTIPLIER = 10 +EVAL_SCALE_DEGRADED_BUCKETS_PER_FLAG = 10 +EVAL_SCALE_FULL_BUCKET_TARGET = EVAL_SCALE_TARGET_FLAGS * EVAL_SCALE_FULL_BUCKETS_PER_FLAG +EVAL_SCALE_PER_FLAG_BUCKET_TARGET = EVAL_SCALE_PER_FLAG_HEADROOM_MULTIPLIER * EVAL_SCALE_USERS_PER_FLAG +EVAL_SCALE_DEGRADED_BUCKET_TARGET = EVAL_SCALE_TARGET_FLAGS * EVAL_SCALE_DEGRADED_BUCKETS_PER_FLAG +GLOBAL_CAP = 131_072 # bounds full-tier buckets +PER_FLAG_CAP = EVAL_SCALE_PER_FLAG_BUCKET_TARGET # bounds full-tier buckets per flag +DEGRADED_CAP = 32_768 # bounds degraded-tier buckets; overflow is drop-counted + +# Async hand-off queue size. +QUEUE_SIZE = 4_096 + +# Flush interval: dedicated 10 s timer, separate from ExposureWriter's 1 s interval. +DEFAULT_FLUSH_INTERVAL = 10.0 + +# Queue drain interval. This keeps the hand-off queue bounded while allowing a flush +# window to accumulate more buckets than QUEUE_SIZE. +DRAIN_INTERVAL = 0.1 + +# Flag metadata key where the provider stamps the evaluation timestamp (ms). +EVAL_TIMESTAMP_METADATA_KEY = "dd.eval.timestamp_ms" + +# Type-tag bytes for the canonical context key encoding (mirrors Go's ctxTag* constants). +_TAG_STR = b"s" +_TAG_BOOL = b"b" +_TAG_INT = b"i" +_TAG_FLOAT = b"f" +_TAG_OTHER = b"o" + +FLAG_EVALUATION_DROPPED_METRIC = "flagevaluation.rows.dropped" +FLAG_EVALUATION_DEGRADED_METRIC = "flagevaluation.rows.degraded" +FLAG_EVALUATION_SPLITS_METRIC = "flagevaluation.payload.splits" + +FLAG_EVALUATION_REASON_QUEUE_OVERFLOW = "queue_overflow" +FLAG_EVALUATION_REASON_CLOSED = "closed" +FLAG_EVALUATION_REASON_DEGRADED_CAP = "degraded_cap" +FLAG_EVALUATION_REASON_PAYLOAD_LIMIT = "payload_limit" +FLAG_EVALUATION_REASON_CARDINALITY_CAP = "cardinality_cap" + + +def _json_dumps(obj: typing.Any) -> bytes: + return json.dumps(obj, default=_json_default, separators=_JSON_SEPARATORS).encode("utf-8") + + +def _json_default(value: typing.Any) -> str: + if hasattr(value, "isoformat"): + try: + return str(value.isoformat()) + except Exception: + return str(value) + return str(value) + + +def _count_metric(name: str, value: int, reason: typing.Optional[str] = None) -> None: + if value <= 0: + return + tags = (("reason", reason),) if reason else tuple() + telemetry_writer.add_count_metric(TELEMETRY_NAMESPACE.TRACERS, name, value, tags) + + +# --------------------------------------------------------------------------- +# Canonical context key — type-tagged, length-delimited, sorted +# --------------------------------------------------------------------------- + + +def _length_delimited(data: bytes) -> bytes: + """Prepend a fixed 8-byte big-endian length to data.""" + return struct.pack(">Q", len(data)) + data + + +def _encode_context_value(v: typing.Any) -> bytes: + """Encode a single context value with a type tag + length-delimited value.""" + if isinstance(v, bool): + # bool must be checked before int because bool is a subclass of int in Python. + tag = _TAG_BOOL + raw = b"true" if v else b"false" + elif isinstance(v, int): + tag = _TAG_INT + raw = str(v).encode() + elif isinstance(v, float): + tag = _TAG_FLOAT + raw = repr(v).encode() + elif isinstance(v, str): + tag = _TAG_STR + raw = v.encode("utf-8", errors="replace") + else: + tag = _TAG_OTHER + raw = str(v).encode("utf-8", errors="replace") + return tag + _length_delimited(raw) + + +def canonical_context_key(attrs: dict[str, typing.Any]) -> str: + """ + Build the EXACT, comparable canonical-context string key for a pruned context dict. + + Uses sorted(attrs.items()) so the encoding is deterministic regardless of Python + dict insertion order. Each entry is encoded as: + length_delimited(key_bytes) + type_tag_byte + length_delimited(value_bytes) + + Because the full encoding is used as the map key (not a hash), distinct contexts + ALWAYS produce distinct keys — no hash collisions, no misattribution. + + Returns "" for empty/None attrs. + """ + if not attrs: + return "" + parts = [] + for k in sorted(attrs.keys()): + parts.append(_length_delimited(k.encode("utf-8", errors="replace"))) + parts.append(_encode_context_value(attrs[k])) + return b"".join(parts).decode("latin-1") # lossless binary → str for dict key + + +def flatten_and_prune_context(attrs: dict[str, typing.Any]) -> dict[str, typing.Any]: + """ + Flatten nested dicts (dot-notation) and apply 256-field / 256-char prune. + + Returns a new dict with at most MAX_CONTEXT_FIELDS entries, skipping string values + that exceed MAX_FIELD_LENGTH. Keys are chosen deterministically (sorted order) so + identical contexts always produce identical pruned maps. + + Returns {} when the input is empty or all values are pruned. + """ + if not attrs: + return {} + + flat: dict[str, typing.Any] = {} + _flatten_recursive("", attrs, flat, set()) + if not flat: + return {} + + # Fast path: no pruning needed. + needs_prune = len(flat) > MAX_CONTEXT_FIELDS + if not needs_prune: + for v in flat.values(): + if isinstance(v, str) and len(v) > MAX_FIELD_LENGTH: + needs_prune = True + break + if not needs_prune: + return flat + + # Deterministic prune: sort keys, keep first MAX_CONTEXT_FIELDS non-oversized values. + out: dict[str, typing.Any] = {} + count = 0 + for k in sorted(flat.keys()): + if count >= MAX_CONTEXT_FIELDS: + break + v = flat[k] + if isinstance(v, str) and len(v) > MAX_FIELD_LENGTH: + continue + out[k] = v + count += 1 + return out + + +def _json_safe_context_value(value: typing.Any) -> typing.Any: + return json.loads(_json_dumps(value).decode("utf-8")) + + +def _json_safe_context(attrs: dict[str, typing.Any]) -> dict[str, typing.Any]: + return {k: _json_safe_context_value(v) for k, v in attrs.items()} + + +def _flatten_recursive(prefix: str, attrs: typing.Any, out: dict[str, typing.Any], seen: set[int]) -> None: + """Recursively flatten nested maps and sequences into dot/bracket notation keys.""" + if isinstance(attrs, Mapping): + attrs_id = id(attrs) + if attrs_id in seen: + return + seen.add(attrs_id) + try: + for k, v in attrs.items(): + if not prefix and k in DEDICATED_TARGETING_KEY_CONTEXT_FIELDS: + continue + full_key = f"{prefix}.{k}" if prefix else str(k) + _flatten_recursive(full_key, v, out, seen) + finally: + seen.remove(attrs_id) + return + + if isinstance(attrs, Sequence) and not isinstance(attrs, (str, bytes, bytearray)): + attrs_id = id(attrs) + if attrs_id in seen: + return + seen.add(attrs_id) + try: + for idx, value in enumerate(attrs): + full_key = f"{prefix}[{idx}]" if prefix else f"[{idx}]" + _flatten_recursive(full_key, value, out, seen) + finally: + seen.remove(attrs_id) + return + + if prefix: + out[prefix] = attrs + + +# --------------------------------------------------------------------------- +# Internal types +# --------------------------------------------------------------------------- + + +class _Entry: + """Per-bucket aggregation state.""" + + __slots__ = ( + "count", + "first_evaluation", + "last_evaluation", + "runtime_default", + "targeting_key", + "context_attrs", + "error_message", + ) + + def __init__( + self, + eval_time_ms: int, + runtime_default: bool, + targeting_key: str, + context_attrs: dict[str, typing.Any], + error_message: str, + ) -> None: + self.count: int = 1 + self.first_evaluation: int = eval_time_ms + self.last_evaluation: int = eval_time_ms + self.runtime_default: bool = runtime_default + # Full-tier only: + self.targeting_key: str = targeting_key + self.context_attrs: dict[str, typing.Any] = context_attrs + self.error_message: str = error_message + + def observe(self, eval_time_ms: int) -> None: + """Update count and first/last bounds for a repeated evaluation.""" + self.count += 1 + if eval_time_ms < self.first_evaluation: + self.first_evaluation = eval_time_ms + if eval_time_ms > self.last_evaluation: + self.last_evaluation = eval_time_ms + + +class _EvalEvent(typing.NamedTuple): + """Minimal snapshot handed from finally_after to the background worker.""" + + flag_key: str + variant: str # "" when absent (= runtime_default) + allocation_key: str + targeting_key: str + attrs: dict[str, typing.Any] # flattened and pruned context snapshot + runtime_default: bool + error_message: str + eval_time_ms: int + + +class _FlagEvaluationConnection(typing.Protocol): + def request(self, method: str, url: str, body: bytes, headers: dict[str, str]) -> None: + pass + + def getresponse(self) -> httplib.HTTPResponse: + pass + + def close(self) -> None: + pass + + +class _PayloadEventResult(typing.NamedTuple): + encoded: typing.Optional[bytes] + degraded_payload_limit: bool = False + dropped_payload_limit: bool = False + + +class _PayloadBuildResult(typing.NamedTuple): + payloads: list[tuple[bytes, int]] + degraded_payload_limit: int = 0 + dropped_payload_limit: int = 0 + + +# --------------------------------------------------------------------------- +# FlagEvaluationWriter +# --------------------------------------------------------------------------- + + +class FlagEvaluationWriter(PeriodicService): + """ + SDK-native EVP `flagevaluation` writer. + + Two-tier aggregation design: + - full-tier: keyed by schema-visible dimensions only: flag, variant, allocation, + runtime_default_used, error.message, targeting_key, canonical_context + - degraded-tier: keyed by schema-visible retained dimensions: flag, variant, allocation, + runtime_default_used, error.message + - drop-counted: beyond degradedCap, increment _dropped_degraded_overflow + + The finally_after hook enqueues _EvalEvent snapshots through enqueue(), which bounds + context before buffering; the PeriodicService background thread drains the queue, + aggregates, and flushes via HTTP every 10 s. + """ + + def __init__(self, interval: float = DEFAULT_FLUSH_INTERVAL, timeout: float = 2.0) -> None: + super().__init__(interval=interval) + self._timeout = timeout + self._intake: str = agent_config.trace_agent_url + self._endpoint: str = FLAGEVALUATIONS_ENDPOINT + self._headers: dict[str, str] = { + "Content-Type": "application/json", + EVP_SUBDOMAIN_HEADER_NAME: EVP_SUBDOMAIN_VALUE, + } + + # Async hand-off queue: non-blocking, bounded. + self._queue: "queue.Queue[_EvalEvent]" = queue.Queue(maxsize=QUEUE_SIZE) + + # Aggregation maps (drained under _lock on each periodic() call). Keys are tuples of + # the enumerable dimensions plus (full tier) the canonical context string. + self._lock = threading.Lock() + self._full: dict[tuple[typing.Any, ...], _Entry] = {} + self._degraded: dict[tuple[typing.Any, ...], _Entry] = {} + self._per_flag_count: dict[str, int] = {} # flag_key → full-tier bucket count + self._global_count: int = 0 + + # Observable drop counters. + self._dropped_queue: int = 0 # queue.Full drops (hook path) + self._dropped_degraded_overflow: int = 0 # degraded-cap overflow drops + + self._drain_worker: typing.Optional[PeriodicThread] = None + self._accepting_events = True + + # ------------------------------------------------------------------ + # Public API used by FlagEvalEVPHook + # ------------------------------------------------------------------ + + def enqueue(self, event: _EvalEvent) -> None: + """ + Non-blocking enqueue from the finally_after hook thread. + + Context is flattened/pruned before it enters the queue so queue length is not the + only memory bound. On queue.Full, increments _dropped_queue (observable) and + returns immediately — never blocks the evaluation hot path. + """ + if not self._accepting_events: + _count_metric(FLAG_EVALUATION_DROPPED_METRIC, 1, FLAG_EVALUATION_REASON_CLOSED) + logger.debug("FlagEvaluationWriter: dropped flag evaluation event after shutdown started") + return + + bounded_event = _EvalEvent( + flag_key=event.flag_key, + variant=event.variant, + allocation_key=event.allocation_key, + targeting_key=event.targeting_key, + attrs=flatten_and_prune_context(event.attrs) if event.attrs else {}, + runtime_default=event.runtime_default, + error_message=event.error_message, + eval_time_ms=event.eval_time_ms, + ) + + try: + self._queue.put_nowait(bounded_event) + except queue.Full: + with self._lock: + self._dropped_queue += 1 + logger.debug( + "FlagEvaluationWriter: queue full — dropped flag evaluation event for %s", + bounded_event.flag_key, + ) + + # ------------------------------------------------------------------ + # PeriodicService implementation + # ------------------------------------------------------------------ + + def _start_service(self, *args: typing.Any, **kwargs: typing.Any) -> None: + self._accepting_events = True + self._drain_worker = PeriodicThread( + DRAIN_INTERVAL, + target=self._drain_queue, + name="%s:%s:drain" % (self.__class__.__module__, self.__class__.__name__), + no_wait_at_start=False, + ) + self._drain_worker.start() + try: + super()._start_service(*args, **kwargs) + except Exception: + self._stop_drain_worker() + raise + + def _stop_service(self, *args: typing.Any, **kwargs: typing.Any) -> None: + self._accepting_events = False + self._stop_drain_worker() + super()._stop_service(*args, **kwargs) + + def periodic(self) -> None: + """ + Drain the queue, aggregate, and flush to the EVP proxy. + + Called periodically by the PeriodicService thread (every DEFAULT_FLUSH_INTERVAL). + Also callable directly in tests. + """ + # 1. Drain the queue into the aggregation maps. + self._drain_queue() + + # 2. Snapshot and reset under lock. + with self._lock: + dropped_queue = self._dropped_queue + dropped_degraded = self._dropped_degraded_overflow + full = self._full + degraded = self._degraded + if not full and not degraded: + if dropped_queue: + logger.warning( + "FlagEvaluationWriter: queue full — dropped %d evaluation(s) under backpressure", + dropped_queue, + ) + _count_metric( + FLAG_EVALUATION_DROPPED_METRIC, + dropped_queue, + FLAG_EVALUATION_REASON_QUEUE_OVERFLOW, + ) + self._dropped_queue = 0 + if dropped_degraded: + logger.warning( + "FlagEvaluationWriter: degraded cap full — dropped %d evaluation(s)", + dropped_degraded, + ) + _count_metric( + FLAG_EVALUATION_DROPPED_METRIC, + dropped_degraded, + FLAG_EVALUATION_REASON_DEGRADED_CAP, + ) + self._dropped_degraded_overflow = 0 + return + # Reset maps. + self._full = {} + self._degraded = {} + self._per_flag_count = {} + self._global_count = 0 + self._dropped_queue = 0 + self._dropped_degraded_overflow = 0 + + if dropped_queue: + logger.warning( + "FlagEvaluationWriter: queue full — dropped %d evaluation(s) under backpressure", + dropped_queue, + ) + _count_metric(FLAG_EVALUATION_DROPPED_METRIC, dropped_queue, FLAG_EVALUATION_REASON_QUEUE_OVERFLOW) + if dropped_degraded: + logger.warning( + "FlagEvaluationWriter: degraded cap full — dropped %d evaluation(s)", + dropped_degraded, + ) + _count_metric(FLAG_EVALUATION_DROPPED_METRIC, dropped_degraded, FLAG_EVALUATION_REASON_DEGRADED_CAP) + + # 3. Build payload. + flush_time_ms = int(time.time() * 1000) + events = [] + + # Full-tier events: all optional fields present. + for key, entry in full.items(): + flag_key = key[0] + variant = key[1] + allocation_key = key[2] + ev = _base_event(flag_key, entry, flush_time_ms) + if entry.runtime_default: + ev["runtime_default_used"] = True + if entry.targeting_key: + ev["targeting_key"] = entry.targeting_key + if variant: + ev["variant"] = {"key": variant} + if allocation_key: + ev["allocation"] = {"key": allocation_key} + if entry.error_message: + ev["error"] = {"message": entry.error_message} + if entry.context_attrs: + ev["context"] = {"evaluation": entry.context_attrs} + events.append(ev) + + # Degraded-tier events: no targeting_key, no context. + degraded_count = 0 + for key, entry in degraded.items(): + degraded_count += entry.count + flag_key = key[0] + variant = key[1] + allocation_key = key[2] + ev = _base_event(flag_key, entry, flush_time_ms) + if entry.runtime_default: + ev["runtime_default_used"] = True + if variant: + ev["variant"] = {"key": variant} + if allocation_key: + ev["allocation"] = {"key": allocation_key} + if entry.error_message: + ev["error"] = {"message": entry.error_message} + events.append(ev) + _count_metric(FLAG_EVALUATION_DEGRADED_METRIC, degraded_count, FLAG_EVALUATION_REASON_CARDINALITY_CAP) + + if not events: + return + + # 4. Encode under the EVP payload limit and POST. + context: dict[str, str] = {} + if ddconfig.service: + context["service"] = ddconfig.service + if ddconfig.env: + context["env"] = ddconfig.env + if ddconfig.version: + context["version"] = ddconfig.version + + result = _build_payloads_with_stats(events, context, FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT) + _count_metric( + FLAG_EVALUATION_DEGRADED_METRIC, + result.degraded_payload_limit, + FLAG_EVALUATION_REASON_PAYLOAD_LIMIT, + ) + _count_metric( + FLAG_EVALUATION_DROPPED_METRIC, + result.dropped_payload_limit, + FLAG_EVALUATION_REASON_PAYLOAD_LIMIT, + ) + if len(result.payloads) > 1: + _count_metric(FLAG_EVALUATION_SPLITS_METRIC, len(result.payloads) - 1) + + for payload, num_events in result.payloads: + self._send_payload(payload, num_events) + + def on_shutdown(self) -> None: # type: ignore[override] + """Final flush on service shutdown — drains the queue and flushes before exit.""" + self._accepting_events = False + self._stop_drain_worker() + self.periodic() + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _drain_queue(self) -> None: + """Drain all pending events from the queue and aggregate them.""" + while True: + try: + event = self._queue.get_nowait() + except queue.Empty: + break + self._aggregate(event) + + def _stop_drain_worker(self) -> None: + worker = self._drain_worker + if worker is None: + return + self._drain_worker = None + worker.stop() + worker.join(timeout=1.0) + + def _aggregate(self, event: _EvalEvent) -> None: + """ + Aggregate a single evaluation event into the two-tier maps. + + Implements: full-tier → degraded-tier → drop-counted cascade. + Canonical key computation happens here (off the hot path). Context was already + flattened and pruned before enqueue. + """ + context_attrs = event.attrs or {} + + # Build the full-tier key tuple. + ctx_key = canonical_context_key(context_attrs) + full_key = ( + event.flag_key, + event.variant, + event.allocation_key, + event.runtime_default, + event.error_message, + event.targeting_key, + ctx_key, + ) + + with self._lock: + # Fast path: existing full-tier bucket. + if full_key in self._full: + self._full[full_key].observe(event.eval_time_ms) + return + + # Per-flag cap check. + per_flag = self._per_flag_count.get(event.flag_key, 0) + if per_flag >= PER_FLAG_CAP: + self._add_to_degraded(event) + return + + # Increment per-flag attempt count before checking globalCap (matches Go design). + self._per_flag_count[event.flag_key] = per_flag + 1 + + # Global cap check. + if self._global_count >= GLOBAL_CAP: + self._add_to_degraded(event) + return + + # New full-tier bucket. + self._full[full_key] = _Entry( + eval_time_ms=event.eval_time_ms, + runtime_default=event.runtime_default, + targeting_key=event.targeting_key, + context_attrs=_json_safe_context(context_attrs), + error_message=event.error_message, + ) + self._global_count += 1 + + def _add_to_degraded(self, event: _EvalEvent) -> None: + """ + Add to the degraded-tier map (drops targeting_key + context). + Must be called with self._lock held. + """ + deg_key = ( + event.flag_key, + event.variant, + event.allocation_key, + event.runtime_default, + event.error_message, + ) + if deg_key in self._degraded: + self._degraded[deg_key].observe(event.eval_time_ms) + return + + if len(self._degraded) >= DEGRADED_CAP: + self._dropped_degraded_overflow += 1 + return + + self._degraded[deg_key] = _Entry( + eval_time_ms=event.eval_time_ms, + runtime_default=event.runtime_default, + targeting_key="", + context_attrs={}, + error_message=event.error_message, + ) + + def _send_payload(self, payload: bytes, num_events: int) -> None: + """POST the encoded payload to the EVP proxy.""" + conn = typing.cast(_FlagEvaluationConnection, get_connection(self._intake, timeout=self._timeout)) + try: + conn.request("POST", self._endpoint, payload, self._headers) + resp = conn.getresponse() + if resp.status >= 300: + logger.debug( + "FlagEvaluationWriter: failed to send %d events to %s, status=%d: %s", + num_events, + self._intake, + resp.status, + resp.read(), + ) + else: + logger.debug( + "FlagEvaluationWriter: sent %d flag evaluation events to %s", + num_events, + self._intake, + ) + except Exception: + logger.debug( + "FlagEvaluationWriter: error sending %d events to %s", + num_events, + self._intake, + exc_info=True, + ) + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# Payload helpers +# --------------------------------------------------------------------------- + + +def _base_event(flag_key: str, entry: "_Entry", flush_time_ms: int) -> dict[str, typing.Any]: + """Build the required-fields-only event dict for a single aggregation entry.""" + return { + "timestamp": flush_time_ms, + "flag": {"key": flag_key}, + "first_evaluation": entry.first_evaluation, + "last_evaluation": entry.last_evaluation, + "evaluation_count": entry.count, + } + + +def _degraded_payload_event(event: dict[str, typing.Any]) -> dict[str, typing.Any]: + degraded = event.copy() + degraded.pop("context", None) + degraded.pop("targeting_key", None) + return degraded + + +def _encode_payload_event( + event: dict[str, typing.Any], + single_event_payload_limit: int, +) -> _PayloadEventResult: + try: + encoded = _json_dumps(event) + except (TypeError, ValueError): + logger.debug("FlagEvaluationWriter: failed to encode event", exc_info=True) + return _PayloadEventResult(None) + + if len(encoded) <= single_event_payload_limit: + return _PayloadEventResult(encoded) + + degraded_event = _degraded_payload_event(event) + if degraded_event != event: + try: + encoded = _json_dumps(degraded_event) + except (TypeError, ValueError): + logger.debug("FlagEvaluationWriter: failed to encode degraded event", exc_info=True) + return _PayloadEventResult(None) + if len(encoded) <= single_event_payload_limit: + logger.warning( + "FlagEvaluationWriter: degraded oversized flag evaluation event for %s before sending", + event.get("flag", {}).get("key", ""), + ) + return _PayloadEventResult(encoded, degraded_payload_limit=True) + + logger.warning( + "FlagEvaluationWriter: dropped oversized flag evaluation event for %s", + event.get("flag", {}).get("key", ""), + ) + return _PayloadEventResult(None, dropped_payload_limit=True) + + +def _build_payloads_with_stats( + events: list[dict[str, typing.Any]], + context: dict[str, str], + payload_size_limit: int = FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT, +) -> _PayloadBuildResult: + context_suffix = b"" + if context: + context_suffix = b',"context":' + _json_dumps(context) + prefix = b'{"flagEvaluations":[' + suffix = b"]" + context_suffix + b"}" + single_event_payload_limit = payload_size_limit - len(prefix) - len(suffix) + if single_event_payload_limit <= 0: + logger.warning("FlagEvaluationWriter: EVP payload size limit is too small to encode flagevaluation payloads") + return _PayloadBuildResult([]) + + payload = bytearray(prefix) + num_events = 0 + payloads: list[tuple[bytes, int]] = [] + degraded_payload_limit = 0 + dropped_payload_limit = 0 + + for event in events: + event_result = _encode_payload_event(event, single_event_payload_limit) + encoded_event = event_result.encoded + if event_result.dropped_payload_limit: + dropped_payload_limit += int(event.get("evaluation_count", 1) or 1) + if event_result.degraded_payload_limit: + degraded_payload_limit += int(event.get("evaluation_count", 1) or 1) + if encoded_event is None: + continue + + separator_size = 1 if num_events else 0 + candidate_size = len(payload) + separator_size + len(encoded_event) + len(suffix) + if num_events and candidate_size > payload_size_limit: + payload.extend(suffix) + payloads.append((bytes(payload), num_events)) + payload = bytearray(prefix) + num_events = 0 + + if num_events: + payload.extend(b",") + payload.extend(encoded_event) + num_events += 1 + + if num_events: + payload.extend(suffix) + payloads.append((bytes(payload), num_events)) + + return _PayloadBuildResult(payloads, degraded_payload_limit, dropped_payload_limit) diff --git a/ddtrace/internal/openfeature/_provider.py b/ddtrace/internal/openfeature/_provider.py index 258382d04d7..f924b62a2e1 100644 --- a/ddtrace/internal/openfeature/_provider.py +++ b/ddtrace/internal/openfeature/_provider.py @@ -9,12 +9,14 @@ from collections.abc import MutableMapping from importlib.metadata import version import threading +import time import typing from openfeature.evaluation_context import EvaluationContext from openfeature.event import ProviderEventDetails from openfeature.exception import ErrorCode from openfeature.flag_evaluation import FlagResolutionDetails +from openfeature.flag_evaluation import FlagValueType from openfeature.flag_evaluation import Reason from openfeature.provider import Metadata from openfeature.provider import ProviderStatus @@ -23,28 +25,35 @@ from ddtrace.internal.native._native import ffe from ddtrace.internal.openfeature._config import _get_ffe_config from ddtrace.internal.openfeature._exposure import build_exposure_event +from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook from ddtrace.internal.openfeature._flageval_metrics import METADATA_ALLOCATION_KEY -from ddtrace.internal.openfeature._flageval_metrics import FlagEvalHook from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetrics +from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetricsHook +from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_TIMESTAMP_METADATA_KEY +from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter from ddtrace.internal.openfeature._native import VariationType from ddtrace.internal.openfeature._native import resolve_flag from ddtrace.internal.openfeature.writer import get_exposure_writer from ddtrace.internal.openfeature.writer import start_exposure_writer from ddtrace.internal.openfeature.writer import stop_exposure_writer from ddtrace.internal.service import ServiceStatusError +from ddtrace.internal.settings.openfeature import OpenFeatureConfig from ddtrace.internal.settings.openfeature import config as ffe_config +from ddtrace.vendor.packaging.version import parse as parse_version # Handle different import paths between openfeature-sdk versions # Versions 0.7.0+ reorganized submodules pkg_version = version("openfeature-sdk") -if pkg_version >= "0.7.0": +if parse_version(pkg_version) >= parse_version("0.7.0"): from openfeature.provider import AbstractProvider else: from openfeature.provider.provider import AbstractProvider T = typing.TypeVar("T", covariant=True) +ResolvedValue = typing.TypeVar("ResolvedValue") +ObjectFlagValue = typing.Union[typing.Sequence[FlagValueType], typing.Mapping[str, FlagValueType]] K = typing.TypeVar("K") V = typing.TypeVar("V") logger = get_logger(__name__) @@ -91,19 +100,18 @@ class DataDogProvider(AbstractProvider): Feature Flags and Experimentation (FFE) product. """ - def __init__(self, *args: typing.Any, initialization_timeout: typing.Optional[float] = None, **kwargs: typing.Any): + def __init__( + self, + *args: typing.Any, + initialization_timeout: typing.Optional[float] = None, + **kwargs: typing.Any, + ) -> None: super().__init__(*args, **kwargs) self._metadata = Metadata(name="Datadog") self._status = ProviderStatus.NOT_READY - # Initialization timeout: constructor arg takes priority, then env var - if initialization_timeout is not None: - self._initialization_timeout = initialization_timeout - else: - self._initialization_timeout = ffe_config.initialization_timeout_ms / 1000.0 - - # Event used to block initialize() until config arrives. - # Also serves as the "config received" flag via is_set(). + # Event set when the first RC config arrives; used by on_configuration_received() + # to guard the first-config path and by _emit_ready_event() timing. self._config_received = threading.Event() # Cache for reported exposures to prevent duplicates @@ -124,41 +132,77 @@ def __init__(self, *args: typing.Any, initialization_timeout: typing.Optional[fl # Initialize flag evaluation metrics tracking # Metrics are emitted via OTel when DD_METRICS_OTEL_ENABLED=true self._flag_eval_metrics: typing.Optional[FlagEvalMetrics] = None - self._flag_eval_hook: typing.Optional[FlagEvalHook] = None + self._flag_eval_metrics_hook: typing.Optional[FlagEvalMetricsHook] = None if self._enabled: self._flag_eval_metrics = FlagEvalMetrics() - self._flag_eval_hook = FlagEvalHook(self._flag_eval_metrics) + self._flag_eval_metrics_hook = FlagEvalMetricsHook(self._flag_eval_metrics) + + # EVP flagevaluation writer + hook — gated by DD_FLAGGING_EVALUATION_COUNTS_ENABLED + # (default on). Gates ONLY the EVP path; the OTel path above is always registered + # when the provider is enabled (preserves the existing OTel non-regression). + # AIDEV-NOTE: the killswitch is read through the ddtrace config system + # (OpenFeatureConfig.flagging_evaluation_counts_enabled, registered in + # supported-configurations.json) rather than raw os.environ. A fresh + # OpenFeatureConfig instance is constructed here so the value reflects the current + # environment at provider-construction time (the config var parses the live + # environment via the DDConfig var system), which keeps the killswitch overridable + # per-instance in tests. + self._flag_eval_evp_writer: typing.Optional[FlagEvaluationWriter] = None + self._flag_eval_evp_hook: typing.Optional[FlagEvalEVPHook] = None + evp_config = OpenFeatureConfig() + evp_counts_enabled = evp_config.flagging_evaluation_counts_enabled + if self._enabled and evp_counts_enabled: + self._flag_eval_evp_writer = FlagEvaluationWriter() + self._flag_eval_evp_hook = FlagEvalEVPHook(self._flag_eval_evp_writer) def get_metadata(self) -> Metadata: """Returns provider metadata.""" return self._metadata + def attach(self, on_emit: typing.Callable[..., None]) -> None: + """Attach OpenFeature event dispatch and register for RC callbacks.""" + super().attach(on_emit) + if self._enabled: + _register_provider(self) + def get_provider_hooks(self) -> list[typing.Any]: """ Returns provider-level hooks. - The flag evaluation hook is registered here to track metrics for + The OTel metrics hook is registered here to track metrics for every flag evaluation via the finally_after hook stage. + + Hook ordering: + 1. OTel FlagEvalMetricsHook (_flageval_metrics.py) — always registered when the provider + is enabled; emits the feature_flag.evaluations OTel counter (preserved unchanged). + 2. FlagEvalEVPHook (_flag_eval_evp_hook.py) — registered only when + DD_FLAGGING_EVALUATION_COUNTS_ENABLED is enabled (default on); enqueues cheap + snapshots to FlagEvaluationWriter for EVP flagevaluation emission. """ hooks: list[typing.Any] = [] - if self._flag_eval_hook is not None: - hooks.append(self._flag_eval_hook) + if self._flag_eval_metrics_hook is not None: + hooks.append(self._flag_eval_metrics_hook) + if self._flag_eval_evp_hook is not None: + hooks.append(self._flag_eval_evp_hook) return hooks def initialize(self, evaluation_context: EvaluationContext) -> None: """ Initialize the provider. - Blocks until Remote Config delivers the first FFE configuration or - the initialization timeout expires. + Returns immediately. This provider's internal status remains NOT_READY until + Remote Config delivers the first FFE_FLAGS payload via on_configuration_received(). + openfeature-sdk 0.8.x still dispatches PROVIDER_READY after initialize() + returns, so flag resolution itself remains gated on the loaded config rather than + the SDK registry's ready event. - The timeout is configurable via: - - Constructor: DataDogProvider(initialization_timeout=10.0) # seconds - - Env var: DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS=10000 + If RC has already delivered config before initialize() runs (e.g. in the master + process of a pre-fork server), the fast path sets READY synchronously so the SDK + dispatches PROVIDER_READY on return. Provider lifecycle: - NOT_READY -> initialize() blocks -> config arrives -> READY - NOT_READY -> initialize() blocks -> timeout -> raises ProviderNotReadyError + NOT_READY -> initialize() returns -> RC delivers config -> on_configuration_received() + -> READY """ if not self._enabled: return @@ -173,7 +217,16 @@ def initialize(self, evaluation_context: EvaluationContext) -> None: except ServiceStatusError: logger.debug("Exposure writer is already running", exc_info=True) - # Fast path: config already available (RC delivered before set_provider) + # Start the EVP flagevaluation writer (if enabled via killswitch). + if self._flag_eval_evp_writer is not None: + try: + self._flag_eval_evp_writer.start() + logger.debug("FlagEvaluationWriter started") + except ServiceStatusError: + logger.debug("FlagEvaluationWriter is already running", exc_info=True) + + # Fast path: config already available (RC delivered before set_provider — + # common in pre-fork servers where master receives RC before workers fork). config = _get_ffe_config() if config is not None: logger.debug("FFE configuration already available, provider is READY") @@ -181,20 +234,18 @@ def initialize(self, evaluation_context: EvaluationContext) -> None: self._status = ProviderStatus.READY return # SDK will dispatch PROVIDER_READY - # Block until config arrives or timeout expires - logger.debug( - "Waiting up to %.1fs for initial FFE configuration from Remote Config", self._initialization_timeout - ) - if not self._config_received.wait(timeout=self._initialization_timeout): - # Timeout expired without receiving config - from openfeature.exception import ProviderNotReadyError - - raise ProviderNotReadyError( - f"Provider timed out after {self._initialization_timeout:.1f}s waiting for " - "initial configuration from Remote Config" - ) - - # Config received during wait -- on_configuration_received() already set status + # Config not yet available — return without blocking. This provider's + # internal status stays NOT_READY; on_configuration_received() will flip it + # to READY when RC delivers the FFE_FLAGS payload. Note that openfeature-sdk + # 0.8.x dispatches PROVIDER_READY unconditionally after initialize() + # returns, even while our internal status and evaluation path are still + # waiting for config. + # AIDEV-NOTE: Do NOT block here with _config_received.wait(). Blocking + # initialize() breaks gunicorn/uWSGI pre-fork workers: when the OpenFeature + # SDK runs initialize() in a background thread, fork() kills that thread in + # child processes, leaving every worker stuck waiting forever (or timing out + # with PROVIDER_ERROR). The async path (on_configuration_received) is the + # correct contract for server SDK providers. def shutdown(self) -> None: """ @@ -211,11 +262,22 @@ def shutdown(self) -> None: except ServiceStatusError: logger.debug("Exposure writer has already stopped", exc_info=True) + # Stop the EVP flagevaluation writer (if it was started). + if self._flag_eval_evp_writer is not None: + try: + self._flag_eval_evp_writer.stop() + self._flag_eval_evp_writer.join() + logger.debug("FlagEvaluationWriter stopped") + except ServiceStatusError: + logger.debug("FlagEvaluationWriter has already stopped", exc_info=True) + self._flag_eval_evp_writer = None + self._flag_eval_evp_hook = None + # Shutdown flag evaluation metrics if self._flag_eval_metrics is not None: self._flag_eval_metrics.shutdown() self._flag_eval_metrics = None - self._flag_eval_hook = None + self._flag_eval_metrics_hook = None # Clear exposure cache self.clear_exposure_cache() @@ -260,18 +322,18 @@ def resolve_float_details( def resolve_object_details( self, flag_key: str, - default_value: typing.Union[dict, list], + default_value: ObjectFlagValue, evaluation_context: typing.Optional[EvaluationContext] = None, - ) -> FlagResolutionDetails[typing.Union[dict, list]]: + ) -> FlagResolutionDetails[ObjectFlagValue]: return self._resolve_details(flag_key, default_value, evaluation_context, VariationType.Object) def _resolve_details( self, flag_key: str, - default_value: typing.Any, + default_value: ResolvedValue, evaluation_context: typing.Optional[EvaluationContext] = None, variation_type: VariationType = VariationType.Boolean, - ) -> FlagResolutionDetails[T]: + ) -> FlagResolutionDetails[ResolvedValue]: """ Core resolution logic for all flag types. @@ -282,12 +344,18 @@ def _resolve_details( flag is not found in the configuration - Returns error with error_code and error_message on other errors """ + # AIDEV-NOTE: Stamp eval-time at provider entry so every OpenFeature exit path + # can feed the EVP flagevaluation hook first_evaluation/last_evaluation from + # evaluation time, not the later hook/flush time. + flag_metadata: dict[str, typing.Any] = {EVAL_TIMESTAMP_METADATA_KEY: int(time.time() * 1000)} + # If provider is not enabled, return default value if not self._enabled: return FlagResolutionDetails( value=default_value, reason=Reason.DISABLED, variant=None, + flag_metadata=flag_metadata, ) try: @@ -310,6 +378,7 @@ def _resolve_details( reason=Reason.ERROR, error_code=ErrorCode.PROVIDER_NOT_READY, error_message="No FFE configuration loaded", + flag_metadata=flag_metadata, ) # Handle errors from native evaluation @@ -332,6 +401,7 @@ def _resolve_details( reason=Reason.ERROR, error_code=openfeature_error_code, error_message="Flag not found", + flag_metadata=flag_metadata, ) # Other errors - return default with ERROR reason @@ -340,6 +410,7 @@ def _resolve_details( reason=Reason.ERROR, error_code=openfeature_error_code, error_message=details.error_message or "Unknown error", + flag_metadata=flag_metadata, ) # Map native ffe.Reason to OpenFeature Reason @@ -354,8 +425,7 @@ def _resolve_details( evaluation_context=evaluation_context, ) - # Build flag_metadata with allocation_key if present - flag_metadata: dict[str, typing.Any] = {} + # Add allocation_key to the provider-entry timestamp metadata when present. if details.allocation_key: flag_metadata[METADATA_ALLOCATION_KEY] = details.allocation_key @@ -372,7 +442,7 @@ def _resolve_details( # Success - return resolved value (which may be None for JSON flags) return FlagResolutionDetails( - value=details.value, + value=typing.cast(ResolvedValue, details.value), reason=reason, variant=details.variant, flag_metadata=flag_metadata, @@ -385,6 +455,7 @@ def _resolve_details( reason=Reason.ERROR, error_code=ErrorCode.GENERAL, error_message=f"Unexpected error during flag evaluation: {str(e)}", + flag_metadata=flag_metadata, ) def _report_exposure( @@ -490,16 +561,18 @@ def on_configuration_received(self) -> None: """ Called when a Remote Configuration payload is received and processed. - Updates status first, then signals the event to unblock initialize(). - Emits PROVIDER_READY for late arrivals (config received after initialize() timed out). + Updates status first, then signals the event for observers. + Emits PROVIDER_READY for late arrivals after non-blocking initialize(). + Some openfeature-sdk versions also emit PROVIDER_READY immediately after + initialize() returns; this late event is the Datadog config-loaded signal. """ if not self._config_received.is_set(): self._status = ProviderStatus.READY logger.debug("First FFE configuration received, provider is now READY") - # Emit READY for late recovery: config arrived after init timed out + # Emit READY for late recovery: config arrived after initialize() returned. self._emit_ready_event() - # Signal the event last to unblock initialize() after status is updated + # Signal the event last after status is updated. self._config_received.set() def _emit_ready_event(self) -> None: diff --git a/ddtrace/internal/openfeature/writer.py b/ddtrace/internal/openfeature/writer.py index 08a8f51bacf..8dca224d2af 100644 --- a/ddtrace/internal/openfeature/writer.py +++ b/ddtrace/internal/openfeature/writer.py @@ -9,6 +9,10 @@ from typing import TypedDict from ddtrace import config +from ddtrace.internal.evp_proxy.constants import DEFAULT_EVP_PAYLOAD_SIZE_LIMIT +from ddtrace.internal.evp_proxy.constants import EVP_PROXY_AGENT_BASE_PATH +from ddtrace.internal.evp_proxy.constants import EVP_SUBDOMAIN_HEADER_EVENT_PLATFORM_VALUE +from ddtrace.internal.evp_proxy.constants import EVP_SUBDOMAIN_HEADER_NAME from ddtrace.internal.logger import get_logger from ddtrace.internal.periodic import PeriodicService from ddtrace.internal.settings._agent import config as agent_config @@ -21,14 +25,12 @@ logger = get_logger(__name__) -EVP_PROXY_AGENT_BASE_PATH = "/evp_proxy/v2" EXPOSURE_ENDPOINT = "/api/v2/exposures" -EVP_SUBDOMAIN_HEADER_NAME = "X-Datadog-EVP-Subdomain" -EXPOSURE_SUBDOMAIN_NAME = "event-platform-intake" +EXPOSURE_SUBDOMAIN_NAME = EVP_SUBDOMAIN_HEADER_EVENT_PLATFORM_VALUE # Buffer and payload limits BUFFER_LIMIT = 1000 -PAYLOAD_SIZE_LIMIT = 5 << 20 # 5MB +PAYLOAD_SIZE_LIMIT = DEFAULT_EVP_PAYLOAD_SIZE_LIMIT # Default configuration DEFAULT_INTERVAL = 1.0 diff --git a/ddtrace/internal/settings/_supported_configurations.py b/ddtrace/internal/settings/_supported_configurations.py index adc6fddda4e..73c4bde2a29 100644 --- a/ddtrace/internal/settings/_supported_configurations.py +++ b/ddtrace/internal/settings/_supported_configurations.py @@ -222,6 +222,7 @@ "DD_FAST_BUILD", "DD_FFE_INTAKE_ENABLED", "DD_FFE_INTAKE_HEARTBEAT_INTERVAL", + "DD_FLAGGING_EVALUATION_COUNTS_ENABLED", "DD_FLASK_CACHE_SERVICE", "DD_FLASK_SERVICE", "DD_FOO_SERVICE", diff --git a/ddtrace/internal/settings/openfeature.py b/ddtrace/internal/settings/openfeature.py index a6324d1ccb8..d5f740f6445 100644 --- a/ddtrace/internal/settings/openfeature.py +++ b/ddtrace/internal/settings/openfeature.py @@ -17,6 +17,15 @@ class OpenFeatureConfig(DDConfig): default=False, ) + # Killswitch for the EVP `flagevaluation` evaluation-counts path. Default on; gates + # ONLY the EVP flagevaluation writer/hook. The existing OTel `feature_flag.evaluations` + # path is unaffected by this flag. + flagging_evaluation_counts_enabled = DDConfig.var( + bool, + "DD_FLAGGING_EVALUATION_COUNTS_ENABLED", + default=True, + ) + # Feature flag exposure intake configuration ffe_intake_enabled = DDConfig.var( bool, @@ -41,6 +50,7 @@ class OpenFeatureConfig(DDConfig): _openfeature_config_keys = [ "experimental_flagging_provider_enabled", + "flagging_evaluation_counts_enabled", "ffe_intake_enabled", "ffe_intake_heartbeat_interval", "initialization_timeout_ms", diff --git a/mypy.ini b/mypy.ini index 137791651d0..513c85ff42d 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1446,9 +1446,6 @@ disallow_incomplete_defs = false disallow_untyped_defs = false disallow_incomplete_defs = false -[mypy-ddtrace.internal.openfeature._flageval_metrics] -disallow_subclassing_any = false - [mypy-ddtrace.internal.openfeature._native] disallow_untyped_calls = false disallow_untyped_defs = false diff --git a/pyproject.toml b/pyproject.toml index 3215f413af9..71e94a58816 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -103,6 +103,7 @@ lint = [ "mypy==1.15.0", "coverage==7.3.0", "envier==0.6.1", + "openfeature-sdk~=0.8.0", "types-docutils==0.19.1.1", "types-protobuf==3.20.4.5", "types-PyYAML==6.0.12.2", diff --git a/releasenotes/notes/openfeature-flagevaluation-evp-0f1a2b3c4d5e6f70.yaml b/releasenotes/notes/openfeature-flagevaluation-evp-0f1a2b3c4d5e6f70.yaml new file mode 100644 index 00000000000..509aa2c1b01 --- /dev/null +++ b/releasenotes/notes/openfeature-flagevaluation-evp-0f1a2b3c4d5e6f70.yaml @@ -0,0 +1,9 @@ +--- +features: + - | + openfeature: the experimental OpenFeature provider now reports aggregated + server-side flag evaluations to Datadog (the EVP ``flagevaluation`` track), + lighting up server-side flag-evaluation observability. The existing + ``feature_flag.evaluations`` OpenTelemetry metric behavior is unchanged. The + new emission is gated by ``DD_FLAGGING_EVALUATION_COUNTS_ENABLED`` (default + enabled). diff --git a/supported-configurations.json b/supported-configurations.json index 9a9d27c2204..d9f94632e42 100644 --- a/supported-configurations.json +++ b/supported-configurations.json @@ -1654,6 +1654,14 @@ "default": "1.0" } ], + "DD_FLAGGING_EVALUATION_COUNTS_ENABLED": [ + { + "implementation": "A", + "type": "boolean", + "default": "true", + "experimental": true + } + ], "DD_FLASK_CACHE_SERVICE": [ { "implementation": "A", diff --git a/tests/openfeature/conftest.py b/tests/openfeature/conftest.py index 8704d6a0bde..a13d86bfae7 100644 --- a/tests/openfeature/conftest.py +++ b/tests/openfeature/conftest.py @@ -1,22 +1,3 @@ """ Shared fixtures for openfeature tests. """ - -import pytest - -from tests.utils import override_global_config - - -@pytest.fixture(autouse=True) -def fast_initialization_timeout(): - """ - Override the provider initialization timeout to 100ms for all tests. - - The production default is 30s (matching other SDKs), but that makes any - test that calls api.set_provider() without pre-loaded config hang for 30s. - Tests that need to verify timeout/blocking behaviour set their own explicit - initialization_timeout= on DataDogProvider() directly, which takes priority - over the config value. - """ - with override_global_config({"initialization_timeout_ms": 100}): - yield diff --git a/tests/openfeature/test_flag_eval_evp_hook.py b/tests/openfeature/test_flag_eval_evp_hook.py new file mode 100644 index 00000000000..a63ff7eba0e --- /dev/null +++ b/tests/openfeature/test_flag_eval_evp_hook.py @@ -0,0 +1,381 @@ +""" +Tests for FlagEvalEVPHook — finally_after cheap capture + non-blocking enqueue. + +Validates the hook design: +- finally_after does cheap capture only (no aggregation, no I/O) +- variant=None → runtime_default_used True +- eval_time_ms from metadata["dd.eval.timestamp_ms"] when present; fallback to hook-fire time +- DD_FLAGGING_EVALUATION_COUNTS_ENABLED killswitch gates the EVP path only +""" + +import json +import os +import time +import typing +from unittest import mock + +from openfeature.evaluation_context import EvaluationContext +from openfeature.exception import ErrorCode +from openfeature.flag_evaluation import FlagEvaluationDetails +from openfeature.flag_evaluation import FlagType +from openfeature.flag_evaluation import Reason +from openfeature.hook import HookContext +import pytest + + +def _make_hook_context( + flag_key: str = "my-flag", + targeting_key: str = "user-1", + attrs: dict = None, +) -> HookContext: + ctx = EvaluationContext(targeting_key=targeting_key, attributes=attrs or {}) + return HookContext( + flag_key=flag_key, + flag_type=FlagType.BOOLEAN, + default_value=False, + evaluation_context=ctx, + ) + + +def _make_details( + flag_key: str = "my-flag", + value: typing.Any = True, + variant: typing.Optional[str] = "on", + reason: typing.Optional[Reason] = Reason.TARGETING_MATCH, + flag_metadata: dict = None, + error_message: str = None, + error_code: typing.Optional[ErrorCode] = None, +) -> FlagEvaluationDetails: + return FlagEvaluationDetails( + flag_key=flag_key, + value=value, + variant=variant, + reason=reason, + flag_metadata=flag_metadata or {}, + error_message=error_message, + error_code=error_code, + ) + + +@pytest.fixture +def writer(): + from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter + + return mock.MagicMock(spec=FlagEvaluationWriter) + + +@pytest.fixture +def hook(writer): + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + + return FlagEvalEVPHook(writer=writer) + + +class TestFlagEvalEVPHook: + def test_finally_after_calls_writer_enqueue_once(self, hook, writer): + """finally_after must call writer.enqueue exactly once per evaluation.""" + hc = _make_hook_context() + details = _make_details() + hook.finally_after(hc, details, {}) + writer.enqueue.assert_called_once() + + def test_finally_after_enqueues_correct_flag_key(self, hook, writer): + hc = _make_hook_context(flag_key="test-flag") + details = _make_details(flag_key="test-flag") + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.flag_key == "test-flag" + + def test_finally_after_enqueues_correct_variant(self, hook, writer): + hc = _make_hook_context() + details = _make_details(variant="control") + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.variant == "control" + + def test_finally_after_none_variant_sets_runtime_default(self, hook, writer): + """None variant → runtime_default=True.""" + hc = _make_hook_context() + details = _make_details(variant=None) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.runtime_default is True + assert event.variant == "" + + def test_finally_after_present_variant_not_runtime_default(self, hook, writer): + hc = _make_hook_context() + details = _make_details(variant="on") + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.runtime_default is False + + def test_finally_after_present_variant_with_type_mismatch_uses_variant(self, hook, writer): + hc = _make_hook_context() + details = _make_details(variant="wrong-type", error_code=ErrorCode.TYPE_MISMATCH) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.runtime_default is False + assert event.variant == "wrong-type" + + def test_finally_after_does_not_enqueue_openfeature_reason(self, hook, writer): + """OpenFeature reason is not an EVP field and must not enter the event snapshot.""" + hc = _make_hook_context() + details = _make_details(reason=Reason.TARGETING_MATCH) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert not hasattr(event, "reason") + + def test_finally_after_eval_time_from_metadata(self, hook, writer): + """Eval-time must come from metadata["dd.eval.timestamp_ms"] when present.""" + stamp = int(time.time() * 1000) - 500 # 500 ms in the past + hc = _make_hook_context() + details = _make_details(flag_metadata={"dd.eval.timestamp_ms": stamp}) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.eval_time_ms == stamp + + def test_finally_after_eval_time_fallback_to_hook_fire(self, hook, writer): + """When metadata key absent, fallback to hook-fire time (within 1 second).""" + before = int(time.time() * 1000) + hc = _make_hook_context() + details = _make_details(flag_metadata={}) + hook.finally_after(hc, details, {}) + after = int(time.time() * 1000) + event = writer.enqueue.call_args[0][0] + assert before <= event.eval_time_ms <= after + 100 + + def test_finally_after_extracts_targeting_key(self, hook, writer): + hc = _make_hook_context(targeting_key="user-99") + details = _make_details() + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.targeting_key == "user-99" + + def test_finally_after_extracts_attrs_shallow_copy(self, hook, writer): + attrs = {"tier": "premium", "region": "us-west"} + hc = _make_hook_context(attrs=attrs) + details = _make_details() + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.attrs == attrs + # Must be a copy, not the same object. + assert event.attrs is not attrs + + def test_finally_after_extracts_allocation_key(self, hook, writer): + hc = _make_hook_context() + details = _make_details(flag_metadata={"allocation_key": "alloc-xyz"}) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.allocation_key == "alloc-xyz" + + def test_finally_after_falls_back_to_error_code_when_message_absent(self, hook, writer): + hc = _make_hook_context() + details = _make_details( + variant=None, + reason=Reason.ERROR, + error_code=ErrorCode.FLAG_NOT_FOUND, + ) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.error_message == ErrorCode.FLAG_NOT_FOUND.value + + def test_finally_after_does_no_aggregation_on_hook_thread(self, hook, writer): + """The hook must call enqueue only — not build payloads or aggregate.""" + # writer.enqueue is a mock; the only call from finally_after must be enqueue. + hc = _make_hook_context() + details = _make_details() + hook.finally_after(hc, details, {}) + # Confirm ONLY enqueue was called on the writer (no periodic, no aggregate, no send). + called_methods = {c[0] for c in writer.method_calls} + assert called_methods == {"enqueue"}, f"Expected only enqueue to be called, got: {called_methods}" + + def test_finally_after_does_not_propagate_exceptions(self, hook, writer): + """Hook must swallow exceptions — best-effort telemetry.""" + writer.enqueue.side_effect = RuntimeError("boom") + hc = _make_hook_context() + details = _make_details() + # Must not raise. + hook.finally_after(hc, details, {}) + + +class TestAsyncBoundary: + """Prove the hook does NOT aggregate on the eval call path. + + The hook may only enqueue a snapshot; canonical-key/aggregate must run later in the + writer's background worker, never on the eval thread. + """ + + def test_aggregate_not_called_on_hook_path(self): + """Spy on the REAL writer's _aggregate — it must NOT run during finally_after.""" + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter + + real_writer = FlagEvaluationWriter(interval=10.0) + hook = FlagEvalEVPHook(writer=real_writer) + + with mock.patch.object(real_writer, "_aggregate", wraps=real_writer._aggregate) as spy_aggregate: + hc = _make_hook_context(attrs={"tier": "premium", "region": "us"}) + details = _make_details(flag_metadata={"allocation_key": "a1"}) + hook.finally_after(hc, details, {}) + + # The event must be queued but NOT aggregated on the hook path. + spy_aggregate.assert_not_called() + assert real_writer._queue.qsize() == 1 + # The aggregation maps are still empty — no keying/aggregation happened yet. + assert real_writer._full == {} + assert real_writer._degraded == {} + + # Aggregation only happens when the worker drains (periodic), off the hook path. + # _aggregate fires exactly once during the drain, producing one full-tier row. + with mock.patch.object(real_writer, "_aggregate", wraps=real_writer._aggregate) as spy_drain: + with mock.patch.object(real_writer, "_send_payload") as mock_send: + real_writer.periodic() + spy_drain.assert_called_once() + decoded = json.loads(mock_send.call_args[0][0]) + assert len(decoded["flagEvaluations"]) == 1 + + def test_canonical_key_not_computed_on_hook_path(self): + """canonical_context_key (the keying cost) must not run during finally_after.""" + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + import ddtrace.internal.openfeature._flagevaluation_writer as writer_mod + from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter + + real_writer = FlagEvaluationWriter(interval=10.0) + hook = FlagEvalEVPHook(writer=real_writer) + + with mock.patch.object(writer_mod, "canonical_context_key", wraps=writer_mod.canonical_context_key) as spy_key: + hc = _make_hook_context(attrs={"a": "b"}) + hook.finally_after(hc, _make_details(), {}) + spy_key.assert_not_called() + + +class TestMetadataSourceMatchesOTelHook: + """EVP hook reads allocation-key/eval metadata from the SAME source as the OTel hook. + + The existing OTel FlagEvalMetricsHook reads allocation_key from + ``details.flag_metadata[METADATA_ALLOCATION_KEY]``. The EVP hook must read from the + identical source so the two paths agree byte-for-byte on metadata. + """ + + def test_allocation_key_metadata_key_matches_otel_hook(self): + from ddtrace.internal.openfeature import _flageval_metrics + from ddtrace.internal.openfeature import _flagevaluation_writer + + # Same metadata key constant in both modules. + assert _flagevaluation_writer.METADATA_ALLOCATION_KEY == _flageval_metrics.METADATA_ALLOCATION_KEY + + def test_evp_hook_reads_allocation_from_details_flag_metadata(self, hook, writer): + """EVP hook reads allocation_key from details.flag_metadata (not hook_context).""" + from ddtrace.internal.openfeature._flageval_metrics import METADATA_ALLOCATION_KEY + + hc = _make_hook_context() + details = _make_details(flag_metadata={METADATA_ALLOCATION_KEY: "alloc-from-details"}) + hook.finally_after(hc, details, {}) + event = writer.enqueue.call_args[0][0] + assert event.allocation_key == "alloc-from-details" + + def test_metrics_and_evp_hooks_extract_same_allocation_key(self): + """Drive both hooks with identical details; both must surface the same allocation key.""" + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + from ddtrace.internal.openfeature._flageval_metrics import METADATA_ALLOCATION_KEY + from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetrics + from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetricsHook + from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter + + details = _make_details(flag_metadata={METADATA_ALLOCATION_KEY: "shared-alloc"}) + hc = _make_hook_context() + + # OTel side: capture what FlagEvalMetrics.record received as allocation_key. + metrics = mock.MagicMock(spec=FlagEvalMetrics) + otel_hook = FlagEvalMetricsHook(metrics) + otel_hook.finally_after(hc, details, {}) + otel_alloc = metrics.record.call_args.kwargs["allocation_key"] + + # EVP side: capture what the EVP hook enqueued as allocation_key. + evp_writer = mock.MagicMock(spec=FlagEvaluationWriter) + evp_hook = FlagEvalEVPHook(evp_writer) + evp_hook.finally_after(hc, details, {}) + evp_alloc = evp_writer.enqueue.call_args[0][0].allocation_key + + assert otel_alloc == evp_alloc == "shared-alloc" + + +class TestKillswitchGating: + def test_default_enabled_registers_evp_hook(self): + """Default (no env var set) must register the EVP hook + writer.""" + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + + env = {k: v for k, v in os.environ.items() if k != "DD_FLAGGING_EVALUATION_COUNTS_ENABLED"} + # No env var → enabled by default. + with mock.patch.dict(os.environ, env, clear=True): + from tests.utils import override_global_config + + with override_global_config({"experimental_flagging_provider_enabled": True}): + from ddtrace.internal.openfeature._provider import DataDogProvider + + provider = DataDogProvider() + assert provider._flag_eval_evp_writer is not None + assert provider._flag_eval_evp_hook is not None + assert isinstance(provider._flag_eval_evp_hook, FlagEvalEVPHook) + + def test_killswitch_false_does_not_register_evp_hook(self): + """DD_FLAGGING_EVALUATION_COUNTS_ENABLED=false must suppress EVP hook (killswitch).""" + with mock.patch.dict(os.environ, {"DD_FLAGGING_EVALUATION_COUNTS_ENABLED": "false"}): + from tests.utils import override_global_config + + with override_global_config({"experimental_flagging_provider_enabled": True}): + from ddtrace.internal.openfeature._provider import DataDogProvider + + provider = DataDogProvider() + assert provider._flag_eval_evp_writer is None + assert provider._flag_eval_evp_hook is None + + def test_killswitch_false_does_not_affect_otel_hook(self): + """Killswitch must not suppress the OTel FlagEvalMetricsHook (OTel non-regression).""" + with mock.patch.dict(os.environ, {"DD_FLAGGING_EVALUATION_COUNTS_ENABLED": "false"}): + from tests.utils import override_global_config + + with override_global_config( + {"experimental_flagging_provider_enabled": True, "_otel_metrics_enabled": True} + ): + from ddtrace.internal.openfeature._provider import DataDogProvider + + provider = DataDogProvider() + # OTel hook still present. + assert provider._flag_eval_metrics_hook is not None + # EVP hook absent. + assert provider._flag_eval_evp_hook is None + # get_provider_hooks still returns the OTel hook. + hooks = provider.get_provider_hooks() + assert len(hooks) == 1 + assert hooks[0] is provider._flag_eval_metrics_hook + + def test_provider_shutdown_joins_evp_writer_final_flush(self): + """Provider shutdown waits for FlagEvaluationWriter.on_shutdown final flush.""" + from tests.utils import override_global_config + + with override_global_config({"experimental_flagging_provider_enabled": True}): + with mock.patch("ddtrace.internal.openfeature._provider.stop_exposure_writer"): + with mock.patch("ddtrace.internal.openfeature._provider.FlagEvaluationWriter") as writer_cls: + from ddtrace.internal.openfeature._provider import DataDogProvider + + writer = writer_cls.return_value + provider = DataDogProvider() + + provider.shutdown() + + writer.stop.assert_called_once() + writer.join.assert_called_once() + assert writer.mock_calls.index(mock.call.stop()) < writer.mock_calls.index(mock.call.join()) + + def test_killswitch_enabled_true_registers_evp_hook(self): + """DD_FLAGGING_EVALUATION_COUNTS_ENABLED=true must register the EVP hook.""" + with mock.patch.dict(os.environ, {"DD_FLAGGING_EVALUATION_COUNTS_ENABLED": "true"}): + from tests.utils import override_global_config + + with override_global_config({"experimental_flagging_provider_enabled": True}): + from ddtrace.internal.openfeature._provider import DataDogProvider + + provider = DataDogProvider() + assert provider._flag_eval_evp_writer is not None + assert provider._flag_eval_evp_hook is not None diff --git a/tests/openfeature/test_flag_eval_metrics.py b/tests/openfeature/test_flag_eval_metrics.py index 3e656f81283..9e63f7ce480 100644 --- a/tests/openfeature/test_flag_eval_metrics.py +++ b/tests/openfeature/test_flag_eval_metrics.py @@ -1,7 +1,7 @@ """ Tests for flag evaluation metrics tracking. -Tests the FlagEvalMetrics class and FlagEvalHook that emit +Tests the FlagEvalMetrics class and FlagEvalMetricsHook that emit feature_flag.evaluations OTel metrics on every flag evaluation. """ @@ -21,8 +21,8 @@ from ddtrace.internal.openfeature._flageval_metrics import ATTR_REASON from ddtrace.internal.openfeature._flageval_metrics import ATTR_VARIANT from ddtrace.internal.openfeature._flageval_metrics import METADATA_ALLOCATION_KEY -from ddtrace.internal.openfeature._flageval_metrics import FlagEvalHook from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetrics +from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetricsHook from ddtrace.internal.openfeature._native import process_ffe_configuration from ddtrace.openfeature import DataDogProvider from tests.openfeature.config_helpers import create_boolean_flag @@ -260,14 +260,14 @@ def test_record_all_error_types(self): assert error_types == {"flag_not_found", "type_mismatch", "parse_error", "general"}, error_types -class TestFlagEvalHook: - """Test FlagEvalHook class.""" +class TestFlagEvalMetricsHook: + """Test FlagEvalMetricsHook class.""" def test_finally_after_calls_record(self): """finally_after should call metrics.record with correct arguments.""" mock_metrics = MagicMock(spec=FlagEvalMetrics) - hook = FlagEvalHook(mock_metrics) + hook = FlagEvalMetricsHook(mock_metrics) # Create mock hook context hook_context = MagicMock(spec=HookContext) @@ -296,7 +296,7 @@ def test_finally_after_with_error(self): """finally_after should pass error_code when present.""" mock_metrics = MagicMock(spec=FlagEvalMetrics) - hook = FlagEvalHook(mock_metrics) + hook = FlagEvalMetricsHook(mock_metrics) hook_context = MagicMock(spec=HookContext) hook_context.flag_key = "missing-flag" @@ -319,7 +319,7 @@ def test_finally_after_without_allocation_key(self): """finally_after should handle missing allocation_key.""" mock_metrics = MagicMock(spec=FlagEvalMetrics) - hook = FlagEvalHook(mock_metrics) + hook = FlagEvalMetricsHook(mock_metrics) hook_context = MagicMock(spec=HookContext) hook_context.flag_key = "test-flag" @@ -354,23 +354,25 @@ def clear_config(self): yield _set_ffe_config(None) - def test_provider_has_flag_eval_hook(self, provider): - """Provider should have flag evaluation hook when enabled.""" - assert provider._flag_eval_hook is not None + def test_provider_has_flag_eval_metrics_hook(self, provider): + """Provider should have the OTel flag evaluation metrics hook when enabled.""" + assert provider._flag_eval_metrics_hook is not None assert provider._flag_eval_metrics is not None - def test_get_provider_hooks_returns_flag_eval_hook(self, provider): - """get_provider_hooks should return the flag eval hook.""" + def test_get_provider_hooks_returns_flag_eval_metrics_hook(self, provider): + """get_provider_hooks should return the OTel metrics hook and optionally the EVP hook.""" hooks = provider.get_provider_hooks() - assert len(hooks) == 1 - assert hooks[0] is provider._flag_eval_hook + # The OTel FlagEvalMetricsHook is always first when the provider is enabled. + # The FlagEvalEVPHook is also registered by default (DD_FLAGGING_EVALUATION_COUNTS_ENABLED=true). + assert len(hooks) >= 1 + assert hooks[0] is provider._flag_eval_metrics_hook def test_provider_disabled_has_no_hooks(self): """Provider should not have hooks when disabled.""" with override_global_config({"experimental_flagging_provider_enabled": False}): provider = DataDogProvider() - assert provider._flag_eval_hook is None + assert provider._flag_eval_metrics_hook is None assert provider._flag_eval_metrics is None assert provider.get_provider_hooks() == [] @@ -381,7 +383,7 @@ def test_shutdown_cleans_up_metrics(self, provider): provider.shutdown() assert provider._flag_eval_metrics is None - assert provider._flag_eval_hook is None + assert provider._flag_eval_metrics_hook is None def test_flag_metadata_includes_allocation_key(self, provider): """FlagResolutionDetails should include allocation_key in flag_metadata.""" diff --git a/tests/openfeature/test_flagevaluation_e2e.py b/tests/openfeature/test_flagevaluation_e2e.py new file mode 100644 index 00000000000..e04a1699c62 --- /dev/null +++ b/tests/openfeature/test_flagevaluation_e2e.py @@ -0,0 +1,209 @@ +""" +End-to-end tests for the EVP `flagevaluation` path through the real provider eval path. + +These cover the lifecycle/exit paths that unit tests of the hook/writer in isolation +cannot prove: + +- The EVP hook actually fires on the provider's REAL OpenFeature evaluation entrypoint + (driven through the OpenFeature client, not by calling the hook directly). +- ALL evaluation exit paths are covered — success, native engine error, runtime default + (flag-not-found / no-config), and disabled. +- The OTel `feature_flag.evaluations` path is preserved alongside the EVP path (non-regression). +""" + +from unittest import mock + +from openfeature import api +from openfeature.evaluation_context import EvaluationContext +import pytest + +from ddtrace.internal.openfeature._config import _set_ffe_config +from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_TIMESTAMP_METADATA_KEY +from ddtrace.internal.openfeature._native import process_ffe_configuration +from ddtrace.openfeature import DataDogProvider +from tests.openfeature.config_helpers import create_boolean_flag +from tests.openfeature.config_helpers import create_config +from tests.openfeature.config_helpers import create_string_flag +from tests.utils import override_global_config + + +@pytest.fixture(autouse=True) +def clear_config(): + _set_ffe_config(None) + yield + _set_ffe_config(None) + + +@pytest.fixture +def provider_and_client(): + """Set up a DataDogProvider with the EVP path enabled, returning (provider, client). + + The writer's background thread is NOT started (we drive aggregation synchronously via + periodic()), so the eval path enqueues and we drain deterministically in the test. + """ + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider() + # Sanity: the EVP writer/hook are wired (killswitch default on). + assert provider._flag_eval_evp_writer is not None + assert provider._flag_eval_evp_hook is not None + + api.set_provider(provider) + client = api.get_client() + try: + yield provider, client + finally: + api.shutdown() + + +def _drain(provider): + """Aggregate everything the eval path enqueued, returning the emitted rows.""" + writer = provider._flag_eval_evp_writer + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + if not mock_send.called: + return [] + import json + + decoded = json.loads(mock_send.call_args[0][0]) + return decoded.get("flagEvaluations", []) + + +class TestFlagEvalEVPHookFiresOnRealEvalPath: + """The EVP hook fires on the provider's real evaluation entrypoint.""" + + def test_evp_hook_registered_in_provider_hooks(self, provider_and_client): + provider, _ = provider_and_client + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + + hooks = provider.get_provider_hooks() + assert any(isinstance(h, FlagEvalEVPHook) for h in hooks) + + def test_success_eval_enqueues_and_emits_row(self, provider_and_client): + provider, client = provider_and_client + config = create_config(create_boolean_flag("evp-success", enabled=True, default_value=True)) + process_ffe_configuration(config) + + # Real client eval — the OpenFeature SDK runs the registered finally_after hook. + assert client.get_boolean_value("evp-success", False) is True + + rows = _drain(provider) + keys = {r["flag"]["key"] for r in rows} + assert "evp-success" in keys + row = next(r for r in rows if r["flag"]["key"] == "evp-success") + assert row["evaluation_count"] >= 1 + # Successful eval -> a real variant object, no runtime_default. + assert "variant" in row and set(row["variant"].keys()) == {"key"} + assert row.get("runtime_default_used", False) is False + + def test_variant_is_resolution_variant_not_value(self, provider_and_client): + """End-to-end: emitted variant == resolution variant key, distinct from value.""" + provider, client = provider_and_client + # String flag: value "blue", variant key "blue" (here equal); use details to confirm + # the EVP row variant matches details.variant exactly. + config = create_config(create_string_flag("evp-variant", "blue", enabled=True)) + process_ffe_configuration(config) + + details = client.get_string_details("evp-variant", "red") + rows = _drain(provider) + row = next(r for r in rows if r["flag"]["key"] == "evp-variant") + assert row["variant"]["key"] == details.variant + + +class TestFlagEvalLoggingExitPathsCovered: + """success / engine-error / runtime-default / disabled exit paths are all captured.""" + + def test_flag_not_found_runtime_default_path(self, provider_and_client): + provider, client = provider_and_client + # A config is loaded but the requested flag isn't in it -> native FlagNotFound -> ERROR. + config = create_config(create_boolean_flag("present-flag", enabled=True)) + process_ffe_configuration(config) + + assert client.get_boolean_value("absent-flag", False) is False + + rows = _drain(provider) + row = next((r for r in rows if r["flag"]["key"] == "absent-flag"), None) + assert row is not None, "flag-not-found eval must still emit a flagevaluation row" + # No variant -> runtime_default_used True. + assert row.get("runtime_default_used") is True + assert "variant" not in row + + def test_no_config_provider_not_ready_path(self, provider_and_client): + provider, client = provider_and_client + _set_ffe_config(None) # no RC config at all -> PROVIDER_NOT_READY error path + + assert client.get_boolean_value("no-config-flag", True) is True + + rows = _drain(provider) + row = next((r for r in rows if r["flag"]["key"] == "no-config-flag"), None) + assert row is not None, "no-config eval must still emit a flagevaluation row" + assert row.get("runtime_default_used") is True + + def test_type_mismatch_error_path(self, provider_and_client): + provider, client = provider_and_client + config = create_config(create_string_flag("str-flag", "hello", enabled=True)) + process_ffe_configuration(config) + + # Evaluate a string flag as boolean -> type mismatch ERROR. + details = client.get_boolean_details("str-flag", False) + assert details.value is False + + rows = _drain(provider) + row = next((r for r in rows if r["flag"]["key"] == "str-flag"), None) + assert row is not None, "type-mismatch eval must still emit a flagevaluation row" + assert row.get("runtime_default_used") is True + assert row["first_evaluation"] == details.flag_metadata[EVAL_TIMESTAMP_METADATA_KEY] + assert row["last_evaluation"] == details.flag_metadata[EVAL_TIMESTAMP_METADATA_KEY] + + def test_disabled_flag_path(self, provider_and_client): + provider, client = provider_and_client + config = create_config(create_boolean_flag("disabled-flag", enabled=False, default_value=False)) + process_ffe_configuration(config) + + assert client.get_boolean_value("disabled-flag", False) is False + + rows = _drain(provider) + # Disabled flag still produces a flagevaluation row. + row = next((r for r in rows if r["flag"]["key"] == "disabled-flag"), None) + assert row is not None, "disabled-flag eval must still emit a flagevaluation row" + + def test_targeting_key_and_context_captured_on_success(self, provider_and_client): + provider, client = provider_and_client + config = create_config(create_boolean_flag("ctx-flag", enabled=True, default_value=True)) + process_ffe_configuration(config) + + ctx = EvaluationContext(targeting_key="user-77", attributes={"tier": "gold"}) + client.get_boolean_value("ctx-flag", False, ctx) + + rows = _drain(provider) + row = next(r for r in rows if r["flag"]["key"] == "ctx-flag") + assert row.get("targeting_key") == "user-77" + assert row["context"]["evaluation"]["tier"] == "gold" + + +class TestOTelNonRegressionAlongsideFlagEvalEVP: + """The EVP path must NOT change which hooks the provider registers for OTel.""" + + def test_both_metrics_and_evp_hooks_registered(self, provider_and_client): + provider, _ = provider_and_client + from ddtrace.internal.openfeature._flag_eval_evp_hook import FlagEvalEVPHook + from ddtrace.internal.openfeature._flageval_metrics import FlagEvalMetricsHook + + hooks = provider.get_provider_hooks() + assert any(isinstance(h, FlagEvalMetricsHook) for h in hooks), "OTel hook must remain registered" + assert any(isinstance(h, FlagEvalEVPHook) for h in hooks), "EVP hook must be registered" + + def test_eval_drives_otel_record_and_evp_enqueue_together(self, provider_and_client): + """A single eval feeds BOTH the OTel metric record and the EVP enqueue.""" + provider, client = provider_and_client + config = create_config(create_boolean_flag("dual-flag", enabled=True, default_value=True)) + process_ffe_configuration(config) + + # Spy the OTel metrics record and the EVP writer enqueue. + with ( + mock.patch.object(provider._flag_eval_metrics, "record") as otel_record, + mock.patch.object(provider._flag_eval_evp_writer, "enqueue") as evp_enqueue, + ): + client.get_boolean_value("dual-flag", False) + + assert otel_record.called, "OTel feature_flag.evaluations record must still fire" + assert evp_enqueue.called, "EVP flagevaluation enqueue must fire" diff --git a/tests/openfeature/test_flagevaluation_writer.py b/tests/openfeature/test_flagevaluation_writer.py new file mode 100644 index 00000000000..5a63886cbdc --- /dev/null +++ b/tests/openfeature/test_flagevaluation_writer.py @@ -0,0 +1,992 @@ +""" +Unit tests for FlagEvaluationWriter — two-tier aggregation, canonical key, EVP transport. + +Tests validate the two-tier aggregation spec: +- canonical_context_key: sorted, type-tagged, length-delimited (NOT a hash) +- Two-tier aggregation (full → degraded → drop-counted) +- Caps GLOBAL_CAP=131072 / PER_FLAG_CAP=10000 / DEGRADED_CAP=32768 +- 256-field / 256-char context pruning +- runtime_default_used from absent/None variant +- Non-blocking enqueue with drop-and-count on queue.Full +- EVP POST to /evp_proxy/v2/api/v2/flagevaluation with correct headers +""" + +from datetime import datetime +from datetime import timezone +from decimal import Decimal +import json +import time +from unittest import mock + +import pytest + +from ddtrace.internal.openfeature._flagevaluation_writer import DEGRADED_CAP +from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_SCALE_DEGRADED_BUCKET_TARGET +from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_SCALE_FULL_BUCKET_TARGET +from ddtrace.internal.openfeature._flagevaluation_writer import EVAL_SCALE_PER_FLAG_BUCKET_TARGET +from ddtrace.internal.openfeature._flagevaluation_writer import EVP_SUBDOMAIN_HEADER_NAME +from ddtrace.internal.openfeature._flagevaluation_writer import EVP_SUBDOMAIN_VALUE +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_DEGRADED_METRIC +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_DROPPED_METRIC +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_REASON_CARDINALITY_CAP +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_REASON_CLOSED +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_REASON_DEGRADED_CAP +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_REASON_PAYLOAD_LIMIT +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_REASON_QUEUE_OVERFLOW +from ddtrace.internal.openfeature._flagevaluation_writer import FLAG_EVALUATION_SPLITS_METRIC +from ddtrace.internal.openfeature._flagevaluation_writer import FLAGEVALUATIONS_ENDPOINT +from ddtrace.internal.openfeature._flagevaluation_writer import GLOBAL_CAP +from ddtrace.internal.openfeature._flagevaluation_writer import MAX_CONTEXT_FIELDS +from ddtrace.internal.openfeature._flagevaluation_writer import MAX_FIELD_LENGTH +from ddtrace.internal.openfeature._flagevaluation_writer import PER_FLAG_CAP +from ddtrace.internal.openfeature._flagevaluation_writer import QUEUE_SIZE +from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter +from ddtrace.internal.openfeature._flagevaluation_writer import _build_payloads_with_stats +from ddtrace.internal.openfeature._flagevaluation_writer import _EvalEvent +from ddtrace.internal.openfeature._flagevaluation_writer import canonical_context_key +from ddtrace.internal.openfeature._flagevaluation_writer import flatten_and_prune_context +from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE + + +TELEMETRY_COUNT_PATCH = "ddtrace.internal.openfeature._flagevaluation_writer.telemetry_writer.add_count_metric" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_event( + flag_key: str = "my-flag", + variant: str = "on", + allocation_key: str = "alloc-1", + targeting_key: str = "user-1", + attrs: dict = None, + runtime_default: bool = False, + error_message: str = "", + eval_time_ms: int = None, +) -> _EvalEvent: + if eval_time_ms is None: + eval_time_ms = int(time.time() * 1000) + return _EvalEvent( + flag_key=flag_key, + variant=variant, + allocation_key=allocation_key, + targeting_key=targeting_key, + attrs=attrs or {}, + runtime_default=runtime_default, + error_message=error_message, + eval_time_ms=eval_time_ms, + ) + + +def _wait_until(predicate, timeout: float = 2.0) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.01) + return predicate() + + +def _assert_count_metric(mock_add_count, name: str, value: int, reason: str = None) -> None: + tags = (("reason", reason),) if reason else tuple() + mock_add_count.assert_any_call(TELEMETRY_NAMESPACE.TRACERS, name, value, tags) + + +def _assert_no_count_metric(mock_add_count, name: str, reason: str = None) -> None: + tags = (("reason", reason),) if reason else tuple() + for call in mock_add_count.call_args_list: + if call.args == (TELEMETRY_NAMESPACE.TRACERS, name, mock.ANY, tags): + raise AssertionError(f"unexpected metric {name} tags={tags}: {call}") + + +@pytest.fixture +def writer(): + """Create a FlagEvaluationWriter that is NOT started (no background thread).""" + return FlagEvaluationWriter(interval=10.0) + + +# --------------------------------------------------------------------------- +# canonical_context_key tests +# --------------------------------------------------------------------------- + + +class TestCanonicalContextKey: + def test_empty_attrs_returns_empty_string(self): + assert canonical_context_key({}) == "" + assert canonical_context_key(None) == "" + + def test_same_dict_same_key(self): + attrs = {"user": "alice", "tier": "premium"} + assert canonical_context_key(attrs) == canonical_context_key(attrs) + + def test_different_insertion_order_same_key(self): + """Dict order must not affect the key (sorted).""" + a = {"b": "2", "a": "1"} + b = {"a": "1", "b": "2"} + assert canonical_context_key(a) == canonical_context_key(b) + + def test_int_vs_string_distinct_keys(self): + """int 1 vs string '1' must produce different keys (type-tagged keys).""" + k_int = canonical_context_key({"x": 1}) + k_str = canonical_context_key({"x": "1"}) + assert k_int != k_str, "int 1 and str '1' must not alias into the same bucket" + + def test_bool_vs_int_distinct_keys(self): + k_bool = canonical_context_key({"x": True}) + k_int = canonical_context_key({"x": 1}) + assert k_bool != k_int + + def test_float_vs_int_distinct_keys(self): + k_float = canonical_context_key({"x": 1.0}) + k_int = canonical_context_key({"x": 1}) + assert k_float != k_int + + def test_datetime_vs_string_distinct_keys(self): + timestamp = datetime(2026, 6, 23, 12, 0, tzinfo=timezone.utc) + k_datetime = canonical_context_key({"x": timestamp}) + k_string = canonical_context_key({"x": timestamp.isoformat()}) + assert k_datetime != k_string + + def test_value_with_equals_or_newline_no_boundary_confusion(self): + r"""'=' and '\n' in values must not fake a field boundary (length-prefix protocol).""" + k_with = canonical_context_key({"a": "foo=bar\nbaz"}) + k_without = canonical_context_key({"a": "foo", "bar\nbaz": ""}) + assert k_with != k_without + + def test_no_hashlib_or_md5_used(self): + """Verify no hash function is used by inspecting the module source.""" + import inspect + + import ddtrace.internal.openfeature._flagevaluation_writer as mod_src + + src = inspect.getsource(mod_src) + assert "hashlib" not in src, "hashlib must not appear in the writer" + assert "md5" not in src, "md5 must not appear in the writer" + + def test_returns_string_not_bytes(self): + k = canonical_context_key({"x": "y"}) + assert isinstance(k, str) + + +# --------------------------------------------------------------------------- +# flatten_and_prune_context tests +# --------------------------------------------------------------------------- + + +class TestFlattenAndPruneContext: + def test_empty_returns_empty(self): + assert flatten_and_prune_context({}) == {} + + def test_flat_attrs_passthrough(self): + attrs = {"a": "1", "b": 2} + result = flatten_and_prune_context(attrs) + assert result == attrs + + def test_nested_attrs_flattened(self): + attrs = {"user": {"id": "u1", "tier": "free"}} + result = flatten_and_prune_context(attrs) + assert "user.id" in result + assert "user.tier" in result + assert result["user.id"] == "u1" + + def test_nested_sequences_are_flattened_with_bracket_indexes(self): + attrs = {"cohorts": ["beta", {"name": "ga"}], "user": {"roles": ["admin"]}} + result = flatten_and_prune_context(attrs) + assert result["cohorts[0]"] == "beta" + assert result["cohorts[1].name"] == "ga" + assert result["user.roles[0]"] == "admin" + + def test_cycles_do_not_recurse_forever(self): + attrs = {} + attrs["self"] = attrs + result = flatten_and_prune_context(attrs) + assert result == {} + + def test_dedicated_targeting_key_aliases_are_not_context_attrs(self): + attrs = {"targetingKey": "user-1", "targeting_key": "user-1", "tier": "premium"} + result = flatten_and_prune_context(attrs) + assert result == {"tier": "premium"} + + def test_prune_beyond_256_fields(self): + attrs = {str(i): str(i) for i in range(300)} + result = flatten_and_prune_context(attrs) + assert len(result) <= MAX_CONTEXT_FIELDS + + def test_prune_skips_oversized_string_values(self): + long_val = "x" * (MAX_FIELD_LENGTH + 1) + attrs = {"short": "ok", "long": long_val} + result = flatten_and_prune_context(attrs) + assert "short" in result + assert "long" not in result + + def test_prune_deterministic_sorted_order(self): + """Pruned result must be reproducible across repeated calls.""" + attrs = {str(i): str(i) for i in range(300)} + r1 = flatten_and_prune_context(attrs) + r2 = flatten_and_prune_context(attrs) + assert r1 == r2 + + def test_context_with_256_fields_not_pruned(self): + attrs = {str(i): str(i) for i in range(256)} + result = flatten_and_prune_context(attrs) + assert len(result) == 256 + + +# --------------------------------------------------------------------------- +# Aggregation tests (full → degraded → drop-counted) +# --------------------------------------------------------------------------- + + +class TestAggregation: + def test_two_identical_evals_aggregate_into_one_bucket_count_2(self, writer): + t0 = int(time.time() * 1000) + t1 = t0 + 100 + e1 = _make_event(eval_time_ms=t0) + e2 = _make_event(eval_time_ms=t1) + writer._aggregate(e1) + writer._aggregate(e2) + + assert len(writer._full) == 1 + entry = list(writer._full.values())[0] + assert entry.count == 2 + assert entry.first_evaluation <= entry.last_evaluation + assert entry.first_evaluation == t0 + assert entry.last_evaluation == t1 + + def test_two_evals_differing_context_value_type_produce_two_buckets(self, writer): + """int 1 vs str '1' in context produce two distinct full-tier buckets.""" + e_int = _make_event(attrs={"x": 1}) + e_str = _make_event(attrs={"x": "1"}) + writer._aggregate(e_int) + writer._aggregate(e_str) + assert len(writer._full) == 2 + + def test_full_tier_overflow_routes_to_degraded(self, writer): + """Overflow past globalCap routes to the degraded tier.""" + writer._global_count = GLOBAL_CAP # simulate full global cap + + # Inject per-flag count below PER_FLAG_CAP so only the global cap triggers. + writer._per_flag_count["flag-x"] = 0 + + e = _make_event(flag_key="flag-x", attrs={"unique": "ctx"}) + writer._aggregate(e) + + assert len(writer._full) == 0 + assert len(writer._degraded) == 1 + + def test_degraded_overflow_increments_dropped_counter(self, writer): + """Beyond degradedCap, increment _dropped_degraded_overflow.""" + # Fill the degraded map to the cap. + for i in range(DEGRADED_CAP): + key = (f"flag-{i}", "on", "alloc", False, "") + from ddtrace.internal.openfeature._flagevaluation_writer import _Entry + + writer._degraded[key] = _Entry(1000, False, "", {}, "") + + with writer._lock: + writer._add_to_degraded(_make_event(flag_key="overflow-flag")) + + assert writer._dropped_degraded_overflow == 1 + + def test_per_flag_cap_routes_to_degraded(self, writer): + """Per-flag cap exceeded → route to degraded even when globalCap has room.""" + writer._per_flag_count["my-flag"] = PER_FLAG_CAP # flag is at cap + e = _make_event(flag_key="my-flag", attrs={"ctx": "x"}) + writer._aggregate(e) + + assert len(writer._degraded) == 1 + assert len(writer._full) == 0 + + def test_runtime_default_when_variant_is_absent(self, writer): + """Absent/empty variant → runtime_default_used True.""" + e = _make_event(variant="", runtime_default=True) + writer._aggregate(e) + + assert len(writer._full) == 1 + entry = list(writer._full.values())[0] + assert entry.runtime_default is True + + def test_degraded_event_omits_targeting_key_and_context(self, writer): + """Degraded tier strips targeting_key + context.""" + with writer._lock: + writer._add_to_degraded(_make_event(targeting_key="some-key", attrs={"k": "v"})) + + entry = list(writer._degraded.values())[0] + assert entry.targeting_key == "" + assert entry.context_attrs == {} + + +# --------------------------------------------------------------------------- +# Enqueue non-blocking tests +# --------------------------------------------------------------------------- + + +class TestEnqueue: + def test_enqueue_non_blocking_on_full_queue(self, writer): + """When queue is full, enqueue must NOT block and must increment _dropped_queue.""" + # Fill the queue. + for i in range(QUEUE_SIZE): + writer._queue.put_nowait(_make_event(flag_key=f"f{i}")) + + # This should not raise and should not block. + writer.enqueue(_make_event(flag_key="overflow")) + assert writer._dropped_queue == 1 + + def test_enqueue_succeeds_when_queue_has_capacity(self, writer): + writer.enqueue(_make_event()) + assert writer._queue.qsize() == 1 + + def test_enqueue_queues_pruned_context_snapshot(self, writer): + attrs = {f"field-{i:03d}": f"value-{i:03d}" for i in range(MAX_CONTEXT_FIELDS + 50)} + attrs["zzz-oversized"] = "x" * (MAX_FIELD_LENGTH + 1) + + writer.enqueue(_make_event(attrs=attrs)) + + queued = writer._queue.get_nowait() + assert len(queued.attrs) == MAX_CONTEXT_FIELDS + assert "zzz-oversized" not in queued.attrs + + def test_enqueue_flattens_nested_context_snapshot(self, writer): + writer.enqueue(_make_event(attrs={"user": {"id": 123, "plan": "pro"}})) + + queued = writer._queue.get_nowait() + assert queued.attrs == {"user.id": 123, "user.plan": "pro"} + + def test_enqueue_after_shutdown_started_is_dropped_and_counted(self, writer): + writer.on_shutdown() + + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + writer.enqueue(_make_event(flag_key="late")) + + assert writer._queue.qsize() == 0 + _assert_count_metric(mock_count, FLAG_EVALUATION_DROPPED_METRIC, 1, FLAG_EVALUATION_REASON_CLOSED) + + +# --------------------------------------------------------------------------- +# Periodic flush + EVP POST tests +# --------------------------------------------------------------------------- + + +class TestPeriodicFlush: + def test_periodic_drains_queue_and_builds_payload(self, writer): + writer.enqueue(_make_event()) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + mock_send.assert_called_once() + payload_bytes, num_events = mock_send.call_args[0] + decoded = json.loads(payload_bytes) + assert "flagEvaluations" in decoded + assert len(decoded["flagEvaluations"]) == 1 + ev = decoded["flagEvaluations"][0] + assert ev["flag"]["key"] == "my-flag" + assert "first_evaluation" in ev + assert "last_evaluation" in ev + assert "evaluation_count" in ev + assert ev["evaluation_count"] == 1 + + def test_periodic_no_send_when_empty(self, writer): + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + mock_send.assert_not_called() + + def test_periodic_resets_maps_after_flush(self, writer): + writer.enqueue(_make_event()) + with mock.patch.object(writer, "_send_payload"): + writer.periodic() + assert writer._full == {} + assert writer._degraded == {} + assert writer._global_count == 0 + + @mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.get_connection") + def test_post_to_correct_endpoint_with_evp_header(self, mock_get_conn, writer): + """Payload goes to /evp_proxy/v2/api/v2/flagevaluation with EVP subdomain header.""" + mock_conn = mock.Mock() + mock_resp = mock.Mock() + mock_resp.status = 200 + mock_resp.read.return_value = b"OK" + mock_conn.getresponse.return_value = mock_resp + mock_get_conn.return_value = mock_conn + + writer.enqueue(_make_event()) + writer.periodic() + + mock_get_conn.assert_called_once() + mock_conn.request.assert_called_once() + call_args = mock_conn.request.call_args + method, endpoint, _payload, headers = call_args[0] + assert method == "POST" + assert endpoint == FLAGEVALUATIONS_ENDPOINT + assert headers[EVP_SUBDOMAIN_HEADER_NAME] == EVP_SUBDOMAIN_VALUE + assert "Content-Type" in headers + + def test_two_evals_same_dims_aggregate_count_2(self, writer): + t0 = int(time.time() * 1000) + writer.enqueue(_make_event(eval_time_ms=t0)) + writer.enqueue(_make_event(eval_time_ms=t0 + 50)) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + evals = decoded["flagEvaluations"] + assert len(evals) == 1 + assert evals[0]["evaluation_count"] == 2 + assert "reason" not in evals[0] + assert evals[0]["first_evaluation"] <= evals[0]["last_evaluation"] + + def test_context_pruning_above_256_fields(self, writer): + """Context with >256 fields is pruned before keying.""" + attrs = {str(i): str(i) for i in range(300)} + e = _make_event(attrs=attrs) + writer.enqueue(e) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + ev = decoded["flagEvaluations"][0] + # The context.evaluation map should exist but have ≤256 fields. + assert "context" in ev + assert len(ev["context"]["evaluation"]) <= MAX_CONTEXT_FIELDS + + def test_context_value_exceeding_256_chars_pruned(self, writer): + """Context values >256 chars are skipped.""" + long_val = "x" * (MAX_FIELD_LENGTH + 10) + attrs = {"short": "ok", "long_field": long_val} + e = _make_event(attrs=attrs) + writer.enqueue(e) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + ev = decoded["flagEvaluations"][0] + ctx_eval = ev.get("context", {}).get("evaluation", {}) + assert "short" in ctx_eval + assert "long_field" not in ctx_eval + + def test_openfeature_datetime_context_value_is_json_serialized(self, writer): + """OpenFeature allows datetime context values; payload JSON should stringify them.""" + timestamp = datetime(2026, 6, 23, 12, 30, tzinfo=timezone.utc) + writer.enqueue(_make_event(attrs={"seen_at": timestamp, "nested": {"created_at": timestamp}})) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + ctx_eval = decoded["flagEvaluations"][0]["context"]["evaluation"] + assert ctx_eval["seen_at"] == timestamp.isoformat() + assert ctx_eval["nested.created_at"] == timestamp.isoformat() + + def test_openfeature_non_json_context_values_are_json_serialized(self, writer): + """Non-JSON-safe context values must not make payload encoding drop the batch.""" + + class CustomContextValue: + def __str__(self): + return "custom-value" + + writer.enqueue( + _make_event( + attrs={ + "amount": Decimal("12.34"), + "custom": CustomContextValue(), + "list": [Decimal("1.5"), CustomContextValue()], + } + ) + ) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + ctx_eval = decoded["flagEvaluations"][0]["context"]["evaluation"] + assert ctx_eval["amount"] == "12.34" + assert ctx_eval["custom"] == "custom-value" + assert ctx_eval["list[0]"] == "1.5" + assert ctx_eval["list[1]"] == "custom-value" + + def test_degraded_event_has_no_context_or_targeting_key(self, writer): + """Degraded-tier events must not include targeting_key or context fields.""" + # Force to degraded by saturating per-flag cap. + writer._per_flag_count["my-flag"] = PER_FLAG_CAP + + e = _make_event(targeting_key="tgt-user", attrs={"k": "v"}) + writer.enqueue(e) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + payload_bytes = mock_send.call_args[0][0] + decoded = json.loads(payload_bytes) + ev = decoded["flagEvaluations"][0] + assert "targeting_key" not in ev + assert "context" not in ev + + def test_targeting_key_is_not_duplicated_in_context_evaluation(self, writer): + writer.enqueue( + _make_event( + targeting_key="user-1", + attrs={"targetingKey": "user-1", "targeting_key": "user-1", "tier": "premium"}, + ) + ) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + decoded = json.loads(mock_send.call_args[0][0]) + ev = decoded["flagEvaluations"][0] + assert ev["targeting_key"] == "user-1" + assert ev["context"]["evaluation"] == {"tier": "premium"} + + def test_writer_endpoint_constant(self): + assert FLAGEVALUATIONS_ENDPOINT == "/evp_proxy/v2/api/v2/flagevaluation" + + def test_cap_sizing_constants(self): + assert EVAL_SCALE_FULL_BUCKET_TARGET == 125_000 + assert EVAL_SCALE_PER_FLAG_BUCKET_TARGET == 10_000 + assert EVAL_SCALE_DEGRADED_BUCKET_TARGET == 25_000 + assert GLOBAL_CAP == 131_072 + assert PER_FLAG_CAP == 10_000 + assert DEGRADED_CAP == 32_768 + + def test_class_exists_and_inherits_periodic_service(self): + from ddtrace.internal.periodic import PeriodicService + + assert issubclass(FlagEvaluationWriter, PeriodicService) + + def test_payloads_are_split_under_evp_payload_size_limit(self, writer): + for i in range(5): + writer.enqueue( + _make_event( + flag_key=f"split-{i}", + targeting_key=f"user-{i}", + attrs={"blob": "x" * 200}, + ) + ) + + sent = [] + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT", 900): + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload", side_effect=lambda p, n: sent.append((p, n))): + writer.periodic() + + assert len(sent) > 1 + _assert_count_metric(mock_count, FLAG_EVALUATION_SPLITS_METRIC, len(sent) - 1) + _assert_no_count_metric(mock_count, FLAG_EVALUATION_DROPPED_METRIC, FLAG_EVALUATION_REASON_PAYLOAD_LIMIT) + _assert_no_count_metric(mock_count, FLAG_EVALUATION_DEGRADED_METRIC, FLAG_EVALUATION_REASON_PAYLOAD_LIMIT) + seen_flags = set() + for payload, num_events in sent: + assert len(payload) <= 900 + decoded = json.loads(payload) + _assert_batch_contract_valid(decoded) + assert num_events == len(decoded["flagEvaluations"]) + seen_flags.update(row["flag"]["key"] for row in decoded["flagEvaluations"]) + assert seen_flags == {f"split-{i}" for i in range(5)} + + def test_single_oversized_full_event_is_degraded_before_send(self, writer): + writer.enqueue( + _make_event( + flag_key="oversized-full", + targeting_key="user-with-context", + attrs={"blob": "x" * 200}, + ) + ) + + sent = [] + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT", 300): + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload", side_effect=lambda p, n: sent.append((p, n))): + writer.periodic() + + assert len(sent) == 1 + _assert_count_metric(mock_count, FLAG_EVALUATION_DEGRADED_METRIC, 1, FLAG_EVALUATION_REASON_PAYLOAD_LIMIT) + payload, num_events = sent[0] + assert len(payload) <= 300 + assert num_events == 1 + row = json.loads(payload)["flagEvaluations"][0] + assert row["flag"]["key"] == "oversized-full" + assert "context" not in row + assert "targeting_key" not in row + + def test_single_oversized_degraded_event_is_dropped_and_counted(self, writer): + writer.enqueue(_make_event(flag_key="f" * 256, targeting_key="", attrs={})) + + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.FLAGEVALUATIONS_PAYLOAD_SIZE_LIMIT", 100): + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + mock_send.assert_not_called() + _assert_count_metric(mock_count, FLAG_EVALUATION_DROPPED_METRIC, 1, FLAG_EVALUATION_REASON_PAYLOAD_LIMIT) + + def test_build_payload_stats_count_payload_limit_degraded_and_dropped_rows(self): + now_ms = int(time.time() * 1000) + degradable = { + "timestamp": now_ms, + "flag": {"key": "large"}, + "first_evaluation": now_ms, + "last_evaluation": now_ms, + "evaluation_count": 7, + "targeting_key": "user-with-context", + "context": {"evaluation": {"blob": "x" * 256}}, + } + degraded = dict(degradable) + degraded.pop("targeting_key") + degraded.pop("context") + degraded_payload = _build_payloads_with_stats([degraded], {}, 1 << 30).payloads[0][0] + + result = _build_payloads_with_stats([degradable], {}, len(degraded_payload)) + + assert result.degraded_payload_limit == 7 + assert result.dropped_payload_limit == 0 + assert len(result.payloads) == 1 + + undegreadable = { + "timestamp": now_ms, + "flag": {"key": "f" * 256}, + "first_evaluation": now_ms, + "last_evaluation": now_ms, + "evaluation_count": 11, + } + oversized_payload = _build_payloads_with_stats([undegreadable], {}, 1 << 30).payloads[0][0] + + result = _build_payloads_with_stats([undegreadable], {}, len(oversized_payload) - 1) + + assert result.degraded_payload_limit == 0 + assert result.dropped_payload_limit == 11 + assert result.payloads == [] + + +# --------------------------------------------------------------------------- +# Stable payload contract for emitted rows (full + degraded) +# --------------------------------------------------------------------------- + +# Required fields that EVERY flagevaluation row (full or degraded) must carry. +_REQUIRED_EVENT_FIELDS = { + "timestamp": int, + "flag": dict, + "first_evaluation": int, + "last_evaluation": int, + "evaluation_count": int, +} +_OPTIONAL_EVENT_FIELDS = { + "runtime_default_used", + "targeting_key", + "context", + "variant", + "allocation", + "targeting_rule", + "error", +} +_ALLOWED_EVENT_FIELDS = set(_REQUIRED_EVENT_FIELDS).union(_OPTIONAL_EVENT_FIELDS) +_ALLOWED_BATCH_FIELDS = {"flagEvaluations", "context"} +_ALLOWED_BATCH_CONTEXT_FIELDS = {"service", "env", "version"} +_ALLOWED_ROW_CONTEXT_FIELDS = {"evaluation", "dd"} + + +def _assert_row_contract_valid(ev: dict) -> None: + """Assert one flagevaluation row uses only the SDK-owned stable EVP fields.""" + extra_fields = set(ev) - _ALLOWED_EVENT_FIELDS + assert not extra_fields, f"unknown flagevaluation row fields: {sorted(extra_fields)}" + + # Required fields present with the right scalar types. + for field, typ in _REQUIRED_EVENT_FIELDS.items(): + assert field in ev, f"required field {field!r} missing from row: {ev}" + assert isinstance(ev[field], typ), f"{field} must be {typ}, got {type(ev[field])}" + + # flag.key is the one required nested field. + assert "key" in ev["flag"] and isinstance(ev["flag"]["key"], str) + + # first <= last evaluation bound. + assert ev["first_evaluation"] <= ev["last_evaluation"] + assert ev["evaluation_count"] >= 1 + + # variant/allocation, when present, MUST be {"key": "..."} objects, NOT bare strings. + for obj_field in ("variant", "allocation"): + if obj_field in ev: + assert isinstance(ev[obj_field], dict), f"{obj_field} must serialize as an object" + assert set(ev[obj_field].keys()) == {"key"}, f"{obj_field} must be exactly {{key}}" + assert isinstance(ev[obj_field]["key"], str) + + # error, when present, is {"message": "..."}. + if "error" in ev: + assert isinstance(ev["error"], dict) + assert "message" in ev["error"] + + # context, when present, nests an "evaluation" map. + if "context" in ev: + assert isinstance(ev["context"], dict) + extra_context_fields = set(ev["context"]) - _ALLOWED_ROW_CONTEXT_FIELDS + assert not extra_context_fields, f"unknown row context fields: {sorted(extra_context_fields)}" + assert "evaluation" in ev["context"] + assert isinstance(ev["context"]["evaluation"], dict) + + # runtime_default_used, when present, is a bool. + if "runtime_default_used" in ev: + assert isinstance(ev["runtime_default_used"], bool) + + +def _assert_batch_contract_valid(payload: dict) -> None: + """Assert the batch envelope uses only the stable fields this SDK emits.""" + extra_fields = set(payload) - _ALLOWED_BATCH_FIELDS + assert not extra_fields, f"unknown flagevaluation batch fields: {sorted(extra_fields)}" + assert "flagEvaluations" in payload + assert isinstance(payload["flagEvaluations"], list) + if "context" in payload: + assert isinstance(payload["context"], dict) + extra_context_fields = set(payload["context"]) - _ALLOWED_BATCH_CONTEXT_FIELDS + assert not extra_context_fields, f"unknown batch context fields: {sorted(extra_context_fields)}" + for value in payload["context"].values(): + assert isinstance(value, str) + for row in payload["flagEvaluations"]: + _assert_row_contract_valid(row) + + +class TestPayloadContractConformance: + def test_full_tier_row_uses_stable_contract_with_object_variant_and_allocation(self, writer): + """A full-tier row carries variant/allocation as {key} objects + context.evaluation.""" + writer.enqueue( + _make_event( + variant="on", + allocation_key="alloc-1", + attrs={"tier": "premium"}, + ) + ) + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + decoded = json.loads(mock_send.call_args[0][0]) + _assert_batch_contract_valid(decoded) + assert "flagEvaluations" in decoded + row = decoded["flagEvaluations"][0] + _assert_row_contract_valid(row) + # Specifically the {key} object shape (NOT bare strings). + assert row["variant"] == {"key": "on"} + assert row["allocation"] == {"key": "alloc-1"} + assert row["context"]["evaluation"]["tier"] == "premium" + + def test_degraded_tier_row_uses_stable_contract_and_omits_context(self, writer): + """A degraded-tier row uses variant/allocation objects, no context.""" + writer._per_flag_count["my-flag"] = PER_FLAG_CAP # force degraded + writer.enqueue( + _make_event( + variant="on", + allocation_key="alloc-1", + attrs={"k": "v"}, + error_message="degraded failure", + ) + ) + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + decoded = json.loads(mock_send.call_args[0][0]) + _assert_batch_contract_valid(decoded) + row = decoded["flagEvaluations"][0] + _assert_row_contract_valid(row) + _assert_count_metric(mock_count, FLAG_EVALUATION_DEGRADED_METRIC, 1, FLAG_EVALUATION_REASON_CARDINALITY_CAP) + assert row["variant"] == {"key": "on"} + assert row["error"] == {"message": "degraded failure"} + assert "context" not in row + assert "targeting_key" not in row + + def test_error_row_carries_error_message_object(self, writer): + """An error evaluation produces a stable row with error.message.""" + writer.enqueue( + _make_event( + variant="", + runtime_default=True, + error_message="Flag not found", + ) + ) + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + decoded = json.loads(mock_send.call_args[0][0]) + _assert_batch_contract_valid(decoded) + row = decoded["flagEvaluations"][0] + _assert_row_contract_valid(row) + assert row["error"] == {"message": "Flag not found"} + # Absent variant -> runtime_default_used True, no variant object emitted. + assert row["runtime_default_used"] is True + assert "variant" not in row + + def test_batch_payload_validates_full_and_degraded_rows_together(self, writer): + """A single flush emits BOTH a full row and a degraded row under the stable contract.""" + # Full-tier event. + writer.enqueue(_make_event(flag_key="full-flag", variant="on", attrs={"a": "b"})) + # Degraded-tier event (different flag forced to degraded). + writer._per_flag_count["deg-flag"] = PER_FLAG_CAP + writer.enqueue(_make_event(flag_key="deg-flag", variant="off")) + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.periodic() + + decoded = json.loads(mock_send.call_args[0][0]) + _assert_batch_contract_valid(decoded) + rows = decoded["flagEvaluations"] + assert len(rows) == 2 + for row in rows: + _assert_row_contract_valid(row) + flags = {r["flag"]["key"] for r in rows} + assert flags == {"full-flag", "deg-flag"} + + def test_contract_rejects_top_level_reason(self): + bad = { + "flagEvaluations": [ + { + "timestamp": int(time.time() * 1000), + "flag": {"key": "reason-flag"}, + "first_evaluation": int(time.time() * 1000), + "last_evaluation": int(time.time() * 1000), + "evaluation_count": 1, + "reason": "targeting_match", + } + ] + } + with pytest.raises(AssertionError, match="reason"): + _assert_batch_contract_valid(bad) + + +# --------------------------------------------------------------------------- +# Shutdown drains the queue + final-flush before exit +# --------------------------------------------------------------------------- + + +class TestShutdownDrain: + def test_on_shutdown_drains_queue_and_flushes(self, writer): + """on_shutdown (the PeriodicService shutdown callback) must drain + flush queued events.""" + writer.enqueue(_make_event(flag_key="pending-1")) + writer.enqueue(_make_event(flag_key="pending-2")) + assert writer._queue.qsize() == 2 + + with mock.patch.object(writer, "_send_payload") as mock_send: + writer.on_shutdown() + + # The queued events were drained, aggregated, and flushed in a final POST. + mock_send.assert_called_once() + decoded = json.loads(mock_send.call_args[0][0]) + flags = {r["flag"]["key"] for r in decoded["flagEvaluations"]} + assert flags == {"pending-1", "pending-2"} + assert writer._queue.qsize() == 0 + + def test_real_start_stop_lifecycle_drains_pending_event(self): + """Real PeriodicService start()->enqueue->stop() drains the queue via on_shutdown.""" + # Long interval so the periodic timer never fires; only stop() triggers the flush. + w = FlagEvaluationWriter(interval=3600.0) + sent = [] + with mock.patch.object(w, "_send_payload", side_effect=lambda p, n: sent.append((p, n))): + w.start() + w.enqueue(_make_event(flag_key="lifecycle-flag")) + w.stop() # stop() runs on_shutdown -> periodic() -> drain + flush + # stop() requests shutdown; join() blocks until the worker (and its + # on_shutdown final flush) has fully completed before we assert. + w.join(timeout=5.0) + assert len(sent) == 1, "stop() must trigger a final drain+flush" + decoded = json.loads(sent[0][0]) + assert decoded["flagEvaluations"][0]["flag"]["key"] == "lifecycle-flag" + + def test_background_drain_accumulates_beyond_queue_size_before_flush(self): + """A flush window can exceed the bounded queue size and naturally degrade.""" + sent = [] + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.QUEUE_SIZE", 8): + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.PER_FLAG_CAP", 12): + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.DRAIN_INTERVAL", 0.01): + w = FlagEvaluationWriter(interval=3600.0) + with mock.patch.object(w, "_send_payload", side_effect=lambda p, n: sent.append((p, n))): + w.start() + try: + for i in range(14): + assert _wait_until(lambda: not w._queue.full()) + w.enqueue( + _make_event( + flag_key="natural-degrade", + targeting_key=f"user-{i}", + attrs={"user": i}, + ) + ) + assert _wait_until(lambda: w._queue.empty()) + with w._lock: + assert w._dropped_queue == 0 + assert w._degraded + finally: + w.stop() + w.join(timeout=5.0) + + assert sent, "shutdown flush must emit the accumulated evaluation rows" + decoded = json.loads(sent[0][0]) + degraded_rows = [ + row + for row in decoded["flagEvaluations"] + if row["flag"]["key"] == "natural-degrade" and "context" not in row and "targeting_key" not in row + ] + assert len(degraded_rows) == 1 + assert degraded_rows[0]["evaluation_count"] == 2 + + +# --------------------------------------------------------------------------- +# Backpressure drop counters are observable (emitted on flush) +# --------------------------------------------------------------------------- + + +class TestObservableDropCounters: + def test_queue_overflow_drop_count_is_logged_on_flush(self, writer): + """Queue-full drops increment _dropped_queue AND are surfaced (logged) on flush.""" + # Fill the queue so the next enqueue drops. + for i in range(QUEUE_SIZE): + writer._queue.put_nowait(_make_event(flag_key=f"f{i}")) + writer.enqueue(_make_event(flag_key="dropped")) + assert writer._dropped_queue == 1 + + # Drain everything (so maps are populated) and assert the drop count is emitted. + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.logger") as mock_logger: + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload"): + writer.periodic() + # A warning naming the queue-full drop count must have been emitted. + warnings = [c for c in mock_logger.warning.call_args_list if "queue full" in str(c).lower()] + assert warnings, "queue-full drop count must be logged (observable)" + _assert_count_metric(mock_count, FLAG_EVALUATION_DROPPED_METRIC, 1, FLAG_EVALUATION_REASON_QUEUE_OVERFLOW) + # Counter resets after emission. + assert writer._dropped_queue == 0 + + def test_degraded_overflow_drop_count_is_logged_on_flush(self, writer): + """Degraded-cap overflow drops increment _dropped_degraded_overflow AND are logged.""" + from ddtrace.internal.openfeature._flagevaluation_writer import _Entry + + # Saturate the degraded map to its cap. + for i in range(DEGRADED_CAP): + writer._degraded[(f"flag-{i}", "on", "alloc", False, "")] = _Entry(1000, False, "", {}, "") + with writer._lock: + writer._add_to_degraded(_make_event(flag_key="overflow")) + assert writer._dropped_degraded_overflow == 1 + + with mock.patch("ddtrace.internal.openfeature._flagevaluation_writer.logger") as mock_logger: + with mock.patch(TELEMETRY_COUNT_PATCH) as mock_count: + with mock.patch.object(writer, "_send_payload"): + writer.periodic() + warnings = [c for c in mock_logger.warning.call_args_list if "degraded cap" in str(c).lower()] + assert warnings, "degraded-cap overflow count must be logged (observable)" + _assert_count_metric(mock_count, FLAG_EVALUATION_DROPPED_METRIC, 1, FLAG_EVALUATION_REASON_DEGRADED_CAP) + assert writer._dropped_degraded_overflow == 0 + + def test_drop_accounting_is_complete_no_silent_loss(self, writer): + """Σ(tier counts + drops) == events processed (no silent loss).""" + # 3 distinct full-tier buckets + 2 degraded-overflow drops. + writer._aggregate(_make_event(flag_key="a", attrs={"x": 1})) + writer._aggregate(_make_event(flag_key="b", attrs={"x": 2})) + writer._aggregate(_make_event(flag_key="a", attrs={"x": 1})) # repeat -> count 2 on bucket a + + full_counts = sum(e.count for e in writer._full.values()) + assert full_counts == 3 # 2 (a) + 1 (b) + assert writer._dropped_degraded_overflow == 0 + assert writer._dropped_queue == 0 diff --git a/tests/openfeature/test_provider.py b/tests/openfeature/test_provider.py index b9b953ae6ec..0e4e71290ff 100644 --- a/tests/openfeature/test_provider.py +++ b/tests/openfeature/test_provider.py @@ -61,6 +61,11 @@ def test_provider_initialization(self, provider, evaluation_context): # Should not raise provider.initialize(evaluation_context) + def test_provider_accepts_legacy_initialization_timeout_kwarg(self): + """Existing callers can still pass initialization_timeout after async initialization.""" + with override_global_config({"experimental_flagging_provider_enabled": True}): + DataDogProvider(initialization_timeout=0.1) + def test_provider_shutdown(self, provider): """Provider should shutdown without errors.""" # Should not raise diff --git a/tests/openfeature/test_provider_status.py b/tests/openfeature/test_provider_status.py index 8af9375fd3c..d61f6c9f494 100644 --- a/tests/openfeature/test_provider_status.py +++ b/tests/openfeature/test_provider_status.py @@ -5,13 +5,14 @@ - NOT_READY by default - READY when first Remote Config payload is received - Event emission on status change -- Blocking initialization until config arrives or timeout +- Non-blocking initialization while config arrives asynchronously """ import threading import time from openfeature import api +from openfeature.evaluation_context import EvaluationContext from openfeature.provider import ProviderStatus import pytest @@ -38,6 +39,13 @@ def clear_config(): _set_ffe_config(None) +def _set_provider(provider): + if hasattr(api, "set_provider_and_wait"): + api.set_provider_and_wait(provider) + else: + api.set_provider(provider) + + class TestProviderStatus: """Test provider status lifecycle.""" @@ -93,30 +101,36 @@ def test_provider_ready_event_emitted(self): def test_provider_ready_event_only_once(self): """Test that PROVIDER_READY event is only emitted once, not on subsequent configs.""" ready_events = [] + ready_event = threading.Event() def on_provider_ready(event_details): ready_events.append(event_details) + ready_event.set() api.add_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) try: with override_global_config({"experimental_flagging_provider_enabled": True}): provider = DataDogProvider() - api.set_provider(provider) + _set_provider(provider) # Clear events from initialization ready_events.clear() + ready_event.clear() # First configuration config1 = create_config(create_boolean_flag("flag1", enabled=True)) process_ffe_configuration(config1) + assert ready_event.wait(timeout=1.0) count_after_first = len(ready_events) assert count_after_first >= 1 # Should have emitted # Second configuration + ready_event.clear() config2 = create_config(create_boolean_flag("flag2", enabled=True)) process_ffe_configuration(config2) + assert not ready_event.wait(timeout=0.05) count_after_second = len(ready_events) # Should not have emitted again @@ -125,6 +139,31 @@ def on_provider_ready(event_details): api.remove_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) api.clear_providers() + @pytest.mark.skipif(ProviderEvent is None, reason="ProviderEvent not available in SDK 0.6.0") + def test_sdk_ready_event_can_fire_before_datadog_config_ready(self): + """SDK-level PROVIDER_READY does not mean Datadog config has loaded.""" + ready_events = [] + ready_event = threading.Event() + + def on_provider_ready(event_details): + ready_events.append(event_details) + ready_event.set() + + api.add_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) + + try: + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider() + _set_provider(provider) + + assert ready_event.wait(timeout=1.0) + assert len(ready_events) >= 1 + assert provider._status == ProviderStatus.NOT_READY + assert not provider._config_received.is_set() + finally: + api.remove_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) + api.clear_providers() + def test_provider_status_after_shutdown(self): """Test that provider returns to NOT_READY after shutdown.""" with override_global_config({"experimental_flagging_provider_enabled": True}): @@ -190,7 +229,7 @@ def on_provider_ready(event_details): api.add_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) try: - api.set_provider(provider) + _set_provider(provider) # Provider should detect existing config and emit READY assert provider._status == ProviderStatus.READY @@ -199,34 +238,43 @@ def on_provider_ready(event_details): api.remove_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) api.clear_providers() + @pytest.mark.skipif(ProviderEvent is None, reason="ProviderEvent not available in SDK 0.6.0") + def test_attached_provider_receives_config_before_async_initialize(self): + """OpenFeature SDK 0.10 attaches synchronously, then initializes asynchronously.""" + ready_events = [] -class TestProviderInitializationBlocking: - """Test that initialize() blocks until config arrives or timeout expires.""" + def on_emit(provider, event, details): + ready_events.append(event) - def test_initialize_blocks_until_config_arrives(self): - """initialize() should block and return once config is delivered mid-wait.""" with override_global_config({"experimental_flagging_provider_enabled": True}): - provider = DataDogProvider(initialization_timeout=5.0) + provider = DataDogProvider() + provider.attach(on_emit) + + config = create_config(create_boolean_flag("test-flag", enabled=True)) + process_ffe_configuration(config) + + assert provider._status == ProviderStatus.READY + assert ProviderEvent.PROVIDER_READY in ready_events - # Deliver config from a background thread after 0.5s - def deliver_config(): - time.sleep(0.5) - config = create_config(create_boolean_flag("test-flag", enabled=True)) - process_ffe_configuration(config) - timer = threading.Thread(target=deliver_config, daemon=True) - timer.start() +class TestProviderInitializationAsync: + """Test that initialize() returns immediately and READY arrives asynchronously.""" + + def test_initialize_returns_immediately_without_config(self): + """initialize() should return immediately even if no config is available yet.""" + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider() try: start = time.monotonic() - api.set_provider(provider) + # initialize() is called inside set_provider; it must not block + provider.initialize(EvaluationContext()) elapsed = time.monotonic() - start - # Should have blocked for ~0.5s (not instant, not full timeout) - assert elapsed >= 0.3, f"initialize() returned too fast ({elapsed:.2f}s)" - assert elapsed < 4.0, f"initialize() took too long ({elapsed:.2f}s), should have unblocked at ~0.5s" - assert provider._status == ProviderStatus.READY - assert provider._config_received.is_set() + # Should return near-instantly (no blocking wait) + assert elapsed < 0.5, f"initialize() blocked for {elapsed:.2f}s — must not block" + # Provider is NOT_READY; READY arrives via on_configuration_received() + assert provider._status == ProviderStatus.NOT_READY finally: api.clear_providers() @@ -237,11 +285,11 @@ def test_initialize_fast_path_when_config_exists(self): config = create_config(create_boolean_flag("test-flag", enabled=True)) process_ffe_configuration(config) - provider = DataDogProvider(initialization_timeout=5.0) + provider = DataDogProvider() try: start = time.monotonic() - api.set_provider(provider) + _set_provider(provider) elapsed = time.monotonic() - start # Should be near-instant (config already available) @@ -250,41 +298,38 @@ def test_initialize_fast_path_when_config_exists(self): finally: api.clear_providers() - def test_initialize_timeout_raises(self): - """initialize() should raise ProviderNotReadyError after timeout expires.""" + def test_ready_after_config_arrives_async(self): + """Provider transitions to READY when config arrives after initialize() returns.""" with override_global_config({"experimental_flagging_provider_enabled": True}): - provider = DataDogProvider(initialization_timeout=0.5) + provider = DataDogProvider() try: - start = time.monotonic() - # set_provider catches the exception and dispatches PROVIDER_ERROR - api.set_provider(provider) - elapsed = time.monotonic() - start + provider.initialize(EvaluationContext()) + # Still NOT_READY immediately after initialize() + assert provider._status == ProviderStatus.NOT_READY - # Should have blocked for ~0.5s (the timeout) - assert elapsed >= 0.3, f"initialize() returned too fast ({elapsed:.2f}s)" - assert elapsed < 2.0, f"initialize() took too long ({elapsed:.2f}s)" + # Config arrives later (simulating RC delivery) + config = create_config(create_boolean_flag("test-flag", enabled=True)) + process_ffe_configuration(config) - # Provider should be in ERROR state (SDK caught ProviderNotReadyError) - client = api.get_client() - assert client.get_provider_status() == ProviderStatus.ERROR + # Provider should now be READY + assert provider._status == ProviderStatus.READY + assert provider._config_received.is_set() finally: api.clear_providers() - def test_late_recovery_after_timeout(self): - """Config arriving after timeout should transition provider to READY.""" + def test_late_config_delivery_transitions_to_ready(self): + """Config arriving after set_provider should transition provider to READY.""" with override_global_config({"experimental_flagging_provider_enabled": True}): - provider = DataDogProvider(initialization_timeout=0.5) + provider = DataDogProvider() try: - # Let it timeout - api.set_provider(provider) + provider.initialize(EvaluationContext()) - # Provider should be in ERROR state - client = api.get_client() - assert client.get_provider_status() == ProviderStatus.ERROR + # Provider is NOT_READY at this point + assert provider._status == ProviderStatus.NOT_READY - # Now deliver config (late recovery) + # Simulate delayed RC delivery config = create_config(create_boolean_flag("test-flag", enabled=True)) process_ffe_configuration(config)