feat(openfeature): add datadog openfeature exposure (#15073)

## Description

This PR adds **Feature Flag Exposure reporting** to the Datadog
OpenFeature Provider, along with centralized configuration management
for OpenFeature settings.

### Key Changes

#### **Feature Flag Exposure Reporting**
- Implements exposure event tracking for feature flag evaluations
- Sends exposure events to Datadog Agent's EVP proxy endpoint
(`/evp_proxy/v2/api/v2/exposures`)
- Tracks flag key, variant, allocation, and user context for each
evaluation
- Includes automatic retry mechanism with fibonacci backoff

#### **Centralized Configuration System**
Created
[ddtrace/settings/openfeature.py](cci:7://file:///home/alberto.vara/projects/dd-python/dd-trace-py/ddtrace/settings/openfeature.py:0:0-0:0)
with three new environment variables:

- **`DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED`** (default: `false`)
  - Controls whether the experimental OpenFeature provider is enabled
  - Must be explicitly enabled to use the provider

- **`DD_FFE_INTAKE_ENABLED`** (default: `true`)
  - Enables/disables exposure event reporting to Datadog
  - Allows disabling telemetry without disabling the provider

- **`DD_FFE_INTAKE_HEARTBEAT_INTERVAL`** (default: `1.0`)
  - Controls the flush interval for exposure events in seconds
  - Configurable for performance tuning


## Additional Notes
Related PR: #15051

Author: avara1986

ddtrace/internal/openfeature/__init__.py

@@ -1,6 +1,6 @@
 """
-Feature Flagging and Experimentation (FFAndE) product module.
+Feature Flagging and Experimentation (FFE) product module.
 
 This module handles Feature Flag configuration rules from Remote Configuration
-and forwards the raw bytes to the native FFAndE processor.
+and forwards the raw bytes to the native FFE processor.
 """

ddtrace/internal/openfeature/_config.py

@@ -1,15 +1,15 @@
-import contextvars
-from typing import Optional
+from typing import Mapping
 
 
-FFE_CONFIG: contextvars.ContextVar[Optional[int]] = contextvars.ContextVar("ffe_config", default=None)
+FFE_CONFIG: Mapping = {}
 
 
 def _get_ffe_config():
     """Retrieve the current IAST context identifier from the ContextVar."""
-    return FFE_CONFIG.get()
+    return FFE_CONFIG
 
 
 def _set_ffe_config(data):
+    global FFE_CONFIG
     """Retrieve the current IAST context identifier from the ContextVar."""
-    return FFE_CONFIG.set(data)
+    FFE_CONFIG = data

ddtrace/internal/openfeature/_exposure.py

@@ -0,0 +1,100 @@
+"""
+Exposure event building for Feature Flag evaluation results.
+"""
+
+import time
+from typing import Any
+from typing import Dict
+from typing import Optional
+
+from openfeature.evaluation_context import EvaluationContext
+
+from ddtrace.internal.logger import get_logger
+from ddtrace.internal.openfeature.writer import ExposureEvent
+
+
+logger = get_logger(__name__)
+
+
+def build_exposure_event(
+    flag_key: str,
+    variant_key: Optional[str],
+    allocation_key: Optional[str],
+    evaluation_context: Optional[EvaluationContext],
+) -> Optional[ExposureEvent]:
+    """
+    Build an exposure event that will be batched and sent with context.
+
+    Individual events are collected and sent in batches with shared context
+    (service, env, version) to the EVP proxy intake endpoint.
+
+    Args:
+        flag_key: The feature flag key
+        variant_key: The variant key returned by the evaluation
+        allocation_key: The allocation key (same as variant_key in basic cases)
+        evaluation_context: The evaluation context with subject information
+    """
+    # Validate required fields
+    if not flag_key:
+        logger.debug("Cannot build exposure event: flag_key is required")
+        return None
+
+    if variant_key is None:
+        variant_key = ""
+
+    # Build subject from evaluation context
+    subject = _build_subject(evaluation_context)
+    if not subject:
+        logger.debug("Cannot build exposure event: valid subject is required")
+        return None
+
+    # Use variant_key as allocation_key if not explicitly provided
+    if allocation_key is None:
+        allocation_key = variant_key
+
+    # Build the exposure event
+    exposure_event: ExposureEvent = {
+        "timestamp": int(time.time() * 1000),  # milliseconds since epoch
+        "allocation": {"key": allocation_key},
+        "flag": {"key": flag_key},
+        "variant": {"key": variant_key},
+        "subject": subject,
+    }
+
+    return exposure_event
+
+
+def _build_subject(evaluation_context: Optional[EvaluationContext]) -> Optional[Dict[str, Any]]:
+    """
+    Build subject object from OpenFeature EvaluationContext.
+
+    The subject must have at minimum an 'id' field.
+
+    Args:
+        evaluation_context: The OpenFeature evaluation context
+
+    Returns:
+        Dictionary with subject information, or None if id cannot be determined
+    """
+    if evaluation_context is None:
+        return None
+
+    # Get targeting_key as the subject id
+    subject_id = evaluation_context.targeting_key
+    if not subject_id:
+        logger.debug("evaluation_context missing targeting_key for subject.id")
+        return None
+
+    subject: Dict[str, Any] = {"id": subject_id}
+
+    # Add optional subject type if available in attributes
+    attributes = evaluation_context.attributes or {}
+    if "subject_type" in attributes:
+        subject["type"] = str(attributes["subject_type"])
+
+    # Add remaining attributes (excluding subject_type which we already handled)
+    remaining_attrs = {k: v for k, v in attributes.items() if k != "subject_type"}
+    if remaining_attrs:
+        subject["attributes"] = remaining_attrs
+
+    return subject

ddtrace/internal/openfeature/_ffe_mock.py

@@ -97,8 +97,7 @@ def mock_get_assignment(
     if not flag or not flag.get("enabled", True):
         return None
 
-    # Handle both enum objects and string values for variation_type
-    variation_type_raw = flag["variation_type"]
+    variation_type_raw = flag["variationType"]
     if isinstance(variation_type_raw, str):
         found_type = VariationType(variation_type_raw)
     else:
@@ -111,15 +110,13 @@ def mock_get_assignment(
             found=found_type,
         )
 
-    # Handle both enum objects and string values for reason
     reason_raw = flag.get("reason", AssignmentReason.STATIC)
     if isinstance(reason_raw, str):
         reason = AssignmentReason(reason_raw)
     else:
         reason = reason_raw
 
-    # Build assignment
-    value = flag["value"]
+    value = list(flag["variations"].values())[0]["value"]
     assignment_value = AssignmentValue(variation_type=found_type, value=value)
     return Assignment(
         value=assignment_value,

ddtrace/internal/openfeature/_native.py

@@ -22,7 +22,7 @@
     # Provide a no-op fallback
     def ffande_process_config(config_bytes: bytes) -> Optional[bool]:
         """Fallback implementation when native module is not available."""
-        log.warning("FFAndE native module not available, ignoring configuration")
+        log.warning("FFE native module not available, ignoring configuration")
         return None
 
 
@@ -47,5 +47,5 @@ def process_ffe_configuration(config_bytes: bytes) -> bool:
             return False
         return result
     except Exception as e:
-        log.error("Error processing FFE configuration: %s", e, exc_info=True)
+        log.debug("Error processing FFE configuration: %s", e, exc_info=True)
         return False

ddtrace/internal/openfeature/_provider.py

@@ -1,8 +1,8 @@
 """
-Feature Flagging and Experimentation (FFAndE) product module.
+Feature Flagging and Experimentation (FFE) product module.
 
 This module handles Feature Flag configuration rules from Remote Configuration
-and forwards the raw bytes to the native FFAndE processor.
+and forwards the raw bytes to the native FFE processor.
 """
 
 import datetime
@@ -16,13 +16,20 @@
 from openfeature.flag_evaluation import Reason
 from openfeature.provider import Metadata
 
+from ddtrace.internal.logger import get_logger
 from ddtrace.internal.openfeature._config import _get_ffe_config
+from ddtrace.internal.openfeature._exposure import build_exposure_event
 from ddtrace.internal.openfeature._ffe_mock import AssignmentReason
 from ddtrace.internal.openfeature._ffe_mock import EvaluationError
 from ddtrace.internal.openfeature._ffe_mock import VariationType
 from ddtrace.internal.openfeature._ffe_mock import mock_get_assignment
 from ddtrace.internal.openfeature._remoteconfiguration import disable_featureflags_rc
 from ddtrace.internal.openfeature._remoteconfiguration import enable_featureflags_rc
+from ddtrace.internal.openfeature.writer import get_exposure_writer
+from ddtrace.internal.openfeature.writer import start_exposure_writer
+from ddtrace.internal.openfeature.writer import stop_exposure_writer
+from ddtrace.internal.service import ServiceStatusError
+from ddtrace.settings.openfeature import config as ffe_config
 
 
 # Handle different import paths between openfeature-sdk versions
@@ -35,20 +42,29 @@
 
 
 T = typing.TypeVar("T", covariant=True)
+logger = get_logger(__name__)
 
 
 class DataDogProvider(AbstractProvider):
     """
     Datadog OpenFeature Provider.
 
     Implements the OpenFeature provider interface for Datadog's
-    Feature Flags and Experimentation (FFAndE) product.
+    Feature Flags and Experimentation (FFE) product.
     """
 
     def __init__(self, *args: typing.Any, **kwargs: typing.Any):
         super().__init__(*args, **kwargs)
         self._metadata = Metadata(name="Datadog")
 
+        # Check if experimental flagging provider is enabled
+        self._enabled = ffe_config.experimental_flagging_provider_enabled
+        if not self._enabled:
+            logger.warning(
+                "openfeature: experimental flagging provider is not enabled, "
+                "please set DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED=true to enable it",
+            )
+
     def get_metadata(self) -> Metadata:
         """Returns provider metadata."""
         return self._metadata
@@ -59,15 +75,32 @@ def initialize(self, evaluation_context: EvaluationContext) -> None:
 
         Called by the OpenFeature SDK when the provider is set.
         """
+        if not self._enabled:
+            return
+
         enable_featureflags_rc()
 
+        try:
+            # Start the exposure writer for reporting
+            start_exposure_writer()
+        except ServiceStatusError:
+            logger.debug("Exposure writer is already running", exc_info=True)
+
     def shutdown(self) -> None:
         """
         Shutdown the provider and disable remote configuration.
 
         Called by the OpenFeature SDK when the provider is being replaced or shutdown.
         """
+        if not self._enabled:
+            return
+
         disable_featureflags_rc()
+        try:
+            # Stop the exposure writer
+            stop_exposure_writer()
+        except ServiceStatusError:
+            logger.debug("Exposure writer has already stopped", exc_info=True)
 
     def resolve_boolean_details(
         self,
@@ -124,6 +157,14 @@ def _resolve_details(
         - Returns default value with DEFAULT reason when flag not found
         - Returns error with error_code and error_message on errors
         """
+        # If provider is not enabled, return default value
+        if not self._enabled:
+            return FlagResolutionDetails(
+                value=default_value,
+                reason=Reason.DISABLED,
+                variant=None,
+            )
+
         try:
             config_raw = _get_ffe_config()
             # Parse JSON config if it's a string
@@ -142,6 +183,12 @@ def _resolve_details(
 
             # Flag not found or disabled - return default
             if result is None:
+                self._report_exposure(
+                    flag_key=flag_key,
+                    variant_key=None,
+                    allocation_key=None,
+                    evaluation_context=evaluation_context,
+                )
                 return FlagResolutionDetails(
                     value=default_value,
                     reason=Reason.DEFAULT,
@@ -156,6 +203,14 @@ def _resolve_details(
             }
             reason = reason_map.get(result.reason, Reason.UNKNOWN)
 
+            # Report exposure event
+            self._report_exposure(
+                flag_key=flag_key,
+                variant_key=result.variation_key,
+                allocation_key=result.variation_key,
+                evaluation_context=evaluation_context,
+            )
+
             # Success - return resolved value
             return FlagResolutionDetails(
                 value=result.value.value,
@@ -187,3 +242,27 @@ def _resolve_details(
                 error_code=ErrorCode.GENERAL,
                 error_message=f"Unexpected error during flag evaluation: {str(e)}",
             )
+
+    def _report_exposure(
+        self,
+        flag_key: str,
+        variant_key: typing.Optional[str],
+        allocation_key: typing.Optional[str],
+        evaluation_context: typing.Optional[EvaluationContext],
+    ) -> None:
+        """
+        Report a feature flag exposure event to the EVP proxy intake.
+        """
+        try:
+            exposure_event = build_exposure_event(
+                flag_key=flag_key,
+                variant_key=variant_key,
+                allocation_key=allocation_key,
+                evaluation_context=evaluation_context,
+            )
+
+            if exposure_event:
+                writer = get_exposure_writer()
+                writer.enqueue(exposure_event)
+        except Exception as e:
+            logger.debug("Failed to report exposure event: %s", e, exc_info=True)

ddtrace/internal/openfeature/_remoteconfiguration.py

@@ -1,8 +1,8 @@
 """
-FFAndE (Feature Flagging and Experimentation) product implementation.
+FFE (Feature Flagging and Experimentation) product implementation.
 
 This product receives feature flag configuration rules from Remote Configuration
-and processes them through the native FFAndE processor.
+and processes them through the native FFE processor.
 """
 import enum
 import json
@@ -24,15 +24,15 @@
 FFE_FLAGS_PRODUCT = "FFE_FLAGS"
 
 
-class FFAndECapabilities(enum.IntFlag):
-    """FFAndE Remote Configuration capabilities."""
+class FFECapabilities(enum.IntFlag):
+    """FFE Remote Configuration capabilities."""
 
     FFE_FLAG_CONFIGURATION_RULES = 1 << 46
 
 
-class FFAndEAdapter(PubSub):
+class FFEAdapter(PubSub):
     """
-    FFAndE Remote Configuration adapter.
+    FFE Remote Configuration adapter.
 
     Receives feature flag configuration rules and forwards raw bytes to native processor.
     """
@@ -78,17 +78,17 @@ def featureflag_rc_callback(payloads: t.Sequence[Payload]) -> None:
             mock_process_ffe_configuration(payload.content)
             log.debug("Processing FFE config ID: %s, size: %d bytes", payload.metadata.id, len(config_bytes))
         except Exception as e:
-            log.error("Error processing FFE config payload: %s", e, exc_info=True)
+            log.debug("Error processing FFE config payload: %s", e, exc_info=True)
 
 
 def enable_featureflags_rc() -> None:
     log.debug("[%s][P: %s] Register ASM Remote Config Callback", os.getpid(), os.getppid())
-    feature_flag_rc = FFAndEAdapter(featureflag_rc_callback)
+    feature_flag_rc = FFEAdapter(featureflag_rc_callback)
     remoteconfig_poller.register(
         FFE_FLAGS_PRODUCT,
         feature_flag_rc,
         restart_on_fork=True,
-        capabilities=[FFAndECapabilities.FFE_FLAG_CONFIGURATION_RULES],
+        capabilities=[FFECapabilities.FFE_FLAG_CONFIGURATION_RULES],
     )