feat(openfeature): emit server-side EVP flagevaluation

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -24,6 +24,11 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake) {
+    return createBackendApi(intake, null, true);
+  }
+
+  public @Nullable BackendApi createBackendApi(
+      Intake intake, @Nullable String preferredEvpProxyEndpoint, boolean responseCompression) {
     HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true);
 
     if (intake.isAgentlessEnabled(config)) {
@@ -49,6 +54,15 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
     if (featuresDiscovery.supportsEvpProxy()) {
       String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
       String evpProxyEndpoint = featuresDiscovery.getEvpProxyEndpoint();
+      if (preferredEvpProxyEndpoint != null
+          && featuresDiscovery.supportsEvpProxyEndpoint(preferredEvpProxyEndpoint)) {
+        evpProxyEndpoint = preferredEvpProxyEndpoint;
+      }
+      log.debug(
+          "Creating EVP proxy client for {} using endpoint {} with responseCompression={}",
+          intake,
+          evpProxyEndpoint,
+          responseCompression);
       HttpUrl evpProxyUrl = sharedCommunicationObjects.agentUrl.resolve(evpProxyEndpoint);
       String subdomain = intake.getUrlPrefix();
       return new EvpProxyApi(
@@ -57,7 +71,7 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
           subdomain,
           retryPolicyFactory,
           sharedCommunicationObjects.agentHttpClient,
-          true);
+          responseCompression);
     }
 
     log.warn(

communication/src/main/java/datadog/communication/EvpProxy.java

@@ -0,0 +1,12 @@
+package datadog.communication;
+
+/** Shared EVP proxy constants. */
+public final class EvpProxy {
+
+  public static final String SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain";
+
+  /** EVP uncompressed request-body limit in bytes. */
+  public static final int PAYLOAD_SIZE_LIMIT_BYTES = 5 * 1024 * 1024;
+
+  private EvpProxy() {}
+}

communication/src/main/java/datadog/communication/EvpProxyApi.java

@@ -20,7 +20,6 @@ public class EvpProxyApi implements BackendApi {
   private static final Logger log = LoggerFactory.getLogger(EvpProxyApi.class);
 
   private static final String API_VERSION = "v2";
-  private static final String X_DATADOG_EVP_SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain";
   private static final String X_DATADOG_TRACE_ID_HEADER = "x-datadog-trace-id";
   private static final String X_DATADOG_PARENT_ID_HEADER = "x-datadog-parent-id";
   private static final String ACCEPT_ENCODING_HEADER = "Accept-Encoding";
@@ -62,7 +61,7 @@ public <T> T post(
     Request.Builder requestBuilder =
         new Request.Builder()
             .url(url)
-            .addHeader(X_DATADOG_EVP_SUBDOMAIN_HEADER, subdomain)
+            .addHeader(EvpProxy.SUBDOMAIN_HEADER, subdomain)
             .addHeader(X_DATADOG_TRACE_ID_HEADER, traceId)
             .addHeader(X_DATADOG_PARENT_ID_HEADER, traceId);
 
@@ -79,6 +78,11 @@ public <T> T post(
     }
 
     final Request request = requestBuilder.post(requestBody).build();
+    log.debug(
+        "Posting EVP request to {} with responseCompression={} requestCompression={}",
+        url,
+        responseCompression,
+        requestCompression);
 
     try (okhttp3.Response response =
         OkHttpUtils.sendWithRetries(httpClient, retryPolicyFactory, request)) {

communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java

@@ -98,6 +98,7 @@ private static class State {
     String debuggerSnapshotEndpoint;
     String debuggerDiagnosticsEndpoint;
     String evpProxyEndpoint;
+    Set<String> evpProxyEndpoints = emptySet();
     String version;
     String telemetryProxyEndpoint;
     Set<String> peerTags = emptySet();
@@ -288,6 +289,7 @@ private boolean processInfoResponse(State newState, String response) {
           break;
         }
       }
+      newState.evpProxyEndpoints = unmodifiableSet(endpoints);
 
       for (String endpoint : telemetryProxyEndpoints) {
         if (containsEndpoint(endpoints, endpoint)) {
@@ -426,6 +428,10 @@ public String getEvpProxyEndpoint() {
     return discoveryState.evpProxyEndpoint;
   }
 
+  public boolean supportsEvpProxyEndpoint(String endpoint) {
+    return containsEndpoint(discoveryState.evpProxyEndpoints, endpoint);
+  }
+
   public HttpUrl buildUrl(String endpoint) {
     return agentBaseUrl.resolve(endpoint);
   }

dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java

@@ -3,4 +3,12 @@
 public class FeatureFlaggingConfig {
 
   public static final String FLAGGING_PROVIDER_ENABLED = "experimental.flagging.provider.enabled";
+
+  /**
+   * Killswitch for the EVP {@code flagevaluation} emission path. Default: enabled. Disabling it
+   * turns off EVP flag-evaluation counts while leaving the OTel {@code feature_flag.evaluations}
+   * metric path untouched. Maps to {@code DD_FLAGGING_EVALUATION_COUNTS_ENABLED}.
+   */
+  public static final String FLAGGING_EVALUATION_COUNTS_ENABLED =
+      "flagging.evaluation.counts.enabled";
 }

dd-trace-core/src/main/java/datadog/trace/common/writer/ddintake/DDEvpProxyApi.java

@@ -3,6 +3,7 @@
 import static datadog.trace.common.writer.DDIntakeWriter.DEFAULT_INTAKE_TIMEOUT;
 import static datadog.trace.common.writer.DDIntakeWriter.DEFAULT_INTAKE_VERSION;
 
+import datadog.communication.EvpProxy;
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.communication.http.OkHttpUtils;
 import datadog.trace.api.civisibility.InstrumentationBridge;
@@ -26,7 +27,6 @@ public class DDEvpProxyApi extends RemoteApi {
 
   private static final Logger log = LoggerFactory.getLogger(DDEvpProxyApi.class);
 
-  private static final String DD_EVP_SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain";
   private static final String CONTENT_ENCODING_HEADER = "Content-Encoding";
   private static final String GZIP_CONTENT_TYPE = "gzip";
 
@@ -131,7 +131,7 @@ public Response sendSerializedTraces(Payload payload) {
     Request.Builder builder =
         new Request.Builder()
             .url(proxiedApiUrl)
-            .addHeader(DD_EVP_SUBDOMAIN_HEADER, subdomain)
+            .addHeader(EvpProxy.SUBDOMAIN_HEADER, subdomain)
             .tag(OkHttpUtils.CustomListener.class, telemetryListener);
 
     if (isCompressionEnabled()) {

dd-trace-core/src/main/java/datadog/trace/llmobs/writer/ddintake/LLMObsSpanMapper.java

@@ -2,6 +2,7 @@
 
 import static datadog.communication.http.OkHttpUtils.gzippedMsgpackRequestBodyOf;
 
+import datadog.communication.EvpProxy;
 import datadog.communication.serialization.GrowableBuffer;
 import datadog.communication.serialization.Writable;
 import datadog.communication.serialization.msgpack.MsgPackWriter;
@@ -99,7 +100,7 @@ public class LLMObsSpanMapper implements RemoteMapper {
   private int spansWritten;
 
   public LLMObsSpanMapper() {
-    this(5 << 20);
+    this(EvpProxy.PAYLOAD_SIZE_LIMIT_BYTES);
   }
 
   private LLMObsSpanMapper(int size) {

dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddintake/DDEvpProxyApiTest.groovy

@@ -2,6 +2,7 @@ package datadog.trace.common.writer.ddintake
 
 import com.fasterxml.jackson.core.type.TypeReference
 import com.fasterxml.jackson.databind.ObjectMapper
+import datadog.communication.EvpProxy
 import datadog.communication.serialization.ByteBufferConsumer
 import datadog.communication.serialization.FlushingBuffer
 import datadog.communication.serialization.msgpack.MsgPackWriter
@@ -64,7 +65,7 @@ class DDEvpProxyApiTest extends DDCoreSpecification {
     clientResponse.status().present
     clientResponse.status().asInt == 200
     agentEvpProxy.getLastRequest().path == path
-    agentEvpProxy.getLastRequest().getHeader(DDEvpProxyApi.DD_EVP_SUBDOMAIN_HEADER) == intakeSubdomain
+    agentEvpProxy.getLastRequest().getHeader(EvpProxy.SUBDOMAIN_HEADER) == intakeSubdomain
 
     cleanup:
     agentEvpProxy.close()
@@ -100,7 +101,7 @@ class DDEvpProxyApiTest extends DDCoreSpecification {
     clientResponse.status().present
     clientResponse.status().asInt == 200
     agentEvpProxy.getLastRequest().path == path
-    agentEvpProxy.getLastRequest().getHeader(DDEvpProxyApi.DD_EVP_SUBDOMAIN_HEADER) == intakeSubdomain
+    agentEvpProxy.getLastRequest().getHeader(EvpProxy.SUBDOMAIN_HEADER) == intakeSubdomain
 
     cleanup:
     agentEvpProxy.close()

internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java

@@ -29,6 +29,14 @@ private CoreMetricCollector() {
     this.metricsQueue = new ArrayBlockingQueue<>(RAW_QUEUE_SIZE);
   }
 
+  public void count(String metricName, long value, String tag) {
+    if (value <= 0) {
+      return;
+    }
+    this.metricsQueue.offer(
+        new CoreMetric(METRIC_NAMESPACE, true, metricName, "count", value, tag));
+  }
+
   @Override
   public void prepareMetrics() {
     // Collect span metrics

internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java

@@ -66,7 +66,9 @@ public enum AgentThread {
 
     LLMOBS_EVALS_PROCESSOR("dd-llmobs-evals-processor"),
 
-    FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor");
+    FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor"),
+
+    FEATURE_FLAG_EVALUATION_PROCESSOR("dd-ffe-evaluation-processor");
 
     public final String threadName;
 

internal-api/src/test/groovy/datadog/trace/api/telemetry/CoreMetricCollectorTest.groovy

@@ -53,4 +53,24 @@ class CoreMetricCollectorTest extends DDSpecification {
     collector.prepareMetrics()
     collector.drain().size() == limit
   }
+
+  def "direct count core metric"() {
+    setup:
+    def collector = CoreMetricCollector.getInstance()
+    collector.drain()
+
+    when:
+    collector.count('flagevaluation.rows.dropped', 3, 'reason:queue_overflow')
+    def metrics = collector.drain()
+
+    then:
+    metrics.size() == 1
+
+    def metric = metrics[0]
+    metric.type == 'count'
+    metric.value == 3
+    metric.namespace == 'tracers'
+    metric.metricName == 'flagevaluation.rows.dropped'
+    metric.tags == ['reason:queue_overflow']
+  }
 }

metadata/supported-configurations.json

@@ -1489,6 +1489,14 @@
         "aliases": []
       }
     ],
+    "DD_FLAGGING_EVALUATION_COUNTS_ENABLED": [
+      {
+        "version": "A",
+        "type": "boolean",
+        "default": "true",
+        "aliases": []
+      }
+    ],
     "DD_FORCE_CLEAR_TEXT_HTTP_FOR_INTAKE_CLIENT": [
       {
         "version": "A",

products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java

@@ -2,6 +2,8 @@
 
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.trace.api.Config;
+import datadog.trace.api.config.FeatureFlaggingConfig;
+import datadog.trace.api.featureflag.flagevaluation.FlagEvaluationWriter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -11,6 +13,7 @@ public class FeatureFlaggingSystem {
 
   private static volatile RemoteConfigService CONFIG_SERVICE;
   private static volatile ExposureWriter EXPOSURE_WRITER;
+  private static volatile FlagEvaluationWriter FLAG_EVAL_WRITER;
 
   private FeatureFlaggingSystem() {}
 
@@ -27,10 +30,31 @@ public static void start(final SharedCommunicationObjects sco) {
     EXPOSURE_WRITER = new ExposureWriterImpl(sco, config);
     EXPOSURE_WRITER.init();
 
+    // EVP flagevaluation writer — gated by the killswitch
+    // DD_FLAGGING_EVALUATION_COUNTS_ENABLED (default: on), read through the tracer config system.
+    final boolean evalCountsEnabled =
+        config
+            .configProvider()
+            .getBoolean(FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, true);
+    if (evalCountsEnabled) {
+      final FlagEvaluationWriterImpl evalWriter = new FlagEvaluationWriterImpl(sco, config);
+      evalWriter.start(); // registers with FeatureFlaggingGateway
+      FLAG_EVAL_WRITER = evalWriter;
+      LOGGER.debug("Flag evaluation EVP writer started");
+    } else {
+      LOGGER.debug(
+          "Flag evaluation EVP writer disabled ({}=false)",
+          FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED);
+    }
+
     LOGGER.debug("Feature Flagging system started");
   }
 
   public static void stop() {
+    if (FLAG_EVAL_WRITER != null) {
+      FLAG_EVAL_WRITER.close(); // also deregisters from gateway
+      FLAG_EVAL_WRITER = null;
+    }
     if (EXPOSURE_WRITER != null) {
       EXPOSURE_WRITER.close();
       EXPOSURE_WRITER = null;

products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java

@@ -387,11 +387,15 @@ private static <T> ProviderEvaluation<T> resolveVariant(
               + e.getMessage());
     }
 
+    // Stamp eval-time at the resolution point so first/last_evaluation reflect evaluation time,
+    // not hook-fire time. Passed to the hook via provider metadata "dd.eval.timestamp_ms".
+    final long evalTimestampMs = System.currentTimeMillis();
     final ImmutableMetadata.ImmutableMetadataBuilder metadataBuilder =
         ImmutableMetadata.builder()
             .addString("flagKey", flag.key)
             .addString("variationType", flag.variationType.name())
-            .addString("allocationKey", allocation.key);
+            .addString("allocationKey", allocation.key)
+            .addLong("dd.eval.timestamp_ms", evalTimestampMs);
     final ProviderEvaluation<T> result =
         ProviderEvaluation.<T>builder()
             .value(mappedValue)