diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/build.gradle b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/build.gradle index 900c517f4b8..ce6b586c0c1 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/build.gradle +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/build.gradle @@ -2,7 +2,15 @@ plugins { id 'dd-trace-java.module.instrumentation' } -// TODO Add a muzzle pass directive on com.datadoghq:dd-openfeature once the SDK is published. +// The SDK releases with the agent bridge. +// TODO APMLP-1512 Enable once dd-openfeature 1.69.0 is published. +// muzzle { +// pass { +// group = 'com.datadoghq' +// module = 'dd-openfeature' +// versions = '[1.69.0,)' +// } +// } tracerJava { addSourceSetFor(JavaVersion.VERSION_11) 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 20b6256dce2..972858e5a01 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 + ".DisabledAgentBridge", 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 a2608c075cd..eab5e524928 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; @@ -8,10 +11,23 @@ import com.datadog.openfeature.internal.bridge.v1.SpanEnricher; import datadog.trace.api.InstrumenterConfig; 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. + // TODO APMLP-1512 Drop this check once dd-openfeature 1.69.0 is released: the japicmpBridgeV1 + // task then keeps the frozen bridge.v1 SPI compatible with the helpers implementing it. + 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) + && implementsSpi(AgentBridge.class, DisabledAgentBridge.class); + private final FeatureFlaggingGateway.Backend backend; private final HealthMetrics healthMetrics; private final SpanEnricher spanEnricher; @@ -25,11 +41,15 @@ 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, the disabled one if the agent configuration disables Feature Flags, * or the fallback one. */ public static AgentBridge connect(final AgentBridge fallback) { + if (!SPI_COMPATIBLE) { + return fallback; + } if (!InstrumenterConfig.get().isFeatureFlaggingEnabled()) { return DisabledAgentBridge.INSTANCE; } @@ -37,6 +57,25 @@ public static AgentBridge connect(final AgentBridge fallback) { return backend == null ? 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 @Override public String setting(final String key) { @@ -69,7 +108,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/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/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/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()); + } +} 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/build.gradle.kts b/dd-openfeature/build.gradle.kts index 9a8e13e1b68..3d9772ec925 100644 --- a/dd-openfeature/build.gradle.kts +++ b/dd-openfeature/build.gradle.kts @@ -62,7 +62,8 @@ tasks.named("jar") { // The agent bridge SPI versions, with the first SDK release shipping each of them. A released // version is frozen (only default methods can be added): check it against that release. -// TODO Set the v1 baseline once the first SDK release with the agent bridge is published. +// TODO APMLP-1512 Set the v1 baseline to 1.69.0, the first SDK release with the agent bridge, +// once it is published. val bridgeSpiBaselines = mapOf( "v1" to null, ) @@ -101,7 +102,8 @@ val japicmpPublicApi = tasks.register("japicmpPublicApi") { packageIncludes.addAll("com.datadog.openfeature", "datadog.trace.api.openfeature") failOnModification.set(true) failOnSourceIncompatibility.set(true) - // TODO Remove once the baseline is 1.69.0 or later: earlier releases link the agent bootstrap API. + // TODO APMLP-1512 Remove once the baseline is 1.69.0 or later: earlier releases link the agent + // bootstrap API. ignoreMissingClasses.set(true) txtOutputFile.set(layout.buildDirectory.file("reports/japicmp/public-api.txt")) } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index 4466e1a53c6..2a34bf5a990 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -104,7 +104,6 @@ static FeatureFlagsRuntime create(final AgentBridge bridge, final Settings setti final ConfigurationService source; final EventTransport transport; if (remoteConfig) { - // TODO Support Remote Configuration without the Datadog Java agent. final ConfigurationSource remoteConfiguration; try { remoteConfiguration = bridge.remoteConfiguration(); 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/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/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/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-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/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureConfiguration.java b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureConfiguration.java index 8ed5c9083a4..b1b41c0f9be 100644 --- a/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureConfiguration.java +++ b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureConfiguration.java @@ -3,16 +3,25 @@ import com.datadog.openfeature.Provider; import dev.openfeature.sdk.Client; import dev.openfeature.sdk.OpenFeatureAPI; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class OpenFeatureConfiguration { + /** + * @param lazyProvider whether to wait for {@code POST /openfeature/provider} to register the + * provider, instead of registering it at startup. + * @return the OpenFeature client. + */ @Bean - public Client openFeatureClient() { + public Client openFeatureClient( + @Value("${openfeature.lazy-provider:false}") final boolean lazyProvider) { OpenFeatureAPI api = OpenFeatureAPI.getInstance(); - api.setProviderAndWait(new Provider()); + if (!lazyProvider) { + api.setProviderAndWait(new Provider()); + } return api.getClient(); } } diff --git a/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureController.java b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureController.java index 47b5e6b18ab..9599854cbbf 100644 --- a/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureController.java +++ b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/OpenFeatureController.java @@ -26,9 +26,23 @@ public class OpenFeatureController { private static final Logger LOGGER = LoggerFactory.getLogger(OpenFeatureController.class); private final Client client; + private final ProviderRegistration providerRegistration; - public OpenFeatureController(final Client client) { + public OpenFeatureController( + final Client client, final ProviderRegistration providerRegistration) { this.client = client; + this.providerRegistration = providerRegistration; + } + + /** + * Registers the provider without waiting for its initialization. + * + * @return an accepted response. + */ + @PostMapping("/provider") + public ResponseEntity registerProvider() { + this.providerRegistration.request(); + return ResponseEntity.accepted().build(); } @PostMapping( diff --git a/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/ProviderRegistration.java b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/ProviderRegistration.java new file mode 100644 index 00000000000..840243d547e --- /dev/null +++ b/dd-smoke-tests/openfeature/application/src/main/java/datadog/smoketest/springboot/openfeature/ProviderRegistration.java @@ -0,0 +1,36 @@ +package datadog.smoketest.springboot.openfeature; + +import com.datadog.openfeature.Provider; +import dev.openfeature.sdk.OpenFeatureAPI; +import java.util.concurrent.CountDownLatch; +import org.springframework.stereotype.Component; + +/** + * Registers the provider on request, from a thread started outside any request trace, so the + * provider initialization does not capture the trace context of the request. + */ +@Component +public class ProviderRegistration { + private final CountDownLatch requested = new CountDownLatch(1); + + public ProviderRegistration() { + final Thread thread = + new Thread(this::registerWhenRequested, "openfeature-provider-registration"); + thread.setDaemon(true); + thread.start(); + } + + /** Requests the provider registration, without waiting for its initialization. */ + public void request() { + this.requested.countDown(); + } + + private void registerWhenRequested() { + try { + this.requested.await(); + OpenFeatureAPI.getInstance().setProvider(new Provider()); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} 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/OpenFeatureMalformedRemoteConfigSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureMalformedRemoteConfigSmokeTest.groovy new file mode 100644 index 00000000000..3108fe0bb8a --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureMalformedRemoteConfigSmokeTest.groovy @@ -0,0 +1,95 @@ +package datadog.smoketest.springboot + +import datadog.smoketest.AbstractServerSmokeTest +import okhttp3.Request +import okhttp3.RequestBody + +/** + * A malformed flag configuration that Remote Configuration delivers before the SDK subscribes is + * acknowledged by the agent, then reported as not applied once the SDK rejects it. + */ +class OpenFeatureMalformedRemoteConfigSmokeTest extends AbstractServerSmokeTest { + private static final String CONFIG_PATH = 'datadog/2/FFE_FLAGS/1/config' + private static final String MALFORMED_CONFIG = '{"flags":[]}' + private static final int APPLY_STATE_ACKNOWLEDGED = 2 + private static final int APPLY_STATE_ERROR = 3 + + @Override + ProcessBuilder createProcessBuilder() { + setRemoteConfig(CONFIG_PATH, MALFORMED_CONFIG) + + final springBootShadowJar = System.getProperty("datadog.smoketest.springboot.shadowJar.path") + final command = [javaPath()] + command.addAll(defaultJavaProperties) + command.add('-Ddd.remote_config.enabled=true') + command.add("-Ddd.remote_config.url=http://localhost:${server.address.port}/v0.7/config".toString()) + command.addAll([ + '-jar', + springBootShadowJar, + "--server.port=${httpPort}".toString(), + '--openfeature.lazy-provider=true' + ]) + final builder = new ProcessBuilder(command).directory(new File(buildDirectory)) + builder.environment().put('DD_FEATURE_FLAGS_CONFIGURATION_SOURCE', 'remote_config') + return builder + } + + /** The provider never gets a valid configuration, so the OpenFeature SDK logs its failed initialization. */ + @Override + boolean isErrorLog(final String line) { + return super.isErrorLog(line) && !line.contains(' d.o.sdk.') && !line.contains(' dev.openfeature.sdk.') + } + + void 'test malformed configuration received before the SDK is reported as not applied'() { + setup: + // The base resets the Remote Configuration response before each test. + setRemoteConfig(CONFIG_PATH, MALFORMED_CONFIG) + final served = rcClientMessages.size() + + when: 'the agent receives the configuration while no SDK has subscribed' + final acknowledged = waitForRequestAfter(served) { applyState(it) == APPLY_STATE_ACKNOWLEDGED } + final response = client.newCall(new Request.Builder() + .url("http://localhost:${httpPort}/openfeature/provider") + .post(RequestBody.create(null, new byte[0])) + .build()).execute() + + then: 'the SDK rejects it and the agent reports it as not applied' + response.code() == 202 + final rejected = waitForRequestAfter(acknowledged) { applyState(it) == APPLY_STATE_ERROR } + (configState(rejected).apply_error as String).contains('Malformed flag configuration') + + cleanup: + response?.close() + } + + /** + * Waits for an agent Remote Configuration request matching the predicate. + * + * @param from the index of the first request to consider. + * @param predicate the predicate to match. + * @return the index of the first matching request. + */ + private int waitForRequestAfter(final int from, final Closure predicate) { + int index = -1 + defaultPoll.eventually { + index = rcClientMessages.findIndexOf(from, predicate) + assert index >= 0 + } + return index + } + + private Map configState(final int index) { + final request = rcClientMessages[index] + final states = request.client?.state?.config_states as List> + return states?.find { + it.product == 'FFE_FLAGS' + } + } + + private static Integer applyState(final Map request) { + final states = request.client?.state?.config_states as List> + return states?.find { + it.product == 'FFE_FLAGS' + }?.apply_state as Integer + } +} 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 fbc65898777..5187f1f8af1 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 8bdd58f2be8..7df0ec92c7b 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() {} @@ -34,7 +38,11 @@ public static synchronized void start(final SharedCommunicationObjects sco) { return; } LOGGER.debug("Feature Flagging enabled; env={}", config.getEnv()); - final FeatureFlagsBackend backend = new FeatureFlagsBackend(sco, config); + // The tracing shutdown path stops the system before the JVM shutdown hooks, but it is missing + // when tracing or its shutdown instrumentation is disabled. Stop from a shutdown hook as well + // once the SDK subscribes to the shutdown, so it always flushes its pending events. + final FeatureFlagsBackend backend = + new FeatureFlagsBackend(sco, config, FeatureFlaggingSystem::addShutdownHook); try { backend.start(); } catch (final RuntimeException | Error e) { @@ -55,10 +63,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..6bffeb1101e 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,62 @@ 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 keepsAShutdownHookWhenTracingIsEnabled() { + // The tracing shutdown instrumentation can be disabled on its own. + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + FeatureFlaggingGateway.backend().subscribeShutdown(() -> {}); + assertTrue(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..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); @@ -98,5 +99,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/gradle.lockfile b/products/feature-flagging/feature-flagging-lib/gradle.lockfile index 5ebb041416f..17cf71fb002 100644 --- a/products/feature-flagging/feature-flagging-lib/gradle.lockfile +++ b/products/feature-flagging/feature-flagging-lib/gradle.lockfile @@ -60,7 +60,7 @@ org.apache.commons:commons-lang3:3.20.0=spotbugs org.apache.commons:commons-text:1.15.0=spotbugs org.apache.logging.log4j:log4j-api:2.26.1=spotbugs org.apache.logging.log4j:log4j-core:2.26.1=spotbugs -org.apiguardian:apiguardian-api:1.1.2=testCompileClasspath +org.apiguardian:apiguardian-api:1.1.2=testCompileClasspath,testRuntimeClasspath org.codehaus.groovy:groovy-ant:3.0.23=codenarc org.codehaus.groovy:groovy-docgenerator:3.0.23=codenarc org.codehaus.groovy:groovy-groovydoc:3.0.23=codenarc 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..d6f9a72cb78 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().trim().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/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java index a0aae50ed9a..8922955d373 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 @@ -24,6 +24,13 @@ public class RemoteConfigServiceImpl implements ConfigurationChangesTypedListene private final ConfigurationPoller configurationPoller; private final List> subscribers = new CopyOnWriteArrayList<>(); private volatile byte[] current; + + /** + * Whether the current document was received before any subscriber, so the poller acknowledged it + * without validation. Guarded by the subscribers lock. + */ + private boolean unvalidated; + private boolean started; public RemoteConfigServiceImpl(final ConfigurationPoller configurationPoller) { @@ -41,9 +48,13 @@ public synchronized void start() { this.started = true; LOGGER.debug("Feature Flagging starting remote_config configuration source"); this.configurationPoller.addCapabilities(CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); + listen(); + this.configurationPoller.start(); + } + + private void listen() { // Listen as serialized event to prevent depending on the deserializer from agent this.configurationPoller.addListener(FFE_FLAGS, content -> content, this); - this.configurationPoller.start(); } /** @@ -54,18 +65,46 @@ public synchronized void start() { */ public AutoCloseable subscribe(final Consumer subscriber) { start(); + boolean requestAgain = false; // Keep registration and replay atomic with notification snapshots. synchronized (this.subscribers) { this.subscribers.add(subscriber); final byte[] document = this.current; if (document != null) { - subscriber.accept(document); + try { + subscriber.accept(document); + } catch (final RuntimeException e) { + // A rejection of a document received with subscribers is already reported. + requestAgain = this.unvalidated; + LOGGER.debug("Feature Flags configuration subscriber rejected the current document", e); + } + this.unvalidated = false; } } + if (requestAgain) { + requestAgain(); + } return () -> this.subscribers.remove(subscriber); } - /** Unregisters the Feature Flags product and drops the subscribers. */ + /** + * Makes the poller deliver the current document again, so it reports the rejection of a document + * acknowledged before any subscriber validated it. Registering the product again drops its cached + * documents. Runs outside the subscribers lock, as the poller notifies under its own lock. + */ + private synchronized void requestAgain() { + if (!this.started) { + return; + } + LOGGER.debug("Requesting the rejected Feature Flags configuration again to report it"); + this.configurationPoller.removeListeners(FFE_FLAGS); + listen(); + } + + /** + * Unregisters the Feature Flags product, releases the poller started by {@link #start()} and + * drops the subscribers. + */ public synchronized void close() { if (!this.started) { return; @@ -74,10 +113,20 @@ public synchronized void close() { LOGGER.debug("Feature Flagging remote_config configuration source stopped"); this.configurationPoller.removeCapabilities(CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); this.configurationPoller.removeListeners(FFE_FLAGS); + this.configurationPoller.stop(); this.subscribers.clear(); 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,14 +135,20 @@ public void accept( final Iterator> snapshot; synchronized (this.subscribers) { this.current = content; + this.unvalidated = content != null && this.subscribers.isEmpty(); 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); + } } } // Logged last: a diagnostic must not get in the way of the delivery. @@ -105,5 +160,8 @@ public void accept( configKey, content.length); } + if (failure != null) { + throw failure; + } } } 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..9fcbdc11b91 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; @@ -15,6 +16,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.remoteconfig.ConfigurationPoller; import datadog.remoteconfig.Product; @@ -24,7 +26,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 +46,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 +131,62 @@ 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 keepsTheEventProxyAvailableWithABlankApiKey() { + when(this.config.getFeatureFlaggingConfigurationSource()) + .thenReturn(FeatureFlaggingSource.AGENTLESS); + when(this.config.getApiKey()).thenReturn(" \t "); + when(this.sco.featuresDiscovery(this.config)).thenReturn(mock(DDAgentFeaturesDiscovery.class)); + + final FeatureFlagsBackend created = new FeatureFlagsBackend(this.sco, this.config, null); + + assertTrue( + created.isEventProxyAvailable(), + "the proxy stays available for recovery when direct intake has no usable 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); 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 ec70db6d9f8..f4083331c6c 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 @@ -7,6 +7,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; 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; @@ -79,6 +80,110 @@ 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 + @SuppressWarnings("unchecked") + void requestsADocumentReceivedBeforeAnySubscriberAgainWhenRejected() { + final ConfigurationPoller poller = mock(ConfigurationPoller.class); + final RemoteConfigServiceImpl service = new RemoteConfigServiceImpl(poller); + service.start(); + service.accept("key", DOCUMENT, mock(PollingRateHinter.class)); + + service.subscribe( + content -> { + throw new IllegalArgumentException("malformed"); + }); + service.subscribe( + content -> { + throw new IllegalArgumentException("malformed"); + }); + + verify(poller, times(1)).removeListeners(Product.FFE_FLAGS); + verify(poller, times(2)) + .addListener(eq(Product.FFE_FLAGS), any(ConfigurationDeserializer.class), eq(service)); + } + + @Test + void keepsADocumentReceivedBeforeAnySubscriberWhenAccepted() { + final ConfigurationPoller poller = mock(ConfigurationPoller.class); + final RemoteConfigServiceImpl service = new RemoteConfigServiceImpl(poller); + service.start(); + service.accept("key", DOCUMENT, mock(PollingRateHinter.class)); + + service.subscribe(content -> {}); + + verify(poller, never()).removeListeners(any()); + } + + @Test + void doesNotRequestAgainADocumentAlreadyRejectedOnDelivery() { + final ConfigurationPoller poller = mock(ConfigurationPoller.class); + final RemoteConfigServiceImpl service = new RemoteConfigServiceImpl(poller); + service.subscribe( + content -> { + throw new IllegalArgumentException("malformed"); + }); + assertThrows( + IllegalArgumentException.class, + () -> service.accept("key", DOCUMENT, mock(PollingRateHinter.class))); + + service.subscribe( + content -> { + throw new IllegalArgumentException("malformed"); + }); + + verify(poller, never()).removeListeners(any()); + } + @Test void replaysTheCurrentDocumentToLateSubscribers() { final RemoteConfigServiceImpl service = @@ -185,10 +290,24 @@ void closeUnregistersTheProductAndDropsSubscribers() { verify(poller, times(1)) .removeCapabilities(Capabilities.CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); verify(poller, times(1)).removeListeners(Product.FFE_FLAGS); - verify(poller, never()).stop(); + verify(poller, times(1)).stop(); assertEquals(0, received.size()); } + @Test + void restartsBalancePollerReferences() { + final ConfigurationPoller poller = mock(ConfigurationPoller.class); + final RemoteConfigServiceImpl service = new RemoteConfigServiceImpl(poller); + + service.start(); + service.close(); + service.subscribe(content -> {}); + service.close(); + + verify(poller, times(2)).start(); + verify(poller, times(2)).stop(); + } + @Nested class Diagnostics { private final Logger logger = (Logger) LoggerFactory.getLogger(RemoteConfigServiceImpl.class);