From b9e2b1507a24d81770580bc5bcad0b36cc009235 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 11 Aug 2026 19:54:06 -0400 Subject: [PATCH 01/10] Send agentless exposures directly to EVP --- .../communication/BackendApiFactory.java | 56 +++++--- .../datadog/communication/EvpProxyApi.java | 3 +- .../communication/HttpResponseException.java | 18 +++ .../communication/EvpProxyApiTest.java | 65 ++++++++++ .../feature-flagging-lib/build.gradle.kts | 2 + .../AgentlessExposureBackendApi.java | 66 ++++++++++ .../ExposureBackendApiFactory.java | 72 +++++++++++ .../featureflag/ExposureWriterImpl.java | 24 ++-- .../AgentlessExposureBackendApiTest.java | 122 ++++++++++++++++++ .../ExposureBackendApiFactoryTest.java | 103 +++++++++++++++ .../featureflag/ExposureWriterTests.java | 49 ++++++- 11 files changed, 546 insertions(+), 34 deletions(-) create mode 100644 communication/src/main/java/datadog/communication/HttpResponseException.java create mode 100644 communication/src/test/java/datadog/communication/EvpProxyApiTest.java create mode 100644 products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java create mode 100644 products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java create mode 100644 products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java create mode 100644 products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index 3ce78b88c22..5b7d92b29a0 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java @@ -24,25 +24,39 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni } public @Nullable BackendApi createBackendApi(Intake intake) { - 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); } + BackendApi backendApi = createEvpProxyApi(intake); + 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) { + 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); + } + + /** Creates an API client that sends data through a compatible local EVP proxy. */ + public @Nullable BackendApi createEvpProxyApi(Intake intake) { DDAgentFeaturesDiscovery featuresDiscovery = sharedCommunicationObjects.featuresDiscovery(config); featuresDiscovery.discoverIfOutdated(); @@ -55,14 +69,14 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni traceId, evpProxyUrl, subdomain, - retryPolicyFactory, + retryPolicyFactory(), sharedCommunicationObjects.agentHttpClient, true); } - - log.warn( - "Cannot create backend API client since agentless mode is disabled, " - + "and agent does not support EVP proxy"); return null; } + + private static HttpRetryPolicy.Factory retryPolicyFactory() { + return new HttpRetryPolicy.Factory(5, 100, 2.0, true); + } } diff --git a/communication/src/main/java/datadog/communication/EvpProxyApi.java b/communication/src/main/java/datadog/communication/EvpProxyApi.java index 83037ab9663..49f0285aac1 100644 --- a/communication/src/main/java/datadog/communication/EvpProxyApi.java +++ b/communication/src/main/java/datadog/communication/EvpProxyApi.java @@ -95,7 +95,8 @@ public T post( return responseParser.apply(responseBodyStream); } else { - throw new IOException( + throw new HttpResponseException( + response.code(), "Request to " + uri + " returned error response " diff --git a/communication/src/main/java/datadog/communication/HttpResponseException.java b/communication/src/main/java/datadog/communication/HttpResponseException.java new file mode 100644 index 00000000000..ac9b62cdb2d --- /dev/null +++ b/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; + } +} diff --git a/communication/src/test/java/datadog/communication/EvpProxyApiTest.java b/communication/src/test/java/datadog/communication/EvpProxyApiTest.java new file mode 100644 index 00000000000..14c6962bf8e --- /dev/null +++ b/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")); + } +} diff --git a/products/feature-flagging/feature-flagging-lib/build.gradle.kts b/products/feature-flagging/feature-flagging-lib/build.gradle.kts index 425e217e822..266966ca894 100644 --- a/products/feature-flagging/feature-flagging-lib/build.gradle.kts +++ b/products/feature-flagging/feature-flagging-lib/build.gradle.kts @@ -20,6 +20,7 @@ dependencies { api(project(":communication")) implementation(project(":internal-api")) api(project(":products:feature-flagging:feature-flagging-bootstrap")) + compileOnly(project(":products:feature-flagging:feature-flagging-config")) implementation(project(":utils:logging-utils")) api(project(":utils:queue-utils")) @@ -29,6 +30,7 @@ dependencies { testImplementation(libs.bundles.junit5) testImplementation(libs.bundles.mockito) + testImplementation(project(":products:feature-flagging:feature-flagging-config")) testImplementation(project(":utils:test-utils")) testImplementation(project(":dd-java-agent:testing")) } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java new file mode 100644 index 00000000000..97c798f0e1d --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java @@ -0,0 +1,66 @@ +package com.datadog.featureflag; + +import datadog.communication.BackendApi; +import datadog.communication.HttpResponseException; +import datadog.communication.http.OkHttpUtils; +import datadog.communication.util.IOThrowingFunction; +import java.io.IOException; +import java.io.InputStream; +import java.net.ConnectException; +import javax.annotation.Nullable; +import okhttp3.RequestBody; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Sends exposures through a local EVP proxy, with a safe direct intake fallback. */ +final class AgentlessExposureBackendApi implements BackendApi { + + private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessExposureBackendApi.class); + + private final BackendApi localApi; + private final BackendApi directApi; + private volatile BackendApi activeApi; + + AgentlessExposureBackendApi(final BackendApi localApi, final BackendApi directApi) { + this.localApi = localApi; + this.directApi = directApi; + this.activeApi = localApi; + } + + @Override + public T post( + final String uri, + final RequestBody requestBody, + final IOThrowingFunction responseParser, + @Nullable final OkHttpUtils.CustomListener requestListener, + final boolean requestCompression) + throws IOException { + final BackendApi selectedApi = activeApi; + try { + return selectedApi.post( + uri, requestBody, responseParser, requestListener, requestCompression); + } catch (final IOException exception) { + if (selectedApi != localApi || !isDefinitiveRejection(exception)) { + throw exception; + } + + if (activeApi == localApi) { + LOGGER.debug( + "Switching Feature Flagging exposure delivery from the local EVP proxy to direct intake"); + activeApi = directApi; + } + return directApi.post(uri, requestBody, responseParser, requestListener, requestCompression); + } + } + + private static boolean isDefinitiveRejection(final IOException exception) { + if (exception instanceof ConnectException) { + return true; + } + if (exception instanceof HttpResponseException) { + final int statusCode = ((HttpResponseException) exception).getStatusCode(); + return statusCode == 403 || statusCode == 404 || statusCode == 405; + } + return false; + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java new file mode 100644 index 00000000000..f0eb1fa0e3d --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java @@ -0,0 +1,72 @@ +package com.datadog.featureflag; + +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; + +import datadog.communication.BackendApi; +import datadog.communication.BackendApiFactory; +import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.trace.api.Config; +import datadog.trace.api.intake.Intake; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Selects the transport for Feature Flagging exposure events. */ +final class ExposureBackendApiFactory { + + private static final Logger LOGGER = LoggerFactory.getLogger(ExposureBackendApiFactory.class); + + private final Config config; + private final BackendApiFactory backendApiFactory; + + ExposureBackendApiFactory( + final Config config, final SharedCommunicationObjects sharedCommunicationObjects) { + this(config, new BackendApiFactory(config, sharedCommunicationObjects)); + } + + ExposureBackendApiFactory(final Config config, final BackendApiFactory backendApiFactory) { + this.config = config; + this.backendApiFactory = backendApiFactory; + } + + @Nullable + BackendApi create() { + final BackendApi localApi = backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM); + if (!CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { + if (localApi == null) { + LOGGER.warn( + "Feature Flagging exposure delivery is disabled because the local Agent does not support the EVP proxy"); + } + return localApi; + } + + final BackendApi directApi = createDirectApi(); + if (localApi != null && directApi != null) { + return new AgentlessExposureBackendApi(localApi, directApi); + } + if (localApi != null) { + return localApi; + } + if (directApi != null) { + return directApi; + } + + LOGGER.warn( + "Feature Flagging exposure delivery is disabled because no compatible local EVP proxy or direct intake credentials are available"); + return null; + } + + @Nullable + private BackendApi createDirectApi() { + final String apiKey = config.getApiKey(); + if (apiKey == null || apiKey.isEmpty()) { + return null; + } + try { + return backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM); + } catch (final IllegalArgumentException exception) { + LOGGER.debug("Cannot configure direct Feature Flagging exposure delivery", exception); + return null; + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java index 9932a20256b..0d8d967ed56 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java @@ -10,13 +10,11 @@ import datadog.common.queue.MessagePassingBlockingQueue; import datadog.common.queue.Queues; import datadog.communication.BackendApi; -import datadog.communication.BackendApiFactory; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.exposure.ExposureEvent; import datadog.trace.api.featureflag.exposure.ExposuresRequest; -import datadog.trace.api.intake.Intake; import datadog.trace.api.internal.VisibleForTesting; import java.util.ArrayList; import java.util.HashMap; @@ -47,6 +45,15 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con final TimeUnit timeUnit, final SharedCommunicationObjects sco, final Config config) { + this(capacity, flushInterval, timeUnit, new ExposureBackendApiFactory(config, sco), config); + } + + ExposureWriterImpl( + final int capacity, + final long flushInterval, + final TimeUnit timeUnit, + final ExposureBackendApiFactory backendApiFactory, + final Config config) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); final Map context = new HashMap<>(4); context.put("service", config.getServiceName() == null ? "unknown" : config.getServiceName()); @@ -58,12 +65,7 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con } final ExposureSerializingHandler serializer = new ExposureSerializingHandler( - new BackendApiFactory(config, sco), - queue, - flushInterval, - timeUnit, - context, - this::close); + backendApiFactory, queue, flushInterval, timeUnit, context, this::close); this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer); } @@ -102,7 +104,7 @@ private static class ExposureSerializingHandler implements Runnable { private long lastTicks; private final JsonAdapter jsonAdapter; - private final BackendApiFactory backendApiFactory; + private final ExposureBackendApiFactory backendApiFactory; private BackendApi evp; private final Map context; @@ -112,7 +114,7 @@ private static class ExposureSerializingHandler implements Runnable { private final Runnable errorCallback; public ExposureSerializingHandler( - final BackendApiFactory backendApiFactory, + final ExposureBackendApiFactory backendApiFactory, final MessagePassingBlockingQueue queue, final long flushInterval, final TimeUnit timeUnit, @@ -134,7 +136,7 @@ public ExposureSerializingHandler( @Override public void run() { - evp = backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM); + evp = backendApiFactory.create(); if (evp == null) { errorCallback.run(); throw new IllegalArgumentException("EVP Proxy not available"); diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java new file mode 100644 index 00000000000..9455634ec45 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java @@ -0,0 +1,122 @@ +package com.datadog.featureflag; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import datadog.communication.BackendApi; +import datadog.communication.HttpResponseException; +import datadog.communication.http.OkHttpUtils; +import datadog.communication.util.IOThrowingFunction; +import java.io.IOException; +import java.io.InputStream; +import java.net.ConnectException; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.util.ArrayList; +import java.util.List; +import javax.annotation.Nullable; +import okhttp3.MediaType; +import okhttp3.RequestBody; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class AgentlessExposureBackendApiTest { + + @ParameterizedTest + @ValueSource(ints = {403, 404, 405}) + void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throws Exception { + final RecordingBackendApi local = + new RecordingBackendApi(new HttpResponseException(statusCode, "rejected")); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + final RequestBody firstBody = requestBody("first"); + final RequestBody secondBody = requestBody("second"); + + api.post("exposures", firstBody, stream -> null, null, false); + api.post("exposures", secondBody, stream -> null, null, false); + + assertEquals(1, local.calls); + assertEquals(2, direct.calls); + assertSame(firstBody, local.requestBodies.get(0)); + assertSame(firstBody, direct.requestBodies.get(0)); + assertSame(secondBody, direct.requestBodies.get(1)); + } + + @Test + void fallsBackAfterConnectionRefusal() throws Exception { + final RecordingBackendApi local = + new RecordingBackendApi(new ConnectException("connection refused")); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + + api.post("exposures", requestBody("exposure"), stream -> null, null, false); + + assertEquals(1, local.calls); + assertEquals(1, direct.calls); + } + + @ParameterizedTest + @ValueSource(ints = {429, 500}) + void doesNotReplayAmbiguousHttpFailure(final int statusCode) { + assertNoDirectReplay(new HttpResponseException(statusCode, "ambiguous")); + } + + @Test + void doesNotReplayTimeout() { + assertNoDirectReplay(new SocketTimeoutException("timed out")); + } + + @Test + void doesNotReplayConnectionReset() { + assertNoDirectReplay(new SocketException("connection reset")); + } + + private static void assertNoDirectReplay(final IOException failure) { + final RecordingBackendApi local = new RecordingBackendApi(failure); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("exposure"), stream -> null, null, false)); + + assertEquals(1, local.calls); + assertEquals(0, direct.calls); + } + + private static RequestBody requestBody(final String value) { + return RequestBody.create(MediaType.parse("application/json"), value); + } + + private static final class RecordingBackendApi implements BackendApi { + private final IOException failure; + private final List requestBodies = new ArrayList<>(); + private int calls; + + private RecordingBackendApi() { + this(null); + } + + private RecordingBackendApi(@Nullable final IOException failure) { + this.failure = failure; + } + + @Override + public T post( + final String uri, + final RequestBody requestBody, + final IOThrowingFunction responseParser, + @Nullable final OkHttpUtils.CustomListener requestListener, + final boolean requestCompression) + throws IOException { + calls++; + requestBodies.add(requestBody); + if (failure != null) { + throw failure; + } + return null; + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java new file mode 100644 index 00000000000..602a3219ef3 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java @@ -0,0 +1,103 @@ +package com.datadog.featureflag; + +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_REMOTE_CONFIG; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import datadog.communication.BackendApi; +import datadog.communication.BackendApiFactory; +import datadog.trace.api.Config; +import datadog.trace.api.intake.Intake; +import org.junit.jupiter.api.Test; + +class ExposureBackendApiFactoryTest { + + @Test + void remoteConfigUsesOnlyLocalEvpProxy() { + final Config config = config(CONFIGURATION_SOURCE_REMOTE_CONFIG, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final BackendApi localApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM)).thenReturn(localApi); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertSame(localApi, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + + @Test + void agentlessPrefersLocalEvpProxyWithDirectFallback() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM)) + .thenReturn(mock(BackendApi.class)); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)) + .thenReturn(mock(BackendApi.class)); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertInstanceOf(AgentlessExposureBackendApi.class, selected); + } + + @Test + void agentlessUsesDirectIntakeWhenLocalEvpProxyIsUnavailable() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final BackendApi directApi = mock(BackendApi.class); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)).thenReturn(directApi); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertSame(directApi, selected); + } + + @Test + void agentlessUsesLocalEvpProxyWhenApiKeyIsUnavailable() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, null); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final BackendApi localApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM)).thenReturn(localApi); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertSame(localApi, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + + @Test + void agentlessDisablesDeliveryWhenNoRouteIsAvailable() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, null); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertNull(selected); + } + + @Test + void agentlessKeepsLocalRouteWhenDirectUrlIsInvalid() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final BackendApi localApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM)).thenReturn(localApi); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)) + .thenThrow(new IllegalArgumentException("invalid URL")); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertSame(localApi, selected); + } + + private static Config config(final String source, final String apiKey) { + final Config config = mock(Config.class); + when(config.getFeatureFlaggingConfigurationSource()).thenReturn(source); + when(config.getApiKey()).thenReturn(apiKey); + return config; + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java index 76b9e2602d8..cfd65cceb2c 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java @@ -1,5 +1,6 @@ package com.datadog.featureflag; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; import static java.util.Collections.singletonList; import static java.util.Collections.singletonMap; import static java.util.concurrent.TimeUnit.MILLISECONDS; @@ -13,8 +14,11 @@ import com.squareup.moshi.JsonAdapter; import com.squareup.moshi.Moshi; +import datadog.communication.BackendApiFactory; +import datadog.communication.IntakeApi; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.http.HttpRetryPolicy; import datadog.trace.agent.test.server.http.JavaTestHttpServer; import datadog.trace.agent.test.server.http.JavaTestHttpServer.HandlerApi; import datadog.trace.api.Config; @@ -57,6 +61,8 @@ class ExposureWriterTests { private static final String EXPOSURES_ENDPOINT = "/evp_proxy/api/v2/exposures"; + private static final String DIRECT_EXPOSURES_ENDPOINT = "/api/v2/exposures"; + private static final String API_KEY = "test-api-key"; private static final double TIMEOUT_SECONDS = 5; private final PollingConditions poll = new PollingConditions(TIMEOUT_SECONDS); @@ -75,7 +81,11 @@ void setUp() { JavaTestHttpServer.httpServer( s -> s.handlers( - h -> h.prefix(EXPOSURES_ENDPOINT, api -> handleExposureRequest(api, adapter)))); + h -> { + h.prefix(EXPOSURES_ENDPOINT, api -> handleExposureRequest(api, adapter)); + h.prefix( + DIRECT_EXPOSURES_ENDPOINT, api -> handleExposureRequest(api, adapter)); + })); sharedCommunicationObjects = sharedCommunicationObjects(true); } @@ -131,6 +141,43 @@ void testExposureEventWrites(String service, String env, String version) throws } } + @Test + void testAgentlessExposureEventWritesDirectlyWithApiKey() throws Exception { + Config config = mockConfig("test-service"); + when(config.getFeatureFlaggingConfigurationSource()).thenReturn(CONFIGURATION_SOURCE_AGENTLESS); + when(config.getApiKey()).thenReturn(API_KEY); + BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + IntakeApi directApi = + new IntakeApi( + HttpUrl.get(server.getAddress()).resolve("/api/v2/"), + API_KEY, + "123", + HttpRetryPolicy.Factory.NEVER_RETRY, + new OkHttpClient.Builder().build(), + false); + when(backendApiFactory.createDirectIntakeApi(datadog.trace.api.intake.Intake.EVENT_PLATFORM)) + .thenReturn(directApi); + ExposureBackendApiFactory exposureBackendApiFactory = + new ExposureBackendApiFactory(config, backendApiFactory); + List exposures = buildExposures(5); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(1 << 4, 100, MILLISECONDS, exposureBackendApiFactory, config)) { + writer.init(); + for (ExposureEvent exposure : exposures) { + writer.accept(exposure); + } + + poll.eventually( + () -> { + assertEquals(DIRECT_EXPOSURES_ENDPOINT, server.getLastRequest().getPath()); + assertEquals(API_KEY, server.getLastRequest().getHeader("dd-api-key")); + assertNull(server.getLastRequest().getHeader("X-Datadog-EVP-Subdomain")); + assertExposures(allExposures(), exposures); + }); + } + } + @Test void testLruCache() throws Exception { Config config = mockConfig("test-service"); From 085ebc53fbfbbabcb90a841806929bfdf71557d0 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 11 Aug 2026 20:21:14 -0400 Subject: [PATCH 02/10] Prepare exposure delivery before agentless activation --- .../featureflag/FeatureFlaggingSystem.java | 47 ++++++++++++++++++- .../FeatureFlaggingSystemTest.java | 6 ++- 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index 91b32ee1d64..2795c743b2e 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -41,6 +41,12 @@ public static synchronized void start(final SharedCommunicationObjects sco) { } if (CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { + try { + initializeExposureWriter(sco, config); + } catch (final RuntimeException | Error e) { + STARTED = false; + throw e; + } final FeatureFlaggingGateway.ActivationListener activationListener = () -> activateAgentless(sco, config); ACTIVATION_LISTENER = activationListener; @@ -79,8 +85,13 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final LOGGER.debug("Feature Flagging system disabled by unsupported configuration source"); return; } - final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); - initialize(configService, exposureWriter); + final ExposureWriter exposureWriter = EXPOSURE_WRITER; + if (exposureWriter == null) { + final ExposureWriter newExposureWriter = new ExposureWriterImpl(sco, config); + initialize(configService, newExposureWriter); + } else { + initializeConfigurationSource(configService, exposureWriter); + } // APM span enrichment: agent-side listener for flag-evaluation seam events. Uses the process- // wide singleton so a subsystem restart reuses the one already-registered trace interceptor @@ -93,6 +104,34 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final LOGGER.debug("Feature Flagging system started"); } + private static void initializeExposureWriter( + final SharedCommunicationObjects sco, final Config config) { + final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); + try { + exposureWriter.init(); + EXPOSURE_WRITER = exposureWriter; + } catch (final RuntimeException | Error e) { + exposureWriter.close(); + throw e; + } + } + + private static void initializeConfigurationSource( + final ConfigurationSourceService configService, final ExposureWriter exposureWriter) { + try { + configService.init(); + CONFIG_SERVICE = configService; + } catch (final RuntimeException | Error e) { + EXPOSURE_WRITER = null; + try { + exposureWriter.close(); + } finally { + configService.close(); + } + throw e; + } + } + static void initialize( final ConfigurationSourceService configService, final ExposureWriter exposureWriter) { try { @@ -167,4 +206,8 @@ public static synchronized void stop() { static boolean isAwaitingApplicationActivation() { return ACTIVATION_LISTENER != null; } + + static boolean isExposureWriterStarted() { + return EXPOSURE_WRITER != null; + } } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index d408d91da4e..8d18f752e8c 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -38,7 +38,7 @@ class FeatureFlaggingSystemTest { @WithConfig( key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, value = "http://127.0.0.1:1") - void agentlessStartWaitsForApplicationProviderActivation() { + void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivation() { SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); clearInvocations(sharedCommunicationObjects); @@ -46,7 +46,7 @@ void agentlessStartWaitsForApplicationProviderActivation() { FeatureFlaggingSystem.start(sharedCommunicationObjects); assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); - verifyNoInteractions(sharedCommunicationObjects); + assertTrue(FeatureFlaggingSystem.isExposureWriterStarted()); FeatureFlaggingGateway.activate(); @@ -56,6 +56,7 @@ void agentlessStartWaitsForApplicationProviderActivation() { } assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + assertFalse(FeatureFlaggingSystem.isExposureWriterStarted()); } @Test @@ -72,6 +73,7 @@ void agentlessStopRemovesPendingApplicationProviderActivation() { assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); FeatureFlaggingSystem.stop(); + clearInvocations(sharedCommunicationObjects); FeatureFlaggingGateway.activate(); assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); From 9992388df3e12fb18b062b9897cb68cf36e53bdc Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 11 Aug 2026 20:33:21 -0400 Subject: [PATCH 03/10] Assert lazy agentless configuration startup --- .../java/com/datadog/featureflag/FeatureFlaggingSystem.java | 4 ++++ .../com/datadog/featureflag/FeatureFlaggingSystemTest.java | 3 +++ 2 files changed, 7 insertions(+) diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index 2795c743b2e..0d1e9147af5 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -210,4 +210,8 @@ static boolean isAwaitingApplicationActivation() { static boolean isExposureWriterStarted() { return EXPOSURE_WRITER != null; } + + static boolean isConfigurationSourceStarted() { + return CONFIG_SERVICE != null; + } } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index 8d18f752e8c..40b1f59136c 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -47,16 +47,19 @@ void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivat assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); assertTrue(FeatureFlaggingSystem.isExposureWriterStarted()); + assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); FeatureFlaggingGateway.activate(); assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + assertTrue(FeatureFlaggingSystem.isConfigurationSourceStarted()); } finally { FeatureFlaggingSystem.stop(); } assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); assertFalse(FeatureFlaggingSystem.isExposureWriterStarted()); + assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); } @Test From 37165945d5324cef95a58254d39e52eda7035047 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 11 Aug 2026 20:52:57 -0400 Subject: [PATCH 04/10] log message Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../src/main/java/datadog/communication/BackendApiFactory.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index 5b7d92b29a0..0f2a9c9fac1 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java @@ -43,7 +43,7 @@ public BackendApi createDirectIntakeApi(Intake intake) { 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"); + "Agentless mode is enabled and API key is not set. Please set DD_API_KEY"); } String traceId = config.getIdGenerationStrategy().generateTraceId().toString(); return new IntakeApi( From 9ed502c5c1d5794299a57fc817ba8aea315a112f Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Thu, 13 Aug 2026 16:51:56 -0400 Subject: [PATCH 05/10] Defer direct exposure intake fallback --- .../AgentlessExposureBackendApi.java | 39 +++++++++++++-- .../ExposureBackendApiFactory.java | 17 ++++--- .../AgentlessExposureBackendApiTest.java | 48 +++++++++++++++++-- .../ExposureBackendApiFactoryTest.java | 18 ++++++- 4 files changed, 106 insertions(+), 16 deletions(-) diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java index 97c798f0e1d..28f046832ab 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java @@ -7,6 +7,7 @@ import java.io.IOException; import java.io.InputStream; import java.net.ConnectException; +import java.util.function.Supplier; import javax.annotation.Nullable; import okhttp3.RequestBody; import org.slf4j.Logger; @@ -18,12 +19,14 @@ final class AgentlessExposureBackendApi implements BackendApi { private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessExposureBackendApi.class); private final BackendApi localApi; - private final BackendApi directApi; + private final Supplier directApiSupplier; private volatile BackendApi activeApi; + private volatile boolean directApiCreationAttempted; - AgentlessExposureBackendApi(final BackendApi localApi, final BackendApi directApi) { + AgentlessExposureBackendApi( + final BackendApi localApi, final Supplier directApiSupplier) { this.localApi = localApi; - this.directApi = directApi; + this.directApiSupplier = directApiSupplier; this.activeApi = localApi; } @@ -44,12 +47,38 @@ public T post( throw exception; } - if (activeApi == localApi) { + final BackendApi directApi = getOrCreateDirectApi(); + if (directApi == null) { + throw exception; + } + return directApi.post(uri, requestBody, responseParser, requestListener, requestCompression); + } + } + + @Nullable + private BackendApi getOrCreateDirectApi() { + final BackendApi selectedApi = activeApi; + if (selectedApi != localApi) { + return selectedApi; + } + + synchronized (this) { + final BackendApi currentApi = activeApi; + if (currentApi != localApi) { + return currentApi; + } + if (directApiCreationAttempted) { + return null; + } + + final BackendApi directApi = directApiSupplier.get(); + if (directApi != null) { LOGGER.debug( "Switching Feature Flagging exposure delivery from the local EVP proxy to direct intake"); activeApi = directApi; } - return directApi.post(uri, requestBody, responseParser, requestListener, requestCompression); + directApiCreationAttempted = true; + return directApi; } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java index f0eb1fa0e3d..8381cbec078 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java @@ -40,13 +40,14 @@ BackendApi create() { return localApi; } - final BackendApi directApi = createDirectApi(); - if (localApi != null && directApi != null) { - return new AgentlessExposureBackendApi(localApi, directApi); - } if (localApi != null) { + if (hasDirectCredentials()) { + return new AgentlessExposureBackendApi(localApi, this::createDirectApi); + } return localApi; } + + final BackendApi directApi = createDirectApi(); if (directApi != null) { return directApi; } @@ -56,10 +57,14 @@ BackendApi create() { return null; } + private boolean hasDirectCredentials() { + final String apiKey = config.getApiKey(); + return apiKey != null && !apiKey.isEmpty(); + } + @Nullable private BackendApi createDirectApi() { - final String apiKey = config.getApiKey(); - if (apiKey == null || apiKey.isEmpty()) { + if (!hasDirectCredentials()) { return null; } try { diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java index 9455634ec45..c8c4f48165d 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java @@ -15,6 +15,7 @@ import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nullable; import okhttp3.MediaType; import okhttp3.RequestBody; @@ -30,13 +31,22 @@ void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throw final RecordingBackendApi local = new RecordingBackendApi(new HttpResponseException(statusCode, "rejected")); final RecordingBackendApi direct = new RecordingBackendApi(); - final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + final AtomicInteger directApiCreations = new AtomicInteger(); + final AgentlessExposureBackendApi api = + new AgentlessExposureBackendApi( + local, + () -> { + directApiCreations.incrementAndGet(); + return direct; + }); final RequestBody firstBody = requestBody("first"); final RequestBody secondBody = requestBody("second"); + assertEquals(0, directApiCreations.get()); api.post("exposures", firstBody, stream -> null, null, false); api.post("exposures", secondBody, stream -> null, null, false); + assertEquals(1, directApiCreations.get()); assertEquals(1, local.calls); assertEquals(2, direct.calls); assertSame(firstBody, local.requestBodies.get(0)); @@ -49,7 +59,7 @@ void fallsBackAfterConnectionRefusal() throws Exception { final RecordingBackendApi local = new RecordingBackendApi(new ConnectException("connection refused")); final RecordingBackendApi direct = new RecordingBackendApi(); - final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, () -> direct); api.post("exposures", requestBody("exposure"), stream -> null, null, false); @@ -73,10 +83,41 @@ void doesNotReplayConnectionReset() { assertNoDirectReplay(new SocketException("connection reset")); } + @Test + void doesNotRetryDirectApiCreationWhenFallbackIsUnavailable() { + final RecordingBackendApi local = + new RecordingBackendApi(new HttpResponseException(404, "rejected")); + final AtomicInteger directApiCreations = new AtomicInteger(); + final AgentlessExposureBackendApi api = + new AgentlessExposureBackendApi( + local, + () -> { + directApiCreations.incrementAndGet(); + return null; + }); + + assertThrows( + HttpResponseException.class, + () -> api.post("exposures", requestBody("first"), stream -> null, null, false)); + assertThrows( + HttpResponseException.class, + () -> api.post("exposures", requestBody("second"), stream -> null, null, false)); + + assertEquals(2, local.calls); + assertEquals(1, directApiCreations.get()); + } + private static void assertNoDirectReplay(final IOException failure) { final RecordingBackendApi local = new RecordingBackendApi(failure); final RecordingBackendApi direct = new RecordingBackendApi(); - final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, direct); + final AtomicInteger directApiCreations = new AtomicInteger(); + final AgentlessExposureBackendApi api = + new AgentlessExposureBackendApi( + local, + () -> { + directApiCreations.incrementAndGet(); + return direct; + }); assertThrows( IOException.class, @@ -84,6 +125,7 @@ private static void assertNoDirectReplay(final IOException failure) { assertEquals(1, local.calls); assertEquals(0, direct.calls); + assertEquals(0, directApiCreations.get()); } private static RequestBody requestBody(final String value) { diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java index 602a3219ef3..ac77a02ef92 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java @@ -43,6 +43,7 @@ void agentlessPrefersLocalEvpProxyWithDirectFallback() { final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); assertInstanceOf(AgentlessExposureBackendApi.class, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); } @Test @@ -81,7 +82,7 @@ void agentlessDisablesDeliveryWhenNoRouteIsAvailable() { } @Test - void agentlessKeepsLocalRouteWhenDirectUrlIsInvalid() { + void agentlessDoesNotValidateDirectUrlWhileLocalRouteIsAvailable() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); final BackendApi localApi = mock(BackendApi.class); @@ -91,7 +92,20 @@ void agentlessKeepsLocalRouteWhenDirectUrlIsInvalid() { final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); - assertSame(localApi, selected); + assertInstanceOf(AgentlessExposureBackendApi.class, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + + @Test + void agentlessDisablesDeliveryWhenDirectUrlIsInvalidAndLocalRouteIsUnavailable() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)) + .thenThrow(new IllegalArgumentException("invalid URL")); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertNull(selected); } private static Config config(final String source, final String apiKey) { From a83643118af6b66318dd6a16c04bd99e037592ab Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Thu, 13 Aug 2026 19:23:02 -0400 Subject: [PATCH 06/10] test(openfeature): cover direct exposure fallback branches --- .../AgentlessExposureBackendApiTest.java | 19 +++++++++++++++- .../ExposureBackendApiFactoryTest.java | 22 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java index c8c4f48165d..de34668f0f5 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java @@ -67,6 +67,23 @@ void fallsBackAfterConnectionRefusal() throws Exception { assertEquals(1, direct.calls); } + @Test + void doesNotReturnToLocalRouteAfterSwitchingToDirectIntake() throws Exception { + final RecordingBackendApi local = + new RecordingBackendApi(new ConnectException("connection refused")); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, () -> direct); + + api.post("exposures", requestBody("first"), stream -> null, null, false); + direct.failure = new IOException("direct intake failed"); + + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("second"), stream -> null, null, false)); + assertEquals(1, local.calls); + assertEquals(2, direct.calls); + } + @ParameterizedTest @ValueSource(ints = {429, 500}) void doesNotReplayAmbiguousHttpFailure(final int statusCode) { @@ -133,7 +150,7 @@ private static RequestBody requestBody(final String value) { } private static final class RecordingBackendApi implements BackendApi { - private final IOException failure; + private IOException failure; private final List requestBodies = new ArrayList<>(); private int calls; diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java index ac77a02ef92..2f966924f44 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java @@ -31,6 +31,17 @@ void remoteConfigUsesOnlyLocalEvpProxy() { verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); } + @Test + void remoteConfigDisablesDeliveryWhenLocalEvpProxyIsUnavailable() { + final Config config = config(CONFIGURATION_SOURCE_REMOTE_CONFIG, "api-key"); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertNull(selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + @Test void agentlessPrefersLocalEvpProxyWithDirectFallback() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); @@ -81,6 +92,17 @@ void agentlessDisablesDeliveryWhenNoRouteIsAvailable() { assertNull(selected); } + @Test + void agentlessDisablesDeliveryWhenApiKeyIsEmpty() { + final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, ""); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + + final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + + assertNull(selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + @Test void agentlessDoesNotValidateDirectUrlWhileLocalRouteIsAvailable() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); From 3e668a120a616771669716d84e910703917254ea Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Thu, 13 Aug 2026 21:47:06 -0400 Subject: [PATCH 07/10] fix(feature-flags): prevent lost agentless activation --- .../featureflag/FeatureFlaggingSystem.java | 30 +++++++--- .../FeatureFlaggingSystemTest.java | 59 +++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index aaa8e4a7aba..b3fd4a062f2 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -14,6 +14,11 @@ public class FeatureFlaggingSystem { + @FunctionalInterface + interface ExposureWriterFactory { + ExposureWriter create(SharedCommunicationObjects sco, Config config); + } + private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class); private static volatile ConfigurationSourceService CONFIG_SERVICE; @@ -25,11 +30,16 @@ public class FeatureFlaggingSystem { private FeatureFlaggingSystem() {} + public static void start(final SharedCommunicationObjects sco) { + start(sco, ExposureWriterImpl::new); + } + @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) { + static synchronized void start( + final SharedCommunicationObjects sco, final ExposureWriterFactory exposureWriterFactory) { if (STARTED) { LOGGER.debug("Feature Flagging system already started"); return; @@ -44,16 +54,16 @@ public static synchronized void start(final SharedCommunicationObjects sco) { } if (CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { - try { - initializeExposureWriter(sco, config); - } catch (final RuntimeException | Error e) { - STARTED = false; - throw e; - } final FeatureFlaggingGateway.ActivationListener activationListener = () -> activateAgentless(sco, config); ACTIVATION_LISTENER = activationListener; FeatureFlaggingGateway.addActivationListener(activationListener); + try { + initializeExposureWriter(sco, config, exposureWriterFactory); + } catch (final RuntimeException | Error e) { + stop(); + throw e; + } LOGGER.debug("Feature Flagging system awaiting application provider activation"); return; } @@ -134,8 +144,10 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final } private static void initializeExposureWriter( - final SharedCommunicationObjects sco, final Config config) { - final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); + final SharedCommunicationObjects sco, + final Config config, + final ExposureWriterFactory exposureWriterFactory) { + final ExposureWriter exposureWriter = exposureWriterFactory.create(sco, config); try { exposureWriter.init(); EXPOSURE_WRITER = exposureWriter; diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index dd85b6c6700..f4f47274e93 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -14,6 +14,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -32,6 +33,11 @@ import datadog.trace.api.featureflag.flagevaluation.FlagEvaluationWriter; import datadog.trace.test.junit.utils.config.WithConfig; import datadog.trace.test.util.PollingConditions; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -78,6 +84,59 @@ void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivat assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); } + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") + @WithConfig(key = FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, value = "false") + @WithConfig( + key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, + value = "http://127.0.0.1:1") + void agentlessActivationDuringEarlyWriterInitializationIsNotLost() throws Exception { + final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); + final ExposureWriter exposureWriter = mock(ExposureWriter.class); + final CountDownLatch writerInitializationStarted = new CountDownLatch(1); + final CountDownLatch activationAttempted = new CountDownLatch(1); + final CountDownLatch finishWriterInitialization = new CountDownLatch(1); + final ExecutorService executor = Executors.newFixedThreadPool(2); + doAnswer( + ignored -> { + writerInitializationStarted.countDown(); + assertTrue(finishWriterInitialization.await(5, TimeUnit.SECONDS)); + return null; + }) + .when(exposureWriter) + .init(); + + try { + final Future start = + executor.submit( + () -> + FeatureFlaggingSystem.start( + sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> exposureWriter)); + + assertTrue(writerInitializationStarted.await(5, TimeUnit.SECONDS)); + assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + + final Future activation = + executor.submit( + () -> { + activationAttempted.countDown(); + FeatureFlaggingGateway.activate(); + }); + assertTrue(activationAttempted.await(5, TimeUnit.SECONDS)); + + finishWriterInitialization.countDown(); + start.get(5, TimeUnit.SECONDS); + + new PollingConditions(TIMEOUT_SECONDS) + .eventually(() -> assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation())); + verify(exposureWriter).init(); + activation.cancel(true); + } finally { + finishWriterInitialization.countDown(); + executor.shutdownNow(); + } + } + @Test @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") @WithConfig(key = API_KEY, value = "") From 3a108403cfd9bb836617624aed12b50040ad0741 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Thu, 13 Aug 2026 22:02:38 -0400 Subject: [PATCH 08/10] test(feature-flags): cover agentless startup rollback --- .../FeatureFlaggingSystemTest.java | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index f4f47274e93..c7ebfa71f01 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -9,6 +9,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; @@ -137,6 +138,38 @@ void agentlessActivationDuringEarlyWriterInitializationIsNotLost() throws Except } } + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") + @WithConfig(key = FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, value = "false") + void agentlessExposureWriterInitializationFailureCleansUpAndAllowsRetry() { + final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); + final ExposureWriter failedWriter = mock(ExposureWriter.class); + final IllegalStateException initializationFailure = + new IllegalStateException("writer initialization failed"); + doThrow(initializationFailure).when(failedWriter).init(); + + final IllegalStateException thrown = + assertThrows( + IllegalStateException.class, + () -> + FeatureFlaggingSystem.start( + sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> failedWriter)); + + assertSame(initializationFailure, thrown); + assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + assertFalse(FeatureFlaggingSystem.isExposureWriterStarted()); + assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); + verify(failedWriter).close(); + + final ExposureWriter retryWriter = mock(ExposureWriter.class); + FeatureFlaggingSystem.start( + sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> retryWriter); + + assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + assertTrue(FeatureFlaggingSystem.isExposureWriterStarted()); + verify(retryWriter).init(); + } + @Test @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") @WithConfig(key = API_KEY, value = "") From 614ec21d2e601acceed4e590eeca07399cfd21c3 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Fri, 14 Aug 2026 00:18:56 -0400 Subject: [PATCH 09/10] fix(feature-flags): remove obsolete SpotBugs suppression --- .../java/com/datadog/featureflag/FeatureFlaggingSystem.java | 4 ---- 1 file changed, 4 deletions(-) diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index b3fd4a062f2..9759e90feed 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -34,10 +34,6 @@ public static void start(final SharedCommunicationObjects sco) { start(sco, ExposureWriterImpl::new); } - @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.") static synchronized void start( final SharedCommunicationObjects sco, final ExposureWriterFactory exposureWriterFactory) { if (STARTED) { From 6b3be03f3e7df23f4e4ff18a63c17187ab63535e Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sat, 15 Aug 2026 13:26:30 -0700 Subject: [PATCH 10/10] fix(feature-flagging): defer agentless initialization --- .../featureflag/FeatureFlaggingSystem.java | 81 +++-------- .../FeatureFlaggingSystemTest.java | 136 ++++-------------- 2 files changed, 46 insertions(+), 171 deletions(-) diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index 9759e90feed..7f105d00213 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -15,8 +15,8 @@ public class FeatureFlaggingSystem { @FunctionalInterface - interface ExposureWriterFactory { - ExposureWriter create(SharedCommunicationObjects sco, Config config); + interface SystemInitializer { + void initialize(SharedCommunicationObjects sco, Config config); } private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class); @@ -31,11 +31,11 @@ interface ExposureWriterFactory { private FeatureFlaggingSystem() {} public static void start(final SharedCommunicationObjects sco) { - start(sco, ExposureWriterImpl::new); + start(sco, FeatureFlaggingSystem::initializeSystem); } static synchronized void start( - final SharedCommunicationObjects sco, final ExposureWriterFactory exposureWriterFactory) { + final SharedCommunicationObjects sco, final SystemInitializer systemInitializer) { if (STARTED) { LOGGER.debug("Feature Flagging system already started"); return; @@ -51,39 +51,37 @@ static synchronized void start( if (CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { final FeatureFlaggingGateway.ActivationListener activationListener = - () -> activateAgentless(sco, config); + () -> activateAgentless(sco, config, systemInitializer); ACTIVATION_LISTENER = activationListener; FeatureFlaggingGateway.addActivationListener(activationListener); - try { - initializeExposureWriter(sco, config, exposureWriterFactory); - } catch (final RuntimeException | Error e) { - stop(); - throw e; - } 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; @@ -96,19 +94,8 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final LOGGER.debug("Feature Flagging system disabled by unsupported configuration source"); return; } - ExposureWriter exposureWriter = EXPOSURE_WRITER; - if (exposureWriter instanceof ExposureWriterImpl - && !((ExposureWriterImpl) exposureWriter).isSerializerThreadAlive()) { - closeQuietly(exposureWriter); - EXPOSURE_WRITER = null; - exposureWriter = null; - } - if (exposureWriter == null) { - final ExposureWriter newExposureWriter = new ExposureWriterImpl(sco, config); - initialize(configService, newExposureWriter); - } else { - initializeConfigurationSource(configService, exposureWriter); - } + final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); + initialize(configService, exposureWriter); final boolean evalCountsEnabled = config @@ -139,36 +126,6 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final LOGGER.debug("Feature Flagging system started"); } - private static void initializeExposureWriter( - final SharedCommunicationObjects sco, - final Config config, - final ExposureWriterFactory exposureWriterFactory) { - final ExposureWriter exposureWriter = exposureWriterFactory.create(sco, config); - try { - exposureWriter.init(); - EXPOSURE_WRITER = exposureWriter; - } catch (final RuntimeException | Error e) { - exposureWriter.close(); - throw e; - } - } - - private static void initializeConfigurationSource( - final ConfigurationSourceService configService, final ExposureWriter exposureWriter) { - try { - configService.init(); - CONFIG_SERVICE = configService; - } catch (final RuntimeException | Error e) { - EXPOSURE_WRITER = null; - try { - exposureWriter.close(); - } finally { - configService.close(); - } - throw e; - } - } - static void initialize( final ConfigurationSourceService configService, final ExposureWriter exposureWriter) { try { @@ -241,12 +198,6 @@ static boolean isExposureWriterStarted() { return EXPOSURE_WRITER != null; } - static boolean isExposureWriterRunning() { - final ExposureWriter exposureWriter = EXPOSURE_WRITER; - return exposureWriter instanceof ExposureWriterImpl - && ((ExposureWriterImpl) exposureWriter).isSerializerThreadAlive(); - } - static boolean isConfigurationSourceStarted() { return CONFIG_SERVICE != null; } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index c7ebfa71f01..5b88389aefc 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -1,6 +1,5 @@ package com.datadog.featureflag; -import static datadog.trace.api.config.GeneralConfig.API_KEY; import static datadog.trace.api.config.RemoteConfigConfig.REMOTE_CONFIGURATION_ENABLED; import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL; @@ -15,7 +14,6 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.clearInvocations; -import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -33,22 +31,12 @@ import datadog.trace.api.featureflag.config.FeatureFlaggingConfig; import datadog.trace.api.featureflag.flagevaluation.FlagEvaluationWriter; import datadog.trace.test.junit.utils.config.WithConfig; -import datadog.trace.test.util.PollingConditions; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; class FeatureFlaggingSystemTest { - - private static final double TIMEOUT_SECONDS = 5; - @AfterEach void resetFlagEvaluationGateway() { FeatureFlaggingSystem.stop(); @@ -61,7 +49,7 @@ void resetFlagEvaluationGateway() { @WithConfig( key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, value = "http://127.0.0.1:1") - void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivation() { + void agentlessStartWaitsForApplicationProviderActivationWithoutPreparingDelivery() { SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); clearInvocations(sharedCommunicationObjects); @@ -69,13 +57,9 @@ void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivat FeatureFlaggingSystem.start(sharedCommunicationObjects); assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); - assertTrue(FeatureFlaggingSystem.isExposureWriterStarted()); + assertFalse(FeatureFlaggingSystem.isExposureWriterStarted()); assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); - - FeatureFlaggingGateway.activate(); - - assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); - assertTrue(FeatureFlaggingSystem.isConfigurationSourceStarted()); + verifyNoInteractions(sharedCommunicationObjects); } finally { FeatureFlaggingSystem.stop(); } @@ -87,113 +71,51 @@ void agentlessStartPreparesExposureDeliveryAndWaitsForApplicationProviderActivat @Test @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") - @WithConfig(key = FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, value = "false") - @WithConfig( - key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, - value = "http://127.0.0.1:1") - void agentlessActivationDuringEarlyWriterInitializationIsNotLost() throws Exception { + void agentlessActivationInitializesSystemOnce() { final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); - final ExposureWriter exposureWriter = mock(ExposureWriter.class); - final CountDownLatch writerInitializationStarted = new CountDownLatch(1); - final CountDownLatch activationAttempted = new CountDownLatch(1); - final CountDownLatch finishWriterInitialization = new CountDownLatch(1); - final ExecutorService executor = Executors.newFixedThreadPool(2); - doAnswer( - ignored -> { - writerInitializationStarted.countDown(); - assertTrue(finishWriterInitialization.await(5, TimeUnit.SECONDS)); - return null; - }) - .when(exposureWriter) - .init(); + final FeatureFlaggingSystem.SystemInitializer systemInitializer = + mock(FeatureFlaggingSystem.SystemInitializer.class); - try { - final Future start = - executor.submit( - () -> - FeatureFlaggingSystem.start( - sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> exposureWriter)); + FeatureFlaggingSystem.start(sharedCommunicationObjects, systemInitializer); - assertTrue(writerInitializationStarted.await(5, TimeUnit.SECONDS)); - assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + verifyNoInteractions(systemInitializer); + assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); - final Future activation = - executor.submit( - () -> { - activationAttempted.countDown(); - FeatureFlaggingGateway.activate(); - }); - assertTrue(activationAttempted.await(5, TimeUnit.SECONDS)); - - finishWriterInitialization.countDown(); - start.get(5, TimeUnit.SECONDS); - - new PollingConditions(TIMEOUT_SECONDS) - .eventually(() -> assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation())); - verify(exposureWriter).init(); - activation.cancel(true); - } finally { - finishWriterInitialization.countDown(); - executor.shutdownNow(); - } + FeatureFlaggingGateway.activate(); + FeatureFlaggingGateway.activate(); + + verify(systemInitializer).initialize(eq(sharedCommunicationObjects), any(Config.class)); + assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); } @Test @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") - @WithConfig(key = FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, value = "false") - void agentlessExposureWriterInitializationFailureCleansUpAndAllowsRetry() { + void agentlessInitializationFailureCleansUpAndAllowsRetry() { final SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); - final ExposureWriter failedWriter = mock(ExposureWriter.class); + final FeatureFlaggingSystem.SystemInitializer failedInitializer = + mock(FeatureFlaggingSystem.SystemInitializer.class); final IllegalStateException initializationFailure = - new IllegalStateException("writer initialization failed"); - doThrow(initializationFailure).when(failedWriter).init(); + 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, - () -> - FeatureFlaggingSystem.start( - sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> failedWriter)); + assertThrows(IllegalStateException.class, FeatureFlaggingGateway::activate); assertSame(initializationFailure, thrown); assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); assertFalse(FeatureFlaggingSystem.isExposureWriterStarted()); assertFalse(FeatureFlaggingSystem.isConfigurationSourceStarted()); - verify(failedWriter).close(); - - final ExposureWriter retryWriter = mock(ExposureWriter.class); - FeatureFlaggingSystem.start( - sharedCommunicationObjects, (ignoredSco, ignoredConfig) -> retryWriter); - - assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); - assertTrue(FeatureFlaggingSystem.isExposureWriterStarted()); - verify(retryWriter).init(); - } - - @Test - @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") - @WithConfig(key = API_KEY, value = "") - @WithConfig(key = FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, value = "false") - @WithConfig( - key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, - value = "http://127.0.0.1:1") - void agentlessActivationRestartsExposureWriterAfterEarlyDiscoveryMiss() { - final AtomicBoolean evpProxyAvailable = new AtomicBoolean(false); - final DDAgentFeaturesDiscovery discovery = mock(DDAgentFeaturesDiscovery.class); - when(discovery.supportsEvpProxy()).thenAnswer(ignored -> evpProxyAvailable.get()); - when(discovery.getEvpProxyEndpoint()).thenReturn("/evp_proxy/"); - final SharedCommunicationObjects sharedCommunicationObjects = - sharedCommunicationObjects(discovery); - final PollingConditions poll = new PollingConditions(TIMEOUT_SECONDS); - - FeatureFlaggingSystem.start(sharedCommunicationObjects); - poll.eventually(() -> assertFalse(FeatureFlaggingSystem.isExposureWriterRunning())); - evpProxyAvailable.set(true); + final FeatureFlaggingSystem.SystemInitializer retryInitializer = + mock(FeatureFlaggingSystem.SystemInitializer.class); + FeatureFlaggingSystem.start(sharedCommunicationObjects, retryInitializer); FeatureFlaggingGateway.activate(); - poll.eventually(() -> assertTrue(FeatureFlaggingSystem.isExposureWriterRunning())); - assertTrue(FeatureFlaggingSystem.isConfigurationSourceStarted()); + verify(retryInitializer).initialize(eq(sharedCommunicationObjects), any(Config.class)); } @Test @@ -349,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 {