From 612ef527e1c060eee5f0b84b17b5f4603285cfa4 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sat, 12 Sep 2026 03:50:17 +0000 Subject: [PATCH 1/2] Harden agentless Feature Flags EVP delivery Add capability-gated local discovery, safe sticky direct fallback, shared route state, send-once semantics, and bounded lifecycle handling for Java Feature Flags telemetry. Environment: Datadog workspace --- .../communication/BackendApiFactory.java | 163 +++++++- .../java/datadog/communication/EvpProxy.java | 9 + .../ddagent/DDAgentFeaturesDiscovery.java | 37 +- .../communication/BackendApiFactoryTest.java | 394 +++++++++++++++++- .../java/lang/ShutdownInstrumentation.java | 10 +- ...ceDisabledFeatureFlaggingShutdownTest.java | 45 ++ .../featureflag/FeatureFlaggingSystem.java | 6 +- .../AgentlessFeatureFlagBackendApi.java | 155 +++++-- .../featureflag/ExposureWriterImpl.java | 172 ++++++-- .../FeatureFlagBackendApiFactory.java | 103 +++-- .../featureflag/FeatureFlagRouteSelector.java | 79 ++++ .../featureflag/FlagEvaluationWriterImpl.java | 14 + .../AgentlessFeatureFlagBackendApiTest.java | 345 ++++++++++++++- .../featureflag/ExposureWriterTests.java | 217 +++++++++- .../FeatureFlagBackendApiFactoryTest.java | 90 +--- 15 files changed, 1611 insertions(+), 228 deletions(-) create mode 100644 dd-java-agent/src/test/java/datadog/trace/agent/TraceDisabledFeatureFlaggingShutdownTest.java create mode 100644 products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagRouteSelector.java diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java index c1b471ebfc8..08c2b08d897 100644 --- a/communication/src/main/java/datadog/communication/BackendApiFactory.java +++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java @@ -1,27 +1,61 @@ package datadog.communication; +import static java.util.Collections.emptyMap; +import static java.util.Collections.unmodifiableMap; + import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.communication.http.HttpRetryPolicy; import datadog.trace.api.Config; import datadog.trace.api.intake.Intake; import datadog.trace.util.throwable.FatalAgentMisconfigurationError; +import java.util.HashMap; +import java.util.Map; import javax.annotation.Nullable; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; +import okhttp3.Request; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class BackendApiFactory { private static final Logger log = LoggerFactory.getLogger(BackendApiFactory.class); + private static final int MAX_DNS_LABEL_LENGTH = 63; + private static final int MAX_DNS_HOST_LENGTH = 253; private final Config config; private final SharedCommunicationObjects sharedCommunicationObjects; + private final Map requestHeaders; + private final boolean sendOnce; public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommunicationObjects) { + this(config, sharedCommunicationObjects, emptyMap()); + } + + public BackendApiFactory( + Config config, + SharedCommunicationObjects sharedCommunicationObjects, + Map requestHeaders) { + this(config, sharedCommunicationObjects, requestHeaders, false); + } + + /** + * Creates a backend factory with per-request headers and optional send-once transport semantics. + * + *

When {@code sendOnce} is true, both the explicit HTTP retry policy and OkHttp's automatic + * connection retry are disabled. This is required for event payloads that do not carry an + * idempotency key. + */ + public BackendApiFactory( + Config config, + SharedCommunicationObjects sharedCommunicationObjects, + Map requestHeaders, + boolean sendOnce) { this.config = config; this.sharedCommunicationObjects = sharedCommunicationObjects; + this.requestHeaders = unmodifiableMap(new HashMap<>(requestHeaders)); + this.sendOnce = sendOnce; } public @Nullable BackendApi createBackendApi(Intake intake) { @@ -67,7 +101,9 @@ public BackendApi createDirectIntakeApi( apiKey, traceId, retryPolicyFactory(), - directIntakeHttpClient(sharedCommunicationObjects.getIntakeHttpClient(), followRedirects), + configureHttpClient( + directIntakeHttpClient( + sharedCommunicationObjects.getIntakeHttpClient(), followRedirects)), responseCompression); } @@ -87,11 +123,14 @@ private static HttpUrl buildDirectIntakeUrl(Intake intake, Config config) { } static HttpUrl buildEventPlatformIntakeUrl(String site) { - if (site == null || site.isEmpty()) { + if (!isValidDnsSuffix(site)) { throw new IllegalArgumentException("Invalid Datadog site"); } String expectedHost = Intake.EVENT_PLATFORM.getUrlPrefix() + "." + site; + if (expectedHost.length() > MAX_DNS_HOST_LENGTH) { + throw new IllegalArgumentException("Invalid Datadog site"); + } HttpUrl url = new HttpUrl.Builder() .scheme("https") @@ -106,6 +145,38 @@ static HttpUrl buildEventPlatformIntakeUrl(String site) { return url; } + private static boolean isValidDnsSuffix(@Nullable String site) { + if (site == null || site.isEmpty()) { + return false; + } + + int labelLength = 0; + for (int i = 0; i < site.length(); i++) { + final char character = site.charAt(i); + if (character == '.') { + if (labelLength == 0 || labelLength > MAX_DNS_LABEL_LENGTH || site.charAt(i - 1) == '-') { + return false; + } + labelLength = 0; + } else { + if ((!isAsciiLetterOrDigit(character) && character != '-') + || (labelLength == 0 && character == '-')) { + return false; + } + labelLength++; + } + } + return labelLength > 0 + && labelLength <= MAX_DNS_LABEL_LENGTH + && site.charAt(site.length() - 1) != '-'; + } + + private static boolean isAsciiLetterOrDigit(final char character) { + return (character >= 'a' && character <= 'z') + || (character >= 'A' && character <= 'Z') + || (character >= '0' && character <= '9'); + } + /** 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); @@ -119,32 +190,104 @@ static HttpUrl buildEventPlatformIntakeUrl(String site) { /** Creates an API client that sends data through a compatible local EVP proxy. */ public @Nullable BackendApi createEvpProxyApi( Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) { + return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, false, false); + } + + /** + * Creates an EVP proxy client after Agent discovery, optionally forcing a fresh discovery and + * requiring the Agent to advertise every configured request header. + * + *

The {@code forceDiscovery} form is intended for bounded unavailable-route recovery probes. + */ + public @Nullable BackendApi createEvpProxyApi( + Intake intake, + boolean responseCompression, + HttpRetryPolicy.Factory retryPolicyFactory, + boolean forceDiscovery, + boolean requireConfiguredRequestHeaders) { DDAgentFeaturesDiscovery featuresDiscovery = sharedCommunicationObjects.featuresDiscovery(config); - featuresDiscovery.discoverIfOutdated(); - if (!featuresDiscovery.supportsEvpProxy()) { - return null; + if (forceDiscovery) { + featuresDiscovery.discover(); + } else { + featuresDiscovery.discoverIfOutdated(); } String evpProxyEndpoint = featuresDiscovery.getEvpProxyEndpoint(); + if (evpProxyEndpoint != null + && requireConfiguredRequestHeaders + && !featuresDiscovery.supportsEvpProxyHeaders(requestHeaders.keySet())) { + evpProxyEndpoint = null; + } + if (evpProxyEndpoint == null) { + return null; + } + + return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, evpProxyEndpoint); + } + + /** Creates an EVP proxy client for a fixed compatibility endpoint without Agent discovery. */ + public BackendApi createEvpProxyApiForEndpoint( + Intake intake, + boolean responseCompression, + HttpRetryPolicy.Factory retryPolicyFactory, + String evpProxyEndpoint) { + return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, evpProxyEndpoint); + } + private BackendApi createEvpProxyApi( + Intake intake, + boolean responseCompression, + HttpRetryPolicy.Factory retryPolicyFactory, + String evpProxyEndpoint) { String traceId = config.getIdGenerationStrategy().generateTraceId().toString(); log.debug( "Creating EVP proxy client for {} using endpoint {} with responseCompression={}", intake, evpProxyEndpoint, responseCompression); - HttpUrl evpProxyUrl = sharedCommunicationObjects.agentUrl.resolve(evpProxyEndpoint); + HttpUrl evpProxyUrl = appendPath(sharedCommunicationObjects.agentUrl, evpProxyEndpoint); String subdomain = intake.getUrlPrefix(); return new EvpProxyApi( traceId, evpProxyUrl, subdomain, - retryPolicyFactory, - sharedCommunicationObjects.agentHttpClient, + sendOnce ? HttpRetryPolicy.Factory.NEVER_RETRY : retryPolicyFactory, + configureHttpClient(sharedCommunicationObjects.agentHttpClient), responseCompression); } - private static HttpRetryPolicy.Factory retryPolicyFactory() { - return new HttpRetryPolicy.Factory(5, 100, 2.0, true); + static HttpUrl appendPath(final HttpUrl baseUrl, final String path) { + int firstCharacter = 0; + while (firstCharacter < path.length() && path.charAt(firstCharacter) == '/') { + firstCharacter++; + } + return baseUrl.newBuilder().addPathSegments(path.substring(firstCharacter)).build(); + } + + OkHttpClient configureHttpClient(final OkHttpClient httpClient) { + if (requestHeaders.isEmpty() && !sendOnce) { + return httpClient; + } + final OkHttpClient.Builder builder = httpClient.newBuilder(); + if (sendOnce) { + builder.retryOnConnectionFailure(false); + } + if (!requestHeaders.isEmpty()) { + builder.addInterceptor( + chain -> { + final Request.Builder requestBuilder = chain.request().newBuilder(); + for (Map.Entry header : requestHeaders.entrySet()) { + requestBuilder.header(header.getKey(), header.getValue()); + } + return chain.proceed(requestBuilder.build()); + }); + } + return builder.build(); + } + + private HttpRetryPolicy.Factory retryPolicyFactory() { + return sendOnce + ? HttpRetryPolicy.Factory.NEVER_RETRY + : new HttpRetryPolicy.Factory(5, 100, 2.0, true); } } diff --git a/communication/src/main/java/datadog/communication/EvpProxy.java b/communication/src/main/java/datadog/communication/EvpProxy.java index c2453bccb25..05ce9757339 100644 --- a/communication/src/main/java/datadog/communication/EvpProxy.java +++ b/communication/src/main/java/datadog/communication/EvpProxy.java @@ -5,6 +5,15 @@ public final class EvpProxy { public static final String SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain"; + /** Identifies the SDK that produced an EVP request. */ + public static final String ORIGIN_HEADER = "DD-EVP-ORIGIN"; + + /** Identifies the version of the SDK that produced an EVP request. */ + public static final String ORIGIN_VERSION_HEADER = "DD-EVP-ORIGIN-VERSION"; + + /** Origin header value identifying this tracing library. */ + public static final String JAVA_TRACING_LIBRARY = "dd-trace-java"; + /** * Default SDK-side target for uncompressed EVP request bodies. Writers may split batches at or * below this size to keep Agent proxy requests comfortably bounded. diff --git a/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java b/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java index 16be2e84b98..c7fc60f1d1f 100644 --- a/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java +++ b/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java @@ -24,6 +24,7 @@ import java.nio.ByteBuffer; import java.util.HashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; import okhttp3.HttpUrl; @@ -98,6 +99,7 @@ private static class State { String debuggerSnapshotEndpoint; String debuggerDiagnosticsEndpoint; String evpProxyEndpoint; + Set evpProxyAllowedHeaders = emptySet(); String version; String telemetryProxyEndpoint; Set peerTags = emptySet(); @@ -156,7 +158,7 @@ private void doDiscovery(State newState) { try (Recording recording = discoveryTimer.start()) { boolean fallback = true; final Request request = - prepareRequest(agentBaseUrl.resolve("info"), emptyMap()).get().build(); + prepareRequest(appendPath(agentBaseUrl, "info"), emptyMap()).get().build(); try (Response response = client.newCall(request).execute()) { if (response.isSuccessful()) { processInfoResponseHeaders(response); @@ -206,7 +208,7 @@ private String probeTracesEndpoint(State newState, List endpoints) { try (Response response = client .newCall( - prepareRequest(agentBaseUrl.resolve(candidate), emptyMap()) + prepareRequest(appendPath(agentBaseUrl, candidate), emptyMap()) .put(msgpackRequestBodyOf(singletonList(ByteBuffer.wrap(PROBE_MESSAGE)))) .build()) .execute()) { @@ -289,6 +291,16 @@ private boolean processInfoResponse(State newState, String response) { break; } } + final Object allowedHeadersObj = map.get("evp_proxy_allowed_headers"); + if (allowedHeadersObj instanceof List) { + final Set allowedHeaders = new HashSet<>(); + for (Object header : (List) allowedHeadersObj) { + if (header instanceof String) { + allowedHeaders.add(((String) header).toLowerCase(Locale.ROOT)); + } + } + newState.evpProxyAllowedHeaders = unmodifiableSet(allowedHeaders); + } for (String endpoint : telemetryProxyEndpoints) { if (containsEndpoint(endpoints, endpoint)) { @@ -424,7 +436,7 @@ public String getEvpProxyEndpoint() { } public HttpUrl buildUrl(String endpoint) { - return agentBaseUrl.resolve(endpoint); + return appendPath(agentBaseUrl, endpoint); } public boolean supportsDataStreams() { @@ -435,6 +447,17 @@ public boolean supportsEvpProxy() { return discoveryState.evpProxyEndpoint != null; } + /** Returns whether the Agent advertises forwarding every required EVP request header. */ + public boolean supportsEvpProxyHeaders(final Iterable requiredHeaders) { + final Set allowedHeaders = discoveryState.evpProxyAllowedHeaders; + for (String requiredHeader : requiredHeaders) { + if (!allowedHeaders.contains(requiredHeader.toLowerCase(Locale.ROOT))) { + return false; + } + } + return true; + } + public boolean supportsContentEncodingHeadersWithEvpProxy() { // content encoding headers are supported in /v4 and above final String evpProxyEndpoint = discoveryState.evpProxyEndpoint; @@ -469,4 +492,12 @@ public boolean active() { public boolean supportsTelemetryProxy() { return discoveryState.telemetryProxyEndpoint != null; } + + private static HttpUrl appendPath(final HttpUrl baseUrl, final String path) { + int firstCharacter = 0; + while (firstCharacter < path.length() && path.charAt(firstCharacter) == '/') { + firstCharacter++; + } + return baseUrl.newBuilder().addPathSegments(path.substring(firstCharacter)).build(); + } } diff --git a/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java b/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java index 44690ef1c3f..5ead5496322 100644 --- a/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java +++ b/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java @@ -1,10 +1,19 @@ package datadog.communication; +import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY; +import static datadog.communication.EvpProxy.ORIGIN_HEADER; +import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V4_EVP_PROXY_ENDPOINT; +import static datadog.trace.api.config.CiVisibilityConfig.CIVISIBILITY_AGENTLESS_URL; +import static datadog.trace.api.config.GeneralConfig.API_KEY; +import static java.util.Collections.singletonMap; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; @@ -15,7 +24,12 @@ import datadog.trace.api.intake.Intake; import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashMap; import java.util.Locale; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Stream; import okhttp3.HttpUrl; import okhttp3.MediaType; import okhttp3.OkHttpClient; @@ -25,6 +39,7 @@ import okhttp3.mockwebserver.RecordedRequest; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.NullAndEmptySource; import org.junit.jupiter.params.provider.ValueSource; @@ -33,8 +48,18 @@ class BackendApiFactoryTest { private static final MediaType JSON = MediaType.parse("application/json"); @ParameterizedTest - @ValueSource(strings = {"datadoghq.com", "custom.example", "DATADOGHQ.EU"}) + @ValueSource(strings = {"datadoghq.com", "custom.example", "DATADOGHQ.EU", "mock-intake.invalid"}) void eventPlatformDirectIntakeUsesExactHttpsHost(String site) { + assertEventPlatformIntakeUrl(site); + } + + @ParameterizedTest + @MethodSource("boundaryValidSites") + void eventPlatformDirectIntakeAcceptsDnsLengthBoundaries(String site) { + assertEventPlatformIntakeUrl(site); + } + + private static void assertEventPlatformIntakeUrl(final String site) { final HttpUrl url = BackendApiFactory.buildEventPlatformIntakeUrl(site); assertEquals("https", url.scheme()); @@ -63,13 +88,58 @@ void eventPlatformDirectIntakeUsesExactHttpsHost(String site) { "data doghq.com", " datadoghq.com", "datadoghq.com ", - "datadoghq.com\\evil.example" + "datadoghq.com\\evil.example", + "-foo.bar", + "foo-.bar", + "foo.-bar", + "foo.bar-", + "foo_bar.com", + ".foo.bar", + "foo..bar", + "foo.bar." }) void eventPlatformDirectIntakeRejectsUnsafeSite(String site) { assertThrows( IllegalArgumentException.class, () -> BackendApiFactory.buildEventPlatformIntakeUrl(site)); } + @ParameterizedTest + @MethodSource("invalidLengthSites") + void eventPlatformDirectIntakeRejectsDnsLengthOverflow(String site) { + assertThrows( + IllegalArgumentException.class, () -> BackendApiFactory.buildEventPlatformIntakeUrl(site)); + } + + private static Stream boundaryValidSites() { + return Stream.of( + repeatedAsciiLabel(63) + ".invalid", + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(39)); + } + + private static Stream invalidLengthSites() { + return Stream.of( + repeatedAsciiLabel(64) + ".invalid", + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(63) + + "." + + repeatedAsciiLabel(40)); + } + + private static String repeatedAsciiLabel(final int length) { + final char[] label = new char[length]; + Arrays.fill(label, 'a'); + return new String(label); + } + @ParameterizedTest @ValueSource(ints = {301, 302, 307, 308}) void featureFlagDirectIntakeDoesNotFollowRedirects(final int statusCode) throws Exception { @@ -153,6 +223,296 @@ void advertisedEvpProxyEndpointSupportsDisabledResponseCompression() throws Exce } } + @Test + void evpProxyPreservesConfiguredAgentBasePath() throws Exception { + final MockWebServer agent = new MockWebServer(); + agent.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + agent.start(); + try { + final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), sharedCommunicationObjects(discovery, agent.url("/agent/base"))); + final BackendApi api = factory.createBackendApi(Intake.EVENT_PLATFORM, false); + + assertNotNull(api); + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false); + + assertEquals("/agent/base/evp_proxy/v4/api/v2/flagevaluation", agent.takeRequest().getPath()); + } finally { + agent.shutdown(); + } + } + + @Test + void discoveryPreservesAgentBasePathAndAcceptsCaseInsensitiveIdentityCapabilities() + throws Exception { + final MockWebServer agent = new MockWebServer(); + agent.enqueue( + new MockResponse() + .setResponseCode(200) + .setBody( + "{\"endpoints\":[\"v0.5/traces\",\"/evp_proxy/v4/\"]," + + "\"evp_proxy_allowed_headers\":[\"dd-evp-origin\"," + + "\"DD-EVP-ORIGIN-VERSION\"]}")); + agent.start(); + try { + final DDAgentFeaturesDiscovery discovery = + new DDAgentFeaturesDiscovery( + new OkHttpClient(), + Monitoring.DISABLED, + agent.url("/agent/base"), + ProtocolVersion.V0_5, + true, + false); + + discovery.discover(); + + assertEquals("/agent/base/info", agent.takeRequest().getPath()); + assertEquals(V4_EVP_PROXY_ENDPOINT, discovery.getEvpProxyEndpoint()); + assertTrue(discovery.supportsEvpProxyHeaders(sdkHeaders().keySet())); + } finally { + agent.shutdown(); + } + } + + @ParameterizedTest + @ValueSource( + strings = { + "{}", + "{\"evp_proxy_allowed_headers\":null}", + "{\"evp_proxy_allowed_headers\":[\"DD-EVP-ORIGIN\"]}", + "{\"evp_proxy_allowed_headers\":[\"DD-EVP-ORIGIN-VERSION\"]}" + }) + void discoveryRejectsMissingOrPartialIdentityCapabilities(final String capabilityJson) + throws Exception { + final MockWebServer agent = new MockWebServer(); + final String fields = capabilityJson.substring(1, capabilityJson.length() - 1); + agent.enqueue( + new MockResponse() + .setResponseCode(200) + .setBody( + "{\"endpoints\":[\"v0.5/traces\",\"evp_proxy/v4/\"]" + + (fields.isEmpty() ? "" : "," + fields) + + "}")); + agent.start(); + try { + final DDAgentFeaturesDiscovery discovery = + new DDAgentFeaturesDiscovery( + new OkHttpClient(), + Monitoring.DISABLED, + agent.url("/"), + ProtocolVersion.V0_5, + true, + false); + + discovery.discover(); + + assertEquals(V4_EVP_PROXY_ENDPOINT, discovery.getEvpProxyEndpoint()); + assertFalse(discovery.supportsEvpProxyHeaders(sdkHeaders().keySet())); + } finally { + agent.shutdown(); + } + } + + @Test + void featureFlagProxyRequestCarriesSdkIdentityWithoutDirectCredentials() throws Exception { + final MockWebServer agent = new MockWebServer(); + agent.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + agent.start(); + try { + final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), + sharedCommunicationObjects(discovery, agent.url("/")), + sdkHeaders(), + true); + final BackendApi api = + factory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true); + + assertNotNull(api); + api.post( + "exposures", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false); + + final RecordedRequest request = agent.takeRequest(); + assertEquals("/evp_proxy/v4/api/v2/exposures", request.getPath()); + assertEquals("event-platform-intake", request.getHeader("X-Datadog-EVP-Subdomain")); + assertEquals(JAVA_TRACING_LIBRARY, request.getHeader(ORIGIN_HEADER)); + assertEquals("test-version", request.getHeader(ORIGIN_VERSION_HEADER)); + assertNull(request.getHeader("DD-API-KEY")); + assertEquals(1, agent.getRequestCount()); + } finally { + agent.shutdown(); + } + } + + @Test + void directIntakeSendsConfiguredRequestHeaders() throws Exception { + final MockWebServer intake = new MockWebServer(); + intake.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + intake.start(); + try { + final Properties properties = new Properties(); + properties.setProperty(API_KEY, "api-key"); + properties.setProperty( + CIVISIBILITY_AGENTLESS_URL, intake.url("/").toString().replaceAll("/$", "")); + final Config config = Config.get(properties); + final BackendApiFactory factory = + new BackendApiFactory( + config, + sharedCommunicationObjects(new FakeFeaturesDiscovery(null), null), + singletonMap(ORIGIN_HEADER, JAVA_TRACING_LIBRARY)); + + // followRedirects=false mirrors the feature-flagging caller, so this also covers the + // interaction between the redirect-scoped client and the header interceptor. + final BackendApi api = factory.createDirectIntakeApi(Intake.API, false, false); + + assertNotNull(api); + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false); + + final RecordedRequest request = intake.takeRequest(); + assertEquals(JAVA_TRACING_LIBRARY, request.getHeader(ORIGIN_HEADER)); + assertEquals("api-key", request.getHeader("DD-API-KEY")); + } finally { + intake.shutdown(); + } + } + + @Test + void featureFlagDirectRequestCarriesCredentialsAndIdentityExactlyOnce() throws Exception { + final MockWebServer intake = new MockWebServer(); + intake.enqueue(new MockResponse().setResponseCode(500).setBody("failed")); + intake.enqueue(new MockResponse().setResponseCode(200).setBody("{}")); + intake.start(); + final OkHttpClient sharedClient = new OkHttpClient.Builder().build(); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), + sharedCommunicationObjects(new FakeFeaturesDiscovery(null), null), + sdkHeaders(), + true); + final OkHttpClient directClient = factory.configureHttpClient(sharedClient); + try { + final IntakeApi api = + new IntakeApi( + intake.url("/api/v2/"), + "api-key", + "123", + HttpRetryPolicy.Factory.NEVER_RETRY, + directClient, + false); + + assertThrows( + IOException.class, + () -> + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false)); + + final RecordedRequest request = intake.takeRequest(); + assertEquals("/api/v2/flagevaluation", request.getPath()); + assertEquals("api-key", request.getHeader("DD-API-KEY")); + assertEquals("dd-trace-java", request.getHeader("DD-EVP-ORIGIN")); + assertEquals("test-version", request.getHeader("DD-EVP-ORIGIN-VERSION")); + assertNull(request.getHeader("X-Datadog-EVP-Subdomain")); + assertEquals(1, intake.getRequestCount()); + assertEquals(false, directClient.retryOnConnectionFailure()); + } finally { + directClient.dispatcher().executorService().shutdownNow(); + directClient.connectionPool().evictAll(); + sharedClient.dispatcher().executorService().shutdownNow(); + sharedClient.connectionPool().evictAll(); + intake.shutdown(); + } + } + + @Test + void fixedV2CompatibilityRouteDoesNotDependOnDiscovery() 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(null); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), sharedCommunicationObjects(discovery, agent.url("/"))); + + final BackendApi api = + factory.createEvpProxyApiForEndpoint( + Intake.EVENT_PLATFORM, + false, + HttpRetryPolicy.Factory.NEVER_RETRY, + V2_EVP_PROXY_ENDPOINT); + + assertNotNull(api); + assertThrows( + HttpResponseException.class, + () -> + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false)); + api.post( + "flagevaluation", + RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)), + stream -> null, + null, + false); + assertEquals("/evp_proxy/v2/api/v2/flagevaluation", agent.takeRequest().getPath()); + assertEquals("/evp_proxy/v2/api/v2/flagevaluation", agent.takeRequest().getPath()); + assertEquals(2, agent.getRequestCount()); + } finally { + agent.shutdown(); + } + } + + @Test + void capabilityGatedProxyRequiresConfiguredRequestHeaders() { + final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT, false); + final BackendApiFactory factory = + new BackendApiFactory( + Config.get(), sharedCommunicationObjects(discovery, null), sdkHeaders(), true); + + assertNull( + factory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true)); + } + + @Test + void recoveryRequestForcesFreshDiscovery() { + final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(null); + final BackendApiFactory factory = + new BackendApiFactory(Config.get(), sharedCommunicationObjects(discovery, null)); + + assertNull( + factory.createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, true, true)); + assertEquals(1, discovery.forcedDiscoveries); + assertEquals(0, discovery.outdatedDiscoveries); + } + @Test void explicitNoRetryProxyPolicyDoesNotReplayAmbiguousFailure() throws Exception { final MockWebServer agent = new MockWebServer(); @@ -193,6 +553,13 @@ private static SharedCommunicationObjects sharedCommunicationObjects( return sco; } + private static Map sdkHeaders() { + final Map headers = new HashMap<>(2); + headers.put("DD-EVP-ORIGIN", "dd-trace-java"); + headers.put("DD-EVP-ORIGIN-VERSION", "test-version"); + return headers; + } + private static final class TestSharedCommunicationObjects extends SharedCommunicationObjects { private final DDAgentFeaturesDiscovery discovery; @@ -208,8 +575,16 @@ public DDAgentFeaturesDiscovery featuresDiscovery(final Config config) { private static final class FakeFeaturesDiscovery extends DDAgentFeaturesDiscovery { private final String evpProxyEndpoint; + private final boolean supportsRequestHeaders; + private int forcedDiscoveries; + private int outdatedDiscoveries; private FakeFeaturesDiscovery(final String evpProxyEndpoint) { + this(evpProxyEndpoint, true); + } + + private FakeFeaturesDiscovery( + final String evpProxyEndpoint, final boolean supportsRequestHeaders) { super( new OkHttpClient(), Monitoring.DISABLED, @@ -218,10 +593,18 @@ private FakeFeaturesDiscovery(final String evpProxyEndpoint) { true, false); this.evpProxyEndpoint = evpProxyEndpoint; + this.supportsRequestHeaders = supportsRequestHeaders; } @Override - public void discoverIfOutdated() {} + public void discover() { + forcedDiscoveries++; + } + + @Override + public void discoverIfOutdated() { + outdatedDiscoveries++; + } @Override public String getEvpProxyEndpoint() { @@ -232,5 +615,10 @@ public String getEvpProxyEndpoint() { public boolean supportsEvpProxy() { return evpProxyEndpoint != null; } + + @Override + public boolean supportsEvpProxyHeaders(final Iterable requiredHeaders) { + return supportsRequestHeaders; + } } } diff --git a/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/main/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentation.java b/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/main/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentation.java index 9aa879ad705..d51def6fa3d 100644 --- a/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/main/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentation.java +++ b/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/main/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentation.java @@ -8,6 +8,7 @@ import datadog.trace.agent.tooling.Instrumenter; import datadog.trace.agent.tooling.InstrumenterModule; import datadog.trace.bootstrap.instrumentation.shutdown.ShutdownHelper; +import java.util.Set; import net.bytebuddy.asm.Advice; /** @@ -15,7 +16,7 @@ * before the shutdown hooks are called.
*/ @AutoService(InstrumenterModule.class) -public class ShutdownInstrumentation extends InstrumenterModule.Tracing +public class ShutdownInstrumentation extends InstrumenterModule implements Instrumenter.ForBootstrap, Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { public ShutdownInstrumentation() { @@ -27,6 +28,13 @@ public String instrumentedType() { return "java.lang.Shutdown"; } + @Override + public boolean isApplicable(Set enabledSystems) { + // Agent-owned subsystems such as Feature Flagging can run while tracing is disabled. Their + // bounded final drains still depend on ShutdownHelper running before application hooks. + return true; + } + @Override public void methodAdvice(MethodTransformer transformer) { transformer.applyAdvice( diff --git a/dd-java-agent/src/test/java/datadog/trace/agent/TraceDisabledFeatureFlaggingShutdownTest.java b/dd-java-agent/src/test/java/datadog/trace/agent/TraceDisabledFeatureFlaggingShutdownTest.java new file mode 100644 index 00000000000..7dc4596c447 --- /dev/null +++ b/dd-java-agent/src/test/java/datadog/trace/agent/TraceDisabledFeatureFlaggingShutdownTest.java @@ -0,0 +1,45 @@ +package datadog.trace.agent; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.agent.test.IntegrationTestUtils; +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.TimeUnit; +import jvmbootstraptest.AgentLoadedChecker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +class TraceDisabledFeatureFlaggingShutdownTest { + @Test + @Timeout(value = 30, unit = TimeUnit.SECONDS) + void featureFlaggingIsStoppedByTheRealAgentWhenTracingIsDisabled() throws Exception { + try (ByteArrayOutputStream output = new ByteArrayOutputStream(); + PrintStream printStream = new PrintStream(output, true, StandardCharsets.UTF_8.name())) { + int exitCode = + IntegrationTestUtils.runOnSeparateJvm( + AgentLoadedChecker.class.getName(), + Arrays.asList( + "-Ddatadog.slf4j.simpleLogger.defaultLogLevel=debug", + "-Ddd.trace.enabled=false", + "-Ddd.feature.flags.enabled=true", + "-Ddd.feature.flags.configuration.source=agentless", + "-Ddd.jmxfetch.enabled=false", + "-Ddd.profiling.enabled=false", + "-Ddd.remote_config.enabled=false", + "-Ddd.telemetry.enabled=false"), + Collections.emptyList(), + Collections.emptyMap(), + printStream); + + String logs = output.toString(StandardCharsets.UTF_8.name()); + assertEquals(0, exitCode); + assertTrue(logs.contains("Shutting down agent")); + assertTrue(logs.contains("Feature Flagging system stopped")); + } + } +} 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 7f105d00213..038c409cf64 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 @@ -94,7 +94,8 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final LOGGER.debug("Feature Flagging system disabled by unsupported configuration source"); return; } - final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config, routeSelector); initialize(configService, exposureWriter); final boolean evalCountsEnabled = @@ -103,7 +104,8 @@ private static void initializeSystem(final SharedCommunicationObjects sco, final .getBoolean(FeatureFlaggingConfig.FLAGGING_EVALUATION_COUNTS_ENABLED, true); FeatureFlaggingGateway.setFlagEvaluationEnqueueEnabled(evalCountsEnabled); if (evalCountsEnabled) { - final FlagEvaluationWriterImpl evalWriter = new FlagEvaluationWriterImpl(sco, config); + final FlagEvaluationWriterImpl evalWriter = + new FlagEvaluationWriterImpl(sco, config, routeSelector); // Publish before start() so a failed start is still reachable by the rollback in stop(). FLAG_EVAL_WRITER = evalWriter; evalWriter.start(); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java index 769ebfd1dd1..bfb5d42b759 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApi.java @@ -7,32 +7,74 @@ import java.io.IOException; import java.io.InputStream; import java.net.ConnectException; +import java.util.function.LongSupplier; import java.util.function.Supplier; import javax.annotation.Nullable; import okhttp3.RequestBody; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** Sends Feature Flag events through a local EVP proxy, with a safe direct intake fallback. */ +/** Sends Feature Flag events through the process-wide Agentless EVP route selector. */ final class AgentlessFeatureFlagBackendApi implements BackendApi { private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessFeatureFlagBackendApi.class); - private final BackendApi proxyApi; + private final FeatureFlagRouteSelector routeSelector; + private final Supplier proxyApiSupplier; private final Supplier directApiSupplier; private final String eventType; - private volatile BackendApi activeApi; + private volatile BackendApi proxyApi; + private volatile BackendApi directApi; private volatile boolean directApiCreationAttempted; AgentlessFeatureFlagBackendApi( - final BackendApi proxyApi, + @Nullable final BackendApi proxyApi, + @Nullable final BackendApi directApi, + final Supplier proxyApiSupplier, final Supplier directApiSupplier, final String eventType) { + this( + proxyApi, + directApi, + proxyApiSupplier, + directApiSupplier, + eventType, + new FeatureFlagRouteSelector()); + } + + AgentlessFeatureFlagBackendApi( + @Nullable final BackendApi proxyApi, + @Nullable final BackendApi directApi, + final Supplier proxyApiSupplier, + final Supplier directApiSupplier, + final String eventType, + final LongSupplier nanoTime, + final long recoveryIntervalNanos) { + this( + proxyApi, + directApi, + proxyApiSupplier, + directApiSupplier, + eventType, + new FeatureFlagRouteSelector(nanoTime, recoveryIntervalNanos)); + } + + AgentlessFeatureFlagBackendApi( + @Nullable final BackendApi proxyApi, + @Nullable final BackendApi directApi, + final Supplier proxyApiSupplier, + final Supplier directApiSupplier, + final String eventType, + final FeatureFlagRouteSelector routeSelector) { this.proxyApi = proxyApi; + this.directApi = directApi; + this.proxyApiSupplier = proxyApiSupplier; this.directApiSupplier = directApiSupplier; this.eventType = eventType; - this.activeApi = proxyApi; + this.routeSelector = routeSelector; + this.directApiCreationAttempted = directApi != null; + routeSelector.initialize(proxyApi != null, directApi != null); } @Override @@ -43,59 +85,110 @@ public T post( @Nullable final OkHttpUtils.CustomListener requestListener, final boolean requestCompression) throws IOException { - final BackendApi selectedApi = activeApi; + final SelectedApi selected = selectApi(); try { - return selectedApi.post( + return selected.api.post( uri, requestBody, responseParser, requestListener, requestCompression); } catch (final IOException exception) { - if (selectedApi != proxyApi || !isDefinitiveRejection(exception)) { + if (!selected.local) { throw exception; } - final BackendApi directApi = getOrCreateDirectApi(); - if (directApi == null) { + final BackendApi fallbackApi = getOrCreateDirectApi(); + routeSelector.localFailure(fallbackApi != null); + if (fallbackApi == null || !isSafeToReplayDirectly(exception)) { throw exception; } - return directApi.post(uri, requestBody, responseParser, requestListener, requestCompression); + return fallbackApi.post( + uri, requestBody, responseParser, requestListener, requestCompression); } } - @Nullable - private BackendApi getOrCreateDirectApi() { - final BackendApi selectedApi = activeApi; - if (selectedApi != proxyApi) { - return selectedApi; + private SelectedApi selectApi() throws IOException { + FeatureFlagRouteSelector.Route selectedRoute = routeSelector.current(); + if (selectedRoute == FeatureFlagRouteSelector.Route.UNAVAILABLE + && routeSelector.tryBeginLocalRecovery()) { + final BackendApi recoveredProxyApi = discoverProxyApi(); + if (recoveredProxyApi != null) { + proxyApi = recoveredProxyApi; + routeSelector.localRecovered(); + } + selectedRoute = routeSelector.current(); } - synchronized (this) { - final BackendApi currentApi = activeApi; - if (currentApi != proxyApi) { - return currentApi; + if (selectedRoute == FeatureFlagRouteSelector.Route.LOCAL) { + BackendApi selectedProxyApi = proxyApi; + if (selectedProxyApi == null) { + selectedProxyApi = discoverProxyApi(); + if (selectedProxyApi != null) { + proxyApi = selectedProxyApi; + } else { + final BackendApi selectedDirectApi = getOrCreateDirectApi(); + routeSelector.localFailure(selectedDirectApi != null); + if (selectedDirectApi != null) { + return new SelectedApi(selectedDirectApi, false); + } + throw unavailableRoute(); + } } - if (directApiCreationAttempted) { - return null; + return new SelectedApi(selectedProxyApi, true); + } + + if (selectedRoute == FeatureFlagRouteSelector.Route.DIRECT) { + final BackendApi selectedDirectApi = getOrCreateDirectApi(); + if (selectedDirectApi != null) { + return new SelectedApi(selectedDirectApi, false); } + } + throw unavailableRoute(); + } - final BackendApi directApi = directApiSupplier.get(); - if (directApi != null) { - LOGGER.debug( - "Switching Feature Flagging {} delivery from the local EVP proxy to direct intake", - eventType); - activeApi = directApi; + @Nullable + private BackendApi discoverProxyApi() { + try { + return proxyApiSupplier.get(); + } catch (final RuntimeException exception) { + LOGGER.debug("Could not discover the local Feature Flagging {} route", eventType, exception); + return null; + } + } + + @Nullable + private BackendApi getOrCreateDirectApi() { + if (directApiCreationAttempted) { + return directApi; + } + synchronized (this) { + if (!directApiCreationAttempted) { + directApi = directApiSupplier.get(); + directApiCreationAttempted = true; } - directApiCreationAttempted = true; return directApi; } } - private static boolean isDefinitiveRejection(final IOException exception) { + private IOException unavailableRoute() { + return new IOException("No Feature Flagging " + eventType + " delivery route is available"); + } + + private static boolean isSafeToReplayDirectly(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 statusCode == 404 || statusCode == 405; } return false; } + + private static final class SelectedApi { + private final BackendApi api; + private final boolean local; + + private SelectedApi(final BackendApi api, final boolean local) { + this.api = api; + this.local = local; + } + } } 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 fcd50e5dc34..7ca34d4c6cb 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,6 +7,7 @@ import datadog.common.queue.MessagePassingBlockingQueue; import datadog.common.queue.Queues; +import datadog.communication.BackendApi; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; @@ -17,6 +18,9 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,14 +31,31 @@ public class ExposureWriterImpl implements ExposureWriter { private static final int DEFAULT_FLUSH_INTERVAL_IN_SECONDS = 1; private static final int FLUSH_THRESHOLD = 100; private static final String EXPOSURES_ROUTE = "exposures"; + static final long SHUTDOWN_TIMEOUT_MILLIS = SECONDS.toMillis(5); private final MessagePassingBlockingQueue queue; + private final ExposureSerializingHandler serializer; private final Thread serializerThread; + private final ReentrantReadWriteLock lifecycleLock = new ReentrantReadWriteLock(); + private final AtomicBoolean closed = new AtomicBoolean(false); + private final long shutdownTimeoutMillis; public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config config) { this(DEFAULT_CAPACITY, DEFAULT_FLUSH_INTERVAL_IN_SECONDS, SECONDS, sco, config); } + ExposureWriterImpl( + final SharedCommunicationObjects sco, + final Config config, + final FeatureFlagRouteSelector routeSelector) { + this( + DEFAULT_CAPACITY, + DEFAULT_FLUSH_INTERVAL_IN_SECONDS, + SECONDS, + new FeatureFlagBackendApiFactory(config, sco, FeatureFlagEventType.EXPOSURE, routeSelector), + config); + } + ExposureWriterImpl( final int capacity, final long flushInterval, @@ -55,35 +76,94 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con final TimeUnit timeUnit, final FeatureFlagBackendApiFactory backendApiFactory, final Config config) { + this( + capacity, + flushInterval, + timeUnit, + backendApiFactory::create, + config, + SHUTDOWN_TIMEOUT_MILLIS); + } + + ExposureWriterImpl( + final int capacity, + final long flushInterval, + final TimeUnit timeUnit, + final Supplier backendApiSupplier, + final Config config, + final long shutdownTimeoutMillis) { this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity); - final ExposureSerializingHandler serializer = + this.serializer = new ExposureSerializingHandler( - backendApiFactory, + backendApiSupplier, queue, flushInterval, timeUnit, FeatureFlagEvpContext.from(config), this::close); this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer); + this.shutdownTimeoutMillis = shutdownTimeoutMillis; } @Override public void init() { - FeatureFlaggingGateway.addExposureListener(this); - this.serializerThread.start(); + lifecycleLock.writeLock().lock(); + try { + if (closed.get()) { + return; + } + FeatureFlaggingGateway.addExposureListener(this); + this.serializerThread.start(); + } finally { + lifecycleLock.writeLock().unlock(); + } } @Override public void close() { - FeatureFlaggingGateway.removeExposureListener(this); - if (this.serializerThread.isAlive()) { - this.serializerThread.interrupt(); + final boolean workerRunning; + lifecycleLock.writeLock().lock(); + try { + if (!closed.compareAndSet(false, true)) { + return; + } + // Exclude all producers before asking the single consumer to perform its final drain. + FeatureFlaggingGateway.removeExposureListener(this); + workerRunning = this.serializerThread.isAlive(); + if (workerRunning) { + serializer.requestShutdown(); + this.serializerThread.interrupt(); + } + } finally { + lifecycleLock.writeLock().unlock(); + } + + // start() failure invokes close() from the serializer itself. It cannot join itself, and no + // final flush is possible when no backend route was created. + if (!workerRunning || Thread.currentThread() == this.serializerThread) { + return; + } + try { + // Bound application shutdown even if the final best-effort network request does not return. + this.serializerThread.join(shutdownTimeoutMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } } @Override public void accept(final ExposureEvent event) { - queue.offer(event); + // Multiple producers may still enqueue concurrently under the read lock. close() takes the + // write lock, so every accepted event is in the queue before the worker's final drain begins, + // while stale CopyOnWriteArrayList dispatch snapshots are rejected after shutdown starts. + lifecycleLock.readLock().lock(); + try { + if (!closed.get()) { + queue.offer(event); + } + } finally { + lifecycleLock.readLock().unlock(); + } } @VisibleForTesting @@ -107,9 +187,10 @@ private static class ExposureSerializingHandler implements Runnable { private final List buffer = new ArrayList<>(); private final Runnable errorCallback; + private final AtomicBoolean shutdownRequested = new AtomicBoolean(false); ExposureSerializingHandler( - final FeatureFlagBackendApiFactory backendApiFactory, + final Supplier backendApiSupplier, final MessagePassingBlockingQueue queue, final long flushInterval, final TimeUnit timeUnit, @@ -117,8 +198,7 @@ private static class ExposureSerializingHandler implements Runnable { final Runnable errorCallback) { this.queue = queue; this.cache = new LRUExposureCache(queue.capacity()); - this.evpPublisher = - new FeatureFlagEvpPublisher<>(backendApiFactory::create, ExposuresRequest.class); + this.evpPublisher = new FeatureFlagEvpPublisher<>(backendApiSupplier, ExposuresRequest.class); this.context = context; this.lastTicks = System.nanoTime(); @@ -129,6 +209,10 @@ private static class ExposureSerializingHandler implements Runnable { LOGGER.debug("starting exposure serializer"); } + void requestShutdown() { + shutdownRequested.set(true); + } + @Override public void run() { if (!evpPublisher.start()) { @@ -140,13 +224,25 @@ public void run() { runDutyCycle(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + } finally { + // close() interrupts this thread to wake queue.poll(). OkHttp fails fast when the calling + // thread is interrupted, so clear the flag for the final best-effort drain and restore it + // only after the final send-once attempt has completed. + final boolean wasInterrupted = Thread.interrupted(); + try { + drainAndFlush(); + } finally { + if (wasInterrupted) { + Thread.currentThread().interrupt(); + } + } } LOGGER.debug("exposure processor worker exited. submitting exposures stopped."); } private void runDutyCycle() throws InterruptedException { final Thread thread = Thread.currentThread(); - while (!thread.isInterrupted()) { + while (!thread.isInterrupted() && !shutdownRequested.get()) { ExposureEvent event; while ((event = queue.poll(100, TimeUnit.MILLISECONDS)) != null) { if (addToBuffer(event)) { @@ -162,6 +258,14 @@ private void consumeBatch() { queue.drain(this::addToBuffer, queue.size()); } + private void drainAndFlush() { + ExposureEvent event; + while ((event = queue.poll()) != null) { + addToBuffer(event); + } + flush(); + } + /** Adds an element to the buffer taking care of duplicated exposures thanks to the LRU cache */ private boolean addToBuffer(final ExposureEvent event) { if (cache.add(event)) { @@ -172,35 +276,39 @@ private boolean addToBuffer(final ExposureEvent event) { } protected void flushIfNecessary() { + if (!buffer.isEmpty() && shouldFlush()) { + flush(); + } + } + + private void flush() { if (buffer.isEmpty()) { return; } - if (shouldFlush()) { - final byte[] payload; - try { - final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer); - payload = evpPublisher.serialize(exposures); - } catch (RuntimeException e) { - LOGGER.error(EXCLUDE_TELEMETRY, "Could not serialize exposures; dropping batch", e); - this.buffer.clear(); - return; - } - try { - evpPublisher.post(EXPOSURES_ROUTE, payload); - } 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(); - } + final byte[] payload; + try { + final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer); + payload = evpPublisher.serialize(exposures); + } catch (RuntimeException e) { + LOGGER.error(EXCLUDE_TELEMETRY, "Could not serialize exposures; dropping batch", e); + this.buffer.clear(); + return; + } + try { + evpPublisher.post(EXPOSURES_ROUTE, payload); + } 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(); } } private boolean shouldFlush() { long nanoTime = System.nanoTime(); long ticks = nanoTime - lastTicks; - if (ticks > ticksRequiredToFlush || queue.size() >= FLUSH_THRESHOLD) { + if (ticks > ticksRequiredToFlush || buffer.size() >= FLUSH_THRESHOLD) { lastTicks = nanoTime; return true; } 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 index 451b903ecc7..7c41efcc551 100644 --- 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 @@ -1,13 +1,21 @@ package com.datadog.featureflag; +import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY; +import static datadog.communication.EvpProxy.ORIGIN_HEADER; +import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; +import static java.util.Collections.unmodifiableMap; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.ddagent.TracerVersion; import datadog.communication.http.HttpRetryPolicy; import datadog.trace.api.Config; import datadog.trace.api.intake.Intake; +import java.util.HashMap; +import java.util.Map; import javax.annotation.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -16,66 +24,95 @@ final class FeatureFlagBackendApiFactory { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagBackendApiFactory.class); + private static final Map REQUEST_HEADERS = requestHeaders(); private final Config config; private final BackendApiFactory backendApiFactory; private final FeatureFlagEventType eventType; + private final FeatureFlagRouteSelector routeSelector; FeatureFlagBackendApiFactory( final Config config, final SharedCommunicationObjects sharedCommunicationObjects, final FeatureFlagEventType eventType) { - this(config, new BackendApiFactory(config, sharedCommunicationObjects), eventType); + this(config, sharedCommunicationObjects, eventType, new FeatureFlagRouteSelector()); + } + + FeatureFlagBackendApiFactory( + final Config config, + final SharedCommunicationObjects sharedCommunicationObjects, + final FeatureFlagEventType eventType, + final FeatureFlagRouteSelector routeSelector) { + this( + config, + new BackendApiFactory(config, sharedCommunicationObjects, REQUEST_HEADERS, true), + eventType, + routeSelector); } FeatureFlagBackendApiFactory( final Config config, final BackendApiFactory backendApiFactory, final FeatureFlagEventType eventType) { + this(config, backendApiFactory, eventType, new FeatureFlagRouteSelector()); + } + + FeatureFlagBackendApiFactory( + final Config config, + final BackendApiFactory backendApiFactory, + final FeatureFlagEventType eventType, + final FeatureFlagRouteSelector routeSelector) { this.config = config; this.backendApiFactory = backendApiFactory; this.eventType = eventType; + this.routeSelector = routeSelector; } @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 boolean agentless = + CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource()); + if (!agentless) { + // Preserve the historical Agent-backed client: fixed EVP v2, no /info discovery, no direct + // credentials, and no Agentless route state. + return backendApiFactory.createEvpProxyApiForEndpoint( + Intake.EVENT_PLATFORM, + eventType.responseCompressionEnabled(), + HttpRetryPolicy.Factory.NEVER_RETRY, + V2_EVP_PROXY_ENDPOINT); } + final BackendApi proxyApi = createProxyApi(false); final BackendApi directApi = createDirectApi(); - if (directApi != null) { - return directApi; + if (proxyApi == null && directApi == null) { + LOGGER.warn( + "Feature Flagging {} delivery is waiting for a compatible local EVP proxy because direct intake credentials are unavailable", + eventType.logName()); } + return new AgentlessFeatureFlagBackendApi( + proxyApi, + directApi, + () -> createProxyApi(true), + this::createDirectApi, + eventType.logName(), + routeSelector); + } + + @Nullable + private BackendApi createProxyApi(final boolean forceDiscovery) { + return backendApiFactory.createEvpProxyApi( + Intake.EVENT_PLATFORM, + eventType.responseCompressionEnabled(), + HttpRetryPolicy.Factory.NEVER_RETRY, + forceDiscovery, + true); + } - LOGGER.warn( - "Feature Flagging {} delivery is disabled because no compatible local EVP proxy or direct intake credentials are available", - eventType.logName()); - return null; + private static Map requestHeaders() { + final Map headers = new HashMap<>(2); + headers.put(ORIGIN_HEADER, JAVA_TRACING_LIBRARY); + headers.put(ORIGIN_VERSION_HEADER, TracerVersion.TRACER_VERSION); + return unmodifiableMap(headers); } private boolean hasDirectCredentials() { diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagRouteSelector.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagRouteSelector.java new file mode 100644 index 00000000000..da3a2da2125 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagRouteSelector.java @@ -0,0 +1,79 @@ +package com.datadog.featureflag; + +import java.util.concurrent.TimeUnit; +import java.util.function.LongSupplier; + +/** Process-wide route state shared by all Feature Flagging event writers. */ +final class FeatureFlagRouteSelector { + + static final long DEFAULT_RECOVERY_INTERVAL_NANOS = TimeUnit.MINUTES.toNanos(1); + + enum Route { + UNINITIALIZED, + LOCAL, + DIRECT, + UNAVAILABLE + } + + private final LongSupplier nanoTime; + private final long recoveryIntervalNanos; + private volatile Route route = Route.UNINITIALIZED; + private volatile long nextLocalDiscoveryNanos; + + FeatureFlagRouteSelector() { + this(System::nanoTime, DEFAULT_RECOVERY_INTERVAL_NANOS); + } + + FeatureFlagRouteSelector(final LongSupplier nanoTime, final long recoveryIntervalNanos) { + this.nanoTime = nanoTime; + this.recoveryIntervalNanos = recoveryIntervalNanos; + } + + synchronized Route initialize(final boolean localAvailable, final boolean directAvailable) { + if (route == Route.UNINITIALIZED || route == Route.UNAVAILABLE) { + if (localAvailable) { + route = Route.LOCAL; + } else if (directAvailable) { + route = Route.DIRECT; + } else { + route = Route.UNAVAILABLE; + scheduleLocalDiscovery(); + } + } + return route; + } + + Route current() { + return route; + } + + synchronized Route localFailure(final boolean directAvailable) { + if (route == Route.LOCAL) { + if (directAvailable) { + route = Route.DIRECT; + } else { + route = Route.UNAVAILABLE; + scheduleLocalDiscovery(); + } + } + return route; + } + + synchronized boolean tryBeginLocalRecovery() { + if (route != Route.UNAVAILABLE || nanoTime.getAsLong() - nextLocalDiscoveryNanos < 0) { + return false; + } + scheduleLocalDiscovery(); + return true; + } + + synchronized void localRecovered() { + if (route == Route.UNAVAILABLE) { + route = Route.LOCAL; + } + } + + private void scheduleLocalDiscovery() { + nextLocalDiscoveryNanos = nanoTime.getAsLong() + recoveryIntervalNanos; + } +} 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 2da8ffe0256..4969376771c 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 @@ -108,6 +108,20 @@ public FlagEvaluationWriterImpl(final SharedCommunicationObjects sco, final Conf config); } + FlagEvaluationWriterImpl( + final SharedCommunicationObjects sco, + final Config config, + final FeatureFlagRouteSelector routeSelector) { + this( + DEFAULT_CAPACITY, + FLUSH_INTERVAL_SECONDS, + SECONDS, + new FeatureFlagBackendApiFactory( + config, sco, FeatureFlagEventType.FLAG_EVALUATION, routeSelector) + ::create, + config); + } + FlagEvaluationWriterImpl( final int capacity, final long flushInterval, diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java index 117c72ba2a1..012ba26ea01 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java @@ -3,6 +3,7 @@ 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 static org.junit.jupiter.api.Assertions.assertTrue; import datadog.communication.BackendApi; import datadog.communication.HttpResponseException; @@ -15,7 +16,11 @@ import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; import javax.annotation.Nullable; import okhttp3.MediaType; @@ -29,7 +34,7 @@ class AgentlessFeatureFlagBackendApiTest { @ParameterizedTest - @ValueSource(ints = {403, 404, 405}) + @ValueSource(ints = {404, 405}) void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throws Exception { final RecordingBackendApi local = new RecordingBackendApi(new HttpResponseException(statusCode, "rejected")); @@ -38,6 +43,8 @@ void replaysRejectedBatchDirectlyAndKeepsDirectRoute(final int statusCode) throw final AgentlessFeatureFlagBackendApi api = new AgentlessFeatureFlagBackendApi( local, + null, + () -> local, () -> { directApiCreations.incrementAndGet(); return direct; @@ -66,7 +73,7 @@ void fallsBackAfterConnectionRefusal(final String route, final String eventType) new RecordingBackendApi(new ConnectException("connection refused")); final RecordingBackendApi direct = new RecordingBackendApi(); final AgentlessFeatureFlagBackendApi api = - new AgentlessFeatureFlagBackendApi(local, () -> direct, eventType); + new AgentlessFeatureFlagBackendApi(local, null, () -> local, () -> direct, eventType); api.post(route, requestBody(eventType), stream -> null, null, false); @@ -75,37 +82,154 @@ void fallsBackAfterConnectionRefusal(final String route, final String eventType) } @Test - void doesNotReturnToLocalRouteAfterSwitchingToDirectIntake() throws Exception { - final RecordingBackendApi local = + void keepsDirectRouteStickyAfterLocalFailure() throws Exception { + final AtomicLong clock = new AtomicLong(); + final RecordingBackendApi unavailableLocal = new RecordingBackendApi(new ConnectException("connection refused")); + final RecordingBackendApi recoveredLocal = new RecordingBackendApi(); final RecordingBackendApi direct = new RecordingBackendApi(); final AgentlessFeatureFlagBackendApi api = - new AgentlessFeatureFlagBackendApi(local, () -> direct, "exposure"); + new AgentlessFeatureFlagBackendApi( + unavailableLocal, null, () -> recoveredLocal, () -> direct, "exposure", clock::get, 10); api.post("exposures", requestBody("first"), stream -> null, null, false); - direct.failure = new IOException("direct intake failed"); + clock.set(9); + api.post("exposures", requestBody("second"), stream -> null, null, false); + clock.set(10); + api.post("exposures", requestBody("third"), stream -> null, null, false); - assertThrows( - IOException.class, - () -> api.post("exposures", requestBody("second"), stream -> null, null, false)); - assertEquals(1, local.calls); - assertEquals(2, direct.calls); + assertEquals(1, unavailableLocal.calls); + assertEquals(3, direct.calls); + assertEquals(0, recoveredLocal.calls); } @ParameterizedTest - @ValueSource(ints = {429, 500}) - void doesNotReplayAmbiguousHttpFailure(final int statusCode) { - assertNoDirectReplay(new HttpResponseException(statusCode, "ambiguous")); + @ValueSource(ints = {403, 429, 500}) + void doesNotReplayAmbiguousHttpFailureButSwitchesFutureBatches(final int statusCode) + throws Exception { + assertNoSameBatchReplayButUsesDirectForNext(new HttpResponseException(statusCode, "ambiguous")); + } + + @Test + void doesNotReplayTimeoutButSwitchesFutureBatches() throws Exception { + assertNoSameBatchReplayButUsesDirectForNext(new SocketTimeoutException("timed out")); + } + + @Test + void doesNotReplayConnectionResetButSwitchesFutureBatches() throws Exception { + assertNoSameBatchReplayButUsesDirectForNext(new SocketException("connection reset")); } @Test - void doesNotReplayTimeout() { - assertNoDirectReplay(new SocketTimeoutException("timed out")); + void startsDirectAndNeverProbesLocal() throws Exception { + final AtomicLong clock = new AtomicLong(); + final RecordingBackendApi direct = new RecordingBackendApi(); + final RecordingBackendApi recoveredLocal = new RecordingBackendApi(); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + direct, + () -> { + proxyApiCreations.incrementAndGet(); + return recoveredLocal; + }, + () -> direct, + "exposure", + clock::get, + 10); + + api.post("exposures", requestBody("first"), stream -> null, null, false); + clock.set(9); + api.post("exposures", requestBody("second"), stream -> null, null, false); + clock.set(10); + api.post("exposures", requestBody("third"), stream -> null, null, false); + + assertEquals(0, proxyApiCreations.get()); + assertEquals(3, direct.calls); + assertEquals(0, recoveredLocal.calls); + } + + @Test + void concurrentSendersDoNotBlockOnOrDuplicateARecoveryProbe() throws Exception { + final AtomicLong clock = new AtomicLong(10); + final RecordingBackendApi recoveredLocal = new RecordingBackendApi(); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final CountDownLatch probeStarted = new CountDownLatch(1); + final CountDownLatch releaseProbe = new CountDownLatch(1); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + null, + () -> { + proxyApiCreations.incrementAndGet(); + probeStarted.countDown(); + try { + assertTrue(releaseProbe.await(5, TimeUnit.SECONDS)); + } catch (final InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new AssertionError(exception); + } + return recoveredLocal; + }, + () -> null, + "flag evaluation", + clock::get, + 10); + + // Construction schedules the first probe for t=20. + clock.set(20); + final CompletableFuture recoveringPost = + CompletableFuture.runAsync( + () -> { + try { + api.post("flagevaluation", requestBody("probe"), stream -> null, null, false); + } catch (final IOException exception) { + throw new AssertionError(exception); + } + }); + assertTrue(probeStarted.await(5, TimeUnit.SECONDS)); + + assertThrows( + IOException.class, + () -> api.post("flagevaluation", requestBody("parallel"), stream -> null, null, false)); + releaseProbe.countDown(); + recoveringPost.get(5, TimeUnit.SECONDS); + + assertEquals(1, proxyApiCreations.get()); + assertEquals(1, recoveredLocal.calls); } @Test - void doesNotReplayConnectionReset() { - assertNoDirectReplay(new SocketException("connection reset")); + void failedRecoveryIsStickyForAnotherCooldown() throws Exception { + final AtomicLong clock = new AtomicLong(); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + null, + () -> { + proxyApiCreations.incrementAndGet(); + return null; + }, + () -> null, + "exposure", + clock::get, + 10); + + clock.set(10); + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("first"), stream -> null, null, false)); + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("second"), stream -> null, null, false)); + clock.set(20); + assertThrows( + IOException.class, + () -> api.post("exposures", requestBody("third"), stream -> null, null, false)); + + assertEquals(2, proxyApiCreations.get()); } @Test @@ -116,6 +240,8 @@ void doesNotRetryDirectApiCreationWhenFallbackIsUnavailable() { final AgentlessFeatureFlagBackendApi api = new AgentlessFeatureFlagBackendApi( local, + null, + () -> local, () -> { directApiCreations.incrementAndGet(); return null; @@ -126,20 +252,183 @@ void doesNotRetryDirectApiCreationWhenFallbackIsUnavailable() { HttpResponseException.class, () -> api.post("exposures", requestBody("first"), stream -> null, null, false)); assertThrows( - HttpResponseException.class, + IOException.class, () -> api.post("exposures", requestBody("second"), stream -> null, null, false)); - assertEquals(2, local.calls); + assertEquals(1, local.calls); + assertEquals(1, directApiCreations.get()); + } + + @Test + void permitsUnavailableStartupSoLocalDeliveryCanRecover() { + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi(null, null, () -> null, () -> null, "flag evaluation"); + + assertThrows( + IOException.class, + () -> api.post("flagevaluation", requestBody("first"), stream -> null, null, false)); + } + + @Test + void propagatesDirectFailuresWithoutTryingToFallback() { + final SocketTimeoutException failure = new SocketTimeoutException("direct timeout"); + final RecordingBackendApi direct = new RecordingBackendApi(failure); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + direct, + () -> { + proxyApiCreations.incrementAndGet(); + return null; + }, + () -> direct, + "exposure"); + + assertSame( + failure, + assertThrows( + SocketTimeoutException.class, + () -> api.post("exposures", requestBody("direct"), stream -> null, null, false))); + assertEquals(0, proxyApiCreations.get()); + assertEquals(1, direct.calls); + } + + @Test + void concurrentProxyFailuresOnlyTransitionTheMatchingRouteOnce() throws Exception { + final CountDownLatch localCallsStarted = new CountDownLatch(2); + final CountDownLatch releaseFailures = new CountDownLatch(1); + final BackendApi local = + new BackendApi() { + @Override + public T post( + final String uri, + final RequestBody requestBody, + final IOThrowingFunction responseParser, + @Nullable final OkHttpUtils.CustomListener requestListener, + final boolean requestCompression) + throws IOException { + localCallsStarted.countDown(); + try { + assertTrue(releaseFailures.await(5, TimeUnit.SECONDS)); + } catch (final InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new AssertionError(exception); + } + throw new HttpResponseException(404, "local route missing"); + } + }; + final AtomicInteger directCalls = new AtomicInteger(); + final BackendApi direct = + new BackendApi() { + @Override + public T post( + final String uri, + final RequestBody requestBody, + final IOThrowingFunction responseParser, + @Nullable final OkHttpUtils.CustomListener requestListener, + final boolean requestCompression) { + directCalls.incrementAndGet(); + return null; + } + }; + final AtomicInteger directApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + local, + null, + () -> local, + () -> { + directApiCreations.incrementAndGet(); + return direct; + }, + "exposure"); + + final CompletableFuture first = + CompletableFuture.runAsync(() -> postWithoutFailure(api, "first concurrent proxy request")); + final CompletableFuture second = + CompletableFuture.runAsync( + () -> postWithoutFailure(api, "second concurrent proxy request")); + assertTrue(localCallsStarted.await(5, TimeUnit.SECONDS)); + releaseFailures.countDown(); + first.get(5, TimeUnit.SECONDS); + second.get(5, TimeUnit.SECONDS); + assertEquals(1, directApiCreations.get()); + assertEquals(2, directCalls.get()); + } + + @Test + void sharesRouteTransitionsAcrossExposureAndFlagEvaluationWriters() throws Exception { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + final RecordingBackendApi exposureLocal = + new RecordingBackendApi(new SocketTimeoutException("ambiguous timeout")); + final RecordingBackendApi exposureDirect = new RecordingBackendApi(); + final RecordingBackendApi evaluationLocal = new RecordingBackendApi(); + final RecordingBackendApi evaluationDirect = new RecordingBackendApi(); + final AgentlessFeatureFlagBackendApi exposureApi = + new AgentlessFeatureFlagBackendApi( + exposureLocal, + exposureDirect, + () -> exposureLocal, + () -> exposureDirect, + "exposure", + routeSelector); + final AgentlessFeatureFlagBackendApi evaluationApi = + new AgentlessFeatureFlagBackendApi( + evaluationLocal, + evaluationDirect, + () -> evaluationLocal, + () -> evaluationDirect, + "flag evaluation", + routeSelector); + + assertThrows( + SocketTimeoutException.class, + () -> exposureApi.post("exposures", requestBody("first"), stream -> null, null, false)); + evaluationApi.post("flagevaluation", requestBody("second"), stream -> null, null, false); + + assertEquals(1, exposureLocal.calls); + assertEquals(0, exposureDirect.calls); + assertEquals(0, evaluationLocal.calls); + assertEquals(1, evaluationDirect.calls); + } + + @Test + void workingDirectRouteDoesNotAttemptRecovery() throws Exception { + final AtomicLong clock = new AtomicLong(); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + direct, + () -> { + proxyApiCreations.incrementAndGet(); + throw new IllegalStateException("discovery failed"); + }, + () -> direct, + "exposure", + clock::get, + 10); + + clock.set(10); + api.post("exposures", requestBody("survives recovery failure"), stream -> null, null, false); + + assertEquals(1, direct.calls); + assertEquals(0, proxyApiCreations.get()); } - private static void assertNoDirectReplay(final IOException failure) { + private static void assertNoSameBatchReplayButUsesDirectForNext(final IOException failure) + throws Exception { final RecordingBackendApi local = new RecordingBackendApi(failure); final RecordingBackendApi direct = new RecordingBackendApi(); final AtomicInteger directApiCreations = new AtomicInteger(); final AgentlessFeatureFlagBackendApi api = new AgentlessFeatureFlagBackendApi( local, + null, + () -> local, () -> { directApiCreations.incrementAndGet(); return direct; @@ -149,16 +438,26 @@ private static void assertNoDirectReplay(final IOException failure) { assertThrows( IOException.class, () -> api.post("flagevaluation", requestBody("evaluation"), stream -> null, null, false)); + api.post("flagevaluation", requestBody("next"), stream -> null, null, false); assertEquals(1, local.calls); - assertEquals(0, direct.calls); - assertEquals(0, directApiCreations.get()); + assertEquals(1, direct.calls); + assertEquals(1, directApiCreations.get()); } private static RequestBody requestBody(final String value) { return RequestBody.create(MediaType.parse("application/json"), value); } + private static void postWithoutFailure( + final AgentlessFeatureFlagBackendApi api, final String body) { + try { + api.post("exposures", requestBody(body), stream -> null, null, false); + } catch (final IOException exception) { + throw new AssertionError(exception); + } + } + private static Stream featureFlagRoutes() { return Stream.of( Arguments.of("exposures", "exposure"), Arguments.of("flagevaluation", "flag evaluation")); 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 f70a4ff0fa1..a0c4f7d59ee 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 @@ -4,6 +4,7 @@ import static java.util.Collections.singletonList; import static java.util.Collections.singletonMap; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -57,6 +58,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; import okhttp3.RequestBody; @@ -65,20 +67,19 @@ 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 { - private static final String EXPOSURES_ENDPOINT = "/evp_proxy/api/v2/exposures"; + private static final String EXPOSURES_ENDPOINT = "/evp_proxy/v2/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); private Queue requests; + private Queue requestAttempts; private Set failed; private JavaTestHttpServer server; private SharedCommunicationObjects sharedCommunicationObjects; @@ -86,6 +87,7 @@ class ExposureWriterTests { @BeforeEach void setUp() { requests = new ConcurrentLinkedQueue<>(); + requestAttempts = new ConcurrentLinkedQueue<>(); failed = Collections.newSetFromMap(new ConcurrentHashMap()); JsonAdapter adapter = new Moshi.Builder().build().adapter(ExposuresRequest.class); @@ -114,6 +116,7 @@ private void handleExposureRequest(HandlerApi api, JsonAdapter adapter.fromJson( Okio.buffer(Okio.source(new ByteArrayInputStream(api.getRequest().getBody())))); String serviceName = exposuresRequest.context.get("service"); + requestAttempts.add(serviceName); boolean failForever = "fail-forever".equals(serviceName); boolean fail = serviceName.startsWith("fail") && (failed.add(serviceName) || failForever); if (fail) { @@ -262,10 +265,183 @@ void testHighLoadScenario() throws Exception { } } - @ParameterizedTest - @ValueSource(booleans = {false, true}) - void testFailuresAreRetried(boolean finallyFail) throws Exception { - String serviceName = finallyFail ? "fail-forever" : "fail-once"; + @Test + void testQueueThresholdFlushesWithoutWaitingForTheInterval() throws Exception { + Config config = mockConfig("threshold-service"); + List exposures = buildExposures(101); + + try (ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 8, Long.MAX_VALUE, NANOSECONDS, sharedCommunicationObjects, config)) { + for (ExposureEvent exposure : exposures) { + writer.accept(exposure); + } + writer.init(); + + poll.eventually(() -> assertExposures(allExposures(), exposures)); + } + } + + @Test + void testCloseDrainsAndFinalFlushesExactlyOnceWithoutWaitingForTheInterval() throws Exception { + Config config = mockConfig("shutdown-service"); + List exposures = buildExposures(5); + ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 4, Long.MAX_VALUE, NANOSECONDS, sharedCommunicationObjects, config); + + writer.init(); + for (ExposureEvent exposure : exposures) { + writer.accept(exposure); + } + + writer.close(); + + assertFalse(writer.isSerializerThreadAlive()); + assertEquals(1, requests.size()); + assertExposures(allExposures(), exposures); + + // A repeated close and a stale listener invocation after close must neither replay the batch + // nor leave an event stranded in the queue. + writer.close(); + writer.accept(buildExposure()); + MILLISECONDS.sleep(200); + assertEquals(1, requests.size()); + assertEquals(0, writer.queueSize()); + } + + @Test + void testFinalFlushRunsWithoutTheInterruptFlagSet() throws Exception { + BackendApi backendApi = mock(BackendApi.class); + AtomicBoolean interruptedDuringPost = new AtomicBoolean(true); + when(backendApi.post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false))) + .thenAnswer( + invocation -> { + interruptedDuringPost.set(Thread.currentThread().isInterrupted()); + return null; + }); + ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 4, + Long.MAX_VALUE, + NANOSECONDS, + () -> backendApi, + mockConfig("shutdown-service"), + ExposureWriterImpl.SHUTDOWN_TIMEOUT_MILLIS); + + writer.init(); + writer.accept(buildExposure()); + writer.close(); + + verify(backendApi, times(1)) + .post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false)); + assertFalse( + interruptedDuringPost.get(), + "The final HTTP request must not inherit the interrupt used to wake queue polling"); + assertFalse(writer.isSerializerThreadAlive()); + } + + @Test + void testCloseWaitIsBoundedWhenFinalPostDoesNotReturn() throws Exception { + BackendApi backendApi = mock(BackendApi.class); + CountDownLatch postStarted = new CountDownLatch(1); + CountDownLatch releasePost = new CountDownLatch(1); + when(backendApi.post(eq("exposures"), any(RequestBody.class), any(), any(), eq(false))) + .thenAnswer( + invocation -> { + postStarted.countDown(); + releasePost.await(); + return null; + }); + ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 4, + Long.MAX_VALUE, + NANOSECONDS, + () -> backendApi, + mockConfig("blocked-shutdown-service"), + 100); + + try { + writer.init(); + writer.accept(buildExposure()); + long start = System.nanoTime(); + + writer.close(); + + long elapsedMillis = NANOSECONDS.toMillis(System.nanoTime() - start); + assertTrue(postStarted.await(1, java.util.concurrent.TimeUnit.SECONDS)); + assertTrue(elapsedMillis < 2000, "close exceeded its configured bounded wait"); + assertTrue(writer.isSerializerThreadAlive()); + } finally { + releasePost.countDown(); + } + poll.eventually(() -> assertFalse(writer.isSerializerThreadAlive())); + } + + @Test + void testStaleGatewayDispatchCannotEnqueueAfterClose() throws Exception { + CountDownLatch dispatchSnapshotTaken = new CountDownLatch(1); + CountDownLatch releaseDispatch = new CountDownLatch(1); + FeatureFlaggingGateway.ExposureListener blocker = + ignored -> { + dispatchSnapshotTaken.countDown(); + try { + releaseDispatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }; + FeatureFlaggingGateway.addExposureListener(blocker); + ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 4, + Long.MAX_VALUE, + NANOSECONDS, + sharedCommunicationObjects, + mockConfig("stale-dispatch-service")); + Thread dispatcher = + new Thread(() -> FeatureFlaggingGateway.dispatch(buildExposure()), "exposure-dispatcher"); + + try { + writer.init(); + dispatcher.start(); + assertTrue(dispatchSnapshotTaken.await(5, java.util.concurrent.TimeUnit.SECONDS)); + + writer.close(); + releaseDispatch.countDown(); + dispatcher.join(5000); + + assertFalse(dispatcher.isAlive()); + assertFalse(writer.isSerializerThreadAlive()); + assertEquals(0, writer.queueSize()); + assertTrue(requests.isEmpty()); + } finally { + releaseDispatch.countDown(); + dispatcher.join(5000); + FeatureFlaggingGateway.removeExposureListener(blocker); + writer.close(); + } + } + + @Test + void testCloseBeforeInitPreventsLaterStartAndAccept() { + ExposureWriterImpl writer = + new ExposureWriterImpl(sharedCommunicationObjects, mockConfig("never-started-service")); + + writer.close(); + writer.init(); + writer.accept(buildExposure()); + writer.close(); + + assertFalse(writer.isSerializerThreadAlive()); + assertEquals(0, writer.queueSize()); + assertTrue(requests.isEmpty()); + } + + @Test + void testHttpFailureIsNotRetriedAtTransportLayer() throws Exception { + String serviceName = "fail-once"; Config config = mockConfig(serviceName); try (ExposureWriterImpl writer = @@ -273,13 +449,11 @@ void testFailuresAreRetried(boolean finallyFail) throws Exception { writer.init(); writer.accept(buildExposure()); - MILLISECONDS.sleep(500); // wait for a flush to happen - ExposuresRequest found = findRequest(serviceName); - if (finallyFail) { - assertNull(found, requests.toString()); - } else { - poll.eventually(() -> assertNotNull(findRequest(serviceName), requests.toString())); - } + poll.eventually(() -> assertEquals(1, Collections.frequency(requestAttempts, serviceName))); + MILLISECONDS.sleep(500); + + assertEquals(1, Collections.frequency(requestAttempts, serviceName)); + assertNull(findRequest(serviceName), requests.toString()); } } @@ -309,7 +483,7 @@ void testAmbiguousExposureBatchIsNotRetriedOrReplayedDirectly() throws Exception final BackendApi proxyApi = mock(BackendApi.class); final BackendApi directApi = mock(BackendApi.class); when(backendApiFactory.createEvpProxyApi( - Intake.EVENT_PLATFORM, true, HttpRetryPolicy.Factory.NEVER_RETRY)) + Intake.EVENT_PLATFORM, true, HttpRetryPolicy.Factory.NEVER_RETRY, false, true)) .thenReturn(proxyApi); when(backendApiFactory.createDirectIntakeApi(eq(Intake.EVENT_PLATFORM), eq(true), eq(false))) .thenReturn(directApi); @@ -353,17 +527,14 @@ void testAmbiguousExposureBatchIsNotRetriedOrReplayedDirectly() throws Exception } @Test - void testWriterStopsReceivingExposuresIfEvpProxyIsNotAvailable() throws Exception { + void testAgentlessWriterWaitsForUnavailableProxyRecovery() throws Exception { SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(false); + Config config = mockConfig("unavailable-service"); + when(config.getFeatureFlaggingConfigurationSource()).thenReturn(CONFIGURATION_SOURCE_AGENTLESS); - try (ExposureWriterImpl writer = - new ExposureWriterImpl(sharedCommunicationObjects, Config.get())) { + try (ExposureWriterImpl writer = new ExposureWriterImpl(sharedCommunicationObjects, config)) { writer.init(); - poll.eventually(() -> assertFalse(writer.isSerializerThreadAlive())); - - FeatureFlaggingGateway.dispatch(buildExposure()); - - assertEquals(0, writer.queueSize()); + poll.eventually(() -> assertTrue(writer.isSerializerThreadAlive())); } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagBackendApiFactoryTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagBackendApiFactoryTest.java index b89cf3ce2b1..58811e0bfb6 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagBackendApiFactoryTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagBackendApiFactoryTest.java @@ -2,10 +2,10 @@ import static com.datadog.featureflag.FeatureFlagEventType.EXPOSURE; import static com.datadog.featureflag.FeatureFlagEventType.FLAG_EVALUATION; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; 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; @@ -22,38 +22,33 @@ class FeatureFlagBackendApiFactoryTest { @Test - void remoteConfigUsesOnlyLocalEvpProxy() { + void remoteConfigUsesFixedV2WithoutDiscoveryOrDirectIntake() { final Config config = config(CONFIGURATION_SOURCE_REMOTE_CONFIG, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); final BackendApi proxyApi = mock(BackendApi.class); - when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(proxyApi); + when(backendApiFactory.createEvpProxyApiForEndpoint( + Intake.EVENT_PLATFORM, + false, + HttpRetryPolicy.Factory.NEVER_RETRY, + V2_EVP_PROXY_ENDPOINT)) + .thenReturn(proxyApi); final BackendApi selected = new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); assertSame(proxyApi, selected); + verify(backendApiFactory, never()) + .createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true); verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false); } @Test - void remoteConfigDisablesDeliveryWhenLocalEvpProxyIsUnavailable() { - final Config config = config(CONFIGURATION_SOURCE_REMOTE_CONFIG, "api-key"); - final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - - final BackendApi selected = - new FeatureFlagBackendApiFactory(config, backendApiFactory, EXPOSURE).create(); - - assertNull(selected); - verify(backendApiFactory).createEvpProxyApi(Intake.EVENT_PLATFORM, true); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, true, false); - } - - @Test - void agentlessPrefersLocalEvpProxyWithDirectFallback() { + void agentlessPrefersCapabilityGatedLocalRouteWithDirectFallbackReady() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); when(backendApiFactory.createEvpProxyApi( - Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY)) + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true)) .thenReturn(mock(BackendApi.class)); when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false)) .thenReturn(mock(BackendApi.class)); @@ -63,68 +58,43 @@ void agentlessPrefersLocalEvpProxyWithDirectFallback() { assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); verify(backendApiFactory) - .createEvpProxyApi(Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false); + .createEvpProxyApi( + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true); + verify(backendApiFactory).createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false); } @Test - void agentlessUsesDirectIntakeWhenLocalEvpProxyIsUnavailable() { + void agentlessUsesDirectIntakeWhenLocalRouteIsUnavailable() { 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, false, false)) - .thenReturn(directApi); - - final BackendApi selected = - new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - - assertSame(directApi, selected); - } - - @Test - void agentlessUsesLocalEvpProxyWhenApiKeyIsUnavailable() { - final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, null); - final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - final BackendApi proxyApi = mock(BackendApi.class); - when(backendApiFactory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(proxyApi); + .thenReturn(mock(BackendApi.class)); final BackendApi selected = new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - assertSame(proxyApi, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false); + assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); } @Test - void agentlessDisablesDeliveryWhenNoRouteIsAvailable() { + void agentlessKeepsWriterAliveWhileEveryRouteIsUnavailable() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, null); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - final BackendApi selected = - new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - - assertNull(selected); - } - - @Test - void agentlessDisablesDeliveryWhenApiKeyIsEmpty() { - final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, ""); - final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); - final BackendApi selected = new FeatureFlagBackendApiFactory(config, backendApiFactory, EXPOSURE).create(); - assertNull(selected); + assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, true, false); } @Test - void agentlessDoesNotValidateDirectUrlWhileLocalRouteIsAvailable() { + void agentlessKeepsCompatibleLocalRouteWhenDirectUrlIsInvalid() { final Config config = config(CONFIGURATION_SOURCE_AGENTLESS, "api-key"); final BackendApiFactory backendApiFactory = mock(BackendApiFactory.class); final BackendApi proxyApi = mock(BackendApi.class); when(backendApiFactory.createEvpProxyApi( - Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY)) + Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY, false, true)) .thenReturn(proxyApi); when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false)) .thenThrow(new IllegalArgumentException("invalid URL")); @@ -133,20 +103,6 @@ void agentlessDoesNotValidateDirectUrlWhileLocalRouteIsAvailable() { new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); assertInstanceOf(AgentlessFeatureFlagBackendApi.class, selected); - verify(backendApiFactory, never()).createDirectIntakeApi(Intake.EVENT_PLATFORM, false, 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, false, false)) - .thenThrow(new IllegalArgumentException("invalid URL")); - - final BackendApi selected = - new FeatureFlagBackendApiFactory(config, backendApiFactory, FLAG_EVALUATION).create(); - - assertNull(selected); } private static Config config(final String source, final String apiKey) { From e20a9d77c333ddb0c0444835f23fafba70952373 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sat, 12 Sep 2026 04:26:30 +0000 Subject: [PATCH 2/2] test(openfeature): cover EVP fallback edge states --- .../AgentlessFeatureFlagBackendApiTest.java | 75 +++++++++++++++++++ .../featureflag/ExposureWriterTests.java | 24 ++++++ .../FeatureFlagRouteSelectorTest.java | 33 ++++++++ 3 files changed, 132 insertions(+) create mode 100644 products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagRouteSelectorTest.java diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java index 012ba26ea01..69792ae35bf 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessFeatureFlagBackendApiTest.java @@ -394,6 +394,81 @@ void sharesRouteTransitionsAcrossExposureAndFlagEvaluationWriters() throws Excep assertEquals(1, evaluationDirect.calls); } + @Test + void sharedLocalRouteDiscoversWriterSpecificProxy() throws Exception { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + routeSelector.initialize(true, false); + final RecordingBackendApi recoveredLocal = new RecordingBackendApi(); + final AtomicInteger proxyApiCreations = new AtomicInteger(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + null, + () -> { + proxyApiCreations.incrementAndGet(); + return recoveredLocal; + }, + () -> null, + "flag evaluation", + routeSelector); + + api.post("flagevaluation", requestBody("local"), stream -> null, null, false); + + assertEquals(1, proxyApiCreations.get()); + assertEquals(1, recoveredLocal.calls); + assertEquals(FeatureFlagRouteSelector.Route.LOCAL, routeSelector.current()); + } + + @Test + void sharedLocalRouteUsesDirectWhenWriterSpecificProxyIsUnavailable() throws Exception { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + routeSelector.initialize(true, false); + final RecordingBackendApi direct = new RecordingBackendApi(); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, direct, () -> null, () -> direct, "flag evaluation", routeSelector); + + api.post("flagevaluation", requestBody("direct"), stream -> null, null, false); + + assertEquals(1, direct.calls); + assertEquals(FeatureFlagRouteSelector.Route.DIRECT, routeSelector.current()); + } + + @Test + void sharedLocalRouteBecomesUnavailableWhenWriterHasNoRoute() { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + routeSelector.initialize(true, false); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, + null, + () -> { + throw new IllegalStateException("discovery failed"); + }, + () -> null, + "flag evaluation", + routeSelector); + + assertThrows( + IOException.class, + () -> api.post("flagevaluation", requestBody("missing"), stream -> null, null, false)); + assertEquals(FeatureFlagRouteSelector.Route.UNAVAILABLE, routeSelector.current()); + } + + @Test + void sharedDirectRouteWithoutWriterClientIsUnavailable() { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + routeSelector.initialize(false, true); + final AgentlessFeatureFlagBackendApi api = + new AgentlessFeatureFlagBackendApi( + null, null, () -> null, () -> null, "flag evaluation", routeSelector); + + assertThrows( + IOException.class, + () -> api.post("flagevaluation", requestBody("missing"), stream -> null, null, false)); + assertEquals(FeatureFlagRouteSelector.Route.DIRECT, routeSelector.current()); + } + @Test void workingDirectRouteDoesNotAttemptRecovery() throws Exception { final AtomicLong clock = new AtomicLong(); 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 a0c4f7d59ee..65a4057e6b1 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 @@ -439,6 +439,30 @@ void testCloseBeforeInitPreventsLaterStartAndAccept() { assertTrue(requests.isEmpty()); } + @Test + void testMissingBackendClosesWriterFromSerializerThread() throws Exception { + CountDownLatch backendRequested = new CountDownLatch(1); + ExposureWriterImpl writer = + new ExposureWriterImpl( + 1 << 4, + Long.MAX_VALUE, + NANOSECONDS, + () -> { + backendRequested.countDown(); + return null; + }, + mockConfig("missing-backend-service"), + 100); + + writer.init(); + + assertTrue(backendRequested.await(5, java.util.concurrent.TimeUnit.SECONDS)); + poll.eventually(() -> assertFalse(writer.isSerializerThreadAlive())); + writer.accept(buildExposure()); + assertEquals(0, writer.queueSize()); + writer.close(); + } + @Test void testHttpFailureIsNotRetriedAtTransportLayer() throws Exception { String serviceName = "fail-once"; diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagRouteSelectorTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagRouteSelectorTest.java new file mode 100644 index 00000000000..d95cacdb94b --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagRouteSelectorTest.java @@ -0,0 +1,33 @@ +package com.datadog.featureflag; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; + +import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Test; + +class FeatureFlagRouteSelectorTest { + + @Test + void directRouteIsTerminalAcrossInitializationAndRecoverySignals() { + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(); + + assertEquals(FeatureFlagRouteSelector.Route.DIRECT, routeSelector.initialize(false, true)); + assertEquals(FeatureFlagRouteSelector.Route.DIRECT, routeSelector.initialize(true, false)); + assertFalse(routeSelector.tryBeginLocalRecovery()); + + routeSelector.localRecovered(); + + assertEquals(FeatureFlagRouteSelector.Route.DIRECT, routeSelector.current()); + } + + @Test + void unavailableRouteCanBeReinitializedWithLocal() { + final AtomicLong clock = new AtomicLong(); + final FeatureFlagRouteSelector routeSelector = new FeatureFlagRouteSelector(clock::get, 10); + + assertEquals( + FeatureFlagRouteSelector.Route.UNAVAILABLE, routeSelector.initialize(false, false)); + assertEquals(FeatureFlagRouteSelector.Route.LOCAL, routeSelector.initialize(true, false)); + } +}