diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java index 8fa3abcb5e7..e578e3824fc 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentation.java @@ -41,6 +41,7 @@ public String[] helperClassNames() { packageName + ".AgentConfigurationSource", packageName + ".AgentEventProxy", packageName + ".AgentHealthMetrics", + packageName + ".AgentShutdownSource", packageName + ".AgentSpanEnricher", packageName + ".JavaAgentBridge", }; diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java new file mode 100644 index 00000000000..9e73869ded4 --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/AgentShutdownSource.java @@ -0,0 +1,18 @@ +package datadog.trace.instrumentation.openfeature; + +import com.datadog.openfeature.internal.bridge.v1.ShutdownSource; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; + +/** Notifies the agent shutdown, before the JVM shutdown hooks run. */ +public final class AgentShutdownSource implements ShutdownSource { + private final FeatureFlaggingGateway.Backend backend; + + AgentShutdownSource(final FeatureFlaggingGateway.Backend backend) { + this.backend = backend; + } + + @Override + public AutoCloseable subscribe(final Runnable listener) { + return this.backend.subscribeShutdown(listener); + } +} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java index a0122636ff5..d175aee6c44 100644 --- a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/main/java11/datadog/trace/instrumentation/openfeature/JavaAgentBridge.java @@ -1,5 +1,8 @@ package datadog.trace.instrumentation.openfeature; +import static java.lang.reflect.Modifier.isAbstract; +import static java.lang.reflect.Modifier.isStatic; + import com.datadog.openfeature.internal.bridge.v1.AgentBridge; import com.datadog.openfeature.internal.bridge.v1.ConfigurationSource; import com.datadog.openfeature.internal.bridge.v1.EventTransport; @@ -7,10 +10,20 @@ import com.datadog.openfeature.internal.bridge.v1.ShutdownSource; import com.datadog.openfeature.internal.bridge.v1.SpanEnricher; import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import java.lang.reflect.Method; import javax.annotation.Nullable; /** The bridge backed by the agent Feature Flags backend. */ public final class JavaAgentBridge implements AgentBridge { + // Muzzle checks calls, but not the SPI methods implemented by these helpers. + private static final boolean SPI_COMPATIBLE = + implementsSpi(AgentBridge.class, JavaAgentBridge.class) + && implementsSpi(ConfigurationSource.class, AgentConfigurationSource.class) + && implementsSpi(EventTransport.class, AgentEventProxy.class) + && implementsSpi(HealthMetrics.class, AgentHealthMetrics.class) + && implementsSpi(SpanEnricher.class, AgentSpanEnricher.class) + && implementsSpi(ShutdownSource.class, AgentShutdownSource.class); + private final FeatureFlaggingGateway.Backend backend; private final HealthMetrics healthMetrics; private final SpanEnricher spanEnricher; @@ -24,12 +37,32 @@ public final class JavaAgentBridge implements AgentBridge { /** * Connects the SDK to the agent backend. * - * @param fallback the bridge to use if the agent backend is not started. + * @param fallback the bridge to use if the agent backend is not started or the SDK SPI is + * incompatible. * @return the agent bridge, or the fallback one. */ public static AgentBridge connect(final AgentBridge fallback) { final FeatureFlaggingGateway.Backend backend = FeatureFlaggingGateway.backend(); - return backend == null ? fallback : new JavaAgentBridge(backend); + return backend == null || !SPI_COMPATIBLE ? fallback : new JavaAgentBridge(backend); + } + + private static boolean implementsSpi(final Class spi, final Class implementation) { + try { + for (final Method method : spi.getMethods()) { + if (!isStatic(method.getModifiers())) { + final Method implemented = + implementation.getMethod(method.getName(), method.getParameterTypes()); + if (isAbstract(implemented.getModifiers()) + || isStatic(implemented.getModifiers()) + || implemented.getReturnType() != method.getReturnType()) { + return false; + } + } + } + return true; + } catch (final NoSuchMethodException e) { + return false; + } } @Nullable @@ -64,7 +97,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/DDOpenFeatureClassLoaderTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java new file mode 100644 index 00000000000..2c9999641c2 --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureClassLoaderTest.java @@ -0,0 +1,197 @@ +package datadog.trace.instrumentation.openfeature; + +import static java.nio.file.Files.createDirectories; +import static java.nio.file.Files.write; +import static net.bytebuddy.jar.asm.Opcodes.ACC_ABSTRACT; +import static net.bytebuddy.jar.asm.Opcodes.ACC_PUBLIC; +import static net.bytebuddy.jar.asm.Opcodes.ACONST_NULL; +import static net.bytebuddy.jar.asm.Opcodes.ARETURN; +import static net.bytebuddy.jar.asm.Opcodes.ASM9; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datadog.openfeature.internal.bridge.v1.AgentBridge; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import java.io.IOException; +import java.io.InputStream; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.file.Path; +import java.util.function.UnaryOperator; +import net.bytebuddy.jar.asm.ClassReader; +import net.bytebuddy.jar.asm.ClassVisitor; +import net.bytebuddy.jar.asm.ClassWriter; +import net.bytebuddy.jar.asm.MethodVisitor; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.tabletest.junit.TableTest; + +class DDOpenFeatureClassLoaderTest extends AbstractInstrumentationTest { + @TempDir Path sdkOverrides; + + @AfterEach + void unregisterBackend() { + FeatureFlaggingGateway.register(null); + } + + @Test + void injectsIntoAnIsolatedSdkClassLoader() throws Exception { + final FeatureFlaggingGateway.Backend backend = mock(FeatureFlaggingGateway.Backend.class); + when(backend.setting("site")).thenReturn("datadoghq.eu"); + FeatureFlaggingGateway.register(backend); + + try (URLClassLoader loader = sdkClassLoader()) { + final Object bridge = detect(loader); + assertSame(loader, bridge.getClass().getClassLoader()); + assertEquals(JavaAgentBridge.class.getName(), bridge.getClass().getName()); + assertEquals( + "datadoghq.eu", + loader + .loadClass(AgentBridge.class.getName()) + .getMethod("setting", String.class) + .invoke(bridge, "site")); + } + } + + @TableTest({ + "Scenario | SPI | Method | Return type ", + "settings changed | AgentBridge | setting | ", + "configuration changed | ConfigurationSource | subscribe | ", + "transport changed | EventTransport | post | ", + "metrics changed | HealthMetrics | count | ", + "enrichment changed | SpanEnricher | serialId | ", + "shutdown changed | ShutdownSource | subscribe | ", + "return type changed | AgentBridge | setting | Ljava/lang/Object;" + }) + void keepsTheStandaloneBridgeWhenAnSpiMethodChanges( + final String spi, final String method, final String returnType) throws Exception { + rewriteSpi( + spi, + bytes -> { + final ClassWriter writer = new ClassWriter(0); + new ClassReader(bytes) + .accept( + new ClassVisitor(ASM9, writer) { + @Override + public MethodVisitor visitMethod( + int access, + String name, + String descriptor, + String signature, + String[] exceptions) { + if (name.equals(method)) { + if (returnType == null) { + name = "renamed"; + } else { + descriptor = + descriptor.substring(0, descriptor.indexOf(')') + 1) + returnType; + } + } + return super.visitMethod(access, name, descriptor, signature, exceptions); + } + }, + 0); + return writer.toByteArray(); + }); + FeatureFlaggingGateway.register(mock(FeatureFlaggingGateway.Backend.class)); + + try (URLClassLoader loader = sdkClassLoader()) { + // Implemented methods are not calls, so Muzzle alone accepts this incompatible SPI. + assertTrue(new DDOpenFeatureInstrumentation().getInstrumentationMuzzle().matches(loader)); + assertSame( + loader.loadClass(AgentBridge.class.getName()).getField("NONE").get(null), detect(loader)); + } + } + + @TableTest({ + "Scenario | Abstract | Static conflict", + "new abstract method | true | false ", + "new default method | false | false ", + "abstract conflicts with static | true | true ", + "default conflicts with static | false | true " + }) + void checksNewSpiMethods(final boolean isAbstract, final boolean staticConflict) + throws Exception { + rewriteSpi( + "AgentBridge", + bytes -> { + final ClassWriter writer = new ClassWriter(0); + new ClassReader(bytes) + .accept( + new ClassVisitor(ASM9, writer) { + @Override + public void visitEnd() { + final MethodVisitor method = + super.visitMethod( + ACC_PUBLIC | (isAbstract ? ACC_ABSTRACT : 0), + staticConflict ? "connect" : "additionalMethod", + staticConflict + ? "(Lcom/datadog/openfeature/internal/bridge/v1/AgentBridge;)" + + "Lcom/datadog/openfeature/internal/bridge/v1/AgentBridge;" + : "()Ljava/lang/String;", + null, + null); + if (!isAbstract) { + method.visitCode(); + if (staticConflict) { + method.visitInsn(ACONST_NULL); + } else { + method.visitLdcInsn("compatible"); + } + method.visitInsn(ARETURN); + method.visitMaxs(1, staticConflict ? 2 : 1); + } + method.visitEnd(); + super.visitEnd(); + } + }, + 0); + return writer.toByteArray(); + }); + FeatureFlaggingGateway.register(mock(FeatureFlaggingGateway.Backend.class)); + + try (URLClassLoader loader = sdkClassLoader()) { + final Class contract = loader.loadClass(AgentBridge.class.getName()); + final Object bridge = detect(loader); + if (isAbstract || staticConflict) { + assertSame(contract.getField("NONE").get(null), bridge); + } else { + assertEquals(JavaAgentBridge.class.getName(), bridge.getClass().getName()); + assertEquals("compatible", contract.getMethod("additionalMethod").invoke(bridge)); + } + } + } + + private URLClassLoader sdkClassLoader() throws IOException { + return new URLClassLoader( + new URL[] { + sdkOverrides.toUri().toURL(), + AgentBridge.class.getProtectionDomain().getCodeSource().getLocation() + }, + null); + } + + private static Object detect(final ClassLoader loader) throws Exception { + return loader + .loadClass("com.datadog.openfeature.internal.bridge.v1.AgentBridges") + .getMethod("detect") + .invoke(null); + } + + private void rewriteSpi(final String spi, final UnaryOperator rewrite) + throws IOException { + final String resource = "com/datadog/openfeature/internal/bridge/v1/" + spi + ".class"; + final Path target = sdkOverrides.resolve(resource); + createDirectories(target.getParent()); + try (InputStream input = AgentBridge.class.getClassLoader().getResourceAsStream(resource)) { + assertNotNull(input); + write(target, rewrite.apply(input.readAllBytes())); + } + } +} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java new file mode 100644 index 00000000000..31485765b7d --- /dev/null +++ b/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/DDOpenFeatureInstrumentationNoTracingForkedTest.java @@ -0,0 +1,6 @@ +package datadog.trace.instrumentation.openfeature; + +import datadog.trace.test.junit.utils.config.WithConfig; + +@WithConfig(key = "trace.enabled", value = "false") +class DDOpenFeatureInstrumentationNoTracingForkedTest extends DDOpenFeatureInstrumentationTest {} diff --git a/dd-java-agent/instrumentation/datadog/openfeature/dd-openfeature-1.69/src/test/java/datadog/trace/instrumentation/openfeature/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/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/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy new file mode 100644 index 00000000000..240131577d1 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/AbstractOpenFeatureShutdownSmokeTest.groovy @@ -0,0 +1,135 @@ +package datadog.smoketest.springboot + +import static datadog.trace.agent.test.server.http.TestHttpServer.httpServer +import static java.util.concurrent.TimeUnit.SECONDS + +import datadog.smoketest.AbstractSmokeTest +import datadog.trace.agent.test.server.http.TestHttpServer +import datadog.trace.agent.test.server.http.TestHttpServer.HandlerApi.RequestApi +import datadog.trace.agent.test.utils.OkHttpUtils +import datadog.trace.agent.test.utils.PortUtils +import groovy.json.JsonOutput +import groovy.json.JsonSlurper +import okhttp3.MediaType +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.RequestBody +import spock.lang.AutoCleanup +import spock.lang.Shared +import spock.util.concurrent.PollingConditions + +/** The SDK with the Datadog Java agent flushes its pending events when the application exits. */ +abstract class AbstractOpenFeatureShutdownSmokeTest extends AbstractSmokeTest { + + @Shared + protected int httpPort = PortUtils.randomOpenPort() + + protected OkHttpClient client = OkHttpUtils.client() + + @Shared + protected final ufcConfig = new JsonSlurper().parse(fetchResource("ffe-system-test-data/ufc-config.json")) as Map + + /** Serves the flag configuration as the Datadog managed CDN does. */ + @Shared + @AutoCleanup + protected TestHttpServer cdn = httpServer { + handlers { + prefix("/api/v2/feature-flagging/config/rules-based/server") { + response.status(200).send(JsonOutput.toJson([ + data: [type: 'universal-flag-configuration', attributes: ufcConfig] + ])) + } + } + } + + /** @return whether the application runs with tracing enabled. */ + abstract boolean tracing() + + def setupSpec() { + PortUtils.waitForPortToOpen(httpPort, 240, SECONDS, testedProcess) + } + + @Override + ProcessBuilder createProcessBuilder() { + final springBootShadowJar = System.getProperty("datadog.smoketest.springboot.shadowJar.path") + final command = [javaPath()] + command.addAll(defaultJavaProperties) + command.add("-Ddd.trace.enabled=${tracing()}".toString()) + command.addAll(['-jar', springBootShadowJar, "--server.port=${httpPort}".toString()]) + final builder = new ProcessBuilder(command).directory(new File(buildDirectory)) + builder.environment().put('DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL', "http://localhost:${cdn.address.port}".toString()) + return builder + } + + @Override + boolean testTelemetry() { + return false + } + + @Override + Closure decodedTracesCallback() { + return {} + } + + @Override + Closure decodedEvpProxyMessageCallback() { + return { String path, RequestApi request -> + if (!path.contains('api/v2/flagevaluation')) { + return null + } + return [ + origin: request.getHeader('DD-EVP-ORIGIN'), + originVersion: request.getHeader('DD-EVP-ORIGIN-VERSION'), + body: new JsonSlurper().parse(request.body) + ] + } + } + + void 'test pending flag evaluations are flushed at application exit'() { + when: + // The first evaluation is flushed within the 10 seconds flush interval, which then restarts: + // the next evaluation stays pending until the application exits. + evaluate('test-case-boolean-false-assignment.json') + new PollingConditions(timeout: 20).eventually { + assert evaluatedFlags().contains('boolean-false-assignment') + } + evaluate('test-case-boolean-one-of-matches.json') + + then: + !evaluatedFlags().contains('boolean-one-of-matches') + + when: + testedProcess.destroy() + + then: + testedProcess.waitFor(30, SECONDS) + evaluatedFlags().contains('boolean-one-of-matches') + evaluationRequests().every { it.origin == 'dd-trace-java' && it.originVersion } + } + + protected void evaluate(final String testCaseFile) { + final testCase = (new JsonSlurper().parse(fetchResource("ffe-system-test-data/evaluation-cases/${testCaseFile}")) as List>).first() + final request = new Request.Builder() + .url("http://localhost:${httpPort}/openfeature/evaluate") + .post(RequestBody.create(MediaType.parse('application/json'), JsonOutput.toJson(testCase))) + .build() + final response = client.newCall(request).execute() + try { + assert response.code() == 200 + } finally { + response.close() + } + } + + protected List evaluatedFlags() { + return evaluationRequests()*.body*.flagEvaluations.flatten()*.flag*.key as List + } + + protected List> evaluationRequests() { + return evpProxyMessages*.getV2() as List> + } + + protected static URL fetchResource(final String name) { + return Thread.currentThread().getContextClassLoader().getResource(name) + } +} diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy new file mode 100644 index 00000000000..70192aa9331 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureShutdownSmokeTest.groovy @@ -0,0 +1,10 @@ +package datadog.smoketest.springboot + +/** The tracing shutdown hook flushes the SDK pending events. */ +class OpenFeatureShutdownSmokeTest extends AbstractOpenFeatureShutdownSmokeTest { + + @Override + boolean tracing() { + return true + } +} diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy new file mode 100644 index 00000000000..c300962d260 --- /dev/null +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureTraceDisabledShutdownSmokeTest.groovy @@ -0,0 +1,13 @@ +package datadog.smoketest.springboot + +/** + * Without tracing, the Feature Flagging system registers its own shutdown hook to flush the SDK + * pending events. + */ +class OpenFeatureTraceDisabledShutdownSmokeTest extends AbstractOpenFeatureShutdownSmokeTest { + + @Override + boolean tracing() { + return false + } +} diff --git a/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy b/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy index 28b71306349..0afa31b8809 100644 --- a/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy +++ b/dd-smoke-tests/src/main/groovy/datadog/smoketest/AbstractSmokeTest.groovy @@ -96,7 +96,9 @@ abstract class AbstractSmokeTest extends ProcessManager { "Accept-Encoding", "Content-Encoding", "User-Agent", - "DD-CI-PROVIDER-NAME" + "DD-CI-PROVIDER-NAME", + "DD-EVP-ORIGIN", + "DD-EVP-ORIGIN-VERSION" ], "config": {} }""") diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java index 9a232111344..a2e6a5378f4 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java @@ -13,5 +13,20 @@ public final class FeatureFlaggingConfig { public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE = "feature.flags.configuration.source"; + /** + * Turns off sending exposures to Datadog while customer exposure hooks keep receiving them. + * Default: enabled. Maps to {@code DD_FEATURE_FLAGS_EXPOSURES_DATADOG_LOGGING_ENABLED}. + */ + public static final String FEATURE_FLAGS_EXPOSURES_DATADOG_LOGGING_ENABLED = + "feature.flags.exposures.datadog.logging.enabled"; + + /** + * Lets the {@code dd_evaluation_telemetry_opt_out} evaluation context attribute opt subjects out + * of evaluation telemetry. Default: disabled. Maps to {@code + * DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED}. + */ + public static final String FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED = + "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; + private FeatureFlaggingConfig() {} } diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index 0b971d3f6f0..300a0d36861 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -1,5 +1,8 @@ package com.datadog.featureflag; +import static datadog.trace.util.AgentThreadFactory.AgentThread.FEATURE_FLAG_SHUTDOWN; +import static datadog.trace.util.AgentThreadFactory.newAgentThread; + import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; @@ -16,6 +19,7 @@ public class FeatureFlaggingSystem { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class); private static volatile FeatureFlagsBackend BACKEND; + private static volatile Thread SHUTDOWN_HOOK; private FeatureFlaggingSystem() {} @@ -33,7 +37,12 @@ public static synchronized void start(final SharedCommunicationObjects sco) { LOGGER.debug("Feature Flagging system disabled"); return; } - final FeatureFlagsBackend backend = new FeatureFlagsBackend(sco, config); + // Tracing owns the shutdown path running before the JVM shutdown hooks. Without it, stop from a + // shutdown hook once the SDK subscribes to the shutdown, so it still flushes its pending + // events. + final FeatureFlagsBackend backend = + new FeatureFlagsBackend( + sco, config, config.isTraceEnabled() ? null : FeatureFlaggingSystem::addShutdownHook); try { backend.start(); } catch (final RuntimeException | Error e) { @@ -52,10 +61,37 @@ public static synchronized void start(final SharedCommunicationObjects sco) { public static synchronized void stop() { final FeatureFlagsBackend backend = BACKEND; BACKEND = null; + final Thread shutdownHook = SHUTDOWN_HOOK; + SHUTDOWN_HOOK = null; + if (shutdownHook != null && shutdownHook != Thread.currentThread()) { + try { + Runtime.getRuntime().removeShutdownHook(shutdownHook); + } catch (final IllegalStateException ignored) { + // The JVM is already shutting down; the hook finds the system stopped. + } + } FeatureFlaggingGateway.register(null); if (backend != null) { backend.close(); } LOGGER.debug("Feature Flagging system stopped"); } + + private static synchronized void addShutdownHook() { + if (BACKEND == null || SHUTDOWN_HOOK != null) { + return; + } + final Thread hook = newAgentThread(FEATURE_FLAG_SHUTDOWN, FeatureFlaggingSystem::stop); + try { + Runtime.getRuntime().addShutdownHook(hook); + } catch (final IllegalStateException | SecurityException e) { + LOGGER.debug("Could not register the Feature Flagging shutdown hook", e); + return; + } + SHUTDOWN_HOOK = hook; + } + + static boolean isShutdownHookRegistered() { + return SHUTDOWN_HOOK != null; + } } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index 0935d1914cb..c2a265544d1 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -3,10 +3,13 @@ import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED; import static datadog.trace.api.config.RemoteConfigConfig.REMOTE_CONFIGURATION_ENABLED; +import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_ENABLED; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; @@ -19,6 +22,7 @@ import datadog.remoteconfig.Product; import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.test.junit.utils.config.WithConfig; +import java.lang.reflect.Field; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -65,6 +69,61 @@ void failedStartRollsBackTheRemoteConfigRegistration() { verify(poller).removeListeners(Product.FFE_FLAGS); } + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void stopsFromAShutdownHookOnceTheSdkSubscribesWhenTracingIsDisabled() { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + assertFalse( + FeatureFlaggingSystem.isShutdownHookRegistered(), "no hook while the SDK is unused"); + + FeatureFlaggingGateway.backend().subscribeShutdown(() -> {}); + assertTrue(FeatureFlaggingSystem.isShutdownHookRegistered()); + + FeatureFlaggingSystem.stop(); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void shutdownHookFlushesListenersAndStopsTheBackend() throws Exception { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + final Runnable listener = mock(Runnable.class); + FeatureFlaggingGateway.backend().subscribeShutdown(listener); + + final Field field = FeatureFlaggingSystem.class.getDeclaredField("SHUTDOWN_HOOK"); + field.setAccessible(true); + final Thread hook = (Thread) field.get(null); + // Run the real hook without shutting down the test JVM. + assertTrue(Runtime.getRuntime().removeShutdownHook(hook)); + hook.start(); + hook.join(5000); + + assertFalse(hook.isAlive(), "the shutdown hook must finish"); + assertNull(FeatureFlaggingGateway.backend()); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + verify(listener).run(); + } + + @Test + @WithConfig(key = TRACE_ENABLED, value = "false") + void lateShutdownSubscriptionDoesNotRegisterAHookAfterStop() throws Exception { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + final FeatureFlaggingGateway.Backend backend = FeatureFlaggingGateway.backend(); + FeatureFlaggingSystem.stop(); + + try (final AutoCloseable ignored = backend.subscribeShutdown(() -> {})) { + assertNull(FeatureFlaggingGateway.backend()); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + } + + @Test + void leavesTheShutdownToTracingWhenEnabled() { + FeatureFlaggingSystem.start(mock(SharedCommunicationObjects.class)); + FeatureFlaggingGateway.backend().subscribeShutdown(() -> {}); + assertFalse(FeatureFlaggingSystem.isShutdownHookRegistered()); + } + @Test @WithConfig(key = FEATURE_FLAGS_ENABLED, value = "false") void startDoesNothingWhenFeatureFlagsIsDisabled() { diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java index 1f1cfcaca76..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/src/main/java/com/datadog/featureflag/EventProxy.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java index ccc28f1e858..6b00fca54e4 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/EventProxy.java @@ -2,45 +2,108 @@ import static com.datadog.featureflag.FeatureFlagEventType.EXPOSURE; import static com.datadog.featureflag.FeatureFlagEventType.FLAG_EVALUATION; +import static datadog.communication.EvpProxy.JAVA_TRACING_LIBRARY; +import static datadog.communication.EvpProxy.ORIGIN_HEADER; +import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; import static datadog.communication.http.HttpRetryPolicy.Factory.NEVER_RETRY; import static datadog.trace.api.intake.Intake.EVENT_PLATFORM; +import static java.util.Arrays.asList; +import static java.util.Collections.unmodifiableMap; +import static java.util.concurrent.TimeUnit.MINUTES; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.HttpResponseException; +import datadog.communication.ddagent.TracerVersion; +import datadog.communication.http.HttpRetryPolicy; import java.io.IOException; import java.net.ConnectException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.LongSupplier; import javax.annotation.Nullable; import okhttp3.MediaType; import okhttp3.RequestBody; +import okhttp3.Response; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -/** Posts Feature Flags events through the Datadog Agent event platform proxy. */ +/** + * Posts Feature Flags events through the Datadog Agent event platform proxy. + * + *

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

Payloads are sent once: only a connection refusal, proven to happen before sending, is + * retried, and only with the Remote Configuration source. + */ class EventProxy { + private static final Logger LOGGER = LoggerFactory.getLogger(EventProxy.class); private static final MediaType JSON = MediaType.parse("application/json"); static final String EXPOSURES_ROUTE = "exposures"; + static final long RECOVERY_INTERVAL_NANOS = MINUTES.toNanos(1); + + /** Identifies the library sending the events, on every proxy request. */ + static final Map IDENTITY_HEADERS = identityHeaders(); + + /** The headers a discovered proxy endpoint must forward to keep the event identity. */ + static final List IDENTITY_HEADER_NAMES = asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER); + + /** Retries connection refusal only; automatic retries and redirects remain disabled. */ + static final HttpRetryPolicy.Factory PRE_SEND_RETRY = + new HttpRetryPolicy.Factory(3, 100, 2.0) { + @Override + public HttpRetryPolicy create() { + return new PreSendRetryPolicy(); + } + }; private final BackendApiFactory backendApiFactory; + private final boolean agentless; private final boolean directFallbackAvailable; - private volatile BackendApi exposures; - private volatile BackendApi flagEvaluations; - private volatile boolean resolved; + private final LongSupplier nanoTime; + private volatile Clients clients; + private boolean discovered; + private long nextDiscoveryNanos; /** - * @param backendApiFactory the factory of event platform proxy clients. + * @param backendApiFactory the send-once factory of event platform proxy clients. + * @param agentless whether the configuration source is agentless, so events use the discovered + * proxy endpoint, or direct intake when the SDK can fall back to it. * @param directFallbackAvailable whether the SDK falls back to direct intake when the proxy is - * unavailable, in which case the proxy must not retry what the intake would accept. + * unavailable. */ - EventProxy(final BackendApiFactory backendApiFactory, final boolean directFallbackAvailable) { + EventProxy( + final BackendApiFactory backendApiFactory, + final boolean agentless, + final boolean directFallbackAvailable) { + this(backendApiFactory, agentless, directFallbackAvailable, System::nanoTime); + } + + EventProxy( + final BackendApiFactory backendApiFactory, + final boolean agentless, + final boolean directFallbackAvailable, + final LongSupplier nanoTime) { this.backendApiFactory = backendApiFactory; + this.agentless = agentless; this.directFallbackAvailable = directFallbackAvailable; + this.nanoTime = nanoTime; } /** - * @return whether the Datadog Agent supports the event platform proxy. + * @return whether events go through the event platform proxy: always with the Remote + * Configuration source, and with the agentless source when the proxy endpoint is discovered + * or when direct intake is not available. */ boolean isAvailable() { - resolve(); - return this.exposures != null; + // Only the agentless source can lack proxy clients. + return current() != null || !this.directFallbackAvailable; } /** @@ -48,57 +111,167 @@ boolean isAvailable() { * * @param route the event platform route. * @param json the UTF-8 JSON payload. - * @return {@code true} if delivered, {@code false} if the proxy is definitively unavailable. + * @return {@code true} if delivered, {@code false} if the proxy definitively did not receive the + * payload. * @throws IOException if the payload could not be delivered. */ boolean post(final String route, final byte[] json) throws IOException { - resolve(); - final BackendApi api = EXPOSURES_ROUTE.equals(route) ? this.exposures : this.flagEvaluations; - if (api == null) { - return false; + Clients current = current(); + if (current == null) { + current = recover(); + if (current == null) { + return false; + } } + final BackendApi api = + EXPOSURES_ROUTE.equals(route) ? current.exposures : current.flagEvaluations; try { api.post(route, RequestBody.create(JSON, json), stream -> null, null, false); return true; } catch (final IOException e) { if (isDefinitiveRejection(e)) { + onDefinitiveRejection(current); return false; } throw e; } } - private void resolve() { - if (this.resolved) { - return; + @Nullable + private Clients current() { + final Clients current = this.clients; + if (current != null) { + return current; } synchronized (this) { - if (this.resolved) { - return; + if (!this.discovered) { + this.discovered = true; + this.clients = + this.agentless ? discover(false) : clients(PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + if (this.clients == null) { + onUnavailable(); + } } - this.exposures = create(EXPOSURE); - this.flagEvaluations = create(FLAG_EVALUATION); - this.resolved = true; + return this.clients; } } @Nullable - private BackendApi create(final FeatureFlagEventType eventType) { - return this.directFallbackAvailable - ? this.backendApiFactory.createEvpProxyApi( - EVENT_PLATFORM, eventType.responseCompressionEnabled(), NEVER_RETRY) - : this.backendApiFactory.createEvpProxyApi( - EVENT_PLATFORM, eventType.responseCompressionEnabled()); + private synchronized Clients recover() { + if (this.clients != null) { + return this.clients; + } + if (this.directFallbackAvailable || this.nanoTime.getAsLong() - this.nextDiscoveryNanos < 0) { + return null; + } + this.clients = discover(true); + if (this.clients == null) { + scheduleDiscovery(); + } else { + LOGGER.info("Feature Flags event delivery recovered through the Datadog Agent event proxy"); + } + return this.clients; } + private synchronized void onDefinitiveRejection(final Clients failed) { + // Only the agentless proxy route without direct intake waits for a new endpoint. An in-flight + // send on an old endpoint cannot invalidate a newly recovered one. + if (this.agentless && !this.directFallbackAvailable && this.clients == failed) { + this.clients = null; + onUnavailable(); + } + } + + /** Waits for a new proxy endpoint when it is the only route. Runs once per outage. */ + private void onUnavailable() { + if (!this.directFallbackAvailable) { + scheduleDiscovery(); + LOGGER.warn( + "Feature Flags events are being dropped because no local EVP proxy or direct intake" + + " route is available. Make the Agent reachable, or configure DD_API_KEY and a" + + " valid DD_SITE."); + } + } + + private void scheduleDiscovery() { + this.nextDiscoveryNanos = this.nanoTime.getAsLong() + RECOVERY_INTERVAL_NANOS; + } + + @Nullable + private Clients discover(final boolean force) { + final String endpoint; + try { + endpoint = this.backendApiFactory.discoverEvpProxyEndpoint(force, IDENTITY_HEADER_NAMES); + } catch (final RuntimeException e) { + LOGGER.debug("Could not discover the Datadog Agent event proxy", e); + return null; + } + return endpoint == null ? null : clients(NEVER_RETRY, endpoint); + } + + private Clients clients(final HttpRetryPolicy.Factory retryPolicy, final String endpoint) { + return new Clients( + this.backendApiFactory.createEvpProxyApiForEndpoint( + EVENT_PLATFORM, EXPOSURE.responseCompressionEnabled(), retryPolicy, endpoint), + this.backendApiFactory.createEvpProxyApiForEndpoint( + EVENT_PLATFORM, FLAG_EVALUATION.responseCompressionEnabled(), retryPolicy, endpoint)); + } + + private static Map identityHeaders() { + final Map headers = new HashMap<>(2); + headers.put(ORIGIN_HEADER, JAVA_TRACING_LIBRARY); + headers.put(ORIGIN_VERSION_HEADER, TracerVersion.TRACER_VERSION); + return unmodifiableMap(headers); + } + + /** + * Tells whether the proxy definitively did not receive a payload, so it can be replayed through + * direct intake. + * + * @param exception the post failure. + * @return whether the failure is a connection refusal or a missing proxy route. + */ static boolean isDefinitiveRejection(final IOException exception) { if (exception instanceof ConnectException) { return true; } if (exception instanceof HttpResponseException) { final int statusCode = ((HttpResponseException) exception).getStatusCode(); - return statusCode == 403 || statusCode == 404 || statusCode == 405; + return statusCode == 404 || statusCode == 405; } return false; } + + /** The proxy clients of both routes, for one endpoint. */ + private static final class Clients { + final BackendApi exposures; + final BackendApi flagEvaluations; + + Clients(final BackendApi exposures, final BackendApi flagEvaluations) { + this.exposures = exposures; + this.flagEvaluations = flagEvaluations; + } + } + + private static final class PreSendRetryPolicy extends HttpRetryPolicy { + private int attemptsLeft = 3; + + private PreSendRetryPolicy() { + super(3, 100, 2.0, false); + } + + @Override + public boolean shouldRetry(final Exception exception) { + if (exception instanceof ConnectException && this.attemptsLeft > 0) { + this.attemptsLeft--; + return true; + } + return false; + } + + @Override + public boolean shouldRetry(@Nullable final Response response) { + return false; + } + } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java index 0fdc3e3d712..10d2d69837f 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/FeatureFlagsBackend.java @@ -13,11 +13,17 @@ import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.telemetry.CoreMetricCollector; import java.io.IOException; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** The agent services exposed to the Feature Flags SDK instrumentation. */ public final class FeatureFlagsBackend implements FeatureFlaggingGateway.Backend, AutoCloseable { + private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagsBackend.class); private static final String API_KEY = "api-key"; private static final String SITE = "site"; private static final String ENV = "env"; @@ -29,18 +35,33 @@ public final class FeatureFlagsBackend implements FeatureFlaggingGateway.Backend private final EventProxy eventProxy; private final SpanEnrichmentWriter spanEnrichmentWriter; private final CoreMetricCollector metrics; + private final List shutdownListeners = new CopyOnWriteArrayList<>(); + @Nullable private final Runnable onFirstShutdownSubscription; + private final AtomicBoolean shutdownSubscribed = new AtomicBoolean(); private RemoteConfigServiceImpl remoteConfig; - public FeatureFlagsBackend(final SharedCommunicationObjects sco, final Config config) { + /** + * @param sco the shared communication objects. + * @param config the agent configuration. + * @param onFirstShutdownSubscription the action to run when the SDK first subscribes to the + * shutdown, or {@code null} for none. + */ + public FeatureFlagsBackend( + final SharedCommunicationObjects sco, + final Config config, + @Nullable final Runnable onFirstShutdownSubscription) { this( sco, config, new EventProxy( - new BackendApiFactory(config, sco), + new BackendApiFactory(config, sco, EventProxy.IDENTITY_HEADERS, true), + config.getFeatureFlaggingConfigurationSource() == AGENTLESS, config.getFeatureFlaggingConfigurationSource() == AGENTLESS - && config.getApiKey() != null), + && config.getApiKey() != null + && !config.getApiKey().isEmpty()), SpanEnrichmentWriter.getInstance(), - CoreMetricCollector.getInstance()); + CoreMetricCollector.getInstance(), + onFirstShutdownSubscription); } FeatureFlagsBackend( @@ -48,12 +69,14 @@ public FeatureFlagsBackend(final SharedCommunicationObjects sco, final Config co final Config config, final EventProxy eventProxy, final SpanEnrichmentWriter spanEnrichmentWriter, - final CoreMetricCollector metrics) { + final CoreMetricCollector metrics, + @Nullable final Runnable onFirstShutdownSubscription) { this.sco = sco; this.config = config; this.eventProxy = eventProxy; this.spanEnrichmentWriter = spanEnrichmentWriter; this.metrics = metrics; + this.onFirstShutdownSubscription = onFirstShutdownSubscription; } /** Registers the Remote Configuration product early when it is the configuration source. */ @@ -136,7 +159,34 @@ public void enrichRuntimeDefault(final String flagKey, @Nullable final Object de } @Override - public synchronized void close() { + public AutoCloseable subscribeShutdown(final Runnable listener) { + this.shutdownListeners.add(listener); + if (this.onFirstShutdownSubscription != null + && this.shutdownSubscribed.compareAndSet(false, true)) { + this.onFirstShutdownSubscription.run(); + } + return () -> this.shutdownListeners.remove(listener); + } + + /** + * Notifies the shutdown listeners, flushing the SDK while the agent services are still running, + * then releases the backend resources. + */ + @Override + public void close() { + // Run outside the lock: the SDK flush posts events and counts metrics through this backend. + for (final Runnable listener : this.shutdownListeners) { + try { + listener.run(); + } catch (final RuntimeException | LinkageError e) { + LOGGER.debug("Feature Flags shutdown listener failed", e); + } + } + this.shutdownListeners.clear(); + closeResources(); + } + + private synchronized void closeResources() { if (this.remoteConfig != null) { this.remoteConfig.close(); this.remoteConfig = null; diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java index 837e377dec4..f98eb8ed29d 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java @@ -58,7 +58,12 @@ public AutoCloseable subscribe(final Consumer subscriber) { this.subscribers.add(subscriber); final byte[] document = this.current; if (document != null) { - subscriber.accept(document); + try { + subscriber.accept(document); + } catch (final RuntimeException e) { + // Already reported to Remote Configuration when the document was received. + LOGGER.debug("Feature Flags configuration subscriber rejected the current document", e); + } } } return () -> this.subscribers.remove(subscriber); @@ -76,6 +81,15 @@ public synchronized void close() { this.current = null; } + /** + * Forwards a configuration document to every subscriber. + * + * @param configKey the Remote Configuration key. + * @param content the raw document, or {@code null} when the configuration is removed. + * @param pollingRateHinter the polling rate hinter. + * @throws RuntimeException the first subscriber failure, once every subscriber is notified, so + * the poller reports the document as not applied. + */ @Override public void accept( final String configKey, @@ -86,13 +100,21 @@ public void accept( this.current = content; snapshot = this.subscribers.iterator(); } + RuntimeException failure = null; while (snapshot.hasNext()) { final Consumer subscriber = snapshot.next(); try { subscriber.accept(content); } catch (final RuntimeException e) { - LOGGER.debug("Feature Flags configuration subscriber failed", e); + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } } } + if (failure != null) { + throw failure; + } } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java index c0492e04819..5d44674e4f3 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/EventProxyTest.java @@ -1,85 +1,196 @@ package com.datadog.featureflag; +import static com.datadog.featureflag.EventProxy.IDENTITY_HEADERS; +import static com.datadog.featureflag.EventProxy.IDENTITY_HEADER_NAMES; +import static com.datadog.featureflag.EventProxy.PRE_SEND_RETRY; +import static com.datadog.featureflag.EventProxy.RECOVERY_INTERVAL_NANOS; +import static datadog.communication.EvpProxy.ORIGIN_HEADER; +import static datadog.communication.EvpProxy.ORIGIN_VERSION_HEADER; +import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT; +import static datadog.communication.http.HttpRetryPolicy.Factory.NEVER_RETRY; +import static datadog.trace.api.intake.Intake.EVENT_PLATFORM; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Arrays.asList; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyIterable; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import datadog.communication.BackendApi; import datadog.communication.BackendApiFactory; import datadog.communication.HttpResponseException; +import datadog.communication.ddagent.TracerVersion; import datadog.communication.http.HttpRetryPolicy; -import datadog.trace.api.intake.Intake; import java.io.IOException; import java.net.ConnectException; +import java.util.concurrent.atomic.AtomicLong; +import okhttp3.Response; import org.junit.jupiter.api.Test; import org.tabletest.junit.TableTest; class EventProxyTest { private static final byte[] PAYLOAD = "{}".getBytes(UTF_8); + private static final String ENDPOINT = "evp_proxy/v4/"; + + private final BackendApiFactory factory = mock(BackendApiFactory.class); + private final BackendApi exposures = mock(BackendApi.class); + private final BackendApi flagEvaluations = mock(BackendApi.class); + private final AtomicLong now = new AtomicLong(); + + @Test + void sendsRemoteConfigurationEventsToTheFixedV2EndpointWithPreSendRetry() throws IOException { + clientsFor(V2_EVP_PROXY_ENDPOINT); + final EventProxy proxy = proxy(false, false); + + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); + assertTrue(proxy.post("flagevaluation", PAYLOAD)); + + verify(this.factory) + .createEvpProxyApiForEndpoint(EVENT_PLATFORM, true, PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + verify(this.factory) + .createEvpProxyApiForEndpoint(EVENT_PLATFORM, false, PRE_SEND_RETRY, V2_EVP_PROXY_ENDPOINT); + verify(this.factory, never()).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); + verify(this.flagEvaluations).post(eq("flagevaluation"), any(), any(), any(), eq(false)); + } + + @Test + void sendsAgentlessEventsOnceThroughTheDiscoveredEndpoint() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + final EventProxy proxy = proxy(true, true); + + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); + + verify(this.factory).createEvpProxyApiForEndpoint(EVENT_PLATFORM, true, NEVER_RETRY, ENDPOINT); + verify(this.factory).createEvpProxyApiForEndpoint(EVENT_PLATFORM, false, NEVER_RETRY, ENDPOINT); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + } @Test - void isUnavailableWithoutAgentProxySupport() throws IOException { - final EventProxy proxy = new EventProxy(mock(BackendApiFactory.class), false); + void leavesAgentlessEventsToDirectIntakeWithoutTheAgentProxy() throws IOException { + final EventProxy proxy = proxy(true, true); assertFalse(proxy.isAvailable()); + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + assertFalse(proxy.post("exposures", PAYLOAD)); + + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + } + + @Test + void recoversTheAgentProxyWhenItIsTheOnlyRoute() throws IOException { + final EventProxy proxy = proxy(true, false); + + assertTrue(proxy.isAvailable(), "the proxy route stays in place to recover"); assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, never()).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + when(this.factory.discoverEvpProxyEndpoint(true, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + + assertTrue(proxy.post("exposures", PAYLOAD)); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); } @Test - void postsEachRouteThroughItsOwnClient() throws IOException { - final BackendApiFactory factory = mock(BackendApiFactory.class); - final BackendApi exposures = mock(BackendApi.class); - final BackendApi flagEvaluations = mock(BackendApi.class); - when(factory.createEvpProxyApi(Intake.EVENT_PLATFORM, true)).thenReturn(exposures); - when(factory.createEvpProxyApi(Intake.EVENT_PLATFORM, false)).thenReturn(flagEvaluations); - final EventProxy proxy = new EventProxy(factory, false); + void failedDiscoveryRetriesAtMostOncePerRecoveryInterval() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)) + .thenThrow(new IllegalStateException("discovery failed")); + final EventProxy proxy = proxy(true, false); assertTrue(proxy.isAvailable()); + assertFalse(proxy.post("exposures", PAYLOAD)); + this.now.addAndGet(RECOVERY_INTERVAL_NANOS); + assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.addAndGet(RECOVERY_INTERVAL_NANOS - 1); + assertFalse(proxy.post("exposures", PAYLOAD)); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + + this.now.incrementAndGet(); + when(this.factory.discoverEvpProxyEndpoint(true, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); assertTrue(proxy.post("exposures", PAYLOAD)); - assertTrue(proxy.post("flagevaluation", PAYLOAD)); - verify(exposures).post(eq("exposures"), any(), any(), any(), eq(false)); - verify(flagEvaluations).post(eq("flagevaluation"), any(), any(), any(), eq(false)); + verify(this.factory, times(2)).discoverEvpProxyEndpoint(eq(true), anyIterable()); + verify(this.exposures).post(eq("exposures"), any(), any(), any(), eq(false)); } @Test - void doesNotRetryWhenDirectFallbackIsAvailable() { - final BackendApiFactory factory = mock(BackendApiFactory.class); - final EventProxy proxy = new EventProxy(factory, true); + void doesNotRediscoverAfterRejectionWhenDirectFallbackIsAvailable() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")) + .thenReturn(null); + final EventProxy proxy = proxy(true, true); + + assertFalse(proxy.post("exposures", PAYLOAD)); + assertTrue(proxy.isAvailable()); + assertTrue(proxy.post("exposures", PAYLOAD)); - proxy.isAvailable(); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures, times(2)).post(any(), any(), any(), any(), eq(false)); + } + + @Test + void propagatesAmbiguousTransportFailuresWithoutRetryOrRediscovery() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + final IOException failure = new IOException("connection reset after sending"); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())).thenThrow(failure); + final EventProxy proxy = proxy(true, false); + + assertSame(failure, assertThrows(IOException.class, () -> proxy.post("exposures", PAYLOAD))); + assertTrue(proxy.isAvailable()); - verify(factory) - .createEvpProxyApi(Intake.EVENT_PLATFORM, true, HttpRetryPolicy.Factory.NEVER_RETRY); - verify(factory) - .createEvpProxyApi(Intake.EVENT_PLATFORM, false, HttpRetryPolicy.Factory.NEVER_RETRY); + verify(this.factory, times(1)).discoverEvpProxyEndpoint(anyBoolean(), anyIterable()); + verify(this.exposures).post(any(), any(), any(), any(), eq(false)); } @Test - void reportsConnectionFailuresAsUnavailable() throws IOException { - final BackendApi api = proxyApi(new ConnectException("refused")); + void waitsForANewEndpointAfterADefinitiveRejectionWhenItIsTheOnlyRoute() throws IOException { + when(this.factory.discoverEvpProxyEndpoint(false, IDENTITY_HEADER_NAMES)).thenReturn(ENDPOINT); + clientsFor(ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")); + final EventProxy proxy = proxy(true, false); + + assertFalse(proxy.post("exposures", PAYLOAD)); + assertFalse(proxy.post("exposures", PAYLOAD)); - assertFalse(new EventProxy(factoryOf(api), false).post("exposures", PAYLOAD)); + verify(this.exposures, times(1)).post(any(), any(), any(), any(), anyBoolean()); } @TableTest({ "Scenario | Status | Definitive", - "forbidden | 403 | true ", + "forbidden | 403 | false ", "not found | 404 | true ", "not allowed | 405 | true ", "server error | 500 | false ", "rate limited | 429 | false " }) void classifiesProxyRejections(final int status, final boolean definitive) throws IOException { - final BackendApi api = proxyApi(new HttpResponseException(status, "rejected")); - final EventProxy proxy = new EventProxy(factoryOf(api), false); + clientsFor(V2_EVP_PROXY_ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new HttpResponseException(status, "rejected")); + final EventProxy proxy = proxy(false, false); if (definitive) { assertFalse(proxy.post("exposures", PAYLOAD)); @@ -88,15 +199,52 @@ void classifiesProxyRejections(final int status, final boolean definitive) throw } } - private static BackendApi proxyApi(final IOException failure) throws IOException { - final BackendApi api = mock(BackendApi.class); - when(api.post(any(), any(), any(), any(), anyBoolean())).thenThrow(failure); - return api; + @Test + void reportsConnectionRefusalAsDefinitive() throws IOException { + clientsFor(V2_EVP_PROXY_ENDPOINT); + when(this.exposures.post(any(), any(), any(), any(), anyBoolean())) + .thenThrow(new ConnectException("refused")); + + assertFalse(proxy(false, false).post("exposures", PAYLOAD)); + } + + @Test + void identifiesTheJavaTracingLibrary() { + assertEquals("dd-trace-java", IDENTITY_HEADERS.get(ORIGIN_HEADER)); + assertEquals(TracerVersion.TRACER_VERSION, IDENTITY_HEADERS.get(ORIGIN_VERSION_HEADER)); + assertNotEquals("0.0.0", IDENTITY_HEADERS.get(ORIGIN_VERSION_HEADER)); + } + + @Test + void discoversOnlyAgentlessEndpointsForwardingTheIdentity() { + proxy(true, true).isAvailable(); + + verify(this.factory) + .discoverEvpProxyEndpoint(false, asList(ORIGIN_HEADER, ORIGIN_VERSION_HEADER)); + } + + @Test + void preSendRetryRetriesConnectionRefusalOnly() { + final HttpRetryPolicy policy = PRE_SEND_RETRY.create(); + + assertFalse(policy.shouldRetry(new IOException("reset"))); + assertFalse(policy.shouldRetry((Response) null)); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertTrue(policy.shouldRetry(new ConnectException("refused"))); + assertFalse(policy.shouldRetry(new ConnectException("refused")), "at most three retries"); + } + + private EventProxy proxy(final boolean agentless, final boolean directFallbackAvailable) { + return new EventProxy(this.factory, agentless, directFallbackAvailable, this.now::get); } - private static BackendApiFactory factoryOf(final BackendApi api) { - final BackendApiFactory factory = mock(BackendApiFactory.class); - when(factory.createEvpProxyApi(eq(Intake.EVENT_PLATFORM), anyBoolean())).thenReturn(api); - return factory; + private void clientsFor(final String endpoint) { + when(this.factory.createEvpProxyApiForEndpoint( + eq(EVENT_PLATFORM), eq(true), any(), eq(endpoint))) + .thenReturn(this.exposures); + when(this.factory.createEvpProxyApiForEndpoint( + eq(EVENT_PLATFORM), eq(false), any(), eq(endpoint))) + .thenReturn(this.flagEvaluations); } } diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java index 0ba9c1a50e4..ed5c042adbb 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/FeatureFlagsBackendTest.java @@ -3,6 +3,7 @@ import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; import static datadog.trace.api.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED; import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Arrays.asList; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -24,7 +25,10 @@ import datadog.trace.bootstrap.config.provider.ConfigProvider; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.test.junit.utils.config.WithConfigExtension; +import java.util.List; import java.util.Properties; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.tabletest.junit.TableTest; @@ -41,7 +45,7 @@ class FeatureFlagsBackendTest { private final CoreMetricCollector metrics = mock(CoreMetricCollector.class); private final FeatureFlagsBackend backend = new FeatureFlagsBackend( - this.sco, this.config, this.eventProxy, this.spanEnrichmentWriter, this.metrics); + this.sco, this.config, this.eventProxy, this.spanEnrichmentWriter, this.metrics, null); private static AgentSpan rootSpan() { final AgentSpan root = mock(AgentSpan.class); @@ -126,11 +130,48 @@ void createsTheBackendFromTheAgentConfiguration( when(this.config.getFeatureFlaggingConfigurationSource()).thenReturn(source); when(this.config.getApiKey()).thenReturn(apiKey); - final FeatureFlagsBackend created = new FeatureFlagsBackend(this.sco, this.config); + final FeatureFlagsBackend created = new FeatureFlagsBackend(this.sco, this.config, null); assertEquals(apiKey, created.setting("api-key")); } + @Test + void runsTheFirstShutdownSubscriptionActionOnce() { + final AtomicInteger firstSubscriptions = new AtomicInteger(); + final FeatureFlagsBackend backend = + new FeatureFlagsBackend( + this.sco, + this.config, + this.eventProxy, + this.spanEnrichmentWriter, + this.metrics, + firstSubscriptions::incrementAndGet); + + backend.subscribeShutdown(() -> {}); + backend.subscribeShutdown(() -> {}); + + assertEquals(1, firstSubscriptions.get()); + } + + @Test + void closeNotifiesTheShutdownListenersBeforeReleasingResources() throws Exception { + final List calls = new CopyOnWriteArrayList<>(); + this.backend.subscribeShutdown(() -> calls.add("first")); + this.backend.subscribeShutdown( + () -> { + throw new IllegalStateException("failing listener"); + }); + final AutoCloseable unregistered = + this.backend.subscribeShutdown(() -> calls.add("unregistered")); + this.backend.subscribeShutdown(() -> calls.add("last")); + unregistered.close(); + + this.backend.close(); + this.backend.close(); + + assertEquals(asList("first", "last"), calls); + } + @Test void closeWithoutRemoteConfigClearsSpanEnrichment() { this.backend.enrichSerialId(1, false, null); diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java index 6bac29d44f9..db1f668495a 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java @@ -6,6 +6,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -69,6 +70,56 @@ void forwardsDocumentsAndRemovalsToAllSubscribers() { assertEquals(2, second.size()); } + @Test + void reportsRejectedDocumentsOnceEverySubscriberIsNotified() { + final RemoteConfigServiceImpl service = + new RemoteConfigServiceImpl(mock(ConfigurationPoller.class)); + final IllegalArgumentException rejection = new IllegalArgumentException("malformed"); + final IllegalStateException otherRejection = new IllegalStateException("malformed too"); + final List received = new ArrayList<>(); + service.subscribe( + content -> { + throw rejection; + }); + service.subscribe(received::add); + service.subscribe( + content -> { + throw otherRejection; + }); + + final IllegalArgumentException reported = + assertThrows( + IllegalArgumentException.class, + () -> service.accept("key", DOCUMENT, mock(PollingRateHinter.class))); + + assertSame(rejection, reported); + assertArrayEquals(new Throwable[] {otherRejection}, reported.getSuppressed()); + assertEquals(1, received.size()); + assertArrayEquals(DOCUMENT, received.get(0)); + } + + @Test + void lateSubscribersRejectingTheCurrentDocumentStaySubscribed() throws Exception { + final RemoteConfigServiceImpl service = + new RemoteConfigServiceImpl(mock(ConfigurationPoller.class)); + service.accept("key", DOCUMENT, mock(PollingRateHinter.class)); + final List received = new ArrayList<>(); + + final AutoCloseable subscription = + service.subscribe( + content -> { + if (content != null) { + throw new IllegalArgumentException("malformed"); + } + received.add(null); + }); + service.accept("key", null, mock(PollingRateHinter.class)); + subscription.close(); + + assertEquals(1, received.size()); + assertNull(received.get(0)); + } + @Test void replaysTheCurrentDocumentToLateSubscribers() { final RemoteConfigServiceImpl service =