diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java b/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java new file mode 100644 index 00000000000..0916ea3d5f2 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/EvaluationTelemetryOptOutHook.java @@ -0,0 +1,55 @@ +package com.datadog.openfeature; + +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.FlagEvaluationDetails; +import dev.openfeature.sdk.Hook; +import dev.openfeature.sdk.HookContext; +import dev.openfeature.sdk.Value; +import java.util.Map; +import java.util.Optional; + +/** Gates a Datadog finally-only evaluation telemetry hook, not exposure or application hooks. */ +final class EvaluationTelemetryOptOutHook implements Hook { + private static final String PARAMETER = "dd_evaluation_telemetry_opt_out"; + private static final String DECISION = "datadog.evaluationTelemetryOptOut"; + private final Hook delegate; + + EvaluationTelemetryOptOutHook(final Hook delegate) { + this.delegate = delegate; + } + + @Override + public Optional before( + final HookContext context, final Map hints) { + // Provider before hooks see the merged context, including application before hooks. + // HookData belongs to this evaluation, including nested and concurrent calls. + context.getHookData().set(DECISION, isOptedOut(context.getCtx())); + return Optional.empty(); + } + + @Override + public void error( + final HookContext context, final Exception error, final Map hints) { + // A failing application before hook can prevent our before hook from running. + // Preserve its decision before application error hooks can change the context. + if (context.getHookData().get(DECISION) == null) { + context.getHookData().set(DECISION, isOptedOut(context.getCtx())); + } + } + + @Override + public void finallyAfter( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if (Boolean.TRUE.equals(context.getHookData().get(DECISION)) || isOptedOut(context.getCtx())) { + return; + } + delegate.finallyAfter(context, details, hints); + } + + private static boolean isOptedOut(final EvaluationContext context) { + final Value value = context == null ? null : context.getValue(PARAMETER); + return value != null && value.isBoolean() && Boolean.TRUE.equals(value.asBoolean()); + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java index cc421c310c4..b8b72493c9d 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/Provider.java @@ -3,6 +3,7 @@ import static java.util.concurrent.TimeUnit.SECONDS; import com.datadog.openfeature.internal.FeatureFlagsRuntime; +import com.datadog.openfeature.internal.FeatureFlagsRuntime.RemoteConfigurationUnavailableException; import com.datadog.openfeature.internal.bridge.AgentBridge; import com.datadog.openfeature.internal.bridge.AgentBridges; import com.datadog.openfeature.internal.config.Settings; @@ -111,7 +112,13 @@ public Provider(final Options options) { if (spanEnrichmentHook != null) { hooks.add(spanEnrichmentHook); } - // Exposure hook: always registered last; no-op when the evaluator is not a Datadog one or is + if (options.isEvaluationTelemetryOptOutParameterEnabled(this.settings)) { + for (int i = 0; i < hooks.size(); i++) { + hooks.set(i, new EvaluationTelemetryOptOutHook(hooks.get(i))); + } + } + // Exposure hook: always registered last, and independent of evaluation telemetry opt-out; no-op + // when the evaluator is not a Datadog one or is // not initialized. hooks.add(new ExposureLoggingHook<>(this::datadogEvaluator)); this.providerHooks = @@ -144,6 +151,14 @@ public void initialize(final EvaluationContext context) throws Exception { throw new ProviderNotReadyError( "Provider timed-out while waiting for initial configuration"); } + } catch (final RemoteConfigurationUnavailableException e) { + // No RC connector means no configuration can arrive in this JVM. Complete registration + // without waiting or switching sources. The evaluator has no runtime/configuration and + // returns caller defaults with PROVIDER_NOT_READY, never successful flag assignments. + // OpenFeature marks the completed registration READY; that does not imply RC is available. + initializationState.set(InitializationState.FALLBACK); + log.warn("{}; returning caller defaults", e.getMessage()); + log.debug("Feature Flags Remote Configuration connector unavailable", e); } catch (final OpenFeatureError e) { markInitializationError(); throw e; @@ -310,14 +325,45 @@ private enum InitializationState { INITIALIZING, INITIAL_CONFIG_RECEIVED, READY, - ERROR + ERROR, + FALLBACK } /** The provider options. */ public static class Options { - private long timeout; - private TimeUnit unit; + private long timeout = 30; + private TimeUnit unit = SECONDS; + private Boolean evaluationTelemetryOptOutParameterEnabled; + + /** + * Enables per-evaluation telemetry opt-out through the {@code dd_evaluation_telemetry_opt_out} + * context attribute. Overrides {@code + * DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED}, including when false. + * Without an explicit option, the provider reads configuration at construction. Disabled by + * default. + * + *

Only boolean {@code true} suppresses new Datadog evaluation events, evaluation metrics, + * and automatic flag span enrichment. Missing or other values preserve telemetry. This does not + * change flag results, exposure logging, application hooks, or previously recorded data. + * + *

The attribute remains available for targeting and exposure reporting. Existing privacy and + * reporting settings apply. + * + * @param enabled whether to honor the attribute; configure before constructing the provider. + * @return these options. + */ + public Options enableEvaluationTelemetryOptOutParameter(final boolean enabled) { + this.evaluationTelemetryOptOutParameterEnabled = enabled; + return this; + } + + private boolean isEvaluationTelemetryOptOutParameterEnabled(final Settings settings) { + if (this.evaluationTelemetryOptOutParameterEnabled != null) { + return this.evaluationTelemetryOptOutParameterEnabled; + } + return settings.isEvaluationTelemetryPerSubjectOptOutEnabled(); + } /** * Sets how long provider initialization waits for the initial flag configuration. diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java new file mode 100644 index 00000000000..dba55339c80 --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/EventPipeline.java @@ -0,0 +1,56 @@ +package com.datadog.openfeature.internal; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +/** + * An event pipeline stopping in two phases, so that several pipelines can share one shutdown + * budget: every pipeline is first asked to flush, then each one is awaited until the same deadline. + */ +public interface EventPipeline extends AutoCloseable { + /** The time the pipelines get to flush their pending events when stopping. */ + long SHUTDOWN_TIMEOUT_MILLIS = 800; + + /** Stops accepting events and asks the worker to flush the pending ones, without waiting. */ + void requestShutdown(); + + /** + * Waits for the worker to flush the pending events and exit. + * + * @param deadlineNanos the {@link System#nanoTime()} value to wait until. + */ + void awaitTermination(long deadlineNanos); + + /** Stops the pipeline, waiting at most {@link #SHUTDOWN_TIMEOUT_MILLIS} for the final flush. */ + @Override + default void close() { + final long deadline = shutdownDeadline(); + requestShutdown(); + awaitTermination(deadline); + } + + /** + * @return the deadline of a shutdown starting now. + */ + static long shutdownDeadline() { + return System.nanoTime() + MILLISECONDS.toNanos(SHUTDOWN_TIMEOUT_MILLIS); + } + + /** + * Waits for a worker thread to exit. + * + * @param worker the worker thread. + * @param deadlineNanos the {@link System#nanoTime()} value to wait until. + */ + static void awaitThread(final Thread worker, final long deadlineNanos) { + final long remaining = deadlineNanos - System.nanoTime(); + if (worker == Thread.currentThread() || remaining <= 0) { + return; + } + try { + NANOSECONDS.timedJoin(worker, remaining); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java index 89a9b0fd4b1..4ff8fecadac 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/FeatureFlagsRuntime.java @@ -5,6 +5,7 @@ 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.ShutdownSource; import com.datadog.openfeature.internal.config.Settings; import com.datadog.openfeature.internal.delivery.CdnConfigurationSource; import com.datadog.openfeature.internal.delivery.DirectIntakeTransport; @@ -20,6 +21,7 @@ import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -27,7 +29,9 @@ /** * The Feature Flags runtime: configuration delivery and event pipelines. A single runtime is shared - * by all the providers of a class loader, and stopped when the last one releases it. + * by all the providers of a class loader, and stopped when the last one releases it, or when the + * application shuts down: by the Datadog Java agent before the JVM shutdown hooks run when + * attached, or by a JVM shutdown hook otherwise. */ public final class FeatureFlagsRuntime { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlagsRuntime.class); @@ -42,6 +46,8 @@ public final class FeatureFlagsRuntime { @Nullable private final FlagEvaluationPipeline evaluations; private final List> listeners = new CopyOnWriteArrayList<>(); private volatile ServerConfiguration configuration; + private final AtomicBoolean stopped = new AtomicBoolean(); + @Nullable private volatile AutoCloseable shutdownRegistration; FeatureFlagsRuntime( final ConfigurationService source, @@ -66,6 +72,7 @@ public static FeatureFlagsRuntime acquire(final AgentBridge bridge, final Settin if (shared == null) { final FeatureFlagsRuntime created = create(bridge, settings); created.start(); + created.registerShutdown(bridge); shared = created; } references++; @@ -97,11 +104,14 @@ static FeatureFlagsRuntime create(final AgentBridge bridge, final Settings setti final EventTransport transport; if (remoteConfig) { // TODO Support Remote Configuration without the Datadog Java agent. - final ConfigurationSource remoteConfiguration = bridge.remoteConfiguration(); + final ConfigurationSource remoteConfiguration; + try { + remoteConfiguration = bridge.remoteConfiguration(); + } catch (final LinkageError e) { + throw new RemoteConfigurationUnavailableException(e); + } if (remoteConfiguration == null) { - throw new IllegalStateException( - "Feature Flags Remote Configuration source requires the Datadog Java agent with" - + " Remote Configuration enabled"); + throw new RemoteConfigurationUnavailableException(null); } source = new RemoteConfigurationService(remoteConfiguration); // Remote Configuration never falls back to direct intake. @@ -142,6 +152,17 @@ static FeatureFlagsRuntime create(final AgentBridge bridge, final Settings setti return new FeatureFlagsRuntime(source, exposures, evaluations); } + /** Signals an unavailable RC connector before any runtime resources have been started. */ + public static final class RemoteConfigurationUnavailableException extends IllegalStateException { + private RemoteConfigurationUnavailableException(@Nullable final LinkageError cause) { + super( + "Feature Flags Remote Configuration requires a compatible Datadog Java agent with" + + " Remote Configuration enabled; attach a compatible agent or configure" + + " DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=agentless for CDN delivery", + cause); + } + } + private static Map eventContext(final Settings settings) { final Map context = new HashMap<>(); putIfNotNull(context, "service", settings.service()); @@ -172,13 +193,67 @@ void start() { } } + /** + * Stops the runtime at application shutdown: through the agent when it supports it, otherwise + * through a JVM shutdown hook. + * + * @param bridge the bridge to the Datadog Java agent. + */ + void registerShutdown(final AgentBridge bridge) { + AutoCloseable registration = null; + try { + final ShutdownSource shutdown = bridge.shutdown(); + if (shutdown != null) { + registration = shutdown.subscribe(this::stop); + } + } catch (final RuntimeException | LinkageError e) { + LOGGER.debug("Could not subscribe to the agent shutdown, using a JVM shutdown hook", e); + } + this.shutdownRegistration = registration == null ? addShutdownHook() : registration; + } + + @Nullable + private AutoCloseable addShutdownHook() { + final Thread hook = new Thread(this::stop, "dd-openfeature-shutdown"); + try { + Runtime.getRuntime().addShutdownHook(hook); + } catch (final IllegalStateException | SecurityException e) { + LOGGER.debug("Could not register the Feature Flags shutdown hook", e); + return null; + } + return () -> { + if (Thread.currentThread() != hook) { + Runtime.getRuntime().removeShutdownHook(hook); + } + }; + } + void stop() { + if (!this.stopped.compareAndSet(false, true)) { + return; + } + final AutoCloseable registration = this.shutdownRegistration; + if (registration != null) { + try { + registration.close(); + } catch (final Exception e) { + // The JVM is already shutting down, or the agent backend is gone. + } + } + // Signal both pipelines before waiting for either, so they flush in parallel within one budget. + final long deadline = EventPipeline.shutdownDeadline(); + if (this.evaluations != null) { + this.evaluations.requestShutdown(); + } + if (this.exposures != null) { + this.exposures.requestShutdown(); + } this.source.close(); if (this.evaluations != null) { - this.evaluations.close(); + this.evaluations.awaitTermination(deadline); } if (this.exposures != null) { - this.exposures.close(); + this.exposures.awaitTermination(deadline); } } 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 index 1ff84479cec..342c2c2bc8c 100644 --- 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 @@ -38,4 +38,11 @@ public interface AgentBridge { * @return the span enricher. */ SpanEnricher spanEnricher(); + + /** + * @return the agent shutdown notifications, sent before the JVM shutdown hooks run, or {@code + * null} if not available, so the SDK must flush from its own JVM shutdown hook. + */ + @Nullable + ShutdownSource shutdown(); } 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 index dab3e42db97..0a8ff1709ae 100644 --- 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 @@ -30,4 +30,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Nullable + @Override + public ShutdownSource shutdown() { + return null; + } } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java new file mode 100644 index 00000000000..145bf413b7b --- /dev/null +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/bridge/ShutdownSource.java @@ -0,0 +1,12 @@ +package com.datadog.openfeature.internal.bridge; + +/** A source of application shutdown notifications. */ +public interface ShutdownSource { + /** + * Subscribes to the application shutdown. + * + * @param listener the listener run at shutdown, typically flushing the pending events. + * @return the handle to close to unsubscribe. + */ + AutoCloseable subscribe(Runnable listener); +} 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 index 11bf5f4e91c..1213c5d2ca4 100644 --- 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 @@ -36,6 +36,8 @@ public final class Settings { 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 EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED = + "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; static final String API_KEY = "api-key"; static final String SITE = "site"; static final String ENV = "env"; @@ -162,6 +164,16 @@ public boolean isExposuresDatadogLoggingEnabled() { return !Boolean.FALSE.equals(getBoolean(EXPOSURES_DATADOG_LOGGING_ENABLED)); } + /** + * Tells whether the {@code dd_evaluation_telemetry_opt_out} evaluation context attribute can opt + * subjects out of evaluation telemetry. The provider option takes precedence over this setting. + * + * @return whether per-subject evaluation telemetry opt-out is enabled (off by default). + */ + public boolean isEvaluationTelemetryPerSubjectOptOutEnabled() { + return Boolean.TRUE.equals(getBoolean(EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED)); + } + /** * @return the Datadog API key, or {@code null} if not set. */ diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java index 9c004fd54c6..bd2c0171a6b 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransport.java @@ -6,40 +6,33 @@ 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. */ +/** + * Posts events directly to the Datadog event platform intake, authenticated with an API key. Each + * batch is sent once, without retries or redirects, so an ambiguously accepted batch is never sent + * twice. + */ 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) { + DirectIntakeTransport(final URI intake, final String apiKey, final HttpClient httpClient) { 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; } /** @@ -58,8 +51,7 @@ public static DirectIntakeTransport create(final Settings settings) { 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); + HttpClient.newBuilder().connectTimeout(REQUEST_TIMEOUT).followRedirects(NEVER).build()); } @Override @@ -76,14 +68,8 @@ public void post(final String route, final byte[] json) throws IOException { final int status; try { status = - Retries.send( - () -> - new HttpCall<>( - this.httpClient, request, HttpResponse.BodyHandlers.discarding()), - new IntakeRetryPolicy(), - MAX_RETRIES + 1, - this.retrySleeper, - () -> false) + new HttpCall<>(this.httpClient, request, HttpResponse.BodyHandlers.discarding()) + .execute() .statusCode(); } catch (final InterruptedException e) { Thread.currentThread().interrupt(); @@ -93,44 +79,4 @@ public void post(final String route, final byte[] json) throws IOException { 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 index e01bf019672..3c4ca077f0b 100644 --- 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 @@ -7,8 +7,9 @@ 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. + * Posts through the event proxy, switching to direct intake for good on the first proxy failure. + * Only a definitively rejected batch is replayed through direct intake, an ambiguously failed batch + * is not to avoid sending it twice. */ public final class FallbackTransport implements EventTransport { private static final Logger LOGGER = LoggerFactory.getLogger(FallbackTransport.class); @@ -29,11 +30,17 @@ public void post(final String route, final byte[] json) throws IOException { 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; + switchToDirectIntake(); + } catch (final IOException e) { + switchToDirectIntake(); + throw e; } } this.direct.post(route, json); } + + private void switchToDirectIntake() { + LOGGER.debug("Switching Feature Flags event delivery from the event proxy to direct intake"); + this.proxyUnavailable = true; + } } diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java index 8397b208647..9577b1df211 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/exposure/ExposurePipeline.java @@ -1,5 +1,8 @@ package com.datadog.openfeature.internal.exposure; +import static com.datadog.openfeature.internal.EventPipeline.awaitThread; + +import com.datadog.openfeature.internal.EventPipeline; import com.datadog.openfeature.internal.EventQueue; import com.datadog.openfeature.internal.RuntimeServices; import com.datadog.openfeature.internal.bridge.EventTransport; @@ -13,14 +16,13 @@ import org.slf4j.LoggerFactory; /** Batches exposure events to the event platform {@code exposures} route. */ -public class ExposurePipeline implements Consumer, AutoCloseable { +public class ExposurePipeline implements Consumer, EventPipeline { private static final Logger LOGGER = LoggerFactory.getLogger(ExposurePipeline.class); public static final int DEFAULT_CAPACITY = 1 << 16; // 65536 elements public static final int DEFAULT_FLUSH_INTERVAL_IN_SECONDS = 1; private static final int FLUSH_THRESHOLD = 100; private static final String EXPOSURES_ROUTE = "exposures"; - private static final long SHUTDOWN_TIMEOUT_MILLIS = 800; private final EventQueue queue; private final ExposureSerializingHandler serializer; @@ -51,22 +53,17 @@ public void start() { } @Override - public void close() { - final long deadline = - System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SHUTDOWN_TIMEOUT_MILLIS); + public void requestShutdown() { synchronized (this.lifecycleLock) { if (this.closed.compareAndSet(false, true)) { this.serializer.requestShutdown(); } } - final long remaining = deadline - System.nanoTime(); - if (Thread.currentThread() != this.serializerThread && remaining > 0) { - try { - TimeUnit.NANOSECONDS.timedJoin(this.serializerThread, remaining); - } catch (final InterruptedException e) { - Thread.currentThread().interrupt(); - } - } + } + + @Override + public void awaitTermination(final long deadlineNanos) { + awaitThread(this.serializerThread, deadlineNanos); } @Override diff --git a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java index d2f3bd924fd..e3daf400dae 100644 --- a/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java +++ b/dd-openfeature/src/main/java/com/datadog/openfeature/internal/flagevaluation/FlagEvaluationPipeline.java @@ -1,5 +1,8 @@ package com.datadog.openfeature.internal.flagevaluation; +import static com.datadog.openfeature.internal.EventPipeline.awaitThread; + +import com.datadog.openfeature.internal.EventPipeline; import com.datadog.openfeature.internal.EventQueue; import com.datadog.openfeature.internal.RuntimeServices; import com.datadog.openfeature.internal.bridge.EventTransport; @@ -38,7 +41,7 @@ * the worker has been joined, counting any remainder as a closed drop so shutdown loss is * observable rather than silent. */ -public class FlagEvaluationPipeline implements FlagEvaluationWriter, AutoCloseable { +public class FlagEvaluationPipeline implements FlagEvaluationWriter, EventPipeline { private static final Logger LOGGER = LoggerFactory.getLogger(FlagEvaluationPipeline.class); @@ -122,37 +125,30 @@ int aggregatorFullTierSizeForTest() { } @Override - public void close() { - final boolean workerRunning; + public void requestShutdown() { synchronized (lifecycleLock) { - if (!closed.compareAndSet(false, true)) { - return; - } - workerRunning = this.serializerThread.isAlive(); - if (workerRunning) { - // Ask the worker to drain the queue and final-flush, then interrupt to wake it from poll(). + if (closed.compareAndSet(false, true)) { + // The worker polls every 100 ms: do not interrupt a current or final HTTP request. serializer.requestShutdown(); - this.serializerThread.interrupt(); } } + } + + @Override + public void awaitTermination(final long deadlineNanos) { if (Thread.currentThread() == this.serializerThread) { return; } - if (workerRunning) { - try { - // Bounded wait for the worker's final flush so queued events are not lost on shutdown. - this.serializerThread.join(TimeUnit.SECONDS.toMillis(5)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - } + awaitThread(this.serializerThread, deadlineNanos); // enqueue() is lock-free, so a producer that passed the closed check before close() ran can // still land an event after the worker's final drain. Sweep the remainder so that loss is // counted rather than silently stranded in a queue nobody polls again. Only safe once the // worker is gone: the queue is single-consumer, so sweeping alongside a live worker would - // break that contract. If join() timed out the worker is still draining, so skip the sweep. - if (!this.serializerThread.isAlive()) { - sweepAndCountResidualEvents(); + // break that contract. If the wait timed out the worker is still draining, so skip the sweep. + synchronized (lifecycleLock) { + if (!this.serializerThread.isAlive()) { + sweepAndCountResidualEvents(); + } } } @@ -294,12 +290,10 @@ public void run() { // On exit (interrupt or shutdown request), drain everything still buffered and flush it so // queued events are not lost on shutdown. // - // close() interrupts this thread to break it out of poll(). The final flush does socket - // I/O, which fails fast on a thread whose interrupt flag is set, so clear the flag - // for the duration of the drain and restore it afterwards. Without this the flush that - // close() exists to guarantee throws IOException, the aggregated rows are discarded by - // the clear() in flush(), and the loss is invisible to the drop counters because those - // rows already left the queue. + // Normal close only signals shutdown. If an unrelated interrupt ends the duty cycle, clear + // it during the final socket I/O and restore it afterwards: the flush otherwise fails + // fast, the aggregated rows are discarded by the clear() in flush(), and the loss is + // invisible to the drop counters because those rows already left the queue. final boolean wasInterrupted = Thread.interrupted(); try { drainAndFlush(); diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java index 846b4bc31cc..7f75c8ab611 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/DatadogExposureLoggingTest.java @@ -14,6 +14,7 @@ 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.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; import com.datadog.openfeature.internal.exposure.ExposureEvent; @@ -150,5 +151,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java new file mode 100644 index 00000000000..d5e3ab218c0 --- /dev/null +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/EvaluationTelemetryOptOutTest.java @@ -0,0 +1,766 @@ +package com.datadog.openfeature; + +import static java.util.Arrays.asList; +import static java.util.Collections.emptyList; +import static java.util.Collections.emptyMap; +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +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.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datadog.openfeature.internal.ConfigurationService; +import com.datadog.openfeature.internal.FeatureFlagsRuntime; +import com.datadog.openfeature.internal.RuntimeServices; +import com.datadog.openfeature.internal.bridge.AgentBridge; +import com.datadog.openfeature.internal.bridge.HealthMetrics; +import com.datadog.openfeature.internal.bridge.SpanEnricher; +import com.datadog.openfeature.internal.config.TestSettings; +import com.datadog.openfeature.internal.exposure.ExposureDeduplicationCache; +import com.datadog.openfeature.internal.exposure.ExposureEvent; +import com.datadog.openfeature.internal.exposure.ExposurePipeline; +import com.datadog.openfeature.internal.flagevaluation.FlagEvalEvent; +import com.datadog.openfeature.internal.flagevaluation.FlagEvaluationPipeline; +import com.datadog.openfeature.internal.ufc.Allocation; +import com.datadog.openfeature.internal.ufc.ConditionConfiguration; +import com.datadog.openfeature.internal.ufc.ConditionOperator; +import com.datadog.openfeature.internal.ufc.Flag; +import com.datadog.openfeature.internal.ufc.Rule; +import com.datadog.openfeature.internal.ufc.ServerConfiguration; +import com.datadog.openfeature.internal.ufc.Split; +import com.datadog.openfeature.internal.ufc.ValueType; +import com.datadog.openfeature.internal.ufc.Variant; +import dev.openfeature.sdk.Client; +import dev.openfeature.sdk.ErrorCode; +import dev.openfeature.sdk.EvaluationContext; +import dev.openfeature.sdk.FlagEvaluationDetails; +import dev.openfeature.sdk.Hook; +import dev.openfeature.sdk.HookContext; +import dev.openfeature.sdk.ImmutableContext; +import dev.openfeature.sdk.MutableContext; +import dev.openfeature.sdk.OpenFeatureAPI; +import dev.openfeature.sdk.Value; +import dev.openfeature.sdk.multiprovider.MultiProvider; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.NullSource; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Exercises real OpenFeature resolution and the provider telemetry hooks, capturing each output + * boundary: flag evaluation events (EVP), OTel evaluation metrics, span enrichment, exposures and + * {@link ExposureHook} evaluations. + * + *

The provider reads its settings from the real system properties and environment through {@link + * com.datadog.openfeature.internal.bridge.AgentBridges#detect()}, so the configuration scenarios + * use the system property; the environment fallback is covered by {@code SettingsTest}. + */ +class EvaluationTelemetryOptOutTest { + private static final String PARAMETER = "dd_evaluation_telemetry_opt_out"; + private static final String SYSTEM_PROPERTY = + "dd.feature.flags.evaluation.telemetry.per.subject.opt.out.enabled"; + + private final List events = new CopyOnWriteArrayList<>(); + private final List spans = new CopyOnWriteArrayList<>(); + private final List exposures = new CopyOnWriteArrayList<>(); + private final List customerEvaluations = new CopyOnWriteArrayList<>(); + private final AtomicInteger metrics = new AtomicInteger(); + private final ExposureDeduplicationCache exposureCache = new ExposureDeduplicationCache(); + private final SpanEnricher spanEnricher = + new SpanEnricher() { + @Override + public void serialId(final int serialId, final boolean doLog, final String targetingKey) { + EvaluationTelemetryOptOutTest.this.spans.add("serial-id:" + targetingKey); + } + + @Override + public void runtimeDefault(final String flagKey, final Object defaultValue) { + EvaluationTelemetryOptOutTest.this.spans.add("runtime-default:" + flagKey); + } + }; + private final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + private String previousProperty; + private FeatureFlagsRuntime runtime; + private CapturingEvaluator evaluator; + private Client client; + + @BeforeEach + void setUp() throws Exception { + this.previousProperty = System.getProperty(SYSTEM_PROPERTY); + System.clearProperty(SYSTEM_PROPERTY); + this.api.setEvaluationContext(new MutableContext()); + GlobalOpenTelemetry.resetForTest(); + final OpenTelemetry otel = mock(OpenTelemetry.class, RETURNS_DEEP_STUBS); + final LongCounter counter = mock(LongCounter.class); + when(otel.getMeterProvider() + .meterBuilder("ddtrace.openfeature") + .build() + .counterBuilder("feature_flag.evaluations") + .setUnit("{evaluation}") + .setDescription("Number of feature flag evaluations") + .build()) + .thenReturn(counter); + doAnswer(invocation -> this.metrics.incrementAndGet()) + .when(counter) + .add(eq(1L), any(Attributes.class)); + GlobalOpenTelemetry.set(otel); + this.runtime = capturingRuntime(new CapturingPipeline(this.events)); + this.api.addHooks(new ExposureHook(this.customerEvaluations::add)); + } + + @AfterEach + void tearDown() { + if (this.previousProperty == null) { + System.clearProperty(SYSTEM_PROPERTY); + } else { + System.setProperty(SYSTEM_PROPERTY, this.previousProperty); + } + this.api.clearHooks(); + this.api.setEvaluationContext(new MutableContext()); + this.api.shutdown(); + GlobalOpenTelemetry.resetForTest(); + } + + @Test + void systemPropertyEnablesPerSubjectOptOut() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + start(new Provider.Options()); + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + assertEquals(1, details.getValue()); + assertEquals("on", details.getVariant()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + evaluate("subject-2", new Value(false)); + evaluate("subject-3", null); + assertTelemetryCounts(2); + assertEquals(3, this.exposures.size()); + } + + @ParameterizedTest + @NullSource + @ValueSource(strings = {"false", "invalid", ""}) + void systemPropertyDefaultsToDisabled(final String value) throws Exception { + if (value != null) { + System.setProperty(SYSTEM_PROPERTY, value); + } + start(new Provider.Options()); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void explicitFalseOverridesEnabledSystemProperty() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(false)); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void explicitTrueOverridesDisabledSystemProperty() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "false"); + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(true)); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void systemPropertyIsCapturedAtProviderConstruction() throws Exception { + System.setProperty(SYSTEM_PROPERTY, "true"); + final Provider.Options options = new Provider.Options(); + start(options); + System.setProperty(SYSTEM_PROPERTY, "false"); + evaluate("subject-1", new Value(true)); + assertTelemetryCounts(0); + this.api.shutdown(); + start(options); + evaluate("subject-2", new Value(true)); + assertTelemetryCounts(1); + } + + @Test + void optedOutSubjectKeepsAssignmentAndExposureButEmitsNoEvaluationTelemetry() throws Exception { + start(); + + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + + assertNull(details.getErrorCode()); + assertEquals(1, details.getValue()); + assertEquals("on", details.getVariant()); + assertEquals(1, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + assertTelemetryCounts(0); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void telemetryAttributeRemainsAvailableForTargetingAndExposure(final boolean enabled) + throws Exception { + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(enabled)); + + for (final boolean observeFullEvaluationData : new boolean[] {false, true}) { + this.evaluator.accept(targetingConfiguration(observeFullEvaluationData)); + final String subject = "subject-" + observeFullEvaluationData; + final int previousEvents = this.events.size(); + final int previousExposures = this.exposures.size(); + + final FlagEvaluationDetails allowed = evaluate(subject, new Value(false)); + final FlagEvaluationDetails optedOut = evaluate(subject, new Value(true)); + final FlagEvaluationDetails repeated = evaluate(subject, new Value(true)); + final FlagEvaluationDetails stringValue = + evaluate(subject + "-string", new Value("true")); + + assertNull(allowed.getErrorCode()); + assertEquals(1, allowed.getValue()); + assertEquals("on", allowed.getVariant()); + for (final FlagEvaluationDetails details : asList(optedOut, repeated, stringValue)) { + assertNull(details.getErrorCode()); + assertEquals(2, details.getValue(), "the ordinary targeting rule sees the attribute"); + assertEquals("targeted", details.getVariant()); + } + assertEquals( + previousExposures + 3, this.exposures.size(), "normal exposure deduplication applies"); + assertEquals(false, exposureAttribute(previousExposures)); + assertEquals(true, exposureAttribute(previousExposures + 1)); + assertEquals("true", exposureAttribute(previousExposures + 2)); + assertTelemetryCounts(previousEvents + (enabled ? 2 : 4)); + + if (observeFullEvaluationData) { + assertEquals(false, this.events.get(previousEvents).attrs.get(PARAMETER)); + assertEquals("true", this.events.get(this.events.size() - 1).attrs.get(PARAMETER)); + if (!enabled) { + assertEquals(true, this.events.get(previousEvents + 1).attrs.get(PARAMETER)); + } + } else { + for (final FlagEvalEvent event : this.events.subList(previousEvents, this.events.size())) { + assertTrue(event.attrs.isEmpty(), "normal evaluation privacy rules still apply"); + } + } + } + } + + @Test + void allowedSubjectEmitsEachSignal() throws Exception { + start(); + + evaluate("subject-1", new Value(false)); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + @ParameterizedTest + @MethodSource("nonOptOutValues") + void onlyBooleanTrueSuppressesTelemetry(final Value value) throws Exception { + start(); + + evaluate("subject-1", value); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + static Stream nonOptOutValues() { + return Stream.of( + Arguments.of((Value) null), + Arguments.of(new Value()), + Arguments.of(new Value(false)), + Arguments.of(new Value("true")), + Arguments.of(new Value(1)), + Arguments.of(new Value(singletonList(new Value(true))))); + } + + @Test + void theParameterHasNoEffectUnlessEnabled() throws Exception { + start(new Provider.Options()); + + evaluate("subject-1", new Value(true)); + + assertTelemetryCounts(1); + assertEquals(1, this.exposures.size()); + } + + @Test + void theParameterIsCaseSensitiveAndDoesNotAcceptTheOldName() throws Exception { + start(); + + this.client.getIntegerDetails( + "flag", + 0, + new MutableContext("subject-1") + .add("dd_telemetry_opt_out", true) + .add("ddEvaluationTelemetryOptOut", true)); + + assertTelemetryCounts(1); + } + + @Test + void providerOptionsAreCapturedAtConstruction() throws Exception { + final Provider.Options options = + new Provider.Options().enableEvaluationTelemetryOptOutParameter(true); + start(options); + options.enableEvaluationTelemetryOptOutParameter(false); + + evaluate("subject-1", new Value(true)); + + assertTelemetryCounts(0); + } + + @Test + void eachEvaluationUsesItsOwnDecisionWithoutChangingExistingEvents() throws Exception { + start(); + + evaluate("same-subject", new Value(false)); + evaluate("same-subject", new Value(true)); + assertTelemetryCounts(1); + evaluate("same-subject", new Value(false)); + + assertTelemetryCounts(2); + assertEquals(1, this.exposures.size(), "exposure deduplication remains independent"); + assertEquals(3, this.customerEvaluations.size()); + } + + @Test + void missingAndWrongTypeFlagsKeepDefaultsWithoutEmittingTelemetry() throws Exception { + start(); + final EvaluationContext context = context("subject-1", new Value(true)); + + final FlagEvaluationDetails missing = + this.client.getIntegerDetails("missing", 42, context); + final FlagEvaluationDetails wrongType = + this.client.getStringDetails("flag", "default", context); + + assertEquals(ErrorCode.FLAG_NOT_FOUND, missing.getErrorCode()); + assertEquals(42, missing.getValue()); + assertEquals(ErrorCode.TYPE_MISMATCH, wrongType.getErrorCode()); + assertEquals("default", wrongType.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(2, this.customerEvaluations.size()); + + this.client.getIntegerDetails("missing", 42, context("subject-2", new Value(false))); + assertTelemetryCounts(1); + } + + @Test + void contextMergingHonorsThePerEvaluationOverride() throws Exception { + start(); + this.api.setEvaluationContext(new MutableContext().add(PARAMETER, true)); + + this.client.getIntegerDetails("flag", 0, new ImmutableContext("subject-1")); + evaluate("subject-2", new Value(false)); + + assertTelemetryCounts(1); + assertEquals("subject-2", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } + + @Test + void applicationBeforeHooksCanSupplyTheOptOutDecision() throws Exception { + start(); + this.api.addHooks( + new Hook() { + @Override + public Optional before( + final HookContext context, final Map hints) { + return Optional.of(new ImmutableContext(singletonMap(PARAMETER, new Value(true)))); + } + }); + + evaluate("subject-1", new Value(false)); + + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void applicationAfterHooksCannotUndoTheEvaluationDecision() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, true); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + invocationContext.add(PARAMETER, false); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertNull(details.getErrorCode()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void optingOutInAnAfterHookStopsTelemetryBeforeEmission() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, false); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + invocationContext.add(PARAMETER, true); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertNull(details.getErrorCode()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + } + + @Test + void aFailingBeforeHookDoesNotLeakErrorTelemetry() throws Exception { + start(); + final MutableContext invocationContext = new MutableContext("subject-1").add(PARAMETER, true); + this.api.addHooks( + new Hook() { + @Override + public Optional before( + final HookContext context, final Map hints) { + throw new IllegalStateException("application hook failed"); + } + + @Override + public void error( + final HookContext context, + final Exception error, + final Map hints) { + invocationContext.add(PARAMETER, false); + } + }); + + final FlagEvaluationDetails details = + this.client.getIntegerDetails("flag", 0, invocationContext); + + assertEquals(ErrorCode.GENERAL, details.getErrorCode()); + assertEquals(0, details.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + @Test + void nestedEvaluationsKeepIndependentDecisions() throws Exception { + start(); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if ("outer".equals(context.getCtx().getTargetingKey())) { + assertEquals(1, evaluate("inner", new Value(false)).getValue()); + } + } + }); + + assertEquals(1, evaluate("outer", new Value(true)).getValue()); + + assertTelemetryCounts(1); + assertEquals("inner", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + assertEquals(2, this.customerEvaluations.size()); + } + + @Test + void concurrentSubjectsCannotChangeEachOthersDecision() throws Exception { + start(); + final CountDownLatch optedOutEvaluated = new CountDownLatch(1); + final CountDownLatch allowOptedOutToFinish = new CountDownLatch(1); + this.api.addHooks( + new Hook() { + @Override + public void after( + final HookContext context, + final FlagEvaluationDetails details, + final Map hints) { + if ("opted-out".equals(context.getCtx().getTargetingKey())) { + optedOutEvaluated.countDown(); + try { + assertTrue(allowOptedOutToFinish.await(10, SECONDS)); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + } + }); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + final Future> pending = + executor.submit(() -> evaluate("opted-out", new Value(true))); + assertTrue(optedOutEvaluated.await(10, SECONDS)); + assertEquals(1, evaluate("allowed", new Value(false)).getValue()); + allowOptedOutToFinish.countDown(); + assertNull(pending.get(10, SECONDS).getErrorCode()); + + assertTelemetryCounts(1); + assertEquals("allowed", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } finally { + allowOptedOutToFinish.countDown(); + executor.shutdownNow(); + } + } + + @Test + void independentProvidersKeepTheirOwnActivationSetting() throws Exception { + start(); + this.api.setProviderAndWait( + "unrestricted", + provider(new Provider.Options(), new CapturingEvaluator(defaultConfiguration()))); + + evaluate("opted-out", new Value(true)); + this.api + .getClient("unrestricted") + .getIntegerDetails("flag", 0, context("unrestricted-subject", new Value(true))); + + assertTelemetryCounts(1); + assertEquals("unrestricted-subject", this.events.get(0).targetingKey); + assertEquals(2, this.exposures.size()); + } + + @Test + void optOutWorksInsideAMultiProvider() throws Exception { + start(); + this.api.setProviderAndWait( + new MultiProvider( + singletonList( + provider( + new Provider.Options().enableEvaluationTelemetryOptOutParameter(true), + new CapturingEvaluator(defaultConfiguration()))))); + + assertEquals(1, evaluate("opted-out", new Value(true)).getValue()); + assertTelemetryCounts(0); + assertEquals(1, this.exposures.size()); + + assertEquals(1, evaluate("allowed", new Value(false)).getValue()); + assertTelemetryCounts(1); + assertEquals(2, this.exposures.size()); + } + + @Test + void missingConfigurationKeepsTheDefaultWithoutEmittingTelemetry() throws Exception { + start(); + this.evaluator.accept(null); + + final FlagEvaluationDetails details = evaluate("subject-1", new Value(true)); + + assertEquals(ErrorCode.PROVIDER_NOT_READY, details.getErrorCode()); + assertEquals(0, details.getValue()); + assertTelemetryCounts(0); + assertEquals(0, this.exposures.size()); + assertEquals(1, this.customerEvaluations.size()); + } + + private void start() throws Exception { + start(new Provider.Options().enableEvaluationTelemetryOptOutParameter(true)); + } + + private void start(final Provider.Options options) throws Exception { + this.evaluator = new CapturingEvaluator(defaultConfiguration()); + this.api.setProviderAndWait(provider(options, this.evaluator)); + this.client = this.api.getClient(); + } + + /** + * Creates a provider with span enrichment on, recording span enrichment to {@link #spanEnricher}. + */ + private Provider provider(final Provider.Options options, final DDEvaluator evaluator) + throws ReflectiveOperationException { + final Provider provider = new Provider(options, evaluator, Boolean.TRUE); + // No seam injects a span enricher: the provider takes it from the detected agent bridge. + final Field enricher = SpanEnrichmentHook.class.getDeclaredField("enricher"); + enricher.setAccessible(true); + enricher.set(provider.spanEnrichmentHook(), this.spanEnricher); + return provider; + } + + /** Creates a runtime only holding a flag evaluation pipeline, without starting it. */ + private static FeatureFlagsRuntime capturingRuntime(final FlagEvaluationPipeline evaluations) + throws ReflectiveOperationException { + final Constructor constructor = + FeatureFlagsRuntime.class.getDeclaredConstructor( + ConfigurationService.class, ExposurePipeline.class, FlagEvaluationPipeline.class); + constructor.setAccessible(true); + return constructor.newInstance(mock(ConfigurationService.class), null, evaluations); + } + + private static ServerConfiguration defaultConfiguration() { + final Allocation allocation = + new Allocation( + "allocation", + null, + null, + null, + singletonList(new Split(emptyList(), "on", emptyMap(), 7)), + true); + final Flag flag = + new Flag( + "flag", + true, + ValueType.INTEGER, + singletonMap("on", new Variant("on", 1)), + singletonList(allocation)); + return new ServerConfiguration("", "", false, null, singletonMap("flag", flag)); + } + + private static ServerConfiguration targetingConfiguration( + final boolean observeFullEvaluationData) { + final Rule rule = + new Rule( + singletonList( + new ConditionConfiguration( + ConditionOperator.ONE_OF, PARAMETER, singletonList(true)))); + final Allocation targeted = + new Allocation( + "targeted", + singletonList(rule), + null, + null, + singletonList(new Split(emptyList(), "targeted", emptyMap(), 7)), + true); + final Allocation fallback = + new Allocation( + "fallback", + null, + null, + null, + singletonList(new Split(emptyList(), "on", emptyMap(), 7)), + true); + final Map variants = new HashMap<>(); + variants.put("on", new Variant("on", 1)); + variants.put("targeted", new Variant("targeted", 2)); + final Flag flag = + new Flag("flag", true, ValueType.INTEGER, variants, asList(targeted, fallback)); + return new ServerConfiguration( + "", "", observeFullEvaluationData, null, singletonMap("flag", flag)); + } + + private FlagEvaluationDetails evaluate(final String subject, final Value optOut) { + return this.client.getIntegerDetails("flag", 0, context(subject, optOut)); + } + + private static EvaluationContext context(final String subject, final Value optOut) { + return new MutableContext( + subject, optOut == null ? emptyMap() : singletonMap(PARAMETER, optOut)); + } + + private Object exposureAttribute(final int index) { + return this.exposures.get(index).subject.attributes.get(PARAMETER); + } + + private void assertTelemetryCounts(final int expected) { + assertAll( + () -> assertEquals(expected, this.events.size(), "evaluation EVP enqueue count"), + () -> assertEquals(expected, this.metrics.get(), "OTel evaluation count"), + () -> assertEquals(expected, this.spans.size(), "automatic span enrichment count")); + } + + /** + * A Datadog evaluator holding a configuration, sharing the test runtime for flag evaluation + * events, and the test exposure cache and capture for exposures. + */ + private final class CapturingEvaluator extends DDEvaluator { + CapturingEvaluator(final ServerConfiguration configuration) { + super( + () -> {}, + AgentBridge.NONE, + TestSettings.of("experimental.flagging.provider.span.enrichment.enabled", "true")); + accept(configuration); + } + + @Override + public boolean initialize( + final long timeout, final TimeUnit unit, final EvaluationContext context) { + return hasConfiguration(); + } + + @Override + FeatureFlagsRuntime runtime() { + return EvaluationTelemetryOptOutTest.this.runtime; + } + + @Override + ExposureDeduplicationCache exposureCache() { + return EvaluationTelemetryOptOutTest.this.exposureCache; + } + + @Override + boolean isExposureDeliveryEnabled() { + return true; + } + + @Override + void recordExposure(final ExposureEvent event) { + EvaluationTelemetryOptOutTest.this.exposures.add(event); + } + } + + /** A flag evaluation pipeline capturing enqueued events, never started. */ + private static final class CapturingPipeline extends FlagEvaluationPipeline { + private final List events; + + CapturingPipeline(final List events) { + super( + 1, 1, SECONDS, (route, json) -> {}, emptyMap(), new RuntimeServices(HealthMetrics.NOOP)); + this.events = events; + } + + @Override + public boolean hasCapacityForEnqueue() { + return true; + } + + @Override + public void enqueue(final FlagEvalEvent event) { + this.events.add(event); + } + } +} diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java index 9f08140abd0..97325594548 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/ProviderTest.java @@ -7,7 +7,9 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; 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; @@ -25,6 +27,7 @@ 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.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; import com.datadog.openfeature.internal.ufc.Allocation; @@ -369,18 +372,58 @@ public void testDisabledProductFailsInitialization() { assertThat(error.getCause().getMessage(), equalTo("Feature Flags is disabled")); } + @MethodSource("providerMethods") + @ParameterizedTest + public void testRemoteConfigurationWithoutCompatibleAgentUsesDefaults( + final String flag, final E defaultValue, final EvaluateMethod method) throws Exception { + final Provider provider = newRemoteConfigurationProvider(AgentBridge.NONE); + final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + + assertDoesNotThrow(() -> api.setProviderAndWait(provider)); + // READY means initialization completed; evaluation details still report unavailable config. + assertThat(api.getClient().getProviderState(), equalTo(ProviderState.READY)); + assertThat(initializationState(provider), equalTo("FALLBACK")); + assertFalse(this.evaluator.hasConfiguration()); + assertNull(this.evaluator.runtime(), "no polling or event delivery starts without RC"); + final FlagEvaluationDetails result = method.evaluate(api.getClient(), flag, defaultValue); + assertThat(result.getValue(), equalTo(defaultValue)); + assertThat(result.getErrorCode(), equalTo(ErrorCode.PROVIDER_NOT_READY)); + assertThat(result.getReason(), equalTo("ERROR")); + assertNull(result.getVariant()); + assertTrue(this.cdn.requests().isEmpty(), "explicit RC must not silently switch to CDN"); + } + @Test - public void testRemoteConfigurationWithoutAgentFailsInitialization() { + public void testIncompatibleRemoteConfigurationBridgeUsesDefaults() throws Exception { + final AgentBridge bridge = mock(AgentBridge.class); + when(bridge.healthMetrics()).thenReturn(HealthMetrics.NOOP); + when(bridge.remoteConfiguration()).thenThrow(new NoSuchMethodError("older agent RC bridge")); + final Provider provider = newRemoteConfigurationProvider(bridge); + final OpenFeatureAPI api = OpenFeatureAPI.getInstance(); + + assertDoesNotThrow(() -> api.setProviderAndWait(provider)); + assertThat(initializationState(provider), equalTo("FALLBACK")); + final FlagEvaluationDetails result = + api.getClient().getStringDetails("missing", "caller-default"); + assertThat(result.getValue(), equalTo("caller-default")); + assertThat(result.getErrorCode(), equalTo(ErrorCode.PROVIDER_NOT_READY)); + assertNull(this.evaluator.runtime()); + assertDoesNotThrow(api::shutdown); + } + + private Provider newRemoteConfigurationProvider(final AgentBridge bridge) { final Provider[] providerRef = new Provider[1]; - final DDEvaluator remoteConfig = + this.evaluator = new DDEvaluator( () -> providerRef[0].onConfigurationChange(), - AgentBridge.NONE, - TestSettings.of("feature.flags.configuration.source", "remote_config")); - providerRef[0] = new Provider(new Options().initTimeout(10, MILLISECONDS), remoteConfig); - - assertThrows( - FatalError.class, () -> OpenFeatureAPI.getInstance().setProviderAndWait(providerRef[0])); + bridge, + TestSettings.of( + "feature.flags.configuration.source", + "remote_config", + "feature.flags.configuration.source.agentless.base.url", + this.cdn.uri("/").toString())); + providerRef[0] = new Provider(new Options().initTimeout(10, MILLISECONDS), this.evaluator); + return providerRef[0]; } @Test @@ -609,5 +652,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java index 2d719897fce..995174b8fa4 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/FeatureFlagsRuntimeTest.java @@ -1,6 +1,8 @@ package com.datadog.openfeature.internal; import static java.util.Collections.emptyMap; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +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.assertNotNull; @@ -15,13 +17,16 @@ 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.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import com.datadog.openfeature.internal.config.TestSettings; +import com.datadog.openfeature.internal.flagevaluation.FlagEvalEvent; import com.datadog.openfeature.internal.ufc.ServerConfiguration; import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; import java.util.function.Consumer; import org.junit.jupiter.api.Test; @@ -51,7 +56,7 @@ void remoteConfigurationRequiresTheJavaAgent() { AgentBridge.NONE, TestSettings.of("feature.flags.configuration.source", "remote_config"))); - assertTrue(error.getMessage().contains("requires the Datadog Java agent")); + assertTrue(error.getMessage().contains("requires a compatible Datadog Java agent")); } @Test @@ -239,6 +244,97 @@ void lastReleaseFlushesPendingExposures() { assertEquals(1, ((List) requests.get(0).get("exposures")).size()); } + @Test + void agentShutdownFlushesPendingExposuresOnce() { + final TestConfigurationSource source = new TestConfigurationSource(); + final List> requests = new CopyOnWriteArrayList<>(); + final TestAgentBridge bridge = + new TestAgentBridge(source, (route, json) -> requests.add(JsonReading.readObject(json))) + .withAgentShutdown(); + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.acquire( + bridge, + TestSettings.of( + "feature.flags.configuration.source", "remote_config", + "flagging.evaluation.counts.enabled", "false")); + try { + assertNotNull(bridge.shutdownFlush, "the runtime registers its flush with the agent"); + runtime.recordExposure(exposure()); + + bridge.shutdownFlush.run(); + + assertTrue(source.closed); + assertTrue(bridge.shutdownUnregistered); + assertEquals(1, requests.size()); + } finally { + FeatureFlagsRuntime.release(); + } + assertEquals(1, requests.size(), "a release after the agent shutdown does not flush again"); + } + + @Test + void releaseUnregistersTheAgentShutdownFlush() { + final TestAgentBridge bridge = + new TestAgentBridge(new TestConfigurationSource(), null).withAgentShutdown(); + FeatureFlagsRuntime.acquire( + bridge, TestSettings.of("feature.flags.configuration.source", "remote_config")); + + FeatureFlagsRuntime.release(); + + assertTrue(bridge.shutdownUnregistered); + } + + @Test + void stopFlushesBothPipelinesWithinOneShutdownBudget() throws Exception { + final CountDownLatch exposuresPosted = new CountDownLatch(1); + final CountDownLatch evaluationsPosted = new CountDownLatch(1); + final CountDownLatch releasePosts = new CountDownLatch(1); + final EventTransport stalledTransport = + (route, json) -> { + ("exposures".equals(route) ? exposuresPosted : evaluationsPosted).countDown(); + try { + releasePosts.await(10, SECONDS); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }; + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.create( + new TestAgentBridge(new TestConfigurationSource(), stalledTransport), + TestSettings.of("feature.flags.configuration.source", "remote_config")); + runtime.start(); + try { + runtime.recordExposure(exposure()); + runtime.evaluations().enqueue(new FlagEvalEvent("f", "v", "a", "s", 1L, emptyMap())); + + final long started = System.nanoTime(); + runtime.stop(); + final long elapsedMillis = NANOSECONDS.toMillis(System.nanoTime() - started); + + assertEquals(0, evaluationsPosted.getCount(), "the evaluations final flush started"); + assertEquals(0, exposuresPosted.getCount(), "the exposures final flush started"); + assertTrue(elapsedMillis < 1300, "Both pipeline waits took " + elapsedMillis + " ms"); + } finally { + releasePosts.countDown(); + } + } + + @Test + void failingAgentShutdownSubscriptionFallsBackToTheJvmShutdownHook() { + final TestConfigurationSource source = new TestConfigurationSource(); + final TestAgentBridge bridge = + new TestAgentBridge(source, null) + .withFailingAgentShutdown(new IllegalStateException("Shutdown in progress")); + + final FeatureFlagsRuntime runtime = + FeatureFlagsRuntime.acquire( + bridge, TestSettings.of("feature.flags.configuration.source", "remote_config")); + assertNotNull(runtime); + FeatureFlagsRuntime.release(); + + assertTrue(source.closed, "the started runtime is stopped on release"); + } + @Test void failedStartDoesNotCacheTheSharedRuntime() { final ConfigurationSource failing = @@ -280,6 +376,16 @@ void recordsExposuresOnlyWhenDeliveryIsEnabled() { null)); } + private static com.datadog.openfeature.internal.exposure.ExposureEvent exposure() { + return new com.datadog.openfeature.internal.exposure.ExposureEvent( + 1L, + new com.datadog.openfeature.internal.exposure.Allocation("a"), + new com.datadog.openfeature.internal.exposure.Flag("f"), + new com.datadog.openfeature.internal.exposure.Variant("v"), + new com.datadog.openfeature.internal.exposure.Subject("s", emptyMap()), + null); + } + private static final class TestConfigurationSource implements ConfigurationSource { volatile Consumer listener; volatile boolean closed; @@ -294,6 +400,10 @@ public AutoCloseable subscribe(final Consumer listener) { private static final class TestAgentBridge implements AgentBridge { private final ConfigurationSource remoteConfiguration; private final EventTransport eventProxy; + private boolean agentShutdown; + private RuntimeException shutdownFailure; + volatile Runnable shutdownFlush; + volatile boolean shutdownUnregistered; TestAgentBridge( final ConfigurationSource remoteConfiguration, final EventTransport eventProxy) { @@ -325,5 +435,29 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + if (!this.agentShutdown) { + return null; + } + return listener -> { + if (this.shutdownFailure != null) { + throw this.shutdownFailure; + } + this.shutdownFlush = listener; + return () -> this.shutdownUnregistered = true; + }; + } + + TestAgentBridge withAgentShutdown() { + this.agentShutdown = true; + return this; + } + + TestAgentBridge withFailingAgentShutdown(final RuntimeException failure) { + this.shutdownFailure = failure; + return withAgentShutdown(); + } } } 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 index 1cfd0a827c5..80c3a9258b3 100644 --- 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 @@ -9,6 +9,7 @@ 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.ShutdownSource; import com.datadog.openfeature.internal.bridge.SpanEnricher; import java.util.HashMap; import java.util.Map; @@ -106,6 +107,37 @@ void readsTheExposuresDatadogLoggingToggle(final String value, final boolean ena .isExposuresDatadogLoggingEnabled()); } + @TableTest({ + "scenario | agent | property | environment | enabled", + "default | | | | false ", + "environment enables | | | 'true' | true ", + "environment disabled | | | 'false' | false ", + "invalid environment | | | 'invalid' | false ", + "blank environment | | | '' | false ", + "property overrides environment | | 'false' | 'true' | false ", + "property enables | | 'true' | 'false' | true ", + "blank property falls back | | ' ' | 'true' | true ", + "agent overrides property | 'false' | 'true' | 'true' | false ", + "agent enables | 'true' | 'false' | 'false' | true " + }) + void readsTheEvaluationTelemetryPerSubjectOptOutToggle( + final String agent, final String property, final String environment, final boolean enabled) { + final Map bridge = new HashMap<>(); + putIfNotNull(bridge, "feature.flags.evaluation.telemetry.per.subject.opt.out.enabled", agent); + final Map properties = new HashMap<>(); + putIfNotNull( + properties, "dd.feature.flags.evaluation.telemetry.per.subject.opt.out.enabled", property); + final Map variables = new HashMap<>(); + putIfNotNull( + variables, + "DD_FEATURE_FLAGS_EVALUATION_TELEMETRY_PER_SUBJECT_OPT_OUT_ENABLED", + environment); + final Settings settings = + new Settings(new SettingAgentBridge(bridge), properties::get, variables::get); + + assertEquals(enabled, settings.isEvaluationTelemetryPerSubjectOptOutEnabled()); + } + @TableTest({ "scenario | value | interval", "valid | '10' | 10 ", @@ -199,5 +231,10 @@ public HealthMetrics healthMetrics() { public SpanEnricher spanEnricher() { return SpanEnricher.NOOP; } + + @Override + public ShutdownSource shutdown() { + return null; + } } } diff --git a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java index 8ff26333df8..ed3f92e35cc 100644 --- a/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java +++ b/dd-openfeature/src/test/java/com/datadog/openfeature/internal/delivery/DirectIntakeTransportTest.java @@ -5,26 +5,21 @@ 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 static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; 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; @@ -34,7 +29,6 @@ public class DirectIntakeTransportTest { private LocalHttpServer server; - private final List sleeps = new CopyOnWriteArrayList<>(); @BeforeEach void setUp() throws IOException { @@ -83,26 +77,14 @@ void postsJsonWithApiKeyAndTraceHeaders() throws Exception { "scenario | status", "server error | 500 ", "unavailable | 503 ", - "too many requests | 429 " + "too many requests | 429 ", + "bad request | 400 ", + "forbidden | 403 ", + "not found | 404 ", + "timeout | 408 ", + "redirect | 302 " }) - 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) { + void sendsFailedBatchesOnce(final int status) { this.server.otherwise(status, null); assertThrows(IOException.class, () -> transport().post("exposures", new byte[] {'{', '}'})); @@ -110,72 +92,28 @@ void doesNotRetryOtherFailures(final int status) { 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 " + "scenario | failure ", + "connection | connection ", + "timeout | timeout ", + "TLS failure | tls ", + "DNS failure | dns ", + "interrupted IO | interrupted", + "other IO | other " }) - void retriesOnlyConnectionAndTimeoutFailures(final IOException failure, final int attempts) - throws Exception { + void doesNotRetryIoFailures(final IOException failure) { 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); + new DirectIntakeTransport(this.server.uri("/api/v2/"), "secret", client); assertThrows(failure.getClass(), () -> transport.post("exposures", new byte[] {'{', '}'})); - assertEquals(attempts - 1, this.sleeps.size()); + verify(client, times(1)).sendAsync(any(), any()); } /** - * Creates an I/O failure for a retry scenario. + * Creates an I/O failure for a send scenario. * * @param value the failure kind. * @return the failure. @@ -198,23 +136,8 @@ public static IOException failure(final String value) { } } - @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); + this.server.uri("/api/v2/"), "secret", HttpClient.newHttpClient()); } } 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 index ce05e06d685..585236831f4 100644 --- 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 @@ -42,11 +42,15 @@ void switchesToDirectIntakeForGoodWhenTheProxyIsUnavailable() throws IOException } @Test - void doesNotReplayAmbiguousProxyFailures() throws IOException { + void switchesToDirectIntakeForGoodWithoutReplayingAmbiguousProxyFailures() 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); + + this.transport.post("exposures", PAYLOAD); + + verify(this.proxy, times(1)).post("exposures", PAYLOAD); + verify(this.direct, times(1)).post("exposures", PAYLOAD); } }