Send Java SDK identity with Feature Flagging EVP events

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

@@ -1,16 +1,21 @@
 package datadog.communication;
 
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.unmodifiableMap;
 
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.trace.api.Config;
 import datadog.trace.api.intake.Intake;
 import datadog.trace.util.throwable.FatalAgentMisconfigurationError;
+import java.util.HashMap;
+import java.util.Map;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
+import okhttp3.Request;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -22,24 +27,35 @@ public class BackendApiFactory {
 
   private final Config config;
   private final SharedCommunicationObjects sharedCommunicationObjects;
+  private final Map<String, String> requestHeaders;
   private final boolean sendOnce;
 
   public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommunicationObjects) {
-    this(config, sharedCommunicationObjects, false);
+    this(config, sharedCommunicationObjects, emptyMap(), false);
+  }
+
+  /** Creates a backend factory with optional send-once transport semantics and no extra headers. */
+  public BackendApiFactory(
+      Config config, SharedCommunicationObjects sharedCommunicationObjects, boolean sendOnce) {
+    this(config, sharedCommunicationObjects, emptyMap(), sendOnce);
   }
 
   /**
-   * Creates a backend factory with optional send-once transport semantics.
+   * Creates a backend factory with per-request headers and optional send-once transport semantics.
    *
    * <p>When {@code sendOnce} is true, automatic connection retries and redirects are disabled, and
    * the default HTTP retry policy is {@link HttpRetryPolicy.Factory#NEVER_RETRY}. Explicit caller
    * policies may retry proven pre-send failures, but must not replay an ambiguously accepted event
    * payload without an idempotency key.
    */
   public BackendApiFactory(
-      Config config, SharedCommunicationObjects sharedCommunicationObjects, boolean sendOnce) {
+      Config config,
+      SharedCommunicationObjects sharedCommunicationObjects,
+      Map<String, String> requestHeaders,
+      boolean sendOnce) {
     this.config = config;
     this.sharedCommunicationObjects = sharedCommunicationObjects;
+    this.requestHeaders = unmodifiableMap(new HashMap<>(requestHeaders));
     this.sendOnce = sendOnce;
   }
 
