Send agentless Feature Flags exposures directly to EVP

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

@@ -28,32 +28,53 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake, boolean responseCompression) {
-    HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true);
-
     if (intake.isAgentlessEnabled(config)) {
-      HttpUrl agentlessUrl = HttpUrl.get(intake.getAgentlessUrl(config));
-      String apiKey = config.getApiKey();
-      if (apiKey == null || apiKey.isEmpty()) {
-        throw new FatalAgentMisconfigurationError(
-            "Agentless mode is enabled and api key is not set. Please set application key");
-      }
-      String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
-      return new IntakeApi(
-          agentlessUrl,
-          apiKey,
-          traceId,
-          retryPolicyFactory,
-          sharedCommunicationObjects.getIntakeHttpClient(),
-          true);
+      return createDirectIntakeApi(intake, responseCompression);
+    }
+
+    BackendApi backendApi = createEvpProxyApi(intake, responseCompression);
+    if (backendApi == null) {
+      log.warn(
+          "Cannot create backend API client since agentless mode is disabled, "
+              + "and agent does not support EVP proxy");
+    }
+    return backendApi;
+  }
+
+  /** Creates an authenticated API client that sends data directly to a Datadog intake. */
+  public BackendApi createDirectIntakeApi(Intake intake) {
+    return createDirectIntakeApi(intake, true);
+  }
+
+  /** Creates an authenticated API client that sends data directly to a Datadog intake. */
+  public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompression) {
+    HttpUrl agentlessUrl = HttpUrl.get(intake.getAgentlessUrl(config));
+    String apiKey = config.getApiKey();
+    if (apiKey == null || apiKey.isEmpty()) {
+      throw new FatalAgentMisconfigurationError(
+          "Agentless mode is enabled and API key is not set. Please set DD_API_KEY");
     }
+    String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
+    return new IntakeApi(
+        agentlessUrl,
+        apiKey,
+        traceId,
+        retryPolicyFactory(),
+        sharedCommunicationObjects.getIntakeHttpClient(),
+        responseCompression);
+  }
+
+  /** Creates an API client that sends data through a compatible local EVP proxy. */
+  public @Nullable BackendApi createEvpProxyApi(Intake intake) {
+    return createEvpProxyApi(intake, true);
+  }
 
+  /** Creates an API client that sends data through a compatible local EVP proxy. */
+  public @Nullable BackendApi createEvpProxyApi(Intake intake, boolean responseCompression) {
     DDAgentFeaturesDiscovery featuresDiscovery =
         sharedCommunicationObjects.featuresDiscovery(config);
     featuresDiscovery.discoverIfOutdated();
     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();
@@ -70,8 +91,12 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
         traceId,
         evpProxyUrl,
         subdomain,
-        retryPolicyFactory,
+        retryPolicyFactory(),
         sharedCommunicationObjects.agentHttpClient,
         responseCompression);
   }
+
+  private static HttpRetryPolicy.Factory retryPolicyFactory() {
+    return new HttpRetryPolicy.Factory(5, 100, 2.0, true);
+  }
 }

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

