diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index e48974605e3..2ac0447fc7d 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java @@ -64,13 +64,19 @@ public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompressi responseCompression); } - /** Creates an API client that sends data through a compatible local EVP proxy. */ + /** Creates an API client that uses the specified retry policy with a compatible local 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) { + return createEvpProxyApi(intake, responseCompression, retryPolicyFactory()); + } + + /** Creates an API client that sends data through a compatible local EVP proxy. */ + public @Nullable BackendApi createEvpProxyApi( + Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) { DDAgentFeaturesDiscovery featuresDiscovery = sharedCommunicationObjects.featuresDiscovery(config); featuresDiscovery.discoverIfOutdated(); @@ -91,7 +97,7 @@ public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompressi traceId, evpProxyUrl, subdomain, - retryPolicyFactory(), + retryPolicyFactory, sharedCommunicationObjects.agentHttpClient, responseCompression); } diff --git a/communication/src/main/java/datadog/communication/IntakeApi.java b/communication/src/main/java/datadog/communication/IntakeApi.java index 3f1fe9df67f..1a6f3f91bc7 100644 --- a/communication/src/main/java/datadog/communication/IntakeApi.java +++ b/communication/src/main/java/datadog/communication/IntakeApi.java @@ -25,6 +25,7 @@ public class IntakeApi implements BackendApi { private static final String ACCEPT_ENCODING_HEADER = "Accept-Encoding"; private static final String CONTENT_ENCODING_HEADER = "Content-Encoding"; private static final String GZIP_ENCODING = "gzip"; + private static final String IDENTITY_ENCODING = "identity"; private final String apiKey; private final String traceId; @@ -73,9 +74,10 @@ public T post( requestBuilder.addHeader(CONTENT_ENCODING_HEADER, GZIP_ENCODING); } - if (responseCompression) { - requestBuilder.addHeader(ACCEPT_ENCODING_HEADER, GZIP_ENCODING); - } + // OkHttp adds Accept-Encoding: gzip when this header is absent. Always set the header so a + // caller can disable response compression on the wire. + requestBuilder.addHeader( + ACCEPT_ENCODING_HEADER, responseCompression ? GZIP_ENCODING : IDENTITY_ENCODING); Request request = requestBuilder.build(); try (okhttp3.Response response = diff --git a/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java b/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java index 7aa33de7742..726c34a7f73 100644 --- a/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java +++ b/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java @@ -4,9 +4,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.http.HttpRetryPolicy; import datadog.metrics.api.Monitoring; import datadog.trace.api.Config; import datadog.trace.api.ProtocolVersion; @@ -61,6 +63,38 @@ void advertisedEvpProxyEndpointSupportsDisabledResponseCompression() throws Exce } } + @Test + void explicitNoRetryProxyPolicyDoesNotReplayAmbiguousFailure() throws Exception { + final MockWebServer agent = new MockWebServer(); + agent.enqueue(new MockResponse().setResponseCode(500).setBody("ambiguous")); + agent.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + agent.start(); + try { + final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), sharedCommunicationObjects(discovery, agent.url("/"))); + final BackendApi api = + factory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY); + + assertNotNull(api); + assertThrows( + HttpResponseException.class, + () -> + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false)); + + assertEquals(1, agent.getRequestCount()); + } finally { + agent.shutdown(); + } + } + private static SharedCommunicationObjects sharedCommunicationObjects( final DDAgentFeaturesDiscovery discovery, final HttpUrl agentUrl) { final TestSharedCommunicationObjects sco = new TestSharedCommunicationObjects(discovery); diff --git a/communication/src/test/java/datadog/communication/IntakeApiTest.java b/communication/src/test/java/datadog/communication/IntakeApiTest.java new file mode 100644 index 00000000000..326cf21ca84 --- /dev/null +++ b/communication/src/test/java/datadog/communication/IntakeApiTest.java @@ -0,0 +1,65 @@ +package datadog.communication; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +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 IntakeApiTest { + + private static final MediaType JSON = MediaType.parse("application/json"); + + 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 requestsGzipResponseCompressionWhenEnabled() throws Exception { + assertEquals("gzip", postAndReadAcceptEncoding(true)); + } + + @Test + void requestsIdentityResponseEncodingWhenCompressionIsDisabled() throws Exception { + assertEquals("identity", postAndReadAcceptEncoding(false)); + } + + private String postAndReadAcceptEncoding(final boolean responseCompression) throws Exception { + server.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + final IntakeApi api = + new IntakeApi( + server.url("/api/v2/"), + "api-key", + "123", + HttpRetryPolicy.Factory.NEVER_RETRY, + client, + responseCompression); + + api.post("flagevaluation", RequestBody.create(JSON, "{}"), responseBody -> null, null, false); + + final RecordedRequest request = server.takeRequest(); + assertEquals("/api/v2/flagevaluation", request.getPath()); + return request.getHeader("Accept-Encoding"); + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java index 8da76d4a88e..bbec5d65229 100644 --- a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java +++ b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationEnqueueContentionBenchmark.java @@ -7,6 +7,7 @@ import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.intake.Intake; import de.thetaphi.forbiddenapis.SuppressForbidden; import java.util.HashMap; import java.util.Map; @@ -90,7 +91,13 @@ public void setUp() { final Config config = Config.get(); final BackendApiFactory factory = new BackendApiFactory(config, null); // Capacity well above what the batch-draining consumer should ever let build up. - writer = new FlagEvaluationWriterImpl(1 << 20, Long.MAX_VALUE, NANOSECONDS, factory, config); + writer = + new FlagEvaluationWriterImpl( + 1 << 20, + Long.MAX_VALUE, + NANOSECONDS, + () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false), + config); } /** diff --git a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java index e22d05bee07..fbd467b88d2 100644 --- a/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java +++ b/products/feature-flagging/feature-flagging-lib/src/jmh/java/com/datadog/featureflag/FlagEvaluationHotPathBenchmark.java @@ -6,6 +6,7 @@ import datadog.communication.BackendApiFactory; import datadog.trace.api.Config; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.intake.Intake; import java.util.HashMap; import java.util.Map; import org.openjdk.jmh.annotations.Benchmark; @@ -79,10 +80,18 @@ public void setUp() { final BackendApiFactory factory = new BackendApiFactory(config, null); final Map ddContext = new HashMap<>(); ddContext.put("service", "bench-service"); - handler = FlagEvaluationWriterImpl.createHandlerForTest(factory, ddContext); + handler = + FlagEvaluationWriterImpl.createHandlerForTest( + () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false), ddContext); // Capacity large enough that the benchmark never overflows within a measurement window. - writer = new FlagEvaluationWriterImpl(1 << 20, Long.MAX_VALUE, NANOSECONDS, factory, config); + writer = + new FlagEvaluationWriterImpl( + 1 << 20, + Long.MAX_VALUE, + NANOSECONDS, + () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false), + config); } /** 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/AgentlessFeatureFlagBackendApi.java similarity index 74% rename from products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessExposureBackendApi.java rename to products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java index 28f046832ab..769ebfd1dd1 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/AgentlessFeatureFlagBackendApi.java @@ -13,21 +13,26 @@ 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 { +/** Sends Feature Flag events through a local EVP proxy, with a safe direct intake fallback. */ +final class AgentlessFeatureFlagBackendApi implements BackendApi { - private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessExposureBackendApi.class); + private static final Logger LOGGER = + LoggerFactory.getLogger(AgentlessFeatureFlagBackendApi.class); - private final BackendApi localApi; + private final BackendApi proxyApi; private final Supplier directApiSupplier; + private final String eventType; private volatile BackendApi activeApi; private volatile boolean directApiCreationAttempted; - AgentlessExposureBackendApi( - final BackendApi localApi, final Supplier directApiSupplier) { - this.localApi = localApi; + AgentlessFeatureFlagBackendApi( + final BackendApi proxyApi, + final Supplier directApiSupplier, + final String eventType) { + this.proxyApi = proxyApi; this.directApiSupplier = directApiSupplier; - this.activeApi = localApi; + this.eventType = eventType; + this.activeApi = proxyApi; } @Override @@ -43,7 +48,7 @@ public T post( return selectedApi.post( uri, requestBody, responseParser, requestListener, requestCompression); } catch (final IOException exception) { - if (selectedApi != localApi || !isDefinitiveRejection(exception)) { + if (selectedApi != proxyApi || !isDefinitiveRejection(exception)) { throw exception; } @@ -58,13 +63,13 @@ public T post( @Nullable private BackendApi getOrCreateDirectApi() { final BackendApi selectedApi = activeApi; - if (selectedApi != localApi) { + if (selectedApi != proxyApi) { return selectedApi; } synchronized (this) { final BackendApi currentApi = activeApi; - if (currentApi != localApi) { + if (currentApi != proxyApi) { return currentApi; } if (directApiCreationAttempted) { @@ -74,7 +79,8 @@ private BackendApi getOrCreateDirectApi() { final BackendApi directApi = directApiSupplier.get(); if (directApi != null) { LOGGER.debug( - "Switching Feature Flagging exposure delivery from the local EVP proxy to direct intake"); + "Switching Feature Flagging {} delivery from the local EVP proxy to direct intake", + eventType); activeApi = directApi; } directApiCreationAttempted = true; 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 deleted file mode 100644 index 8381cbec078..00000000000 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureBackendApiFactory.java +++ /dev/null @@ -1,77 +0,0 @@ -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 928c2da6681..fcd50e5dc34 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 @@ -41,14 +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); + this( + capacity, + flushInterval, + timeUnit, + new FeatureFlagBackendApiFactory(config, sco, FeatureFlagEventType.EXPOSURE), + config); } ExposureWriterImpl( final int capacity, final long flushInterval, final TimeUnit timeUnit, - final ExposureBackendApiFactory backendApiFactory, + final FeatureFlagBackendApiFactory backendApiFactory, final Config config) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); final ExposureSerializingHandler serializer = @@ -104,7 +109,7 @@ private static class ExposureSerializingHandler implements Runnable { private final Runnable errorCallback; ExposureSerializingHandler( - final ExposureBackendApiFactory backendApiFactory, + final FeatureFlagBackendApiFactory backendApiFactory, final MessagePassingBlockingQueue queue, final long flushInterval, final TimeUnit timeUnit, @@ -182,9 +187,12 @@ protected void flushIfNecessary() { } try { evpPublisher.post(EXPOSURES_ROUTE, payload); - this.buffer.clear(); } catch (Exception e) { LOGGER.debug("Could not submit exposures", e); + } finally { + // Best-effort delivery must not retry an ambiguously accepted batch. A later definitive + // proxy rejection could otherwise replay the same exposures through direct intake. + this.buffer.clear(); } } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagBackendApiFactory.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagBackendApiFactory.java new file mode 100644 index 00000000000..0dbf9c74254 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagBackendApiFactory.java @@ -0,0 +1,100 @@ +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.communication.http.HttpRetryPolicy; +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 events. */ +final class FeatureFlagBackendApiFactory { + + private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagBackendApiFactory.class); + + private final Config config; + private final BackendApiFactory backendApiFactory; + private final FeatureFlagEventType eventType; + + FeatureFlagBackendApiFactory( + final Config config, + final SharedCommunicationObjects sharedCommunicationObjects, + final FeatureFlagEventType eventType) { + this(config, new BackendApiFactory(config, sharedCommunicationObjects), eventType); + } + + FeatureFlagBackendApiFactory( + final Config config, + final BackendApiFactory backendApiFactory, + final FeatureFlagEventType eventType) { + this.config = config; + this.backendApiFactory = backendApiFactory; + this.eventType = eventType; + } + + @Nullable + BackendApi create() { + final boolean directFallbackAvailable = + CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource()) + && hasDirectCredentials(); + final BackendApi proxyApi = + directFallbackAvailable + ? backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, + eventType.responseCompressionEnabled(), + HttpRetryPolicy.Factory.NEVER_RETRY) + : backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, eventType.responseCompressionEnabled()); + if (!CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { + if (proxyApi == null) { + LOGGER.warn( + "Feature Flagging {} delivery is disabled because the local Agent does not support the EVP proxy", + eventType.logName()); + } + return proxyApi; + } + + if (proxyApi != null) { + if (directFallbackAvailable) { + return new AgentlessFeatureFlagBackendApi( + proxyApi, this::createDirectApi, eventType.logName()); + } + return proxyApi; + } + + final BackendApi directApi = createDirectApi(); + if (directApi != null) { + return directApi; + } + + LOGGER.warn( + "Feature Flagging {} delivery is disabled because no compatible local EVP proxy or direct intake credentials are available", + eventType.logName()); + 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, eventType.responseCompressionEnabled()); + } catch (final IllegalArgumentException exception) { + LOGGER.debug( + "Cannot configure direct Feature Flagging {} delivery", eventType.logName(), exception); + return null; + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEventType.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEventType.java new file mode 100644 index 00000000000..dee8b424fa3 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagEventType.java @@ -0,0 +1,26 @@ +package com.datadog.featureflag; + +/** Defines event-specific transport behavior for Feature Flag delivery. */ +enum FeatureFlagEventType { + // Keep the established exposure transport behavior for compatibility. + EXPOSURE("exposure", true), + + // Flag evaluation writers ignore successful response bodies, so gzip negotiation adds no value. + FLAG_EVALUATION("flag evaluation", false); + + private final String logName; + private final boolean responseCompressionEnabled; + + FeatureFlagEventType(final String logName, final boolean responseCompressionEnabled) { + this.logName = logName; + this.responseCompressionEnabled = responseCompressionEnabled; + } + + String logName() { + return logName; + } + + boolean responseCompressionEnabled() { + return responseCompressionEnabled; + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java index e15666aa10a..2da8ffe0256 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FlagEvaluationWriterImpl.java @@ -6,7 +6,7 @@ import datadog.common.queue.MessagePassingBlockingQueue; import datadog.common.queue.Queues; -import datadog.communication.BackendApiFactory; +import datadog.communication.BackendApi; import datadog.communication.EvpProxy; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; @@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,8 +31,8 @@ * EVP flagevaluation writer for Java. * *

Uses the same EVP publisher path as ExposureWriterImpl, with two-tier aggregation replacing - * the single-exposure buffer. Routes to the Agent-advertised EVP proxy endpoint for - * /api/v2/flagevaluation. + * the single-exposure buffer. Uses a local EVP proxy when available. Agentless mode can use direct + * intake when no compatible local route is available. * *

Two-tier aggregation contract: Full key: (flagKey, variant, allocationKey, runtimeDefault, * errorMessage, targetingKey, canonical-context-key). Degraded key: (flagKey, variant, @@ -99,36 +100,32 @@ private static void countMetric(final String metricName, final long value, final new ConcurrentHashMap<>(); public FlagEvaluationWriterImpl(final SharedCommunicationObjects sco, final Config config) { - this(DEFAULT_CAPACITY, FLUSH_INTERVAL_SECONDS, SECONDS, sco, config); + this( + DEFAULT_CAPACITY, + FLUSH_INTERVAL_SECONDS, + SECONDS, + new FeatureFlagBackendApiFactory(config, sco, FeatureFlagEventType.FLAG_EVALUATION)::create, + config); } FlagEvaluationWriterImpl( final int capacity, final long flushInterval, final TimeUnit timeUnit, - final SharedCommunicationObjects sco, - final Config config) { - this(capacity, flushInterval, timeUnit, new BackendApiFactory(config, sco), config); - } - - /** Package-private constructor allowing a BackendApiFactory to be injected for tests. */ - FlagEvaluationWriterImpl( - final int capacity, - final long flushInterval, - final TimeUnit timeUnit, - final BackendApiFactory backendApiFactory, + final Supplier backendApiSupplier, final Config config) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); this.serializer = new FlagEvaluationSerializingHandler( - backendApiFactory, + backendApiSupplier, queue, flushInterval, timeUnit, FeatureFlagEvpContext.from(config), droppedQueueOverflow, contextTruncatedCounts, - this::close); + this::close, + FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); this.serializerThread = newAgentThread(FEATURE_FLAG_EVALUATION_PROCESSOR, serializer); } @@ -309,28 +306,7 @@ static class FlagEvaluationSerializingHandler implements Runnable { private final CountDownLatch finalFlushDone = new CountDownLatch(1); FlagEvaluationSerializingHandler( - final BackendApiFactory backendApiFactory, - final MessagePassingBlockingQueue queue, - final long flushInterval, - final TimeUnit timeUnit, - final Map context, - final AtomicLong droppedQueueOverflow, - final ConcurrentHashMap contextTruncatedCounts, - final Runnable errorCallback) { - this( - backendApiFactory, - queue, - flushInterval, - timeUnit, - context, - droppedQueueOverflow, - contextTruncatedCounts, - errorCallback, - FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); - } - - FlagEvaluationSerializingHandler( - final BackendApiFactory backendApiFactory, + final Supplier backendApiSupplier, final MessagePassingBlockingQueue queue, final long flushInterval, final TimeUnit timeUnit, @@ -342,7 +318,7 @@ static class FlagEvaluationSerializingHandler implements Runnable { this.queue = queue; this.evpPublisher = new FeatureFlagEvpPublisher<>( - backendApiFactory, FlagEvaluationPayloads.FlagEvaluationsRequest.class, false); + backendApiSupplier, FlagEvaluationPayloads.FlagEvaluationsRequest.class); this.context = context; this.droppedQueueOverflow = droppedQueueOverflow; this.contextTruncatedCounts = contextTruncatedCounts; @@ -545,16 +521,17 @@ private boolean shouldFlush() { */ static class SerializingHandlerForTest extends FlagEvaluationSerializingHandler { - SerializingHandlerForTest(final BackendApiFactory factory, final Map context) { - this(factory, context, FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); + SerializingHandlerForTest( + final Supplier backendApiSupplier, final Map context) { + this(backendApiSupplier, context, FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); } SerializingHandlerForTest( - final BackendApiFactory factory, + final Supplier backendApiSupplier, final Map context, final int payloadSizeLimitBytes) { super( - factory, + backendApiSupplier, Queues.mpscBlockingConsumerArrayQueue(DEFAULT_CAPACITY), Long.MAX_VALUE, // effectively never auto-flush TimeUnit.NANOSECONDS, @@ -618,14 +595,14 @@ int fullTierSizeForTest() { /** Factory method for test use - creates a SerializingHandlerForTest. */ static SerializingHandlerForTest createHandlerForTest( - final BackendApiFactory factory, final Map context) { - return new SerializingHandlerForTest(factory, context); + final Supplier backendApiSupplier, final Map context) { + return new SerializingHandlerForTest(backendApiSupplier, context); } static SerializingHandlerForTest createHandlerForTest( - final BackendApiFactory factory, + final Supplier backendApiSupplier, final Map context, final int payloadSizeLimitBytes) { - return new SerializingHandlerForTest(factory, context, payloadSizeLimitBytes); + return new SerializingHandlerForTest(backendApiSupplier, context, payloadSizeLimitBytes); } } 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/AgentlessFeatureFlagBackendApiTest.java similarity index 78% rename from products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessExposureBackendApiTest.java rename to products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java index de34668f0f5..117c72ba2a1 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/AgentlessFeatureFlagBackendApiTest.java @@ -16,14 +16,17 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; 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.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; -class AgentlessExposureBackendApiTest { +class AgentlessFeatureFlagBackendApiTest { @ParameterizedTest @ValueSource(ints = {403, 404, 405}) @@ -32,19 +35,20 @@ void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throw new RecordingBackendApi(new HttpResponseException(statusCode, "rejected")); final RecordingBackendApi direct = new RecordingBackendApi(); final AtomicInteger directApiCreations = new AtomicInteger(); - final AgentlessExposureBackendApi api = - new AgentlessExposureBackendApi( + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( local, () -> { directApiCreations.incrementAndGet(); return direct; - }); + }, + "flag evaluation"); 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); + api.post("flagevaluation", firstBody, stream -> null, null, false); + api.post("flagevaluation", secondBody, stream -> null, null, false); assertEquals(1, directApiCreations.get()); assertEquals(1, local.calls); @@ -54,14 +58,17 @@ void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throw assertSame(secondBody, direct.requestBodies.get(1)); } - @Test - void fallsBackAfterConnectionRefusal() throws Exception { + @ParameterizedTest + @MethodSource("featureFlagRoutes") + void fallsBackAfterConnectionRefusal(final String route, final String eventType) + throws Exception { final RecordingBackendApi local = new RecordingBackendApi(new ConnectException("connection refused")); final RecordingBackendApi direct = new RecordingBackendApi(); - final AgentlessExposureBackendApi api = new AgentlessExposureBackendApi(local, () -> direct); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi(local, () -> direct, eventType); - api.post("exposures", requestBody("exposure"), stream -> null, null, false); + api.post(route, requestBody(eventType), stream -> null, null, false); assertEquals(1, local.calls); assertEquals(1, direct.calls); @@ -72,7 +79,8 @@ 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); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi(local, () -> direct, "exposure"); api.post("exposures", requestBody("first"), stream -> null, null, false); direct.failure = new IOException("direct intake failed"); @@ -105,13 +113,14 @@ void doesNotRetryDirectApiCreationWhenFallbackIsUnavailable() { final RecordingBackendApi local = new RecordingBackendApi(new HttpResponseException(404, "rejected")); final AtomicInteger directApiCreations = new AtomicInteger(); - final AgentlessExposureBackendApi api = - new AgentlessExposureBackendApi( + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( local, () -> { directApiCreations.incrementAndGet(); return null; - }); + }, + "exposure"); assertThrows( HttpResponseException.class, @@ -128,17 +137,18 @@ 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( + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( local, () -> { directApiCreations.incrementAndGet(); return direct; - }); + }, + "flag evaluation"); assertThrows( IOException.class, - () -> api.post("exposures", requestBody("exposure"), stream -> null, null, false)); + () -> api.post("flagevaluation", requestBody("evaluation"), stream -> null, null, false)); assertEquals(1, local.calls); assertEquals(0, direct.calls); @@ -149,6 +159,11 @@ private static RequestBody requestBody(final String value) { return RequestBody.create(MediaType.parse("application/json"), value); } + private static Stream featureFlagRoutes() { + return Stream.of( + Arguments.of("exposures", "exposure"), Arguments.of("flagevaluation", "flag evaluation")); + } + private static final class RecordingBackendApi implements BackendApi { private IOException failure; private final List requestBodies = new ArrayList<>(); 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 cfd65cceb2c..daaf213ebfb 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 @@ -9,11 +9,17 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.squareup.moshi.JsonAdapter; import com.squareup.moshi.Moshi; +import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.IntakeApi; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; @@ -30,8 +36,11 @@ import datadog.trace.api.featureflag.exposure.Flag; import datadog.trace.api.featureflag.exposure.Subject; import datadog.trace.api.featureflag.exposure.Variant; +import datadog.trace.api.intake.Intake; import datadog.trace.test.util.PollingConditions; import java.io.ByteArrayInputStream; +import java.net.ConnectException; +import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; @@ -50,12 +59,15 @@ import java.util.concurrent.Future; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; +import okhttp3.RequestBody; +import okio.Buffer; import okio.Okio; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; import org.tabletest.junit.TableTest; class ExposureWriterTests { @@ -155,10 +167,11 @@ void testAgentlessExposureEventWritesDirectlyWithApiKey() throws Exception { HttpRetryPolicy.Factory.NEVER_RETRY, new OkHttpClient.Builder().build(), false); - when(backendApiFactory.createDirectIntakeApi(datadog.trace.api.intake.Intake.EVENT_PLATFORM)) + when(backendApiFactory.createDirectIntakeApi( + datadog.trace.api.intake.Intake.EVENT_PLATFORM, true)) .thenReturn(directApi); - ExposureBackendApiFactory exposureBackendApiFactory = - new ExposureBackendApiFactory(config, backendApiFactory); + FeatureFlagBackendApiFactory exposureBackendApiFactory = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FeatureFlagEventType.EXPOSURE); List exposures = buildExposures(5); try (ExposureWriterImpl writer = @@ -287,6 +300,58 @@ void testSerializationFailureDoesNotPoisonFollowingExposures() throws Exception } } + @Test + void testAmbiguousExposureBatchIsNotRetriedOrReplayedDirectly() throws Exception { + final Config config = mockConfig("test-service"); + when(config.getFeatureFlaggingConfigurationSource()).thenReturn(CONFIGURATION_SOURCE_AGENTLESS); + when(config.getApiKey()).thenReturn(API_KEY); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final BackendApi proxyApi = mock(BackendApi.class); + final BackendApi directApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, true, HttpRetryPolicy.Factory.NEVER_RETRY)) + .thenReturn(proxyApi); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, true)) + .thenReturn(directApi); + when(proxyApi.post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false))) + .thenThrow(new SocketTimeoutException("ambiguous timeout")) + .thenThrow(new ConnectException("definitive refusal")); + final FeatureFlagBackendApiFactory featureFlagBackendApiFactory = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FeatureFlagEventType.EXPOSURE); + final List exposures = buildExposures(2); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl(1 << 4, 100, MILLISECONDS, featureFlagBackendApiFactory, config)) { + writer.init(); + writer.accept(exposures.get(0)); + + poll.eventually( + () -> + verify(proxyApi) + .post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false))); + MILLISECONDS.sleep(300); + verify(proxyApi, times(1)) + .post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false)); + verify(directApi, never()) + .post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false)); + + writer.accept(exposures.get(1)); + poll.eventually( + () -> + verify(directApi) + .post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false))); + + final ArgumentCaptor directBody = ArgumentCaptor.forClass(RequestBody.class); + verify(directApi).post(eq("exposures"), directBody.capture(), any(), any(), eq(false)); + final Buffer buffer = new Buffer(); + directBody.getValue().writeTo(buffer); + final ExposuresRequest directRequest = + new Moshi.Builder().build().adapter(ExposuresRequest.class).fromJson(buffer.readUtf8()); + assertNotNull(directRequest); + assertExposures(directRequest.exposures, singletonList(exposures.get(1))); + } + } + @Test void testWriterStopsReceivingExposuresIfEvpProxyIsNotAvailable() throws Exception { SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(false); 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/FeatureFlagBackendApiFactoryTest.java similarity index 62% rename from products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureBackendApiFactoryTest.java rename to products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagBackendApiFactoryTest.java index 2f966924f44..8b117742715 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/FeatureFlagBackendApiFactoryTest.java @@ -1,5 +1,7 @@ package com.datadog.featureflag; +import static com.datadog.featureflag.FeatureFlagEventType.EXPOSURE; +import static com.datadog.featureflag.FeatureFlagEventType.FLAG_EVALUATION; 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; @@ -12,23 +14,25 @@ import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; +import datadog.communication.http.HttpRetryPolicy; import datadog.trace.api.Config; import datadog.trace.api.intake.Intake; import org.junit.jupiter.api.Test; -class ExposureBackendApiFactoryTest { +class FeatureFlagBackendApiFactoryTest { @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 proxyApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(proxyApi); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - assertSame(localApi, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + assertSame(proxyApi, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false); } @Test @@ -36,25 +40,31 @@ 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(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, EXPOSURE).create(); assertNull(selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + verify(backendApiFactory).createEvpProxyApi(Intake.EVENT_PLATFORM, true); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, true); } @Test void agentlessPrefersLocalEvpProxyWithDirectFallback() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM)) + when(backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY)) .thenReturn(mock(BackendApi.class)); - when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)) + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false)) .thenReturn(mock(BackendApi.class)); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - assertInstanceOf(AgentlessExposureBackendApi.class, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); + verify(backendApiFactory) + .createEvpProxyApi(Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false); } @Test @@ -62,9 +72,11 @@ 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); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false)) + .thenReturn(directApi); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); assertSame(directApi, selected); } @@ -73,13 +85,14 @@ void agentlessUsesDirectIntakeWhenLocalEvpProxyIsUnavailable() { 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 proxyApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(proxyApi); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - assertSame(localApi, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + assertSame(proxyApi, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false); } @Test @@ -87,7 +100,8 @@ void agentlessDisablesDeliveryWhenNoRouteIsAvailable() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, null); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); assertNull(selected); } @@ -97,35 +111,40 @@ void agentlessDisablesDeliveryWhenApiKeyIsEmpty() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, ""); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, EXPOSURE).create(); assertNull(selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, true); } @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)) + final BackendApi proxyApi = mock(BackendApi.class); + when(backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY)) + .thenReturn(proxyApi); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false)) .thenThrow(new IllegalArgumentException("invalid URL")); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - assertInstanceOf(AgentlessExposureBackendApi.class, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM); + assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); + verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false); } @Test void agentlessDisablesDeliveryWhenDirectUrlIsInvalidAndLocalRouteIsUnavailable() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM)) + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false)) .thenThrow(new IllegalArgumentException("invalid URL")); - final BackendApi selected = new ExposureBackendApiFactory(config, backendApiFactory).create(); + final BackendApi selected = + new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); assertNull(selected); } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationTestSupport.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationTestSupport.java index 4a65a81a8bf..502f0e76c69 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationTestSupport.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationTestSupport.java @@ -16,6 +16,7 @@ import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; +import datadog.trace.api.intake.Intake; import datadog.trace.api.telemetry.CoreMetricCollector; import datadog.trace.api.telemetry.MetricCollector; import java.lang.reflect.Type; @@ -25,6 +26,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import okhttp3.RequestBody; import okio.Buffer; @@ -41,6 +43,10 @@ final class FlagEvaluationTestSupport { private FlagEvaluationTestSupport() {} + static Supplier backendApiSupplier(final BackendApiFactory factory) { + return () -> factory.createBackendApi(Intake.EVENT_PLATFORM, false); + } + static void clearCoreMetrics() { CoreMetricCollector.getInstance().drain(); } @@ -98,7 +104,7 @@ static TestWriterSetup buildTestWriter(final BackendApi mockEvp) { context.put("service", "test-service"); final FlagEvaluationWriterImpl.SerializingHandlerForTest handler = - FlagEvaluationWriterImpl.createHandlerForTest(factory, context); + FlagEvaluationWriterImpl.createHandlerForTest(backendApiSupplier(factory), context); return new TestWriterSetup(handler, mockEvp, factory); } @@ -112,7 +118,8 @@ static TestWriterSetup buildTestWriter( context.put("service", "test-service"); final FlagEvaluationWriterImpl.SerializingHandlerForTest handler = - FlagEvaluationWriterImpl.createHandlerForTest(factory, context, payloadSizeLimitBytes); + FlagEvaluationWriterImpl.createHandlerForTest( + backendApiSupplier(factory), context, payloadSizeLimitBytes); return new TestWriterSetup(handler, mockEvp, factory); } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java index b354fdbb6c4..ba3d03ec6d5 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FlagEvaluationWriterImplTest.java @@ -1,6 +1,7 @@ package com.datadog.featureflag; import static com.datadog.featureflag.FlagEvaluationTestSupport.JSON_MAP; +import static com.datadog.featureflag.FlagEvaluationTestSupport.backendApiSupplier; import static com.datadog.featureflag.FlagEvaluationTestSupport.buildTestWriter; import static com.datadog.featureflag.FlagEvaluationTestSupport.cfg; import static com.datadog.featureflag.FlagEvaluationTestSupport.clearCoreMetrics; @@ -11,6 +12,7 @@ import static com.datadog.featureflag.FlagEvaluationTestSupport.metricSum; import static com.datadog.featureflag.FlagEvaluationTestSupport.repeat; import static com.datadog.featureflag.FlagEvaluationTestSupport.simpleEvent; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; import static java.util.Collections.emptyMap; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -31,13 +33,18 @@ import datadog.common.queue.Queues; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; +import datadog.communication.IntakeApi; import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.http.HttpRetryPolicy; +import datadog.trace.agent.test.server.http.JavaTestHttpServer; +import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.flagevaluation.FlagEvalEvent; import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; import datadog.trace.api.intake.Intake; import datadog.trace.api.telemetry.CoreMetricCollector; import datadog.trace.api.telemetry.MetricCollector; +import datadog.trace.test.util.PollingConditions; import java.io.IOException; import java.lang.reflect.Field; import java.util.Collection; @@ -46,6 +53,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import okhttp3.HttpUrl; +import okhttp3.OkHttpClient; import okhttp3.RequestBody; import okio.Buffer; import org.junit.jupiter.api.AfterEach; @@ -54,6 +63,10 @@ class FlagEvaluationWriterImplTest { + private static final String DIRECT_FLAG_EVALUATION_ENDPOINT = "/api/v2/flagevaluation"; + private static final String API_KEY = "test-api-key"; + private static final double TIMEOUT_SECONDS = 5; + @BeforeEach void clearCoreMetricsBefore() { clearCoreMetrics(); @@ -93,7 +106,8 @@ void startRegistersWriterAndCloseDeregistersIt() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); writer.start(); assertEquals(writer, FeatureFlaggingGateway.getFlagEvalWriter()); @@ -111,7 +125,7 @@ void queueOverflowIncrementsObservableDropCounter() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(2, 10L, TimeUnit.SECONDS, factory, cfg()); + new FlagEvaluationWriterImpl(2, 10L, TimeUnit.SECONDS, backendApiSupplier(factory), cfg()); for (int i = 0; i < 100; i++) { writer.enqueue(simpleEvent("of-flag", "on")); @@ -136,7 +150,8 @@ void enqueueAfterCloseIsDroppedAndCounted() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); writer.close(); writer.enqueue(simpleEvent("closed-flag", "on")); @@ -160,7 +175,8 @@ void enqueueDisabledDropsAndCountsAsClosedDrop() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); FeatureFlaggingGateway.setFlagEvaluationEnqueueEnabled(false); writer.enqueue(simpleEvent("disabled-flag", "on")); @@ -182,7 +198,8 @@ void closeSweepsAndCountsEventsLeftInTheQueue() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); // The worker is never started, so nothing drains these; close() must account for them rather // than leave them silently stranded. Stands in for the narrow window where a lock-free @@ -208,7 +225,8 @@ void enqueueIgnoresNullEvent() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); writer.enqueue(null); @@ -222,7 +240,8 @@ void enqueueDoesNotAggregateOnTheCallingThread() { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); writer.enqueue(simpleEvent("g2-flag", "on")); writer.enqueue(simpleEvent("g2-flag", "on")); @@ -235,7 +254,7 @@ void enqueueDoesNotAggregateOnTheCallingThread() { void handlerRunFailsFastWhenEvpProxyIsUnavailable() { final BackendApiFactory factory = mock(BackendApiFactory.class); final FlagEvaluationWriterImpl.SerializingHandlerForTest handler = - FlagEvaluationWriterImpl.createHandlerForTest(factory, context()); + FlagEvaluationWriterImpl.createHandlerForTest(backendApiSupplier(factory), context()); assertThrows(IllegalArgumentException.class, handler::run); } @@ -258,14 +277,15 @@ void flushIfNecessaryDoesNotReturnEarlyWhenOnlyQueueDropsArePending() { final AtomicLong queueDrops = new AtomicLong(1); final FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler handler = new FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler( - mock(BackendApiFactory.class), - Queues.mpscBlockingConsumerArrayQueue(16), + () -> null, + Queues.mpscBlockingConsumerArrayQueue(16), Long.MAX_VALUE, TimeUnit.NANOSECONDS, context(), queueDrops, new java.util.concurrent.ConcurrentHashMap<>(), - () -> {}); + () -> {}, + FlagEvaluationWriterImpl.FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); handler.flushIfNecessary(); @@ -288,14 +308,15 @@ void workerHandlesEmptyPolls() throws Exception { }); final FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler handler = new FlagEvaluationWriterImpl.FlagEvaluationSerializingHandler( - factory, + backendApiSupplier(factory), queue, Long.MAX_VALUE, TimeUnit.NANOSECONDS, context(), new AtomicLong(0), new java.util.concurrent.ConcurrentHashMap<>(), - () -> {}); + () -> {}, + FlagEvaluationWriterImpl.FLAG_EVALUATION_PAYLOAD_SIZE_LIMIT_BYTES); handler.run(); @@ -367,7 +388,7 @@ void finalFlushRunsWithoutTheInterruptFlagSet() throws Exception { when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = new FlagEvaluationWriterImpl( - 64, TimeUnit.DAYS.toSeconds(1), TimeUnit.SECONDS, factory, cfg()); + 64, TimeUnit.DAYS.toSeconds(1), TimeUnit.SECONDS, backendApiSupplier(factory), cfg()); writer.startForTest(); writer.enqueue(simpleEvent("interrupt-flag", "on")); @@ -396,7 +417,7 @@ void closeDrainsAndFinalFlushesQueuedEvents() throws Exception { when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = new FlagEvaluationWriterImpl( - 64, TimeUnit.DAYS.toSeconds(1), TimeUnit.SECONDS, factory, cfg()); + 64, TimeUnit.DAYS.toSeconds(1), TimeUnit.SECONDS, backendApiSupplier(factory), cfg()); writer.startForTest(); writer.enqueue(simpleEvent("shutdown-flag", "on")); @@ -416,7 +437,8 @@ void continuousTrafficFlushesWithoutWaitingForIdle() throws Exception { final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(1 << 12, 1, TimeUnit.MILLISECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 1 << 12, 1, TimeUnit.MILLISECONDS, backendApiSupplier(factory), cfg()); writer.startForTest(); boolean posted = false; @@ -650,13 +672,68 @@ void scoConstructorCreatesUsableWriter() { writer.close(); } + @Test + void agentlessWritesFlagEvaluationsDirectlyWhenLocalProxyIsUnavailable() throws Exception { + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.prefix( + DIRECT_FLAG_EVALUATION_ENDPOINT, + api -> api.getResponse().status(200).send("OK"))))) { + final Config config = cfg(); + when(config.getFeatureFlaggingConfigurationSource()) + .thenReturn(CONFIGURATION_SOURCE_AGENTLESS); + when(config.getApiKey()).thenReturn(API_KEY); + final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); + final OkHttpClient client = new OkHttpClient.Builder().build(); + try { + final IntakeApi directApi = + new IntakeApi( + HttpUrl.get(server.getAddress()).resolve("/api/v2/"), + API_KEY, + "123", + HttpRetryPolicy.Factory.NEVER_RETRY, + client, + false); + when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false)) + .thenReturn(directApi); + final FeatureFlagBackendApiFactory featureFlagBackendApiFactory = + new FeatureFlagBackendApiFactory( + config, backendApiFactory, FeatureFlagEventType.FLAG_EVALUATION); + final PollingConditions poll = new PollingConditions(TIMEOUT_SECONDS); + + try (FlagEvaluationWriterImpl writer = + new FlagEvaluationWriterImpl( + 16, 1, TimeUnit.MILLISECONDS, featureFlagBackendApiFactory::create, config)) { + writer.startForTest(); + writer.enqueue(simpleEvent("direct-flag", "on")); + + poll.eventually( + () -> { + assertNotNull(server.getLastRequest()); + assertEquals(DIRECT_FLAG_EVALUATION_ENDPOINT, server.getLastRequest().getPath()); + assertEquals(API_KEY, server.getLastRequest().getHeader("dd-api-key")); + assertNull(server.getLastRequest().getHeader("X-Datadog-EVP-Subdomain")); + assertTrue(server.getLastRequest().getBody().length > 0); + }); + } + } finally { + client.dispatcher().executorService().shutdownNow(); + client.connectionPool().evictAll(); + } + } + } + @Test void countContextTruncatedAccumulatesPerReason() { final BackendApi mockEvp = mock(BackendApi.class); final BackendApiFactory factory = mock(BackendApiFactory.class); when(factory.createBackendApi(any(), anyBoolean())).thenReturn(mockEvp); final FlagEvaluationWriterImpl writer = - new FlagEvaluationWriterImpl(16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + new FlagEvaluationWriterImpl( + 16, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); writer.countContextTruncated("field_count"); writer.countContextTruncated("field_count"); @@ -687,7 +764,7 @@ void hasCapacityForEnqueueReflectsQueueSaturationAndCountsPreQueueOverflow() { final int capacity = 2; final FlagEvaluationWriterImpl writer = new FlagEvaluationWriterImpl( - capacity, Long.MAX_VALUE, TimeUnit.NANOSECONDS, factory, cfg()); + capacity, Long.MAX_VALUE, TimeUnit.NANOSECONDS, backendApiSupplier(factory), cfg()); assertTrue(writer.hasCapacityForEnqueue());