@@ -182,7 +198,8 @@ private static boolean isAsciiLetterOrDigit(final char character) {
    * Creates an EVP proxy client after Agent discovery, optionally forcing a fresh discovery and
    * requiring the Agent to advertise every specified forwarding header.
    *
-   * <p>Callers supply the forwarding capabilities their event protocol requires.
+   * <p>Callers supply the forwarding capabilities their event protocol requires. These are
+   * independent of any headers configured on this factory.
    *
    * <p>The {@code forceDiscovery} form is intended for bounded unavailable-route recovery probes.
    */
@@ -250,14 +267,24 @@ private BackendApi createEvpProxyApi(
   }
 
   OkHttpClient configureHttpClient(final OkHttpClient httpClient) {
-    return sendOnce
-        ? httpClient
-            .newBuilder()
-            .retryOnConnectionFailure(false)
-            .followRedirects(false)
-            .followSslRedirects(false)
-            .build()
-        : httpClient;
+    if (requestHeaders.isEmpty() && !sendOnce) {
+      return httpClient;
+    }
+    final OkHttpClient.Builder builder = httpClient.newBuilder();
+    if (sendOnce) {
+      builder.retryOnConnectionFailure(false).followRedirects(false).followSslRedirects(false);
+    }
+    if (!requestHeaders.isEmpty()) {
+      builder.addInterceptor(
+          chain -> {
+            final Request.Builder requestBuilder = chain.request().newBuilder();
+            for (Map.Entry<String, String> header : requestHeaders.entrySet()) {
+              requestBuilder.header(header.getKey(), header.getValue());
+            }
+            return chain.proceed(requestBuilder.build());
+          });
+    }
+    return builder.build();
   }
 
   private HttpRetryPolicy.Factory retryPolicyFactory() {

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

@@ -11,6 +11,9 @@ public final class EvpProxy {
   /** Identifies the version of the SDK that produced an EVP request. */
   public static final String ORIGIN_VERSION_HEADER = "DD-EVP-ORIGIN-VERSION";
 
+  /** Origin header value identifying this tracing library. */
+  public static final String JAVA_TRACING_LIBRARY = "dd-trace-java";
+
   /**
    * 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.

communication/src/test/java/datadog/communication/BackendApiFactoryTest.java

@@ -1,10 +1,14 @@
 package datadog.communication;
 
+import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY;
 import static datadog.communication.EvpProxy.ORIGIN_HEADER;
 import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER;
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT;
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V4_EVP_PROXY_ENDPOINT;
+import static datadog.trace.api.config.CiVisibilityConfig.CIVISIBILITY_AGENTLESS_URL;
+import static datadog.trace.api.config.GeneralConfig.API_KEY;
 import static java.util.Arrays.asList;
+import static java.util.Collections.singletonMap;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -22,7 +26,10 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
 import java.util.stream.Stream;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
@@ -327,15 +334,18 @@ void discoveryRejectsMissingOrPartialIdentityCapabilities(final String capabilit
   }
 
   @Test
-  void featureFlagProxyRequestNeverCarriesDirectCredentials() throws Exception {
+  void featureFlagProxyRequestCarriesSdkIdentityWithoutDirectCredentials() 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("/")), true);
+              Config.get(),
+              sharedCommunicationObjects(discovery, agent.url("/")),
+              sdkHeaders(),
+              true);
       final BackendApi api =
           factory.createEvpProxyApi(
               Intake.EVENT_PLATFORM,
@@ -355,6 +365,8 @@ void featureFlagProxyRequestNeverCarriesDirectCredentials() throws Exception {
       final RecordedRequest request = agent.takeRequest();
       assertEquals("/evp_proxy/v4/api/v2/exposures", request.getPath());
       assertEquals("event-platform-intake", request.getHeader("X-Datadog-EVP-Subdomain"));
+      assertEquals(JAVA_TRACING_LIBRARY, request.getHeader(ORIGIN_HEADER));
+      assertEquals("test-version", request.getHeader(ORIGIN_VERSION_HEADER));
       assertNull(request.getHeader("DD-API-KEY"));
       assertEquals(1, agent.getRequestCount());
     } finally {
@@ -363,15 +375,56 @@ void featureFlagProxyRequestNeverCarriesDirectCredentials() throws Exception {
   }
 
   @Test
-  void featureFlagDirectRequestCarriesCredentialsExactlyOnce() throws Exception {
+  void directIntakeSendsConfiguredRequestHeaders() throws Exception {
+    final MockWebServer intake = new MockWebServer();
+    intake.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));
+    intake.start();
+    final SharedCommunicationObjects sco =
+        sharedCommunicationObjects(new FakeFeaturesDiscovery(null), null);
+    try {
+      final Properties properties = new Properties();
+      properties.setProperty(API_KEY, "api-key");
+      properties.setProperty(
+          CIVISIBILITY_AGENTLESS_URL, intake.url("/").toString().replaceAll("/$", ""));
+      final Config config = Config.get(properties);
+      final BackendApiFactory factory =
+          new BackendApiFactory(
+              config, sco, singletonMap(ORIGIN_HEADER, JAVA_TRACING_LIBRARY), false);
+
+      // followRedirects=false mirrors the feature-flagging caller, so this also covers the
+      // interaction between the redirect-scoped client and the header interceptor.
+      final BackendApi api = factory.createDirectIntakeApi(Intake.API, false, false);
+
+      assertNotNull(api);
+      api.post(
+          "flagevaluation",
+          RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)),
+          stream -> null,
+          null,
+          false);
+
+      final RecordedRequest request = intake.takeRequest();
+      assertEquals(JAVA_TRACING_LIBRARY, request.getHeader(ORIGIN_HEADER));
+      assertEquals("api-key", request.getHeader("DD-API-KEY"));
+    } finally {
+      sco.getIntakeHttpClient().connectionPool().evictAll();
+      intake.shutdown();
+    }
+  }
+
+  @Test
+  void featureFlagDirectRequestCarriesCredentialsAndIdentityExactlyOnce() throws Exception {
     final MockWebServer intake = new MockWebServer();
     intake.enqueue(new MockResponse().setResponseCode(500).setBody("failed"));
     intake.enqueue(new MockResponse().setResponseCode(200).setBody("{}"));
     intake.start();
     final OkHttpClient sharedClient = new OkHttpClient.Builder().build();
     final BackendApiFactory factory =
         new BackendApiFactory(
-            Config.get(), sharedCommunicationObjects(new FakeFeaturesDiscovery(null), null), true);
+            Config.get(),
+            sharedCommunicationObjects(new FakeFeaturesDiscovery(null), null),
+            sdkHeaders(),
+            true);
     final OkHttpClient directClient = factory.configureHttpClient(sharedClient);
     try {
       final IntakeApi api =
@@ -396,6 +449,8 @@ void featureFlagDirectRequestCarriesCredentialsExactlyOnce() throws Exception {
       final RecordedRequest request = intake.takeRequest();
       assertEquals("/api/v2/flagevaluation", request.getPath());
       assertEquals("api-key", request.getHeader("DD-API-KEY"));
+      assertEquals("dd-trace-java", request.getHeader("DD-EVP-ORIGIN"));
+      assertEquals("test-version", request.getHeader("DD-EVP-ORIGIN-VERSION"));
       assertNull(request.getHeader("X-Datadog-EVP-Subdomain"));
       assertEquals(1, intake.getRequestCount());
       assertEquals(false, directClient.retryOnConnectionFailure());
@@ -523,6 +578,13 @@ private static SharedCommunicationObjects sharedCommunicationObjects(
     return sco;
   }
 
+  private static Map<String, String> sdkHeaders() {
+    final Map<String, String> headers = new HashMap<>(2);
+    headers.put("DD-EVP-ORIGIN", "dd-trace-java");
+    headers.put("DD-EVP-ORIGIN-VERSION", "test-version");
+    return headers;
+  }
+
   private static final class TestSharedCommunicationObjects extends SharedCommunicationObjects {
     private final DDAgentFeaturesDiscovery discovery;
 

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagBackendApiFactory.java

@@ -1,16 +1,23 @@
 package com.datadog.featureflag;
 
+import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY;
+import static datadog.communication.EvpProxy.ORIGIN_HEADER;
+import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER;
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT;
 import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS;
-import static java.util.Collections.emptyList;
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableMap;
 
 import datadog.communication.BackendApi;
 import datadog.communication.BackendApiFactory;
 import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.communication.ddagent.TracerVersion;
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.trace.api.Config;
 import datadog.trace.api.intake.Intake;
 import java.net.ConnectException;
+import java.util.HashMap;
+import java.util.Map;
 import javax.annotation.Nullable;
 import okhttp3.Response;
 import org.slf4j.Logger;
@@ -20,6 +27,7 @@
 final class FeatureFlagBackendApiFactory {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagBackendApiFactory.class);
+  private static final Map<String, String> REQUEST_HEADERS = requestHeaders();
 
   /** Retries connection refusal only; automatic retries and redirects remain disabled. */
   static final HttpRetryPolicy.Factory PRE_SEND_RETRY =
@@ -85,7 +93,7 @@ private static BackendApiFactory initializedBackendFactory(
       // cached HTTP client uses configured timeouts even when Feature Flagging starts first.
       sharedCommunicationObjects.createRemaining(config);
     }
-    return new BackendApiFactory(config, sharedCommunicationObjects, true);
+    return new BackendApiFactory(config, sharedCommunicationObjects, REQUEST_HEADERS, true);
   }
 
   FeatureFlagBackendApiFactory(
@@ -139,14 +147,22 @@ BackendApi create() {
   @Nullable
   private String discoverProxyEndpoint(final boolean forceDiscovery) {
     try {
-      // Identity-forwarding requirements belong with SDK identity emission in the identity PR.
-      return backendApiFactory.discoverEvpProxyEndpoint(forceDiscovery, emptyList());
+      // A local route must preserve the identity headers that this sender emits on every batch.
+      return backendApiFactory.discoverEvpProxyEndpoint(
+          forceDiscovery, asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER));
     } catch (final RuntimeException exception) {
       LOGGER.debug("Could not discover the local Feature Flagging route", exception);
       return null;
     }
   }
 
+  private static Map<String, String> requestHeaders() {
+    final Map<String, String> headers = new HashMap<>(2);
+    headers.put(ORIGIN_HEADER, JAVA_TRACING_LIBRARY);
+    headers.put(ORIGIN_VERSION_HEADER, TracerVersion.TRACER_VERSION);
+    return unmodifiableMap(headers);
+  }
+
   private boolean hasDirectCredentials() {
     final String apiKey = config.getApiKey();
     return apiKey != null && !apiKey.isEmpty();