diff --git a/dd-openfeature/build.gradle.kts b/dd-openfeature/build.gradle.kts index 17e9f6ee412..c218e9603d7 100644 --- a/dd-openfeature/build.gradle.kts +++ b/dd-openfeature/build.gradle.kts @@ -5,6 +5,7 @@ plugins { `java-library` idea id("dd-trace-java.conventions.java") + id("dd-trace-java.version-file") } description = "Datadog OpenFeature provider SDK." diff --git a/dd-openfeature/gradle.lockfile b/dd-openfeature/gradle.lockfile index c878fe4af10..8fa9448ffd5 100644 --- a/dd-openfeature/gradle.lockfile +++ b/dd-openfeature/gradle.lockfile @@ -2,77 +2,84 @@ # Manual edits can break the build and are not advised. # This file is expected to be part of source control. # To regenerate this file, run: ./gradlew :dd-openfeature:dependencies --write-locks -ch.qos.logback:logback-classic:1.2.13=testCompileClasspath,testRuntimeClasspath -ch.qos.logback:logback-core:1.2.13=testCompileClasspath,testRuntimeClasspath -com.fasterxml.jackson.core:jackson-core:2.20.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -com.fasterxml.jackson:jackson-bom:2.20.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath +ch.qos.logback:logback-classic:1.2.13=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +ch.qos.logback:logback-core:1.2.13=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +com.fasterxml.jackson.core:jackson-core:2.20.0=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath +com.fasterxml.jackson:jackson-bom:2.20.0=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath com.github.javaparser:javaparser-core:3.25.6=codenarc -com.github.spotbugs:spotbugs-annotations:4.10.4=compileClasspath,spotbugs +com.github.spotbugs:spotbugs-annotations:4.10.4=compileClasspath,jmhCompileClasspath,spotbugs com.github.spotbugs:spotbugs:4.10.4=spotbugs com.github.stephenc.jcip:jcip-annotations:1.0-1=spotbugs -com.google.code.findbugs:jsr305:3.0.2=compileClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath +com.google.code.findbugs:jsr305:3.0.2=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath com.google.code.gson:gson:2.14.0=spotbugs com.google.errorprone:error_prone_annotations:2.48.0=spotbugs com.thoughtworks.qdox:qdox:1.12.1=codenarc commons-io:commons-io:2.21.0=spotbugs -de.thetaphi:forbiddenapis:3.11=compileClasspath -dev.openfeature:sdk:1.22.0=compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -io.leangen.geantyref:geantyref:1.3.16=testRuntimeClasspath +de.thetaphi:forbiddenapis:3.11=compileClasspath,jmhCompileClasspath +dev.openfeature:sdk:1.22.0=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath +io.leangen.geantyref:geantyref:1.3.16=jmhRuntimeClasspath,testRuntimeClasspath jaxen:jaxen:2.0.6=spotbugs -net.bytebuddy:byte-buddy-agent:1.18.12=testCompileClasspath,testRuntimeClasspath -net.bytebuddy:byte-buddy:1.18.12=testCompileClasspath,testRuntimeClasspath +net.bytebuddy:byte-buddy-agent:1.18.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +net.bytebuddy:byte-buddy:1.18.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +net.sf.jopt-simple:jopt-simple:5.0.4=jmh,jmhCompileClasspath,jmhRuntimeClasspath net.sf.saxon:Saxon-HE:12.10=spotbugs org.apache.ant:ant-antlr:1.10.14=codenarc org.apache.ant:ant-junit:1.10.14=codenarc org.apache.bcel:bcel:6.12.0=spotbugs org.apache.commons:commons-lang3:3.20.0=spotbugs +org.apache.commons:commons-math3:3.6.1=jmh,jmhCompileClasspath,jmhRuntimeClasspath 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,testRuntimeClasspath +org.apiguardian:apiguardian-api:1.1.2=jmhRuntimeClasspath,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 org.codehaus.groovy:groovy-json:3.0.23=codenarc -org.codehaus.groovy:groovy-json:3.0.25=testCompileClasspath,testRuntimeClasspath +org.codehaus.groovy:groovy-json:3.0.25=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.codehaus.groovy:groovy-templates:3.0.23=codenarc org.codehaus.groovy:groovy-xml:3.0.23=codenarc org.codehaus.groovy:groovy:3.0.23=codenarc -org.codehaus.groovy:groovy:3.0.25=testCompileClasspath,testRuntimeClasspath +org.codehaus.groovy:groovy:3.0.25=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.codenarc:CodeNarc:3.7.0=codenarc org.dom4j:dom4j:2.2.0=spotbugs org.gmetrics:GMetrics:2.1.0=codenarc -org.hamcrest:hamcrest:3.0=testCompileClasspath,testRuntimeClasspath +org.hamcrest:hamcrest:3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.jacoco:org.jacoco.agent:0.8.15=jacocoAgent,jacocoAnt org.jacoco:org.jacoco.ant:0.8.15=jacocoAnt org.jacoco:org.jacoco.core:0.8.15=jacocoAnt org.jacoco:org.jacoco.report:0.8.15=jacocoAnt -org.junit.jupiter:junit-jupiter-api:5.14.1=testCompileClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter-engine:5.14.1=testRuntimeClasspath -org.junit.jupiter:junit-jupiter-params:5.14.1=testCompileClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter:5.14.1=testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-commons:1.14.1=testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-engine:1.14.1=testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-launcher:1.14.1=testRuntimeClasspath -org.junit:junit-bom:5.14.1=testCompileClasspath,testRuntimeClasspath -org.mockito:mockito-core:4.4.0=testCompileClasspath,testRuntimeClasspath -org.mockito:mockito-junit-jupiter:4.4.0=testCompileClasspath,testRuntimeClasspath -org.objenesis:objenesis:3.3=testCompileClasspath,testRuntimeClasspath -org.opentest4j:opentest4j:1.3.0=testCompileClasspath,testRuntimeClasspath +org.junit.jupiter:junit-jupiter-api:5.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.junit.jupiter:junit-jupiter-engine:5.14.1=jmhRuntimeClasspath,testRuntimeClasspath +org.junit.jupiter:junit-jupiter-params:5.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.junit.jupiter:junit-jupiter:5.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.junit.platform:junit-platform-commons:1.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.junit.platform:junit-platform-engine:1.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.junit.platform:junit-platform-launcher:1.14.1=jmhRuntimeClasspath,testRuntimeClasspath +org.junit:junit-bom:5.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.mockito:mockito-core:4.4.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.mockito:mockito-junit-jupiter:4.4.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.objenesis:objenesis:3.3=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.openjdk.jmh:jmh-core:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath +org.openjdk.jmh:jmh-generator-asm:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath +org.openjdk.jmh:jmh-generator-bytecode:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath +org.openjdk.jmh:jmh-generator-reflection:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath +org.opentest4j:opentest4j:1.3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.ow2.asm:asm-analysis:9.10.1=spotbugs org.ow2.asm:asm-commons:9.10.1=jacocoAnt,spotbugs org.ow2.asm:asm-tree:9.10.1=jacocoAnt,spotbugs org.ow2.asm:asm-util:9.10.1=spotbugs +org.ow2.asm:asm:9.0=jmh,jmhCompileClasspath,jmhRuntimeClasspath org.ow2.asm:asm:9.10.1=jacocoAnt,spotbugs -org.slf4j:jcl-over-slf4j:1.7.30=testCompileClasspath,testRuntimeClasspath -org.slf4j:jul-to-slf4j:1.7.30=testCompileClasspath,testRuntimeClasspath -org.slf4j:log4j-over-slf4j:1.7.30=testCompileClasspath,testRuntimeClasspath +org.slf4j:jcl-over-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.slf4j:jul-to-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.slf4j:log4j-over-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.slf4j:slf4j-api:2.0.17=spotbugsSlf4j -org.slf4j:slf4j-api:2.0.18=compileClasspath,runtimeClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath +org.slf4j:slf4j-api:2.0.18=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath org.slf4j:slf4j-simple:2.0.17=spotbugsSlf4j -org.spockframework:spock-bom:2.4-groovy-3.0=testCompileClasspath,testRuntimeClasspath -org.spockframework:spock-core:2.4-groovy-3.0=testCompileClasspath,testRuntimeClasspath -org.tabletest:tabletest-junit:1.2.2=testCompileClasspath,testRuntimeClasspath -org.tabletest:tabletest-parser:1.2.1=testCompileClasspath,testRuntimeClasspath +org.spockframework:spock-bom:2.4-groovy-3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.spockframework:spock-core:2.4-groovy-3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.tabletest:tabletest-junit:1.2.2=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath +org.tabletest:tabletest-parser:1.2.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath org.xmlresolver:xmlresolver:5.3.3=spotbugs -empty=annotationProcessor,spotbugsPlugins,testAnnotationProcessor +empty=annotationProcessor,jmhAnnotationProcessor,spotbugsPlugins,testAnnotationProcessor diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/ConfigurationService.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/ConfigurationService.java new file mode 100644 index 00000000000..465db6b0444 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/ConfigurationService.java @@ -0,0 +1,19 @@ +package com.datadog.openfeature.internal; + +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import java.util.function.Consumer; + +/** Delivers flag configurations to the runtime. */ +public interface ConfigurationService extends AutoCloseable { + /** + * Starts delivering configurations. Implementations may deliver the first configuration before + * returning. + * + * @param listener the listener receiving configurations, and {@code null} when the configuration + * is removed. + */ + void start(Consumer listener); + + @Override + void close(); +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventQueue.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventQueue.java new file mode 100644 index 00000000000..d2de2f8f239 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventQueue.java @@ -0,0 +1,110 @@ +package com.datadog.openfeature.internal; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.LockSupport; +import java.util.function.Consumer; + +/** + * A bounded, lock-free, multiple-producer single-consumer queue. Producers never block; the + * consumer polls with a timeout by parking, as producers do not signal it. + * + * @param the element type. + */ +public final class EventQueue { + private static final long PARK_NANOS = TimeUnit.MILLISECONDS.toNanos(10); + + private final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>(); + private final AtomicInteger size = new AtomicInteger(); + private final int capacity; + + public EventQueue(final int capacity) { + this.capacity = capacity; + } + + /** + * Offers an element without blocking. + * + * @param element the element to offer. + * @return {@code true} if the element was queued, {@code false} if the queue is full. + */ + public boolean offer(final E element) { + if (element == null) { + throw new NullPointerException("element"); + } + if (this.size.incrementAndGet() > this.capacity) { + this.size.decrementAndGet(); + return false; + } + this.queue.offer(element); + return true; + } + + /** + * @return the head element, or {@code null} if the queue is empty. + */ + public E poll() { + final E element = this.queue.poll(); + if (element != null) { + this.size.decrementAndGet(); + } + return element; + } + + /** + * Polls the head element, waiting up to the given timeout for one to be available. + * + * @param timeout the maximum time to wait. + * @param unit the timeout unit. + * @return the head element, or {@code null} if none was available in time. + * @throws InterruptedException if the consumer thread is interrupted while waiting. + */ + public E poll(final long timeout, final TimeUnit unit) throws InterruptedException { + final long deadline = System.nanoTime() + unit.toNanos(timeout); + while (true) { + final E element = poll(); + if (element != null) { + return element; + } + final long remaining = deadline - System.nanoTime(); + if (remaining <= 0) { + return null; + } + LockSupport.parkNanos(this, Math.min(remaining, PARK_NANOS)); + if (Thread.interrupted()) { + throw new InterruptedException(); + } + } + } + + /** + * Drains up to {@code limit} elements to the consumer. + * + * @param consumer the consumer of the drained elements. + * @param limit the maximum number of elements to drain. + */ + public void drain(final Consumer consumer, final int limit) { + for (int i = 0; i < limit; i++) { + final E element = poll(); + if (element == null) { + return; + } + consumer.accept(element); + } + } + + /** + * @return the current number of queued elements. + */ + public int size() { + return this.size.get(); + } + + /** + * @return the queue capacity. + */ + public int capacity() { + return this.capacity; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/JsonWriting.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/JsonWriting.java new file mode 100644 index 00000000000..807f76ab428 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/JsonWriting.java @@ -0,0 +1,96 @@ +package com.datadog.openfeature.internal; + +import com.fasterxml.jackson.core.JsonEncoding; +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.Collection; +import java.util.Map; +import javax.annotation.Nullable; + +/** JSON serialization helpers for event payloads. */ +public final class JsonWriting { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + + private JsonWriting() {} + + /** + * Serializes a JSON document. + * + * @param writer the document writer. + * @return the UTF-8 encoded document. + */ + public static byte[] write(final Writer writer) { + final ByteArrayOutputStream output = new ByteArrayOutputStream(256); + try (JsonGenerator generator = JSON_FACTORY.createGenerator(output, JsonEncoding.UTF8)) { + writer.write(generator); + } catch (final IOException e) { + throw new UncheckedIOException(e); + } + return output.toByteArray(); + } + + /** + * Writes an arbitrary value: maps, collections, strings, numbers, booleans and {@code null}. + * Other values are written with their string representation. + * + * @param generator the generator to write to. + * @param value the value to write. + * @throws IOException if the value could not be written. + */ + public static void writeValue(final JsonGenerator generator, @Nullable final Object value) + throws IOException { + if (value == null) { + generator.writeNull(); + } else if (value instanceof String) { + generator.writeString((String) value); + } else if (value instanceof Boolean) { + generator.writeBoolean((Boolean) value); + } else if (value instanceof Integer || value instanceof Long || value instanceof Short) { + generator.writeNumber(((Number) value).longValue()); + } else if (value instanceof Number) { + generator.writeNumber(((Number) value).doubleValue()); + } else if (value instanceof Map) { + generator.writeStartObject(); + for (final Map.Entry entry : ((Map) value).entrySet()) { + generator.writeFieldName(String.valueOf(entry.getKey())); + writeValue(generator, entry.getValue()); + } + generator.writeEndObject(); + } else if (value instanceof Collection) { + generator.writeStartArray(); + for (final Object element : (Collection) value) { + writeValue(generator, element); + } + generator.writeEndArray(); + } else { + generator.writeString(value.toString()); + } + } + + /** + * Writes a {@code {"key": ...}} object field, skipped when the key is {@code null}. + * + * @param generator the generator to write to. + * @param fieldName the field name. + * @param key the key value. + * @throws IOException if the field could not be written. + */ + public static void writeKeyObject( + final JsonGenerator generator, final String fieldName, @Nullable final String key) + throws IOException { + if (key != null) { + generator.writeObjectFieldStart(fieldName); + generator.writeStringField("key", key); + generator.writeEndObject(); + } + } + + /** Writes a JSON document. */ + @FunctionalInterface + public interface Writer { + void write(JsonGenerator generator) throws IOException; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/RuntimeServices.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/RuntimeServices.java new file mode 100644 index 00000000000..8227e19d851 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/RuntimeServices.java @@ -0,0 +1,40 @@ +package com.datadog.openfeature.internal; + +import com.datadog.openfeature.internal.bridge.HealthMetrics; +import javax.annotation.Nullable; + +/** Thread creation and health metrics shared by the runtime components. */ +public final class RuntimeServices { + private final HealthMetrics healthMetrics; + + public RuntimeServices(final HealthMetrics healthMetrics) { + this.healthMetrics = healthMetrics; + } + + /** + * Creates a daemon thread. + * + * @param role the thread role, used to name the thread. + * @param task the thread task. + * @return the created thread, not started yet. + */ + public Thread newThread(final String role, final Runnable task) { + final Thread thread = new Thread(task, "dd-openfeature-" + role); + thread.setDaemon(true); + thread.setContextClassLoader(null); + return thread; + } + + /** + * Increments a health counter, skipping non-positive values. + * + * @param name the metric name. + * @param value the increment. + * @param reason the optional {@code reason} tag value. + */ + public void countMetric(final String name, final long value, @Nullable final String reason) { + if (value > 0) { + this.healthMetrics.count(name, value, reason); + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java new file mode 100644 index 00000000000..1ff84479cec --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridge.java @@ -0,0 +1,41 @@ +package com.datadog.openfeature.internal.bridge; + +import javax.annotation.Nullable; + +/** Components supplied by the Datadog Java agent when attached, or no-op ones otherwise. */ +public interface AgentBridge { + /** The bridge used when no Datadog Java agent is attached. */ + AgentBridge NONE = new NoopAgentBridge(); + + /** + * Looks up a setting from the agent configuration. + * + * @param key the setting key, using the {@code dd.} system property notation without the prefix. + * @return the setting value, or {@code null} to fall back to system properties and environment. + */ + @Nullable + String setting(String key); + + /** + * @return the Remote Configuration source, or {@code null} if not available. + */ + @Nullable + ConfigurationSource remoteConfiguration(); + + /** + * @return the transport through the Datadog Agent event platform proxy, or {@code null} if not + * available. + */ + @Nullable + EventTransport eventProxy(); + + /** + * @return the health metrics sink. + */ + HealthMetrics healthMetrics(); + + /** + * @return the span enricher. + */ + SpanEnricher spanEnricher(); +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridges.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridges.java new file mode 100644 index 00000000000..f11022880ae --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/AgentBridges.java @@ -0,0 +1,20 @@ +package com.datadog.openfeature.internal.bridge; + +/** + * Resolves the {@link AgentBridge} to use. Kept apart from {@link AgentBridge}: the agent injects + * bridge implementations when instrumenting {@link #detect()}, and they cannot be defined while + * their {@link AgentBridge} interface is itself being defined. + */ +public final class AgentBridges { + private AgentBridges() {} + + /** + * Detects the bridge to use. The Datadog Java agent instruments this method to return its own + * bridge. + * + * @return the bridge to use. + */ + public static AgentBridge detect() { + return AgentBridge.NONE; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ConfigurationSource.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ConfigurationSource.java new file mode 100644 index 00000000000..263938ee973 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ConfigurationSource.java @@ -0,0 +1,15 @@ +package com.datadog.openfeature.internal.bridge; + +import java.util.function.Consumer; + +/** A source of Universal Flag Configuration (UFC) documents. */ +public interface ConfigurationSource { + /** + * Subscribes to configuration updates. + * + * @param listener the listener receiving raw UFC JSON documents, or {@code null} when the + * configuration is removed. + * @return the handle to close to unsubscribe. + */ + AutoCloseable subscribe(Consumer listener); +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventProxyUnavailableException.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventProxyUnavailableException.java new file mode 100644 index 00000000000..b141a2fce00 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventProxyUnavailableException.java @@ -0,0 +1,13 @@ +package com.datadog.openfeature.internal.bridge; + +import java.io.IOException; + +/** + * Thrown by an event proxy {@link EventTransport} when the proxy definitively rejects the payload + * as unavailable, so the SDK can fall back to direct intake. + */ +public final class EventProxyUnavailableException extends IOException { + public EventProxyUnavailableException(final String message) { + super(message); + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventTransport.java new file mode 100644 index 00000000000..6f2eb3aca6e --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/EventTransport.java @@ -0,0 +1,16 @@ +package com.datadog.openfeature.internal.bridge; + +import java.io.IOException; + +/** Delivers serialized product events to the Datadog event platform. */ +@FunctionalInterface +public interface EventTransport { + /** + * Posts a JSON payload. + * + * @param route the event platform route, like {@code exposures} or {@code flagevaluation}. + * @param json the UTF-8 JSON payload. + * @throws IOException if the payload could not be delivered. + */ + void post(String route, byte[] json) throws IOException; +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/HealthMetrics.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/HealthMetrics.java new file mode 100644 index 00000000000..2b1c247cb91 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/HealthMetrics.java @@ -0,0 +1,19 @@ +package com.datadog.openfeature.internal.bridge; + +import javax.annotation.Nullable; + +/** Records SDK health counters. */ +@FunctionalInterface +public interface HealthMetrics { + /** The sink discarding all counters. */ + HealthMetrics NOOP = (metric, value, reason) -> {}; + + /** + * Increments a counter. + * + * @param metric the metric name. + * @param value the increment. + * @param reason the optional {@code reason} tag value. + */ + void count(String metric, long value, @Nullable String reason); +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java new file mode 100644 index 00000000000..dab3e42db97 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/NoopAgentBridge.java @@ -0,0 +1,33 @@ +package com.datadog.openfeature.internal.bridge; + +import javax.annotation.Nullable; + +final class NoopAgentBridge implements AgentBridge { + @Nullable + @Override + public String setting(final String key) { + return null; + } + + @Nullable + @Override + public ConfigurationSource remoteConfiguration() { + return null; + } + + @Nullable + @Override + public EventTransport eventProxy() { + return null; + } + + @Override + public HealthMetrics healthMetrics() { + return HealthMetrics.NOOP; + } + + @Override + public SpanEnricher spanEnricher() { + return SpanEnricher.NOOP; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/SpanEnricher.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/SpanEnricher.java new file mode 100644 index 00000000000..768481bb6e8 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/SpanEnricher.java @@ -0,0 +1,34 @@ +package com.datadog.openfeature.internal.bridge; + +import javax.annotation.Nullable; + +/** Reflects flag evaluations onto the active local root span. */ +public interface SpanEnricher { + /** The enricher discarding all evaluations. */ + SpanEnricher NOOP = + new SpanEnricher() { + @Override + public void serialId( + final int serialId, final boolean doLog, @Nullable final String targetingKey) {} + + @Override + public void runtimeDefault(final String flagKey, @Nullable final Object defaultValue) {} + }; + + /** + * Records an evaluation that resolved to a split with a serial id. + * + * @param serialId the split serial id. + * @param doLog whether the allocation logs exposures, to also record the subject. + * @param targetingKey the optional targeting key of the subject. + */ + void serialId(int serialId, boolean doLog, @Nullable String targetingKey); + + /** + * Records an evaluation that resolved to its runtime default value. + * + * @param flagKey the flag key. + * @param defaultValue the default value, unwrapped to a native Java type (map, list or scalar). + */ + void runtimeDefault(String flagKey, @Nullable Object defaultValue); +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/package-info.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/package-info.java new file mode 100644 index 00000000000..f4d9dead3c4 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/package-info.java @@ -0,0 +1,8 @@ +/** + * Extension points the Datadog Java agent fills in through instrumentation. + * + *

