diff --git a/communication/src/main/java/datadog/communication/BackendApiFactory.java b/communication/src/main/java/datadog/communication/BackendApiFactory.java
index c1b471ebfc8..b233094f048 100644
--- a/communication/src/main/java/datadog/communication/BackendApiFactory.java
+++ b/communication/src/main/java/datadog/communication/BackendApiFactory.java
@@ -1,5 +1,7 @@
package datadog.communication;
+import static java.util.Collections.emptyList;
+
import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
import datadog.communication.ddagent.SharedCommunicationObjects;
import datadog.communication.http.HttpRetryPolicy;
@@ -15,13 +17,29 @@
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 boolean sendOnce;
public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommunicationObjects) {
+ this(config, sharedCommunicationObjects, false);
+ }
+
+ /**
+ * Creates a backend factory with 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, boolean sendOnce) {
this.config = config;
this.sharedCommunicationObjects = sharedCommunicationObjects;
+ this.sendOnce = sendOnce;
}
public @Nullable BackendApi createBackendApi(Intake intake) {
@@ -67,7 +85,9 @@ public BackendApi createDirectIntakeApi(
apiKey,
traceId,
retryPolicyFactory(),
- directIntakeHttpClient(sharedCommunicationObjects.getIntakeHttpClient(), followRedirects),
+ configureHttpClient(
+ directIntakeHttpClient(
+ sharedCommunicationObjects.getIntakeHttpClient(), followRedirects)),
responseCompression);
}
@@ -87,11 +107,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 +129,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 +174,88 @@ 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, emptyList());
+ }
+
+ /**
+ * Creates an EVP proxy client after Agent discovery, optionally forcing a fresh discovery and
+ * requiring the Agent to advertise every specified forwarding header.
+ *
+ *
Callers supply the forwarding capabilities their event protocol requires.
+ *
+ *
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,
+ Iterable requiredProxyHeaders) {
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
+ && !featuresDiscovery.supportsEvpProxyHeaders(requiredProxyHeaders)) {
+ 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) {
+ return sendOnce ? httpClient.newBuilder().retryOnConnectionFailure(false).build() : httpClient;
+ }
+
+ 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..301c48f0ff5 100644
--- a/communication/src/main/java/datadog/communication/EvpProxy.java
+++ b/communication/src/main/java/datadog/communication/EvpProxy.java
@@ -5,6 +5,12 @@ 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";
+
/**
* 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..f3479237c01 100644
--- a/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java
+++ b/communication/src/test/java/datadog/communication/BackendApiFactoryTest.java
@@ -1,10 +1,16 @@
package datadog.communication;
+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 java.util.Arrays.asList;
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 +21,9 @@
import datadog.trace.api.intake.Intake;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import java.util.Locale;
+import java.util.stream.Stream;
import okhttp3.HttpUrl;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
@@ -25,6 +33,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 +42,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 +82,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 +217,272 @@ 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(asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)));
+ } 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(asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)));
+
+ // The capability gate must not depend on whether the factory emits SDK identity headers.
+ final BackendApiFactory factory =
+ new BackendApiFactory(
+ Config.get(), sharedCommunicationObjects(discovery, agent.url("/")), true);
+ assertNull(
+ factory.createEvpProxyApi(
+ Intake.EVENT_PLATFORM,
+ false,
+ HttpRetryPolicy.Factory.NEVER_RETRY,
+ false,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)));
+ } finally {
+ agent.shutdown();
+ }
+ }
+
+ @Test
+ void featureFlagProxyRequestNeverCarriesDirectCredentials() 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("/")), true);
+ final BackendApi api =
+ factory.createEvpProxyApi(
+ Intake.EVENT_PLATFORM,
+ false,
+ HttpRetryPolicy.Factory.NEVER_RETRY,
+ false,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER));
+
+ 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"));
+ assertNull(request.getHeader("DD-API-KEY"));
+ assertEquals(1, agent.getRequestCount());
+ } finally {
+ agent.shutdown();
+ }
+ }
+
+ @Test
+ void featureFlagDirectRequestCarriesCredentialsExactlyOnce() 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), 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"));
+ 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 capabilityGatedProxyRequiresHeadersWithoutRequestHeaderConfiguration() {
+ final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(V4_EVP_PROXY_ENDPOINT, false);
+ final BackendApiFactory factory =
+ new BackendApiFactory(Config.get(), sharedCommunicationObjects(discovery, null), true);
+
+ assertNull(
+ factory.createEvpProxyApi(
+ Intake.EVENT_PLATFORM,
+ false,
+ HttpRetryPolicy.Factory.NEVER_RETRY,
+ false,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)));
+ }
+
+ @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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)));
+ assertEquals(1, discovery.forcedDiscoveries);
+ assertEquals(0, discovery.outdatedDiscoveries);
+ }
+
@Test
void explicitNoRetryProxyPolicyDoesNotReplayAmbiguousFailure() throws Exception {
final MockWebServer agent = new MockWebServer();
@@ -208,8 +538,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 +556,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 +578,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..f5b75e2f815 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,6 +1,10 @@
package com.datadog.featureflag;
+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.Arrays.asList;
import datadog.communication.BackendApi;
import datadog.communication.BackendApiFactory;
@@ -20,62 +24,83 @@ final class FeatureFlagBackendApiFactory {
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, 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);
+ }
- LOGGER.warn(
- "Feature Flagging {} delivery is disabled because no compatible local EVP proxy or direct intake credentials are available",
- eventType.logName());
- return null;
+ @Nullable
+ private BackendApi createProxyApi(final boolean forceDiscovery) {
+ return backendApiFactory.createEvpProxyApi(
+ Intake.EVENT_PLATFORM,
+ eventType.responseCompressionEnabled(),
+ HttpRetryPolicy.Factory.NEVER_RETRY,
+ forceDiscovery,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER));
}
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..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
@@ -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 doesNotReplayTimeout() {
- assertNoDirectReplay(new SocketTimeoutException("timed out"));
+ void doesNotReplayTimeoutButSwitchesFutureBatches() throws Exception {
+ assertNoSameBatchReplayButUsesDirectForNext(new SocketTimeoutException("timed out"));
}
@Test
- void doesNotReplayConnectionReset() {
- assertNoDirectReplay(new SocketException("connection reset"));
+ void doesNotReplayConnectionResetButSwitchesFutureBatches() throws Exception {
+ assertNoSameBatchReplayButUsesDirectForNext(new SocketException("connection reset"));
+ }
+
+ @Test
+ 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 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,258 @@ 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 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());
}
- private static void assertNoDirectReplay(final IOException failure) {
+ @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 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 +513,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..8c243ae5a95 100644
--- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java
+++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/ExposureWriterTests.java
@@ -1,9 +1,13 @@
package com.datadog.featureflag;
+import static datadog.communication.EvpProxy.ORIGIN_HEADER;
+import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER;
import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS;
+import static java.util.Arrays.asList;
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 +61,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 +70,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 +90,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 +119,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 +268,207 @@ 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 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";
Config config = mockConfig(serviceName);
try (ExposureWriterImpl writer =
@@ -273,13 +476,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 +510,11 @@ 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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)))
.thenReturn(proxyApi);
when(backendApiFactory.createDirectIntakeApi(eq(Intake.EVENT_PLATFORM), eq(true), eq(false)))
.thenReturn(directApi);
@@ -353,17 +558,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..f77d064b9c0 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,13 @@
import static com.datadog.featureflag.FeatureFlagEventType.EXPOSURE;
import static com.datadog.featureflag.FeatureFlagEventType.FLAG_EVALUATION;
+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 datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_REMOTE_CONFIG;
+import static java.util.Arrays.asList;
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 +25,41 @@
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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER));
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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)))
.thenReturn(mock(BackendApi.class));
when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false))
.thenReturn(mock(BackendApi.class));
@@ -63,68 +69,51 @@ 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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER));
+ 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,
+ asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)))
.thenReturn(proxyApi);
when(backendApiFactory.createDirectIntakeApi(Intake.EVENT_PLATFORM, false, false))
.thenThrow(new IllegalArgumentException("invalid URL"));
@@ -133,20 +122,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) {
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));
+ }
+}