diff --git a/communication/src/main/java/datadog/communication/http/HttpRetryPolicy.java b/communication/src/main/java/datadog/communication/http/HttpRetryPolicy.java index 3578f0e1c93..a645f53e29a 100644 --- a/communication/src/main/java/datadog/communication/http/HttpRetryPolicy.java +++ b/communication/src/main/java/datadog/communication/http/HttpRetryPolicy.java @@ -54,7 +54,13 @@ public class HttpRetryPolicy implements AutoCloseable { private final double delayFactor; private final boolean suppressInterrupts; - private HttpRetryPolicy( + /** + * Creates a retry policy. + * + *

Protected so products with a stricter cross-SDK retry contract can reuse the shared HTTP + * retry loop while supplying their own response, exception, and backoff rules. + */ + protected HttpRetryPolicy( int retriesLeft, long delay, double delayFactor, boolean suppressInterrupts) { this.retriesLeft = retriesLeft; this.delay = delay; diff --git a/communication/src/main/java/datadog/communication/http/OkHttpUtils.java b/communication/src/main/java/datadog/communication/http/OkHttpUtils.java index 9dc13229131..44105ae772d 100644 --- a/communication/src/main/java/datadog/communication/http/OkHttpUtils.java +++ b/communication/src/main/java/datadog/communication/http/OkHttpUtils.java @@ -20,6 +20,7 @@ import java.util.List; import java.util.Map; import javax.annotation.Nullable; +import okhttp3.Call; import okhttp3.ConnectionPool; import okhttp3.ConnectionSpec; import okhttp3.Credentials; @@ -405,19 +406,39 @@ public void writeTo(BufferedSink sink) throws IOException { public static Response sendWithRetries( OkHttpClient httpClient, HttpRetryPolicy.Factory retryPolicyFactory, Request request) throws IOException { + return sendWithRetries((Call.Factory) httpClient, retryPolicyFactory, request); + } + + public static Response sendWithRetries( + Call.Factory callFactory, HttpRetryPolicy.Factory retryPolicyFactory, Request request) + throws IOException { + return sendWithRetries(callFactory, retryPolicyFactory, request, response -> response); + } + + public static T sendWithRetries( + Call.Factory callFactory, + HttpRetryPolicy.Factory retryPolicyFactory, + Request request, + ResponseMapper responseMapper) + throws IOException { try (HttpRetryPolicy retryPolicy = retryPolicyFactory.create()) { while (true) { + Response response = null; try { - Response response = httpClient.newCall(request).execute(); + response = callFactory.newCall(request).execute(); if (response.isSuccessful()) { - return response; + return responseMapper.map(response); } if (!retryPolicy.shouldRetry(response)) { - return response; + return responseMapper.map(response); } else { closeQuietly(response); + response = null; } } catch (Exception ex) { + if (response != null) { + closeQuietly(response); + } if (!retryPolicy.shouldRetry(ex)) { throw ex; } @@ -428,6 +449,11 @@ public static Response sendWithRetries( } } + @FunctionalInterface + public interface ResponseMapper { + T map(Response response) throws IOException; + } + private static void closeQuietly(Response response) { try { response.close(); diff --git a/communication/src/test/java/datadog/communication/http/OkHttpUtilsRetryTest.java b/communication/src/test/java/datadog/communication/http/OkHttpUtilsRetryTest.java new file mode 100644 index 00000000000..45f84a4810a --- /dev/null +++ b/communication/src/test/java/datadog/communication/http/OkHttpUtilsRetryTest.java @@ -0,0 +1,53 @@ +package datadog.communication.http; + +import static datadog.communication.http.OkHttpUtils.sendWithRetries; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.net.ConnectException; +import java.util.concurrent.atomic.AtomicInteger; +import okhttp3.Call; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.ResponseBody; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import org.junit.jupiter.api.Test; + +class OkHttpUtilsRetryTest { + + @Test + void retriesResponseMappingFailureThroughCallFactory() throws Exception { + final MockWebServer server = new MockWebServer(); + final OkHttpClient client = new OkHttpClient(); + final AtomicInteger mappingAttempts = new AtomicInteger(); + server.enqueue(new MockResponse().setBody("first")); + server.enqueue(new MockResponse().setBody("second")); + server.start(); + + try { + final Request request = new Request.Builder().url(server.url("/configuration")).build(); + + final String body = + sendWithRetries( + (Call.Factory) client, + new HttpRetryPolicy.Factory(1, 0, 1), + request, + response -> { + try (ResponseBody responseBody = response.body()) { + if (mappingAttempts.getAndIncrement() == 0) { + throw new ConnectException("response body could not be mapped"); + } + return responseBody.string(); + } + }); + + assertEquals("second", body); + assertEquals(2, mappingAttempts.get()); + assertEquals(2, server.getRequestCount()); + } finally { + client.dispatcher().executorService().shutdownNow(); + client.connectionPool().evictAll(); + server.shutdown(); + } + } +} diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java index 555e027d498..2e316eeb253 100644 --- a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java +++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java @@ -135,8 +135,7 @@ private enum AgentFeature { AGENTLESS_LOG_SUBMISSION(GeneralConfig.AGENTLESS_LOG_SUBMISSION_ENABLED, false), APP_LOGS_COLLECTION(GeneralConfig.APP_LOGS_COLLECTION_ENABLED, false), LLMOBS(LlmObsConfig.LLMOBS_ENABLED, false), - LLMOBS_AGENTLESS(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED, false), - FEATURE_FLAGGING(FeatureFlaggingConfig.FLAGGING_PROVIDER_ENABLED, false); + LLMOBS_AGENTLESS(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED, false); private final String configKey; private final String systemProp; @@ -283,7 +282,7 @@ public static void start( agentlessLogSubmissionEnabled = isFeatureEnabled(AgentFeature.AGENTLESS_LOG_SUBMISSION); appLogsCollectionEnabled = isFeatureEnabled(AgentFeature.APP_LOGS_COLLECTION); llmObsEnabled = isFeatureEnabled(AgentFeature.LLMOBS); - featureFlaggingEnabled = isFeatureEnabled(AgentFeature.FEATURE_FLAGGING); + featureFlaggingEnabled = isFeatureFlaggingEnabled(); // setup writers when llmobs is enabled to accomodate apm and llmobs if (llmObsEnabled) { @@ -531,6 +530,9 @@ public static void shutdown(final boolean sync) { if (flareEnabled) { stopFlarePoller(); } + if (featureFlaggingEnabled) { + shutdownFeatureFlagging(AGENT_CLASSLOADER); + } if (agentlessLogSubmissionEnabled) { shutdownLogsIntake(); @@ -1287,6 +1289,20 @@ private static void maybeStartFeatureFlagging(final Class scoClass, final Obj } } + static void shutdownFeatureFlagging(final ClassLoader agentClassLoader) { + if (agentClassLoader == null) { + return; + } + try { + final Class ffSysClass = + agentClassLoader.loadClass("com.datadog.featureflag.FeatureFlaggingSystem"); + final Method stopMethod = ffSysClass.getMethod("stop"); + stopMethod.invoke(null); + } catch (final Throwable e) { + log.warn("Unable to stop Feature Flagging subsystem", e); + } + } + private static void maybeInstallLogsIntake(Class scoClass, Object sco) { if (agentlessLogSubmissionEnabled || appLogsCollectionEnabled) { StaticEventLogger.begin("Logs Intake"); @@ -1739,6 +1755,45 @@ private static boolean isFeatureEnabled(AgentFeature feature) { } } + private static boolean isFeatureFlaggingEnabled() { + final Boolean providerEnabled = + featureFlaggingBooleanSetting(FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED); + final String configurationSource = + featureFlaggingSetting(FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE); + final Boolean legacyProviderEnabled = + featureFlaggingBooleanSetting(FeatureFlaggingConfig.EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED); + + return FeatureFlaggingConfig.resolveConfiguration( + providerEnabled, configurationSource, legacyProviderEnabled) + .isEnabled(); + } + + @SuppressFBWarnings( + value = "NP_BOOLEAN_RETURN_NULL", + justification = "A null value preserves the distinction between absent and explicitly false") + private static Boolean featureFlaggingBooleanSetting(final String configKey) { + final String value = featureFlaggingSetting(configKey); + if (value == null) { + return null; + } + return Boolean.parseBoolean(value) || "1".equals(value); + } + + private static String featureFlaggingSetting(final String configKey) { + final String systemProperty = propertyNameToSystemPropertyName(configKey); + String value = SystemProperties.get(systemProperty); + if (value == null) { + value = getStableConfig(FLEET, configKey); + } + if (value == null) { + value = ddGetEnv(systemProperty); + } + if (value == null) { + value = getStableConfig(LOCAL, configKey); + } + return value; + } + /** * @see datadog.trace.api.ProductActivation#fromString(String) */ diff --git a/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/AgentFeatureFlaggingLifecycleTest.java b/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/AgentFeatureFlaggingLifecycleTest.java new file mode 100644 index 00000000000..75e9987c4da --- /dev/null +++ b/dd-java-agent/agent-bootstrap/src/test/java/datadog/trace/bootstrap/AgentFeatureFlaggingLifecycleTest.java @@ -0,0 +1,48 @@ +package datadog.trace.bootstrap; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class AgentFeatureFlaggingLifecycleTest { + + @BeforeEach + void reset() { + FakeFeatureFlaggingSystem.stopCalls.set(0); + } + + @Test + void shutdownInvokesFeatureFlaggingSystemStopThroughAgentClassLoader() { + final ClassLoader classLoader = + new ClassLoader(null) { + @Override + public Class loadClass(final String name) throws ClassNotFoundException { + if ("com.datadog.featureflag.FeatureFlaggingSystem".equals(name)) { + return FakeFeatureFlaggingSystem.class; + } + return super.loadClass(name); + } + }; + + Agent.shutdownFeatureFlagging(classLoader); + + assertEquals(1, FakeFeatureFlaggingSystem.stopCalls.get()); + } + + @Test + void shutdownIsNoopBeforeAgentClassLoaderExists() { + Agent.shutdownFeatureFlagging(null); + + assertEquals(0, FakeFeatureFlaggingSystem.stopCalls.get()); + } + + public static final class FakeFeatureFlaggingSystem { + private static final AtomicInteger stopCalls = new AtomicInteger(); + + public static void stop() { + stopCalls.incrementAndGet(); + } + } +} diff --git a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureProviderSmokeTest.groovy b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureProviderSmokeTest.groovy index cb4c641d667..ca80ca47f8b 100644 --- a/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureProviderSmokeTest.groovy +++ b/dd-smoke-tests/openfeature/src/test/groovy/datadog/smoketest/springboot/OpenFeatureProviderSmokeTest.groovy @@ -42,6 +42,7 @@ class OpenFeatureProviderSmokeTest extends AbstractServerSmokeTest { command.addAll(['-jar', springBootShadowJar, "--server.port=${httpPort}".toString()]) final builder = new ProcessBuilder(command).directory(new File(buildDirectory)) builder.environment().put('DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED', 'true') + builder.environment().put('DD_FEATURE_FLAGS_CONFIGURATION_SOURCE', 'remote_config') return builder } diff --git a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java index 40d5753c478..78d0eddc9ae 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java @@ -48,6 +48,9 @@ public final class ConfigDefaults { static final boolean DEFAULT_INJECT_DATADOG_ATTRIBUTE = true; static final String DEFAULT_SITE = "datadoghq.com"; + static final String DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE = "agentless"; + static final int DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS = 30; + static final int DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS = 5; static final boolean DEFAULT_CODE_ORIGIN_FOR_SPANS_INTERFACE_SUPPORT = false; static final int DEFAULT_CODE_ORIGIN_MAX_USER_FRAMES = 8; diff --git a/internal-api/build.gradle.kts b/internal-api/build.gradle.kts index 65be1563eda..e96d4d34e9d 100644 --- a/internal-api/build.gradle.kts +++ b/internal-api/build.gradle.kts @@ -269,6 +269,7 @@ dependencies { api(project(":components:context")) api(project(":components:environment")) api(project(":components:json")) + implementation(project(":products:feature-flagging:feature-flagging-config")) api(project(":utils:config-utils")) api(project(":utils:time-utils")) diff --git a/internal-api/src/main/java/datadog/trace/api/Config.java b/internal-api/src/main/java/datadog/trace/api/Config.java index 0fa16a5b838..fc472c17b3d 100644 --- a/internal-api/src/main/java/datadog/trace/api/Config.java +++ b/internal-api/src/main/java/datadog/trace/api/Config.java @@ -87,6 +87,8 @@ import static datadog.trace.api.ConfigDefaults.DEFAULT_ELASTICSEARCH_BODY_ENABLED; import static datadog.trace.api.ConfigDefaults.DEFAULT_ELASTICSEARCH_PARAMS_ENABLED; import static datadog.trace.api.ConfigDefaults.DEFAULT_EXPERIMENTATAL_JEE_SPLIT_BY_DEPLOYMENT; +import static datadog.trace.api.ConfigDefaults.DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS; +import static datadog.trace.api.ConfigDefaults.DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS; import static datadog.trace.api.ConfigDefaults.DEFAULT_GRPC_CLIENT_ERROR_STATUSES; import static datadog.trace.api.ConfigDefaults.DEFAULT_GRPC_SERVER_ERROR_STATUSES; import static datadog.trace.api.ConfigDefaults.DEFAULT_HEALTH_METRICS_ENABLED; @@ -725,6 +727,14 @@ import static datadog.trace.api.config.TracerConfig.WRITER_BAGGAGE_INJECT; import static datadog.trace.api.config.TracerConfig.WRITER_LINKS_INJECT; import static datadog.trace.api.config.TracerConfig.WRITER_TYPE; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.isSupportedConfigurationSource; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.resolveConfiguration; import static datadog.trace.api.telemetry.LogCollector.SEND_TELEMETRY; import static datadog.trace.bootstrap.instrumentation.api.WriterConstants.OTLP_WRITER_TYPE; import static datadog.trace.util.CollectionUtils.tryMakeImmutableList; @@ -741,6 +751,7 @@ import datadog.trace.api.config.OtlpConfig; import datadog.trace.api.config.ProfilingConfig; import datadog.trace.api.config.TracerConfig; +import datadog.trace.api.featureflag.config.FeatureFlaggingConfig; import datadog.trace.api.iast.IastContext; import datadog.trace.api.iast.IastDetectionMode; import datadog.trace.api.iast.telemetry.Verbosity; @@ -1219,6 +1230,12 @@ public static String getHostName() { private final int remoteConfigMaxExtraServices; + private final boolean featureFlaggingProviderEnabled; + private final String featureFlaggingConfigurationSource; + private final String featureFlaggingConfigurationSourceAgentlessBaseUrl; + private final int featureFlaggingConfigurationSourcePollIntervalSeconds; + private final int featureFlaggingConfigurationSourceRequestTimeoutSeconds; + private final boolean dbmInjectSqlBaseHash; private final String dbmPropagationMode; private final boolean dbmTracePreparedStatements; @@ -2849,6 +2866,63 @@ PROFILING_DATADOG_PROFILER_ENABLED, isDatadogProfilerSafeInCurrentEnvironment()) configProvider.getInteger( REMOTE_CONFIG_MAX_EXTRA_SERVICES, DEFAULT_REMOTE_CONFIG_MAX_EXTRA_SERVICES); + final Boolean configuredFeatureFlaggingProviderEnabled = + configProvider.getBoolean(FEATURE_FLAGS_ENABLED); + final Boolean legacyFeatureFlaggingProviderEnabled = + configProvider.getBoolean(EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED); + final String configuredFeatureFlaggingConfigurationSource = + configProvider.isSet(FEATURE_FLAGS_CONFIGURATION_SOURCE) + ? configProvider.getString(FEATURE_FLAGS_CONFIGURATION_SOURCE) + : null; + final FeatureFlaggingConfig.Resolution resolvedFeatureFlaggingConfiguration = + resolveConfiguration( + configuredFeatureFlaggingProviderEnabled, + configuredFeatureFlaggingConfigurationSource, + legacyFeatureFlaggingProviderEnabled); + if (legacyFeatureFlaggingProviderEnabled != null) { + log.warn( + "Setting {} is deprecated. Use {} and {} instead.", + EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED, + FEATURE_FLAGS_ENABLED, + FEATURE_FLAGS_CONFIGURATION_SOURCE); + } + if (!isSupportedConfigurationSource(configuredFeatureFlaggingConfigurationSource)) { + log.warn( + "Unsupported Feature Flagging configuration source; provider disabled: '{}'", + resolvedFeatureFlaggingConfiguration.getSource()); + } + featureFlaggingProviderEnabled = resolvedFeatureFlaggingConfiguration.isEnabled(); + featureFlaggingConfigurationSource = resolvedFeatureFlaggingConfiguration.getSource(); + featureFlaggingConfigurationSourceAgentlessBaseUrl = + configProvider.getStringNotEmpty( + FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, null); + int configuredFeatureFlaggingPollIntervalSeconds = + configProvider.getInteger( + FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS, + DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS); + if (configuredFeatureFlaggingPollIntervalSeconds <= 0) { + log.warn( + "Invalid Feature Flagging agentless poll interval: {}. The value must be positive", + configuredFeatureFlaggingPollIntervalSeconds); + configuredFeatureFlaggingPollIntervalSeconds = + DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS; + } + featureFlaggingConfigurationSourcePollIntervalSeconds = + configuredFeatureFlaggingPollIntervalSeconds; + int configuredFeatureFlaggingRequestTimeoutSeconds = + configProvider.getInteger( + FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS, + DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS); + if (configuredFeatureFlaggingRequestTimeoutSeconds <= 0) { + log.warn( + "Invalid Feature Flagging agentless request timeout: {}. The value must be positive", + configuredFeatureFlaggingRequestTimeoutSeconds); + configuredFeatureFlaggingRequestTimeoutSeconds = + DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS; + } + featureFlaggingConfigurationSourceRequestTimeoutSeconds = + configuredFeatureFlaggingRequestTimeoutSeconds; + dynamicInstrumentationEnabled = configProvider.getBoolean( DYNAMIC_INSTRUMENTATION_ENABLED, DEFAULT_DYNAMIC_INSTRUMENTATION_ENABLED); @@ -4682,6 +4756,26 @@ public int getRemoteConfigMaxExtraServices() { return remoteConfigMaxExtraServices; } + public boolean isFeatureFlaggingProviderEnabled() { + return featureFlaggingProviderEnabled; + } + + public String getFeatureFlaggingConfigurationSource() { + return featureFlaggingConfigurationSource; + } + + public String getFeatureFlaggingConfigurationSourceAgentlessBaseUrl() { + return featureFlaggingConfigurationSourceAgentlessBaseUrl; + } + + public int getFeatureFlaggingConfigurationSourcePollIntervalSeconds() { + return featureFlaggingConfigurationSourcePollIntervalSeconds; + } + + public int getFeatureFlaggingConfigurationSourceRequestTimeoutSeconds() { + return featureFlaggingConfigurationSourceRequestTimeoutSeconds; + } + public boolean isDynamicInstrumentationEnabled() { return dynamicInstrumentationEnabled; } @@ -6540,6 +6634,16 @@ public String toString() { + remoteConfigMaxPayloadSize + ", remoteConfigIntegrityCheckEnabled=" + remoteConfigIntegrityCheckEnabled + + ", featureFlaggingProviderEnabled=" + + featureFlaggingProviderEnabled + + ", featureFlaggingConfigurationSource=" + + featureFlaggingConfigurationSource + + ", featureFlaggingConfigurationSourceAgentlessBaseUrl=" + + featureFlaggingConfigurationSourceAgentlessBaseUrl + + ", featureFlaggingConfigurationSourcePollIntervalSeconds=" + + featureFlaggingConfigurationSourcePollIntervalSeconds + + ", featureFlaggingConfigurationSourceRequestTimeoutSeconds=" + + featureFlaggingConfigurationSourceRequestTimeoutSeconds + ", debuggerEnabled=" + dynamicInstrumentationEnabled + ", debuggerUploadTimeout=" diff --git a/internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java b/internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java index 752adb8899d..e9ef282cc7d 100644 --- a/internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java +++ b/internal-api/src/main/java/datadog/trace/util/AgentThreadFactory.java @@ -66,7 +66,8 @@ public enum AgentThread { LLMOBS_EVALS_PROCESSOR("dd-llmobs-evals-processor"), - FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor"); + FEATURE_FLAG_EXPOSURE_PROCESSOR("dd-ffe-exposure-processor"), + FEATURE_FLAG_CONFIGURATION_POLLER("dd-feature-flagging-http-poller"); public final String threadName; diff --git a/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy index 11f100efc42..3b96c713974 100644 --- a/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy +++ b/internal-api/src/test/groovy/datadog/trace/api/ConfigTest.groovy @@ -2,6 +2,8 @@ package datadog.trace.api import static datadog.trace.api.ConfigDefaults.DEFAULT_HTTP_CLIENT_ERROR_STATUSES import static datadog.trace.api.ConfigDefaults.DEFAULT_HTTP_SERVER_ERROR_STATUSES +import static datadog.trace.api.ConfigDefaults.DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS +import static datadog.trace.api.ConfigDefaults.DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS import static datadog.trace.api.ConfigDefaults.DEFAULT_PARTIAL_FLUSH_MIN_SPANS import static datadog.trace.api.ConfigDefaults.DEFAULT_SERVICE_NAME import static datadog.trace.api.ConfigDefaults.DEFAULT_TRACE_LONG_RUNNING_FLUSH_INTERVAL @@ -55,6 +57,11 @@ import static datadog.trace.api.config.GeneralConfig.TAGS import static datadog.trace.api.config.GeneralConfig.TRACER_METRICS_IGNORED_RESOURCES import static datadog.trace.api.config.GeneralConfig.TRACE_OTEL_SEMANTICS_ENABLED import static datadog.trace.api.config.GeneralConfig.VERSION +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_ENABLED import static datadog.trace.api.config.JmxFetchConfig.JMX_FETCH_CHECK_PERIOD import static datadog.trace.api.config.JmxFetchConfig.JMX_FETCH_ENABLED import static datadog.trace.api.config.JmxFetchConfig.JMX_FETCH_METRICS_CONFIGS @@ -3478,4 +3485,89 @@ class ConfigTest extends DDSpecification { "1" | true "0" | false } + + def "agentless feature flag timing uses positive configured values"() { + setup: + Properties properties = new Properties() + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS, "60") + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS, "4") + + when: + def config = new Config(ConfigProvider.withPropertiesOverride(properties)) + + then: + config.featureFlaggingConfigurationSourcePollIntervalSeconds == 60 + config.featureFlaggingConfigurationSourceRequestTimeoutSeconds == 4 + } + + def "feature flag configuration source normalizes #value to #expected"() { + setup: + Properties properties = new Properties() + if (value != null) { + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE, value) + } + + when: + def config = new Config(ConfigProvider.withPropertiesOverride(properties)) + + then: + config.featureFlaggingConfigurationSource == expected + + where: + value | expected + null | "agentless" + "" | "agentless" + " " | "agentless" + " ReMoTe_ConFiG " | "remote_config" + "not-a-real-source" | "not-a-real-source" + " OFFLINE " | "offline" + } + + def "feature flag configuration applies migration precedence"() { + setup: + Properties properties = new Properties() + if (providerEnabled != null) { + properties.setProperty(FEATURE_FLAGS_ENABLED, providerEnabled.toString()) + } + if (source != null) { + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE, source) + } + if (legacyProviderEnabled != null) { + properties.setProperty(EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED, legacyProviderEnabled.toString()) + } + + when: + def config = new Config(ConfigProvider.withPropertiesOverride(properties)) + + then: + config.featureFlaggingProviderEnabled == expectedEnabled + config.featureFlaggingConfigurationSource == expectedSource + + where: + providerEnabled | source | legacyProviderEnabled | expectedEnabled | expectedSource + null | null | null | true | "agentless" + true | null | null | true | "agentless" + null | null | true | true | "remote_config" + null | null | false | false | null + null | "agentless" | true | true | "agentless" + null | "remote_config" | false | true | "remote_config" + false | "agentless" | true | false | "agentless" + true | null | false | false | null + null | "not-a-source" | null | false | "not-a-source" + null | "offline" | true | false | "offline" + } + + def "agentless feature flag timing falls back for non-positive values"() { + setup: + Properties properties = new Properties() + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS, "0") + properties.setProperty(FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS, "-1") + + when: + def config = new Config(ConfigProvider.withPropertiesOverride(properties)) + + then: + config.featureFlaggingConfigurationSourcePollIntervalSeconds == DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_POLL_INTERVAL_SECONDS + config.featureFlaggingConfigurationSourceRequestTimeoutSeconds == DEFAULT_FEATURE_FLAGGING_CONFIGURATION_SOURCE_REQUEST_TIMEOUT_SECONDS + } } diff --git a/metadata/agent-jar-checks.properties b/metadata/agent-jar-checks.properties index d2565d0f6fd..a096b2cf519 100644 --- a/metadata/agent-jar-checks.properties +++ b/metadata/agent-jar-checks.properties @@ -1,7 +1,7 @@ # Agent jar structural invariants — edit intentionally, commit with the change that justified it. -# Max agent jar size in bytes. Raise only when the size growth is intentional (~33.02 MiB). -jar.size.budget = 34619392 +# Max agent jar size in bytes. Raise only when the size growth is intentional (34 MiB = 35651584). +jar.size.budget = 35651584 # Minimum combined class + classdata count in the assembled agent jar. # Set to ~98% of the actual count at the time of the last intentional change. diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json index d6b02fbfa28..779def0bb27 100644 --- a/metadata/supported-configurations.json +++ b/metadata/supported-configurations.json @@ -1513,6 +1513,46 @@ "aliases": [] } ], + "DD_FEATURE_FLAGS_ENABLED": [ + { + "version": "A", + "type": "boolean", + "default": "true", + "aliases": [] + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE": [ + { + "version": "A", + "type": "string", + "default": "agentless", + "aliases": [] + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL": [ + { + "version": "A", + "type": "string", + "default": null, + "aliases": [] + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS": [ + { + "version": "A", + "type": "int", + "default": "30", + "aliases": [] + } + ], + "DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS": [ + { + "version": "A", + "type": "int", + "default": "5", + "aliases": [] + } + ], "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": [ { "version": "A", @@ -11970,5 +12010,7 @@ } ] }, - "deprecations": {} + "deprecations": { + "DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED": "DD_FEATURE_FLAGS_ENABLED" + } } diff --git a/products/feature-flagging/feature-flagging-agent/build.gradle.kts b/products/feature-flagging/feature-flagging-agent/build.gradle.kts index a219a48456c..d2dc4baea2b 100644 --- a/products/feature-flagging/feature-flagging-agent/build.gradle.kts +++ b/products/feature-flagging/feature-flagging-agent/build.gradle.kts @@ -17,9 +17,11 @@ dependencies { api(libs.slf4j) api(project(":products:feature-flagging:feature-flagging-lib")) api(project(":internal-api")) + compileOnly(project(":products:feature-flagging:feature-flagging-config")) testImplementation(libs.bundles.junit5) testImplementation(libs.bundles.mockito) + testImplementation(project(":products:feature-flagging:feature-flagging-config")) testImplementation(project(":utils:test-utils")) testRuntimeOnly(project(":dd-trace-core")) } diff --git a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java index 9761caf3895..adbae0a6e44 100644 --- a/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java +++ b/products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java @@ -1,7 +1,11 @@ package com.datadog.featureflag; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_REMOTE_CONFIG; + import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -9,24 +13,69 @@ public class FeatureFlaggingSystem { private static final Logger LOGGER = LoggerFactory.getLogger(FeatureFlaggingSystem.class); - private static volatile RemoteConfigService CONFIG_SERVICE; + private static volatile ConfigurationSourceService CONFIG_SERVICE; private static volatile ExposureWriter EXPOSURE_WRITER; private static volatile SpanEnrichmentWriter SPAN_ENRICHMENT_WRITER; + private static volatile FeatureFlaggingGateway.ActivationListener ACTIVATION_LISTENER; + private static volatile boolean STARTED; private FeatureFlaggingSystem() {} - public static void start(final SharedCommunicationObjects sco) { + public static synchronized void start(final SharedCommunicationObjects sco) { + if (STARTED) { + LOGGER.debug("Feature Flagging system already started"); + return; + } LOGGER.debug("Feature Flagging system starting"); final Config config = Config.get(); + STARTED = true; + + if (!config.isFeatureFlaggingProviderEnabled()) { + LOGGER.debug("Feature Flagging system disabled"); + return; + } + + if (CONFIGURATION_SOURCE_AGENTLESS.equals(config.getFeatureFlaggingConfigurationSource())) { + final FeatureFlaggingGateway.ActivationListener activationListener = + () -> activateAgentless(sco, config); + ACTIVATION_LISTENER = activationListener; + FeatureFlaggingGateway.addActivationListener(activationListener); + LOGGER.debug("Feature Flagging system awaiting application provider activation"); + return; + } - if (!config.isRemoteConfigEnabled()) { - throw new IllegalStateException("Feature Flagging system started without RC"); + try { + initializeSystem(sco, config); + } catch (final RuntimeException | Error e) { + STARTED = false; + throw e; + } + } + + private static synchronized void activateAgentless( + final SharedCommunicationObjects sco, final Config config) { + final FeatureFlaggingGateway.ActivationListener activationListener = ACTIVATION_LISTENER; + if (!STARTED || activationListener == null) { + return; + } + ACTIVATION_LISTENER = null; + FeatureFlaggingGateway.removeActivationListener(activationListener); + try { + initializeSystem(sco, config); + } catch (final RuntimeException | Error e) { + STARTED = false; + throw e; } - CONFIG_SERVICE = new RemoteConfigServiceImpl(sco, config); - CONFIG_SERVICE.init(); + } - EXPOSURE_WRITER = new ExposureWriterImpl(sco, config); - EXPOSURE_WRITER.init(); + private static void initializeSystem(final SharedCommunicationObjects sco, final Config config) { + final ConfigurationSourceService configService = createConfigurationSourceService(sco, config); + if (configService == null) { + LOGGER.debug("Feature Flagging system disabled by unsupported configuration source"); + return; + } + final ExposureWriter exposureWriter = new ExposureWriterImpl(sco, config); + initialize(configService, exposureWriter); // APM span enrichment: agent-side listener for flag-evaluation seam events. Uses the process- // wide singleton so a subsystem restart reuses the one already-registered trace interceptor @@ -39,19 +88,74 @@ public static void start(final SharedCommunicationObjects sco) { LOGGER.debug("Feature Flagging system started"); } - public static void stop() { - if (SPAN_ENRICHMENT_WRITER != null) { - SPAN_ENRICHMENT_WRITER.close(); - SPAN_ENRICHMENT_WRITER = null; + static void initialize( + final ConfigurationSourceService configService, final ExposureWriter exposureWriter) { + try { + if (configService != null) { + configService.init(); + } + exposureWriter.init(); + CONFIG_SERVICE = configService; + EXPOSURE_WRITER = exposureWriter; + } catch (final RuntimeException | Error e) { + try { + exposureWriter.close(); + } finally { + if (configService != null) { + configService.close(); + } + } + throw e; + } + } + + static ConfigurationSourceService createConfigurationSourceService( + final SharedCommunicationObjects sco, final Config config) { + final String configurationSource = config.getFeatureFlaggingConfigurationSource(); + if (CONFIGURATION_SOURCE_REMOTE_CONFIG.equals(configurationSource)) { + if (!config.isRemoteConfigEnabled()) { + throw new IllegalStateException("Feature Flagging system started without RC"); + } + return new RemoteConfigServiceImpl(sco, config); } - if (EXPOSURE_WRITER != null) { - EXPOSURE_WRITER.close(); - EXPOSURE_WRITER = null; + if (CONFIGURATION_SOURCE_AGENTLESS.equals(configurationSource)) { + return new AgentlessConfigurationSource(config); } - if (CONFIG_SERVICE != null) { - CONFIG_SERVICE.close(); - CONFIG_SERVICE = null; + return null; + } + + public static synchronized void stop() { + final FeatureFlaggingGateway.ActivationListener activationListener = ACTIVATION_LISTENER; + final SpanEnrichmentWriter spanEnrichmentWriter = SPAN_ENRICHMENT_WRITER; + final ExposureWriter exposureWriter = EXPOSURE_WRITER; + final ConfigurationSourceService configService = CONFIG_SERVICE; + STARTED = false; + ACTIVATION_LISTENER = null; + SPAN_ENRICHMENT_WRITER = null; + EXPOSURE_WRITER = null; + CONFIG_SERVICE = null; + if (activationListener != null) { + FeatureFlaggingGateway.removeActivationListener(activationListener); + } + try { + if (spanEnrichmentWriter != null) { + spanEnrichmentWriter.close(); + } + } finally { + try { + if (exposureWriter != null) { + exposureWriter.close(); + } + } finally { + if (configService != null) { + configService.close(); + } + } } LOGGER.debug("Feature Flagging system stopped"); } + + static boolean isAwaitingApplicationActivation() { + return ACTIVATION_LISTENER != null; + } } diff --git a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java index a78e3fb9e49..d408d91da4e 100644 --- a/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java +++ b/products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java @@ -1,11 +1,21 @@ package com.datadog.featureflag; import static datadog.trace.api.config.RemoteConfigConfig.REMOTE_CONFIGURATION_ENABLED; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.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.clearInvocations; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; @@ -15,6 +25,7 @@ import datadog.remoteconfig.ConfigurationPoller; import datadog.remoteconfig.Product; import datadog.trace.api.Config; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.test.junit.utils.config.WithConfig; import okhttp3.HttpUrl; import okhttp3.OkHttpClient; @@ -23,6 +34,56 @@ class FeatureFlaggingSystemTest { @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") + @WithConfig( + key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, + value = "http://127.0.0.1:1") + void agentlessStartWaitsForApplicationProviderActivation() { + SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); + clearInvocations(sharedCommunicationObjects); + + try { + FeatureFlaggingSystem.start(sharedCommunicationObjects); + + assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + verifyNoInteractions(sharedCommunicationObjects); + + FeatureFlaggingGateway.activate(); + + assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + } finally { + FeatureFlaggingSystem.stop(); + } + + assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") + @WithConfig( + key = FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL, + value = "http://127.0.0.1:1") + void agentlessStopRemovesPendingApplicationProviderActivation() { + SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); + clearInvocations(sharedCommunicationObjects); + + try { + FeatureFlaggingSystem.start(sharedCommunicationObjects); + assertTrue(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + + FeatureFlaggingSystem.stop(); + FeatureFlaggingGateway.activate(); + + assertFalse(FeatureFlaggingSystem.isAwaitingApplicationActivation()); + verifyNoInteractions(sharedCommunicationObjects); + } finally { + FeatureFlaggingSystem.stop(); + } + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "remote_config") + @WithConfig(key = REMOTE_CONFIGURATION_ENABLED, value = "true") void testFeatureFlagSystemInitialization() { ConfigurationPoller poller = mock(ConfigurationPoller.class); DDAgentFeaturesDiscovery discovery = mock(DDAgentFeaturesDiscovery.class); @@ -34,12 +95,14 @@ void testFeatureFlagSystemInitialization() { sharedCommunicationObjects.agentUrl = HttpUrl.get("http://localhost"); sharedCommunicationObjects.agentHttpClient = new OkHttpClient.Builder().build(); + FeatureFlaggingSystem.start(sharedCommunicationObjects); FeatureFlaggingSystem.start(sharedCommunicationObjects); verify(poller).addCapabilities(Capabilities.CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); verify(poller).addListener(eq(Product.FFE_FLAGS), any(ConfigurationDeserializer.class), any()); verify(poller).start(); + FeatureFlaggingSystem.stop(); FeatureFlaggingSystem.stop(); verify(poller).removeCapabilities(Capabilities.CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); @@ -48,6 +111,7 @@ void testFeatureFlagSystemInitialization() { } @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "remote_config") @WithConfig(key = REMOTE_CONFIGURATION_ENABLED, value = "false") void testThatRemoteConfigIsRequired() { SharedCommunicationObjects sharedCommunicationObjects = mock(SharedCommunicationObjects.class); @@ -60,4 +124,131 @@ void testThatRemoteConfigIsRequired() { FeatureFlaggingSystem.stop(); } } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless") + @WithConfig(key = REMOTE_CONFIGURATION_ENABLED, value = "false") + void agentlessConfigurationSourceUsesHttpServiceWithoutRemoteConfig() { + assertInstanceOf( + AgentlessConfigurationSource.class, + FeatureFlaggingSystem.createConfigurationSourceService( + sharedCommunicationObjects(), Config.get())); + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "remote_config") + @WithConfig(key = REMOTE_CONFIGURATION_ENABLED, value = "true") + void explicitRemoteConfigUsesRemoteConfigService() { + SharedCommunicationObjects sharedCommunicationObjects = sharedCommunicationObjects(); + when(sharedCommunicationObjects.configurationPoller(any(Config.class))) + .thenReturn(mock(ConfigurationPoller.class)); + + assertInstanceOf( + RemoteConfigServiceImpl.class, + FeatureFlaggingSystem.createConfigurationSourceService( + sharedCommunicationObjects, Config.get())); + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "invalid") + void invalidConfigurationSourceDoesNotStartNetworkSource() { + assertNull( + FeatureFlaggingSystem.createConfigurationSourceService( + sharedCommunicationObjects(), Config.get())); + } + + @Test + void unsupportedNormalizedConfigurationSourceDoesNotStartNetworkSource() { + Config config = mock(Config.class); + when(config.getFeatureFlaggingConfigurationSource()).thenReturn("invalid"); + + assertNull( + FeatureFlaggingSystem.createConfigurationSourceService( + sharedCommunicationObjects(), config)); + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "offline") + void offlineConfigurationSourceDoesNotStartNetworkSource() { + assertNull( + FeatureFlaggingSystem.createConfigurationSourceService( + sharedCommunicationObjects(), Config.get())); + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "offline") + void startWithOfflineConfigurationSourceDisablesSystem() { + SharedCommunicationObjects sharedCommunicationObjects = mock(SharedCommunicationObjects.class); + + try { + assertDoesNotThrow(() -> FeatureFlaggingSystem.start(sharedCommunicationObjects)); + verifyNoInteractions(sharedCommunicationObjects); + } finally { + FeatureFlaggingSystem.stop(); + } + } + + @Test + @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "invalid") + void startWithInvalidConfigurationSourceDisablesSystem() { + SharedCommunicationObjects sharedCommunicationObjects = mock(SharedCommunicationObjects.class); + + try { + assertDoesNotThrow(() -> FeatureFlaggingSystem.start(sharedCommunicationObjects)); + verifyNoInteractions(sharedCommunicationObjects); + } finally { + FeatureFlaggingSystem.stop(); + } + } + + @Test + void initializationFailureClosesConfigurationSourceAndExposureWriter() { + ConfigurationSourceService configService = mock(ConfigurationSourceService.class); + ExposureWriter exposureWriter = mock(ExposureWriter.class); + doThrow(new IllegalStateException("exposure init failed")).when(exposureWriter).init(); + + assertThrows( + IllegalStateException.class, + () -> FeatureFlaggingSystem.initialize(configService, exposureWriter)); + + verify(configService).init(); + verify(configService).close(); + verify(exposureWriter).close(); + } + + @Test + void initializationFailureWithoutConfigurationSourceClosesExposureWriter() { + ExposureWriter exposureWriter = mock(ExposureWriter.class); + doThrow(new IllegalStateException("exposure init failed")).when(exposureWriter).init(); + + assertThrows( + IllegalStateException.class, () -> FeatureFlaggingSystem.initialize(null, exposureWriter)); + + verify(exposureWriter).close(); + } + + @Test + void initializationFailureClosesConfigurationSourceWhenExposureWriterCloseFails() { + ConfigurationSourceService configService = mock(ConfigurationSourceService.class); + ExposureWriter exposureWriter = mock(ExposureWriter.class); + doThrow(new IllegalStateException("exposure init failed")).when(exposureWriter).init(); + doThrow(new IllegalArgumentException("exposure close failed")).when(exposureWriter).close(); + + assertThrows( + IllegalArgumentException.class, + () -> FeatureFlaggingSystem.initialize(configService, exposureWriter)); + + verify(configService).close(); + } + + private static SharedCommunicationObjects sharedCommunicationObjects() { + DDAgentFeaturesDiscovery discovery = mock(DDAgentFeaturesDiscovery.class); + when(discovery.supportsEvpProxy()).thenReturn(true); + when(discovery.getEvpProxyEndpoint()).thenReturn("/evp_proxy/"); + SharedCommunicationObjects sharedCommunicationObjects = mock(SharedCommunicationObjects.class); + when(sharedCommunicationObjects.featuresDiscovery(any(Config.class))).thenReturn(discovery); + sharedCommunicationObjects.agentUrl = HttpUrl.get("http://localhost"); + sharedCommunicationObjects.agentHttpClient = new OkHttpClient.Builder().build(); + return sharedCommunicationObjects; + } } diff --git a/products/feature-flagging/feature-flagging-api/README.md b/products/feature-flagging/feature-flagging-api/README.md index 47733020559..74bbccca2e2 100644 --- a/products/feature-flagging/feature-flagging-api/README.md +++ b/products/feature-flagging/feature-flagging-api/README.md @@ -85,5 +85,19 @@ OTEL_EXPORTER_OTLP_PROTOCOL=grpc ## Requirements - Java 11+ -- Datadog Agent with Remote Configuration enabled -- `DD_EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED=true` +- `DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=agentless` uses the Datadog agentless + backend. Set `DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL` to a + different HTTP backend while keeping agentless delivery semantics. A bare + host uses the standard rules-based server path; a URL with a path is used as + the exact UFC endpoint. Configured URLs are opaque: the SDK does not add the + Datadog-managed `dd_env` query parameter, so custom backends must include any + required tenant or environment scope in the configured URL. The derived + Datadog-managed endpoint is + `https://ufc-server.ff-cdn./api/v2/feature-flagging/config/rules-based/server` + and expects UFC under the JSON:API `data.attributes` response member. It is + intended for supported commercial sites; use an explicit base URL elsewhere. + Agentless responses do not have an SDK-imposed payload-size limit. + `remote_config` uses the existing Agent Remote + Configuration path. `offline` is reserved for startup-provided UFC bytes; + until those bytes are implemented, no network source starts and evaluations + use defaults. diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java index 654fb6e8f6c..6d55c54a5b3 100644 --- a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java +++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java @@ -67,6 +67,7 @@ public DDEvaluator(final Runnable configCallback) { @Override public boolean initialize( final long timeout, final TimeUnit unit, final EvaluationContext context) throws Exception { + FeatureFlaggingGateway.activate(); FeatureFlaggingGateway.addConfigListener(this); return initializationLatch.await(timeout, unit) || hasConfiguration(); } diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java index 3421fe3454e..7b6cc01e021 100644 --- a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java +++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java @@ -23,6 +23,7 @@ import com.squareup.moshi.JsonWriter; import com.squareup.moshi.Moshi; import com.squareup.moshi.Types; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.ufc.v1.Flag; import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; import dev.openfeature.sdk.ErrorCode; @@ -64,6 +65,22 @@ public class DDEvaluatorTest { private static final JsonAdapter> FIXTURE_LIST_ADAPTER = MOSHI.adapter(FIXTURE_LIST_TYPE); + @Test + public void testInitializeSignalsApplicationProviderActivation() throws Exception { + final FeatureFlaggingGateway.ActivationListener listener = + mock(FeatureFlaggingGateway.ActivationListener.class); + final DDEvaluator evaluator = new DDEvaluator(mock(Runnable.class)); + FeatureFlaggingGateway.addActivationListener(listener); + try { + evaluator.initialize(1, MILLISECONDS, mock(EvaluationContext.class)); + + verify(listener).activate(); + } finally { + evaluator.shutdown(); + FeatureFlaggingGateway.removeActivationListener(listener); + } + } + private static Arguments[] valueMappingTestCases() { return new Arguments[] { // String mappings diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java index 2704b4be341..c8f5625c855 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/FeatureFlaggingGateway.java @@ -11,11 +11,16 @@ public abstract class FeatureFlaggingGateway { public interface ConfigListener extends Consumer {} + public interface ActivationListener { + void activate(); + } + public interface ExposureListener extends Consumer {} public interface SpanEnrichmentListener extends Consumer {} private static final List CONFIG_LISTENERS = new CopyOnWriteArrayList<>(); + private static final List ACTIVATION_LISTENERS = new CopyOnWriteArrayList<>(); private static final List EXPOSURE_LISTENERS = new CopyOnWriteArrayList<>(); private static final List SPAN_ENRICHMENT_LISTENERS = new CopyOnWriteArrayList<>(); @@ -42,6 +47,19 @@ public static void dispatch(final ServerConfiguration config) { CONFIG_LISTENERS.forEach(listener -> listener.accept(config)); } + public static void addActivationListener(final ActivationListener listener) { + ACTIVATION_LISTENERS.add(listener); + } + + public static void removeActivationListener(final ActivationListener listener) { + ACTIVATION_LISTENERS.remove(listener); + } + + /** Signals that application code initialized the Datadog OpenFeature provider. */ + public static void activate() { + ACTIVATION_LISTENERS.forEach(ActivationListener::activate); + } + public static void addExposureListener(final ExposureListener listener) { EXPOSURE_LISTENERS.add(listener); } diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java index 3af2ed5add8..daaaf8d7001 100644 --- a/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java +++ b/products/feature-flagging/feature-flagging-bootstrap/src/test/java/datadog/trace/api/featureflag/FeatureFlaggingGatewayTest.java @@ -13,6 +13,7 @@ class FeatureFlaggingGatewayTest { private FeatureFlaggingGateway.ConfigListener configListener; + private FeatureFlaggingGateway.ActivationListener activationListener; private FeatureFlaggingGateway.ExposureListener exposureListener; private FeatureFlaggingGateway.SpanEnrichmentListener spanEnrichmentListener; private ServerConfiguration firstConfiguration; @@ -23,6 +24,7 @@ class FeatureFlaggingGatewayTest { @BeforeEach void setUp() { configListener = mock(FeatureFlaggingGateway.ConfigListener.class); + activationListener = mock(FeatureFlaggingGateway.ActivationListener.class); exposureListener = mock(FeatureFlaggingGateway.ExposureListener.class); spanEnrichmentListener = mock(FeatureFlaggingGateway.SpanEnrichmentListener.class); firstConfiguration = mock(ServerConfiguration.class); @@ -34,10 +36,21 @@ void setUp() { @AfterEach void tearDown() { FeatureFlaggingGateway.removeConfigListener(configListener); + FeatureFlaggingGateway.removeActivationListener(activationListener); FeatureFlaggingGateway.removeExposureListener(exposureListener); FeatureFlaggingGateway.removeSpanEnrichmentListener(spanEnrichmentListener); } + @Test + void testProviderActivationListener() { + FeatureFlaggingGateway.addActivationListener(activationListener); + + FeatureFlaggingGateway.activate(); + + verify(activationListener).activate(); + verifyNoMoreInteractions(activationListener); + } + @Test void testAttachingAConfigListener() { clearCurrentServerConfiguration(); diff --git a/products/feature-flagging/feature-flagging-config/build.gradle.kts b/products/feature-flagging/feature-flagging-config/build.gradle.kts index 14109d8dfd9..97dd3aa9f0d 100644 --- a/products/feature-flagging/feature-flagging-config/build.gradle.kts +++ b/products/feature-flagging/feature-flagging-config/build.gradle.kts @@ -4,9 +4,8 @@ plugins { apply(from = "$rootDir/gradle/java.gradle") -description = "Feature flagging configuration keys (compile-time constants)" +description = "Feature flagging configuration keys and source resolution" -extra["excludedClassesCoverage"] = listOf( - // Constants-only holder — no executable logic to cover. - "datadog.trace.api.featureflag.config.FeatureFlaggingConfig", -) +dependencies { + testImplementation(libs.bundles.junit5) +} diff --git a/products/feature-flagging/feature-flagging-config/src/main/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfig.java b/products/feature-flagging/feature-flagging-config/src/main/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfig.java index 9d75961e9d4..bae6fda1890 100644 --- a/products/feature-flagging/feature-flagging-config/src/main/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfig.java +++ b/products/feature-flagging/feature-flagging-config/src/main/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfig.java @@ -2,13 +2,91 @@ public class FeatureFlaggingConfig { - public static final String FLAGGING_PROVIDER_ENABLED = "experimental.flagging.provider.enabled"; + public static final String CONFIGURATION_SOURCE_AGENTLESS = "agentless"; + public static final String CONFIGURATION_SOURCE_REMOTE_CONFIG = "remote_config"; + + private static final Resolution DISABLED_RESOLUTION = new Resolution(false, null); + private static final Resolution AGENTLESS_CONFIGURATION = + new Resolution(true, CONFIGURATION_SOURCE_AGENTLESS); + private static final Resolution REMOTE_CONFIG_CONFIGURATION = + new Resolution(true, CONFIGURATION_SOURCE_REMOTE_CONFIG); + + public static final String FEATURE_FLAGS_ENABLED = "feature.flags.enabled"; + public static final String EXPERIMENTAL_FLAGGING_PROVIDER_ENABLED = + "experimental.flagging.provider.enabled"; /** * Opt-in gate for APM span enrichment with feature-flag evaluation metadata. DISTINCT from {@link - * #FLAGGING_PROVIDER_ENABLED} and OFF by default — enabling the provider does not enable span + * #FEATURE_FLAGS_ENABLED} and OFF by default — enabling the provider does not enable span * enrichment. */ public static final String EXPERIMENTAL_SPAN_ENRICHMENT_ENABLED = "experimental.flagging.provider.span.enrichment.enabled"; + + public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE = + "feature.flags.configuration.source"; + public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_BASE_URL = + "feature.flags.configuration.source.agentless.base.url"; + public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_POLL_INTERVAL_SECONDS = + "feature.flags.configuration.source.agentless.poll.interval.seconds"; + public static final String FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_REQUEST_TIMEOUT_SECONDS = + "feature.flags.configuration.source.agentless.request.timeout.seconds"; + + public static Resolution resolveConfiguration( + final Boolean providerEnabled, + final String explicitSource, + final Boolean legacyProviderEnabled) { + final String normalizedSource = normalizeConfigurationSource(explicitSource); + if (Boolean.FALSE.equals(providerEnabled)) { + return new Resolution(false, normalizedSource); + } + if (normalizedSource != null) { + if (CONFIGURATION_SOURCE_AGENTLESS.equals(normalizedSource)) { + return AGENTLESS_CONFIGURATION; + } + if (CONFIGURATION_SOURCE_REMOTE_CONFIG.equals(normalizedSource)) { + return REMOTE_CONFIG_CONFIGURATION; + } + return new Resolution(false, normalizedSource); + } + if (legacyProviderEnabled != null) { + return legacyProviderEnabled ? REMOTE_CONFIG_CONFIGURATION : DISABLED_RESOLUTION; + } + return AGENTLESS_CONFIGURATION; + } + + public static boolean isSupportedConfigurationSource(final String source) { + final String normalizedSource = normalizeConfigurationSource(source); + return normalizedSource == null + || CONFIGURATION_SOURCE_AGENTLESS.equals(normalizedSource) + || CONFIGURATION_SOURCE_REMOTE_CONFIG.equals(normalizedSource); + } + + private static String normalizeConfigurationSource(final String source) { + if (source == null) { + return null; + } + final String normalized = source.trim().toLowerCase(java.util.Locale.ROOT); + return normalized.isEmpty() ? null : normalized; + } + + public static final class Resolution { + private final boolean enabled; + private final String source; + + private Resolution(final boolean enabled, final String source) { + this.enabled = enabled; + this.source = source; + } + + public boolean isEnabled() { + return enabled; + } + + public String getSource() { + return source; + } + } + + private FeatureFlaggingConfig() {} } diff --git a/products/feature-flagging/feature-flagging-config/src/test/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfigTest.java b/products/feature-flagging/feature-flagging-config/src/test/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfigTest.java new file mode 100644 index 00000000000..16283869d71 --- /dev/null +++ b/products/feature-flagging/feature-flagging-config/src/test/java/datadog/trace/api/featureflag/config/FeatureFlaggingConfigTest.java @@ -0,0 +1,55 @@ +package datadog.trace.api.featureflag.config; + +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_AGENTLESS; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.CONFIGURATION_SOURCE_REMOTE_CONFIG; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.isSupportedConfigurationSource; +import static datadog.trace.api.featureflag.config.FeatureFlaggingConfig.resolveConfiguration; +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 org.junit.jupiter.api.Test; + +class FeatureFlaggingConfigTest { + + @Test + void appliesConfigurationPrecedence() { + assertResolution(true, CONFIGURATION_SOURCE_AGENTLESS, null, null, null); + assertResolution(true, CONFIGURATION_SOURCE_AGENTLESS, null, " ", null); + assertResolution(true, CONFIGURATION_SOURCE_REMOTE_CONFIG, null, null, true); + assertResolution(false, null, null, null, false); + assertResolution(true, CONFIGURATION_SOURCE_AGENTLESS, null, "agentless", true); + assertResolution(true, CONFIGURATION_SOURCE_REMOTE_CONFIG, null, " remote_CONFIG ", false); + assertResolution(false, CONFIGURATION_SOURCE_AGENTLESS, false, "agentless", true); + assertResolution(false, "invalid", null, "invalid", null); + assertResolution(false, "offline", null, " OFFLINE ", true); + } + + @Test + void recognizesSupportedExplicitSources() { + assertTrue(isSupportedConfigurationSource(null)); + assertTrue(isSupportedConfigurationSource(" ")); + assertTrue(isSupportedConfigurationSource("agentless")); + assertTrue(isSupportedConfigurationSource(" REMOTE_CONFIG ")); + assertFalse(isSupportedConfigurationSource("invalid")); + assertFalse(isSupportedConfigurationSource("offline")); + } + + private static void assertResolution( + final boolean enabled, + final String source, + final Boolean providerEnabled, + final String explicitSource, + final Boolean legacyProviderEnabled) { + final FeatureFlaggingConfig.Resolution resolution = + resolveConfiguration(providerEnabled, explicitSource, legacyProviderEnabled); + + assertEquals(enabled, resolution.isEnabled()); + if (source == null) { + assertNull(resolution.getSource()); + } else { + assertEquals(source, resolution.getSource()); + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/build.gradle.kts b/products/feature-flagging/feature-flagging-lib/build.gradle.kts index 9d659d6ed63..425e217e822 100644 --- a/products/feature-flagging/feature-flagging-lib/build.gradle.kts +++ b/products/feature-flagging/feature-flagging-lib/build.gradle.kts @@ -18,18 +18,17 @@ dependencies { api(libs.moshi) api(libs.jctools) api(project(":communication")) + implementation(project(":internal-api")) api(project(":products:feature-flagging:feature-flagging-bootstrap")) + implementation(project(":utils:logging-utils")) api(project(":utils:queue-utils")) compileOnly(project(":dd-trace-core")) // shading does not work with this one - // Span-enrichment write tier: TraceInterceptor / GlobalTracer / AgentTracer / AgentSpan. - compileOnly(project(":internal-api")) // Platform JSON writer for the ffe_* tag values. compileOnly(project(":components:json")) testImplementation(libs.bundles.junit5) testImplementation(libs.bundles.mockito) - testImplementation(project(":internal-api")) testImplementation(project(":utils:test-utils")) testImplementation(project(":dd-java-agent:testing")) } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessConfigurationSource.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessConfigurationSource.java new file mode 100644 index 00000000000..f4fc35cce30 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/AgentlessConfigurationSource.java @@ -0,0 +1,489 @@ +package com.datadog.featureflag; + +import static datadog.communication.http.OkHttpUtils.prepareRequest; +import static datadog.communication.http.OkHttpUtils.sendWithRetries; +import static datadog.trace.util.AgentThreadFactory.AgentThread.FEATURE_FLAG_CONFIGURATION_POLLER; +import static datadog.trace.util.Strings.isBlank; + +import datadog.communication.http.HttpRetryPolicy; +import datadog.communication.http.OkHttpUtils; +import datadog.logging.RatelimitedLogger; +import datadog.trace.api.Config; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import datadog.trace.util.AgentThreadFactory; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.HttpURLConnection; +import java.util.HashMap; +import java.util.Map; +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.concurrent.atomic.AtomicReference; +import java.util.function.DoubleSupplier; +import javax.annotation.Nullable; +import okhttp3.Call; +import okhttp3.HttpUrl; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.ResponseBody; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +final class AgentlessConfigurationSource implements ConfigurationSourceService { + private static final Logger LOGGER = LoggerFactory.getLogger(AgentlessConfigurationSource.class); + + private static final String DATADOG_UFC_RULES_BASED_SERVER_PATH = + "/api/v2/feature-flagging/config/rules-based/server"; + private static final int MAX_ATTEMPTS = 3; + private static final int MINUTES_BETWEEN_WARNINGS = 5; + 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 final HttpUrl endpoint; + private final Config config; + private final long pollIntervalMillis; + private final UfcHttpClient client; + private final ScheduledExecutorService executor; + private final RatelimitedLogger ratelimitedLogger; + private final Object lifecycleLock = new Object(); + private final AtomicBoolean polling = new AtomicBoolean(); + private final AtomicReference pollingThread = new AtomicReference<>(); + private volatile boolean closed; + private volatile boolean started; + private volatile ScheduledFuture scheduledPoll; + private volatile String etag; + + AgentlessConfigurationSource(final Config config) { + this(config, endpoint(config)); + } + + private AgentlessConfigurationSource(final Config config, final HttpUrl endpoint) { + this( + endpoint, + config, + millis(config.getFeatureFlaggingConfigurationSourcePollIntervalSeconds()), + new OkHttpUfcHttpClient( + OkHttpUtils.buildHttpClient( + endpoint, + millis(config.getFeatureFlaggingConfigurationSourceRequestTimeoutSeconds())), + millis(config.getFeatureFlaggingConfigurationSourcePollIntervalSeconds()), + TimeUnit.MILLISECONDS::sleep, + () -> ThreadLocalRandom.current().nextDouble(1 - RETRY_JITTER, 1 + RETRY_JITTER)), + Executors.newSingleThreadScheduledExecutor( + new AgentThreadFactory(FEATURE_FLAG_CONFIGURATION_POLLER)), + new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_WARNINGS, TimeUnit.MINUTES)); + } + + AgentlessConfigurationSource( + final HttpUrl endpoint, + final Config config, + final long pollIntervalMillis, + final UfcHttpClient client, + final ScheduledExecutorService executor) { + this( + endpoint, + config, + pollIntervalMillis, + client, + executor, + new RatelimitedLogger(LOGGER, MINUTES_BETWEEN_WARNINGS, TimeUnit.MINUTES)); + } + + AgentlessConfigurationSource( + final HttpUrl endpoint, + final Config config, + final long pollIntervalMillis, + final UfcHttpClient client, + final ScheduledExecutorService executor, + final RatelimitedLogger ratelimitedLogger) { + this.endpoint = endpoint; + this.config = config; + this.pollIntervalMillis = pollIntervalMillis; + this.client = client; + this.executor = executor; + this.ratelimitedLogger = ratelimitedLogger; + } + + @Override + public void init() { + synchronized (lifecycleLock) { + if (closed || started) { + return; + } + started = true; + } + + // Complete the first poll cycle on the activation thread. This lets OpenFeature provider + // initialization observe a successful retry before it checks whether configuration is ready. + // No request occurs before application code activates the provider. + pollOnceSafely(); + + synchronized (lifecycleLock) { + if (!closed) { + scheduledPoll = + executor.scheduleWithFixedDelay( + this::pollOnceSafely, + pollIntervalMillis, + pollIntervalMillis, + TimeUnit.MILLISECONDS); + } + } + } + + boolean pollOnce() { + if (closed || !polling.compareAndSet(false, true)) { + return false; + } + pollingThread.set(Thread.currentThread()); + try { + return fetchAndApply(); + } finally { + pollingThread.compareAndSet(Thread.currentThread(), null); + polling.set(false); + } + } + + @Override + public void close() { + final ScheduledFuture poll; + synchronized (lifecycleLock) { + if (closed) { + return; + } + closed = true; + started = false; + poll = scheduledPoll; + scheduledPoll = null; + } + if (poll != null) { + poll.cancel(true); + } + client.cancel(); + final Thread activePollingThread = pollingThread.get(); + if (activePollingThread != null) { + activePollingThread.interrupt(); + } + executor.shutdownNow(); + } + + private void pollOnceSafely() { + try { + pollOnce(); + } catch (final RuntimeException e) { + LOGGER.debug("Unexpected error while polling Feature Flagging HTTP configuration source", e); + } + } + + private boolean fetchAndApply() { + try { + final UfcHttpResponse response = client.fetch(endpoint, config, etag); + if (closed) { + return false; + } + if (isRetryableStatus(response.status)) { + ratelimitedLogger.warn( + "Feature Flagging agentless endpoint failed after {} attempts with HTTP {}", + MAX_ATTEMPTS, + response.status); + return false; + } + synchronized (lifecycleLock) { + return !closed && apply(response); + } + } catch (final IOException e) { + if (!closed) { + ratelimitedLogger.warn( + "Feature Flagging agentless endpoint request failed after {} attempts", + MAX_ATTEMPTS, + e); + } + return false; + } + } + + private boolean apply(final UfcHttpResponse response) { + if (response.status == HttpURLConnection.HTTP_NOT_MODIFIED) { + return true; + } + if (response.status == HttpURLConnection.HTTP_UNAUTHORIZED + || response.status == HttpURLConnection.HTTP_FORBIDDEN) { + ratelimitedLogger.warn( + "Feature Flagging agentless endpoint returned HTTP {}; verify endpoint authentication", + response.status); + return false; + } + if (response.status != HttpURLConnection.HTTP_OK || response.body == null) { + return false; + } + final ServerConfiguration configuration; + try { + configuration = JsonApiUfcResponseParser.INSTANCE.parse(response.body); + } catch (final IOException | RuntimeException e) { + LOGGER.debug("Feature Flagging HTTP configuration source returned malformed UFC payload", e); + return false; + } + if (configuration == null) { + return false; + } + FeatureFlaggingGateway.dispatch(configuration); + updateEtag(response.etag); + return true; + } + + private static boolean isRetryableStatus(final int status) { + return status == HttpURLConnection.HTTP_CLIENT_TIMEOUT + || status == 429 + || (status >= 500 && status <= 599); + } + + private void updateEtag(final String nextEtag) { + etag = isBlank(nextEtag) ? null : nextEtag; + } + + static HttpUrl endpoint(final Config config) { + final String configuredBaseUrl = config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl(); + return configuredBaseUrl == null + ? datadogApiServerDistributionEndpoint(config) + : endpointFromConfiguredBaseUrl(configuredBaseUrl); + } + + private static HttpUrl endpointFromConfiguredBaseUrl(final String configuredBaseUrl) { + final HttpUrl parsed = HttpUrl.parse(configuredBaseUrl.trim()); + if (parsed == null) { + throw new IllegalArgumentException( + "Invalid Feature Flagging HTTP configuration source URL: " + configuredBaseUrl); + } + if ("/".equals(parsed.encodedPath()) || parsed.encodedPath().isEmpty()) { + return parsed + .newBuilder() + .addPathSegments(DATADOG_UFC_RULES_BASED_SERVER_PATH.substring(1)) + .build(); + } + return parsed; + } + + private static HttpUrl datadogApiServerDistributionEndpoint(final Config config) { + final HttpUrl.Builder endpoint = + new HttpUrl.Builder() + .scheme("https") + .host("ufc-server.ff-cdn." + config.getSite()) + .addPathSegments(DATADOG_UFC_RULES_BASED_SERVER_PATH.substring(1)); + final String env = config.getEnv(); + if (env != null && !env.isEmpty()) { + endpoint.addQueryParameter("dd_env", env); + } + return endpoint.build(); + } + + static long millis(final int seconds) { + return TimeUnit.SECONDS.toMillis(seconds); + } + + 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 Flagging 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)); + } + + interface UfcHttpClient { + UfcHttpResponse fetch(HttpUrl endpoint, Config config, String etag) throws IOException; + + void cancel(); + } + + interface RetrySleeper { + void sleep(long delayMillis) throws InterruptedException; + } + + static final class UfcHttpResponse { + final int status; + @Nullable final String etag; + @Nullable final byte[] body; + + UfcHttpResponse(final int status, @Nullable final String etag, @Nullable final byte[] body) { + this.status = status; + this.etag = etag; + this.body = body; + } + } + + static final class OkHttpUfcHttpClient implements UfcHttpClient { + private final OkHttpClient httpClient; + private final long pollIntervalMillis; + private final RetrySleeper retrySleeper; + private final DoubleSupplier jitter; + private final AtomicBoolean fetching = new AtomicBoolean(); + private final AtomicBoolean cancelled = new AtomicBoolean(); + private final AtomicReference activeCall = new AtomicReference<>(); + + OkHttpUfcHttpClient(final OkHttpClient httpClient) { + this( + httpClient, + TimeUnit.SECONDS.toMillis(30), + TimeUnit.MILLISECONDS::sleep, + () -> ThreadLocalRandom.current().nextDouble(1 - RETRY_JITTER, 1 + RETRY_JITTER)); + } + + OkHttpUfcHttpClient( + final OkHttpClient httpClient, + final long pollIntervalMillis, + final RetrySleeper retrySleeper, + final DoubleSupplier jitter) { + this.httpClient = httpClient; + this.pollIntervalMillis = pollIntervalMillis; + this.retrySleeper = retrySleeper; + this.jitter = jitter; + } + + @Override + public UfcHttpResponse fetch(final HttpUrl endpoint, final Config config, final String etag) + throws IOException { + final Map headers = new HashMap<>(); + if (etag != null) { + headers.put("If-None-Match", etag); + } + // Leave Accept-Encoding unset so OkHttp negotiates gzip and transparently decompresses it. + final Request request = + prepareRequest(endpoint, headers, config, isDatadogManagedEndpoint(endpoint, config)) + .get() + .build(); + if (!fetching.compareAndSet(false, true)) { + throw new IllegalStateException("Feature Flagging HTTP request already in flight"); + } + if (cancelled.get()) { + fetching.set(false); + throw new InterruptedIOException("Feature Flagging HTTP client is closed"); + } + try { + final HttpRetryPolicy.Factory retryPolicyFactory = + new HttpRetryPolicy.Factory(0, 0, 0) { + @Override + public HttpRetryPolicy create() { + return new AgentlessRetryPolicy( + cancelled, pollIntervalMillis, retrySleeper, jitter); + } + }; + final Call.Factory callFactory = + retryRequest -> { + final Call call = httpClient.newCall(retryRequest); + activeCall.set(call); + if (cancelled.get()) { + call.cancel(); + } + return call; + }; + return sendWithRetries( + callFactory, + retryPolicyFactory, + request, + response -> { + final int status = response.code(); + final String responseEtag = response.header("ETag"); + try (ResponseBody responseBody = response.body()) { + final byte[] body = responseBody != null ? responseBody.bytes() : null; + return new UfcHttpResponse(status, responseEtag, body); + } + }); + } finally { + activeCall.set(null); + fetching.set(false); + } + } + + @Override + public void cancel() { + cancelled.set(true); + final Call call = activeCall.get(); + if (call != null) { + call.cancel(); + } + } + + private static boolean isDatadogManagedEndpoint(final HttpUrl endpoint, final Config config) { + return config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl() == null + && endpoint.isHttps() + && endpoint.host().equalsIgnoreCase("ufc-server.ff-cdn." + config.getSite()); + } + } + + @SuppressFBWarnings( + value = "AT_NONATOMIC_OPERATIONS_ON_SHARED_VARIABLE", + justification = + "Each retry policy belongs to one synchronous HTTP request and is confined to one thread") + static final class AgentlessRetryPolicy extends HttpRetryPolicy { + private final AtomicBoolean cancelled; + private final long pollIntervalMillis; + private final RetrySleeper retrySleeper; + private final DoubleSupplier jitter; + private int retriesLeft = MAX_ATTEMPTS - 1; + private int retryAttempt; + + AgentlessRetryPolicy( + final AtomicBoolean cancelled, + final long pollIntervalMillis, + final RetrySleeper retrySleeper, + final DoubleSupplier jitter) { + super(0, 0, 0, false); + this.cancelled = cancelled; + this.pollIntervalMillis = pollIntervalMillis; + this.retrySleeper = retrySleeper; + this.jitter = jitter; + } + + @Override + public boolean shouldRetry(final Exception exception) { + return exception instanceof IOException + && !Thread.currentThread().isInterrupted() + && reserveRetry(); + } + + @Override + public boolean shouldRetry(@Nullable final Response response) { + return response != null && isRetryableStatus(response.code()) && reserveRetry(); + } + + private boolean reserveRetry() { + if (cancelled.get() || retriesLeft == 0) { + return false; + } + retriesLeft--; + retryAttempt++; + return true; + } + + @Override + public void backoff() throws IOException { + if (cancelled.get()) { + throw new InterruptedIOException("Feature Flagging HTTP client is closed"); + } + try { + retrySleeper.sleep( + retryDelayMillis(pollIntervalMillis, retryAttempt, jitter.getAsDouble())); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Feature Flagging retry interrupted"); + } + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigService.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ConfigurationSourceService.java similarity index 60% rename from products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigService.java rename to products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ConfigurationSourceService.java index 5f84a78f7d4..83495545717 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigService.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ConfigurationSourceService.java @@ -2,7 +2,7 @@ import java.io.Closeable; -public interface RemoteConfigService extends Closeable { +public interface ConfigurationSourceService extends Closeable { void init(); diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/JsonApiUfcResponseParser.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/JsonApiUfcResponseParser.java new file mode 100644 index 00000000000..0a47d316075 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/JsonApiUfcResponseParser.java @@ -0,0 +1,92 @@ +package com.datadog.featureflag; + +import com.squareup.moshi.JsonReader; +import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import javax.annotation.Nullable; +import okio.BufferedSource; +import okio.Okio; + +final class JsonApiUfcResponseParser { + + private static final String UNIVERSAL_FLAG_CONFIGURATION_TYPE = "universal-flag-configuration"; + private static final JsonReader.Options RESPONSE_FIELDS = JsonReader.Options.of("data"); + private static final JsonReader.Options DATA_FIELDS = JsonReader.Options.of("type", "attributes"); + + static final JsonApiUfcResponseParser INSTANCE = + new JsonApiUfcResponseParser(UniversalFlagConfigParser.INSTANCE); + + private final UniversalFlagConfigParser ufcParser; + + JsonApiUfcResponseParser(final UniversalFlagConfigParser ufcParser) { + this.ufcParser = ufcParser; + } + + @Nullable + ServerConfiguration parse(final byte[] content) throws IOException { + try (BufferedSource source = Okio.buffer(Okio.source(new ByteArrayInputStream(content)))) { + final JsonReader reader = JsonReader.of(source); + if (reader.peek() != JsonReader.Token.BEGIN_OBJECT) { + reader.skipValue(); + return null; + } + ServerConfiguration configuration = null; + reader.beginObject(); + while (reader.hasNext()) { + if (reader.selectName(RESPONSE_FIELDS) == 0) { + configuration = parseData(reader); + } else { + reader.skipName(); + reader.skipValue(); + } + } + reader.endObject(); + requireEndOfDocument(reader); + return configuration; + } + } + + @Nullable + private ServerConfiguration parseData(final JsonReader reader) throws IOException { + if (reader.peek() != JsonReader.Token.BEGIN_OBJECT) { + reader.skipValue(); + return null; + } + String type = null; + ServerConfiguration configuration = null; + reader.beginObject(); + while (reader.hasNext()) { + switch (reader.selectName(DATA_FIELDS)) { + case 0: + if (reader.peek() == JsonReader.Token.STRING) { + type = reader.nextString(); + } else { + reader.skipValue(); + } + break; + case 1: + configuration = ufcParser.parse(reader); + break; + default: + reader.skipName(); + reader.skipValue(); + } + } + reader.endObject(); + return UNIVERSAL_FLAG_CONFIGURATION_TYPE.equals(type) + ? validConfiguration(configuration) + : null; + } + + @Nullable + private static ServerConfiguration validConfiguration( + @Nullable final ServerConfiguration configuration) { + return configuration != null && configuration.flags != null ? configuration : null; + } + + private static void requireEndOfDocument(final JsonReader reader) throws IOException { + // A strict JsonReader throws if another top-level value follows the parsed document. + reader.peek(); + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java index ec28f684a96..e66a6231990 100644 --- a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java @@ -1,38 +1,18 @@ package com.datadog.featureflag; -import com.squareup.moshi.JsonAdapter; -import com.squareup.moshi.JsonDataException; -import com.squareup.moshi.JsonReader; -import com.squareup.moshi.JsonWriter; -import com.squareup.moshi.Moshi; -import com.squareup.moshi.Types; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.remoteconfig.Capabilities; import datadog.remoteconfig.ConfigurationChangesTypedListener; -import datadog.remoteconfig.ConfigurationDeserializer; import datadog.remoteconfig.ConfigurationPoller; import datadog.remoteconfig.PollingRateHinter; import datadog.remoteconfig.Product; import datadog.trace.api.Config; import datadog.trace.api.featureflag.FeatureFlaggingGateway; -import datadog.trace.api.featureflag.ufc.v1.Flag; import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.lang.annotation.Annotation; -import java.lang.reflect.Type; -import java.time.Instant; -import java.time.format.DateTimeFormatter; -import java.util.Date; -import java.util.HashMap; -import java.util.Map; -import java.util.Set; -import javax.annotation.Nonnull; import javax.annotation.Nullable; -import okio.Okio; public class RemoteConfigServiceImpl - implements RemoteConfigService, ConfigurationChangesTypedListener { + implements ConfigurationSourceService, ConfigurationChangesTypedListener { private final ConfigurationPoller configurationPoller; @@ -43,8 +23,7 @@ public RemoteConfigServiceImpl(final SharedCommunicationObjects sco, final Confi @Override public void init() { configurationPoller.addCapabilities(Capabilities.CAPABILITY_FFE_FLAG_CONFIGURATION_RULES); - configurationPoller.addListener( - Product.FFE_FLAGS, UniversalFlagConfigDeserializer.INSTANCE, this); + configurationPoller.addListener(Product.FFE_FLAGS, UniversalFlagConfigParser.INSTANCE, this); configurationPoller.start(); } @@ -62,102 +41,4 @@ public void accept( final PollingRateHinter pollingRateHinter) { FeatureFlaggingGateway.dispatch(configuration); } - - static class UniversalFlagConfigDeserializer - implements ConfigurationDeserializer { - - static final UniversalFlagConfigDeserializer INSTANCE = new UniversalFlagConfigDeserializer(); - - private static final Moshi MOSHI = - new Moshi.Builder().add(Date.class, new DateAdapter()).add(FlagMapAdapter.FACTORY).build(); - private static final JsonAdapter V1_ADAPTER = - MOSHI.adapter(ServerConfiguration.class); - - @Override - public ServerConfiguration deserialize(final byte[] content) throws IOException { - return V1_ADAPTER.fromJson(Okio.buffer(Okio.source(new ByteArrayInputStream(content)))); - } - } - - static class FlagMapAdapter extends JsonAdapter> { - - private static final Type FLAGS_TYPE = - Types.newParameterizedType(Map.class, String.class, Flag.class); - - static final Factory FACTORY = - new Factory() { - @Nullable - @Override - public JsonAdapter create( - @Nonnull final Type type, - @Nonnull final Set annotations, - @Nonnull final Moshi moshi) { - if (!annotations.isEmpty() || !Types.equals(type, FLAGS_TYPE)) { - return null; - } - return new FlagMapAdapter(moshi.adapter(Flag.class)); - } - }; - - private final JsonAdapter flagAdapter; - - FlagMapAdapter(final JsonAdapter flagAdapter) { - this.flagAdapter = flagAdapter; - } - - @Nullable - @Override - public Map fromJson(@Nonnull final JsonReader reader) throws IOException { - if (reader.peek() == JsonReader.Token.NULL) { - return reader.nextNull(); - } - final Map flags = new HashMap<>(); - reader.beginObject(); - while (reader.hasNext()) { - final String flagKey = reader.nextName(); - final Object rawFlag = reader.readJsonValue(); - try { - final Flag flag = flagAdapter.fromJsonValue(rawFlag); - if (flag != null) { - flags.put(flagKey, flag); - } - } catch (JsonDataException | IllegalArgumentException ignored) { - // A malformed flag must not prevent other flags in the same config from evaluating. - } - } - reader.endObject(); - return flags; - } - - @Override - public void toJson(@Nonnull final JsonWriter writer, @Nullable final Map value) - throws IOException { - throw new UnsupportedOperationException("Reading only adapter"); - } - } - - static class DateAdapter extends JsonAdapter { - - @Nullable - @Override - public Date fromJson(@Nonnull final JsonReader reader) throws IOException { - final String date = reader.nextString(); - if (date == null) { - return null; - } - try { - final Instant instant = DateTimeFormatter.ISO_OFFSET_DATE_TIME.parse(date, Instant::from); - return Date.from(instant); - } catch (Exception e) { - // ignore wrongly set dates - return null; - } - } - - @Override - public void toJson(@Nonnull final JsonWriter writer, @Nullable final Date value) - throws IOException { - throw new UnsupportedOperationException("Reading only adapter"); - } - } } diff --git a/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/UniversalFlagConfigParser.java b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/UniversalFlagConfigParser.java new file mode 100644 index 00000000000..ba5536601f7 --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/UniversalFlagConfigParser.java @@ -0,0 +1,139 @@ +package com.datadog.featureflag; + +import com.squareup.moshi.JsonAdapter; +import com.squareup.moshi.JsonDataException; +import com.squareup.moshi.JsonReader; +import com.squareup.moshi.JsonWriter; +import com.squareup.moshi.Moshi; +import com.squareup.moshi.Types; +import datadog.remoteconfig.ConfigurationDeserializer; +import datadog.trace.api.featureflag.ufc.v1.Flag; +import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.lang.annotation.Annotation; +import java.lang.reflect.Type; +import java.time.Instant; +import java.time.format.DateTimeFormatter; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import okio.BufferedSource; +import okio.Okio; + +final class UniversalFlagConfigParser implements ConfigurationDeserializer { + + static final UniversalFlagConfigParser INSTANCE = new UniversalFlagConfigParser(); + + private static final Moshi MOSHI = + new Moshi.Builder().add(Date.class, new DateAdapter()).add(FlagMapAdapter.FACTORY).build(); + private static final JsonAdapter V1_ADAPTER = + MOSHI.adapter(ServerConfiguration.class); + + private UniversalFlagConfigParser() {} + + @Override + public ServerConfiguration deserialize(final byte[] content) throws IOException { + try (BufferedSource source = Okio.buffer(Okio.source(new ByteArrayInputStream(content)))) { + final JsonReader reader = JsonReader.of(source); + final ServerConfiguration configuration = parse(reader); + requireEndOfDocument(reader); + return configuration; + } + } + + @Nullable + ServerConfiguration parse(final JsonReader reader) throws IOException { + return V1_ADAPTER.fromJson(reader); + } + + private static void requireEndOfDocument(final JsonReader reader) throws IOException { + // A strict JsonReader throws if another top-level value follows the parsed document. + reader.peek(); + } + + static final class FlagMapAdapter extends JsonAdapter> { + + private static final Type FLAGS_TYPE = + Types.newParameterizedType(Map.class, String.class, Flag.class); + + static final Factory FACTORY = + new Factory() { + @Nullable + @Override + public JsonAdapter create( + @Nonnull final Type type, + @Nonnull final Set annotations, + @Nonnull final Moshi moshi) { + if (!annotations.isEmpty() || !Types.equals(type, FLAGS_TYPE)) { + return null; + } + return new FlagMapAdapter(moshi.adapter(Flag.class)); + } + }; + + private final JsonAdapter flagAdapter; + + FlagMapAdapter(final JsonAdapter flagAdapter) { + this.flagAdapter = flagAdapter; + } + + @Nullable + @Override + public Map fromJson(@Nonnull final JsonReader reader) throws IOException { + if (reader.peek() == JsonReader.Token.NULL) { + return reader.nextNull(); + } + final Map flags = new HashMap<>(); + reader.beginObject(); + while (reader.hasNext()) { + final String flagKey = reader.nextName(); + final Object rawFlag = reader.readJsonValue(); + try { + final Flag flag = flagAdapter.fromJsonValue(rawFlag); + if (flag != null) { + flags.put(flagKey, flag); + } + } catch (JsonDataException | IllegalArgumentException ignored) { + // A malformed flag must not prevent other flags in the same config from evaluating. + } + } + reader.endObject(); + return flags; + } + + @Override + public void toJson(@Nonnull final JsonWriter writer, @Nullable final Map value) + throws IOException { + throw new UnsupportedOperationException("Reading only adapter"); + } + } + + static final class DateAdapter extends JsonAdapter { + + @Nullable + @Override + public Date fromJson(@Nonnull final JsonReader reader) throws IOException { + final String date = reader.nextString(); + if (date == null) { + return null; + } + try { + final Instant instant = DateTimeFormatter.ISO_OFFSET_DATE_TIME.parse(date, Instant::from); + return Date.from(instant); + } catch (Exception e) { + // ignore wrongly set dates + return null; + } + } + + @Override + public void toJson(@Nonnull final JsonWriter writer, @Nullable final Date value) + throws IOException { + throw new UnsupportedOperationException("Reading only adapter"); + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessConfigurationSourceTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessConfigurationSourceTest.java new file mode 100644 index 00000000000..b02bd05642b --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/AgentlessConfigurationSourceTest.java @@ -0,0 +1,1363 @@ +package com.datadog.featureflag; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import datadog.communication.http.OkHttpUtils; +import datadog.logging.RatelimitedLogger; +import datadog.trace.agent.test.server.http.JavaTestHttpServer; +import datadog.trace.api.Config; +import datadog.trace.api.featureflag.FeatureFlaggingGateway; +import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.HttpURLConnection; +import java.net.SocketTimeoutException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.zip.GZIPOutputStream; +import okhttp3.Call; +import okhttp3.HttpUrl; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Protocol; +import okhttp3.Response; +import okhttp3.ResponseBody; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class AgentlessConfigurationSourceTest { + private static final String CONFIG_PATH = "/api/v2/feature-flagging/config/rules-based/server"; + + @Mock private FeatureFlaggingGateway.ConfigListener listener; + + @AfterEach + void cleanup() { + FeatureFlaggingGateway.removeConfigListener(listener); + FeatureFlaggingGateway.dispatch((ServerConfiguration) null); + } + + @Test + void derivesDatadogUfcCdnEndpointFromSiteAndEnv() { + final Config config = config("datad0g.com", "staging env"); + + assertEquals( + "https://ufc-server.ff-cdn.datad0g.com/api/v2/feature-flagging/config/rules-based/server?dd_env=staging%20env", + AgentlessConfigurationSource.endpoint(config).toString()); + } + + @Test + void derivesDatadogUfcCdnEndpointWithoutEnv() { + assertEquals( + "https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server", + AgentlessConfigurationSource.endpoint(config("datadoghq.com", "")).toString()); + assertEquals( + "https://ufc-server.ff-cdn.datadoghq.com/api/v2/feature-flagging/config/rules-based/server", + AgentlessConfigurationSource.endpoint(config("datadoghq.com", null)).toString()); + } + + @Test + void appendsRulesBasedServerPathToConfiguredAgentlessBaseUrl() { + final Config config = config(); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()) + .thenReturn("http://mock-backend:8080"); + + assertEquals( + "http://mock-backend:8080/api/v2/feature-flagging/config/rules-based/server", + AgentlessConfigurationSource.endpoint(config).toString()); + } + + @Test + void usesConfiguredAgentlessEndpointWithPathUnchanged() { + final Config config = config(); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()) + .thenReturn("http://mock-backend:8080/custom/ufc?tenant=test"); + + assertEquals( + "http://mock-backend:8080/custom/ufc?tenant=test", + AgentlessConfigurationSource.endpoint(config).toString()); + } + + @Test + void rejectsInvalidConfiguredAgentlessBaseUrl() { + final Config config = config(); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()) + .thenReturn("not a URL"); + + assertThrows( + IllegalArgumentException.class, () -> AgentlessConfigurationSource.endpoint(config)); + } + + @Test + void rejectsInvalidDatadogUfcCdnEndpoint() { + assertThrows( + IllegalArgumentException.class, + () -> AgentlessConfigurationSource.endpoint(config("datadoghq.com:bad", ""))); + } + + @Test + void defaultConstructorBuildsHttpClientFromConfig() { + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource(config("datad0g.com", "staging")); + + service.close(); + } + + @Test + void realHttpClientDoesNotSendApiKeyOverHttp() throws Exception { + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.get( + CONFIG_PATH, + api -> + api.getResponse() + .addHeader("ETag", "etag-b") + .send(emptyConfig()))))) { + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient); + + try { + final AgentlessConfigurationSource.UfcHttpResponse response = + client.fetch(HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)), config(), "etag-a"); + + assertEquals(HttpURLConnection.HTTP_OK, response.status); + assertEquals("etag-b", response.etag); + assertEquals(emptyConfig(), new String(response.body, UTF_8)); + assertNull(server.getLastRequest().getHeader("DD-API-KEY")); + assertEquals("etag-a", server.getLastRequest().getHeader("If-None-Match")); + assertEquals("java", server.getLastRequest().getHeader("Datadog-Meta-Lang")); + assertEquals("gzip", server.getLastRequest().getHeader("Accept-Encoding")); + } finally { + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void sendsApiKeyToDefaultDatadogHttpsEndpoint() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient(requests, delay -> {}, () -> 1.0, response(200, "etag-a", emptyConfig())); + final HttpUrl endpoint = HttpUrl.get("https://ufc-server.ff-cdn.datadoghq.com" + CONFIG_PATH); + + client.fetch(endpoint, config(), null); + + assertEquals("test-api-key", requests.get(0).header("DD-API-KEY")); + } + + @Test + void stripsApiKeyFromCustomHttpsEndpoint() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient(requests, delay -> {}, () -> 1.0, response(200, "etag-a", emptyConfig())); + final Config config = config(); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()) + .thenReturn("https://flags.example.test/custom/ufc"); + final HttpUrl endpoint = HttpUrl.get("https://flags.example.test/custom/ufc"); + + client.fetch(endpoint, config, null); + + assertNull(requests.get(0).header("DD-API-KEY")); + } + + @Test + void stripsApiKeyFromUnexpectedHttpsEndpoint() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient(requests, delay -> {}, () -> 1.0, response(200, "etag-a", emptyConfig())); + final HttpUrl endpoint = HttpUrl.get("https://flags.example.test/custom/ufc"); + + client.fetch(endpoint, config(), null); + + assertNull(requests.get(0).header("DD-API-KEY")); + } + + @Test + void handlesGzipAndKeepsLastKnownGoodWhenNextResponseIsTruncated() throws Exception { + final byte[] compressedConfig = gzip(emptyConfig()); + final byte[] truncatedConfig = Arrays.copyOf(compressedConfig, compressedConfig.length - 8); + final AtomicInteger responses = new AtomicInteger(); + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.get( + CONFIG_PATH, + api -> { + final boolean firstResponse = responses.getAndIncrement() == 0; + final byte[] body = + firstResponse ? compressedConfig : truncatedConfig; + api.getResponse() + .addHeader("Content-Encoding", "gzip") + .addHeader("ETag", firstResponse ? "etag-good" : "etag-bad") + .sendWithType("application/json", body); + })))) { + final HttpUrl endpoint = HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)); + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + endpoint, + config(), + 30_000, + new AgentlessConfigurationSource.OkHttpUfcHttpClient( + httpClient, 30_000, delay -> {}, () -> 1.0), + Executors.newSingleThreadScheduledExecutor()); + final ArgumentCaptor configuration = + ArgumentCaptor.forClass(ServerConfiguration.class); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + assertTrue(service.pollOnce()); + assertFalse(service.pollOnce()); + + verify(listener).accept(configuration.capture()); + assertEquals("Staging", configuration.getValue().environment.name); + assertEquals(4, responses.get()); + assertEquals("gzip", server.getLastRequest().getHeader("Accept-Encoding")); + assertEquals("etag-good", server.getLastRequest().getHeader("If-None-Match")); + } finally { + service.close(); + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void downloadsAndAppliesLargeUfcWithoutPayloadLimit() throws Exception { + final int flagCount = 5_000; + final String largeConfig = largeConfig(flagCount); + assertTrue(largeConfig.getBytes(UTF_8).length > 500_000); + + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> s.handlers(h -> h.get(CONFIG_PATH, api -> api.getResponse().send(largeConfig))))) { + final HttpUrl endpoint = HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)); + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + endpoint, + config(), + 30_000, + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient), + Executors.newSingleThreadScheduledExecutor()); + final ArgumentCaptor configuration = + ArgumentCaptor.forClass(ServerConfiguration.class); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + assertTrue(service.pollOnce()); + + verify(listener).accept(configuration.capture()); + assertEquals(flagCount, configuration.getValue().flags.size()); + } finally { + service.close(); + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void realHttpClientAllowsMissingEtagAndEmptyResponseBody() throws Exception { + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.get( + CONFIG_PATH, + api -> + api.getResponse() + .status(HttpURLConnection.HTTP_NO_CONTENT) + .send())))) { + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient); + + try { + final AgentlessConfigurationSource.UfcHttpResponse response = + client.fetch(HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)), config(), null); + + assertEquals(HttpURLConnection.HTTP_NO_CONTENT, response.status); + assertNull(response.etag); + assertEquals(0, response.body.length); + assertNull(server.getLastRequest().getHeader("If-None-Match")); + } finally { + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void httpClientAdapterPreservesMissingResponseBody() throws Exception { + final OkHttpClient httpClient = mock(OkHttpClient.class); + final Call call = mock(Call.class); + final HttpUrl endpoint = HttpUrl.get("http://localhost"); + final okhttp3.Request request = new okhttp3.Request.Builder().url(endpoint).build(); + final Response okHttpResponse = + new Response.Builder() + .request(request) + .protocol(Protocol.HTTP_1_1) + .code(HttpURLConnection.HTTP_OK) + .message("OK") + .build(); + when(httpClient.newCall(any())).thenReturn(call); + when(call.execute()).thenReturn(okHttpResponse); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient); + + final AgentlessConfigurationSource.UfcHttpResponse response = + client.fetch(endpoint, config(), null); + + assertEquals(HttpURLConnection.HTTP_OK, response.status); + assertNull(response.etag); + assertNull(response.body); + } + + @Test + void realHttpClientCancellationInterruptsInFlightRequest() throws Exception { + final CountDownLatch requestStarted = new CountDownLatch(1); + final CountDownLatch releaseRequest = new CountDownLatch(1); + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.get( + CONFIG_PATH, + api -> { + requestStarted.countDown(); + assertTrue(releaseRequest.await(1, TimeUnit.SECONDS)); + api.getResponse().send(emptyConfig()); + })))) { + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient); + final ExecutorService runner = Executors.newSingleThreadExecutor(); + + try { + final Future response = + runner.submit( + () -> + client.fetch( + HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)), config(), null)); + assertTrue(requestStarted.await(1, TimeUnit.SECONDS)); + assertThrows( + IllegalStateException.class, + () -> + client.fetch( + HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)), config(), null)); + + client.cancel(); + + final ExecutionException failure = + assertThrows(ExecutionException.class, () -> response.get(1, TimeUnit.SECONDS)); + assertInstanceOf(IOException.class, failure.getCause()); + } finally { + releaseRequest.countDown(); + runner.shutdownNow(); + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void realHttpClientCancellationBeforeFetchPreventsRequest() throws Exception { + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> h.get(CONFIG_PATH, api -> api.getResponse().send(emptyConfig()))))) { + final OkHttpClient httpClient = new OkHttpClient.Builder().build(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient(httpClient); + + try { + client.cancel(); + + assertThrows( + IOException.class, + () -> + client.fetch( + HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)), config(), null)); + } finally { + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void realHttpClientTimesOutDelayedResponse() throws Exception { + try (JavaTestHttpServer server = + JavaTestHttpServer.httpServer( + s -> + s.handlers( + h -> + h.get( + CONFIG_PATH, + api -> { + TimeUnit.MILLISECONDS.sleep(500); + api.getResponse().send(emptyConfig()); + })))) { + final HttpUrl endpoint = HttpUrl.get(server.getAddress().resolve(CONFIG_PATH)); + final OkHttpClient httpClient = OkHttpUtils.buildHttpClient(endpoint, 50); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + new AgentlessConfigurationSource.OkHttpUfcHttpClient( + httpClient, 30_000, delay -> {}, () -> 1.0); + + try { + assertThrows(IOException.class, () -> client.fetch(endpoint, config(), null)); + } finally { + httpClient.dispatcher().executorService().shutdownNow(); + httpClient.connectionPool().evictAll(); + } + } + } + + @Test + void appliesAcceptedJsonApiUfcThroughGateway() throws Exception { + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + final ArgumentCaptor configuration = + ArgumentCaptor.forClass(ServerConfiguration.class); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + + verify(listener).accept(configuration.capture()); + assertEquals("2026-07-15T19:57:07.219869778Z", configuration.getValue().createdAt); + assertNull(configuration.getValue().format); + assertEquals("Staging", configuration.getValue().environment.name); + assertTrue(configuration.getValue().flags.isEmpty()); + assertNull(client.requests.get(0).etag); + } + + @Test + void customEndpointRequiresJsonApiAndKeepsLastKnownGoodOnRawUfc() throws Exception { + final FakeClient client = + new FakeClient( + response(200, "etag-good", emptyConfig()), + response(200, "etag-raw", emptyConfigAttributes())); + final Config config = config(); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()) + .thenReturn("http://custom-backend/custom/ufc"); + final AgentlessConfigurationSource service = service(client, config); + final ArgumentCaptor configuration = + ArgumentCaptor.forClass(ServerConfiguration.class); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertFalse(service.pollOnce()); + + verify(listener).accept(configuration.capture()); + assertEquals("Staging", configuration.getValue().environment.name); + assertNull(client.requests.get(0).etag); + assertEquals("etag-good", client.requests.get(1).etag); + } + + @Test + void ignoresBlankEtag() throws Exception { + final FakeClient client = + new FakeClient(response(200, " ", emptyConfig()), response(304, null, null)); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + + assertNull(client.requests.get(1).etag); + verify(listener).accept(any(ServerConfiguration.class)); + } + + @Test + void successfulResponseWithoutEtagClearsPreviousEtag() throws Exception { + final FakeClient client = + new FakeClient( + response(200, "etag-a", emptyConfig()), + response(200, null, emptyConfig()), + response(304, null, null)); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + + assertEquals("etag-a", client.requests.get(1).etag); + assertNull(client.requests.get(2).etag); + verify(listener, times(2)).accept(any(ServerConfiguration.class)); + } + + @Test + void usesEtagAndSkipsDispatchOnUnchangedConfig() throws Exception { + final FakeClient client = + new FakeClient(response(200, "etag-a", emptyConfig()), response(304, null, null)); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + + verify(listener).accept(any(ServerConfiguration.class)); + assertEquals("etag-a", client.requests.get(1).etag); + } + + @Test + void coldNotModifiedDoesNotEstablishEtag() throws Exception { + final FakeClient client = + new FakeClient(response(304, "etag-cold", null), response(200, "etag-warm", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + + assertNull(client.requests.get(1).etag); + verify(listener).accept(any(ServerConfiguration.class)); + } + + @Test + void failedGatewayDispatchDoesNotAdvanceEtag() throws Exception { + final FakeClient client = + new FakeClient( + response(200, "etag-a", emptyConfig()), response(200, "etag-b", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + final FeatureFlaggingGateway.ConfigListener failingListener = + configuration -> { + throw new IllegalStateException("listener rejected configuration"); + }; + FeatureFlaggingGateway.addConfigListener(failingListener); + + try { + assertThrows(IllegalStateException.class, service::pollOnce); + FeatureFlaggingGateway.removeConfigListener(failingListener); + + assertTrue(service.pollOnce()); + + assertNull(client.requests.get(1).etag); + } finally { + FeatureFlaggingGateway.removeConfigListener(failingListener); + } + } + + @Test + void keepsLastKnownGoodOnAuthFailureAndMalformedPayload() throws Exception { + final FakeClient client = + new FakeClient( + response(200, "etag-good", emptyConfig()), + response(401, null, null), + response(200, null, "{not-json}"), + response(200, null, "{\"flags\":[]}")); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + + verify(listener).accept(any(ServerConfiguration.class)); + assertEquals("etag-good", client.requests.get(1).etag); + assertEquals("etag-good", client.requests.get(2).etag); + assertEquals("etag-good", client.requests.get(3).etag); + } + + @Test + void rejectsForbiddenNonOkMissingBodyAndNullConfiguration() throws Exception { + final FakeClient client = + new FakeClient( + response(403, null, null), + response(404, null, null), + response(600, null, null), + response(200, null, null), + response(200, null, "null"), + response(200, null, jsonApiResponse("other-configuration", emptyConfigAttributes())), + response(200, null, "{\"data\":null}"), + response( + 200, null, "{\"data\":{\"id\":\"1\",\"type\":\"universal-flag-configuration\"}}"), + response(200, null, emptyConfigAttributes())); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + + verifyNoInteractions(listener); + } + + @Test + void warnsRateLimitedOnUnauthorizedAndForbidden() throws Exception { + final RatelimitedLogger ratelimitedLogger = mock(RatelimitedLogger.class); + final FakeClient client = new FakeClient(response(401, null, null), response(403, null, null)); + final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 30_000, + client, + executor, + ratelimitedLogger); + + try { + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + + verify(ratelimitedLogger) + .warn( + "Feature Flagging agentless endpoint returned HTTP {}; verify endpoint authentication", + HttpURLConnection.HTTP_UNAUTHORIZED); + verify(ratelimitedLogger) + .warn( + "Feature Flagging agentless endpoint returned HTTP {}; verify endpoint authentication", + HttpURLConnection.HTTP_FORBIDDEN); + } finally { + service.close(); + } + } + + @Test + void retriesTimeoutBeforeApplyingConfig() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + new SocketTimeoutException("slow HTTP configuration source"), + new SocketTimeoutException("slow HTTP configuration source"), + response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + + assertEquals(3, requests.size()); + verify(listener).accept(any(ServerConfiguration.class)); + } + + @Test + void retriesClientTimeoutAndRateLimitStatusBeforeApplyingConfig() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + response(408, null, null), + response(200, "etag-a", emptyConfig()), + response(429, null, null), + response(200, "etag-b", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + + assertEquals(4, requests.size()); + verify(listener, times(2)).accept(any(ServerConfiguration.class)); + } + + @Test + void retriesServerErrorThenKeepsColdStateOnNotModified() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, delay -> {}, () -> 1.0, response(500, null, null), response(304, null, null)); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + + assertEquals(2, requests.size()); + verifyNoInteractions(listener); + } + + @Test + void warnsRateLimitedAfterRetryableFailuresAreExhausted() throws Exception { + final RatelimitedLogger ratelimitedLogger = mock(RatelimitedLogger.class); + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + response(503, null, null), + response(503, null, null), + response(503, null, null)); + final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 30_000, + client, + executor, + ratelimitedLogger); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + assertFalse(service.pollOnce()); + + assertEquals(3, requests.size()); + verify(ratelimitedLogger) + .warn( + "Feature Flagging agentless endpoint failed after {} attempts with HTTP {}", 3, 503); + verifyNoInteractions(listener); + } finally { + service.close(); + } + } + + @Test + void warnsRateLimitedAfterIoFailuresAreExhausted() throws Exception { + final RatelimitedLogger ratelimitedLogger = mock(RatelimitedLogger.class); + final SocketTimeoutException finalFailure = + new SocketTimeoutException("slow HTTP configuration source"); + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + new SocketTimeoutException("slow HTTP configuration source"), + new SocketTimeoutException("slow HTTP configuration source"), + finalFailure); + final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 30_000, + client, + executor, + ratelimitedLogger); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + assertFalse(service.pollOnce()); + + assertEquals(3, requests.size()); + verify(ratelimitedLogger) + .warn( + "Feature Flagging agentless endpoint request failed after {} attempts", + 3, + finalFailure); + verifyNoInteractions(listener); + } finally { + service.close(); + } + } + + @Test + void usesIntervalAwareRetryBackoff() throws Exception { + final List delays = new ArrayList<>(); + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delays::add, + () -> 1.0, + response(503, null, null), + new SocketTimeoutException("slow HTTP configuration source"), + response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + + assertEquals(Arrays.asList(5_000L, 10_000L), delays); + assertEquals(3, requests.size()); + verify(listener).accept(any(ServerConfiguration.class)); + } + + @Test + void retryPolicyRejectsNonIoAndInterruptedIoFailures() { + final AgentlessConfigurationSource.AgentlessRetryPolicy policy = + retryPolicy(new AtomicBoolean()); + + assertFalse(policy.shouldRetry(new IllegalStateException("not an I/O failure"))); + + Thread.currentThread().interrupt(); + try { + assertFalse(policy.shouldRetry(new IOException("interrupted"))); + } finally { + assertTrue(Thread.interrupted()); + } + } + + @Test + void retryPolicyRejectsMissingResponse() { + final AgentlessConfigurationSource.AgentlessRetryPolicy policy = + retryPolicy(new AtomicBoolean()); + + assertFalse(policy.shouldRetry((Response) null)); + } + + @Test + void cancelledRetryPolicyRejectsBackoff() { + final AgentlessConfigurationSource.AgentlessRetryPolicy policy = + retryPolicy(new AtomicBoolean(true)); + + assertThrows(InterruptedIOException.class, policy::backoff); + } + + @Test + void clampsAndJittersRetryBackoff() { + assertEquals(2_000, AgentlessConfigurationSource.retryDelayMillis(1_000, 1, 1.0)); + assertEquals(5_000, AgentlessConfigurationSource.retryDelayMillis(1_000, 2, 1.0)); + assertEquals(10_000, AgentlessConfigurationSource.retryDelayMillis(600_000, 1, 1.0)); + assertEquals(30_000, AgentlessConfigurationSource.retryDelayMillis(600_000, 2, 1.0)); + assertEquals(6_000, AgentlessConfigurationSource.retryDelayMillis(30_000, 1, 1.2)); + assertThrows( + IllegalArgumentException.class, + () -> AgentlessConfigurationSource.retryDelayMillis(30_000, 3, 1.0)); + } + + @Test + void rejectsOverlappingPolls() throws Exception { + final CountDownLatch requestStarted = new CountDownLatch(1); + final CountDownLatch releaseRequest = new CountDownLatch(1); + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + client.block(requestStarted, releaseRequest); + final AgentlessConfigurationSource service = service(client); + final ExecutorService runner = Executors.newFixedThreadPool(2); + + try { + final Future first = runner.submit(service::pollOnce); + assertTrue(requestStarted.await(1, TimeUnit.SECONDS)); + final Future second = runner.submit(service::pollOnce); + + assertFalse(second.get(1, TimeUnit.SECONDS)); + releaseRequest.countDown(); + assertTrue(first.get(1, TimeUnit.SECONDS)); + assertEquals(1, client.calls.get()); + } finally { + releaseRequest.countDown(); + runner.shutdownNow(); + } + } + + @Test + void initCompletesFirstPollAndCloseCancelsScheduledFuture() throws Exception { + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 60_000, + client, + Executors.newSingleThreadScheduledExecutor()); + FeatureFlaggingGateway.addConfigListener(listener); + + service.init(); + assertEquals(1, client.calls.get()); + service.close(); + + verify(listener).accept(any(ServerConfiguration.class)); + } + + @Test + void initCompletesInitialRetryCycleBeforeReturning() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + response(500, null, null), + response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 60_000, + client, + Executors.newSingleThreadScheduledExecutor()); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + service.init(); + + assertEquals(2, requests.size()); + verify(listener).accept(any(ServerConfiguration.class)); + } finally { + service.close(); + } + } + + @Test + void repeatedInitStartsOnlyOnePoller() throws Exception { + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + + service.init(); + service.init(); + awaitCalls(client, 1); + service.close(); + + assertEquals(1, client.calls.get()); + } + + @Test + void scheduledPollContinuesAfterListenerRuntimeException() throws Exception { + final FakeClient client = + new FakeClient( + response(200, "etag-a", emptyConfig()), response(200, "etag-b", emptyConfig())); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 10, + client, + Executors.newSingleThreadScheduledExecutor()); + final AtomicInteger listenerCalls = new AtomicInteger(); + // Wait on the listener rather than on FakeClient.calls: the call counter is incremented when + // a request starts, so it reaches 2 before the second configuration has been applied. + final CountDownLatch listenerNotified = new CountDownLatch(2); + final FeatureFlaggingGateway.ConfigListener flakyListener = + configuration -> { + final int call = listenerCalls.incrementAndGet(); + listenerNotified.countDown(); + if (call == 1) { + throw new IllegalStateException("listener rejected first configuration"); + } + }; + FeatureFlaggingGateway.addConfigListener(flakyListener); + + try { + service.init(); + assertTrue(listenerNotified.await(5, TimeUnit.SECONDS)); + + assertEquals(2, listenerCalls.get()); + assertNull(client.requests.get(1).etag); + } finally { + service.close(); + FeatureFlaggingGateway.removeConfigListener(flakyListener); + } + } + + @Test + void closeCancelsInFlightRequestAndIgnoresLateSuccess() throws Exception { + final CountDownLatch requestStarted = new CountDownLatch(1); + final CountDownLatch releaseRequest = new CountDownLatch(1); + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + client.block(requestStarted, releaseRequest); + final AgentlessConfigurationSource service = service(client); + final ExecutorService runner = Executors.newSingleThreadExecutor(); + FeatureFlaggingGateway.addConfigListener(listener); + + try { + final Future poll = runner.submit(service::pollOnce); + assertTrue(requestStarted.await(1, TimeUnit.SECONDS)); + + service.close(); + + assertFalse(poll.get(1, TimeUnit.SECONDS)); + assertEquals(1, client.cancelCalls.get()); + assertEquals(1, client.calls.get()); + verifyNoInteractions(listener); + } finally { + releaseRequest.countDown(); + runner.shutdownNow(); + } + } + + @Test + void closeDuringIoFailurePreventsRetry() throws Exception { + final CountDownLatch requestStarted = new CountDownLatch(1); + final CountDownLatch releaseRequest = new CountDownLatch(1); + final FakeClient client = + new FakeClient(new SocketTimeoutException("slow HTTP configuration source")); + client.block(requestStarted, releaseRequest); + final AgentlessConfigurationSource service = service(client); + final ExecutorService runner = Executors.newSingleThreadExecutor(); + + try { + final Future poll = runner.submit(service::pollOnce); + assertTrue(requestStarted.await(1, TimeUnit.SECONDS)); + + service.close(); + + assertFalse(poll.get(1, TimeUnit.SECONDS)); + assertEquals(1, client.calls.get()); + } finally { + releaseRequest.countDown(); + runner.shutdownNow(); + } + } + + @Test + void closeInterruptsRetryBackoff() throws Exception { + final CountDownLatch backoffStarted = new CountDownLatch(1); + final List requests = new ArrayList<>(); + final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> { + backoffStarted.countDown(); + TimeUnit.MINUTES.sleep(1); + }, + () -> 1.0, + new SocketTimeoutException("slow HTTP configuration source"), + response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = + new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), config(), 30_000, client, executor); + final ExecutorService runner = Executors.newSingleThreadExecutor(); + + try { + final Future initialization = runner.submit(service::init); + assertTrue(backoffStarted.await(1, TimeUnit.SECONDS)); + + service.close(); + + initialization.get(1, TimeUnit.SECONDS); + assertTrue(executor.awaitTermination(1, TimeUnit.SECONDS)); + assertEquals(1, requests.size()); + } finally { + runner.shutdownNow(); + } + } + + @Test + void closePreventsFurtherPolls() throws Exception { + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + + service.close(); + + assertFalse(service.pollOnce()); + assertEquals(0, client.calls.get()); + } + + @Test + void initAfterCloseDoesNotSchedulePoll() throws Exception { + final FakeClient client = new FakeClient(response(200, "etag-a", emptyConfig())); + final AgentlessConfigurationSource service = service(client); + + service.close(); + service.init(); + + assertEquals(0, client.calls.get()); + } + + @Nested + class SystemTestParity { + @Test + void preservesSystemTestSourceTransitionsAndLastKnownGoodState() throws Exception { + final List requests = new ArrayList<>(); + final AgentlessConfigurationSource.OkHttpUfcHttpClient client = + scriptedClient( + requests, + delay -> {}, + () -> 1.0, + response(200, "etag-a", emptyConfig()), + response(304, "etag-must-not-replace-a", null), + response(509, null, null), + response(200, "etag-b", emptyConfig()), + response(200, "etag-c", "{not-json}"), + response(401, null, null)); + final AgentlessConfigurationSource service = service(client); + FeatureFlaggingGateway.addConfigListener(listener); + + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + assertTrue(service.pollOnce()); + assertFalse(service.pollOnce()); + assertFalse(service.pollOnce()); + + verify(listener, times(2)).accept(any(ServerConfiguration.class)); + assertEquals("etag-a", requests.get(1).header("If-None-Match")); + assertEquals("etag-a", requests.get(2).header("If-None-Match")); + assertEquals("etag-a", requests.get(3).header("If-None-Match")); + assertEquals("etag-b", requests.get(4).header("If-None-Match")); + assertEquals("etag-b", requests.get(5).header("If-None-Match")); + } + } + + private static AgentlessConfigurationSource service( + final AgentlessConfigurationSource.UfcHttpClient client) { + return new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config(), + 30_000, + client, + Executors.newSingleThreadScheduledExecutor()); + } + + private static AgentlessConfigurationSource service( + final FakeClient client, final Config config) { + return new AgentlessConfigurationSource( + HttpUrl.get("http://localhost" + CONFIG_PATH), + config, + 30_000, + client, + Executors.newSingleThreadScheduledExecutor()); + } + + private static AgentlessConfigurationSource.OkHttpUfcHttpClient scriptedClient( + final List requests, + final AgentlessConfigurationSource.RetrySleeper retrySleeper, + final java.util.function.DoubleSupplier jitter, + final Object... outcomes) + throws IOException { + final BlockingQueue scriptedOutcomes = new LinkedBlockingQueue<>(); + scriptedOutcomes.addAll(Arrays.asList(outcomes)); + final OkHttpClient httpClient = mock(OkHttpClient.class); + when(httpClient.newCall(any())) + .thenAnswer( + invocation -> { + final okhttp3.Request request = invocation.getArgument(0); + requests.add(request); + final Object outcome = scriptedOutcomes.remove(); + final Call call = mock(Call.class); + when(call.execute()) + .thenAnswer( + ignored -> { + if (outcome instanceof IOException) { + throw (IOException) outcome; + } + return okHttpResponse( + request, (AgentlessConfigurationSource.UfcHttpResponse) outcome); + }); + return call; + }); + return new AgentlessConfigurationSource.OkHttpUfcHttpClient( + httpClient, 30_000, retrySleeper, jitter); + } + + private static AgentlessConfigurationSource.AgentlessRetryPolicy retryPolicy( + final AtomicBoolean cancelled) { + return new AgentlessConfigurationSource.AgentlessRetryPolicy( + cancelled, 30_000, delay -> {}, () -> 1.0); + } + + private static Response okHttpResponse( + final okhttp3.Request request, final AgentlessConfigurationSource.UfcHttpResponse response) { + final Response.Builder builder = + new Response.Builder() + .request(request) + .protocol(Protocol.HTTP_1_1) + .code(response.status) + .message(Integer.toString(response.status)); + if (response.etag != null) { + builder.header("ETag", response.etag); + } + if (response.body != null) { + builder.body(ResponseBody.create(MediaType.get("application/json"), response.body)); + } + return builder.build(); + } + + private static Config config() { + return config("datadoghq.com", ""); + } + + private static Config config(final String site, final String env) { + final Config config = mock(Config.class); + lenient() + .when(config.getFeatureFlaggingConfigurationSourcePollIntervalSeconds()) + .thenReturn(30); + lenient() + .when(config.getFeatureFlaggingConfigurationSourceRequestTimeoutSeconds()) + .thenReturn(5); + lenient().when(config.getFeatureFlaggingConfigurationSourceAgentlessBaseUrl()).thenReturn(null); + lenient().when(config.getApiKey()).thenReturn("test-api-key"); + lenient().when(config.getSite()).thenReturn(site); + lenient().when(config.getEnv()).thenReturn(env); + return config; + } + + private static AgentlessConfigurationSource.UfcHttpResponse response( + final int status, final String etag, final String body) { + return new AgentlessConfigurationSource.UfcHttpResponse( + status, etag, body == null ? null : body.getBytes(UTF_8)); + } + + private static String emptyConfig() { + return jsonApiResponse("universal-flag-configuration", emptyConfigAttributes()); + } + + private static String emptyConfigAttributes() { + return "{" + + "\"createdAt\":\"2026-07-15T19:57:07.219869778Z\"," + + "\"environment\":{\"name\":\"Staging\"}," + + "\"flags\":{}" + + "}"; + } + + private static String jsonApiResponse(final String type, final String attributes) { + return "{\"data\":{" + + "\"id\":\"1\"," + + "\"type\":\"" + + type + + "\"," + + "\"attributes\":" + + attributes + + "}}"; + } + + private static String largeConfig(final int flagCount) { + final StringBuilder json = + new StringBuilder( + "{\"createdAt\":\"2026-07-15T19:57:07.219869778Z\"," + + "\"environment\":{\"name\":\"Large Test\"}," + + "\"flags\":{"); + for (int index = 0; index < flagCount; index++) { + if (index > 0) { + json.append(','); + } + final String flagKey = "large-flag-" + index; + json.append('"') + .append(flagKey) + .append("\":{\"key\":\"") + .append(flagKey) + .append( + "\",\"enabled\":true,\"variationType\":\"STRING\"," + + "\"variations\":{\"on\":{\"key\":\"on\",\"value\":\"on\"}}," + + "\"allocations\":[]}"); + } + return jsonApiResponse("universal-flag-configuration", json.append("}}").toString()); + } + + private static byte[] gzip(final String value) throws IOException { + final ByteArrayOutputStream output = new ByteArrayOutputStream(); + try (GZIPOutputStream gzip = new GZIPOutputStream(output)) { + gzip.write(value.getBytes(UTF_8)); + } + return output.toByteArray(); + } + + private static void awaitCalls(final FakeClient client, final int count) throws Exception { + for (int i = 0; i < 100; i++) { + if (client.calls.get() >= count) { + return; + } + TimeUnit.MILLISECONDS.sleep(10); + } + assertEquals(count, client.calls.get()); + } + + private static final class FakeClient implements AgentlessConfigurationSource.UfcHttpClient { + private final AtomicInteger calls = new AtomicInteger(); + private final AtomicInteger cancelCalls = new AtomicInteger(); + private final List requests = new ArrayList<>(); + private final BlockingQueue responses = new LinkedBlockingQueue<>(); + private CountDownLatch requestStarted; + private CountDownLatch releaseRequest; + + private FakeClient(final Object... responses) { + for (final Object response : responses) { + this.responses.add(response); + } + } + + private void block(final CountDownLatch requestStarted, final CountDownLatch releaseRequest) { + this.requestStarted = requestStarted; + this.releaseRequest = releaseRequest; + } + + @Override + public AgentlessConfigurationSource.UfcHttpResponse fetch( + final HttpUrl endpoint, final Config config, final String etag) throws IOException { + calls.incrementAndGet(); + requests.add(new Request(etag)); + if (requestStarted != null) { + requestStarted.countDown(); + } + if (releaseRequest != null) { + await(releaseRequest); + } + final Object response = responses.remove(); + if (response instanceof IOException) { + throw (IOException) response; + } + if (response instanceof RuntimeException) { + throw (RuntimeException) response; + } + return (AgentlessConfigurationSource.UfcHttpResponse) response; + } + + @Override + public void cancel() { + cancelCalls.incrementAndGet(); + if (releaseRequest != null) { + releaseRequest.countDown(); + } + } + + private static void await(final CountDownLatch latch) throws IOException { + try { + if (!latch.await(1, TimeUnit.SECONDS)) { + throw new SocketTimeoutException("test request did not release"); + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + } + } + + private static final class Request { + private final String etag; + + private Request(final String etag) { + this.etag = etag; + } + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/JsonApiUfcResponseParserTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/JsonApiUfcResponseParserTest.java new file mode 100644 index 00000000000..a31d47889ba --- /dev/null +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/JsonApiUfcResponseParserTest.java @@ -0,0 +1,87 @@ +package com.datadog.featureflag; + +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 datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import java.io.IOException; +import org.junit.jupiter.api.Test; + +class JsonApiUfcResponseParserTest { + + @Test + void parsesJsonApiMembersInAnyOrder() throws Exception { + final ServerConfiguration configuration = + parse( + "{" + + "\"meta\":{\"ignored\":true}," + + "\"data\":{" + + "\"attributes\":" + + emptyConfig() + + ",\"ignored\":true," + + "\"type\":\"universal-flag-configuration\"" + + "}" + + "}"); + + assertNotNull(configuration); + assertEquals("Test", configuration.environment.name); + assertTrue(configuration.flags.isEmpty()); + } + + @Test + void rejectsRawUfc() throws Exception { + assertNull(parse(emptyConfig())); + } + + @Test + void rejectsUnexpectedJsonApiType() throws Exception { + assertNull(parse("{\"data\":{\"type\":\"other-type\",\"attributes\":" + emptyConfig() + "}}")); + } + + @Test + void rejectsNonStringJsonApiType() throws Exception { + assertNull(parse("{\"data\":{\"type\":null,\"attributes\":" + emptyConfig() + "}}")); + } + + @Test + void rejectsConfigurationWithoutFlags() throws Exception { + assertNull( + parse( + "{\"data\":{" + + "\"type\":\"universal-flag-configuration\"," + + "\"attributes\":{\"environment\":{\"name\":\"Test\"}}" + + "}}")); + } + + @Test + void rejectsNonObjectData() throws Exception { + assertNull(parse("{\"data\":[]}")); + } + + @Test + void rejectsTrailingJson() { + assertThrows( + IOException.class, + () -> + parse( + "{\"data\":{\"type\":\"universal-flag-configuration\",\"attributes\":" + + emptyConfig() + + "}}{}")); + } + + private static ServerConfiguration parse(final String json) throws Exception { + return JsonApiUfcResponseParser.INSTANCE.parse(json.getBytes(UTF_8)); + } + + private static String emptyConfig() { + return "{" + + "\"createdAt\":\"2024-04-17T19:40:53.716Z\"," + + "\"environment\":{\"name\":\"Test\"}," + + "\"flags\":{}" + + "}"; + } +} diff --git a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java index d4ff05c75b1..c1f5ef17b87 100644 --- a/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java +++ b/products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java @@ -31,6 +31,7 @@ import datadog.trace.api.featureflag.FeatureFlaggingGateway; import datadog.trace.api.featureflag.ufc.v1.Flag; import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration; +import java.io.IOException; import java.lang.annotation.Annotation; import java.lang.reflect.Type; import java.time.Instant; @@ -137,6 +138,11 @@ void ignoresUnknownTopLevelFields() throws Exception { assertTrue(config.flags.isEmpty()); } + @Test + void rejectsTrailingJson() { + assertThrows(IOException.class, () -> deserialize(emptyConfig() + "{}")); + } + @Test void skipsUnknownOperatorFlagAndKeepsValidFlag() throws Exception { final ServerConfiguration config = @@ -189,14 +195,14 @@ void flagMapAdapterFactoryOnlyCreatesFlagMapAdapterForFlagMapType() { final Type flagsType = Types.newParameterizedType(Map.class, String.class, Flag.class); final JsonAdapter adapter = - RemoteConfigServiceImpl.FlagMapAdapter.FACTORY.create(flagsType, emptySet(), moshi); + UniversalFlagConfigParser.FlagMapAdapter.FACTORY.create(flagsType, emptySet(), moshi); assertNotNull(adapter); - assertTrue(adapter instanceof RemoteConfigServiceImpl.FlagMapAdapter); + assertTrue(adapter instanceof UniversalFlagConfigParser.FlagMapAdapter); assertNull( - RemoteConfigServiceImpl.FlagMapAdapter.FACTORY.create(String.class, emptySet(), moshi)); + UniversalFlagConfigParser.FlagMapAdapter.FACTORY.create(String.class, emptySet(), moshi)); assertNull( - RemoteConfigServiceImpl.FlagMapAdapter.FACTORY.create( + UniversalFlagConfigParser.FlagMapAdapter.FACTORY.create( flagsType, singleton(mock(Annotation.class)), moshi)); } @@ -248,8 +254,8 @@ void skipsNullFlagAndKeepsValidFlag() throws Exception { @Test void flagMapAdapterIsReadOnly() { - final RemoteConfigServiceImpl.FlagMapAdapter adapter = - new RemoteConfigServiceImpl.FlagMapAdapter(moshi().adapter(Flag.class)); + final UniversalFlagConfigParser.FlagMapAdapter adapter = + new UniversalFlagConfigParser.FlagMapAdapter(moshi().adapter(Flag.class)); assertThrows( UnsupportedOperationException.class, @@ -280,7 +286,8 @@ void flagMapAdapterIsReadOnly() { void testDateParsing(final String value, final Long expectedEpochMilli) throws Exception { final JsonReader reader = mock(JsonReader.class); when(reader.nextString()).thenReturn(value); - final RemoteConfigServiceImpl.DateAdapter adapter = new RemoteConfigServiceImpl.DateAdapter(); + final UniversalFlagConfigParser.DateAdapter adapter = + new UniversalFlagConfigParser.DateAdapter(); final Date parsed = adapter.fromJson(reader); if (expectedEpochMilli == null) { @@ -293,7 +300,8 @@ void testDateParsing(final String value, final Long expectedEpochMilli) throws E @Test void testParsingOnlyAdapter() { - final RemoteConfigServiceImpl.DateAdapter adapter = new RemoteConfigServiceImpl.DateAdapter(); + final UniversalFlagConfigParser.DateAdapter adapter = + new UniversalFlagConfigParser.DateAdapter(); assertThrows( UnsupportedOperationException.class, @@ -306,12 +314,11 @@ private ConfigurationDeserializer deserializer() { } private static ServerConfiguration deserialize(final String json) throws Exception { - return RemoteConfigServiceImpl.UniversalFlagConfigDeserializer.INSTANCE.deserialize( - json.getBytes(UTF_8)); + return UniversalFlagConfigParser.INSTANCE.deserialize(json.getBytes(UTF_8)); } private static Moshi moshi() { - return new Moshi.Builder().add(Date.class, new RemoteConfigServiceImpl.DateAdapter()).build(); + return new Moshi.Builder().add(Date.class, new UniversalFlagConfigParser.DateAdapter()).build(); } private static String emptyConfig() {