DataDog
diff --git a/‎communication/src/main/java/datadog/communication/BackendApiFactory.java‎
Lines changed: 25 additions & 16 deletions b/‎communication/src/main/java/datadog/communication/BackendApiFactory.java‎
Lines changed: 25 additions & 16 deletions
diff --git a/‎communication/src/main/java/datadog/communication/EvpProxy.java‎
Lines changed: 15 additions & 0 deletions b/‎communication/src/main/java/datadog/communication/EvpProxy.java‎
Lines changed: 15 additions & 0 deletions
diff --git a/‎communication/src/main/java/datadog/communication/EvpProxyApi.java‎
Lines changed: 12 additions & 2 deletions b/‎communication/src/main/java/datadog/communication/EvpProxyApi.java‎
Lines changed: 12 additions & 2 deletions
diff --git a/‎communication/src/test/java/datadog/communication/BackendApiFactoryTest.java‎
Lines changed: 112 additions & 0 deletions b/‎communication/src/test/java/datadog/communication/BackendApiFactoryTest.java‎
Lines changed: 112 additions & 0 deletions
diff --git a/‎dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java‎
Lines changed: 5 additions & 3 deletions b/‎dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java‎
Lines changed: 5 additions & 3 deletions
diff --git a/‎dd-trace-core/src/main/java/datadog/trace/common/writer/ddintake/DDEvpProxyApi.java‎
Lines changed: 2 additions & 2 deletions b/‎dd-trace-core/src/main/java/datadog/trace/common/writer/ddintake/DDEvpProxyApi.java‎
Lines changed: 2 additions & 2 deletions
diff --git a/‎dd-trace-core/src/main/java/datadog/trace/llmobs/writer/ddintake/LLMObsSpanMapper.java‎
Lines changed: 2 additions & 1 deletion b/‎dd-trace-core/src/main/java/datadog/trace/llmobs/writer/ddintake/LLMObsSpanMapper.java‎
Lines changed: 2 additions & 1 deletion
diff --git a/‎dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddintake/DDEvpProxyApiTest.groovy‎
Lines changed: 3 additions & 2 deletions b/‎dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddintake/DDEvpProxyApiTest.groovy‎
Lines changed: 3 additions & 2 deletions
diff --git a/‎internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java‎
Lines changed: 8 additions & 0 deletions b/‎internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java‎
Lines changed: 8 additions & 0 deletions
diff --git a/‎internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java‎
Lines changed: 1 addition & 0 deletions b/‎internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java‎
Lines changed: 1 addition & 0 deletions
@@ -24,6 +24,10 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake) {
+    return createBackendApi(intake, true);
+  }
+
+  public @Nullable BackendApi createBackendApi(Intake intake, boolean responseCompression) {
     HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true);
 
     if (intake.isAgentlessEnabled(config)) {
@@ -46,23 +50,28 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
     DDAgentFeaturesDiscovery featuresDiscovery =
         sharedCommunicationObjects.featuresDiscovery(config);
     featuresDiscovery.discoverIfOutdated();
-    if (featuresDiscovery.supportsEvpProxy()) {
-      String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
-      String evpProxyEndpoint = featuresDiscovery.getEvpProxyEndpoint();
-      HttpUrl evpProxyUrl = sharedCommunicationObjects.agentUrl.resolve(evpProxyEndpoint);
-      String subdomain = intake.getUrlPrefix();
-      return new EvpProxyApi(
-          traceId,
-          evpProxyUrl,
-          subdomain,
-          retryPolicyFactory,
-          sharedCommunicationObjects.agentHttpClient,
-          true);
+    if (!featuresDiscovery.supportsEvpProxy()) {
+      log.warn(
+          "Cannot create backend API client since agentless mode is disabled, "
+              + "and agent does not support EVP proxy");
+      return null;
     }
+    String evpProxyEndpoint = featuresDiscovery.getEvpProxyEndpoint();
 
-    log.warn(
-        "Cannot create backend API client since agentless mode is disabled, "
-            + "and agent does not support EVP proxy");
-    return null;
+    String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
+    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(
+        traceId,
+        evpProxyUrl,
+        subdomain,
+        retryPolicyFactory,
+        sharedCommunicationObjects.agentHttpClient,
+        responseCompression);
   }
 }
@@ -0,0 +1,15 @@
+package datadog.communication;
+
+/** Shared EVP proxy constants. */
+public final class EvpProxy {
+
+  public static final String SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain";
+
+  /**
+   * Default SDK-side target for uncompressed EVP request bodies. Writers may split batches at or
+   * below this size to keep Agent proxy requests comfortably bounded.
+   */
+  public static final int PAYLOAD_SIZE_LIMIT_BYTES = 5 * 1024 * 1024;
+
+  private EvpProxy() {}
+}
@@ -20,12 +20,12 @@ 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";
   private static final String CONTENT_ENCODING_HEADER = "Content-Encoding";
   private static final String GZIP_ENCODING = "gzip";
+  private static final String IDENTITY_ENCODING = "identity";
 
   private final String traceId;
   private final HttpRetryPolicy.Factory retryPolicyFactory;
@@ -62,7 +62,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);
 
@@ -74,11 +74,21 @@ public <T> T post(
       requestBuilder.addHeader(CONTENT_ENCODING_HEADER, GZIP_ENCODING);
     }
 
