diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index daf35c2c351..e48974605e3 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/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); + } } diff --git a/communication/src/main/java/datadog/communication/EvpProxyApi.java b/communication/src/main/java/datadog/communication/EvpProxyApi.java index 267f4952fe2..8bb768b7e4e 100644 --- a/communication/src/main/java/datadog/communication/EvpProxyApi.java +++ b/communication/src/main/java/datadog/communication/EvpProxyApi.java @@ -105,7 +105,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-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 4cc841e737a..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 @@ -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 { 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 cce19d736ad..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 @@ -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,7 +49,7 @@ 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); @@ -57,16 +57,65 @@ void agentlessStartWaitsForApplicationProviderActivation() { 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"); diff --git a/products/feature-flagging/feature-flagging-lib/build.gradle.kts b/products/feature-flagging/feature-flagging-lib/build.gradle.kts index 6f4413b5ff7..0c00f5d7ed6 100644 --- a/products/feature-flagging/feature-flagging-lib/build.gradle.kts +++ b/products/feature-flagging/feature-flagging-lib/build.gradle.kts @@ -21,6 +21,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")) @@ -30,6 +31,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..28f046832ab --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java @@ -0,0 +1,95 @@ +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 java.util.function.Supplier; +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 Supplier directApiSupplier; + private volatile BackendApi activeApi; + private volatile boolean directApiCreationAttempted; + + AgentlessExposureBackendApi( + final BackendApi localApi, final Supplier directApiSupplier) { + this.localApi = localApi; + this.directApiSupplier = directApiSupplier; + 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; + } + + 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; + } + directApiCreationAttempted = true; + return directApi; + } + } + + 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..8381cbec078 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java @@ -0,0 +1,77 @@ +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; + } + + if (localApi != null) { + if (hasDirectCredentials()) { + return new AgentlessExposureBackendApi(localApi, this::createDirectApi); + } + return localApi; + } + + final BackendApi directApi = createDirectApi(); + 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; + } + + private boolean hasDirectCredentials() { + final String apiKey = config.getApiKey(); + return apiKey != null && !apiKey.isEmpty(); + } + + @Nullable + private BackendApi createDirectApi() { + if (!hasDirectCredentials()) { + 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 46b3fe5a486..928c2da6681 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 @@ -7,7 +7,6 @@ import datadog.common.queue.MessagePassingBlockingQueue; import datadog.common.queue.Queues; -import datadog.communication.BackendApiFactory; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; @@ -42,10 +41,19 @@ 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 ExposureSerializingHandler serializer = new ExposureSerializingHandler( - new BackendApiFactory(config, sco), + backendApiFactory, queue, flushInterval, timeUnit, @@ -95,8 +103,8 @@ private static class ExposureSerializingHandler implements Runnable { private final List buffer = new ArrayList<>(); private final Runnable errorCallback; - public ExposureSerializingHandler( - final BackendApiFactory backendApiFactory, + ExposureSerializingHandler( + final ExposureBackendApiFactory backendApiFactory, final MessagePassingBlockingQueue queue, final long flushInterval, final TimeUnit timeUnit, @@ -104,7 +112,8 @@ public ExposureSerializingHandler( final Runnable errorCallback) { this.queue = queue; this.cache = new LRUExposureCache(queue.capacity()); - this.evpPublisher = new FeatureFlagEvpPublisher<>(backendApiFactory, ExposuresRequest.class); + this.evpPublisher = + new FeatureFlagEvpPublisher<>(backendApiFactory::create, ExposuresRequest.class); this.context = context; this.lastTicks = System.nanoTime(); @@ -119,7 +128,8 @@ public ExposureSerializingHandler( public void run() { if (!evpPublisher.start()) { errorCallback.run(); - throw new IllegalArgumentException("EVP Proxy not available"); + LOGGER.warn("Feature Flagging exposure delivery is disabled"); + return; } try { runDutyCycle(); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java index c871b18165e..8a024b52f43 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEvpPublisher.java @@ -7,6 +7,7 @@ import datadog.trace.api.intake.Intake; import java.io.IOException; import java.io.UnsupportedEncodingException; +import java.util.function.Supplier; import okhttp3.MediaType; import okhttp3.RequestBody; @@ -14,8 +15,7 @@ final class FeatureFlagEvpPublisher { private static final MediaType JSON = MediaType.parse("application/json"); - private final BackendApiFactory backendApiFactory; - private final boolean responseCompression; + private final Supplier backendApiSupplier; private final JsonAdapter jsonAdapter; private BackendApi evp; @@ -27,14 +27,20 @@ final class FeatureFlagEvpPublisher { final BackendApiFactory backendApiFactory, final Class requestType, final boolean responseCompression) { - this.backendApiFactory = backendApiFactory; - this.responseCompression = responseCompression; + this( + () -> backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM, responseCompression), + requestType); + } + + FeatureFlagEvpPublisher( + final Supplier backendApiSupplier, final Class requestType) { + this.backendApiSupplier = backendApiSupplier; this.jsonAdapter = new Moshi.Builder().build().adapter(requestType); } boolean start() { if (evp == null) { - evp = backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM, responseCompression); + evp = backendApiSupplier.get(); } return evp != null; } 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..de34668f0f5 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java @@ -0,0 +1,181 @@ +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 java.util.concurrent.atomic.AtomicInteger; +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 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)); + 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); + } + + @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) { + assertNoDirectReplay(new HttpResponseException(statusCode, "ambiguous")); + } + + @Test + void doesNotReplayTimeout() { + assertNoDirectReplay(new SocketTimeoutException("timed out")); + } + + @Test + 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 AtomicInteger directApiCreations = new AtomicInteger(); + final AgentlessExposureBackendApi api = + new AgentlessExposureBackendApi( + local, + () -> { + directApiCreations.incrementAndGet(); + return direct; + }); + + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("exposure"), stream -> null, null, false)); + + assertEquals(1, local.calls); + assertEquals(0, direct.calls); + assertEquals(0, directApiCreations.get()); + } + + private static RequestBody requestBody(final String value) { + return RequestBody.create(MediaType.parse("application/json"), value); + } + + private static final class RecordingBackendApi implements BackendApi { + private 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..2f966924f44 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java @@ -0,0 +1,139 @@ +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 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"); + 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); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + } + + @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 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"); + 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(); + + 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) { + 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");