These types are not customer API. They form the contract between this SDK and the agent + * instrumentation, which can run a different version than the SDK, so they must evolve in a + * backward compatible way. + */ +package com.datadog.openfeature.internal.bridge; diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/SdkVersion.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/SdkVersion.java new file mode 100644 index 00000000000..b4ea3613ea9 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/SdkVersion.java @@ -0,0 +1,31 @@ +package com.datadog.openfeature.internal.config; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; + +/** The version of this SDK, as written by the build. */ +public final class SdkVersion { + /** The SDK version, or {@code unknown} if the version file is missing. */ + public static final String VERSION = readVersion(); + + private SdkVersion() {} + + private static String readVersion() { + final InputStream stream = SdkVersion.class.getResourceAsStream("/dd-openfeature.version"); + if (stream == null) { + return "unknown"; + } + try (BufferedReader reader = + new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8))) { + final String line = reader.readLine(); + // Strip the git hash suffix. + final int hash = line == null ? -1 : line.indexOf('~'); + return line == null ? "unknown" : hash < 0 ? line.trim() : line.substring(0, hash); + } catch (final IOException e) { + return "unknown"; + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java new file mode 100644 index 00000000000..11bf5f4e91c --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/config/Settings.java @@ -0,0 +1,294 @@ +package com.datadog.openfeature.internal.config; + +import com.datadog.openfeature.internal.bridge.AgentBridge; +import de.thetaphi.forbiddenapis.SuppressForbidden; +import java.util.Locale; +import java.util.function.UnaryOperator; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Resolves the SDK settings. A setting is looked up from the {@link AgentBridge} first, then from + * the {@code dd.}-prefixed system property, then from the {@code DD_}-prefixed environment + * variable. + */ +public final class Settings { + private static final Logger LOGGER = LoggerFactory.getLogger(Settings.class); + + /** The setting sources: bridge, system properties, then environment variables. */ + private static final int SOURCE_COUNT = 3; + + public static final String CONFIGURATION_SOURCE_AGENTLESS = "agentless"; + public static final String CONFIGURATION_SOURCE_REMOTE_CONFIG = "remote_config"; + + static final String FEATURE_FLAGS_ENABLED = "feature.flags.enabled"; + static final String FEATURE_FLAGS_CONFIGURATION_SOURCE = "feature.flags.configuration.source"; + static final String EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED = + "experimental.flagging.provider.enabled"; + static final String AGENTLESS_BASE_URL = "feature.flags.configuration.source.agentless.base.url"; + static final String AGENTLESS_POLL_INTERVAL_SECONDS = + "feature.flags.configuration.source.agentless.poll.interval.seconds"; + static final String AGENTLESS_REQUEST_TIMEOUT_SECONDS = + "feature.flags.configuration.source.agentless.request.timeout.seconds"; + static final String SPAN_ENRICHMENT_ENABLED = + "experimental.flagging.provider.span.enrichment.enabled"; + static final String EVALUATION_COUNTS_ENABLED = "flagging.evaluation.counts.enabled"; + static final String EXPOSURES_DATADOG_LOGGING_ENABLED = + "feature.flags.exposures.datadog.logging.enabled"; + static final String API_KEY = "api-key"; + static final String SITE = "site"; + static final String ENV = "env"; + static final String SERVICE = "service"; + static final String VERSION = "version"; + + static final String DEFAULT_SITE = "datadoghq.com"; + static final int DEFAULT_POLL_INTERVAL_SECONDS = 30; + static final int DEFAULT_REQUEST_TIMEOUT_SECONDS = 5; + + private final AgentBridge bridge; + private final UnaryOperator systemProperties; + private final UnaryOperator environment; + + Settings( + final AgentBridge bridge, + final UnaryOperator systemProperties, + final UnaryOperator environment) { + this.bridge = bridge; + this.systemProperties = systemProperties; + this.environment = environment; + } + + /** + * Creates the settings backed by the given bridge, system properties and environment. + * + * @param bridge the bridge to look settings up from first. + * @return the settings. + */ + // Allowing VM properties and environment variable access + // as the SDK does not ship the agent configuration helpers + @SuppressForbidden + public static Settings load(final AgentBridge bridge) { + return new Settings(bridge, System::getProperty, System::getenv); + } + + /** + * @return whether Feature Flags is enabled. + */ + public boolean isEnabled() { + return resolveSource() != null; + } + + /** + * Resolves the configuration source, with the same semantics as the Datadog Java agent: an + * explicit {@code DD_FEATURE_FLAGS_ENABLED=false} disables the product, an explicit source wins + * over the legacy {@code DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED} toggle, and {@code agentless} + * is the default. + * + * @return the configuration source, or {@code null} if Feature Flags is disabled or the source is + * not supported. + */ + @Nullable + public String configurationSource() { + return resolveSource(); + } + + @Nullable + private String resolveSource() { + final Boolean enabled = getBoolean(FEATURE_FLAGS_ENABLED); + final String source = normalizeSource(get(FEATURE_FLAGS_CONFIGURATION_SOURCE)); + if (Boolean.FALSE.equals(enabled)) { + return null; + } + if (source != null) { + if (CONFIGURATION_SOURCE_AGENTLESS.equals(source) + || CONFIGURATION_SOURCE_REMOTE_CONFIG.equals(source)) { + return source; + } + LOGGER.warn("Unsupported Feature Flags configuration source: {}", source); + return null; + } + final Boolean legacyEnabled = getBoolean(EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED); + if (legacyEnabled != null) { + return legacyEnabled ? CONFIGURATION_SOURCE_REMOTE_CONFIG : null; + } + return CONFIGURATION_SOURCE_AGENTLESS; + } + + /** + * @return the custom CDN base URL, or {@code null} to use the Datadog managed one. + */ + @Nullable + public String agentlessBaseUrl() { + final String url = get(AGENTLESS_BASE_URL); + return url == null || url.trim().isEmpty() ? null : url.trim(); + } + + /** + * @return the CDN poll interval, in seconds. + */ + public int agentlessPollIntervalSeconds() { + return getPositiveInt(AGENTLESS_POLL_INTERVAL_SECONDS, DEFAULT_POLL_INTERVAL_SECONDS); + } + + /** + * @return the CDN request timeout, in seconds. + */ + public int agentlessRequestTimeoutSeconds() { + return getPositiveInt(AGENTLESS_REQUEST_TIMEOUT_SECONDS, DEFAULT_REQUEST_TIMEOUT_SECONDS); + } + + /** + * @return whether span enrichment is enabled (off by default). + */ + public boolean isSpanEnrichmentEnabled() { + return Boolean.TRUE.equals(getBoolean(SPAN_ENRICHMENT_ENABLED)); + } + + /** + * @return whether flag evaluation counts are sent to the event platform (on by default). + */ + public boolean isEvaluationCountsEnabled() { + return !Boolean.FALSE.equals(getBoolean(EVALUATION_COUNTS_ENABLED)); + } + + /** + * Tells whether exposures are sent to Datadog. When off, exposures are still deduplicated so + * exposure hooks get the same decisions, and flag evaluation events are still sent. + * + * @return whether exposures are sent to Datadog (on by default). + */ + public boolean isExposuresDatadogLoggingEnabled() { + return !Boolean.FALSE.equals(getBoolean(EXPOSURES_DATADOG_LOGGING_ENABLED)); + } + + /** + * @return the Datadog API key, or {@code null} if not set. + */ + @Nullable + public String apiKey() { + return nonBlank(get(API_KEY)); + } + + /** + * @return the Datadog site. + */ + public String site() { + final String site = nonBlank(get(SITE)); + return site == null ? DEFAULT_SITE : site; + } + + /** + * @return the environment, or {@code null} if not set. + */ + @Nullable + public String env() { + return nonBlank(get(ENV)); + } + + /** + * @return the service name, or {@code null} if not set. + */ + @Nullable + public String service() { + return nonBlank(get(SERVICE)); + } + + /** + * @return the application version, or {@code null} if not set. + */ + @Nullable + public String version() { + return nonBlank(get(VERSION)); + } + + /** Reads the first non-blank value, in source priority order, like the agent. */ + @Nullable + String get(final String key) { + for (int source = 0; source < SOURCE_COUNT; source++) { + final String value = value(key, source); + if (nonBlank(value) != null) { + return value; + } + } + return null; + } + + /** + * Reads a setting from one source. + * + * @param key the setting key. + * @param source the source index, from the highest priority one. + * @return the raw setting value, or {@code null} if not set. + */ + @Nullable + private String value(final String key, final int source) { + switch (source) { + case 0: + return this.bridge.setting(key); + case 1: + return this.systemProperties.apply("dd." + key); + default: + return this.environment.apply(environmentVariableName(key)); + } + } + + @Nullable + private Boolean getBoolean(final String key) { + final String value = get(key); + if (value == null) { + return null; + } + // Same parsing as the agent: exact values, invalid ones are read as false. + if ("1".equals(value) || "true".equalsIgnoreCase(value)) { + return Boolean.TRUE; + } + if (!"0".equals(value) && !"false".equalsIgnoreCase(value)) { + LOGGER.warn("Invalid boolean value for {}: {}", key, value); + } + return Boolean.FALSE; + } + + /** Reads the first value that parses, in source priority order, like the agent. */ + private int getPositiveInt(final String key, final int defaultValue) { + for (int source = 0; source < SOURCE_COUNT; source++) { + final String value = value(key, source); + if (nonBlank(value) == null) { + continue; + } + try { + final int parsed = Integer.parseInt(value); + if (parsed > 0) { + return parsed; + } + LOGGER.warn("Invalid value for {}: {}. The value must be positive", key, value); + return defaultValue; + } catch (final NumberFormatException e) { + LOGGER.warn("Invalid value for {}: {}", key, value); + } + } + return defaultValue; + } + + static String environmentVariableName(final String key) { + return "DD_" + key.replace('.', '_').replace('-', '_').toUpperCase(Locale.ROOT); + } + + /** + * Normalizes the configuration source like the agent enum parsing: case insensitive, with no + * trimming of surrounding spaces. + */ + @Nullable + private static String normalizeSource(@Nullable final String value) { + return nonBlank(value) == null ? null : value.replace('/', '_').toLowerCase(Locale.ROOT); + } + + @Nullable + private static String nonBlank(@Nullable final String value) { + if (value == null) { + return null; + } + final String trimmed = value.trim(); + return trimmed.isEmpty() ? null : trimmed; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSource.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSource.java new file mode 100644 index 00000000000..cfb38849cab --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSource.java @@ -0,0 +1,348 @@ +package com.datadog.openfeature.internal.delivery; + +import static com.datadog.openfeature.internal.delivery.Retries.isRetryableStatus; + +import com.datadog.openfeature.internal.ConfigurationService; +import com.datadog.openfeature.internal.RuntimeServices; +import com.datadog.openfeature.internal.config.Settings; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import com.datadog.openfeature.internal.ufc.UniversalFlagConfigParser; +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.URI; +import java.net.URLEncoder; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.function.DoubleSupplier; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Polls the flag configuration from the Datadog managed CDN, or a custom HTTP endpoint. */ +public final class CdnConfigurationSource implements ConfigurationService { + private static final Logger LOGGER = LoggerFactory.getLogger(CdnConfigurationSource.class); + + static final String UFC_RULES_BASED_SERVER_PATH = + "/api/v2/feature-flagging/config/rules-based/server"; + static final int MAX_ATTEMPTS = 3; + private static final long FIRST_RETRY_MIN_MILLIS = 2_000; + private static final long FIRST_RETRY_MAX_MILLIS = 10_000; + private static final long SECOND_RETRY_MIN_MILLIS = 5_000; + private static final long SECOND_RETRY_MAX_MILLIS = 30_000; + private static final double RETRY_JITTER = 0.2; + private static final long WARNING_INTERVAL_NANOS = TimeUnit.MINUTES.toNanos(5); + + private final URI endpoint; + @Nullable private final String apiKey; + private final long pollIntervalMillis; + private final Duration requestTimeout; + private final HttpClient httpClient; + private final ScheduledExecutorService executor; + private final Retries.Sleeper retrySleeper; + private final DoubleSupplier jitter; + private final Object lifecycleLock = new Object(); + private final AtomicBoolean polling = new AtomicBoolean(); + private final CountDownLatch closedSignal = new CountDownLatch(1); + private HttpCall activeCall; + private volatile Consumer listener; + private volatile boolean closed; + private volatile boolean started; + private volatile ScheduledFuture scheduledPoll; + private volatile String etag; + private long lastWarningNanos; + + public CdnConfigurationSource(final Settings settings, final RuntimeServices services) { + this( + endpoint(settings), + isManagedEndpoint(settings) ? settings.apiKey() : null, + TimeUnit.SECONDS.toMillis(settings.agentlessPollIntervalSeconds()), + Duration.ofSeconds(settings.agentlessRequestTimeoutSeconds()), + HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(settings.agentlessRequestTimeoutSeconds())) + .followRedirects(HttpClient.Redirect.NEVER) + .build(), + Executors.newSingleThreadScheduledExecutor( + task -> services.newThread("configuration", task)), + null, + () -> ThreadLocalRandom.current().nextDouble(1 - RETRY_JITTER, 1 + RETRY_JITTER)); + } + + CdnConfigurationSource( + final URI endpoint, + @Nullable final String apiKey, + final long pollIntervalMillis, + final Duration requestTimeout, + final HttpClient httpClient, + final ScheduledExecutorService executor, + @Nullable final Retries.Sleeper retrySleeper, + final DoubleSupplier jitter) { + this.endpoint = endpoint; + this.apiKey = apiKey; + this.pollIntervalMillis = pollIntervalMillis; + this.requestTimeout = requestTimeout; + this.httpClient = httpClient; + this.executor = executor; + this.retrySleeper = + retrySleeper != null + ? retrySleeper + : delay -> this.closedSignal.await(delay, TimeUnit.MILLISECONDS); + this.jitter = jitter; + } + + @Override + public void start(final Consumer listener) { + synchronized (this.lifecycleLock) { + if (this.closed || this.started) { + return; + } + this.started = true; + this.listener = listener; + } + // Complete the first poll cycle on the calling thread so provider initialization observes a + // successful retry before it checks whether configuration is ready. + pollOnceSafely(); + synchronized (this.lifecycleLock) { + if (!this.closed) { + this.scheduledPoll = + this.executor.scheduleWithFixedDelay( + this::pollOnceSafely, + this.pollIntervalMillis, + this.pollIntervalMillis, + TimeUnit.MILLISECONDS); + } + } + } + + @Override + public void close() { + final ScheduledFuture poll; + final HttpCall call; + synchronized (this.lifecycleLock) { + if (this.closed) { + return; + } + this.closed = true; + poll = this.scheduledPoll; + this.scheduledPoll = null; + call = this.activeCall; + this.activeCall = null; + } + this.closedSignal.countDown(); + if (call != null) { + call.cancel(); + } + if (poll != null) { + poll.cancel(true); + } + this.executor.shutdownNow(); + } + + /** Test seam: sets the configuration listener without starting the polling. */ + void setListenerForTest(final Consumer listener) { + this.listener = listener; + } + + /** + * Polls the endpoint once. + * + * @return {@code true} if the configuration was applied or is unchanged, {@code false} otherwise. + */ + boolean pollOnce() { + if (this.closed || !this.polling.compareAndSet(false, true)) { + return false; + } + try { + return fetchAndApply(); + } finally { + synchronized (this.lifecycleLock) { + this.activeCall = null; + } + this.polling.set(false); + } + } + + private void pollOnceSafely() { + try { + pollOnce(); + } catch (final RuntimeException e) { + LOGGER.debug("Unexpected error while polling the Feature Flags configuration", e); + } + } + + private boolean fetchAndApply() { + final HttpResponse response; + try { + response = fetchWithRetries(); + } catch (final IOException e) { + if (!this.closed) { + warn("Feature Flags configuration request failed after {} attempts", MAX_ATTEMPTS, e); + } + return false; + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + if (this.closed) { + return false; + } + final int status = response.statusCode(); + if (isRetryableStatus(status)) { + warn( + "Feature Flags configuration endpoint failed after {} attempts with HTTP {}", + MAX_ATTEMPTS, + status); + return false; + } + synchronized (this.lifecycleLock) { + return !this.closed && apply(response); + } + } + + private boolean apply(final HttpResponse response) { + final int status = response.statusCode(); + if (status == HttpURLConnection.HTTP_NOT_MODIFIED) { + return true; + } + if (status == HttpURLConnection.HTTP_UNAUTHORIZED + || status == HttpURLConnection.HTTP_FORBIDDEN) { + warn( + "Feature Flags configuration endpoint returned HTTP {}; verify endpoint authentication", + status); + return false; + } + if (status != HttpURLConnection.HTTP_OK || response.body() == null) { + warn("Feature Flags configuration endpoint returned HTTP {}", status); + return false; + } + final ServerConfiguration configuration; + try { + configuration = UniversalFlagConfigParser.parseJsonApi(Gzip.decode(response)); + } catch (final IOException | RuntimeException e) { + LOGGER.debug("Feature Flags configuration endpoint returned a malformed payload", e); + return false; + } + if (configuration == null) { + return false; + } + this.listener.accept(configuration); + final String nextEtag = response.headers().firstValue("ETag").orElse(null); + this.etag = nextEtag == null || nextEtag.trim().isEmpty() ? null : nextEtag; + return true; + } + + private HttpResponse fetchWithRetries() throws IOException, InterruptedException { + final HttpRequest request = buildRequest(); + return Retries.send( + () -> { + synchronized (this.lifecycleLock) { + if (this.closed) { + throw new IOException("Feature Flags configuration source is closed"); + } + this.activeCall = + new HttpCall<>(this.httpClient, request, HttpResponse.BodyHandlers.ofByteArray()); + return this.activeCall; + } + }, + (response, attempt) -> + response == null || isRetryableStatus(response.statusCode()) + ? retryDelayMillis(this.pollIntervalMillis, attempt, this.jitter.getAsDouble()) + : -1, + MAX_ATTEMPTS, + this.retrySleeper, + () -> this.closed); + } + + private HttpRequest buildRequest() { + final HttpRequest.Builder request = + HttpRequest.newBuilder(this.endpoint) + .timeout(this.requestTimeout) + .header("Accept-Encoding", "gzip"); + Headers.addMetadata(request); + if (this.apiKey != null) { + request.header("DD-API-KEY", this.apiKey); + } + final String currentEtag = this.etag; + if (currentEtag != null) { + request.header("If-None-Match", currentEtag); + } + return request.GET().build(); + } + + private synchronized void warn(final String message, final Object... arguments) { + final long now = System.nanoTime(); + if (this.lastWarningNanos == 0 || now - this.lastWarningNanos >= WARNING_INTERVAL_NANOS) { + this.lastWarningNanos = now; + LOGGER.warn(message, arguments); + } else { + LOGGER.debug(message, arguments); + } + } + + static URI endpoint(final Settings settings) { + final String configuredBaseUrl = settings.agentlessBaseUrl(); + if (configuredBaseUrl == null) { + final String env = settings.env(); + return URI.create( + "https://" + + managedHost(settings) + + UFC_RULES_BASED_SERVER_PATH + + (env == null ? "" : "?dd_env=" + URLEncoder.encode(env, StandardCharsets.UTF_8))); + } + final URI parsed; + try { + parsed = URI.create(configuredBaseUrl); + } catch (final IllegalArgumentException e) { + throw new IllegalArgumentException( + "Invalid Feature Flags configuration source URL: " + configuredBaseUrl, e); + } + final String scheme = parsed.getScheme(); + if ((!"http".equalsIgnoreCase(scheme) && !"https".equalsIgnoreCase(scheme)) + || parsed.getHost() == null) { + throw new IllegalArgumentException( + "Invalid Feature Flags configuration source URL: " + configuredBaseUrl); + } + final String path = parsed.getRawPath(); + if (path == null || path.isEmpty() || "/".equals(path)) { + final String query = parsed.getRawQuery(); + return parsed.resolve( + query == null ? UFC_RULES_BASED_SERVER_PATH : UFC_RULES_BASED_SERVER_PATH + "?" + query); + } + return parsed; + } + + private static boolean isManagedEndpoint(final Settings settings) { + return settings.agentlessBaseUrl() == null; + } + + private static String managedHost(final Settings settings) { + return SiteHost.of("ufc-server.ff-cdn", settings.site()); + } + + static long retryDelayMillis( + final long pollIntervalMillis, final int attempt, final double jitter) { + final long baseDelay; + if (attempt == 1) { + baseDelay = clamp(pollIntervalMillis / 6, FIRST_RETRY_MIN_MILLIS, FIRST_RETRY_MAX_MILLIS); + } else if (attempt == 2) { + baseDelay = clamp(pollIntervalMillis / 3, SECOND_RETRY_MIN_MILLIS, SECOND_RETRY_MAX_MILLIS); + } else { + throw new IllegalArgumentException("Unsupported Feature Flags retry attempt: " + attempt); + } + return Math.max(1, Math.round(baseDelay * jitter)); + } + + private static long clamp(final long value, final long minimum, final long maximum) { + return Math.max(minimum, Math.min(maximum, value)); + } +} 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 new file mode 100644 index 00000000000..9c004fd54c6 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java @@ -0,0 +1,136 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.net.http.HttpClient.Redirect.NEVER; +import static java.net.http.HttpRequest.BodyPublishers.ofByteArray; + +import com.datadog.openfeature.internal.bridge.EventTransport; +import com.datadog.openfeature.internal.config.Settings; +import java.io.IOException; +import java.net.ConnectException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.time.Duration; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; + +/** Posts events directly to the Datadog event platform intake, authenticated with an API key. */ +public final class DirectIntakeTransport implements EventTransport { + static final int MAX_RETRIES = 5; + private static final long INITIAL_RETRY_DELAY_MILLIS = 100; + private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(10); + + private final URI intake; + private final String apiKey; + private final String traceId; + private final HttpClient httpClient; + private final Retries.Sleeper retrySleeper; + + DirectIntakeTransport( + final URI intake, + final String apiKey, + final HttpClient httpClient, + final Retries.Sleeper retrySleeper) { + this.intake = intake; + this.apiKey = apiKey; + // Intake requests are attributed to a stable, random trace id. + this.traceId = Long.toUnsignedString(ThreadLocalRandom.current().nextLong() >>> 1); + this.httpClient = httpClient; + this.retrySleeper = retrySleeper; + } + + /** + * Creates a direct intake transport. + * + * @param settings the SDK settings. + * @return the transport, or {@code null} if no API key is configured. + * @throws IllegalArgumentException if the configured site is invalid. + */ + @Nullable + public static DirectIntakeTransport create(final Settings settings) { + final String apiKey = settings.apiKey(); + if (apiKey == null) { + return null; + } + return new DirectIntakeTransport( + URI.create("https://" + SiteHost.of("event-platform-intake", settings.site()) + "/api/v2/"), + apiKey, + HttpClient.newBuilder().connectTimeout(REQUEST_TIMEOUT).followRedirects(NEVER).build(), + TimeUnit.MILLISECONDS::sleep); + } + + @Override + public void post(final String route, final byte[] json) throws IOException { + final HttpRequest.Builder builder = + HttpRequest.newBuilder(this.intake.resolve(route)) + .timeout(REQUEST_TIMEOUT) + .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); + Headers.addMetadata(builder); + final HttpRequest request = builder.POST(ofByteArray(json)).build(); + final int status; + try { + status = + Retries.send( + () -> + new HttpCall<>( + this.httpClient, request, HttpResponse.BodyHandlers.discarding()), + new IntakeRetryPolicy(), + MAX_RETRIES + 1, + this.retrySleeper, + () -> false) + .statusCode(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Request to " + route + " was interrupted", e); + } + if (status < 200 || status >= 300) { + throw new IOException("Request to " + route + " returned error response " + status); + } + } + + private static final class IntakeRetryPolicy implements Retries.Policy { + private boolean rateLimitRetry; + + @Override + public boolean shouldRetry(final IOException error) { + return error instanceof ConnectException || error instanceof HttpTimeoutException; + } + + @Override + public long retryDelayMillis(@Nullable final HttpResponse response, final int attempt) { + if (this.rateLimitRetry) { + return -1; + } + if (response != null) { + final int status = response.statusCode(); + if (status == 429) { + final String reset = response.headers().firstValue("X-RateLimit-Reset").orElse(null); + if (reset != null) { + try { + final long seconds = Long.parseLong(reset); + if (seconds > 10) { + return -1; + } + if (seconds >= 0) { + this.rateLimitRetry = true; + return TimeUnit.SECONDS.toMillis(seconds) + + ThreadLocalRandom.current().nextInt(401); + } + } catch (final NumberFormatException ignored) { + // An invalid reset header falls back to exponential backoff. + } + } + } else if (status < 500 || status > 599) { + return -1; + } + } + return INITIAL_RETRY_DELAY_MILLIS << (attempt - 1); + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java new file mode 100644 index 00000000000..e01bf019672 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/FallbackTransport.java @@ -0,0 +1,39 @@ +package com.datadog.openfeature.internal.delivery; + +import com.datadog.openfeature.internal.bridge.EventProxyUnavailableException; +import com.datadog.openfeature.internal.bridge.EventTransport; +import java.io.IOException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Posts through the event proxy, switching to direct intake for good once the proxy is definitively + * unavailable. Ambiguous proxy failures are not replayed through direct intake. + */ +public final class FallbackTransport implements EventTransport { + private static final Logger LOGGER = LoggerFactory.getLogger(FallbackTransport.class); + + private final EventTransport proxy; + private final EventTransport direct; + private volatile boolean proxyUnavailable; + + public FallbackTransport(final EventTransport proxy, final EventTransport direct) { + this.proxy = proxy; + this.direct = direct; + } + + @Override + public void post(final String route, final byte[] json) throws IOException { + if (!this.proxyUnavailable) { + try { + this.proxy.post(route, json); + return; + } catch (final EventProxyUnavailableException e) { + LOGGER.debug( + "Switching Feature Flags event delivery from the event proxy to direct intake"); + this.proxyUnavailable = true; + } + } + this.direct.post(route, json); + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Gzip.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Gzip.java new file mode 100644 index 00000000000..0525ebd397d --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Gzip.java @@ -0,0 +1,34 @@ +package com.datadog.openfeature.internal.delivery; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.http.HttpResponse; +import java.util.zip.GZIPInputStream; + +/** Gzip decoding, as the JDK HTTP client does not handle content encoding. */ +final class Gzip { + private Gzip() {} + + /** + * Decodes a response body according to its {@code Content-Encoding} header. + * + * @param response the response to decode the body of. + * @return the decoded body. + * @throws IOException if the body is not valid gzip content. + */ + static byte[] decode(final HttpResponse response) throws IOException { + final boolean gzipped = + response + .headers() + .firstValue("Content-Encoding") + .map("gzip"::equalsIgnoreCase) + .orElse(false); + if (!gzipped) { + return response.body(); + } + try (InputStream input = new GZIPInputStream(new ByteArrayInputStream(response.body()))) { + return input.readAllBytes(); + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Headers.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Headers.java new file mode 100644 index 00000000000..e61ab4735a0 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Headers.java @@ -0,0 +1,27 @@ +package com.datadog.openfeature.internal.delivery; + +import com.datadog.openfeature.internal.config.SdkVersion; +import java.net.http.HttpRequest; + +/** Common Datadog request headers. */ +final class Headers { + private static final String JAVA_VERSION = System.getProperty("java.version", "unknown"); + private static final String JAVA_VM_NAME = System.getProperty("java.vm.name", "unknown"); + private static final String JAVA_VM_VENDOR = System.getProperty("java.vm.vendor", "unknown"); + + private Headers() {} + + /** + * Adds the language and SDK metadata headers. + * + * @param request the request to add the headers to. + */ + static void addMetadata(final HttpRequest.Builder request) { + request + .header("Datadog-Meta-Lang", "java") + .header("Datadog-Meta-Lang-Version", JAVA_VERSION) + .header("Datadog-Meta-Lang-Interpreter", JAVA_VM_NAME) + .header("Datadog-Meta-Lang-Interpreter-Vendor", JAVA_VM_VENDOR) + .header("Datadog-Meta-Tracer-Version", SdkVersion.VERSION); + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/HttpCall.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/HttpCall.java new file mode 100644 index 00000000000..d5a2f9876dd --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/HttpCall.java @@ -0,0 +1,125 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.lang.Math.max; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import java.io.IOException; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Flow; +import java.util.concurrent.TimeoutException; + +/** A cancellable HTTP attempt whose timeout includes response body consumption. */ +final class HttpCall { + private final long startedNanos = System.nanoTime(); + private final long timeoutNanos; + private final CompletableFuture> response; + private volatile Flow.Subscription subscription; + private volatile boolean cancelled; + + HttpCall( + final HttpClient client, + final HttpRequest request, + final HttpResponse.BodyHandler bodyHandler) { + this.timeoutNanos = request.timeout().orElseThrow().toNanos(); + this.response = + client.sendAsync(request, info -> new CancellableBodySubscriber(bodyHandler.apply(info))); + } + + /** + * Waits for the complete response. + * + * @return the response, including its consumed body. + * @throws IOException if the request fails, times out, or is cancelled. + * @throws InterruptedException if the caller is interrupted. + */ + HttpResponse execute() throws IOException, InterruptedException { + try { + return this.response.get( + max(0, this.timeoutNanos - (System.nanoTime() - this.startedNanos)), NANOSECONDS); + } catch (final TimeoutException e) { + cancel(); + throw new HttpTimeoutException("HTTP response timed out"); + } catch (final InterruptedException e) { + cancel(); + throw e; + } catch (final CancellationException e) { + throw new IOException("HTTP request cancelled", e); + } catch (final ExecutionException e) { + final Throwable cause = e.getCause(); + if (cause instanceof IOException) { + throw (IOException) cause; + } + if (cause instanceof CancellationException) { + throw new IOException("HTTP request cancelled", cause); + } + if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } + if (cause instanceof Error) { + throw (Error) cause; + } + throw new IOException(cause); + } + } + + /** Cancels the active exchange and any response body subscription. */ + void cancel() { + if (!this.cancelled && this.response.cancel(true)) { + this.cancelled = true; + final Flow.Subscription subscription = this.subscription; + if (subscription != null) { + subscription.cancel(); + } + } + } + + private void subscribe(final Flow.Subscription subscription) { + this.subscription = subscription; + if (this.cancelled) { + subscription.cancel(); + } + } + + private final class CancellableBodySubscriber implements HttpResponse.BodySubscriber { + private final HttpResponse.BodySubscriber delegate; + + private CancellableBodySubscriber(final HttpResponse.BodySubscriber delegate) { + this.delegate = delegate; + } + + @Override + public CompletionStage getBody() { + return this.delegate.getBody(); + } + + @Override + public void onSubscribe(final Flow.Subscription subscription) { + this.delegate.onSubscribe(subscription); + subscribe(subscription); + } + + @Override + public void onNext(final List buffers) { + this.delegate.onNext(buffers); + } + + @Override + public void onError(final Throwable error) { + this.delegate.onError(error); + } + + @Override + public void onComplete() { + this.delegate.onComplete(); + } + } +} 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 new file mode 100644 index 00000000000..8d1a4c6434a --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/RemoteConfigurationService.java @@ -0,0 +1,59 @@ +package com.datadog.openfeature.internal.delivery; + +import com.datadog.openfeature.internal.ConfigurationService; +import com.datadog.openfeature.internal.bridge.ConfigurationSource; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import com.datadog.openfeature.internal.ufc.UniversalFlagConfigParser; +import java.io.IOException; +import java.util.function.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Receives the flag configuration from Remote Configuration, through the Datadog Java agent. */ +public final class RemoteConfigurationService implements ConfigurationService { + private static final Logger LOGGER = LoggerFactory.getLogger(RemoteConfigurationService.class); + + private final ConfigurationSource source; + private AutoCloseable subscription; + + public RemoteConfigurationService(final ConfigurationSource source) { + this.source = source; + } + + @Override + public synchronized void start(final Consumer listener) { + if (this.subscription != null) { + return; + } + this.subscription = + this.source.subscribe( + content -> { + if (content == null) { + listener.accept(null); + return; + } + final ServerConfiguration configuration; + try { + configuration = UniversalFlagConfigParser.parse(content); + } catch (final IOException | RuntimeException e) { + LOGGER.debug("Remote Configuration returned a malformed flag configuration", e); + return; + } + if (configuration != null) { + listener.accept(configuration); + } + }); + } + + @Override + public synchronized void close() { + if (this.subscription != null) { + try { + this.subscription.close(); + } catch (final Exception e) { + LOGGER.debug("Failed to unsubscribe from Remote Configuration", e); + } + this.subscription = null; + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Retries.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Retries.java new file mode 100644 index 00000000000..7c1544e4708 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/Retries.java @@ -0,0 +1,104 @@ +package com.datadog.openfeature.internal.delivery; + +import java.io.IOException; +import java.net.HttpURLConnection; +import java.net.http.HttpResponse; +import java.util.function.BooleanSupplier; +import javax.annotation.Nullable; + +/** Sends HTTP requests, retrying the failures that can succeed on a later attempt. */ +final class Retries { + private Retries() {} + + /** + * Sends a request until it gets a non-retryable response, the attempts are exhausted, or the + * caller gives up. + * + * @param callFactory creates a new call for each attempt. + * @param policy the retry rules and delays. + * @param maxAttempts the maximum number of attempts. + * @param sleeper the sleeper waiting between attempts. + * @param cancelled whether the caller gave up on the request. + * @param the response body type. + * @return the last response, which can still have a retryable status. + * @throws IOException if the last attempt failed. + * @throws InterruptedException if interrupted while sending or waiting between attempts. + */ + static HttpResponse send( + final CallFactory callFactory, + final Policy policy, + final int maxAttempts, + final Sleeper sleeper, + final BooleanSupplier cancelled) + throws IOException, InterruptedException { + for (int attempt = 1; ; attempt++) { + if (cancelled.getAsBoolean()) { + throw new IOException("HTTP request cancelled"); + } + if (Thread.interrupted()) { + throw new InterruptedException(); + } + long delayMillis; + try { + final HttpResponse response = callFactory.newCall().execute(); + if (attempt == maxAttempts || cancelled.getAsBoolean()) { + return response; + } + delayMillis = policy.retryDelayMillis(response, attempt); + if (delayMillis < 0) { + return response; + } + } catch (final IOException e) { + if (attempt == maxAttempts + || !policy.shouldRetry(e) + || cancelled.getAsBoolean() + || Thread.currentThread().isInterrupted()) { + throw e; + } + delayMillis = policy.retryDelayMillis(null, attempt); + if (delayMillis < 0) { + throw e; + } + } + sleeper.sleep(delayMillis); + } + } + + /** + * Identifies retryable CDN responses. + * + * @param status the HTTP response status. + * @return whether a request that got this status can succeed on a later attempt. + */ + static boolean isRetryableStatus(final int status) { + return status == HttpURLConnection.HTTP_CLIENT_TIMEOUT + || status == 429 + || (status >= 500 && status <= 599); + } + + @FunctionalInterface + interface CallFactory { + HttpCall newCall() throws IOException; + } + + @FunctionalInterface + interface Policy { + /** + * Computes the delay before retrying an attempt. + * + * @param response the failed response, or {@code null} for an I/O failure. + * @param attempt the failed attempt number. + * @return the delay in milliseconds, or a negative value to stop retrying. + */ + long retryDelayMillis(@Nullable HttpResponse response, int attempt); + + default boolean shouldRetry(final IOException error) { + return true; + } + } + + @FunctionalInterface + interface Sleeper { + void sleep(long delayMillis) throws InterruptedException; + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/SiteHost.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/SiteHost.java new file mode 100644 index 00000000000..f12639c3ad2 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/SiteHost.java @@ -0,0 +1,30 @@ +package com.datadog.openfeature.internal.delivery; + +import java.net.URI; +import java.net.URISyntaxException; + +/** Builds the host of a Datadog endpoint from the configured site. */ +final class SiteHost { + private SiteHost() {} + + /** + * Builds the host of a Datadog endpoint. + * + * @param prefix the endpoint host prefix. + * @param site the configured Datadog site. + * @return the endpoint host. + * @throws IllegalArgumentException if the site does not make a plain host name. + */ + static String of(final String prefix, final String site) { + final String host = prefix + "." + site; + try { + final URI uri = new URI("https", null, host, -1, "/", null, null); + if (host.equalsIgnoreCase(uri.getHost())) { + return host; + } + } catch (final URISyntaxException ignored) { + // Reported below + } + throw new IllegalArgumentException("Invalid Datadog site: " + site); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/EventQueueTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/EventQueueTest.java new file mode 100644 index 00000000000..fbee9cc5944 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/EventQueueTest.java @@ -0,0 +1,143 @@ +package com.datadog.openfeature.internal; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class EventQueueTest { + @Test + void rejectsNullWithoutConsumingCapacity() { + final EventQueue queue = new EventQueue<>(1); + + assertThrows(NullPointerException.class, () -> queue.offer(null)); + assertEquals(0, queue.size()); + assertTrue(queue.offer("a")); + + assertThrows(NullPointerException.class, () -> queue.offer(null)); + assertEquals(1, queue.size()); + assertEquals("a", queue.poll()); + assertEquals(0, queue.size()); + } + + @Test + void rejectsOffersBeyondCapacity() { + final EventQueue queue = new EventQueue<>(2); + + assertTrue(queue.offer("a")); + assertTrue(queue.offer("b")); + assertFalse(queue.offer("c")); + + assertEquals(2, queue.size()); + assertEquals(2, queue.capacity()); + assertEquals("a", queue.poll()); + assertTrue(queue.offer("c")); + } + + @Test + void pollsInFifoOrderAndReturnsNullWhenEmpty() { + final EventQueue queue = new EventQueue<>(4); + queue.offer(1); + queue.offer(2); + + assertEquals(1, queue.poll()); + assertEquals(2, queue.poll()); + assertNull(queue.poll()); + assertEquals(0, queue.size()); + } + + @Test + void timedPollWaitsForAnElement() throws Exception { + final EventQueue queue = new EventQueue<>(1); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + executor.submit( + () -> { + MILLISECONDS.sleep(50); + return queue.offer("late"); + }); + + assertEquals("late", queue.poll(5, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + } + + @Test + void timedPollTimesOut() throws Exception { + final EventQueue queue = new EventQueue<>(1); + final long start = System.nanoTime(); + + assertNull(queue.poll(30, MILLISECONDS)); + + assertTrue(System.nanoTime() - start >= MILLISECONDS.toNanos(30)); + } + + @Test + void timedPollIsInterruptible() { + final EventQueue queue = new EventQueue<>(1); + Thread.currentThread().interrupt(); + + assertThrows(InterruptedException.class, () -> queue.poll(5, TimeUnit.SECONDS)); + assertFalse(Thread.currentThread().isInterrupted()); + } + + @Test + void drainsUpToTheLimit() { + final EventQueue queue = new EventQueue<>(8); + for (int i = 0; i < 5; i++) { + queue.offer(i); + } + final List drained = new ArrayList<>(); + + queue.drain(drained::add, 3); + + assertEquals(List.of(0, 1, 2), drained); + assertEquals(2, queue.size()); + queue.drain(drained::add, 10); + assertEquals(List.of(0, 1, 2, 3, 4), drained); + } + + @Test + void concurrentProducersNeverExceedCapacity() throws Exception { + final int capacity = 1000; + final EventQueue queue = new EventQueue<>(capacity); + final int producers = 8; + final ExecutorService executor = Executors.newFixedThreadPool(producers); + final CountDownLatch start = new CountDownLatch(1); + final AtomicInteger accepted = new AtomicInteger(); + try { + for (int p = 0; p < producers; p++) { + executor.submit( + () -> { + start.await(); + for (int i = 0; i < 500; i++) { + if (queue.offer(i)) { + accepted.incrementAndGet(); + } + } + return null; + }); + } + start.countDown(); + executor.shutdown(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + + assertEquals(capacity, accepted.get()); + assertEquals(capacity, queue.size()); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/JsonReading.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/JsonReading.java new file mode 100644 index 00000000000..82e888e5b0a --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/JsonReading.java @@ -0,0 +1,101 @@ +package com.datadog.openfeature.internal; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** Reads JSON documents as generic trees for test assertions. */ +public final class JsonReading { + private static final JsonFactory JSON_FACTORY = new JsonFactory(); + + private JsonReading() {} + + /** Reads a JSON document. Integral numbers are read as {@link Long}, others as {@link Double}. */ + public static Object read(final byte[] json) { + try (JsonParser parser = JSON_FACTORY.createParser(json)) { + parser.nextToken(); + return readValue(parser); + } catch (final IOException e) { + throw new UncheckedIOException(e); + } + } + + public static Object read(final String json) { + return read(json.getBytes(UTF_8)); + } + + @SuppressWarnings("unchecked") + public static Map readObject(final byte[] json) { + return (Map) read(json); + } + + @SuppressWarnings("unchecked") + public static Map readObject(final String json) { + return (Map) read(json); + } + + /** Reads a JSON object with all numbers as {@link Double}, like Moshi's generic adapters. */ + @SuppressWarnings("unchecked") + public static Map readObjectWithDoubles(final byte[] json) { + return (Map) toDoubles(read(json)); + } + + private static Object toDoubles(final Object value) { + if (value instanceof Long) { + return ((Long) value).doubleValue(); + } + if (value instanceof Map) { + final Map map = new LinkedHashMap<>(); + ((Map) value).forEach((k, v) -> map.put((String) k, toDoubles(v))); + return map; + } + if (value instanceof List) { + final List list = new ArrayList<>(); + ((List) value).forEach(v -> list.add(toDoubles(v))); + return list; + } + return value; + } + + private static Object readValue(final JsonParser parser) throws IOException { + final JsonToken token = parser.currentToken(); + switch (token) { + case START_OBJECT: + final Map object = new LinkedHashMap<>(); + while (parser.nextToken() == JsonToken.FIELD_NAME) { + final String name = parser.currentName(); + parser.nextToken(); + object.put(name, readValue(parser)); + } + return object; + case START_ARRAY: + final List array = new ArrayList<>(); + while (parser.nextToken() != JsonToken.END_ARRAY) { + array.add(readValue(parser)); + } + return array; + case VALUE_STRING: + return parser.getText(); + case VALUE_NUMBER_INT: + return parser.getLongValue(); + case VALUE_NUMBER_FLOAT: + return parser.getDoubleValue(); + case VALUE_TRUE: + return Boolean.TRUE; + case VALUE_FALSE: + return Boolean.FALSE; + case VALUE_NULL: + return null; + default: + throw new IOException("Unexpected token " + token); + } + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/LocalHttpServer.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/LocalHttpServer.java new file mode 100644 index 00000000000..c2d2e9dbdc6 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/LocalHttpServer.java @@ -0,0 +1,163 @@ +package com.datadog.openfeature.internal; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.sun.net.httpserver.Headers; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.URI; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.zip.GZIPOutputStream; + +/** A local HTTP server replaying scripted responses and recording the received requests. */ +public final class LocalHttpServer implements AutoCloseable { + private final HttpServer server; + private final ConcurrentLinkedQueue responses = new ConcurrentLinkedQueue<>(); + private final List requests = new CopyOnWriteArrayList<>(); + private final BlockingQueue received = new LinkedBlockingQueue<>(); + private volatile Response fallback = new Response(404, null, Map.of(), false); + + public LocalHttpServer() throws IOException { + this.server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0); + this.server.createContext("/", this::handle); + this.server.start(); + } + + /** + * @param path the request path, with an optional query. + * @return the URI of the given path on this server. + */ + public URI uri(final String path) { + return URI.create( + "http://" + + this.server.getAddress().getHostString() + + ":" + + this.server.getAddress().getPort() + + path); + } + + /** Queues a response. */ + public LocalHttpServer enqueue(final int status, final String body, final String... headers) { + this.responses.add(response(status, body, false, headers)); + return this; + } + + /** Queues a gzip encoded response. */ + public LocalHttpServer enqueueGzip(final int status, final String body, final String... headers) { + this.responses.add(response(status, body, true, headers)); + return this; + } + + /** Sets the response returned once queued responses are exhausted. */ + public LocalHttpServer otherwise(final int status, final String body, final String... headers) { + this.fallback = response(status, body, false, headers); + return this; + } + + public List requests() { + return this.requests; + } + + public BlockingQueue received() { + return this.received; + } + + @Override + public void close() { + this.server.stop(0); + } + + private static Response response( + final int status, final String body, final boolean gzip, final String... headers) { + final Map headerMap = new java.util.HashMap<>(); + for (int i = 0; i < headers.length; i += 2) { + headerMap.put(headers[i], headers[i + 1]); + } + return new Response(status, body, headerMap, gzip); + } + + private void handle(final HttpExchange exchange) throws IOException { + final Request request = + new Request( + exchange.getRequestMethod(), + exchange.getRequestURI(), + exchange.getRequestHeaders(), + exchange.getRequestBody().readAllBytes()); + this.requests.add(request); + this.received.add(request); + Response response = this.responses.poll(); + if (response == null) { + response = this.fallback; + } + for (final Map.Entry header : response.headers.entrySet()) { + exchange.getResponseHeaders().add(header.getKey(), header.getValue()); + } + byte[] body = response.body == null ? new byte[0] : response.body.getBytes(UTF_8); + if (response.gzip) { + exchange.getResponseHeaders().add("Content-Encoding", "gzip"); + final ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(compressed)) { + gzip.write(body); + } + body = compressed.toByteArray(); + } + exchange.sendResponseHeaders(response.status, body.length == 0 ? -1 : body.length); + if (body.length > 0) { + try (OutputStream output = exchange.getResponseBody()) { + output.write(body); + } + } + exchange.close(); + } + + /** A received request. */ + public static final class Request { + public final String method; + public final URI uri; + public final Headers headers; + public final byte[] body; + + Request(final String method, final URI uri, final Headers headers, final byte[] body) { + this.method = method; + this.uri = uri; + this.headers = headers; + this.body = body; + } + + public String header(final String name) { + return this.headers.getFirst(name); + } + + public String bodyAsString() { + return new String(this.body, UTF_8); + } + } + + private static final class Response { + final int status; + final String body; + final Map headers; + final boolean gzip; + + Response( + final int status, + final String body, + final Map headers, + final boolean gzip) { + this.status = status; + this.body = body; + this.headers = headers; + this.gzip = gzip; + } + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java new file mode 100644 index 00000000000..1cfd0a827c5 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/SettingsTest.java @@ -0,0 +1,203 @@ +package com.datadog.openfeature.internal.config; + +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.assertTrue; + +import com.datadog.openfeature.internal.bridge.AgentBridge; +import com.datadog.openfeature.internal.bridge.ConfigurationSource; +import com.datadog.openfeature.internal.bridge.EventTransport; +import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.SpanEnricher; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.tabletest.junit.TableTest; + +class SettingsTest { + @Test + void looksUpBridgeThenSystemPropertiesThenEnvironment() { + final Map bridge = new HashMap<>(); + final Map properties = new HashMap<>(); + final Map environment = new HashMap<>(); + final Settings settings = + new Settings(new SettingAgentBridge(bridge), properties::get, environment::get); + + assertNull(settings.env()); + environment.put("DD_ENV", "from-env"); + assertEquals("from-env", settings.env()); + properties.put("dd.env", "from-property"); + assertEquals("from-property", settings.env()); + bridge.put("env", "from-agent"); + assertEquals("from-agent", settings.env()); + } + + @TableTest({ + "scenario | key | variable ", + "dotted key | 'feature.flags.enabled' | 'DD_FEATURE_FLAGS_ENABLED' ", + "dashed key | 'api-key' | 'DD_API_KEY' ", + "deeply nested key | 'flagging.evaluation.counts.enabled' | 'DD_FLAGGING_EVALUATION_COUNTS_ENABLED'" + }) + void mapsKeysToEnvironmentVariables(final String key, final String variable) { + assertEquals(variable, Settings.environmentVariableName(key)); + } + + @TableTest({ + "scenario | enabled | source | legacy | resolved ", + "default | | | | 'agentless' ", + "explicit agentless | | 'agentless' | | 'agentless' ", + "explicit remote config | | 'remote_config' | | 'remote_config'", + "case insensitive source | | 'Remote_Config' | | 'remote_config'", + "padded source | | ' agentless ' | | ", + "unsupported source | | 'offline' | | ", + "blank source | | ' ' | | 'agentless' ", + "disabled wins over source | 'false' | 'remote_config' | | ", + "enabled keeps default | 'true' | | | 'agentless' ", + "legacy enabled selects rc | | | 'true' | 'remote_config'", + "legacy disabled disables | | | 'false' | ", + "explicit source wins over legacy | | 'agentless' | 'false' | 'agentless' ", + "invalid enabled reads as false | 'maybe' | | | ", + "numeric disabled | '0' | | | ", + "padded enabled reads as false | ' true ' | | | ", + "padded legacy reads as false | | | ' true ' | " + }) + void resolvesConfigurationSource( + final String enabled, final String source, final String legacy, final String resolved) { + final Map properties = new HashMap<>(); + putIfNotNull(properties, "dd.feature.flags.enabled", enabled); + putIfNotNull(properties, "dd.feature.flags.configuration.source", source); + putIfNotNull(properties, "dd.experimental.flagging.provider.enabled", legacy); + final Settings settings = new Settings(AgentBridge.NONE, properties::get, name -> null); + + assertEquals(resolved, settings.configurationSource()); + assertEquals(resolved != null, settings.isEnabled()); + } + + @Test + void appliesDefaults() { + final Settings settings = TestSettings.of(); + + assertEquals("datadoghq.com", settings.site()); + assertNull(settings.apiKey()); + assertNull(settings.service()); + assertNull(settings.version()); + assertNull(settings.agentlessBaseUrl()); + assertEquals(30, settings.agentlessPollIntervalSeconds()); + assertEquals(5, settings.agentlessRequestTimeoutSeconds()); + assertFalse(settings.isSpanEnrichmentEnabled()); + assertTrue(settings.isEvaluationCountsEnabled()); + assertTrue(settings.isExposuresDatadogLoggingEnabled()); + } + + @TableTest({ + "scenario | value | enabled", + "true | 'true' | true ", + "one | '1' | true ", + "false | 'false' | false ", + "zero | '0' | false ", + "invalid | 'nope' | false ", + "blank | ' ' | true " + }) + void readsTheExposuresDatadogLoggingToggle(final String value, final boolean enabled) { + assertEquals( + enabled, + TestSettings.of("feature.flags.exposures.datadog.logging.enabled", value) + .isExposuresDatadogLoggingEnabled()); + } + + @TableTest({ + "scenario | value | interval", + "valid | '10' | 10 ", + "zero | '0' | 30 ", + "negative | '-5' | 30 ", + "invalid | 'ten' | 30 ", + "blank | ' ' | 30 ", + "padded | ' 10 ' | 30 " + }) + void fallsBackToDefaultPollIntervalOnInvalidValues(final String value, final int interval) { + assertEquals( + interval, + TestSettings.of("feature.flags.configuration.source.agentless.poll.interval.seconds", value) + .agentlessPollIntervalSeconds()); + } + + @Test + void fallsBackToLowerPrioritySourcesOnBlankOrInvalidValues() { + final Map properties = new HashMap<>(); + properties.put("dd.flagging.evaluation.counts.enabled", " "); + properties.put("dd.feature.flags.configuration.source.agentless.poll.interval.seconds", "ten"); + final Map environment = new HashMap<>(); + environment.put("DD_FLAGGING_EVALUATION_COUNTS_ENABLED", "false"); + environment.put("DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS", "10"); + final Settings settings = new Settings(AgentBridge.NONE, properties::get, environment::get); + + assertFalse(settings.isEvaluationCountsEnabled()); + assertEquals(10, settings.agentlessPollIntervalSeconds()); + } + + @Test + void trimsAndIgnoresBlankValues() { + final Settings settings = + TestSettings.of( + "api-key", " secret ", + "site", " ", + "feature.flags.configuration.source.agentless.base.url", " "); + + assertEquals("secret", settings.apiKey()); + assertEquals("datadoghq.com", settings.site()); + assertNull(settings.agentlessBaseUrl()); + } + + @Test + void readsToggles() { + final Settings settings = + TestSettings.of( + "experimental.flagging.provider.span.enrichment.enabled", "true", + "flagging.evaluation.counts.enabled", "false"); + + assertTrue(settings.isSpanEnrichmentEnabled()); + assertFalse(settings.isEvaluationCountsEnabled()); + } + + private static void putIfNotNull( + final Map map, final String key, final String value) { + if (value != null) { + map.put(key, value); + } + } + + /** A bridge only providing settings. */ + private static final class SettingAgentBridge implements AgentBridge { + private final Map settings; + + SettingAgentBridge(final Map settings) { + this.settings = settings; + } + + @Override + public String setting(final String key) { + return this.settings.get(key); + } + + @Override + public ConfigurationSource remoteConfiguration() { + return null; + } + + @Override + public EventTransport eventProxy() { + return null; + } + + @Override + public HealthMetrics healthMetrics() { + return HealthMetrics.NOOP; + } + + @Override + public SpanEnricher spanEnricher() { + return SpanEnricher.NOOP; + } + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/TestSettings.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/TestSettings.java new file mode 100644 index 00000000000..6acc0792138 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/config/TestSettings.java @@ -0,0 +1,31 @@ +package com.datadog.openfeature.internal.config; + +import com.datadog.openfeature.internal.bridge.AgentBridge; +import java.util.HashMap; +import java.util.Map; + +/** Builds {@link Settings} from explicit values, ignoring the JVM system properties and env. */ +public final class TestSettings { + private TestSettings() {} + + /** + * @param keyValues alternating setting keys, without the {@code dd.} prefix, and values. + * @return the settings backed by the given values only. + */ + public static Settings of(final String... keyValues) { + return of(AgentBridge.NONE, keyValues); + } + + /** + * @param bridge the bridge to look settings up from first. + * @param keyValues alternating setting keys, without the {@code dd.} prefix, and values. + * @return the settings backed by the bridge and the given values only. + */ + public static Settings of(final AgentBridge bridge, final String... keyValues) { + final Map properties = new HashMap<>(); + for (int i = 0; i < keyValues.length; i += 2) { + properties.put("dd." + keyValues[i], keyValues[i + 1]); + } + return new Settings(bridge, properties::get, name -> null); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSourceTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSourceTest.java new file mode 100644 index 00000000000..7ddcbaba216 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/CdnConfigurationSourceTest.java @@ -0,0 +1,563 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +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 com.datadog.openfeature.internal.LocalHttpServer; +import com.datadog.openfeature.internal.RuntimeServices; +import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.config.Settings; +import com.datadog.openfeature.internal.config.TestSettings; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.net.http.HttpClient; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.FutureTask; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.tabletest.junit.TableTest; + +class CdnConfigurationSourceTest { + private static final long POLL_INTERVAL_MILLIS = 30_000; + private static final String CONFIG = + "{\"data\":{\"type\":\"universal-flag-configuration\",\"attributes\":" + + "{\"createdAt\":\"2024-04-17T19:40:53.716Z\",\"environment\":{\"name\":\"Test\"}," + + "\"flags\":{}}}}"; + private static final String RAW_UFC = + "{\"createdAt\":\"2024-04-17T19:40:53.716Z\",\"environment\":{\"name\":\"Test\"},\"flags\":{}}"; + + private LocalHttpServer server; + private final List dispatched = new CopyOnWriteArrayList<>(); + private final List sleeps = new CopyOnWriteArrayList<>(); + + @BeforeEach + void setUp() throws IOException { + this.server = new LocalHttpServer(); + } + + @AfterEach + void tearDown() { + this.server.close(); + } + + @Test + void derivesDatadogUfcCdnEndpointFromSiteAndEnv() { + final URI endpoint = + CdnConfigurationSource.endpoint(TestSettings.of("site", "datadoghq.eu", "env", "prod")); + + assertEquals( + "https://ufc-server.ff-cdn.datadoghq.eu/api/v2/feature-flagging/config/rules-based/server?dd_env=prod", + endpoint.toString()); + } + + @Test + void derivesDatadogUfcCdnEndpointWithoutEnv() { + assertEquals( + "https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server", + CdnConfigurationSource.endpoint(TestSettings.of()).toString()); + } + + @Test + void encodesEnvInDatadogUfcCdnEndpoint() { + assertEquals( + "https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server?dd_env=a+b%26c", + CdnConfigurationSource.endpoint(TestSettings.of("env", "a b&c")).toString()); + } + + @Test + void rejectsInvalidSiteForDatadogUfcCdnEndpoint() { + assertThrows( + IllegalArgumentException.class, + () -> + CdnConfigurationSource.endpoint( + TestSettings.of("site", "datadoghq.com@other.example"))); + } + + @TableTest({ + "scenario | baseUrl | endpoint ", + "host only | 'http://localhost:8080' | 'http://localhost:8080/api/v2/feature-flagging/config/rules-based/server' ", + "root path | 'http://localhost:8080/' | 'http://localhost:8080/api/v2/feature-flagging/config/rules-based/server' ", + "root query | 'https://proxy.example/?token=a%2Bb' | 'https://proxy.example/api/v2/feature-flagging/config/rules-based/server?token=a%2Bb'", + "host query | 'https://proxy.example?token=a' | 'https://proxy.example/api/v2/feature-flagging/config/rules-based/server?token=a' ", + "custom path | 'https://proxy.example/ufc.json' | 'https://proxy.example/ufc.json' ", + "trimmed | ' https://proxy.example/ufc ' | 'https://proxy.example/ufc' " + }) + void resolvesConfiguredAgentlessBaseUrl(final String baseUrl, final String endpoint) { + assertEquals( + endpoint, + CdnConfigurationSource.endpoint( + TestSettings.of( + "feature.flags.configuration.source.agentless.base.url", + baseUrl, + "env", + "prod")) + .toString()); + } + + @TableTest({ + "scenario | baseUrl ", + "no scheme | 'localhost' ", + "relative path | 'relative/ufc' ", + "invalid | 'http://[::1' ", + "ftp scheme | 'ftp://example.com/ufc'" + }) + void rejectsInvalidConfiguredAgentlessBaseUrl(final String baseUrl) { + final Settings settings = + TestSettings.of("feature.flags.configuration.source.agentless.base.url", baseUrl); + + assertThrows(IllegalArgumentException.class, () -> CdnConfigurationSource.endpoint(settings)); + } + + @Test + void customEndpointDoesNotSendApiKey() throws Exception { + this.server.enqueue(200, CONFIG); + final CdnConfigurationSource source = + new CdnConfigurationSource( + TestSettings.of( + "feature.flags.configuration.source.agentless.base.url", + this.server.uri("/").toString(), + "api-key", + "secret"), + new RuntimeServices(HealthMetrics.NOOP)); + try { + source.start(this.dispatched::add); + } finally { + source.close(); + } + + assertEquals(1, this.dispatched.size()); + assertNull(this.server.requests().get(0).header("DD-API-KEY")); + } + + @Test + void sendsApiKeyAndMetadataHeaders() throws Exception { + this.server.enqueue(200, CONFIG); + + assertTrue(source("secret").pollOnce()); + + final LocalHttpServer.Request request = this.server.requests().get(0); + assertEquals("GET", request.method); + assertEquals("secret", request.header("DD-API-KEY")); + assertEquals("gzip", request.header("Accept-Encoding")); + assertEquals("java", request.header("Datadog-Meta-Lang")); + assertNull(request.header("If-None-Match")); + } + + @Test + void appliesJsonApiConfiguration() { + this.server.enqueue(200, CONFIG, "ETag", "\"v1\""); + + assertTrue(source().pollOnce()); + + assertEquals(1, this.dispatched.size()); + assertEquals("Test", this.dispatched.get(0).environment.name); + } + + @Test + void decodesGzipResponses() { + this.server.enqueueGzip(200, CONFIG); + + assertTrue(source().pollOnce()); + + assertEquals(1, this.dispatched.size()); + } + + @Test + void keepsLastKnownGoodOnRawUfcAndMalformedPayloads() { + this.server.enqueue(200, CONFIG).enqueue(200, RAW_UFC).enqueue(200, "{not json"); + final CdnConfigurationSource source = source(); + + assertTrue(source.pollOnce()); + assertFalse(source.pollOnce()); + assertFalse(source.pollOnce()); + + assertEquals(1, this.dispatched.size()); + } + + @Test + void usesEtagAndSkipsDispatchOnUnchangedConfig() { + this.server.enqueue(200, CONFIG, "ETag", "\"v1\"").enqueue(304, null); + final CdnConfigurationSource source = source(); + + assertTrue(source.pollOnce()); + assertTrue(source.pollOnce()); + + assertEquals(1, this.dispatched.size()); + assertEquals("\"v1\"", this.server.requests().get(1).header("If-None-Match")); + } + + @Test + void ignoresBlankEtagAndClearsPreviousEtag() { + this.server + .enqueue(200, CONFIG, "ETag", "\"v1\"") + .enqueue(200, CONFIG) + .enqueue(200, CONFIG, "ETag", " ") + .enqueue(200, CONFIG); + final CdnConfigurationSource source = source(); + + for (int i = 0; i < 4; i++) { + assertTrue(source.pollOnce()); + } + + assertEquals("\"v1\"", this.server.requests().get(1).header("If-None-Match")); + assertNull(this.server.requests().get(2).header("If-None-Match")); + assertNull(this.server.requests().get(3).header("If-None-Match")); + } + + @Test + void coldNotModifiedDoesNotEstablishEtag() { + this.server.enqueue(304, null, "ETag", "\"v1\"").enqueue(200, CONFIG); + final CdnConfigurationSource source = source(); + + assertTrue(source.pollOnce()); + assertTrue(source.pollOnce()); + + assertNull(this.server.requests().get(1).header("If-None-Match")); + assertEquals(1, this.dispatched.size()); + } + + @Test + void failedDispatchDoesNotAdvanceEtag() { + this.server.enqueue(200, CONFIG, "ETag", "\"v1\"").enqueue(200, CONFIG, "ETag", "\"v2\""); + final CdnConfigurationSource source = source(); + final RuntimeException failure = new IllegalStateException("listener failure"); + source.setListenerForTest( + configuration -> { + throw failure; + }); + + assertThrows(IllegalStateException.class, source::pollOnce); + source.setListenerForTest(this.dispatched::add); + assertTrue(source.pollOnce()); + + assertNull(this.server.requests().get(1).header("If-None-Match")); + } + + @TableTest({ + "scenario | status", + "unauthorized | 401 ", + "forbidden | 403 ", + "not found | 404 ", + "no content | 204 " + }) + void keepsLastKnownGoodOnNonOkResponses(final int status) { + this.server.enqueue(200, CONFIG).enqueue(status, null); + final CdnConfigurationSource source = source(); + + assertTrue(source.pollOnce()); + assertFalse(source.pollOnce()); + + assertEquals(1, this.dispatched.size()); + assertEquals(2, this.server.requests().size()); + } + + @TableTest({ + "scenario | status", + "client timeout | 408 ", + "too many requests | 429 ", + "server error | 503 " + }) + void retriesRetryableStatusBeforeApplyingConfig(final int status) { + this.server.enqueue(status, null).enqueue(status, null).enqueue(200, CONFIG); + + assertTrue(source().pollOnce()); + + assertEquals(3, this.server.requests().size()); + assertEquals(1, this.dispatched.size()); + assertEquals(2, this.sleeps.size()); + } + + @Test + void stopsAfterThreeAttemptsOnRetryableStatus() { + this.server.otherwise(500, null); + + assertFalse(source().pollOnce()); + + assertEquals(CdnConfigurationSource.MAX_ATTEMPTS, this.server.requests().size()); + assertTrue(this.dispatched.isEmpty()); + } + + @Test + void retriesServerErrorThenKeepsColdStateOnNotModified() { + this.server.enqueue(500, null).enqueue(304, null); + + assertTrue(source().pollOnce()); + + assertEquals(2, this.server.requests().size()); + assertTrue(this.dispatched.isEmpty()); + } + + @Test + void retriesIoFailuresUntilAttemptsAreExhausted() throws Exception { + final URI unreachable; + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + unreachable = URI.create("http://127.0.0.1:" + socket.getLocalPort() + "/ufc"); + } + final CdnConfigurationSource source = + source(unreachable, null, mock(ScheduledExecutorService.class)); + + assertFalse(source.pollOnce()); + + assertEquals(CdnConfigurationSource.MAX_ATTEMPTS - 1, this.sleeps.size()); + } + + @Test + void usesIntervalAwareRetryBackoff() { + this.server.enqueue(503, null).enqueue(503, null).enqueue(200, CONFIG); + + assertTrue(source().pollOnce()); + + assertEquals( + List.of( + CdnConfigurationSource.retryDelayMillis(POLL_INTERVAL_MILLIS, 1, 1.0), + CdnConfigurationSource.retryDelayMillis(POLL_INTERVAL_MILLIS, 2, 1.0)), + this.sleeps); + } + + @TableTest({ + "scenario | pollIntervalMillis | attempt | jitter | delay", + "first clamped low | 1000 | 1 | 1.0 | 2000 ", + "first proportional | 30000 | 1 | 1.0 | 5000 ", + "first clamped high | 600000 | 1 | 1.0 | 10000", + "second clamped low | 1000 | 2 | 1.0 | 5000 ", + "second proportional | 30000 | 2 | 1.0 | 10000", + "second clamped high | 600000 | 2 | 1.0 | 30000", + "jitter down | 30000 | 1 | 0.8 | 4000 ", + "jitter up | 30000 | 2 | 1.2 | 12000" + }) + void clampsAndJittersRetryBackoff( + final long pollIntervalMillis, final int attempt, final double jitter, final long delay) { + assertEquals( + delay, CdnConfigurationSource.retryDelayMillis(pollIntervalMillis, attempt, jitter)); + } + + @Test + void rejectsUnsupportedRetryAttempt() { + assertThrows( + IllegalArgumentException.class, + () -> CdnConfigurationSource.retryDelayMillis(POLL_INTERVAL_MILLIS, 3, 1.0)); + } + + @Test + void startCompletesFirstPollAndCloseCancelsScheduledPoll() { + this.server.enqueue(200, CONFIG); + final ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + final ScheduledFuture future = mock(ScheduledFuture.class); + doReturn(future) + .when(executor) + .scheduleWithFixedDelay( + any(Runnable.class), + eq(POLL_INTERVAL_MILLIS), + eq(POLL_INTERVAL_MILLIS), + eq(MILLISECONDS)); + final CdnConfigurationSource source = source(this.server.uri("/ufc"), null, executor); + + source.start(this.dispatched::add); + source.start(this.dispatched::add); + + assertEquals(1, this.dispatched.size()); + assertEquals(1, this.server.requests().size()); + verify(executor, times(1)) + .scheduleWithFixedDelay( + any(Runnable.class), + eq(POLL_INTERVAL_MILLIS), + eq(POLL_INTERVAL_MILLIS), + eq(MILLISECONDS)); + + source.close(); + verify(future).cancel(true); + verify(executor).shutdownNow(); + assertFalse(source.pollOnce()); + } + + @Test + void startSurvivesListenerFailure() { + this.server.enqueue(200, CONFIG); + final ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + final CdnConfigurationSource source = source(this.server.uri("/ufc"), null, executor); + + source.start( + configuration -> { + throw new IllegalStateException("listener failure"); + }); + + verify(executor) + .scheduleWithFixedDelay(any(Runnable.class), any(Long.class), any(Long.class), any()); + } + + @Test + void startAfterCloseDoesNotPoll() { + final ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + final CdnConfigurationSource source = source(this.server.uri("/ufc"), null, executor); + + source.close(); + source.start(this.dispatched::add); + + assertTrue(this.server.requests().isEmpty()); + verify(executor, never()) + .scheduleWithFixedDelay(any(Runnable.class), any(Long.class), any(Long.class), any()); + } + + @Test + void closeWakesRetryBackoffWithoutInterruptingCaller() throws Exception { + this.server.otherwise(503, null); + final CountDownLatch backoff = new CountDownLatch(1); + final CdnConfigurationSource source = + new CdnConfigurationSource( + this.server.uri("/ufc"), + null, + TimeUnit.MINUTES.toMillis(10), + Duration.ofSeconds(5), + HttpClient.newHttpClient(), + mock(ScheduledExecutorService.class), + null, + () -> { + backoff.countDown(); + return 1.0; + }); + final boolean[] result = {true, true}; + final Thread poller = + new Thread( + () -> { + result[0] = source.pollOnce(); + result[1] = Thread.currentThread().isInterrupted(); + }); + poller.setDaemon(true); + poller.start(); + try { + assertTrue(backoff.await(5, SECONDS)); + source.close(); + poller.join(SECONDS.toMillis(5)); + + assertFalse(poller.isAlive(), "close must wake the retry backoff"); + assertFalse(result[0]); + assertFalse(result[1], "close must not interrupt the caller"); + assertEquals(1, this.server.requests().size()); + } finally { + source.close(); + poller.interrupt(); + poller.join(SECONDS.toMillis(5)); + } + } + + @Test + @Timeout(15) + void stalledBodyDoesNotBlockPollingIndefinitely() throws Exception { + try (StallingHttpServer stalled = new StallingHttpServer(true)) { + final CdnConfigurationSource source = + new CdnConfigurationSource( + stalled.uri(), + null, + POLL_INTERVAL_MILLIS, + Duration.ofMillis(500), + HttpClient.newHttpClient(), + mock(ScheduledExecutorService.class), + this.sleeps::add, + () -> 1.0); + final FutureTask polling = new FutureTask<>(source::pollOnce); + final Thread caller = new Thread(polling, "configuration-poll"); + caller.setDaemon(true); + caller.start(); + try { + stalled.awaitRequest(); + + assertFalse(polling.get(5, SECONDS)); + assertTrue(stalled.awaitDisconnect()); + assertEquals(CdnConfigurationSource.MAX_ATTEMPTS - 1, this.sleeps.size()); + } finally { + source.close(); + polling.cancel(true); + caller.join(SECONDS.toMillis(5)); + } + } + } + + @Test + @Timeout(15) + void closeCancelsInitialPollWithoutInterruptingCaller() throws Exception { + try (StallingHttpServer stalled = new StallingHttpServer(true)) { + final CdnConfigurationSource source = + new CdnConfigurationSource( + stalled.uri(), + null, + POLL_INTERVAL_MILLIS, + Duration.ofMinutes(1), + HttpClient.newHttpClient(), + mock(ScheduledExecutorService.class), + null, + () -> 1.0); + final FutureTask initialization = + new FutureTask<>( + () -> { + source.start(this.dispatched::add); + return Thread.currentThread().isInterrupted(); + }); + final Thread caller = new Thread(initialization, "provider-initialization"); + caller.setDaemon(true); + caller.start(); + try { + stalled.awaitRequest(); + source.close(); + + assertFalse(initialization.get(5, SECONDS), "close must not interrupt initialization"); + assertTrue(stalled.awaitDisconnect()); + assertTrue(this.dispatched.isEmpty()); + assertFalse(source.pollOnce()); + } finally { + source.close(); + initialization.cancel(true); + caller.join(SECONDS.toMillis(5)); + } + } + } + + private CdnConfigurationSource source() { + return source(null); + } + + private CdnConfigurationSource source(final String apiKey) { + final CdnConfigurationSource source = + source(this.server.uri("/ufc"), apiKey, mock(ScheduledExecutorService.class)); + source.setListenerForTest(this.dispatched::add); + return source; + } + + private CdnConfigurationSource source( + final URI endpoint, final String apiKey, final ScheduledExecutorService executor) { + final CdnConfigurationSource source = + new CdnConfigurationSource( + endpoint, + apiKey, + POLL_INTERVAL_MILLIS, + Duration.ofSeconds(5), + HttpClient.newHttpClient(), + executor, + this.sleeps::add, + () -> 1.0); + source.setListenerForTest(this.dispatched::add); + return source; + } +} 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 new file mode 100644 index 00000000000..8ff26333df8 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java @@ -0,0 +1,220 @@ +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 static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; + +import com.datadog.openfeature.internal.LocalHttpServer; +import com.datadog.openfeature.internal.config.TestSettings; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.ConnectException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.net.UnknownHostException; +import java.net.http.HttpClient; +import java.net.http.HttpTimeoutException; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import javax.net.ssl.SSLException; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.tabletest.junit.TableTest; +import org.tabletest.junit.TypeConverter; + +public class DirectIntakeTransportTest { + private LocalHttpServer server; + private final List sleeps = new CopyOnWriteArrayList<>(); + + @BeforeEach + void setUp() throws IOException { + this.server = new LocalHttpServer(); + } + + @AfterEach + void tearDown() { + this.server.close(); + } + + @Test + void requiresAnApiKey() { + assertNull(DirectIntakeTransport.create(TestSettings.of())); + assertNotNull(DirectIntakeTransport.create(TestSettings.of("api-key", "secret"))); + } + + @Test + void rejectsInvalidSite() { + assertThrows( + IllegalArgumentException.class, + () -> + DirectIntakeTransport.create( + TestSettings.of("api-key", "secret", "site", "datadoghq.com@other.example"))); + } + + @Test + void postsJsonWithApiKeyAndTraceHeaders() throws Exception { + this.server.enqueue(202, null); + + transport().post("exposures", "{\"a\":1}".getBytes(UTF_8)); + + final LocalHttpServer.Request request = this.server.requests().get(0); + assertEquals("POST", request.method); + assertEquals("/api/v2/exposures", request.uri.getPath()); + assertEquals("{\"a\":1}", request.bodyAsString()); + assertEquals("application/json", request.header("Content-Type")); + assertEquals("secret", request.header("dd-api-key")); + assertNotNull(request.header("x-datadog-trace-id")); + 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")); + } + + @TableTest({ + "scenario | status", + "server error | 500 ", + "unavailable | 503 ", + "too many requests | 429 " + }) + void retriesRetryableStatus(final int status) throws Exception { + this.server.enqueue(status, null).enqueue(status, null).enqueue(200, null); + + transport().post("flagevaluation", new byte[] {'{', '}'}); + + assertEquals(3, this.server.requests().size()); + assertEquals(List.of(100L, 200L), this.sleeps); + } + + @TableTest({ + "scenario | status", + "bad request | 400 ", + "forbidden | 403 ", + "not found | 404 ", + "timeout | 408 ", + "redirect | 302 " + }) + void doesNotRetryOtherFailures(final int status) { + this.server.otherwise(status, null); + + assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); + + assertEquals(1, this.server.requests().size()); + } + + @Test + void givesUpAfterMaxRetries() { + this.server.otherwise(503, null); + + assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); + + assertEquals(DirectIntakeTransport.MAX_RETRIES + 1, this.server.requests().size()); + assertEquals(DirectIntakeTransport.MAX_RETRIES, this.sleeps.size()); + } + + @TableTest({ + "scenario | reset | requests | minDelay | maxDelay", + "immediate | 0 | 2 | 0 | 400 ", + "one second | 1 | 2 | 1000 | 1400 ", + "maximum wait | 10 | 2 | 10000 | 10400 ", + "too long | 11 | 1 | 0 | 0 ", + "overflow | 9223372036854775807 | 1 | 0 | 0 ", + "invalid | invalid | 3 | 100 | 100 ", + "negative | -2 | 3 | 100 | 100 ", + "empty | '' | 3 | 100 | 100 " + }) + void honorsRateLimitReset( + final String reset, final int requests, final long minDelay, final long maxDelay) + throws Exception { + this.server + .enqueue(429, null, "X-RateLimit-Reset", reset) + .enqueue(503, null) + .enqueue(200, null); + + if (requests == 3) { + transport().post("exposures", new byte[] {'{', '}'}); + } else { + assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); + } + + assertEquals(requests, this.server.requests().size()); + assertEquals(requests - 1, this.sleeps.size()); + if (!this.sleeps.isEmpty()) { + assertTrue(this.sleeps.get(0) >= minDelay && this.sleeps.get(0) <= maxDelay); + } + } + + @TableTest({ + "scenario | failure | attempts", + "connection | connection | 6 ", + "timeout | timeout | 6 ", + "TLS failure | tls | 1 ", + "DNS failure | dns | 1 ", + "interrupted IO | interrupted | 1 ", + "other IO | other | 1 " + }) + void retriesOnlyConnectionAndTimeoutFailures(final IOException failure, final int attempts) + throws Exception { + final HttpClient client = mock(HttpClient.class); + doReturn(CompletableFuture.failedFuture(failure)).when(client).sendAsync(any(), any()); + doThrow(failure).when(client).send(any(), any()); + final DirectIntakeTransport transport = + new DirectIntakeTransport(this.server.uri("/api/v2/"), "secret", client, this.sleeps::add); + + assertThrows(failure.getClass(), () -> transport.post("exposures", new byte[] {'{', '}'})); + + assertEquals(attempts - 1, this.sleeps.size()); + } + + /** + * Creates an I/O failure for a retry scenario. + * + * @param value the failure kind. + * @return the failure. + */ + @TypeConverter + public static IOException failure(final String value) { + switch (value) { + case "connection": + return new ConnectException("connection refused"); + case "timeout": + return new HttpTimeoutException("request timed out"); + case "tls": + return new SSLException("TLS handshake failed"); + case "dns": + return new UnknownHostException("unknown host"); + case "interrupted": + return new InterruptedIOException("interrupted"); + default: + return new IOException("I/O failure"); + } + } + + @Test + void retriesConnectionFailures() throws Exception { + final URI unreachable; + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + unreachable = URI.create("http://127.0.0.1:" + socket.getLocalPort() + "/api/v2/"); + } + final DirectIntakeTransport transport = + new DirectIntakeTransport( + unreachable, "secret", HttpClient.newHttpClient(), this.sleeps::add); + + assertThrows(IOException.class, () -> transport.post("exposures", new byte[] {'{', '}'})); + + assertEquals(DirectIntakeTransport.MAX_RETRIES, this.sleeps.size()); + } + + private DirectIntakeTransport transport() { + return new DirectIntakeTransport( + this.server.uri("/api/v2/"), "secret", HttpClient.newHttpClient(), this.sleeps::add); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java new file mode 100644 index 00000000000..ce05e06d685 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/FallbackTransportTest.java @@ -0,0 +1,52 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.doThrow; +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 com.datadog.openfeature.internal.bridge.EventProxyUnavailableException; +import com.datadog.openfeature.internal.bridge.EventTransport; +import java.io.IOException; +import org.junit.jupiter.api.Test; + +class FallbackTransportTest { + private static final byte[] PAYLOAD = "{}".getBytes(UTF_8); + + private final EventTransport proxy = mock(EventTransport.class); + private final EventTransport direct = mock(EventTransport.class); + private final FallbackTransport transport = new FallbackTransport(this.proxy, this.direct); + + @Test + void postsThroughTheProxy() throws IOException { + this.transport.post("exposures", PAYLOAD); + + verify(this.proxy).post("exposures", PAYLOAD); + verify(this.direct, never()).post("exposures", PAYLOAD); + } + + @Test + void switchesToDirectIntakeForGoodWhenTheProxyIsUnavailable() throws IOException { + doThrow(new EventProxyUnavailableException("unavailable")) + .when(this.proxy) + .post("exposures", PAYLOAD); + + this.transport.post("exposures", PAYLOAD); + this.transport.post("exposures", PAYLOAD); + + verify(this.proxy, times(1)).post("exposures", PAYLOAD); + verify(this.direct, times(2)).post("exposures", PAYLOAD); + } + + @Test + void doesNotReplayAmbiguousProxyFailures() throws IOException { + doThrow(new IOException("timeout")).when(this.proxy).post("exposures", PAYLOAD); + + assertThrows(IOException.class, () -> this.transport.post("exposures", PAYLOAD)); + + verify(this.direct, never()).post("exposures", PAYLOAD); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/HttpCallTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/HttpCallTest.java new file mode 100644 index 00000000000..c8452bb8ae4 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/HttpCallTest.java @@ -0,0 +1,206 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +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.doReturn; +import static org.mockito.Mockito.mock; + +import com.datadog.openfeature.internal.LocalHttpServer; +import java.io.IOException; +import java.net.ConnectException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.net.http.HttpTimeoutException; +import java.time.Duration; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CountDownLatch; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.tabletest.junit.TableTest; + +@Timeout(15) +class HttpCallTest { + @TableTest({ + "scenario | discard", + "CDN | false ", + "intake | true " + }) + void timesOutStalledBodyAndClosesConnection(final boolean discard) throws Exception { + try (StallingHttpServer server = new StallingHttpServer(true)) { + final CountDownLatch handlerCalled = new CountDownLatch(1); + final HttpResponse.BodyHandler handler = bodyHandler(discard); + final HttpCall call = + new HttpCall<>( + HttpClient.newHttpClient(), + request(server, Duration.ofSeconds(1)), + info -> { + handlerCalled.countDown(); + return handler.apply(info); + }); + + assertTrue(handlerCalled.await(5, SECONDS)); + assertThrows(HttpTimeoutException.class, call::execute); + assertTrue(server.awaitDisconnect()); + } + } + + @Test + void timesOutStalledHeaders() throws Exception { + try (StallingHttpServer server = new StallingHttpServer(false)) { + final HttpCall call = + new HttpCall<>( + HttpClient.newHttpClient(), + request(server, Duration.ofMillis(500)), + HttpResponse.BodyHandlers.ofByteArray()); + + server.awaitRequest(); + assertThrows(HttpTimeoutException.class, call::execute); + server.sendHeaders(); + assertTrue(server.awaitDisconnect()); + } + } + + @Test + void cancellationBeforeHeadersCancelsLateSubscription() throws Exception { + try (StallingHttpServer server = new StallingHttpServer(false)) { + final HttpCall call = + new HttpCall<>( + HttpClient.newHttpClient(), + request(server, Duration.ofSeconds(10)), + HttpResponse.BodyHandlers.ofByteArray()); + + server.awaitRequest(); + call.cancel(); + assertThrows(IOException.class, call::execute); + server.sendHeaders(); + assertTrue(server.awaitDisconnect()); + } + } + + @Test + void cancellationDuringHandlerCreationCancelsLateSubscription() throws Exception { + final CountDownLatch handlerCalled = new CountDownLatch(1); + final CountDownLatch createSubscriber = new CountDownLatch(1); + try (StallingHttpServer server = new StallingHttpServer(true)) { + final HttpCall call = + new HttpCall<>( + HttpClient.newHttpClient(), + request(server, Duration.ofSeconds(10)), + info -> { + handlerCalled.countDown(); + try { + if (!createSubscriber.await(5, SECONDS)) { + throw new IllegalStateException("Subscriber creation was not released"); + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + return HttpResponse.BodySubscribers.ofByteArray(); + }); + + assertTrue(handlerCalled.await(5, SECONDS)); + call.cancel(); + createSubscriber.countDown(); + assertThrows(IOException.class, call::execute); + assertTrue(server.awaitDisconnect()); + } finally { + createSubscriber.countDown(); + } + } + + @Test + void interruptionClosesConnection() throws Exception { + try (StallingHttpServer server = new StallingHttpServer(true)) { + final HttpCall call = + new HttpCall<>( + HttpClient.newHttpClient(), + request(server, Duration.ofSeconds(10)), + HttpResponse.BodyHandlers.ofByteArray()); + + server.awaitRequest(); + Thread.currentThread().interrupt(); + try { + assertThrows(InterruptedException.class, call::execute); + } finally { + Thread.interrupted(); + } + assertTrue(server.awaitDisconnect()); + } + } + + @Test + void preservesIoExceptionType() { + final HttpClient client = mock(HttpClient.class); + final ConnectException failure = new ConnectException("connection refused"); + doReturn(CompletableFuture.failedFuture(failure)).when(client).sendAsync(any(), any()); + final HttpCall call = + new HttpCall<>( + client, + HttpRequest.newBuilder(URI.create("http://localhost/")) + .timeout(Duration.ofSeconds(1)) + .build(), + HttpResponse.BodyHandlers.ofByteArray()); + + assertSame(failure, assertThrows(ConnectException.class, call::execute)); + } + + @Test + void normalizesWrappedCancellation() { + final HttpClient client = mock(HttpClient.class); + final CancellationException cancelled = new CancellationException("Request cancelled"); + doReturn(CompletableFuture.failedFuture(new CompletionException(cancelled))) + .when(client) + .sendAsync(any(), any()); + final HttpCall call = + new HttpCall<>( + client, + HttpRequest.newBuilder(URI.create("http://localhost/")) + .timeout(Duration.ofSeconds(1)) + .build(), + HttpResponse.BodyHandlers.ofByteArray()); + + assertSame(cancelled, assertThrows(IOException.class, call::execute).getCause()); + } + + @Test + void successfulResponsesRemainUsableAfterCancel() throws Exception { + try (LocalHttpServer server = new LocalHttpServer()) { + server.enqueue(200, "first").enqueue(200, "second"); + final HttpClient client = HttpClient.newHttpClient(); + final HttpRequest request = + HttpRequest.newBuilder(server.uri("/")).timeout(Duration.ofSeconds(5)).build(); + final HttpCall first = + new HttpCall<>(client, request, HttpResponse.BodyHandlers.ofByteArray()); + + final byte[] body = first.execute().body(); + first.cancel(); + + assertArrayEquals("first".getBytes(UTF_8), body); + assertArrayEquals( + "second".getBytes(UTF_8), + new HttpCall<>(client, request, HttpResponse.BodyHandlers.ofByteArray()) + .execute() + .body()); + } + } + + private static HttpRequest request(final StallingHttpServer server, final Duration timeout) { + return HttpRequest.newBuilder(server.uri()).timeout(timeout).build(); + } + + private static HttpResponse.BodyHandler bodyHandler(final boolean discard) { + return discard + ? HttpResponse.BodyHandlers.discarding() + : HttpResponse.BodyHandlers.ofByteArray(); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/SiteHostTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/SiteHostTest.java new file mode 100644 index 00000000000..b47e3558004 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/SiteHostTest.java @@ -0,0 +1,31 @@ +package com.datadog.openfeature.internal.delivery; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import org.tabletest.junit.TableTest; + +class SiteHostTest { + @TableTest({ + "scenario | site | host ", + "default | 'datadoghq.com' | 'intake.datadoghq.com' ", + "eu | 'datadoghq.eu' | 'intake.datadoghq.eu' ", + "subdomain | 'us3.datadoghq.com' | 'intake.us3.datadoghq.com'" + }) + void buildsHostFromSite(final String site, final String host) { + assertEquals(host, SiteHost.of("intake", site)); + } + + @TableTest({ + "scenario | site ", + "user info | 'datadoghq.com@other.example'", + "path | 'datadoghq.com/other' ", + "fragment | 'datadoghq.com#' ", + "port | 'datadoghq.com:8443' ", + "query | 'datadoghq.com?other' ", + "space | 'datadoghq com' " + }) + void rejectsSiteThatIsNotAHostName(final String site) { + assertThrows(IllegalArgumentException.class, () -> SiteHost.of("intake", site)); + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/StallingHttpServer.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/StallingHttpServer.java new file mode 100644 index 00000000000..5ea6d86df8b --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/StallingHttpServer.java @@ -0,0 +1,94 @@ +package com.datadog.openfeature.internal.delivery; + +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.util.concurrent.TimeUnit.SECONDS; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketException; +import java.net.URI; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; + +/** A raw HTTP endpoint that stalls before headers or after the first response body byte. */ +final class StallingHttpServer implements AutoCloseable { + private final ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + private final CompletableFuture received = new CompletableFuture<>(); + private final CompletableFuture disconnected = new CompletableFuture<>(); + private final CountDownLatch headersAllowed; + private final Thread thread; + private final URI uri; + private volatile Socket connection; + + StallingHttpServer(final boolean sendHeaders) throws Exception { + this.uri = + new URI( + "http", + null, + this.server.getInetAddress().getHostAddress(), + this.server.getLocalPort(), + "/", + null, + null); + this.headersAllowed = new CountDownLatch(sendHeaders ? 0 : 1); + this.thread = new Thread(this::serve, "stalling-http-server"); + this.thread.setDaemon(true); + this.thread.start(); + } + + URI uri() { + return this.uri; + } + + void awaitRequest() throws Exception { + this.received.get(5, SECONDS); + } + + void sendHeaders() { + this.headersAllowed.countDown(); + } + + boolean awaitDisconnect() throws Exception { + return this.disconnected.get(5, SECONDS); + } + + private void serve() { + try (Socket socket = this.server.accept()) { + this.connection = socket; + socket.setSoTimeout(10_000); + final BufferedReader reader = + new BufferedReader(new InputStreamReader(socket.getInputStream(), US_ASCII)); + String line; + while ((line = reader.readLine()) != null && !line.isEmpty()) {} + this.received.complete(null); + if (!this.headersAllowed.await(5, SECONDS)) { + throw new IllegalStateException("Response headers were not released"); + } + socket + .getOutputStream() + .write("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nx".getBytes(US_ASCII)); + socket.getOutputStream().flush(); + this.disconnected.complete(reader.read() == -1); + } catch (final SocketException e) { + this.received.completeExceptionally(e); + this.disconnected.complete(true); + } catch (final Exception e) { + this.received.completeExceptionally(e); + this.disconnected.completeExceptionally(e); + } + } + + @Override + public void close() throws Exception { + this.server.close(); + final Socket socket = this.connection; + if (socket != null) { + socket.close(); + } + this.thread.interrupt(); + this.thread.join(SECONDS.toMillis(5)); + } +}