+    // OkHttp's BridgeInterceptor adds a transparent Accept-Encoding: gzip when the caller does not
+    // set one. Set the header explicitly on both paths so responseCompression=false actually
+    // suppresses gzip negotiation on the wire.
     if (responseCompression) {
       requestBuilder.addHeader(ACCEPT_ENCODING_HEADER, GZIP_ENCODING);
+    } else {
+      requestBuilder.addHeader(ACCEPT_ENCODING_HEADER, IDENTITY_ENCODING);
     }
 
     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)) {
 
@@ -0,0 +1,112 @@
+package datadog.communication;
+
+import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V4_EVP_PROXY_ENDPOINT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
+import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.metrics.api.Monitoring;
+import datadog.trace.api.Config;
+import datadog.trace.api.ProtocolVersion;
+import datadog.trace.api.intake.Intake;
+import java.nio.charset.StandardCharsets;
+import okhttp3.HttpUrl;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.RequestBody;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.api.Test;
+
+class BackendApiFactoryTest {
+
+  private static final MediaType JSON = MediaType.parse("application/json");
+
+  @Test
+  void noBackendApiWhenAgentDoesNotAdvertiseEvpProxy() {
+    final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(null);
+    final BackendApiFactory factory =
+        new BackendApiFactory(Config.get(), sharedCommunicationObjects(discovery, null));
+
+    assertNull(factory.createBackendApi(Intake.EVENT_PLATFORM, false));
+  }
+
+  @Test
+  void advertisedEvpProxyEndpointSupportsDisabledResponseCompression() throws Exception {
+    final MockWebServer agent = new MockWebServer();
+    agent.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));
+    agent.start();
+    try {
+      final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT);
+      final BackendApiFactory factory =
+          new BackendApiFactory(
+              Config.get(), sharedCommunicationObjects(discovery, agent.url("/")));
+      final BackendApi api = factory.createBackendApi(Intake.EVENT_PLATFORM, false);
+
+      assertNotNull(api);
+      api.post(
+          "flagevaluation",
+          RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)),
+          stream -> null,
+          null,
+          false);
+
+      final RecordedRequest request = agent.takeRequest();
+      assertEquals("/evp_proxy/v4/api/v2/flagevaluation", request.getPath());
+    } finally {
+      agent.shutdown();
+    }
+  }
+
+  private static SharedCommunicationObjects sharedCommunicationObjects(
+      final DDAgentFeaturesDiscovery discovery, final HttpUrl agentUrl) {
+    final TestSharedCommunicationObjects sco = new TestSharedCommunicationObjects(discovery);
+    sco.agentUrl = agentUrl != null ? agentUrl : HttpUrl.get("http://localhost:8126/");
+    sco.agentHttpClient = new OkHttpClient();
+    return sco;
+  }
+
+  private static final class TestSharedCommunicationObjects extends SharedCommunicationObjects {
+    private final DDAgentFeaturesDiscovery discovery;
+
+    private TestSharedCommunicationObjects(final DDAgentFeaturesDiscovery discovery) {
+      this.discovery = discovery;
+    }
+
+    @Override
+    public DDAgentFeaturesDiscovery featuresDiscovery(final Config config) {
+      return discovery;
+    }
+  }
+
+  private static final class FakeFeaturesDiscovery extends DDAgentFeaturesDiscovery {
+    private final String evpProxyEndpoint;
+
+    private FakeFeaturesDiscovery(final String evpProxyEndpoint) {
+      super(
+          new OkHttpClient(),
+          Monitoring.DISABLED,
+          HttpUrl.get("http://localhost:8126/"),
+          ProtocolVersion.V0_5,
+          true,
+          false);
+      this.evpProxyEndpoint = evpProxyEndpoint;
+    }
+
+    @Override
+    public void discoverIfOutdated() {}
+
+    @Override
+    public String getEvpProxyEndpoint() {
+      return evpProxyEndpoint;
+    }
+
+    @Override
+    public boolean supportsEvpProxy() {
+      return evpProxyEndpoint != null;
+    }
+  }
+}
@@ -524,15 +524,17 @@ public static void shutdown(final boolean sync) {
     if (profilingEnabled) {
       shutdownProfilingAgent(sync);
     }
+    // Before telemetry: the feature flagging writers queue drop/degradation metrics during their
+    // final flush, and only a still-running telemetry worker can drain and transmit them.
+    if (featureFlaggingEnabled) {
+      shutdownFeatureFlagging(AGENT_CLASSLOADER);
+    }
     if (telemetryEnabled) {
       stopTelemetry();
     }
     if (flareEnabled) {
       stopFlarePoller();
     }
-    if (featureFlaggingEnabled) {
-      shutdownFeatureFlagging(AGENT_CLASSLOADER);
-    }
 
     if (agentlessLogSubmissionEnabled) {
       shutdownLogsIntake();
 
@@ -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()) {
 
@@ -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) {
 
@@ -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()
 
@@ -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
 
@@ -67,6 +67,7 @@ public enum AgentThread {
     LLMOBS_EVALS_PROCESSOR("dd-llmobs-evals-processor"),
 
     FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor"),
+    FEATURE_FLAG_EVALUATION_PROCESSOR("dd-ffe-evaluation-processor"),
     FEATURE_FLAG_CONFIGURATION_POLLER("dd-feature-flagging-http-poller");
 
     public final String threadName;