diff --git a/ddtrace/internal/openfeature/_agentless.py b/ddtrace/internal/openfeature/_agentless.py new file mode 100644 index 00000000000..08afc199fd5 --- /dev/null +++ b/ddtrace/internal/openfeature/_agentless.py @@ -0,0 +1,124 @@ +""" +Helpers for the agentless Feature Flagging configuration source. + +These are pure, dependency-light functions (endpoint derivation, JSON:API +validation, gzip decoding) used by the agentless poller. They deliberately take +their inputs explicitly rather than reading global config so they can be unit +tested in isolation. +""" + +import gzip +import json +from typing import Any +from typing import Optional +from urllib.parse import urlencode +from urllib.parse import urlsplit +from urllib.parse import urlunsplit + + +# Canonical rules-based server path appended to the managed CDN host and to +# custom base URLs that only supply an origin. +DEFAULT_AGENTLESS_PATH = "/api/v2/feature-flagging/config/rules-based/server" + + +def build_agentless_endpoint(site: str, env: Optional[str] = None, base_url: Optional[str] = None) -> str: + """Build the agentless UFC endpoint URL. + + Without a custom ``base_url`` the managed Datadog CDN endpoint is derived + from ``site`` (lowercased), adding ``dd_env`` only when ``env`` is set. This + resolves staging (``datad0g.com``) and GovCloud (``ddog-gov.com``) + automatically with no site allowlist. + + A custom ``base_url`` that is a root/origin receives the standard rules-based + path; one with a non-root path is used verbatim as the exact endpoint. HTTP + is permitted for custom endpoints (operator-owned trust). + + :raises ValueError: if a custom ``base_url`` is malformed or uses a scheme + other than http/https. Error messages never include the URL, which is + sensitive. + """ + configured = base_url.strip() if base_url else "" + + if not configured: + netloc = "ufc-server.ff-cdn.{}".format(site.strip().lower()) + query = urlencode({"dd_env": env}) if env else "" + return urlunsplit(("https", netloc, DEFAULT_AGENTLESS_PATH, query, "")) + + # A URL with internal whitespace is malformed; urlsplit is lenient and would + # otherwise accept it. Do not surface the value. + if any(ch.isspace() for ch in configured): + raise ValueError("Invalid Feature Flagging agentless URL") + + try: + parts = urlsplit(configured) + except ValueError: + raise ValueError("Invalid Feature Flagging agentless URL") + + if parts.scheme not in ("http", "https"): + raise ValueError("Feature Flagging agentless URL must use HTTP or HTTPS") + if not parts.netloc: + raise ValueError("Invalid Feature Flagging agentless URL") + + path = parts.path + if path in ("", "/"): + path = DEFAULT_AGENTLESS_PATH + + return urlunsplit((parts.scheme, parts.netloc, path, parts.query, parts.fragment)) + + +def decode_response_body(body: bytes, content_encoding: Optional[str]) -> bytes: + """Return ``body`` decompressed when the response was gzip-encoded. + + gzip is NOT auto-decoded by the HTTP layer, so callers pass the raw body and + the ``Content-Encoding`` header value. The check is case-insensitive. + + :raises: propagates gzip errors (e.g. ``OSError``/``EOFError``) on a body + that claims gzip encoding but cannot be decompressed. + """ + if content_encoding and content_encoding.strip().lower() == "gzip": + return gzip.decompress(body) + return body + + +def parse_ufc_configuration(body: Any) -> "dict[str, Any]": + """Validate a JSON:API UFC response envelope and return ``data.attributes``. + + The envelope must be:: + + {"data": {"type": "universal-flag-configuration", + "attributes": {"format": , "createdAt": , + "environment": {"name": }, "flags": {}}}} + + Only ``data.attributes`` is returned (and passed to the evaluator). Raw UFC + (non-JSON:API) is not accepted, even for custom endpoints. + + :raises ValueError: if the payload is not valid JSON or does not match the + JSON:API Universal Flag Configuration v1 contract. + """ + try: + payload = json.loads(body) + except (ValueError, TypeError) as e: + raise ValueError("Malformed UFC payload") from e + + if not isinstance(payload, dict): + raise ValueError("Expected a JSON:API Universal Flag Configuration resource") + + data = payload.get("data") + if not isinstance(data, dict) or data.get("type") != "universal-flag-configuration": + raise ValueError("Expected a JSON:API Universal Flag Configuration resource") + + attributes = data.get("attributes") + if not isinstance(attributes, dict): + raise ValueError("Expected a Universal Flag Configuration v1 object") + + environment = attributes.get("environment") + if ( + not isinstance(attributes.get("format"), str) + or not isinstance(attributes.get("createdAt"), str) + or not isinstance(environment, dict) + or not isinstance(environment.get("name"), str) + or not isinstance(attributes.get("flags"), dict) + ): + raise ValueError("Expected a Universal Flag Configuration v1 object") + + return attributes diff --git a/ddtrace/internal/openfeature/_agentless_source.py b/ddtrace/internal/openfeature/_agentless_source.py new file mode 100644 index 00000000000..b680a6a95d4 --- /dev/null +++ b/ddtrace/internal/openfeature/_agentless_source.py @@ -0,0 +1,362 @@ +""" +Agentless Feature Flagging configuration source. + +Polls the Datadog UFC CDN (or a configured custom endpoint) for Universal Flag +Configuration and feeds each accepted payload into the same apply function the +Agent Remote Config path uses. A failed poll never replaces last-known-good. + +The poller runs on a background thread via :class:`PeriodicService`. That thread +already implements fixed-delay-after-completion scheduling, so polls never +overlap. The per-poll retry/backoff and per-request timeout live inside a single +:meth:`periodic` tick. +""" + +from collections import namedtuple +import os +import random +import socket +import threading +import time +from typing import Any +from typing import Callable +from typing import Optional +from urllib.parse import urlsplit +from urllib.parse import urlunsplit + +from ddtrace.internal.constants import _HTTPLIB_NO_TRACE_REQUEST +from ddtrace.internal.logger import get_logger +from ddtrace.internal.openfeature._agentless import decode_response_body +from ddtrace.internal.openfeature._agentless import parse_ufc_configuration +from ddtrace.internal.periodic import PeriodicService +from ddtrace.internal.utils.http import get_connection +from ddtrace.internal.utils.retry import RetryError +from ddtrace.internal.utils.retry import retry +from ddtrace.internal.utils.version import _pep440_to_semver + + +log = get_logger(__name__) + +# Polling / retry policy (mirrors the dd-trace-js reference implementation). +MAX_POLL_INTERVAL_SECONDS = 60 * 60 +DEFAULT_POLL_INTERVAL_SECONDS = 30.0 +DEFAULT_REQUEST_TIMEOUT_SECONDS = 5.0 + +MAX_ATTEMPTS = 3 +FIRST_RETRY_MIN_S = 2.0 +FIRST_RETRY_MAX_S = 10.0 +SECOND_RETRY_MIN_S = 5.0 +SECOND_RETRY_MAX_S = 30.0 +RETRY_JITTER = 0.2 + +# Upper bound (seconds) on the randomized delay before the FIRST poll in a +# forked child, so pre-fork workers (gunicorn/uWSGI) don't hit the CDN in +# lockstep. The origin process is never delayed. +FIRST_POLL_JITTER_MAX_S = 5.0 + +# Granularity of in-tick waits (retry backoff, fork jitter). Waits are sliced at +# this interval so a requested shutdown is noticed without an event primitive. +SHUTDOWN_POLL_INTERVAL_S = 0.2 +_WAIT_EPSILON_S = 1e-6 + + +# A single poll outcome. ``status`` is None for a network error / timeout. +_PollResponse = namedtuple("_PollResponse", ["status", "etag", "content_encoding", "body", "error"]) + + +def _clamp(value: float, minimum: float, maximum: float) -> float: + return max(minimum, min(maximum, value)) + + +def _is_retryable_status(status: Optional[int]) -> bool: + if status is None: + return True + return status == 408 or status == 429 or (500 <= status <= 599) + + +class AgentlessConfigurationSource(PeriodicService): + """Background poller that loads UFC from the agentless endpoint.""" + + def __init__( + self, + endpoint: str, + apply_configuration: Callable[["dict[str, Any]"], None], + api_key: Optional[str] = None, + poll_interval: float = DEFAULT_POLL_INTERVAL_SECONDS, + request_timeout: float = DEFAULT_REQUEST_TIMEOUT_SECONDS, + ) -> None: + # A non-positive interval would make PeriodicThread schedule the next run + # immediately, turning polling into a tight loop against the CDN, so fall + # back to the documented default. + if poll_interval <= 0: + log.warning( + "Feature Flagging agentless poll interval must be positive; using the default %.0fs", + DEFAULT_POLL_INTERVAL_SECONDS, + ) + poll_interval = DEFAULT_POLL_INTERVAL_SECONDS + elif poll_interval > MAX_POLL_INTERVAL_SECONDS: + log.warning( + "Feature Flagging agentless poll interval %.0fs exceeds the %ds maximum; clamping", + poll_interval, + MAX_POLL_INTERVAL_SECONDS, + ) + poll_interval = MAX_POLL_INTERVAL_SECONDS + + if request_timeout <= 0: + log.warning( + "Feature Flagging agentless request timeout must be positive; using the default %.0fs", + DEFAULT_REQUEST_TIMEOUT_SECONDS, + ) + request_timeout = DEFAULT_REQUEST_TIMEOUT_SECONDS + + super().__init__(interval=poll_interval, no_wait_at_start=True) + + self._apply_configuration = apply_configuration + self._api_key = api_key + self._request_timeout = request_timeout + + # Split the endpoint into an origin (for the connection) and a request + # target (path + query). get_connection drops the query, so the target + # must carry it. + parts = urlsplit(endpoint) + self._conn_url = urlunsplit((parts.scheme, parts.netloc, "/", "", "")) + self._request_target = urlunsplit(("", "", parts.path, parts.query, "")) or "/" + + self._etag: Optional[str] = None + self._failure_warnings: "set[str]" = set() + self._malformed_payload_logged = False + self._application_failure_logged = False + + # Fork staggering: the process that created the source polls immediately; + # a forked child jitters its first poll (see periodic()). Tracked by PID so + # it is race-free with the automatic post-fork thread restart. + self._origin_pid = os.getpid() + self._jittered_pid: Optional[int] = None + + # Set on shutdown so in-tick waits (retry backoff, fork jitter) return + # early instead of holding the worker thread for their full delay. + self._stopping = False + + # The connection of the poll currently in flight, if any. Shutdown uses it + # to unblock the worker thread mid-request; see _cancel_in_flight_request. + # The lock keeps the shutdown thread from touching a connection the worker + # is concurrently replacing or closing. + self._conn_lock = threading.Lock() + self._in_flight_conn: Optional[Any] = None + + # -- scheduling --------------------------------------------------------- + + def _start_service(self, *args: Any, **kwargs: Any) -> None: + self._stopping = False + super()._start_service(*args, **kwargs) + + def _stop_service(self, *args: Any, **kwargs: Any) -> None: + # Request the stop first so any in-tick wait ends at its next slice and + # the worker thread can finish promptly. + self._stopping = True + # Then tear down any request already on the wire. Without this, a caller + # joining the worker waits out the per-request timeout on every blocking + # socket call still to come (connect, then each read). + self._cancel_in_flight_request() + super()._stop_service(*args, **kwargs) + + def _cancel_in_flight_request(self) -> None: + """Unblock a poll waiting on the network so shutdown does not wait for it. + + Half-closing the socket from this thread makes the worker's pending + connect/recv return or raise at once; close() alone would not, since the + worker is already blocked inside a syscall on that file descriptor. The + worker owns cleanup either way -- _request closes the connection in its + finally block, and the resulting error is reported as a failed poll, which + keeps last-known-good. + + One window stays uncancellable: http.client connects lazily inside + request(), so a stop that lands before the socket exists has nothing to + half-close and the worker blocks in connect() for up to the request + timeout. Requests already past connect -- the long pole, since a poll + spends its time reading the UFC body -- are cancelled immediately. + """ + with self._conn_lock: + conn = self._in_flight_conn + sock = getattr(conn, "sock", None) if conn is not None else None + if sock is None: + return + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + # Already closed, never connected, or shut down by the worker first. + log.debug("Feature Flagging agentless request was not cancellable", exc_info=True) + + def _wait(self, delay: float) -> bool: + """Wait up to ``delay`` seconds. Returns True if a shutdown was requested. + + The wait is sliced rather than done with an event because the library runs + on native threads and has no Python-visible event primitive; polling a + plain flag keeps shutdown responsive without one. + """ + remaining = delay + # The epsilon keeps floating-point drift from adding a final zero-length + # slice (0.6 - 0.2 * 3 does not land exactly on zero). + while remaining > _WAIT_EPSILON_S and not self._stopping: + this_slice = min(remaining, SHUTDOWN_POLL_INTERVAL_S) + time.sleep(this_slice) + remaining -= this_slice + return self._stopping + + def periodic(self) -> None: + """Run one poll with in-tick retries; never let an error escape.""" + if self._stagger_first_poll_after_fork(): + return + + after = [self._retry_delay(1), self._retry_delay(2)] + poll = retry( + after=after, + # Stop retrying on a decisive response, or as soon as a shutdown is + # requested (the backoff wait below returns early in that case). + until=lambda r: self._stopping or not _is_retryable_status(r.status), + sleep_func=self._wait, + )(self._request) + try: + response = poll() + except RetryError as e: + # All attempts failed with retryable outcomes; keep last-known-good. + self._warn_failure(e.args[0], MAX_ATTEMPTS) + return + except Exception: + log.debug("Feature Flagging agentless poll failed unexpectedly", exc_info=True) + return + + # A shutdown mid-poll leaves the response unusable for state transitions; + # keep last-known-good and the current ETag. + if self._stopping: + return + + self._apply(response) + + def _stagger_first_poll_after_fork(self) -> bool: + """Delay the first poll in a forked child so workers don't poll in lockstep. + + The process that created the source (or any single-process run) is never + delayed, so behavior matches dd-trace-js and leaves tests unaffected. A + forked worker waits a random bounded delay before its first poll only. + + Returns True if a shutdown was requested while waiting, in which case the + caller should skip the poll. + """ + pid = os.getpid() + if pid == self._origin_pid or self._jittered_pid == pid: + return self._stopping + self._jittered_pid = pid + delay = random.uniform(0, min(self.interval, FIRST_POLL_JITTER_MAX_S)) # nosec B311 + return self._wait(delay) + + def _retry_delay(self, attempt: int) -> float: + if attempt == 1: + base = _clamp(self.interval / 6, FIRST_RETRY_MIN_S, FIRST_RETRY_MAX_S) + else: + base = _clamp(self.interval / 3, SECOND_RETRY_MIN_S, SECOND_RETRY_MAX_S) + jitter = 1 - RETRY_JITTER + random.random() * RETRY_JITTER * 2 # nosec B311 + return max(1.0, base * jitter) + + # -- request ------------------------------------------------------------ + + def _headers(self) -> "dict[str, str]": + headers = { + "Accept-Encoding": "gzip", + "DD-Client-Library-Language": "python", + "DD-Client-Library-Version": _pep440_to_semver(), + } + if self._api_key: + headers["DD-API-KEY"] = self._api_key + if self._etag: + headers["If-None-Match"] = self._etag + return headers + + def _request(self) -> _PollResponse: + conn = None + try: + conn = get_connection(self._conn_url, timeout=self._request_timeout) + # Publish the connection so a concurrent shutdown can half-close it + # instead of waiting out the request timeout. Claiming the slot and + # re-reading the stop flag under one lock keeps a poll from starting + # after _stop_service has already looked for something to cancel. + with self._conn_lock: + if self._stopping: + return _PollResponse(status=None, etag=None, content_encoding=None, body=None, error=None) + self._in_flight_conn = conn + # Suppress self-tracing: no HTTP span, no trace-header injection. + setattr(conn, _HTTPLIB_NO_TRACE_REQUEST, True) + conn.request("GET", self._request_target, None, self._headers()) # type: ignore[no-untyped-call] + resp = conn.getresponse() + body = resp.read() + return _PollResponse( + status=resp.status, + etag=resp.getheader("ETag"), + content_encoding=resp.getheader("Content-Encoding"), + body=body, + error=None, + ) + except Exception as e: + return _PollResponse(status=None, etag=None, content_encoding=None, body=None, error=e) + finally: + with self._conn_lock: + self._in_flight_conn = None + if conn is not None: + conn.close() + + # -- response handling -------------------------------------------------- + + def _apply(self, response: _PollResponse) -> None: + status = response.status + + if status == 304: + return + if status in (401, 403): + self._warn_failure(response, 1) + return + if status != 200: + # Non-2xx bodies are not decoded as config. + return + + try: + body = decode_response_body(response.body, response.content_encoding) + attributes = parse_ufc_configuration(body) + except Exception: + if not self._malformed_payload_logged: + self._malformed_payload_logged = True + log.error("Feature Flagging agentless endpoint returned malformed UFC payload") + return + + try: + self._apply_configuration(attributes) + except Exception as e: + if not self._application_failure_logged: + self._application_failure_logged = True + log.warning("Feature Flagging agentless UFC payload could not be applied: %s", e) + return + + # Advance the ETag only after parse AND apply both succeed. A blank or + # absent ETag clears the previous one. + etag = (response.etag or "").strip() + self._etag = etag or None + + def _warn_failure(self, response: _PollResponse, attempts: int) -> None: + status = response.status + if status in (401, 403): + category = "authentication" + elif status: + category = "http" + else: + category = "request" + + if category in self._failure_warnings: + return + self._failure_warnings.add(category) + + if status in (401, 403): + log.warning("Feature Flagging agentless endpoint returned HTTP %d; verify endpoint authentication", status) + elif status: + log.warning("Feature Flagging agentless endpoint returned HTTP %d after %d attempts", status, attempts) + elif attempts > 1: + log.warning("Feature Flagging agentless request failed after %d attempts: %s", attempts, response.error) + else: + log.warning("Feature Flagging agentless request failed: %s", response.error) diff --git a/ddtrace/internal/openfeature/_native.py b/ddtrace/internal/openfeature/_native.py index d1a0e422a92..7835d5f1011 100644 --- a/ddtrace/internal/openfeature/_native.py +++ b/ddtrace/internal/openfeature/_native.py @@ -20,7 +20,7 @@ ResolutionDetails = ffe.ResolutionDetails -def process_ffe_configuration(config): +def process_ffe_configuration(config) -> bool: """ Process FFE configuration and store as native Configuration object. @@ -28,6 +28,12 @@ def process_ffe_configuration(config): Args: config: Configuration dict in format {"flags": {...}} or wrapped format + + Returns: + True when the configuration was applied, False when the native library + rejected it. On rejection the previously applied configuration is left + in place, so callers that track delivery state (e.g. the agentless + source's ETag) must not treat a False result as success. """ try: config_json = json.dumps(config) @@ -40,6 +46,7 @@ def process_ffe_configuration(config): from ddtrace.internal.openfeature._provider import _notify_providers_config_received _notify_providers_config_received() + return True except ValueError as e: log.debug( "Failed to parse FFE configuration. The native library expects complete server format with: " @@ -48,6 +55,7 @@ def process_ffe_configuration(config): e, exc_info=True, ) + return False def resolve_flag( diff --git a/ddtrace/internal/openfeature/_provider.py b/ddtrace/internal/openfeature/_provider.py index e78ff789337..dc062ef9177 100644 --- a/ddtrace/internal/openfeature/_provider.py +++ b/ddtrace/internal/openfeature/_provider.py @@ -11,9 +11,14 @@ import time import typing + +if typing.TYPE_CHECKING: + from ddtrace.internal.openfeature._agentless_source import AgentlessConfigurationSource + from openfeature.evaluation_context import EvaluationContext from openfeature.event import ProviderEventDetails from openfeature.exception import ErrorCode +from openfeature.exception import ProviderNotReadyError from openfeature.flag_evaluation import FlagResolutionDetails from openfeature.flag_evaluation import FlagValueType from openfeature.flag_evaluation import Reason @@ -113,9 +118,22 @@ def __init__( self._metadata = Metadata(name="Datadog") self._status = ProviderStatus.NOT_READY - # Event set when the first RC config arrives; used by on_configuration_received() + # Event set when the first config arrives; used by on_configuration_received() # to guard the first-config path and by _emit_ready_event() timing. self._config_received = threading.Event() + self._initialization_timed_out = False + + # Read through a fresh config instance so values reflect the environment at + # provider-construction time rather than at import time (see the killswitch + # note below for the same reasoning). + instance_config = OpenFeatureConfig() + + # How long initialize() waits for that first config, in seconds. The + # constructor argument wins so embedders and tests can override the + # environment; otherwise the configured value applies. + if initialization_timeout is None: + initialization_timeout = instance_config.initialization_timeout_ms / 1000.0 + self._initialization_timeout = initialization_timeout # Cache for reported exposures to prevent duplicates # Stores mapping of (flag_key, subject_id) -> (allocation_key, variant_key) @@ -124,19 +142,33 @@ def __init__( maxsize=65536 ) - # Check if experimental flagging provider is enabled - self._enabled = ffe_config.experimental_flagging_provider_enabled - if not self._enabled: + # Master gate: the resolved configuration source (stable kill switch + + # source selection, with legacy grandfathering). Mirrors dd-trace-js, + # where the provider only activates when a delivery source is selected. + from ddtrace.internal.openfeature._source_selection import DISABLED + from ddtrace.internal.openfeature._source_selection import resolve_configuration_source + + self._active = resolve_configuration_source(ffe_config) != DISABLED + if not self._active: logger.warning( - "openfeature: experimental flagging provider is not enabled, " - "please set DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED=true to enable it", + "openfeature: Feature Flagging provider is disabled; set DD_FEATURE_FLAGS_ENABLED=true and " + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE to a supported value to enable it", ) + # NOTE: the legacy DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED flag is an input + # to source grandfathering (handled during resolution above) only. Evaluation, + # hooks and telemetry all key off the resolved decision (``_active``), matching + # dd-trace-js where the same switch gates the provider and its writers. + + # Agentless configuration-source poller; started in initialize() when + # agentless is the resolved source and stopped in shutdown(). + self._configuration_source: typing.Optional["AgentlessConfigurationSource"] = None + # 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_metrics_hook: typing.Optional[FlagEvalMetricsHook] = None - if self._enabled: + if self._active: self._flag_eval_metrics = FlagEvalMetrics() self._flag_eval_metrics_hook = FlagEvalMetricsHook(self._flag_eval_metrics) @@ -145,16 +177,15 @@ def __init__( # 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 + # supported-configurations.json) rather than raw os.environ. It comes from the + # fresh instance_config built at the top of __init__, so the value reflects the # 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: + evp_counts_enabled = instance_config.flagging_evaluation_counts_enabled + if self._active and evp_counts_enabled: self._flag_eval_evp_writer = FlagEvaluationWriter() self._flag_eval_evp_hook = FlagEvalEVPHook(self._flag_eval_evp_writer) @@ -162,7 +193,7 @@ def __init__( # Constructed ONLY when the gate is on, so nothing is allocated and # nothing subscribes to span finish when it is off (DG-005). self._span_enrichment_hook: typing.Optional[SpanEnrichmentHook] = None - if self._enabled and ffe_config.experimental_flagging_provider_span_enrichment_enabled: + if self._active and ffe_config.experimental_flagging_provider_span_enrichment_enabled: self._span_enrichment_hook = SpanEnrichmentHook() def get_metadata(self) -> Metadata: @@ -172,7 +203,7 @@ def get_metadata(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: + if self._active: _register_provider(self) def get_provider_hooks(self) -> list[typing.Any]: @@ -203,29 +234,37 @@ def get_provider_hooks(self) -> list[typing.Any]: def initialize(self, evaluation_context: EvaluationContext) -> None: """ - Initialize the provider. + Initialize the provider, waiting up to the initialization timeout for config. - 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. + Blocking bounds PROVIDER_READY to a provider that can actually resolve flags, + matching the other server SDKs. A timeout raises ProviderNotReadyError so the + OpenFeature SDK reports PROVIDER_ERROR instead of a false PROVIDER_READY. + on_configuration_received() promotes the provider whenever the payload does land. + Evaluations before that return the caller-provided default with + ErrorCode.PROVIDER_NOT_READY. - 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. + If config already arrived before initialize() runs (e.g. in the master process of + a pre-fork server), the fast path sets READY without waiting. Provider lifecycle: - NOT_READY -> initialize() returns -> RC delivers config -> on_configuration_received() - -> READY + NOT_READY -> initialize() waits -> config arrives -> READY + NOT_READY -> initialize() waits -> timeout -> ERROR + -> config arrives -> on_configuration_received() -> READY """ - if not self._enabled: + if not self._active: return + self._initialization_timed_out = False + # Register for RC config callbacks (in initialize, not __init__, so # re-initialization after shutdown re-registers the provider) _register_provider(self) + # Start the agentless poller when agentless is the resolved source + # (no-op otherwise). Mirrors dd-trace-js starting the source from the + # provider lifecycle rather than at tracer init. + self._start_configuration_source() + try: # Start the exposure writer for reporting start_exposure_writer() @@ -249,18 +288,24 @@ def initialize(self, evaluation_context: EvaluationContext) -> None: self._status = ProviderStatus.READY return # SDK will dispatch PROVIDER_READY - # 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. + # Config not yet available — wait for the source to deliver it, so a caller + # that observes PROVIDER_READY can trust the first evaluation. + if self._config_received.wait(timeout=self._initialization_timeout): + self._status = ProviderStatus.READY + return # SDK will dispatch PROVIDER_READY + + # AIDEV-NOTE: the wait above must stay bounded. A blocked initialize() blocks + # set_provider() in openfeature-sdk 0.8.x and set_provider_and_wait() in 0.10+. + # The 10s default stays inside gunicorn's 30s worker timeout. Raising the + # OpenFeature error is also required: the SDK converts it to PROVIDER_ERROR for + # non-blocking registration and propagates it for blocking registration. + logger.warning( + "openfeature: no Feature Flagging configuration received after %.1fs; evaluations will return " + "default values until configuration arrives", + self._initialization_timeout, + ) + self._initialization_timed_out = True + raise ProviderNotReadyError("No Feature Flagging configuration received before the initialization timeout") def shutdown(self) -> None: """ @@ -268,9 +313,12 @@ def shutdown(self) -> None: Called by the OpenFeature SDK when the provider is being replaced or shutdown. """ - if not self._enabled: + if not self._active: return + # Stop the agentless poller if it was started. + self._stop_configuration_source() + try: # Stop the exposure writer stop_exposure_writer() @@ -308,6 +356,36 @@ def shutdown(self) -> None: _unregister_provider(self) self._status = ProviderStatus.NOT_READY self._config_received.clear() + self._initialization_timed_out = False + + def _start_configuration_source(self) -> None: + """Start the agentless poller when agentless is the resolved source.""" + if self._configuration_source is not None: + return + + from ddtrace.internal.openfeature._source_selection import create_agentless_source + + source = create_agentless_source(ffe_config, _apply_agentless_configuration) + if source is None: + return + + try: + source.start() + except ServiceStatusError: + logger.debug("Agentless configuration source is already running", exc_info=True) + self._configuration_source = source + + def _stop_configuration_source(self) -> None: + """Stop the agentless poller if it was started.""" + if self._configuration_source is None: + return + + try: + self._configuration_source.stop() + self._configuration_source.join() + except ServiceStatusError: + logger.debug("Agentless configuration source is already stopped", exc_info=True) + self._configuration_source = None def resolve_boolean_details( self, @@ -371,8 +449,8 @@ def _resolve_details( # 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: + # If provider is not active, return default value + if not self._active: return FlagResolutionDetails( value=default_value, reason=Reason.DISABLED, @@ -590,15 +668,15 @@ def on_configuration_received(self) -> None: Called when a Remote Configuration payload is received and processed. 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. + Emits PROVIDER_READY for late arrivals after initialization fails. """ 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 initialize() returned. - self._emit_ready_event() + # The SDK emits READY after successful initialization. The provider only + # owns the recovery event after initialization reported an error. + if self._initialization_timed_out: + self._emit_ready_event() # Signal the event last after status is updated. self._config_received.set() @@ -625,6 +703,22 @@ def clear_exposure_cache(self) -> None: logger.debug("Exposure cache cleared") +def _apply_agentless_configuration(configuration: "dict[str, typing.Any]") -> None: + """Apply a UFC payload delivered by the agentless source. + + ``process_ffe_configuration`` reports a payload the native evaluator refused + by returning False rather than raising, which the agentless source would + otherwise read as success and advance its ETag past a configuration it never + loaded (the next poll would then get a 304 and keep the stale config + indefinitely). Translate a rejection into an error so the source keeps + last-known-good and retries the payload on the next poll. + """ + from ddtrace.internal.openfeature._native import process_ffe_configuration + + if not process_ffe_configuration(configuration): + raise ValueError("Feature Flagging configuration was rejected by the evaluator") + + # Module-level registry for active provider instances _provider_instances: list[DataDogProvider] = [] diff --git a/ddtrace/internal/openfeature/_source_selection.py b/ddtrace/internal/openfeature/_source_selection.py new file mode 100644 index 00000000000..a1cc515ecd9 --- /dev/null +++ b/ddtrace/internal/openfeature/_source_selection.py @@ -0,0 +1,102 @@ +""" +Feature Flagging configuration-source selection. + +Resolves which delivery source is active (agentless CDN, Agent Remote Config, or +disabled) from the stable kill switch, the explicit source setting, and the +legacy experimental flag used for grandfathering. Mirrors the dd-trace-js +resolution (kill switch -> explicit source -> grandfathering -> default) and the +agentless factory, adapted to dd-trace-py config conventions. +""" + +from typing import Any +from typing import Callable +from typing import Optional + +from ddtrace.internal.logger import get_logger +from ddtrace.internal.openfeature._agentless import build_agentless_endpoint +from ddtrace.internal.openfeature._agentless_source import AgentlessConfigurationSource +from ddtrace.internal.settings._core import ValueSource +from ddtrace.internal.settings.openfeature import OpenFeatureConfig + + +log = get_logger(__name__) + +AGENTLESS = "agentless" +REMOTE_CONFIG = "remote_config" +DISABLED = "disabled" + +_SOURCE_ENV = "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE" +_LEGACY_ENV = "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED" + + +def _provided(ffe_config: OpenFeatureConfig, env_name: str) -> bool: + """True when the value came from an external source rather than the default.""" + return ffe_config.value_source(env_name) != ValueSource.DEFAULT + + +def resolve_configuration_source(ffe_config: OpenFeatureConfig) -> str: + """Resolve the active source: ``agentless``, ``remote_config`` or ``disabled``. + + Precedence (mirrors dd-trace-js): + + 1. Stable kill switch off -> disabled. + 2. Explicit source -> use it; an unsupported/reserved value (e.g. ``offline``) + fails closed to disabled without contacting any source. + 3. Source absent -> grandfather on the legacy experimental flag: explicitly + true -> remote_config, explicitly false -> disabled. + 4. Otherwise the default -> agentless. + """ + if not ffe_config.feature_flags_enabled: + return DISABLED + + source = ffe_config.configuration_source or "" + if _provided(ffe_config, _SOURCE_ENV) and source: + if source == AGENTLESS: + return AGENTLESS + if source == REMOTE_CONFIG: + return REMOTE_CONFIG + log.warning("Unsupported Feature Flagging configuration source %r; provider disabled", source) + return DISABLED + + # Source absent (unset or blank): preserve legacy Remote Config grandfathering. + if _provided(ffe_config, _LEGACY_ENV): + return REMOTE_CONFIG if ffe_config.experimental_flagging_provider_enabled else DISABLED + + return AGENTLESS + + +def create_agentless_source( + ffe_config: OpenFeatureConfig, apply_configuration: Callable[["dict[str, Any]"], None] +) -> Optional[AgentlessConfigurationSource]: + """Build the agentless poller when agentless is the resolved source, else None. + + The default Datadog endpoint requires ``DD_API_KEY`` and sends it. A custom + endpoint is operator-owned trust: it starts without an API key and omits the + header, letting the endpoint report any authentication failure itself. + """ + if resolve_configuration_source(ffe_config) != AGENTLESS: + return None + + from ddtrace import config as dd_config + + base_url = ffe_config.configuration_source_agentless_base_url + has_custom_endpoint = bool(base_url and base_url.strip()) + api_key = dd_config._dd_api_key + + if not has_custom_endpoint and not api_key: + log.error("DD_API_KEY is required for the default Datadog Feature Flagging endpoint") + return None + + try: + endpoint = build_agentless_endpoint(dd_config._dd_site, dd_config.env, base_url) + except ValueError as e: + log.error("Unable to configure Feature Flagging configuration source: %s", e) + return None + + return AgentlessConfigurationSource( + endpoint=endpoint, + apply_configuration=apply_configuration, + api_key=None if has_custom_endpoint else api_key, + poll_interval=ffe_config.configuration_source_agentless_poll_interval_seconds, + request_timeout=ffe_config.configuration_source_agentless_request_timeout_seconds, + ) diff --git a/ddtrace/internal/openfeature/product.py b/ddtrace/internal/openfeature/product.py index a691f5e7697..3d6c365c0be 100644 --- a/ddtrace/internal/openfeature/product.py +++ b/ddtrace/internal/openfeature/product.py @@ -9,13 +9,23 @@ def post_preload(): def enabled(): - return ffe_config.experimental_flagging_provider_enabled + from ddtrace.internal.openfeature._source_selection import DISABLED + from ddtrace.internal.openfeature._source_selection import resolve_configuration_source + + return resolve_configuration_source(ffe_config) != DISABLED def start(): - from ddtrace.internal.openfeature._remoteconfiguration import enable_featureflags_rc + # Agent Remote Config delivery is activated only when it is the resolved + # source. The agentless source is started from the provider lifecycle + # (mirroring dd-trace-js), so there is nothing to start here for agentless. + from ddtrace.internal.openfeature._source_selection import REMOTE_CONFIG + from ddtrace.internal.openfeature._source_selection import resolve_configuration_source + + if resolve_configuration_source(ffe_config) == REMOTE_CONFIG: + from ddtrace.internal.openfeature._remoteconfiguration import enable_featureflags_rc - enable_featureflags_rc() + enable_featureflags_rc() def restart(join=False): @@ -23,6 +33,10 @@ def restart(join=False): def stop(join=False): - from ddtrace.internal.openfeature._remoteconfiguration import disable_featureflags_rc + from ddtrace.internal.openfeature._source_selection import REMOTE_CONFIG + from ddtrace.internal.openfeature._source_selection import resolve_configuration_source + + if resolve_configuration_source(ffe_config) == REMOTE_CONFIG: + from ddtrace.internal.openfeature._remoteconfiguration import disable_featureflags_rc - disable_featureflags_rc() + disable_featureflags_rc() diff --git a/ddtrace/internal/settings/_supported_configurations.py b/ddtrace/internal/settings/_supported_configurations.py index 053a04aa6b9..1c501b475db 100644 --- a/ddtrace/internal/settings/_supported_configurations.py +++ b/ddtrace/internal/settings/_supported_configurations.py @@ -229,6 +229,11 @@ "DD_FASTAPI_ASYNC_BODY_TIMEOUT_SECONDS", "DD_FASTAPI_SERVICE", "DD_FAST_BUILD", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS", + "DD_FEATURE_FLAGS_ENABLED", "DD_FFE_INTAKE_ENABLED", "DD_FFE_INTAKE_HEARTBEAT_INTERVAL", "DD_FLAGGING_EVALUATION_COUNTS_ENABLED", @@ -972,6 +977,7 @@ { "DD_API_KEY", "DD_APP_KEY", + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL", "OTEL_EXPORTER_OTLP_HEADERS", "OTEL_EXPORTER_OTLP_LOGS_HEADERS", "OTEL_EXPORTER_OTLP_METRICS_HEADERS", diff --git a/ddtrace/internal/settings/openfeature.py b/ddtrace/internal/settings/openfeature.py index a410686f17b..78a653263f5 100644 --- a/ddtrace/internal/settings/openfeature.py +++ b/ddtrace/internal/settings/openfeature.py @@ -2,6 +2,8 @@ OpenFeature configuration settings. """ +from typing import Optional + from ddtrace.internal.settings._core import DDConfig @@ -48,15 +50,61 @@ class OpenFeatureConfig(DDConfig): default=1.0, ) - # Provider initialization timeout in milliseconds. - # Controls how long initialize() blocks waiting for the first Remote Config payload. - # Default is 10000ms (10 seconds). + # Provider initialization timeout in milliseconds. Controls how long initialize() + # blocks waiting for the first configuration payload, from either configuration + # source. Expiry is not an error; the provider stays NOT_READY and becomes READY when + # configuration arrives. + # Default is 10000ms: long enough for a healthy delivery path, and short enough that a + # pre-fork worker boots inside gunicorn's 30s default worker timeout. Raising it much + # further risks the worker being killed before it finishes starting. initialization_timeout_ms = DDConfig.var( int, "DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS", default=10000, ) + # Stable Feature Flagging kill switch. When False, the provider is disabled + # regardless of the configured source. Default on. + feature_flags_enabled = DDConfig.var( + bool, + "DD_FEATURE_FLAGS_ENABLED", + default=True, + ) + + # Where Feature Flagging loads Universal Flag Configuration from. + # Supported: "agentless" (default) and "remote_config"; "offline" is reserved + # and currently unsupported. Normalized to trimmed lowercase; validity and + # grandfathering are resolved by the source-selection layer. + configuration_source = DDConfig.var( + str, + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE", + default="agentless", + parser=lambda v: v.strip().lower(), + ) + + # Optional override of the agentless UFC endpoint or base URL. A root/origin + # URL receives the standard rules-based path; a non-root URL is used verbatim. + configuration_source_agentless_base_url = DDConfig.var( + Optional[str], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL", + default=None, + parser=lambda v: v.strip() or None, + ) + + # Agentless UFC polling interval in seconds, capped at one hour by the source. + configuration_source_agentless_poll_interval_seconds = DDConfig.var( + int, + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS", + default=30, + ) + + # Agentless UFC per-request timeout in seconds. + configuration_source_agentless_request_timeout_seconds = DDConfig.var( + int, + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS", + default=5, + ) + _openfeature_config_keys = [ "experimental_flagging_provider_enabled", "experimental_flagging_provider_span_enrichment_enabled", @@ -64,6 +112,11 @@ class OpenFeatureConfig(DDConfig): "ffe_intake_enabled", "ffe_intake_heartbeat_interval", "initialization_timeout_ms", + "feature_flags_enabled", + "configuration_source", + "configuration_source_agentless_base_url", + "configuration_source_agentless_poll_interval_seconds", + "configuration_source_agentless_request_timeout_seconds", ] diff --git a/docs/configuration.rst b/docs/configuration.rst index 5327458a51e..16b5e0593ca 100644 --- a/docs/configuration.rst +++ b/docs/configuration.rst @@ -1206,6 +1206,57 @@ Sampling v1.20.0: added support for "tags" v2.8.0: added lazy sampling support, so that spans are evaluated at the end of the trace, guaranteeing more metadata to evaluate against. +Feature Flagging +---------------- + +.. ddtrace-configuration-options:: + + DD_FEATURE_FLAGS_ENABLED: + type: Boolean + default: True + description: | + Stable kill switch for Feature Flagging. When ``False``, the provider is + disabled regardless of the configured source. + + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE: + type: String + default: agentless + description: | + Selects where Feature Flagging loads Universal Flag Configuration from. + Supported values are ``agentless`` (load directly from the Datadog CDN) + and ``remote_config`` (deliver via the Datadog Agent's Remote + Configuration). ``offline`` is reserved and currently unsupported; any + unsupported value disables the provider without contacting either source. + + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL: + type: String + default: (none) + description: | + Overrides the Datadog-managed agentless Universal Flag Configuration + endpoint, for local development or an operator-managed proxy. The URL must + use HTTP or HTTPS. An origin or root URL receives the standard rules-based + server path, so ``http://localhost:8080`` resolves to + ``http://localhost:8080/api/v2/feature-flagging/config/rules-based/server``; + a URL with a non-root path, such as + ``https://ufc-proxy.internal.example.com/ufc``, is used verbatim as the + exact endpoint. ``DD_API_KEY`` is never sent to a custom endpoint. Only + applies when ``DD_FEATURE_FLAGS_CONFIGURATION_SOURCE`` is ``agentless``. + See `Use a custom agentless endpoint `_. + + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS: + type: Integer + default: 30 + description: | + The agentless Universal Flag Configuration polling interval in seconds, + capped at one hour. + + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS: + type: Integer + default: 5 + description: | + The per-request timeout in seconds for agentless Universal Flag + Configuration polls. + Other ----- diff --git a/releasenotes/notes/agentless-feature-flag-configuration-source-6b2f9ae3c47d105e.yaml b/releasenotes/notes/agentless-feature-flag-configuration-source-6b2f9ae3c47d105e.yaml new file mode 100644 index 00000000000..a6c83c60061 --- /dev/null +++ b/releasenotes/notes/agentless-feature-flag-configuration-source-6b2f9ae3c47d105e.yaml @@ -0,0 +1,8 @@ +--- +features: + - | + openfeature: Adds agentless delivery for the Feature Flagging and Experimentation + (FFE) OpenFeature provider, loading Universal Flag Configuration directly from Datadog + over HTTPS without requiring a Datadog Agent. Agentless is now the default source and + requires ``DD_API_KEY``; set ``DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=remote_config`` to + keep loading configuration through the Datadog Agent's Remote Configuration. diff --git a/releasenotes/notes/fix-openfeature-init-blocking-70c8d5a99287cc49.yaml b/releasenotes/notes/fix-openfeature-init-blocking-70c8d5a99287cc49.yaml index a6de2e99e1d..98e50424254 100644 --- a/releasenotes/notes/fix-openfeature-init-blocking-70c8d5a99287cc49.yaml +++ b/releasenotes/notes/fix-openfeature-init-blocking-70c8d5a99287cc49.yaml @@ -4,11 +4,13 @@ fixes: openfeature: This fix resolves an issue where ``DataDogProvider.initialize()`` returned before configuration was received, causing the OpenFeature SDK to mark the provider as ready to serve evaluations too early and flag evaluations to silently return default values. The provider now - waits for configuration before returning. + waits up to the initialization timeout for configuration before returning. If the timeout + expires, initialization still succeeds and the provider becomes ready once configuration + arrives; evaluations until then return the caller-provided default value. features: - | openfeature: This introduces a configurable initialization timeout for ``DataDogProvider``. The timeout controls how long ``initialize()`` waits for configuration before returning, and defaults to 10 seconds. Set it via the ``DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS`` environment variable or the - ``init_timeout`` constructor parameter. + ``initialization_timeout`` constructor parameter. diff --git a/supported-configurations.json b/supported-configurations.json index 2258bbc6faf..a59badab418 100644 --- a/supported-configurations.json +++ b/supported-configurations.json @@ -1704,6 +1704,42 @@ "default": "false" } ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE": [ + { + "implementation": "A", + "type": "string", + "default": "agentless" + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL": [ + { + "implementation": "A", + "type": "string", + "default": null, + "sensitive": true + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS": [ + { + "implementation": "A", + "type": "int", + "default": "30" + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS": [ + { + "implementation": "A", + "type": "int", + "default": "5" + } + ], + "DD_FEATURE_FLAGS_ENABLED": [ + { + "implementation": "A", + "type": "boolean", + "default": "true" + } + ], "DD_FFE_INTAKE_ENABLED": [ { "implementation": "A", diff --git a/tests/openfeature/conftest.py b/tests/openfeature/conftest.py index a13d86bfae7..2c72a1b0a89 100644 --- a/tests/openfeature/conftest.py +++ b/tests/openfeature/conftest.py @@ -1,3 +1,17 @@ """ Shared fixtures for openfeature tests. """ + +import pytest + + +@pytest.fixture(autouse=True) +def _no_initialization_wait(monkeypatch): + """Stop initialize() from spending its full timeout in tests that never deliver config. + + Most tests construct a provider with no configuration available, so the production + 10s wait would be paid once per provider. Tests that care about the wait itself set + the timeout explicitly, either through this environment variable or the + initialization_timeout constructor argument. + """ + monkeypatch.setenv("DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS", "0") diff --git a/tests/openfeature/test_agentless_poller.py b/tests/openfeature/test_agentless_poller.py new file mode 100644 index 00000000000..99e5a4e49d5 --- /dev/null +++ b/tests/openfeature/test_agentless_poller.py @@ -0,0 +1,469 @@ +import gzip +import json +import socket + +import pytest + +from ddtrace.internal.constants import _HTTPLIB_NO_TRACE_REQUEST +import ddtrace.internal.openfeature._agentless_source as source_mod +from ddtrace.internal.openfeature._agentless_source import MAX_POLL_INTERVAL_SECONDS +from ddtrace.internal.openfeature._agentless_source import AgentlessConfigurationSource + + +ENDPOINT = "https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server?dd_env=prod" + + +def _ufc_body(): + return json.dumps( + { + "data": { + "id": "1", + "type": "universal-flag-configuration", + "attributes": { + "format": "SERVER", + "createdAt": "2024-01-01T00:00:00Z", + "environment": {"name": "production"}, + "flags": {"my-flag": {"enabled": True}}, + }, + } + } + ).encode("utf-8") + + +class _FakeResponse: + def __init__(self, status, body=b"", headers=None): + self.status = status + self._body = body + self._headers = {k.lower(): v for k, v in (headers or {}).items()} + + def read(self): + return self._body + + def getheader(self, name, default=None): + return self._headers.get(name.lower(), default) + + +class _FakeSocket: + """Records the half-close a shutdown performs to cancel an in-flight poll.""" + + def __init__(self, error=None): + self.shutdowns: list = [] + self._error = error + + def shutdown(self, how): + if self._error is not None: + raise self._error + self.shutdowns.append(how) + + +class _FakeConn: + def __init__(self, responses, requests, sock=None): + self._responses = responses + self._requests = requests + self.closed = False + # http.client connects lazily, so a connection only grows a socket once + # the request is on the wire. Tests that exercise cancellation set one. + self.sock = sock + + def request(self, method, target, body, headers): + self._requests.append( + { + "method": method, + "target": target, + "headers": headers, + "no_trace": getattr(self, _HTTPLIB_NO_TRACE_REQUEST, False), + } + ) + + def getresponse(self): + item = self._responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + def close(self): + self.closed = True + + +@pytest.fixture +def harness(monkeypatch): + """Return a factory building a poller with a scripted fake HTTP layer.""" + + def build(responses, api_key=None, poll_interval=30.0, sock=None): + requests: list = [] + applied: list = [] + conns: list = [] + + def fake_get_connection(url, timeout=None): + conn = _FakeConn(responses, requests, sock=sock) + conns.append(conn) + return conn + + monkeypatch.setattr(source_mod, "get_connection", fake_get_connection) + + src = AgentlessConfigurationSource( + endpoint=ENDPOINT, + apply_configuration=applied.append, + api_key=api_key, + poll_interval=poll_interval, + ) + # Keep retries instant. + monkeypatch.setattr(src, "_retry_delay", lambda attempt: 0.0) + src._requests = requests + src._applied = applied + src._conns = conns + return src + + return build + + +def test_200_applies_and_advances_etag(harness): + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})]) + src.periodic() + assert len(src._applied) == 1 + assert src._applied[0]["environment"]["name"] == "production" + assert src._etag == '"v1"' + + +def test_gzip_body_is_decoded(harness): + body = gzip.compress(_ufc_body()) + src = harness([_FakeResponse(200, body, {"Content-Encoding": "gzip", "ETag": '"gz"'})]) + src.periodic() + assert len(src._applied) == 1 + assert src._etag == '"gz"' + + +def test_304_is_noop_and_preserves_state(harness): + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'}), _FakeResponse(304)]) + src.periodic() + src.periodic() + assert len(src._applied) == 1 # only the first 200 applied + assert src._etag == '"v1"' # preserved across the 304 + + +def test_blank_etag_clears_previous(harness): + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'}), _FakeResponse(200, _ufc_body(), {})]) + src.periodic() + assert src._etag == '"v1"' + src.periodic() + assert src._etag is None + + +def test_401_warns_and_does_not_apply(harness): + src = harness([_FakeResponse(401)]) + src.periodic() + assert src._applied == [] + + +def test_malformed_payload_preserves_last_known_good(harness): + src = harness( + [_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'}), _FakeResponse(200, b"not json", {"ETag": '"v2"'})] + ) + src.periodic() + src.periodic() + assert len(src._applied) == 1 # malformed second poll not applied + assert src._etag == '"v1"' # etag not advanced on malformed + + +def test_apply_failure_does_not_advance_etag(monkeypatch, harness): + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})]) + + def boom(_): + raise RuntimeError("apply failed") + + monkeypatch.setattr(src, "_apply_configuration", boom) + src.periodic() + assert src._etag is None + + +def test_retryable_500_then_success(harness): + src = harness([_FakeResponse(500), _FakeResponse(200, _ufc_body(), {"ETag": '"ok"'})]) + src.periodic() + assert len(src._applied) == 1 + assert len(src._requests) == 2 # retried once + + +def test_network_error_is_retryable(harness): + src = harness([OSError("boom"), _FakeResponse(200, _ufc_body(), {"ETag": '"ok"'})]) + src.periodic() + assert len(src._applied) == 1 + assert len(src._requests) == 2 + + +def test_retries_exhausted_no_apply(harness): + src = harness([_FakeResponse(500), _FakeResponse(503), _FakeResponse(500)]) + src.periodic() + assert src._applied == [] + assert len(src._requests) == 3 # MAX_ATTEMPTS + + +def test_non_retryable_status_not_retried(harness): + src = harness([_FakeResponse(404)]) + src.periodic() + assert src._applied == [] + assert len(src._requests) == 1 # 404 is not retried + + +def test_if_none_match_sent_when_etag_held(harness): + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'}), _FakeResponse(304)]) + src.periodic() + src.periodic() + assert "If-None-Match" not in src._requests[0]["headers"] + assert src._requests[1]["headers"]["If-None-Match"] == '"v1"' + + +def test_api_key_header_present_and_absent(harness): + with_key = harness([_FakeResponse(304)], api_key="secret") + with_key.periodic() + assert with_key._requests[0]["headers"]["DD-API-KEY"] == "secret" + + without_key = harness([_FakeResponse(304)], api_key=None) + without_key.periodic() + assert "DD-API-KEY" not in without_key._requests[0]["headers"] + + +def test_client_library_headers_and_gzip_accept(harness): + src = harness([_FakeResponse(304)]) + src.periodic() + headers = src._requests[0]["headers"] + assert headers["Accept-Encoding"] == "gzip" + assert headers["DD-Client-Library-Language"] == "python" + assert headers["DD-Client-Library-Version"] + + +def test_self_tracing_suppressed(harness): + src = harness([_FakeResponse(304)]) + src.periodic() + assert src._requests[0]["no_trace"] is True + + +def _spy_waits(src, monkeypatch): + """Record the positive delays the source waits on (ignores no-op 0 waits).""" + waits: list = [] + + def fake_wait(delay): + if delay > 0: + waits.append(delay) + return False + + monkeypatch.setattr(src, "_wait", fake_wait) + return waits + + +def test_origin_process_first_poll_is_not_jittered(harness, monkeypatch): + src = harness([_FakeResponse(304)]) + waits = _spy_waits(src, monkeypatch) + src.periodic() + assert waits == [] # origin process polls immediately + + +def test_forked_child_first_poll_is_jittered_once(harness, monkeypatch): + src = harness([_FakeResponse(304), _FakeResponse(304)]) + waits = _spy_waits(src, monkeypatch) + # Simulate a forked worker: a PID different from the creating process. + monkeypatch.setattr(source_mod.os, "getpid", lambda: src._origin_pid + 1) + + src.periodic() + assert len(waits) == 1 + assert 0 < waits[0] <= min(src.interval, source_mod.FIRST_POLL_JITTER_MAX_S) + + src.periodic() + assert len(waits) == 1 # only the first poll in the child is staggered + + +# --------------------------------------------------------------------------- +# Shutdown +# --------------------------------------------------------------------------- + + +def test_shutdown_during_backoff_stops_retrying(harness, monkeypatch): + """A shutdown requested while backing off must not start another attempt.""" + src = harness([_FakeResponse(500), _FakeResponse(200, _ufc_body(), {"ETag": '"late"'})]) + # A real backoff, so the fake wait below can tell it apart from the zero-length + # initial_wait retry() performs before the first attempt. Nothing actually + # sleeps: _wait is replaced outright. + monkeypatch.setattr(src, "_retry_delay", lambda attempt: 30.0) + + def wait_then_shutdown(delay): + # The real _wait returns immediately for a zero delay without observing a + # stop, so only a genuine backoff wait may request one here. + if not delay: + return src._stopping + src._stopping = True + return True + + monkeypatch.setattr(src, "_wait", wait_then_shutdown) + + src.periodic() + + assert len(src._requests) == 1 # retry abandoned + assert src._applied == [] + assert src._etag is None + + +def test_shutdown_mid_poll_does_not_apply(harness): + """A response that arrives after shutdown must not replace state.""" + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})]) + src._stopping = True + + src.periodic() + + assert src._applied == [] + assert src._etag is None + + +def test_shutdown_half_closes_the_socket_of_a_poll_in_flight(harness): + """A stop must tear down the open request instead of waiting out its timeout.""" + sock = _FakeSocket() + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})], sock=sock) + cancelled_at: list = [] + + # Stand in for the thread calling stop() while the worker blocks on the read. + original_getresponse = _FakeConn.getresponse + + def getresponse_with_concurrent_stop(conn): + src._stopping = True + src._cancel_in_flight_request() + cancelled_at.append(list(sock.shutdowns)) + return original_getresponse(conn) + + _FakeConn.getresponse = getresponse_with_concurrent_stop + try: + src.periodic() + finally: + _FakeConn.getresponse = original_getresponse + + # Half-closed while the request was still open, not after it returned. + assert cancelled_at == [[socket.SHUT_RDWR]] + assert src._applied == [] # a cancelled poll keeps last-known-good + assert src._etag is None + + +def test_shutdown_releases_the_connection_it_cancelled(harness): + """The worker still owns cleanup: the cancelled connection is closed.""" + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})], sock=_FakeSocket()) + + src.periodic() + + assert src._conns[0].closed is True + assert src._in_flight_conn is None # slot released for the next poll + + +def test_stop_service_cancels_before_joining(harness, monkeypatch): + """_stop_service half-closes the live socket, then defers to PeriodicService.""" + sock = _FakeSocket() + src = harness([_FakeResponse(304)], sock=sock) + joined: list = [] + # PeriodicService._stop_service joins the worker; stub it out so the test needs + # no running thread and can assert the cancel happened before the join. + monkeypatch.setattr( + source_mod.PeriodicService, + "_stop_service", + lambda self, *a, **kw: joined.append(list(sock.shutdowns)), + ) + with src._conn_lock: + src._in_flight_conn = _FakeConn([], [], sock=sock) + + src._stop_service() + + assert src._stopping is True + assert joined == [[socket.SHUT_RDWR]] + + +def test_cancel_is_a_noop_when_no_poll_is_in_flight(harness): + """Stopping an idle poller must not raise.""" + src = harness([_FakeResponse(304)]) + src._cancel_in_flight_request() # no connection published yet + assert src._in_flight_conn is None + + +def test_cancel_is_a_noop_before_the_socket_exists(harness): + """http.client connects lazily; a stop with no socket yet has nothing to close.""" + src = harness([_FakeResponse(304)]) + with src._conn_lock: + src._in_flight_conn = _FakeConn([], [], sock=None) + + src._cancel_in_flight_request() # must not raise on the missing socket + + +def test_cancel_tolerates_an_already_closed_socket(harness): + """A socket the worker closed first raises OSError; the stop swallows it.""" + src = harness([_FakeResponse(304)]) + with src._conn_lock: + src._in_flight_conn = _FakeConn([], [], sock=_FakeSocket(error=OSError("not connected"))) + + src._cancel_in_flight_request() # must not propagate + + +def test_no_request_is_issued_after_a_stop_was_requested(harness): + """A poll that starts after the stop flag is set never reaches the network.""" + src = harness([_FakeResponse(200, _ufc_body(), {"ETag": '"v1"'})]) + src._stopping = True + + src.periodic() + + assert src._requests == [] # returned before conn.request() + assert src._applied == [] + + +def test_backoff_wait_is_interruptible(harness): + """The backoff wait returns as soon as a shutdown is requested.""" + src = harness([_FakeResponse(304)]) + src._stopping = True + # A long delay must return immediately (True) rather than sleeping it out. + assert src._wait(3600) is True + + +def test_wait_stops_at_the_next_slice(harness, monkeypatch): + """A stop requested mid-wait ends it at the next slice, not after the full delay.""" + src = harness([_FakeResponse(304)]) + slept: list = [] + + def fake_sleep(delay): + slept.append(delay) + src._stopping = True # requested while the wait is in progress + + monkeypatch.setattr(source_mod.time, "sleep", fake_sleep) + + assert src._wait(3600) is True + assert slept == [source_mod.SHUTDOWN_POLL_INTERVAL_S] # one slice, not 3600s + + +def test_wait_sleeps_the_full_delay_when_not_stopping(harness, monkeypatch): + """Without a stop request the wait consumes the whole delay in slices.""" + src = harness([_FakeResponse(304)]) + slept: list = [] + monkeypatch.setattr(source_mod.time, "sleep", slept.append) + + assert src._wait(source_mod.SHUTDOWN_POLL_INTERVAL_S * 3) is False + assert len(slept) == 3 + + +def test_poll_interval_clamped_to_one_hour(): + src = AgentlessConfigurationSource( + endpoint=ENDPOINT, + apply_configuration=lambda _: None, + poll_interval=MAX_POLL_INTERVAL_SECONDS * 5, + ) + assert src.interval == MAX_POLL_INTERVAL_SECONDS + + +@pytest.mark.parametrize("bad_interval", [0, -1, -30.0]) +def test_non_positive_poll_interval_falls_back_to_default(bad_interval): + """A non-positive interval would busy-loop against the CDN; use the default.""" + src = AgentlessConfigurationSource( + endpoint=ENDPOINT, + apply_configuration=lambda _: None, + poll_interval=bad_interval, + ) + assert src.interval == source_mod.DEFAULT_POLL_INTERVAL_SECONDS + + +@pytest.mark.parametrize("bad_timeout", [0, -1, -5.0]) +def test_non_positive_request_timeout_falls_back_to_default(bad_timeout): + src = AgentlessConfigurationSource( + endpoint=ENDPOINT, + apply_configuration=lambda _: None, + request_timeout=bad_timeout, + ) + assert src._request_timeout == source_mod.DEFAULT_REQUEST_TIMEOUT_SECONDS diff --git a/tests/openfeature/test_agentless_provider_e2e.py b/tests/openfeature/test_agentless_provider_e2e.py new file mode 100644 index 00000000000..d3ee95e0886 --- /dev/null +++ b/tests/openfeature/test_agentless_provider_e2e.py @@ -0,0 +1,178 @@ +""" +End-to-end wiring tests for agentless Feature Flagging delivery: the JSON:API +CDN response -> parse -> apply -> native config -> OpenFeature evaluation path, +and the provider lifecycle that starts/stops the agentless poller. +""" + +import json + +from openfeature.evaluation_context import EvaluationContext +import pytest + +import ddtrace.internal.openfeature._agentless_source as source_mod +from ddtrace.internal.openfeature._config import _get_ffe_config +from ddtrace.internal.openfeature._config import _set_ffe_config +from ddtrace.internal.openfeature._native import process_ffe_configuration +from ddtrace.internal.openfeature._provider import _apply_agentless_configuration +from ddtrace.internal.openfeature._source_selection import create_agentless_source +from ddtrace.internal.settings.openfeature import config as ffe_config +from ddtrace.openfeature import DataDogProvider +from tests.openfeature.config_helpers import create_boolean_flag +from tests.openfeature.config_helpers import create_config +from tests.utils import override_global_config + + +def _jsonapi_response(*flags): + """Wrap a UFC config as the JSON:API envelope the agentless CDN returns.""" + return json.dumps( + {"data": {"id": "1", "type": "universal-flag-configuration", "attributes": create_config(*flags)}} + ).encode("utf-8") + + +class _FakeResponse: + def __init__(self, status, body=b"", headers=None): + self.status = status + self._body = body + self._headers = {k.lower(): v for k, v in (headers or {}).items()} + + def read(self): + return self._body + + def getheader(self, name, default=None): + return self._headers.get(name.lower(), default) + + +class _FakeConn: + def __init__(self, response): + self._response = response + + def request(self, *args, **kwargs): + pass + + def getresponse(self): + return self._response + + def close(self): + pass + + +@pytest.fixture(autouse=True) +def clear_config(): + _set_ffe_config(None) + yield + _set_ffe_config(None) + + +@pytest.fixture +def mock_cdn(monkeypatch): + def install(body): + monkeypatch.setattr( + source_mod, + "get_connection", + lambda url, timeout=None: _FakeConn(_FakeResponse(200, body, {"ETag": '"v1"'})), + ) + + return install + + +def test_agentless_delivery_evaluates_flag(mock_cdn): + """One agentless poll delivers a flag that the provider then evaluates.""" + mock_cdn(_jsonapi_response(create_boolean_flag("my-flag", enabled=True, default_value=True))) + + with override_global_config({"_dd_api_key": "secret", "_dd_site": "datadoghq.com"}): + provider = DataDogProvider() + + # The provider builds this same source in its lifecycle; drive one poll + # synchronously so the assertion is deterministic (no background thread). + source = create_agentless_source(ffe_config, process_ffe_configuration) + assert source is not None + source._retry_delay = lambda attempt: 0.0 + source.periodic() + + result = provider.resolve_boolean_details("my-flag", False) + + assert result.value is True + assert result.variant == "true" + + +def test_provider_lifecycle_starts_and_stops_source(mock_cdn): + """initialize() starts the agentless poller; shutdown() stops it.""" + mock_cdn(_jsonapi_response(create_boolean_flag("my-flag", enabled=True, default_value=True))) + + with override_global_config({"_dd_api_key": "secret", "_dd_site": "datadoghq.com"}): + provider = DataDogProvider() + try: + provider.initialize(EvaluationContext()) + assert provider._configuration_source is not None + + # The poller runs on a background thread and polls immediately; wait + # for the config to be applied. + assert provider._config_received.wait(timeout=5.0) + + result = provider.resolve_boolean_details("my-flag", False) + assert result.value is True + finally: + provider.shutdown() + + assert provider._configuration_source is None + + +# Attributes that pass JSON:API validation (``createdAt`` is a string) but that the +# native evaluator refuses because the timestamp is unparsable. Note the evaluator +# tolerates malformed individual flags, so an invalid timestamp is the realistic way +# a delivered payload gets rejected. +_REJECTED_ATTRIBUTES = { + "format": "SERVER", + "createdAt": "not-a-timestamp", + "environment": {"name": "production"}, + "flags": {}, +} + + +def test_evaluator_rejection_is_reported_as_failure(): + """A payload the native evaluator refuses must not report success. + + ``process_ffe_configuration`` returns False instead of raising, so the + agentless apply wrapper turns that into an error; otherwise the source would + advance its ETag past a configuration it never loaded. + """ + assert process_ffe_configuration(_REJECTED_ATTRIBUTES) is False + with pytest.raises(ValueError): + _apply_agentless_configuration(_REJECTED_ATTRIBUTES) + + +def test_etag_not_advanced_when_evaluator_rejects_payload(monkeypatch): + """Regression: a rejected payload must leave the ETag (and config) untouched. + + Otherwise the next poll sends If-None-Match, receives 304, and the stale + configuration is kept indefinitely. + """ + body = json.dumps( + {"data": {"id": "1", "type": "universal-flag-configuration", "attributes": _REJECTED_ATTRIBUTES}} + ).encode("utf-8") + monkeypatch.setattr( + source_mod, + "get_connection", + lambda url, timeout=None: _FakeConn(_FakeResponse(200, body, {"ETag": '"rejected"'})), + ) + + src = source_mod.AgentlessConfigurationSource( + endpoint="https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server", + apply_configuration=_apply_agentless_configuration, + api_key="secret", + ) + monkeypatch.setattr(src, "_retry_delay", lambda attempt: 0.0) + + src.periodic() + + assert src._etag is None # not advanced + assert _get_ffe_config() is None # nothing applied + + +def test_disabled_provider_starts_no_source(): + """The kill switch disables the provider and starts no agentless poller.""" + with override_global_config({"feature_flags_enabled": False, "_dd_api_key": "secret"}): + provider = DataDogProvider() + provider.initialize(EvaluationContext()) + + assert provider._configuration_source is None diff --git a/tests/openfeature/test_agentless_source.py b/tests/openfeature/test_agentless_source.py new file mode 100644 index 00000000000..907dde7e5d6 --- /dev/null +++ b/tests/openfeature/test_agentless_source.py @@ -0,0 +1,185 @@ +import gzip +import json + +import pytest + +from ddtrace.internal.openfeature._agentless import DEFAULT_AGENTLESS_PATH +from ddtrace.internal.openfeature._agentless import build_agentless_endpoint +from ddtrace.internal.openfeature._agentless import decode_response_body +from ddtrace.internal.openfeature._agentless import parse_ufc_configuration + + +# --------------------------------------------------------------------------- +# Endpoint derivation +# --------------------------------------------------------------------------- + + +def test_endpoint_managed_default_site(): + assert build_agentless_endpoint("datadoghq.com") == ( + "https://ufc-server.ff-cdn.datadoghq.com" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_site_is_lowercased(): + assert build_agentless_endpoint("DataDogHQ.com") == ( + "https://ufc-server.ff-cdn.datadoghq.com" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_managed_staging_site(): + assert build_agentless_endpoint("datad0g.com") == ("https://ufc-server.ff-cdn.datad0g.com" + DEFAULT_AGENTLESS_PATH) + + +def test_endpoint_managed_govcloud_site(): + assert build_agentless_endpoint("ddog-gov.com") == ( + "https://ufc-server.ff-cdn.ddog-gov.com" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_dd_env_added_when_set(): + url = build_agentless_endpoint("datadoghq.com", env="prod") + assert url == "https://ufc-server.ff-cdn.datadoghq.com" + DEFAULT_AGENTLESS_PATH + "?dd_env=prod" + + +def test_endpoint_dd_env_omitted_when_unset(): + assert "dd_env" not in build_agentless_endpoint("datadoghq.com", env=None) + assert "dd_env" not in build_agentless_endpoint("datadoghq.com", env="") + + +def test_endpoint_dd_env_is_url_encoded(): + url = build_agentless_endpoint("datadoghq.com", env="my env/1") + assert "dd_env=my+env%2F1" in url + + +def test_endpoint_custom_origin_receives_standard_path(): + assert build_agentless_endpoint("datadoghq.com", base_url="https://flags.dev.internal:8080") == ( + "https://flags.dev.internal:8080" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_custom_root_path_receives_standard_path(): + assert build_agentless_endpoint("datadoghq.com", base_url="https://flags.dev.internal/") == ( + "https://flags.dev.internal" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_custom_non_root_path_used_verbatim(): + assert ( + build_agentless_endpoint("datadoghq.com", base_url="https://example.com/custom/ufc?tenant=one") + == "https://example.com/custom/ufc?tenant=one" + ) + + +def test_endpoint_custom_http_allowed_any_host(): + # #9481 removed the loopback-only guard: an explicit custom endpoint is + # operator-owned trust and may be cleartext on any host. + assert build_agentless_endpoint("datadoghq.com", base_url="http://host.docker.internal:8126") == ( + "http://host.docker.internal:8126" + DEFAULT_AGENTLESS_PATH + ) + + +def test_endpoint_custom_base_url_is_trimmed(): + assert build_agentless_endpoint("datadoghq.com", base_url=" https://x.test/ufc ") == "https://x.test/ufc" + + +def test_endpoint_rejects_non_http_scheme(): + with pytest.raises(ValueError): + build_agentless_endpoint("datadoghq.com", base_url="ftp://flags.dev.internal") + + +def test_endpoint_rejects_malformed_url_without_leaking_value(): + sentinel = "sensitive-value" + with pytest.raises(ValueError) as excinfo: + build_agentless_endpoint("datadoghq.com", base_url="https://%s value" % sentinel) + assert sentinel not in str(excinfo.value) + + +# --------------------------------------------------------------------------- +# gzip decoding +# --------------------------------------------------------------------------- + + +def test_decode_gzip_body(): + raw = b'{"hello": "world"}' + assert decode_response_body(gzip.compress(raw), "gzip") == raw + + +def test_decode_gzip_case_insensitive(): + raw = b"payload" + assert decode_response_body(gzip.compress(raw), "GZIP") == raw + + +def test_decode_passthrough_when_not_gzip(): + raw = b"plain" + assert decode_response_body(raw, None) == raw + assert decode_response_body(raw, "identity") == raw + + +def test_decode_raises_on_bad_gzip(): + with pytest.raises((OSError, EOFError)): + decode_response_body(b"not gzip", "gzip") + + +# --------------------------------------------------------------------------- +# JSON:API validation +# --------------------------------------------------------------------------- + + +def _valid_envelope(**attr_overrides): + attributes = { + "format": "SERVER", + "createdAt": "2024-01-01T00:00:00Z", + "environment": {"name": "production"}, + "flags": {"my-flag": {"enabled": True}}, + } + attributes.update(attr_overrides) + return {"data": {"id": "1", "type": "universal-flag-configuration", "attributes": attributes}} + + +def test_parse_accepts_valid_and_returns_attributes_only(): + envelope = _valid_envelope() + attributes = parse_ufc_configuration(json.dumps(envelope)) + assert attributes == envelope["data"]["attributes"] + assert "data" not in attributes + + +def test_parse_accepts_bytes_body(): + attributes = parse_ufc_configuration(json.dumps(_valid_envelope()).encode("utf-8")) + assert attributes["environment"]["name"] == "production" + + +def test_parse_accepts_empty_flags_object(): + attributes = parse_ufc_configuration(json.dumps(_valid_envelope(flags={}))) + assert attributes["flags"] == {} + + +@pytest.mark.parametrize( + "body", + [ + "not json at all", + "", + json.dumps([1, 2, 3]), # top-level not an object + json.dumps({"data": None}), + json.dumps({"data": {"type": "something-else", "attributes": {}}}), + json.dumps({"data": {"type": "universal-flag-configuration"}}), # missing attributes + ], +) +def test_parse_rejects_bad_envelope(body): + with pytest.raises(ValueError): + parse_ufc_configuration(body) + + +@pytest.mark.parametrize( + "attr_overrides", + [ + {"format": 123}, # non-string format + {"createdAt": None}, # non-string createdAt + {"environment": {}}, # missing environment.name + {"environment": "production"}, # environment not an object + {"flags": []}, # flags is an array, not an object + {"flags": None}, # flags missing + ], +) +def test_parse_rejects_bad_attributes(attr_overrides): + with pytest.raises(ValueError): + parse_ufc_configuration(json.dumps(_valid_envelope(**attr_overrides))) diff --git a/tests/openfeature/test_flag_eval_metrics.py b/tests/openfeature/test_flag_eval_metrics.py index 9e63f7ce480..1e93a3facd9 100644 --- a/tests/openfeature/test_flag_eval_metrics.py +++ b/tests/openfeature/test_flag_eval_metrics.py @@ -368,8 +368,8 @@ def test_get_provider_hooks_returns_flag_eval_metrics_hook(self, provider): 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 should not have hooks when disabled by the stable kill switch.""" + with override_global_config({"feature_flags_enabled": False}): provider = DataDogProvider() assert provider._flag_eval_metrics_hook is None diff --git a/tests/openfeature/test_provider_env_var.py b/tests/openfeature/test_provider_env_var.py index 32c802120cd..c2324b08f06 100644 --- a/tests/openfeature/test_provider_env_var.py +++ b/tests/openfeature/test_provider_env_var.py @@ -54,11 +54,11 @@ def test_provider_enabled_with_true_value(self): class TestProviderConfigDisabled: - """Test experimental_flagging_provider_enabled=False or unset behavior.""" + """Test the stable kill switch (DD_FEATURE_FLAGS_ENABLED=false) disabling the provider.""" def test_provider_disabled_returns_default(self): """Provider should return default values when disabled.""" - with override_global_config({"experimental_flagging_provider_enabled": False}): + with override_global_config({"feature_flags_enabled": False}): provider = DataDogProvider() config = create_config(create_boolean_flag("test-flag", enabled=True, default_value=True)) @@ -70,19 +70,19 @@ def test_provider_disabled_returns_default(self): assert result.reason == Reason.DISABLED assert result.variant is None - def test_provider_disabled_by_default(self): - """Provider should be disabled by default.""" - # Don't override config, use defaults - provider = DataDogProvider() + def test_provider_disabled_by_kill_switch(self): + """The stable kill switch disables the provider even with the default agentless source.""" + with override_global_config({"feature_flags_enabled": False}): + provider = DataDogProvider() - result = provider.resolve_string_details("test-flag", "default-value") + result = provider.resolve_string_details("test-flag", "default-value") - assert result.value == "default-value" - assert result.reason == Reason.DISABLED + assert result.value == "default-value" + assert result.reason == Reason.DISABLED def test_provider_disabled_all_types(self): """Provider should return defaults for all flag types when disabled.""" - with override_global_config({"experimental_flagging_provider_enabled": False}): + with override_global_config({"feature_flags_enabled": False}): provider = DataDogProvider() # Boolean @@ -112,7 +112,7 @@ def test_provider_disabled_all_types(self): def test_provider_disabled_skips_initialization(self): """Provider should skip initialization when disabled.""" - with override_global_config({"experimental_flagging_provider_enabled": False}): + with override_global_config({"feature_flags_enabled": False}): provider = DataDogProvider() context = EvaluationContext(targeting_key="user-123") @@ -125,22 +125,22 @@ def test_provider_disabled_skips_initialization(self): def test_provider_disabled_skips_shutdown(self): """Provider should skip shutdown when disabled.""" - with override_global_config({"experimental_flagging_provider_enabled": False}): + with override_global_config({"feature_flags_enabled": False}): provider = DataDogProvider() # Should not raise, just skip shutdown provider.shutdown() def test_provider_disabled_logs_warning(self): - """Provider should log an error when disabled.""" + """Provider should log a warning when disabled.""" from unittest.mock import patch - with override_global_config({"experimental_flagging_provider_enabled": False}): - # Mock the logger to verify error is logged + with override_global_config({"feature_flags_enabled": False}): + # Mock the logger to verify the warning is logged with patch("ddtrace.internal.openfeature._provider.logger") as mock_logger: _ = DataDogProvider() mock_logger.warning.assert_called_once() call_args = mock_logger.warning.call_args - assert "experimental flagging provider is not enabled" in call_args[0][0] - assert "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED" in call_args[0][0] + assert "Feature Flagging provider is disabled" in call_args[0][0] + assert "DD_FEATURE_FLAGS_ENABLED" in call_args[0][0] diff --git a/tests/openfeature/test_provider_status.py b/tests/openfeature/test_provider_status.py index d61f6c9f494..c78e69e5582 100644 --- a/tests/openfeature/test_provider_status.py +++ b/tests/openfeature/test_provider_status.py @@ -5,7 +5,7 @@ - NOT_READY by default - READY when first Remote Config payload is received - Event emission on status change -- Non-blocking initialization while config arrives asynchronously +- Bounded initialization wait, ERROR on timeout, and READY after late delivery """ import threading @@ -13,6 +13,7 @@ from openfeature import api from openfeature.evaluation_context import EvaluationContext +from openfeature.exception import ProviderNotReadyError from openfeature.provider import ProviderStatus import pytest @@ -112,7 +113,7 @@ def on_provider_ready(event_details): try: with override_global_config({"experimental_flagging_provider_enabled": True}): provider = DataDogProvider() - _set_provider(provider) + api.set_provider(provider) # Clear events from initialization ready_events.clear() @@ -140,28 +141,35 @@ def on_provider_ready(event_details): 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 test_sdk_error_event_emitted_when_initialization_times_out(self): + """The SDK reports PROVIDER_ERROR when no configuration arrives in time.""" + error_events = [] + error_event = threading.Event() - def on_provider_ready(event_details): - ready_events.append(event_details) - ready_event.set() + def on_provider_error(event_details): + error_events.append(event_details) + error_event.set() - api.add_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) + api.add_handler(ProviderEvent.PROVIDER_ERROR, on_provider_error) try: with override_global_config({"experimental_flagging_provider_enabled": True}): - provider = DataDogProvider() - _set_provider(provider) + provider = DataDogProvider(initialization_timeout=0.01) + api.set_provider(provider) - assert ready_event.wait(timeout=1.0) - assert len(ready_events) >= 1 + assert error_event.wait(timeout=1.0) + assert len(error_events) == 1 + assert api.get_client().get_provider_status() == ProviderStatus.ERROR assert provider._status == ProviderStatus.NOT_READY assert not provider._config_received.is_set() + + config = create_config(create_boolean_flag("test-flag", enabled=True)) + process_ffe_configuration(config) + + assert api.get_client().get_provider_status() == ProviderStatus.READY + assert provider._status == ProviderStatus.READY finally: - api.remove_handler(ProviderEvent.PROVIDER_READY, on_provider_ready) + api.remove_handler(ProviderEvent.PROVIDER_ERROR, on_provider_error) api.clear_providers() def test_provider_status_after_shutdown(self): @@ -240,7 +248,7 @@ def on_provider_ready(event_details): @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.""" + """Configuration can arrive after attach but before asynchronous initialization.""" ready_events = [] def on_emit(provider, event, details): @@ -254,30 +262,105 @@ def on_emit(provider, event, details): process_ffe_configuration(config) assert provider._status == ProviderStatus.READY - assert ProviderEvent.PROVIDER_READY in ready_events + assert ProviderEvent.PROVIDER_READY not in ready_events class TestProviderInitializationAsync: - """Test that initialize() returns immediately and READY arrives asynchronously.""" + """Test the bounded initialization wait and the READY recovery that can follow it.""" - def test_initialize_returns_immediately_without_config(self): - """initialize() should return immediately even if no config is available yet.""" + def test_initialize_raises_when_the_timeout_expires(self): + """A timeout raises the canonical OpenFeature provider-not-ready error.""" with override_global_config({"experimental_flagging_provider_enabled": True}): - provider = DataDogProvider() + provider = DataDogProvider(initialization_timeout=0.2) try: start = time.monotonic() - # initialize() is called inside set_provider; it must not block - provider.initialize(EvaluationContext()) + with pytest.raises(ProviderNotReadyError): + provider.initialize(EvaluationContext()) elapsed = time.monotonic() - start - # 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() + # Waited for the configured timeout, then gave up rather than hanging. + assert 0.2 <= elapsed < 2.0, f"initialize() waited {elapsed:.2f}s, expected about 0.2s" + # READY still arrives later, via on_configuration_received(). assert provider._status == ProviderStatus.NOT_READY finally: api.clear_providers() + @pytest.mark.skipif(not hasattr(api, "set_provider_and_wait"), reason="Blocking registration requires SDK 0.10+") + def test_set_provider_and_wait_propagates_the_initialization_error(self): + """Blocking registration propagates ProviderNotReadyError to the caller.""" + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider(initialization_timeout=0) + + try: + with pytest.raises(ProviderNotReadyError): + api.set_provider_and_wait(provider) + + assert api.get_client().get_provider_status() == ProviderStatus.ERROR + finally: + api.clear_providers() + + @pytest.mark.skipif( + not hasattr(api, "set_provider_and_wait"), reason="Non-blocking registration requires SDK 0.10+" + ) + def test_set_provider_initializes_in_the_background(self): + """Non-blocking registration returns before provider initialization completes.""" + error_event = threading.Event() + + def on_provider_error(event_details): + error_event.set() + + api.add_handler(ProviderEvent.PROVIDER_ERROR, on_provider_error) + + try: + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider(initialization_timeout=0.5) + + start = time.monotonic() + api.set_provider(provider) + elapsed = time.monotonic() - start + + assert elapsed < 0.2, f"set_provider() waited {elapsed:.2f}s, expected non-blocking registration" + assert error_event.wait(timeout=1.0) + assert api.get_client().get_provider_status() == ProviderStatus.ERROR + finally: + api.remove_handler(ProviderEvent.PROVIDER_ERROR, on_provider_error) + api.clear_providers() + + def test_initialize_waits_for_config_and_becomes_ready(self): + """initialize() blocks until config arrives, so READY implies flags are resolvable.""" + with override_global_config({"experimental_flagging_provider_enabled": True}): + provider = DataDogProvider(initialization_timeout=5.0) + + config = create_config(create_boolean_flag("test-flag", enabled=True)) + delivery = threading.Timer(0.1, process_ffe_configuration, args=(config,)) + + try: + delivery.start() + start = time.monotonic() + provider.initialize(EvaluationContext()) + elapsed = time.monotonic() - start + + # Returned on delivery rather than on the timeout. + assert elapsed < 5.0, f"initialize() waited {elapsed:.2f}s, expected to return on delivery" + assert provider._status == ProviderStatus.READY + assert provider._config_received.is_set() + finally: + delivery.cancel() + api.clear_providers() + + def test_initialization_timeout_comes_from_the_environment(self, monkeypatch): + """The timeout is configurable without a constructor argument.""" + monkeypatch.setenv("DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS", "2500") + with override_global_config({"experimental_flagging_provider_enabled": True}): + assert DataDogProvider()._initialization_timeout == 2.5 + + def test_constructor_timeout_overrides_the_environment(self, monkeypatch): + """An explicit argument wins, so embedders are not bound to the environment.""" + monkeypatch.setenv("DD_EXPERIMENTAL_FLAGGING_PROVIDER_INITIALIZATION_TIMEOUT_MS", "2500") + with override_global_config({"experimental_flagging_provider_enabled": True}): + assert DataDogProvider(initialization_timeout=0.5)._initialization_timeout == 0.5 + def test_initialize_fast_path_when_config_exists(self): """initialize() should return immediately if config already exists.""" with override_global_config({"experimental_flagging_provider_enabled": True}): @@ -299,13 +382,14 @@ def test_initialize_fast_path_when_config_exists(self): api.clear_providers() def test_ready_after_config_arrives_async(self): - """Provider transitions to READY when config arrives after initialize() returns.""" + """Provider transitions to READY when config arrives after initialization fails.""" with override_global_config({"experimental_flagging_provider_enabled": True}): provider = DataDogProvider() try: - provider.initialize(EvaluationContext()) - # Still NOT_READY immediately after initialize() + with pytest.raises(ProviderNotReadyError): + provider.initialize(EvaluationContext()) + # Still NOT_READY immediately after initialization fails. assert provider._status == ProviderStatus.NOT_READY # Config arrives later (simulating RC delivery) @@ -324,7 +408,8 @@ def test_late_config_delivery_transitions_to_ready(self): provider = DataDogProvider() try: - provider.initialize(EvaluationContext()) + with pytest.raises(ProviderNotReadyError): + provider.initialize(EvaluationContext()) # Provider is NOT_READY at this point assert provider._status == ProviderStatus.NOT_READY diff --git a/tests/openfeature/test_source_selection.py b/tests/openfeature/test_source_selection.py new file mode 100644 index 00000000000..dd743b2efae --- /dev/null +++ b/tests/openfeature/test_source_selection.py @@ -0,0 +1,132 @@ +import pytest + +from ddtrace.internal.openfeature._source_selection import AGENTLESS +from ddtrace.internal.openfeature._source_selection import DISABLED +from ddtrace.internal.openfeature._source_selection import REMOTE_CONFIG +from ddtrace.internal.openfeature._source_selection import create_agentless_source +from ddtrace.internal.openfeature._source_selection import resolve_configuration_source +from ddtrace.internal.settings.openfeature import OpenFeatureConfig +from tests.utils import override_global_config + + +def _config(**env): + """Build an OpenFeatureConfig where the given env vars are marked as provided.""" + return OpenFeatureConfig(source=env) + + +# --------------------------------------------------------------------------- +# Source resolution matrix (mirrors the system-tests contract) +# --------------------------------------------------------------------------- + + +def test_default_is_agentless(): + assert resolve_configuration_source(_config()) == AGENTLESS + + +def test_explicit_agentless(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="agentless")) == AGENTLESS + + +def test_explicit_remote_config(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="remote_config")) == REMOTE_CONFIG + + +def test_source_is_case_and_whitespace_insensitive(): + assert ( + resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=" Remote_Config ")) + == REMOTE_CONFIG + ) + + +def test_invalid_source_fails_closed(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="invalid")) == DISABLED + + +def test_reserved_offline_source_fails_closed(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="offline")) == DISABLED + + +def test_blank_source_is_treated_as_absent(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=" ")) == AGENTLESS + + +def test_kill_switch_disables(): + assert resolve_configuration_source(_config(DD_FEATURE_FLAGS_ENABLED="false")) == DISABLED + + +def test_kill_switch_overrides_legacy(): + cfg = _config(DD_FEATURE_FLAGS_ENABLED="false", DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED="true") + assert resolve_configuration_source(cfg) == DISABLED + + +def test_grandfather_legacy_true_selects_remote_config(): + cfg = _config(DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED="true") + assert resolve_configuration_source(cfg) == REMOTE_CONFIG + + +def test_grandfather_legacy_false_disables(): + cfg = _config(DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED="false") + assert resolve_configuration_source(cfg) == DISABLED + + +def test_explicit_agentless_wins_over_legacy_true(): + cfg = _config( + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="agentless", + DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED="true", + ) + assert resolve_configuration_source(cfg) == AGENTLESS + + +def test_explicit_remote_config_wins_over_legacy_false(): + cfg = _config( + DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="remote_config", + DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED="false", + ) + assert resolve_configuration_source(cfg) == REMOTE_CONFIG + + +# --------------------------------------------------------------------------- +# Agentless factory +# --------------------------------------------------------------------------- + + +def test_create_returns_none_when_not_agentless(): + cfg = _config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE="remote_config") + assert create_agentless_source(cfg, lambda _: None) is None + + +def test_create_default_endpoint_requires_api_key(): + cfg = _config() + with override_global_config({"_dd_api_key": None}): + assert create_agentless_source(cfg, lambda _: None) is None + + +def test_create_default_endpoint_with_api_key(): + cfg = _config() + with override_global_config({"_dd_api_key": "secret", "_dd_site": "datadoghq.com"}): + src = create_agentless_source(cfg, lambda _: None) + assert src is not None + assert src._api_key == "secret" + assert src._conn_url == "https://ufc-server.ff-cdn.datadoghq.com/" + + +def test_create_custom_endpoint_omits_api_key(): + cfg = _config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL="http://host.docker.internal:8126") + with override_global_config({"_dd_api_key": "secret"}): + src = create_agentless_source(cfg, lambda _: None) + assert src is not None + assert src._api_key is None # custom endpoint: key omitted even though one is set + + +def test_create_custom_endpoint_starts_without_api_key(): + cfg = _config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL="http://host.docker.internal:8126") + with override_global_config({"_dd_api_key": None}): + src = create_agentless_source(cfg, lambda _: None) + assert src is not None # missing key does not block a custom endpoint + + +@pytest.mark.parametrize("bad_url", ["ftp://flags.example.test", "https://flags.example.test path"]) +def test_create_invalid_endpoint_returns_none(bad_url): + cfg = _config(DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL=bad_url) + with override_global_config({"_dd_api_key": "secret"}): + assert create_agentless_source(cfg, lambda _: None) is None diff --git a/tests/suitespec.yml b/tests/suitespec.yml index bd519e0d005..a26b32eab60 100644 --- a/tests/suitespec.yml +++ b/tests/suitespec.yml @@ -90,6 +90,7 @@ components: - ddtrace/ext/schema.py openfeature: - ddtrace/openfeature/* + - ddtrace/internal/openfeature/* - tests/openfeature/* git: - ddtrace/ext/git.py