Move Feature Flagging evaluation into the provider runtime

dd-smoke-tests/openfeature/application/build.gradle

@@ -31,8 +31,9 @@ if (hasProperty('featureFlaggingApiJar')) {
 }
 
 dependencies {
-  // OpenFeature SDK is an API dependency of feature-flagging-api but is not
-  // transitively resolved when the jar is passed as a files() dependency.
+  // Maven resolves these provider dependencies from the published POM. This smoke test uses the
+  // provider as a files() dependency, so it must declare them explicitly.
   implementation 'dev.openfeature:sdk:1.20.1'
+  implementation 'com.squareup.moshi:moshi:1.11.0'
   implementation 'org.springframework.boot:spring-boot-starter-web'
 }

dd-smoke-tests/openfeature/build.gradle

@@ -21,7 +21,7 @@ smokeTestApp {
 
 dependencies {
   testImplementation project(':dd-smoke-tests')
-  testImplementation project(':products:feature-flagging:feature-flagging-lib')
+  testImplementation project(':products:feature-flagging:feature-flagging-agent-runtime')
 }
 
 spotless {

products/feature-flagging/feature-flagging-lib/build.gradle.kts → products/feature-flagging/feature-flagging-agent-runtime/build.gradle.kts

@@ -5,13 +5,7 @@ plugins {
 
 apply(from = "$rootDir/gradle/java.gradle")
 
-description = "Feature flagging remote config and exposure handling"
-
-extra["excludedClassesCoverage"] = listOf(
-  // POJOs
-  "com.datadog.featureflag.ExposureCache.Key",
-  "com.datadog.featureflag.ExposureCache.Value"
-)
+description = "Java agent runtime for Feature Flagging configuration and telemetry"
 
 dependencies {
   api(libs.slf4j)
@@ -20,13 +14,10 @@ dependencies {
   api(project(":communication"))
   implementation(project(":internal-api"))
   api(project(":products:feature-flagging:feature-flagging-bootstrap"))
+  implementation(project(":products:feature-flagging:feature-flagging-telemetry"))
   implementation(project(":utils:logging-utils"))
   api(project(":utils:queue-utils"))
 
-  compileOnly(project(":dd-trace-core")) // shading does not work with this one
-  // Platform JSON writer for the ffe_* tag values.
-  compileOnly(project(":components:json"))
-
   testImplementation(libs.bundles.junit5)
   testImplementation(libs.bundles.mockito)
   testImplementation(project(":utils:test-utils"))

products/feature-flagging/feature-flagging-lib/gradle.lockfile → products/feature-flagging/feature-flagging-agent-runtime/gradle.lockfile

@@ -1,7 +1,7 @@
 # This is a Gradle generated file for dependency locking.
 # Manual edits can break the build and are not advised.
 # This file is expected to be part of source control.
-# To regenerate this file, run: ./gradlew :products:feature-flagging:feature-flagging-lib:dependencies --write-locks
+# To regenerate this file, run: ./gradlew :products:feature-flagging:feature-flagging-agent-runtime:dependencies --write-locks
 cafe.cryptography:curve25519-elisabeth:0.1.0=runtimeClasspath,testRuntimeClasspath
 cafe.cryptography:ed25519-elisabeth:0.1.0=runtimeClasspath,testRuntimeClasspath
 ch.qos.logback:logback-classic:1.2.13=testCompileClasspath,testRuntimeClasspath
@@ -92,7 +92,6 @@ org.junit.platform:junit-platform-launcher:1.14.1=testRuntimeClasspath
 org.junit.platform:junit-platform-runner:1.14.1=testRuntimeClasspath
 org.junit.platform:junit-platform-suite-api:1.14.1=testRuntimeClasspath
 org.junit.platform:junit-platform-suite-commons:1.14.1=testRuntimeClasspath
-org.junit:junit-bom:5.14.0=spotbugs
 org.junit:junit-bom:5.14.1=testCompileClasspath,testRuntimeClasspath
 org.mockito:mockito-core:4.4.0=testCompileClasspath,testRuntimeClasspath
 org.mockito:mockito-junit-jupiter:4.4.0=testCompileClasspath,testRuntimeClasspath

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java → products/feature-flagging/feature-flagging-agent-runtime/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java

@@ -11,6 +11,7 @@
 import datadog.communication.BackendApi;
 import datadog.communication.BackendApiFactory;
 import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.openfeature.internal.telemetry.ExposureDeduplicationCache;
 import datadog.trace.api.Config;
 import datadog.trace.api.featureflag.FeatureFlaggingGateway;
 import datadog.trace.api.featureflag.exposure.ExposureEvent;
@@ -105,7 +106,7 @@ private static class ExposureSerializingHandler implements Runnable {
     private BackendApi evp;
 
     private final Map<String, String> context;
-    private final ExposureCache cache;
+    private final ExposureDeduplicationCache cache;
 
     private final List<ExposureEvent> buffer = new ArrayList<>();
     private final Runnable errorCallback;
@@ -118,7 +119,7 @@ public ExposureSerializingHandler(
         final Map<String, String> context,
         final Runnable errorCallback) {
       this.queue = queue;
-      this.cache = new LRUExposureCache(queue.capacity());
+      this.cache = new ExposureDeduplicationCache(queue.capacity());
       this.jsonAdapter = new Moshi.Builder().build().adapter(ExposuresRequest.class);
       this.backendApiFactory = backendApiFactory;
       this.context = context;
@@ -166,7 +167,8 @@ private void consumeBatch() {
 
     /** Adds an element to the buffer taking care of duplicated exposures thanks to the LRU cache */
     private boolean addToBuffer(final ExposureEvent event) {
-      if (cache.add(event)) {
+      if (cache.shouldEmit(
+          event.flag.key, event.subject.id, event.variant.key, event.allocation.key)) {
         buffer.add(event);
         return true;
       }

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java → products/feature-flagging/feature-flagging-agent-runtime/src/main/java/com/datadog/featureflag/RemoteConfigServiceImpl.java

@@ -2,17 +2,19 @@
 
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.remoteconfig.Capabilities;
-import datadog.remoteconfig.ConfigurationChangesTypedListener;
+import datadog.remoteconfig.ConfigurationChangesListener;
 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.FeatureFlaggingRawBridge;
 import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration;
+import java.io.IOException;
 import javax.annotation.Nullable;
 
 public class RemoteConfigServiceImpl
-    implements ConfigurationSourceService, ConfigurationChangesTypedListener<ServerConfiguration> {
+    implements ConfigurationSourceService, ConfigurationChangesListener {
 
   private final ConfigurationPoller configurationPoller;
 
@@ -23,7 +25,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, UniversalFlagConfigParser.INSTANCE, this);
+    configurationPoller.addListener(Product.FFE_FLAGS, this);
     configurationPoller.start();
   }
 
@@ -37,8 +39,12 @@ public void close() {
   @Override
   public void accept(
       final String configKey,
-      @Nullable final ServerConfiguration configuration,
-      final PollingRateHinter pollingRateHinter) {
+      @Nullable final byte[] content,
+      final PollingRateHinter pollingRateHinter)
+      throws IOException {
+    final ServerConfiguration configuration =
+        content == null ? null : UniversalFlagConfigParser.INSTANCE.deserialize(content);
     FeatureFlaggingGateway.dispatch(configuration);
+    FeatureFlaggingRawBridge.dispatchConfiguration(content);
   }
 }

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/SpanEnrichmentInterceptor.java → products/feature-flagging/feature-flagging-agent-runtime/src/main/java/com/datadog/featureflag/SpanEnrichmentInterceptor.java

@@ -1,5 +1,6 @@
 package com.datadog.featureflag;
 
+import datadog.openfeature.internal.telemetry.SpanEnrichmentAccumulator;
 import datadog.trace.api.interceptor.MutableSpan;
 import datadog.trace.api.interceptor.TraceInterceptor;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/SpanEnrichmentStates.java → products/feature-flagging/feature-flagging-agent-runtime/src/main/java/com/datadog/featureflag/SpanEnrichmentStates.java

@@ -1,5 +1,6 @@
 package com.datadog.featureflag;
 
+import datadog.openfeature.internal.telemetry.SpanEnrichmentAccumulator;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import java.util.Map;
 import java.util.WeakHashMap;

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/SpanEnrichmentWriter.java → products/feature-flagging/feature-flagging-agent-runtime/src/main/java/com/datadog/featureflag/SpanEnrichmentWriter.java

@@ -1,5 +1,6 @@
 package com.datadog.featureflag;
 
+import datadog.openfeature.internal.telemetry.SpanEnrichmentAccumulator;
 import datadog.trace.api.GlobalTracer;
 import datadog.trace.api.featureflag.FeatureFlaggingGateway;
 import datadog.trace.api.featureflag.SpanEnrichmentEvent;

products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java → products/feature-flagging/feature-flagging-agent-runtime/src/test/java/com/datadog/featureflag/RemoteConfigServiceImplTest.java

@@ -23,12 +23,12 @@
 import com.squareup.moshi.Types;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.remoteconfig.Capabilities;
-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.FeatureFlaggingRawBridge;
 import datadog.trace.api.featureflag.ufc.v1.Flag;
 import datadog.trace.api.featureflag.ufc.v1.ServerConfiguration;
 import java.io.IOException;
@@ -40,8 +40,6 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.tabletest.junit.TableTest;
@@ -50,11 +48,13 @@
 class RemoteConfigServiceImplTest {
 
   @Mock private FeatureFlaggingGateway.ConfigListener listener;
-  @Captor private ArgumentCaptor<ConfigurationDeserializer> deserializerCaptor;
+  @Mock private FeatureFlaggingRawBridge.ConfigurationListener rawListener;
 
   @AfterEach
   void cleanup() {
     FeatureFlaggingGateway.removeConfigListener(listener);
+    FeatureFlaggingRawBridge.removeConfigurationListener(rawListener);
+    FeatureFlaggingRawBridge.dispatchConfiguration(null);
   }
 
   @Test
@@ -63,17 +63,23 @@ void testNewConfigReceived() throws Exception {
     final SharedCommunicationObjects sco = mock(SharedCommunicationObjects.class);
     when(sco.configurationPoller(any(Config.class))).thenReturn(poller);
     FeatureFlaggingGateway.addConfigListener(listener);
+    FeatureFlaggingRawBridge.addConfigurationListener(rawListener);
     final RemoteConfigServiceImpl service = new RemoteConfigServiceImpl(sco, Config.get());
 
     service.init();
 
     verify(poller).addCapabilities(Capabilities.CAPABILITY_FFE_FLAG_CONFIGURATION_RULES);
-    verify(poller).addListener(eq(Product.FFE_FLAGS), deserializerCaptor.capture(), eq(service));
+    verify(poller).addListener(Product.FFE_FLAGS, service);
 
-    final ServerConfiguration config = deserializer().deserialize(emptyConfig().getBytes(UTF_8));
-    service.accept("test", config, mock(PollingRateHinter.class));
+    final byte[] content = emptyConfig().getBytes(UTF_8);
+    service.accept("test", content, mock(PollingRateHinter.class));
 
     verify(listener).accept(any(ServerConfiguration.class));
+    verify(rawListener).accept(eq(content));
+
+    service.accept("test", null, mock(PollingRateHinter.class));
+    verify(listener).accept(null);
+    verify(rawListener).accept(null);
 
     service.close();
 
@@ -308,11 +314,6 @@ void testParsingOnlyAdapter() {
         () -> adapter.toJson(mock(JsonWriter.class), new Date()));
   }
 
-  @SuppressWarnings("unchecked")
-  private ConfigurationDeserializer<ServerConfiguration> deserializer() {
-    return deserializerCaptor.getValue();
-  }
-
   private static ServerConfiguration deserialize(final String json) throws Exception {
     return UniversalFlagConfigParser.INSTANCE.deserialize(json.getBytes(UTF_8));
   }

products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/SpanEnrichmentInterceptorTest.java → products/feature-flagging/feature-flagging-agent-runtime/src/test/java/com/datadog/featureflag/SpanEnrichmentInterceptorTest.java

@@ -10,6 +10,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import datadog.openfeature.internal.telemetry.SpanEnrichmentAccumulator;
 import datadog.trace.api.interceptor.MutableSpan;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import java.util.ArrayList;
@@ -88,7 +89,7 @@ void partialFlushExcludingRootPreservesState() {
     verify(root, never()).setTag(anyString(), anyString());
     final SpanEnrichmentAccumulator surviving = states.peek(root);
     assertNotNull(surviving, "partial flush must NOT remove the accumulator");
-    assertTrue(surviving.serialIdsView().contains(100) && surviving.serialIdsView().contains(108));
+    assertEquals("ZAg=", surviving.toSpanTags().get(SpanEnrichmentAccumulator.TAG_FLAGS_ENC));
 
     // more evaluations, then the FINAL flush with the root present
     states.getOrCreate(root).addSerialId(128);

products/feature-flagging/feature-flagging-lib/src/test/java/com/datadog/featureflag/SpanEnrichmentWriterTest.java → products/feature-flagging/feature-flagging-agent-runtime/src/test/java/com/datadog/featureflag/SpanEnrichmentWriterTest.java

@@ -11,6 +11,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import datadog.openfeature.internal.telemetry.SpanEnrichmentAccumulator;
 import datadog.trace.api.featureflag.SpanEnrichmentEvent;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import java.util.Collections;
@@ -41,8 +42,10 @@ void serialIdWithDoLogAndTargetingKeyRecordsSerialAndSubject() {
 
     final SpanEnrichmentAccumulator state = writer.states().peek(root);
     assertNotNull(state);
-    assertTrue(state.serialIdsView().contains(42));
-    assertEquals(1, state.subjectCount(), "doLog=true + targeting key => subject recorded");
+    assertEquals("Kg==", state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_FLAGS_ENC));
+    assertNotNull(
+        state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_SUBJECTS_ENC),
+        "doLog=true + targeting key => subject recorded");
   }
 
   @Test
@@ -52,8 +55,10 @@ void serialIdWithoutDoLogRecordsNoSubject() {
     writer.accept(SpanEnrichmentEvent.serialId(7, false, "user-1"));
 
     final SpanEnrichmentAccumulator state = writer.states().peek(root);
-    assertTrue(state.serialIdsView().contains(7));
-    assertEquals(0, state.subjectCount(), "doLog=false must not record a subject");
+    assertEquals("Bw==", state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_FLAGS_ENC));
+    assertNull(
+        state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_SUBJECTS_ENC),
+        "doLog=false must not record a subject");
   }
 
   @Test
@@ -63,8 +68,10 @@ void serialIdWithNullTargetingKeyRecordsNoSubject() {
     writer.accept(SpanEnrichmentEvent.serialId(9, true, null));
 
     final SpanEnrichmentAccumulator state = writer.states().peek(root);
-    assertTrue(state.serialIdsView().contains(9));
-    assertEquals(0, state.subjectCount(), "no targeting key => no subject");
+    assertEquals("CQ==", state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_FLAGS_ENC));
+    assertNull(
+        state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_SUBJECTS_ENC),
+        "no targeting key => no subject");
   }
 
   @Test
@@ -75,7 +82,7 @@ void runtimeDefaultRecordsDefault() {
 
     final SpanEnrichmentAccumulator state = writer.states().peek(root);
     assertNotNull(state);
-    assertEquals(1, state.defaultCount());
+    assertNotNull(state.toSpanTags().get(SpanEnrichmentAccumulator.TAG_RUNTIME_DEFAULTS));
   }
 
   @Test

products/feature-flagging/feature-flagging-agent/build.gradle.kts

@@ -15,7 +15,7 @@ description = "Feature flagging agent system"
 
 dependencies {
   api(libs.slf4j)
-  api(project(":products:feature-flagging:feature-flagging-lib"))
+  api(project(":products:feature-flagging:feature-flagging-agent-runtime"))
   api(project(":internal-api"))
   compileOnly(project(":products:feature-flagging:feature-flagging-config"))
 

products/feature-flagging/feature-flagging-agent/src/main/java/com/datadog/featureflag/FeatureFlaggingSystem.java

@@ -70,10 +70,6 @@ private static synchronized void activateAgentless(
 
   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);
 
@@ -119,7 +115,8 @@ static ConfigurationSourceService createConfigurationSourceService(
       return new RemoteConfigServiceImpl(sco, config);
     }
     if (CONFIGURATION_SOURCE_AGENTLESS.equals(configurationSource)) {
-      return new AgentlessConfigurationSource(config);
+      // CDN delivery belongs to the application provider. The agent owns telemetry only.
+      return null;
     }
     return null;
   }

products/feature-flagging/feature-flagging-agent/src/test/java/com/datadog/featureflag/FeatureFlaggingSystemTest.java

@@ -21,7 +21,7 @@
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.remoteconfig.Capabilities;
-import datadog.remoteconfig.ConfigurationDeserializer;
+import datadog.remoteconfig.ConfigurationChangesListener;
 import datadog.remoteconfig.ConfigurationPoller;
 import datadog.remoteconfig.Product;
 import datadog.trace.api.Config;
@@ -99,7 +99,7 @@ void testFeatureFlagSystemInitialization() {
     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).addListener(eq(Product.FFE_FLAGS), any(ConfigurationChangesListener.class));
     verify(poller).start();
 
     FeatureFlaggingSystem.stop();
@@ -128,9 +128,8 @@ void testThatRemoteConfigIsRequired() {
   @Test
   @WithConfig(key = FEATURE_FLAGS_CONFIGURATION_SOURCE, value = "agentless")
   @WithConfig(key = REMOTE_CONFIGURATION_ENABLED, value = "false")
-  void agentlessConfigurationSourceUsesHttpServiceWithoutRemoteConfig() {
-    assertInstanceOf(
-        AgentlessConfigurationSource.class,
+  void agentlessConfigurationSourceIsProviderOwned() {
+    assertNull(
         FeatureFlaggingSystem.createConfigurationSourceService(
             sharedCommunicationObjects(), Config.get()));
   }