@@ -105,7 +105,8 @@ public <T> T post(
 
         return responseParser.apply(responseBodyStream);
       } else {
-        throw new IOException(
+        throw new HttpResponseException(
+            response.code(),
             "Request to "
                 + uri
                 + " returned error response "

communication/src/main/java/datadog/communication/HttpResponseException.java

@@ -0,0 +1,18 @@
+package datadog.communication;
+
+import java.io.IOException;
+
+/** An HTTP request failed with a non-success response. */
+public final class HttpResponseException extends IOException {
+
+  private final int statusCode;
+
+  public HttpResponseException(final int statusCode, final String message) {
+    super(message);
+    this.statusCode = statusCode;
+  }
+
+  public int getStatusCode() {
+    return statusCode;
+  }
+}

communication/src/test/java/datadog/communication/EvpProxyApiTest.java

@@ -0,0 +1,65 @@
+package datadog.communication;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import datadog.communication.http.HttpRetryPolicy;
+import java.io.IOException;
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class EvpProxyApiTest {
+
+  private MockWebServer server;
+  private OkHttpClient client;
+
+  @BeforeEach
+  void setUp() throws IOException {
+    server = new MockWebServer();
+    server.start();
+    client = new OkHttpClient.Builder().build();
+  }
+
+  @AfterEach
+  void tearDown() throws IOException {
+    client.dispatcher().executorService().shutdownNow();
+    client.connectionPool().evictAll();
+    server.shutdown();
+  }
+
+  @Test
+  void reportsHttpStatusForRejectedRequest() throws Exception {
+    server.enqueue(new MockResponse().setResponseCode(404).setBody("not found"));
+    final EvpProxyApi api =
+        new EvpProxyApi(
+            "123",
+            server.url("/evp_proxy/v4/"),
+            "event-platform-intake",
+            HttpRetryPolicy.Factory.NEVER_RETRY,
+            client,
+            false);
+
+    final HttpResponseException exception =
+        assertThrows(
+            HttpResponseException.class,
+            () ->
+                api.post(
+                    "exposures",
+                    RequestBody.create(MediaType.parse("application/json"), "{}"),
+                    stream -> null,
+                    null,
+                    false));
+
+    assertEquals(404, exception.getStatusCode());
+    final RecordedRequest request = server.takeRequest();
+    assertEquals("/evp_proxy/v4/api/v2/exposures", request.getPath());
+    assertEquals("event-platform-intake", request.getHeader("X-Datadog-EVP-Subdomain"));
+  }
+}

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

@@ -14,6 +14,11 @@
 
 public class FeatureFlaggingSystem {
 
+  @FunctionalInterface
+  interface SystemInitializer {
+    void initialize(SharedCommunicationObjects sco, Config config);
+  }
+
   private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class);
 
   private static volatile ConfigurationSourceService CONFIG_SERVICE;
@@ -25,11 +30,12 @@ public class FeatureFlaggingSystem {
 
   private FeatureFlaggingSystem() {}
 
-  @SuppressFBWarnings(
-      value = "USO_UNSAFE_STATIC_METHOD_SYNCHRONIZATION",
-      justification =
-          "Agent-internal class; Class object does not escape to app code and lock only guards the subsystem lifecycle.")
-  public static synchronized void start(final SharedCommunicationObjects sco) {
+  public static void start(final SharedCommunicationObjects sco) {
+    start(sco, FeatureFlaggingSystem::initializeSystem);
+  }
+
+  static synchronized void start(
+      final SharedCommunicationObjects sco, final SystemInitializer systemInitializer) {
     if (STARTED) {
       LOGGER.debug("Feature Flagging system already started");
       return;
@@ -45,33 +51,37 @@ public static synchronized void start(final SharedCommunicationObjects sco) {
 
     if (CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) {
       final FeatureFlaggingGateway.ActivationListener activationListener =
-          () -> activateAgentless(sco, config);
+          () -> activateAgentless(sco, config, systemInitializer);
       ACTIVATION_LISTENER = activationListener;
       FeatureFlaggingGateway.addActivationListener(activationListener);
       LOGGER.debug("Feature Flagging system awaiting application provider activation");
       return;
     }
 
-    initializeOrRollBack(sco, config);
+    initializeOrRollBack(sco, config, systemInitializer);
   }
 
   private static synchronized void activateAgentless(
-      final SharedCommunicationObjects sco, final Config config) {
+      final SharedCommunicationObjects sco,
+      final Config config,
+      final SystemInitializer systemInitializer) {
     final FeatureFlaggingGateway.ActivationListener activationListener = ACTIVATION_LISTENER;
     if (!STARTED || activationListener == null) {
       return;
     }
     ACTIVATION_LISTENER = null;
     FeatureFlaggingGateway.removeActivationListener(activationListener);
-    initializeOrRollBack(sco, config);
+    initializeOrRollBack(sco, config, systemInitializer);
   }
 
   // Any failure leaves the subsystem fully stopped: stop() releases whatever initializeSystem
   // managed to publish before it threw, so a later start() begins from a clean state.
   private static void initializeOrRollBack(
-      final SharedCommunicationObjects sco, final Config config) {
+      final SharedCommunicationObjects sco,
+      final Config config,
+      final SystemInitializer systemInitializer) {
     try {
-      initializeSystem(sco, config);
+      systemInitializer.initialize(sco, config);
     } catch (final RuntimeException | Error e) {
       stop();
       throw e;
@@ -184,6 +194,14 @@ static boolean isAwaitingApplicationActivation() {
     return ACTIVATION_LISTENER != null;
   }
 
+  static boolean isExposureWriterStarted() {
+    return EXPOSURE_WRITER != null;
+  }
+
+  static boolean isConfigurationSourceStarted() {
+    return CONFIG_SERVICE != null;
+  }
+
   private static void closeQuietly(final AutoCloseable resource) {
     if (resource != null) {
       try {

products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java

@@ -8,6 +8,7 @@
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -36,7 +37,6 @@
 import org.junit.jupiter.api.Test;
 
 class FeatureFlaggingSystemTest {
-
   @AfterEach
   void resetFlagEvaluationGateway() {
     FeatureFlaggingSystem.stop();
@@ -49,24 +49,73 @@ void resetFlagEvaluationGateway() {
   @WithConfig(
       key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL,
       value = "http://127.0.0.1:1")
-  void agentlessStartWaitsForApplicationProviderActivation() {
+  void agentlessStartWaitsForApplicationProviderActivationWithoutPreparingDelivery() {
     SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects();
     clearInvocations(sharedCommunicationObjects);
 
     try {
       FeatureFlaggingSystem.start(sharedCommunicationObjects);
 
       assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation());
+      assertFalse(FeatureFlaggingSystem.isExposureWriterStarted());
+      assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted());
       verifyNoInteractions(sharedCommunicationObjects);
-
-      FeatureFlaggingGateway.activate();
-
-      assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation());
     } finally {
       FeatureFlaggingSystem.stop();
     }
 
     assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation());
+    assertFalse(FeatureFlaggingSystem.isExposureWriterStarted());
+    assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted());
+  }
+
+  @Test
+  @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless")
+  void agentlessActivationInitializesSystemOnce() {
+    final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects();
+    final FeatureFlaggingSystem.SystemInitializer systemInitializer =
+        mock(FeatureFlaggingSystem.SystemInitializer.class);
+
+    FeatureFlaggingSystem.start(sharedCommunicationObjects, systemInitializer);
+
+    verifyNoInteractions(systemInitializer);
+    assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation());
+
+    FeatureFlaggingGateway.activate();
+    FeatureFlaggingGateway.activate();
+
+    verify(systemInitializer).initialize(eq(sharedCommunicationObjects), any(Config.class));
+    assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation());
+  }
+
+  @Test
+  @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless")
+  void agentlessInitializationFailureCleansUpAndAllowsRetry() {
+    final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects();
+    final FeatureFlaggingSystem.SystemInitializer failedInitializer =
+        mock(FeatureFlaggingSystem.SystemInitializer.class);
+    final IllegalStateException initializationFailure =
+        new IllegalStateException("system initialization failed");
+    doThrow(initializationFailure)
+        .when(failedInitializer)
+        .initialize(any(SharedCommunicationObjects.class), any(Config.class));
+
+    FeatureFlaggingSystem.start(sharedCommunicationObjects, failedInitializer);
+
+    final IllegalStateException thrown =
+        assertThrows(IllegalStateException.class, FeatureFlaggingGateway::activate);
+
+    assertSame(initializationFailure, thrown);
+    assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation());
+    assertFalse(FeatureFlaggingSystem.isExposureWriterStarted());
+    assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted());
+
+    final FeatureFlaggingSystem.SystemInitializer retryInitializer =
+        mock(FeatureFlaggingSystem.SystemInitializer.class);
+    FeatureFlaggingSystem.start(sharedCommunicationObjects, retryInitializer);
+    FeatureFlaggingGateway.activate();
+
+    verify(retryInitializer).initialize(eq(sharedCommunicationObjects), any(Config.class));
   }
 
   @Test
@@ -83,6 +132,7 @@ void agentlessStopRemovesPendingApplicationProviderActivation() {
       assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation());
 
       FeatureFlaggingSystem.stop();
+      clearInvocations(sharedCommunicationObjects);
       FeatureFlaggingGateway.activate();
 
       assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation());
@@ -221,6 +271,8 @@ void agentlessConfigurationSourceStartsTelemetryWritersWithoutRemoteConfig() {
       // Agentless defers initialization until the application provider activates.
       FeatureFlaggingGateway.activate();
 
+      assertTrue(FeatureFlaggingSystem.isExposureWriterStarted());
+      assertTrue(FeatureFlaggingSystem.isConfigurationSourceStarted());
       assertTrue(FeatureFlaggingGateway.isFlagEvaluationEnqueueEnabled());
       assertNotNull(FeatureFlaggingGateway.getFlagEvalWriter());
     } finally {
@@ -338,6 +390,11 @@ private static SharedCommunicationObjects sharedCommunicationObjects() {
     DDAgentFeaturesDiscovery discovery = mock(DDAgentFeaturesDiscovery.class);
     when(discovery.supportsEvpProxy()).thenReturn(true);
     when(discovery.getEvpProxyEndpoint()).thenReturn("/evp_proxy/");
+    return sharedCommunicationObjects(discovery);
+  }
+
+  private static SharedCommunicationObjects sharedCommunicationObjects(
+      final DDAgentFeaturesDiscovery discovery) {
     SharedCommunicationObjects sharedCommunicationObjects = mock(SharedCommunicationObjects.class);
     when(sharedCommunicationObjects.featuresDiscovery(any(Config.class))).thenReturn(discovery);
     sharedCommunicationObjects.agentUrl = HttpUrl.get("http://localhost");