From 8450e1dedbd7ee8ecad7b4e1b1b8992a02e6cad7 Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Wed, 21 Oct 2026 13:47:21 +0200 Subject: [PATCH 1/3] feat(ffe): Restore the agent Feature Flags behaviors through the bridge Port the agent-side behaviors of the removed writers: send each event batch once with proxy discovery and recovery, and the Java SDK identity headers (#12477, #12575); flush the SDK pending events at agent shutdown, with or without tracing; declare the SDK settings the agent exposes (#12884, #12774). Move the Feature Flags shutdown tests from the agent tests to the OpenFeature smoke tests. Co-Authored-By: Claude Opus 5.5 --- .../DDOpenFeatureInstrumentation.java | 1 + .../openfeature/AgentShutdownSource.java | 18 ++ .../openfeature/JavaAgentBridge.java | 3 +- .../DDOpenFeatureInstrumentationTest.java | 11 + .../lang/ShutdownInstrumentationTest.java | 58 ----- .../delivery/DirectIntakeTransport.java | 8 +- .../delivery/DirectIntakeTransportTest.java | 7 +- ...bstractOpenFeatureShutdownSmokeTest.groovy | 135 ++++++++++ .../OpenFeatureShutdownSmokeTest.groovy | 10 + ...atureTraceDisabledShutdownSmokeTest.groovy | 13 + .../smoketest/AbstractSmokeTest.groovy | 4 +- .../api/config/FeatureFlaggingConfig.java | 15 ++ .../featureflag/FeatureFlaggingSystem.java | 38 ++- .../FeatureFlaggingSystemTest.java | 59 +++++ .../featureflag/FeatureFlaggingGateway.java | 8 + .../com/datadog/featureflag/EventProxy.java | 233 +++++++++++++++--- .../featureflag/FeatureFlagsBackend.java | 62 ++++- .../datadog/featureflag/EventProxyTest.java | 218 +++++++++++++--- .../featureflag/FeatureFlagsBackendTest.java | 45 +++- 19 files changed, 809 insertions(+), 137 deletions(-) create mode 100644 dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java delete mode 100644 dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/test/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentationTest.java create mode 100644 dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy create mode 100644 dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy create mode 100644 dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java index 8fa3abcb5e7..e578e3824fc 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java @@ -41,6 +41,7 @@ public String[] helperClassNames() { packageName + ".AgentConfigurationSource", packageName + ".AgentEventProxy", packageName + ".AgentHealthMetrics", + packageName + ".AgentShutdownSource", packageName + ".AgentSpanEnricher", packageName + ".JavaAgentBridge", }; diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java new file mode 100644 index 00000000000..9e73869ded4 --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java @@ -0,0 +1,18 @@ +package datadog.trace.instrumentation.openfeature; + +import com.datadog.openfeature.internal.bridge.v1.ShutdownSource; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; + +/** Notifies the agent shutdown, before the JVM shutdown hooks run. */ +public final class AgentShutdownSource implements ShutdownSource { + private final FeatureFlaggingGateway.Backend backend; + + AgentShutdownSource(final FeatureFlaggingGateway.Backend backend) { + this.backend = backend; + } + + @Override + public AutoCloseable subscribe(final Runnable listener) { + return this.backend.subscribeShutdown(listener); + } +} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java index a0122636ff5..ad4d2da6366 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java @@ -64,7 +64,6 @@ public SpanEnricher spanEnricher() { @Override public ShutdownSource shutdown() { - // TODO Flush the SDK through the agent shutdown. Until then, the SDK uses a JVM shutdown hook. - return null; + return new AgentShutdownSource(this.backend); } } diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationTest.java index 61470453cd9..18fe2081cf7 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationTest.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationTest.java @@ -88,6 +88,17 @@ void postsThroughTheEventProxyOnlyWhenAvailable() throws Exception { assertThrows(EventProxyUnavailableException.class, () -> proxy.post("exposures", payload)); } + @Test + void subscribesToTheAgentShutdown() throws Exception { + final FeatureFlaggingGateway.Backend backend = mock(FeatureFlaggingGateway.Backend.class); + final AutoCloseable subscription = mock(AutoCloseable.class); + final Runnable listener = mock(Runnable.class); + when(backend.subscribeShutdown(listener)).thenReturn(subscription); + FeatureFlaggingGateway.register(backend); + + assertSame(subscription, AgentBridges.detect().shutdown().subscribe(listener)); + } + @Test void delegatesSpanEnrichment() { final FeatureFlaggingGateway.Backend backend = mock(FeatureFlaggingGateway.Backend.class); diff --git a/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/test/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentationTest.java b/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/test/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentationTest.java deleted file mode 100644 index 13736934cda..00000000000 --- a/dd-java-agent/instrumentation/java/java-lang/java-lang-1.8/src/test/java/datadog/trace/instrumentation/java/lang/ShutdownInstrumentationTest.java +++ /dev/null @@ -1,58 +0,0 @@ -package datadog.trace.instrumentation.java.lang; - -import static datadog.trace.agent.tooling.InstrumenterModule.TargetSystem.TRACING; -import static java.util.Collections.emptySet; -import static java.util.Collections.singleton; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import datadog.trace.agent.tooling.InstrumenterModule.TargetSystem; -import datadog.trace.test.junit.utils.config.WithConfig; -import datadog.trace.test.util.DDJavaSpecification; -import java.util.EnumSet; -import org.junit.jupiter.api.Test; - -class ShutdownInstrumentationTest extends DDJavaSpecification { - @Test - void defaultConfigurationDoesNotEnableWholeAgentShutdownWithoutTracing() { - assertFalse(new ShutdownInstrumentation().isApplicable(emptySet())); - } - - @Test - @WithConfig(key = "feature.flags.enabled", value = "true") - void generatedTargetMaskDoesNotDependOnFeatureFlagBuildConfiguration() { - final EnumSet systems = EnumSet.noneOf(TargetSystem.class); - for (TargetSystem system : TargetSystem.values()) { - if (new ShutdownInstrumentation().isApplicable(singleton(system))) { - systems.add(system); - } - } - assertEquals(EnumSet.of(TRACING), systems); - } - - @Test - @WithConfig(key = "feature.flags.enabled", value = "false") - void doesNotInstrumentShutdownWhenTracingAndFeatureFlagsAreDisabled() { - assertFalse(new ShutdownInstrumentation().isApplicable(emptySet())); - } - - @Test - @WithConfig(key = "feature.flags.enabled", value = "false") - void preservesTracingShutdown() { - assertTrue(new ShutdownInstrumentation().isApplicable(singleton(TRACING))); - } - - @Test - @WithConfig(key = "feature.flags.configuration.source", value = "agentless") - void featureFlagsDoNotEnableWholeAgentShutdownWithoutTracing() { - assertFalse(new ShutdownInstrumentation().isApplicable(emptySet())); - } - - @Test - @WithConfig(key = "feature.flags.enabled", value = "false") - @WithConfig(key = "feature.flags.configuration.source", value = "agentless") - void honorsExplicitDisableEvenWithAConfigurationSource() { - assertFalse(new ShutdownInstrumentation().isApplicable(emptySet())); - } -} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java index 509f24fb699..fa87a169b42 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java @@ -4,6 +4,7 @@ import static java.net.http.HttpRequest.BodyPublishers.ofByteArray; import com.datadog.openfeature.internal.bridge.v1.EventTransport; +import com.datadog.openfeature.internal.config.SdkVersion; import com.datadog.openfeature.internal.config.Settings; import java.io.IOException; import java.net.URI; @@ -22,6 +23,9 @@ public final class DirectIntakeTransport implements EventTransport { private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(10); + /** Identifies the events like the agent event proxy does. */ + static final String ORIGIN = "dd-trace-java"; + private final URI intake; private final String apiKey; private final String traceId; @@ -62,7 +66,9 @@ public void post(final String route, final byte[] json) throws IOException { .header("Content-Type", "application/json") .header("dd-api-key", this.apiKey) .header("x-datadog-trace-id", this.traceId) - .header("x-datadog-parent-id", this.traceId); + .header("x-datadog-parent-id", this.traceId) + .header("DD-EVP-ORIGIN", ORIGIN) + .header("DD-EVP-ORIGIN-VERSION", SdkVersion.VERSION); Headers.addMetadata(builder); final HttpRequest request = builder.POST(ofByteArray(json)).build(); final int status; diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java index ed3f92e35cc..d2b87e44304 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java @@ -2,6 +2,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -12,6 +13,7 @@ import static org.mockito.Mockito.verify; import com.datadog.openfeature.internal.LocalHttpServer; +import com.datadog.openfeature.internal.config.SdkVersion; import com.datadog.openfeature.internal.config.TestSettings; import java.io.IOException; import java.io.InterruptedIOException; @@ -56,7 +58,7 @@ void rejectsInvalidSite() { } @Test - void postsJsonWithApiKeyAndTraceHeaders() throws Exception { + void postsJsonWithApiKeyTraceAndIdentityHeaders() throws Exception { this.server.enqueue(202, null); transport().post("exposures", "{\"a\":1}".getBytes(UTF_8)); @@ -71,6 +73,9 @@ void postsJsonWithApiKeyAndTraceHeaders() throws Exception { assertEquals(request.header("x-datadog-trace-id"), request.header("x-datadog-parent-id")); assertEquals("java", request.header("Datadog-Meta-Lang")); assertNull(request.header("X-Datadog-EVP-Subdomain")); + assertEquals("dd-trace-java", request.header("DD-EVP-ORIGIN")); + assertEquals(SdkVersion.VERSION, request.header("DD-EVP-ORIGIN-VERSION")); + assertNotEquals("unknown", request.header("DD-EVP-ORIGIN-VERSION")); } @TableTest({ diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy new file mode 100644 index 00000000000..240131577d1 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy @@ -0,0 +1,135 @@ +package datadog.smoketest.springboot + +import static datadog.trace.agent.test.server.http.TestHttpServer.httpServer +import static java.util.concurrent.TimeUnit.SECONDS + +import datadog.smoketest.AbstractSmokeTest +import datadog.trace.agent.test.server.http.TestHttpServer +import datadog.trace.agent.test.server.http.TestHttpServer.HandlerApi.RequestApi +import datadog.trace.agent.test.utils.OkHttpUtils +import datadog.trace.agent.test.utils.PortUtils +import groovy.json.JsonOutput +import groovy.json.JsonSlurper +import okhttp3.MediaType +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.RequestBody +import spock.lang.AutoCleanup +import spock.lang.Shared +import spock.util.concurrent.PollingConditions + +/** The SDK with the Datadog Java agent flushes its pending events when the application exits. */ +abstract class AbstractOpenFeatureShutdownSmokeTest extends AbstractSmokeTest { + + @Shared + protected int httpPort = PortUtils.randomOpenPort() + + protected OkHttpClient client = OkHttpUtils.client() + + @Shared + protected final ufcConfig = new JsonSlurper().parse(fetchResource("ffe-system-test-data/ufc-config.json")) as Map + + /** Serves the flag configuration as the Datadog managed CDN does. */ + @Shared + @AutoCleanup + protected TestHttpServer cdn = httpServer { + handlers { + prefix("/api/v2/feature-flagging/config/rules-based/server") { + response.status(200).send(JsonOutput.toJson([ + data: [type: 'universal-flag-configuration', attributes: ufcConfig] + ])) + } + } + } + + /** @return whether the application runs with tracing enabled. */ + abstract boolean tracing() + + def setupSpec() { + PortUtils.waitForPortToOpen(httpPort, 240, SECONDS, testedProcess) + } + + @Override + ProcessBuilder createProcessBuilder() { + final springBootShadowJar = System.getProperty("datadog.smoketest.springboot.shadowJar.path") + final command = [javaPath()] + command.addAll(defaultJavaProperties) + command.add("-Ddd.trace.enabled=${tracing()}".toString()) + command.addAll(['-jar', springBootShadowJar, "--server.port=${httpPort}".toString()]) + final builder = new ProcessBuilder(command).directory(new File(buildDirectory)) + builder.environment().put('DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL', "http://localhost:${cdn.address.port}".toString()) + return builder + } + + @Override + boolean testTelemetry() { + return false + } + + @Override + Closure decodedTracesCallback() { + return {} + } + + @Override + Closure decodedEvpProxyMessageCallback() { + return { String path, RequestApi request -> + if (!path.contains('api/v2/flagevaluation')) { + return null + } + return [ + origin: request.getHeader('DD-EVP-ORIGIN'), + originVersion: request.getHeader('DD-EVP-ORIGIN-VERSION'), + body: new JsonSlurper().parse(request.body) + ] + } + } + + void 'test pending flag evaluations are flushed at application exit'() { + when: + // The first evaluation is flushed within the 10 seconds flush interval, which then restarts: + // the next evaluation stays pending until the application exits. + evaluate('test-case-boolean-false-assignment.json') + new PollingConditions(timeout: 20).eventually { + assert evaluatedFlags().contains('boolean-false-assignment') + } + evaluate('test-case-boolean-one-of-matches.json') + + then: + !evaluatedFlags().contains('boolean-one-of-matches') + + when: + testedProcess.destroy() + + then: + testedProcess.waitFor(30, SECONDS) + evaluatedFlags().contains('boolean-one-of-matches') + evaluationRequests().every { it.origin == 'dd-trace-java' && it.originVersion } + } + + protected void evaluate(final String testCaseFile) { + final testCase = (new JsonSlurper().parse(fetchResource("ffe-system-test-data/evaluation-cases/${testCaseFile}")) as List>).first() + final request = new Request.Builder() + .url("http://localhost:${httpPort}/openfeature/evaluate") + .post(RequestBody.create(MediaType.parse('application/json'), JsonOutput.toJson(testCase))) + .build() + final response = client.newCall(request).execute() + try { + assert response.code() == 200 + } finally { + response.close() + } + } + + protected List evaluatedFlags() { + return evaluationRequests()*.body*.flagEvaluations.flatten()*.flag*.key as List + } + + protected List> evaluationRequests() { + return evpProxyMessages*.getV2() as List> + } + + protected static URL fetchResource(final String name) { + return Thread.currentThread().getContextClassLoader().getResource(name) + } +} diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy new file mode 100644 index 00000000000..70192aa9331 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy @@ -0,0 +1,10 @@ +package datadog.smoketest.springboot + +/** The tracing shutdown hook flushes the SDK pending events. */ +class OpenFeatureShutdownSmokeTest extends AbstractOpenFeatureShutdownSmokeTest { + + @Override + boolean tracing() { + return true + } +} diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy new file mode 100644 index 00000000000..c300962d260 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy @@ -0,0 +1,13 @@ +package datadog.smoketest.springboot + +/** + * Without tracing, the Feature Flagging system registers its own shutdown hook to flush the SDK + * pending events. + */ +class OpenFeatureTraceDisabledShutdownSmokeTest extends AbstractOpenFeatureShutdownSmokeTest { + + @Override + boolean tracing() { + return false + } +} diff --git a/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy b/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy index 28b71306349..0afa31b8809 100644 --- a/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy +++ b/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy @@ -96,7 +96,9 @@ abstract class AbstractSmokeTest extends ProcessManager { "Accept-Encoding", "Content-Encoding", "User-Agent", - "DD-CI-PROVIDER-NAME" + "DD-CI-PROVIDER-NAME", + "DD-EVP-ORIGIN", + "DD-EVP-ORIGIN-VERSION" ], "config": {} }""") diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java index 9a232111344..a2e6a5378f4 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java @@ -13,5 +13,20 @@ public final class FeatureFlaggingConfig { public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE = "feature.flags.configuration.source"; + /** + * Turns off sending exposures to Datadog while customer exposure hooks keep receiving them. + * Default: enabled. Maps to {@code DD_FEATURE_FLAGS_EXPOSURES_DATADOG_LOGGING_ENABLED}. + */ + public static final String FEATURE_FLAGS_EXPOSURES_DATADOG_LOGGING_ENABLED = + "feature.flags.exposures.datadog.logging.enabled"; + + /** + * Lets the {@code dd_evaluation_telemetry_opt_out} evaluation context attribute opt subjects out + * of evaluation telemetry. Default: disabled. Maps to {@code + * DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED}. + */ + public static final String FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED = + "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; + private FeatureFlaggingConfig() {} } 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 0b971d3f6f0..300a0d36861 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 @@ -1,5 +1,8 @@ package com.datadog.featureflag; +import static datadog.trace.util.AgentThreadFactory.AgentThread.FEATURE_FLAG_SHUTDOWN; +import static datadog.trace.util.AgentThreadFactory.newAgentThread; + import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; @@ -16,6 +19,7 @@ public class FeatureFlaggingSystem { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class); private static volatile FeatureFlagsBackend BACKEND; + private static volatile Thread SHUTDOWN_HOOK; private FeatureFlaggingSystem() {} @@ -33,7 +37,12 @@ public static synchronized void start(final SharedCommunicationObjects sco) { LOGGER.debug("Feature Flagging system disabled"); return; } - final FeatureFlagsBackend backend = new FeatureFlagsBackend(sco, config); + // Tracing owns the shutdown path running before the JVM shutdown hooks. Without it, stop from a + // shutdown hook once the SDK subscribes to the shutdown, so it still flushes its pending + // events. + final FeatureFlagsBackend backend = + new FeatureFlagsBackend( + sco, config, config.isTraceEnabled() ? null : FeatureFlaggingSystem::addShutdownHook); try { backend.start(); } catch (final RuntimeException | Error e) { @@ -52,10 +61,37 @@ public static synchronized void start(final SharedCommunicationObjects sco) { public static synchronized void stop() { final FeatureFlagsBackend backend = BACKEND; BACKEND = null; + final Thread shutdownHook = SHUTDOWN_HOOK; + SHUTDOWN_HOOK = null; + if (shutdownHook != null && shutdownHook != Thread.currentThread()) { + try { + Runtime.getRuntime().removeShutdownHook(shutdownHook); + } catch (final IllegalStateException ignored) { + // The JVM is already shutting down; the hook finds the system stopped. + } + } FeatureFlaggingGateway.register(null); if (backend != null) { backend.close(); } LOGGER.debug("Feature Flagging system stopped"); } + + private static synchronized void addShutdownHook() { + if (BACKEND == null || SHUTDOWN_HOOK != null) { + return; + } + final Thread hook = newAgentThread(FEATURE_FLAG_SHUTDOWN, FeatureFlaggingSystem::stop); + try { + Runtime.getRuntime().addShutdownHook(hook); + } catch (final IllegalStateException | SecurityException e) { + LOGGER.debug("Could not register the Feature Flagging shutdown hook", e); + return; + } + SHUTDOWN_HOOK = hook; + } + + static boolean isShutdownHookRegistered() { + return SHUTDOWN_HOOK != null; + } } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index 0935d1914cb..c2a265544d1 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -3,10 +3,13 @@ import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED; import static datadog.trace.api.config.RemoteConfigConfig.REMOTE_CONFIGURATION_ENABLED; +import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_ENABLED; +import static org.junit.jupiter.api.Assertions.assertFalse; 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.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -19,6 +22,7 @@ import datadog.remoteconfig.Product; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.test.junit.utils.config.WithConfig; +import java.lang.reflect.Field; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -65,6 +69,61 @@ void failedStartRollsBackTheRemoteConfigRegistration() { verify(poller).removeListeners(Product.FFE_FLAGS); } + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void stopsFromAShutdownHookOnceTheSdkSubscribesWhenTracingIsDisabled() { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + assertFalse( + FeatureFlaggingSystem.isShutdownHookRegistered(), "no hook while the SDK is unused"); + + FeatureFlaggingGateway.backend().subscribeShutdown(() -> {}); + assertTrue(FeatureFlaggingSystem.isShutdownHookRegistered()); + + FeatureFlaggingSystem.stop(); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void shutdownHookFlushesListenersAndStopsTheBackend() throws Exception { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + final Runnable listener = mock(Runnable.class); + FeatureFlaggingGateway.backend().subscribeShutdown(listener); + + final Field field = FeatureFlaggingSystem.class.getDeclaredField("SHUTDOWN_HOOK"); + field.setAccessible(true); + final Thread hook = (Thread) field.get(null); + // Run the real hook without shutting down the test JVM. + assertTrue(Runtime.getRuntime().removeShutdownHook(hook)); + hook.start(); + hook.join(5000); + + assertFalse(hook.isAlive(), "the shutdown hook must finish"); + assertNull(FeatureFlaggingGateway.backend()); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + verify(listener).run(); + } + + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void lateShutdownSubscriptionDoesNotRegisterAHookAfterStop() throws Exception { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + final FeatureFlaggingGateway.Backend backend = FeatureFlaggingGateway.backend(); + FeatureFlaggingSystem.stop(); + + try (final AutoCloseable ignored = backend.subscribeShutdown(() -> {})) { + assertNull(FeatureFlaggingGateway.backend()); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + } + + @Test + void leavesTheShutdownToTracingWhenEnabled() { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + FeatureFlaggingGateway.backend().subscribeShutdown(() -> {}); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + @Test @WithConfig(key = FEATURE_FLAGS_ENABLED, value = "false") void startDoesNothingWhenFeatureFlagsIsDisabled() { diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java index 1f1cfcaca76..a20b965a96c 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java @@ -98,5 +98,13 @@ public interface Backend { * @param defaultValue the native default value. */ void enrichRuntimeDefault(String flagKey, @Nullable Object defaultValue); + + /** + * Subscribes to the agent shutdown, notified before the JVM shutdown hooks run. + * + * @param listener the listener run at shutdown, typically flushing the SDK pending events. + * @return the handle to close to unsubscribe. + */ + AutoCloseable subscribeShutdown(Runnable listener); } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java index ccc28f1e858..6b00fca54e4 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java @@ -2,45 +2,108 @@ import static com.datadog.featureflag.FeatureFlagEventType.EXPOSURE; import static com.datadog.featureflag.FeatureFlagEventType.FLAG_EVALUATION; +import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY; +import static datadog.communication.EvpProxy.ORIGIN_HEADER; +import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; import static datadog.communication.http.HttpRetryPolicy.Factory.NEVER_RETRY; import static datadog.trace.api.intake.Intake.EVENT_PLATFORM; +import static java.util.Arrays.asList; +import static java.util.Collections.unmodifiableMap; +import static java.util.concurrent.TimeUnit.MINUTES; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.HttpResponseException; +import datadog.communication.ddagent.TracerVersion; +import datadog.communication.http.HttpRetryPolicy; import java.io.IOException; import java.net.ConnectException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.LongSupplier; import javax.annotation.Nullable; import okhttp3.MediaType; import okhttp3.RequestBody; +import okhttp3.Response; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -/** Posts Feature Flags events through the Datadog Agent event platform proxy. */ +/** + * Posts Feature Flags events through the Datadog Agent event platform proxy. + * + *

With the Remote Configuration source, events always go to the Agent through the fixed v2 proxy + * endpoint. With the agentless source, the proxy endpoint is discovered once and shared by both + * routes, provided it forwards the identity headers sent with every event. When it is not available + * and the SDK cannot fall back to direct intake, the proxy route stays in place and the endpoint is + * discovered again at most once per recovery interval. + * + *

Payloads are sent once: only a connection refusal, proven to happen before sending, is + * retried, and only with the Remote Configuration source. + */ class EventProxy { + private static final Logger LOGGER = LoggerFactory.getLogger(EventProxy.class); private static final MediaType JSON = MediaType.parse("application/json"); static final String EXPOSURES_ROUTE = "exposures"; + static final long RECOVERY_INTERVAL_NANOS = MINUTES.toNanos(1); + + /** Identifies the library sending the events, on every proxy request. */ + static final Map IDENTITY_HEADERS = identityHeaders(); + + /** The headers a discovered proxy endpoint must forward to keep the event identity. */ + static final List IDENTITY_HEADER_NAMES = asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER); + + /** Retries connection refusal only; automatic retries and redirects remain disabled. */ + static final HttpRetryPolicy.Factory PRE_SEND_RETRY = + new HttpRetryPolicy.Factory(3, 100, 2.0) { + @Override + public HttpRetryPolicy create() { + return new PreSendRetryPolicy(); + } + }; private final BackendApiFactory backendApiFactory; + private final boolean agentless; private final boolean directFallbackAvailable; - private volatile BackendApi exposures; - private volatile BackendApi flagEvaluations; - private volatile boolean resolved; + private final LongSupplier nanoTime; + private volatile Clients clients; + private boolean discovered; + private long nextDiscoveryNanos; /** - * @param backendApiFactory the factory of event platform proxy clients. + * @param backendApiFactory the send-once factory of event platform proxy clients. + * @param agentless whether the configuration source is agentless, so events use the discovered + * proxy endpoint, or direct intake when the SDK can fall back to it. * @param directFallbackAvailable whether the SDK falls back to direct intake when the proxy is - * unavailable, in which case the proxy must not retry what the intake would accept. + * unavailable. */ - EventProxy(final BackendApiFactory backendApiFactory, final boolean directFallbackAvailable) { + EventProxy( + final BackendApiFactory backendApiFactory, + final boolean agentless, + final boolean directFallbackAvailable) { + this(backendApiFactory, agentless, directFallbackAvailable, System::nanoTime); + } + + EventProxy( + final BackendApiFactory backendApiFactory, + final boolean agentless, + final boolean directFallbackAvailable, + final LongSupplier nanoTime) { this.backendApiFactory = backendApiFactory; + this.agentless = agentless; this.directFallbackAvailable = directFallbackAvailable; + this.nanoTime = nanoTime; } /** - * @return whether the Datadog Agent supports the event platform proxy. + * @return whether events go through the event platform proxy: always with the Remote + * Configuration source, and with the agentless source when the proxy endpoint is discovered + * or when direct intake is not available. */ boolean isAvailable() { - resolve(); - return this.exposures != null; + // Only the agentless source can lack proxy clients. + return current() != null || !this.directFallbackAvailable; } /** @@ -48,57 +111,167 @@ boolean isAvailable() { * * @param route the event platform route. * @param json the UTF-8 JSON payload. - * @return {@code true} if delivered, {@code false} if the proxy is definitively unavailable. + * @return {@code true} if delivered, {@code false} if the proxy definitively did not receive the + * payload. * @throws IOException if the payload could not be delivered. */ boolean post(final String route, final byte[] json) throws IOException { - resolve(); - final BackendApi api = EXPOSURES_ROUTE.equals(route) ? this.exposures : this.flagEvaluations; - if (api == null) { - return false; + Clients current = current(); + if (current == null) { + current = recover(); + if (current == null) { + return false; + } } + final BackendApi api = + EXPOSURES_ROUTE.equals(route) ? current.exposures : current.flagEvaluations; try { api.post(route, RequestBody.create(JSON, json), stream -> null, null, false); return true; } catch (final IOException e) { if (isDefinitiveRejection(e)) { + onDefinitiveRejection(current); return false; } throw e; } } - private void resolve() { - if (this.resolved) { - return; + @Nullable + private Clients current() { + final Clients current = this.clients; + if (current != null) { + return current; } synchronized (this) { - if (this.resolved) { - return; + if (!this.discovered) { + this.discovered = true; + this.clients = + this.agentless ? discover(false) : clients(PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + if (this.clients == null) { + onUnavailable(); + } } - this.exposures = create(EXPOSURE); - this.flagEvaluations = create(FLAG_EVALUATION); - this.resolved = true; + return this.clients; } } @Nullable - private BackendApi create(final FeatureFlagEventType eventType) { - return this.directFallbackAvailable - ? this.backendApiFactory.createEvpProxyApi( - EVENT_PLATFORM, eventType.responseCompressionEnabled(), NEVER_RETRY) - : this.backendApiFactory.createEvpProxyApi( - EVENT_PLATFORM, eventType.responseCompressionEnabled()); + private synchronized Clients recover() { + if (this.clients != null) { + return this.clients; + } + if (this.directFallbackAvailable || this.nanoTime.getAsLong() - this.nextDiscoveryNanos < 0) { + return null; + } + this.clients = discover(true); + if (this.clients == null) { + scheduleDiscovery(); + } else { + LOGGER.info("Feature Flags event delivery recovered through the Datadog Agent event proxy"); + } + return this.clients; } + private synchronized void onDefinitiveRejection(final Clients failed) { + // Only the agentless proxy route without direct intake waits for a new endpoint. An in-flight + // send on an old endpoint cannot invalidate a newly recovered one. + if (this.agentless && !this.directFallbackAvailable && this.clients == failed) { + this.clients = null; + onUnavailable(); + } + } + + /** Waits for a new proxy endpoint when it is the only route. Runs once per outage. */ + private void onUnavailable() { + if (!this.directFallbackAvailable) { + scheduleDiscovery(); + LOGGER.warn( + "Feature Flags events are being dropped because no local EVP proxy or direct intake" + + " route is available. Make the Agent reachable, or configure DD_API_KEY and a" + + " valid DD_SITE."); + } + } + + private void scheduleDiscovery() { + this.nextDiscoveryNanos = this.nanoTime.getAsLong() + RECOVERY_INTERVAL_NANOS; + } + + @Nullable + private Clients discover(final boolean force) { + final String endpoint; + try { + endpoint = this.backendApiFactory.discoverEvpProxyEndpoint(force, IDENTITY_HEADER_NAMES); + } catch (final RuntimeException e) { + LOGGER.debug("Could not discover the Datadog Agent event proxy", e); + return null; + } + return endpoint == null ? null : clients(NEVER_RETRY, endpoint); + } + + private Clients clients(final HttpRetryPolicy.Factory retryPolicy, final String endpoint) { + return new Clients( + this.backendApiFactory.createEvpProxyApiForEndpoint( + EVENT_PLATFORM, EXPOSURE.responseCompressionEnabled(), retryPolicy, endpoint), + this.backendApiFactory.createEvpProxyApiForEndpoint( + EVENT_PLATFORM, FLAG_EVALUATION.responseCompressionEnabled(), retryPolicy, endpoint)); + } + + private static Map identityHeaders() { + final Map headers = new HashMap<>(2); + headers.put(ORIGIN_HEADER, JAVA_TRACING_LIBRARY); + headers.put(ORIGIN_VERSION_HEADER, TracerVersion.TRACER_VERSION); + return unmodifiableMap(headers); + } + + /** + * Tells whether the proxy definitively did not receive a payload, so it can be replayed through + * direct intake. + * + * @param exception the post failure. + * @return whether the failure is a connection refusal or a missing proxy route. + */ static boolean isDefinitiveRejection(final IOException exception) { if (exception instanceof ConnectException) { return true; } if (exception instanceof HttpResponseException) { final int statusCode = ((HttpResponseException) exception).getStatusCode(); - return statusCode == 403 || statusCode == 404 || statusCode == 405; + return statusCode == 404 || statusCode == 405; } return false; } + + /** The proxy clients of both routes, for one endpoint. */ + private static final class Clients { + final BackendApi exposures; + final BackendApi flagEvaluations; + + Clients(final BackendApi exposures, final BackendApi flagEvaluations) { + this.exposures = exposures; + this.flagEvaluations = flagEvaluations; + } + } + + private static final class PreSendRetryPolicy extends HttpRetryPolicy { + private int attemptsLeft = 3; + + private PreSendRetryPolicy() { + super(3, 100, 2.0, false); + } + + @Override + public boolean shouldRetry(final Exception exception) { + if (exception instanceof ConnectException && this.attemptsLeft > 0) { + this.attemptsLeft--; + return true; + } + return false; + } + + @Override + public boolean shouldRetry(@Nullable final Response response) { + return false; + } + } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java index 0fdc3e3d712..10d2d69837f 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java @@ -13,11 +13,17 @@ import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.telemetry.CoreMetricCollector; import java.io.IOException; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** The agent services exposed to the Feature Flags SDK instrumentation. */ public final class FeatureFlagsBackend implements FeatureFlaggingGateway.Backend, AutoCloseable { + private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagsBackend.class); private static final String API_KEY = "api-key"; private static final String SITE = "site"; private static final String ENV = "env"; @@ -29,18 +35,33 @@ public final class FeatureFlagsBackend implements FeatureFlaggingGateway.Backend private final EventProxy eventProxy; private final SpanEnrichmentWriter spanEnrichmentWriter; private final CoreMetricCollector metrics; + private final List shutdownListeners = new CopyOnWriteArrayList<>(); + @Nullable private final Runnable onFirstShutdownSubscription; + private final AtomicBoolean shutdownSubscribed = new AtomicBoolean(); private RemoteConfigServiceImpl remoteConfig; - public FeatureFlagsBackend(final SharedCommunicationObjects sco, final Config config) { + /** + * @param sco the shared communication objects. + * @param config the agent configuration. + * @param onFirstShutdownSubscription the action to run when the SDK first subscribes to the + * shutdown, or {@code null} for none. + */ + public FeatureFlagsBackend( + final SharedCommunicationObjects sco, + final Config config, + @Nullable final Runnable onFirstShutdownSubscription) { this( sco, config, new EventProxy( - new BackendApiFactory(config, sco), + new BackendApiFactory(config, sco, EventProxy.IDENTITY_HEADERS, true), + config.getFeatureFlaggingConfigurationSource() == AGENTLESS, config.getFeatureFlaggingConfigurationSource() == AGENTLESS - && config.getApiKey() != null), + && config.getApiKey() != null + && !config.getApiKey().isEmpty()), SpanEnrichmentWriter.getInstance(), - CoreMetricCollector.getInstance()); + CoreMetricCollector.getInstance(), + onFirstShutdownSubscription); } FeatureFlagsBackend( @@ -48,12 +69,14 @@ public FeatureFlagsBackend(final SharedCommunicationObjects sco, final Config co final Config config, final EventProxy eventProxy, final SpanEnrichmentWriter spanEnrichmentWriter, - final CoreMetricCollector metrics) { + final CoreMetricCollector metrics, + @Nullable final Runnable onFirstShutdownSubscription) { this.sco = sco; this.config = config; this.eventProxy = eventProxy; this.spanEnrichmentWriter = spanEnrichmentWriter; this.metrics = metrics; + this.onFirstShutdownSubscription = onFirstShutdownSubscription; } /** Registers the Remote Configuration product early when it is the configuration source. */ @@ -136,7 +159,34 @@ public void enrichRuntimeDefault(final String flagKey, @Nullable final Object de } @Override - public synchronized void close() { + public AutoCloseable subscribeShutdown(final Runnable listener) { + this.shutdownListeners.add(listener); + if (this.onFirstShutdownSubscription != null + && this.shutdownSubscribed.compareAndSet(false, true)) { + this.onFirstShutdownSubscription.run(); + } + return () -> this.shutdownListeners.remove(listener); + } + + /** + * Notifies the shutdown listeners, flushing the SDK while the agent services are still running, + * then releases the backend resources. + */ + @Override + public void close() { + // Run outside the lock: the SDK flush posts events and counts metrics through this backend. + for (final Runnable listener : this.shutdownListeners) { + try { + listener.run(); + } catch (final RuntimeException | LinkageError e) { + LOGGER.debug("Feature Flags shutdown listener failed", e); + } + } + this.shutdownListeners.clear(); + closeResources(); + } + + private synchronized void closeResources() { if (this.remoteConfig != null) { this.remoteConfig.close(); this.remoteConfig = null; diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java index c0492e04819..5d44674e4f3 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java @@ -1,85 +1,196 @@ package com.datadog.featureflag; +import static com.datadog.featureflag.EventProxy.IDENTITY_HEADERS; +import static com.datadog.featureflag.EventProxy.IDENTITY_HEADER_NAMES; +import static com.datadog.featureflag.EventProxy.PRE_SEND_RETRY; +import static com.datadog.featureflag.EventProxy.RECOVERY_INTERVAL_NANOS; +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.http.HttpRetryPolicy.Factory.NEVER_RETRY; +import static datadog.trace.api.intake.Intake.EVENT_PLATFORM; import static java.nio.charset.StandardCharsets.UTF_8; +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.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyIterable; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.HttpResponseException; +import datadog.communication.ddagent.TracerVersion; import datadog.communication.http.HttpRetryPolicy; -import datadog.trace.api.intake.Intake; import java.io.IOException; import java.net.ConnectException; +import java.util.concurrent.atomic.AtomicLong; +import okhttp3.Response; import org.junit.jupiter.api.Test; import org.tabletest.junit.TableTest; class EventProxyTest { private static final byte[] PAYLOAD = "{}".getBytes(UTF_8); + private static final String ENDPOINT = "evp_proxy/v4/"; + + private final BackendApiFactory factory = mock(BackendApiFactory.class); + private final BackendApi exposures = mock(BackendApi.class); + private final BackendApi flagEvaluations = mock(BackendApi.class); + private final AtomicLong now = new AtomicLong(); + + @Test + void sendsRemoteConfigurationEventsToTheFixedV2EndpointWithPreSendRetry() throws IOException { + clientsFor(V2_EVP_PROXY_ENDPOINT); + final EventProxy proxy = proxy(false, false); + + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); + assertTrue(proxy.post("flagevaluation", PAYLOAD)); + + verify(this.factory) + .createEvpProxyApiForEndpoint(EVENT_PLATFORM, true, PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + verify(this.factory) + .createEvpProxyApiForEndpoint(EVENT_PLATFORM, false, PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + verify(this.factory, never()).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); + verify(this.flagEvaluations).post(eq("flagevaluation"), any(), any(), any(), eq(false)); + } + + @Test + void sendsAgentlessEventsOnceThroughTheDiscoveredEndpoint() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + final EventProxy proxy = proxy(true, true); + + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); + + verify(this.factory).createEvpProxyApiForEndpoint(EVENT_PLATFORM, true, NEVER_RETRY, ENDPOINT); + verify(this.factory).createEvpProxyApiForEndpoint(EVENT_PLATFORM, false, NEVER_RETRY, ENDPOINT); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + } @Test - void isUnavailableWithoutAgentProxySupport() throws IOException { - final EventProxy proxy = new EventProxy(mock(BackendApiFactory.class), false); + void leavesAgentlessEventsToDirectIntakeWithoutTheAgentProxy() throws IOException { + final EventProxy proxy = proxy(true, true); assertFalse(proxy.isAvailable()); + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + assertFalse(proxy.post("exposures", PAYLOAD)); + + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + } + + @Test + void recoversTheAgentProxyWhenItIsTheOnlyRoute() throws IOException { + final EventProxy proxy = proxy(true, false); + + assertTrue(proxy.isAvailable(), "the proxy route stays in place to recover"); assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, never()).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + when(this.factory.discoverEvpProxyEndpoint(true, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + + assertTrue(proxy.post("exposures", PAYLOAD)); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); } @Test - void postsEachRouteThroughItsOwnClient() throws IOException { - final BackendApiFactory factory = mock(BackendApiFactory.class); - final BackendApi exposures = mock(BackendApi.class); - final BackendApi flagEvaluations = mock(BackendApi.class); - when(factory.createEvpProxyApi(Intake.EVENT_PLATFORM, true)).thenReturn(exposures); - when(factory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(flagEvaluations); - final EventProxy proxy = new EventProxy(factory, false); + void failedDiscoveryRetriesAtMostOncePerRecoveryInterval() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)) + .thenThrow(new IllegalStateException("discovery failed")); + final EventProxy proxy = proxy(true, false); assertTrue(proxy.isAvailable()); + assertFalse(proxy.post("exposures", PAYLOAD)); + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.addAndGet(RECOVERY_INTERVAL_NANOS - 1); + assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.incrementAndGet(); + when(this.factory.discoverEvpProxyEndpoint(true, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); assertTrue(proxy.post("exposures", PAYLOAD)); - assertTrue(proxy.post("flagevaluation", PAYLOAD)); - verify(exposures).post(eq("exposures"), any(), any(), any(), eq(false)); - verify(flagEvaluations).post(eq("flagevaluation"), any(), any(), any(), eq(false)); + verify(this.factory, times(2)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); } @Test - void doesNotRetryWhenDirectFallbackIsAvailable() { - final BackendApiFactory factory = mock(BackendApiFactory.class); - final EventProxy proxy = new EventProxy(factory, true); + void doesNotRediscoverAfterRejectionWhenDirectFallbackIsAvailable() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")) + .thenReturn(null); + final EventProxy proxy = proxy(true, true); + + assertFalse(proxy.post("exposures", PAYLOAD)); + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); - proxy.isAvailable(); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures, times(2)).post(any(), any(), any(), any(), eq(false)); + } + + @Test + void propagatesAmbiguousTransportFailuresWithoutRetryOrRediscovery() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + final IOException failure = new IOException("connection reset after sending"); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())).thenThrow(failure); + final EventProxy proxy = proxy(true, false); + + assertSame(failure, assertThrows(IOException.class, () -> proxy.post("exposures", PAYLOAD))); + assertTrue(proxy.isAvailable()); - verify(factory) - .createEvpProxyApi(Intake.EVENT_PLATFORM, true, HttpRetryPolicy.Factory.NEVER_RETRY); - verify(factory) - .createEvpProxyApi(Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures).post(any(), any(), any(), any(), eq(false)); } @Test - void reportsConnectionFailuresAsUnavailable() throws IOException { - final BackendApi api = proxyApi(new ConnectException("refused")); + void waitsForANewEndpointAfterADefinitiveRejectionWhenItIsTheOnlyRoute() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")); + final EventProxy proxy = proxy(true, false); + + assertFalse(proxy.post("exposures", PAYLOAD)); + assertFalse(proxy.post("exposures", PAYLOAD)); - assertFalse(new EventProxy(factoryOf(api), false).post("exposures", PAYLOAD)); + verify(this.exposures, times(1)).post(any(), any(), any(), any(), anyBoolean()); } @TableTest({ "Scenario | Status | Definitive", - "forbidden | 403 | true ", + "forbidden | 403 | false ", "not found | 404 | true ", "not allowed | 405 | true ", "server error | 500 | false ", "rate limited | 429 | false " }) void classifiesProxyRejections(final int status, final boolean definitive) throws IOException { - final BackendApi api = proxyApi(new HttpResponseException(status, "rejected")); - final EventProxy proxy = new EventProxy(factoryOf(api), false); + clientsFor(V2_EVP_PROXY_ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new HttpResponseException(status, "rejected")); + final EventProxy proxy = proxy(false, false); if (definitive) { assertFalse(proxy.post("exposures", PAYLOAD)); @@ -88,15 +199,52 @@ void classifiesProxyRejections(final int status, final boolean definitive) throw } } - private static BackendApi proxyApi(final IOException failure) throws IOException { - final BackendApi api = mock(BackendApi.class); - when(api.post(any(), any(), any(), any(), anyBoolean())).thenThrow(failure); - return api; + @Test + void reportsConnectionRefusalAsDefinitive() throws IOException { + clientsFor(V2_EVP_PROXY_ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")); + + assertFalse(proxy(false, false).post("exposures", PAYLOAD)); + } + + @Test + void identifiesTheJavaTracingLibrary() { + assertEquals("dd-trace-java", IDENTITY_HEADERS.get(ORIGIN_HEADER)); + assertEquals(TracerVersion.TRACER_VERSION, IDENTITY_HEADERS.get(ORIGIN_VERSION_HEADER)); + assertNotEquals("0.0.0", IDENTITY_HEADERS.get(ORIGIN_VERSION_HEADER)); + } + + @Test + void discoversOnlyAgentlessEndpointsForwardingTheIdentity() { + proxy(true, true).isAvailable(); + + verify(this.factory) + .discoverEvpProxyEndpoint(false, asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)); + } + + @Test + void preSendRetryRetriesConnectionRefusalOnly() { + final HttpRetryPolicy policy = PRE_SEND_RETRY.create(); + + assertFalse(policy.shouldRetry(new IOException("reset"))); + assertFalse(policy.shouldRetry((Response) null)); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertFalse(policy.shouldRetry(new ConnectException("refused")), "at most three retries"); + } + + private EventProxy proxy(final boolean agentless, final boolean directFallbackAvailable) { + return new EventProxy(this.factory, agentless, directFallbackAvailable, this.now::get); } - private static BackendApiFactory factoryOf(final BackendApi api) { - final BackendApiFactory factory = mock(BackendApiFactory.class); - when(factory.createEvpProxyApi(eq(Intake.EVENT_PLATFORM), anyBoolean())).thenReturn(api); - return factory; + private void clientsFor(final String endpoint) { + when(this.factory.createEvpProxyApiForEndpoint( + eq(EVENT_PLATFORM), eq(true), any(), eq(endpoint))) + .thenReturn(this.exposures); + when(this.factory.createEvpProxyApiForEndpoint( + eq(EVENT_PLATFORM), eq(false), any(), eq(endpoint))) + .thenReturn(this.flagEvaluations); } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java index 0ba9c1a50e4..ed5c042adbb 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java @@ -3,6 +3,7 @@ import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Arrays.asList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -24,7 +25,10 @@ import datadog.trace.bootstrap.config.provider.ConfigProvider; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.test.junit.utils.config.WithConfigExtension; +import java.util.List; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.tabletest.junit.TableTest; @@ -41,7 +45,7 @@ class FeatureFlagsBackendTest { private final CoreMetricCollector metrics = mock(CoreMetricCollector.class); private final FeatureFlagsBackend backend = new FeatureFlagsBackend( - this.sco, this.config, this.eventProxy, this.spanEnrichmentWriter, this.metrics); + this.sco, this.config, this.eventProxy, this.spanEnrichmentWriter, this.metrics, null); private static AgentSpan rootSpan() { final AgentSpan root = mock(AgentSpan.class); @@ -126,11 +130,48 @@ void createsTheBackendFromTheAgentConfiguration( when(this.config.getFeatureFlaggingConfigurationSource()).thenReturn(source); when(this.config.getApiKey()).thenReturn(apiKey); - final FeatureFlagsBackend created = new FeatureFlagsBackend(this.sco, this.config); + final FeatureFlagsBackend created = new FeatureFlagsBackend(this.sco, this.config, null); assertEquals(apiKey, created.setting("api-key")); } + @Test + void runsTheFirstShutdownSubscriptionActionOnce() { + final AtomicInteger firstSubscriptions = new AtomicInteger(); + final FeatureFlagsBackend backend = + new FeatureFlagsBackend( + this.sco, + this.config, + this.eventProxy, + this.spanEnrichmentWriter, + this.metrics, + firstSubscriptions::incrementAndGet); + + backend.subscribeShutdown(() -> {}); + backend.subscribeShutdown(() -> {}); + + assertEquals(1, firstSubscriptions.get()); + } + + @Test + void closeNotifiesTheShutdownListenersBeforeReleasingResources() throws Exception { + final List calls = new CopyOnWriteArrayList<>(); + this.backend.subscribeShutdown(() -> calls.add("first")); + this.backend.subscribeShutdown( + () -> { + throw new IllegalStateException("failing listener"); + }); + final AutoCloseable unregistered = + this.backend.subscribeShutdown(() -> calls.add("unregistered")); + this.backend.subscribeShutdown(() -> calls.add("last")); + unregistered.close(); + + this.backend.close(); + this.backend.close(); + + assertEquals(asList("first", "last"), calls); + } + @Test void closeWithoutRemoteConfigClearsSpanEnrichment() { this.backend.enrichSerialId(1, false, null); From 53c469bee0172ee3f7ed9ae520b4d6c368bf4097 Mon Sep 17 00:00:00 2001 From: Sarah Chen Date: Thu, 22 Oct 2026 14:56:02 -0500 Subject: [PATCH 2/3] fix(ffe): Guard OpenFeature agent bridge against incompatible SDK interfaces --- .../openfeature/JavaAgentBridge.java | 37 +++- .../DDOpenFeatureClassLoaderTest.java | 197 ++++++++++++++++++ ...ureInstrumentationNoTracingForkedTest.java | 6 + ...nFeatureIntegrationDisabledForkedTest.java | 26 +++ 4 files changed, 264 insertions(+), 2 deletions(-) create mode 100644 dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java create mode 100644 dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java create mode 100644 dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureIntegrationDisabledForkedTest.java diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java index ad4d2da6366..d175aee6c44 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java @@ -1,5 +1,8 @@ package datadog.trace.instrumentation.openfeature; +import static java.lang.reflect.Modifier.isAbstract; +import static java.lang.reflect.Modifier.isStatic; + import com.datadog.openfeature.internal.bridge.v1.AgentBridge; import com.datadog.openfeature.internal.bridge.v1.ConfigurationSource; import com.datadog.openfeature.internal.bridge.v1.EventTransport; @@ -7,10 +10,20 @@ import com.datadog.openfeature.internal.bridge.v1.ShutdownSource; import com.datadog.openfeature.internal.bridge.v1.SpanEnricher; import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import java.lang.reflect.Method; import javax.annotation.Nullable; /** The bridge backed by the agent Feature Flags backend. */ public final class JavaAgentBridge implements AgentBridge { + // Muzzle checks calls, but not the SPI methods implemented by these helpers. + private static final boolean SPI_COMPATIBLE = + implementsSpi(AgentBridge.class, JavaAgentBridge.class) + && implementsSpi(ConfigurationSource.class, AgentConfigurationSource.class) + && implementsSpi(EventTransport.class, AgentEventProxy.class) + && implementsSpi(HealthMetrics.class, AgentHealthMetrics.class) + && implementsSpi(SpanEnricher.class, AgentSpanEnricher.class) + && implementsSpi(ShutdownSource.class, AgentShutdownSource.class); + private final FeatureFlaggingGateway.Backend backend; private final HealthMetrics healthMetrics; private final SpanEnricher spanEnricher; @@ -24,12 +37,32 @@ public final class JavaAgentBridge implements AgentBridge { /** * Connects the SDK to the agent backend. * - * @param fallback the bridge to use if the agent backend is not started. + * @param fallback the bridge to use if the agent backend is not started or the SDK SPI is + * incompatible. * @return the agent bridge, or the fallback one. */ public static AgentBridge connect(final AgentBridge fallback) { final FeatureFlaggingGateway.Backend backend = FeatureFlaggingGateway.backend(); - return backend == null ? fallback : new JavaAgentBridge(backend); + return backend == null || !SPI_COMPATIBLE ? fallback : new JavaAgentBridge(backend); + } + + private static boolean implementsSpi(final Class spi, final Class implementation) { + try { + for (final Method method : spi.getMethods()) { + if (!isStatic(method.getModifiers())) { + final Method implemented = + implementation.getMethod(method.getName(), method.getParameterTypes()); + if (isAbstract(implemented.getModifiers()) + || isStatic(implemented.getModifiers()) + || implemented.getReturnType() != method.getReturnType()) { + return false; + } + } + } + return true; + } catch (final NoSuchMethodException e) { + return false; + } } @Nullable diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java new file mode 100644 index 00000000000..2c9999641c2 --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java @@ -0,0 +1,197 @@ +package datadog.trace.instrumentation.openfeature; + +import static java.nio.file.Files.createDirectories; +import static java.nio.file.Files.write; +import static net.bytebuddy.jar.asm.Opcodes.ACC_ABSTRACT; +import static net.bytebuddy.jar.asm.Opcodes.ACC_PUBLIC; +import static net.bytebuddy.jar.asm.Opcodes.ACONST_NULL; +import static net.bytebuddy.jar.asm.Opcodes.ARETURN; +import static net.bytebuddy.jar.asm.Opcodes.ASM9; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datadog.openfeature.internal.bridge.v1.AgentBridge; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import java.io.IOException; +import java.io.InputStream; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Path; +import java.util.function.UnaryOperator; +import net.bytebuddy.jar.asm.ClassReader; +import net.bytebuddy.jar.asm.ClassVisitor; +import net.bytebuddy.jar.asm.ClassWriter; +import net.bytebuddy.jar.asm.MethodVisitor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.tabletest.junit.TableTest; + +class DDOpenFeatureClassLoaderTest extends AbstractInstrumentationTest { + @TempDir Path sdkOverrides; + + @AfterEach + void unregisterBackend() { + FeatureFlaggingGateway.register(null); + } + + @Test + void injectsIntoAnIsolatedSdkClassLoader() throws Exception { + final FeatureFlaggingGateway.Backend backend = mock(FeatureFlaggingGateway.Backend.class); + when(backend.setting("site")).thenReturn("datadoghq.eu"); + FeatureFlaggingGateway.register(backend); + + try (URLClassLoader loader = sdkClassLoader()) { + final Object bridge = detect(loader); + assertSame(loader, bridge.getClass().getClassLoader()); + assertEquals(JavaAgentBridge.class.getName(), bridge.getClass().getName()); + assertEquals( + "datadoghq.eu", + loader + .loadClass(AgentBridge.class.getName()) + .getMethod("setting", String.class) + .invoke(bridge, "site")); + } + } + + @TableTest({ + "Scenario | SPI | Method | Return type ", + "settings changed | AgentBridge | setting | ", + "configuration changed | ConfigurationSource | subscribe | ", + "transport changed | EventTransport | post | ", + "metrics changed | HealthMetrics | count | ", + "enrichment changed | SpanEnricher | serialId | ", + "shutdown changed | ShutdownSource | subscribe | ", + "return type changed | AgentBridge | setting | Ljava/lang/Object;" + }) + void keepsTheStandaloneBridgeWhenAnSpiMethodChanges( + final String spi, final String method, final String returnType) throws Exception { + rewriteSpi( + spi, + bytes -> { + final ClassWriter writer = new ClassWriter(0); + new ClassReader(bytes) + .accept( + new ClassVisitor(ASM9, writer) { + @Override + public MethodVisitor visitMethod( + int access, + String name, + String descriptor, + String signature, + String[] exceptions) { + if (name.equals(method)) { + if (returnType == null) { + name = "renamed"; + } else { + descriptor = + descriptor.substring(0, descriptor.indexOf(')') + 1) + returnType; + } + } + return super.visitMethod(access, name, descriptor, signature, exceptions); + } + }, + 0); + return writer.toByteArray(); + }); + FeatureFlaggingGateway.register(mock(FeatureFlaggingGateway.Backend.class)); + + try (URLClassLoader loader = sdkClassLoader()) { + // Implemented methods are not calls, so Muzzle alone accepts this incompatible SPI. + assertTrue(new DDOpenFeatureInstrumentation().getInstrumentationMuzzle().matches(loader)); + assertSame( + loader.loadClass(AgentBridge.class.getName()).getField("NONE").get(null), detect(loader)); + } + } + + @TableTest({ + "Scenario | Abstract | Static conflict", + "new abstract method | true | false ", + "new default method | false | false ", + "abstract conflicts with static | true | true ", + "default conflicts with static | false | true " + }) + void checksNewSpiMethods(final boolean isAbstract, final boolean staticConflict) + throws Exception { + rewriteSpi( + "AgentBridge", + bytes -> { + final ClassWriter writer = new ClassWriter(0); + new ClassReader(bytes) + .accept( + new ClassVisitor(ASM9, writer) { + @Override + public void visitEnd() { + final MethodVisitor method = + super.visitMethod( + ACC_PUBLIC | (isAbstract ? ACC_ABSTRACT : 0), + staticConflict ? "connect" : "additionalMethod", + staticConflict + ? "(Lcom/datadog/openfeature/internal/bridge/v1/AgentBridge;)" + + "Lcom/datadog/openfeature/internal/bridge/v1/AgentBridge;" + : "()Ljava/lang/String;", + null, + null); + if (!isAbstract) { + method.visitCode(); + if (staticConflict) { + method.visitInsn(ACONST_NULL); + } else { + method.visitLdcInsn("compatible"); + } + method.visitInsn(ARETURN); + method.visitMaxs(1, staticConflict ? 2 : 1); + } + method.visitEnd(); + super.visitEnd(); + } + }, + 0); + return writer.toByteArray(); + }); + FeatureFlaggingGateway.register(mock(FeatureFlaggingGateway.Backend.class)); + + try (URLClassLoader loader = sdkClassLoader()) { + final Class contract = loader.loadClass(AgentBridge.class.getName()); + final Object bridge = detect(loader); + if (isAbstract || staticConflict) { + assertSame(contract.getField("NONE").get(null), bridge); + } else { + assertEquals(JavaAgentBridge.class.getName(), bridge.getClass().getName()); + assertEquals("compatible", contract.getMethod("additionalMethod").invoke(bridge)); + } + } + } + + private URLClassLoader sdkClassLoader() throws IOException { + return new URLClassLoader( + new URL[] { + sdkOverrides.toUri().toURL(), + AgentBridge.class.getProtectionDomain().getCodeSource().getLocation() + }, + null); + } + + private static Object detect(final ClassLoader loader) throws Exception { + return loader + .loadClass("com.datadog.openfeature.internal.bridge.v1.AgentBridges") + .getMethod("detect") + .invoke(null); + } + + private void rewriteSpi(final String spi, final UnaryOperator rewrite) + throws IOException { + final String resource = "com/datadog/openfeature/internal/bridge/v1/" + spi + ".class"; + final Path target = sdkOverrides.resolve(resource); + createDirectories(target.getParent()); + try (InputStream input = AgentBridge.class.getClassLoader().getResourceAsStream(resource)) { + assertNotNull(input); + write(target, rewrite.apply(input.readAllBytes())); + } + } +} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java new file mode 100644 index 00000000000..31485765b7d --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java @@ -0,0 +1,6 @@ +package datadog.trace.instrumentation.openfeature; + +import datadog.trace.test.junit.utils.config.WithConfig; + +@WithConfig(key = "trace.enabled", value = "false") +class DDOpenFeatureInstrumentationNoTracingForkedTest extends DDOpenFeatureInstrumentationTest {} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureIntegrationDisabledForkedTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureIntegrationDisabledForkedTest.java new file mode 100644 index 00000000000..d7a52e5a6c1 --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureIntegrationDisabledForkedTest.java @@ -0,0 +1,26 @@ +package datadog.trace.instrumentation.openfeature; + +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; + +import com.datadog.openfeature.internal.bridge.v1.AgentBridge; +import com.datadog.openfeature.internal.bridge.v1.AgentBridges; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import datadog.trace.test.junit.utils.config.WithConfig; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +@WithConfig(key = "trace.openfeature.enabled", value = "false") +class DDOpenFeatureIntegrationDisabledForkedTest extends AbstractInstrumentationTest { + @AfterEach + void unregisterBackend() { + FeatureFlaggingGateway.register(null); + } + + @Test + void doesNotInstrumentWhenTheIntegrationIsDisabled() { + FeatureFlaggingGateway.register(mock(FeatureFlaggingGateway.Backend.class)); + assertSame(AgentBridge.NONE, AgentBridges.detect()); + } +} From 96895533c1d09f255fe39ba2e5b72f572630bab1 Mon Sep 17 00:00:00 2001 From: Bruce Bujon Date: Fri, 23 Oct 2026 16:20:17 +0200 Subject: [PATCH 3/3] fix(ffe): Report rejected Remote Configuration flag documents as apply errors The SDK dropped malformed Remote Configuration documents and the agent swallowed subscriber failures, so the poller acknowledged documents the SDK never applied. The SDK now rejects malformed documents by throwing, and the agent rethrows the first subscriber failure once every subscriber is notified, so the poller reports an apply error as before the switch. Co-Authored-By: Claude Opus 5.5 --- .../bridge/v1/ConfigurationSource.java | 3 +- .../delivery/RemoteConfigurationService.java | 5 +- .../internal/FeatureFlagsRuntimeTest.java | 4 +- .../RemoteConfigurationServiceTest.java | 77 +++++++++++++++++++ .../featureflag/FeatureFlaggingGateway.java | 3 +- .../featureflag/RemoteConfigServiceImpl.java | 26 ++++++- .../RemoteConfigServiceImplTest.java | 51 ++++++++++++ 7 files changed, 162 insertions(+), 7 deletions(-) create mode 100644 dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationServiceTest.java diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/v1/ConfigurationSource.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/v1/ConfigurationSource.java index 5cda6a93009..79505f916c1 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/v1/ConfigurationSource.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/v1/ConfigurationSource.java @@ -8,7 +8,8 @@ public interface ConfigurationSource { * Subscribes to configuration updates. * * @param listener the listener receiving raw UFC JSON documents, or {@code null} when the - * configuration is removed. + * configuration is removed. It throws a {@link RuntimeException} to reject a malformed + * document, which the source reports as not applied. * @return the handle to close to unsubscribe. */ AutoCloseable subscribe(Consumer listener); diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationService.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationService.java index 63cae4527cb..c594c2a2702 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationService.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationService.java @@ -36,8 +36,9 @@ public synchronized void start(final Consumer listener) { try { configuration = UniversalFlagConfigParser.parse(content); } catch (final IOException | RuntimeException e) { - LOGGER.debug("Remote Configuration returned a malformed flag configuration", e); - return; + // Rejects the document so the agent reports it as not applied. + throw new IllegalArgumentException( + "Malformed flag configuration: " + e.getMessage(), e); } if (configuration != null) { listener.accept(configuration); diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index 1104b34f8fc..8e930d27ebe 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -125,7 +125,9 @@ void remoteConfigurationDeliversParsedConfigurationsAndReplaysToLateListeners() runtime.start(); try { source.listener.accept(EMPTY_UFC); - source.listener.accept("{not json".getBytes(StandardCharsets.UTF_8)); + assertThrows( + IllegalArgumentException.class, + () -> source.listener.accept("{not json".getBytes(StandardCharsets.UTF_8))); final List late = new CopyOnWriteArrayList<>(); runtime.addConfigurationListener(late::add); diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationServiceTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationServiceTest.java new file mode 100644 index 00000000000..8706b279348 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationServiceTest.java @@ -0,0 +1,77 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.datadog.openfeature.internal.bridge.v1.ConfigurationSource; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import org.junit.jupiter.api.Test; + +class RemoteConfigurationServiceTest { + @Test + void forwardsParsedConfigurationsAndRemovals() { + final CapturingSource source = new CapturingSource(); + final List received = new ArrayList<>(); + final RemoteConfigurationService service = new RemoteConfigurationService(source); + service.start(received::add); + + source.listener.accept("{\"flags\":{}}".getBytes(UTF_8)); + source.listener.accept(null); + + assertEquals(2, received.size()); + assertNotNull(received.get(0)); + assertNull(received.get(1)); + } + + @Test + void rejectsMalformedDocuments() { + final CapturingSource source = new CapturingSource(); + final List received = new ArrayList<>(); + final RemoteConfigurationService service = new RemoteConfigurationService(source); + service.start(received::add); + + final IllegalArgumentException rejection = + assertThrows( + IllegalArgumentException.class, + () -> source.listener.accept("{\"flags\":[]}".getBytes(UTF_8))); + + assertTrue(rejection.getCause() instanceof IOException); + assertTrue(rejection.getMessage().startsWith("Malformed flag configuration: ")); + assertEquals(0, received.size()); + } + + @Test + void closeUnsubscribesOnce() { + final CapturingSource source = new CapturingSource(); + final RemoteConfigurationService service = new RemoteConfigurationService(source); + service.start(configuration -> {}); + service.start(configuration -> {}); + + service.close(); + service.close(); + + assertEquals(1, source.subscriptions); + assertEquals(1, source.closes); + } + + private static final class CapturingSource implements ConfigurationSource { + private Consumer listener; + private int subscriptions; + private int closes; + + @Override + public AutoCloseable subscribe(final Consumer listener) { + this.listener = listener; + this.subscriptions++; + return () -> this.closes++; + } + } +} diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java index a20b965a96c..e5b4760c1b2 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java @@ -52,7 +52,8 @@ public interface Backend { * Subscribes to the Feature Flags Remote Configuration product. * * @param listener the listener receiving raw configuration documents, or {@code null} when the - * configuration is removed. + * configuration is removed. It throws a {@link RuntimeException} to reject a malformed + * document, reported to Remote Configuration as an apply error. * @return the handle to close to unsubscribe. */ AutoCloseable subscribeRemoteConfig(Consumer listener); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java index 837e377dec4..f98eb8ed29d 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java @@ -58,7 +58,12 @@ public AutoCloseable subscribe(final Consumer subscriber) { this.subscribers.add(subscriber); final byte[] document = this.current; if (document != null) { - subscriber.accept(document); + try { + subscriber.accept(document); + } catch (final RuntimeException e) { + // Already reported to Remote Configuration when the document was received. + LOGGER.debug("Feature Flags configuration subscriber rejected the current document", e); + } } } return () -> this.subscribers.remove(subscriber); @@ -76,6 +81,15 @@ public synchronized void close() { this.current = null; } + /** + * Forwards a configuration document to every subscriber. + * + * @param configKey the Remote Configuration key. + * @param content the raw document, or {@code null} when the configuration is removed. + * @param pollingRateHinter the polling rate hinter. + * @throws RuntimeException the first subscriber failure, once every subscriber is notified, so + * the poller reports the document as not applied. + */ @Override public void accept( final String configKey, @@ -86,13 +100,21 @@ public void accept( this.current = content; snapshot = this.subscribers.iterator(); } + RuntimeException failure = null; while (snapshot.hasNext()) { final Consumer subscriber = snapshot.next(); try { subscriber.accept(content); } catch (final RuntimeException e) { - LOGGER.debug("Feature Flags configuration subscriber failed", e); + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } } } + if (failure != null) { + throw failure; + } } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java index 6bac29d44f9..db1f668495a 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java @@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -69,6 +70,56 @@ void forwardsDocumentsAndRemovalsToAllSubscribers() { assertEquals(2, second.size()); } + @Test + void reportsRejectedDocumentsOnceEverySubscriberIsNotified() { + final RemoteConfigServiceImpl service = + new RemoteConfigServiceImpl(mock(ConfigurationPoller.class)); + final IllegalArgumentException rejection = new IllegalArgumentException("malformed"); + final IllegalStateException otherRejection = new IllegalStateException("malformed too"); + final List received = new ArrayList<>(); + service.subscribe( + content -> { + throw rejection; + }); + service.subscribe(received::add); + service.subscribe( + content -> { + throw otherRejection; + }); + + final IllegalArgumentException reported = + assertThrows( + IllegalArgumentException.class, + () -> service.accept("key", DOCUMENT, mock(PollingRateHinter.class))); + + assertSame(rejection, reported); + assertArrayEquals(new Throwable[] {otherRejection}, reported.getSuppressed()); + assertEquals(1, received.size()); + assertArrayEquals(DOCUMENT, received.get(0)); + } + + @Test + void lateSubscribersRejectingTheCurrentDocumentStaySubscribed() throws Exception { + final RemoteConfigServiceImpl service = + new RemoteConfigServiceImpl(mock(ConfigurationPoller.class)); + service.accept("key", DOCUMENT, mock(PollingRateHinter.class)); + final List received = new ArrayList<>(); + + final AutoCloseable subscription = + service.subscribe( + content -> { + if (content != null) { + throw new IllegalArgumentException("malformed"); + } + received.add(null); + }); + service.accept("key", null, mock(PollingRateHinter.class)); + subscription.close(); + + assertEquals(1, received.size()); + assertNull(received.get(0)); + } + @Test void replaysTheCurrentDocumentToLateSubscribers() { final RemoteConfigServiceImpl service =