Implementation of the open feature SDK in the java tracer

.github/CODEOWNERS

@@ -139,3 +139,9 @@ dd-trace-core/src/test/groovy/datadog/trace/llmobs/  @DataDog/ml-observability
 /internal-api/src/test/groovy/datadog/trace/api/rum/ @DataDog/rum
 /telemetry/src/main/java/datadog/telemetry/rum/      @DataDog/rum
 /telemetry/src/test/groovy/datadog/telemetry/rum/    @DataDog/rum
+
+# @DataDog/feature-flagging-and-experimentation-sdk
+/dd-java-agent/agent-feature-flagging/                      @DataDog/feature-flagging-and-experimentation-sdk
+/internal-api/src/main/java/datadog/trace/api/featureflag   @DataDog/feature-flagging-and-experimentation-sdk
+/internal-api/src/test/groovy/datadog/trace/api/featureflag @DataDog/feature-flagging-and-experimentation-sdk
+/products/openfeature/                                      @DataDog/feature-flagging-and-experimentation-sdk

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java

@@ -32,6 +32,7 @@
 import datadog.trace.api.config.CrashTrackingConfig;
 import datadog.trace.api.config.CwsConfig;
 import datadog.trace.api.config.DebuggerConfig;
+import datadog.trace.api.config.FeatureFlaggingConfig;
 import datadog.trace.api.config.GeneralConfig;
 import datadog.trace.api.config.IastConfig;
 import datadog.trace.api.config.JmxFetchConfig;
@@ -125,7 +126,8 @@ private enum AgentFeature {
     DATA_JOBS(GeneralConfig.DATA_JOBS_ENABLED, false),
     AGENTLESS_LOG_SUBMISSION(GeneralConfig.AGENTLESS_LOG_SUBMISSION_ENABLED, false),
     LLMOBS(LlmObsConfig.LLMOBS_ENABLED, false),
-    LLMOBS_AGENTLESS(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED, false);
+    LLMOBS_AGENTLESS(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED, false),
+    FEATURE_FLAGGING(FeatureFlaggingConfig.FLAGGING_PROVIDER_ENABLED, false);
 
     private final String configKey;
     private final String systemProp;
@@ -184,6 +186,7 @@ public boolean isEnabledByDefault() {
   private static boolean codeOriginEnabled = false;
   private static boolean distributedDebuggerEnabled = false;
   private static boolean agentlessLogSubmissionEnabled = false;
+  private static boolean featureFlaggingEnabled = false;
 
   private static void safelySetContextClassLoader(ClassLoader classLoader) {
     try {
@@ -268,6 +271,7 @@ public static void start(
     codeOriginEnabled = isFeatureEnabled(AgentFeature.CODE_ORIGIN);
     agentlessLogSubmissionEnabled = isFeatureEnabled(AgentFeature.AGENTLESS_LOG_SUBMISSION);
     llmObsEnabled = isFeatureEnabled(AgentFeature.LLMOBS);
+    featureFlaggingEnabled = isFeatureEnabled(AgentFeature.FEATURE_FLAGGING);
 
     // setup writers when llmobs is enabled to accomodate apm and llmobs
     if (llmObsEnabled) {
@@ -662,6 +666,7 @@ public void execute() {
       maybeStartDebugger(instrumentation, scoClass, sco);
       maybeStartRemoteConfig(scoClass, sco);
       maybeStartAiGuard();
+      maybeStartFeatureFlagging(scoClass, sco);
 
       if (telemetryEnabled) {
         startTelemetry(instrumentation, scoClass, sco);
@@ -1083,6 +1088,23 @@ private static void maybeStartLLMObs(Instrumentation inst, Class<?> scoClass, Ob
     }
   }
 
+  private static void maybeStartFeatureFlagging(final Class<?> scoClass, final Object sco) {
+    if (featureFlaggingEnabled) {
+      StaticEventLogger.begin("Feature Flagging");
+
+      try {
+        final Class<?> ffSysClass =
+            AGENT_CLASSLOADER.loadClass("com.datadog.featureflag.FeatureFlaggingSystem");
+        final Method ffSysMethod = ffSysClass.getMethod("start", scoClass);
+        ffSysMethod.invoke(null, sco);
+      } catch (final Throwable e) {
+        log.warn("Not starting Feature Flagging subsystem", e);
+      }
+
+      StaticEventLogger.end("Feature Flagging");
+    }
+  }
+
   private static void maybeInstallLogsIntake(Class<?> scoClass, Object sco) {
     if (agentlessLogSubmissionEnabled) {
       StaticEventLogger.begin("Logs Intake");

dd-java-agent/agent-feature-flagging/build.gradle

@@ -0,0 +1,42 @@
+import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+
+plugins {
+  id 'com.gradleup.shadow'
+}
+
+apply from: "$rootDir/gradle/java.gradle"
+apply from: "$rootDir/gradle/version.gradle"
+
+java {
+  sourceCompatibility = JavaVersion.VERSION_1_8
+  targetCompatibility = JavaVersion.VERSION_1_8
+}
+
+excludedClassesCoverage += [
+  // POJOs
+  'com.datadog.featureflag.ExposureCache.Key',
+  'com.datadog.featureflag.ExposureCache.Value'
+]
+
+dependencies {
+  api libs.slf4j
+  implementation libs.moshi
+  implementation libs.jctools
+
+  api project(':dd-trace-api')
+  compileOnly project(':dd-trace-core')
+  implementation project(':internal-api')
+  implementation project(':communication')
+
+  testImplementation project(':utils:test-utils')
+  testImplementation project(':dd-java-agent:testing')
+}
+
+tasks.named("shadowJar", ShadowJar) {
+  dependencies deps.excludeShared
+}
+
+tasks.named("jar", Jar) {
+  archiveClassifier = 'unbundled'
+}
+

dd-java-agent/agent-feature-flagging/src/main/java/com/datadog/featureflag/ExposureCache.java

@@ -0,0 +1,61 @@
+package com.datadog.featureflag;
+
+import datadog.trace.api.featureflag.exposure.ExposureEvent;
+import java.util.Objects;
+
+public interface ExposureCache {
+
+  boolean add(ExposureEvent event);
+
+  Value get(Key key);
+
+  int size();
+
+  final class Key {
+    public final String flag;
+    public final String subject;
+
+    public Key(final ExposureEvent event) {
+      this.flag = event.flag == null ? null : event.flag.key;
+      this.subject = event.subject == null ? null : event.subject.id;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      final Key key = (Key) o;
+      return Objects.equals(flag, key.flag) && Objects.equals(subject, key.subject);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(flag, subject);
+    }
+  }
+
+  final class Value {
+    public final String variant;
+    public final String allocation;
+
+    public Value(final ExposureEvent event) {
+      this.variant = event.variant == null ? null : event.variant.key;
+      this.allocation = event.allocation == null ? null : event.allocation.key;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      final Value value = (Value) o;
+      return Objects.equals(variant, value.variant) && Objects.equals(allocation, value.allocation);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(variant, allocation);
+    }
+  }
+}

dd-java-agent/agent-feature-flagging/src/main/java/com/datadog/featureflag/ExposureWriter.java

@@ -0,0 +1,14 @@
+package com.datadog.featureflag;
+
+import datadog.trace.api.featureflag.FeatureFlaggingGateway;
+
+/**
+ * Defines an exposure writer responsible for sending exposure events to the EVP proxy.
+ * Implementations should use a background thread to perform these operations asynchronously.
+ */
+public interface ExposureWriter extends AutoCloseable, FeatureFlaggingGateway.ExposureListener {
+
+  void init();
+
+  void close();
+}

dd-java-agent/agent-feature-flagging/src/main/java/com/datadog/featureflag/ExposureWriterImpl.java

@@ -0,0 +1,203 @@
+package com.datadog.featureflag;
+
+import static datadog.trace.util.AgentThreadFactory.AgentThread.FEATURE_FLAG_EXPOSURE_PROCESSOR;
+import static datadog.trace.util.AgentThreadFactory.newAgentThread;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.Moshi;
+import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
+import datadog.communication.http.HttpRetryPolicy;
+import datadog.communication.http.OkHttpUtils;
+import datadog.trace.api.Config;
+import datadog.trace.api.featureflag.FeatureFlaggingGateway;
+import datadog.trace.api.featureflag.exposure.ExposureEvent;
+import datadog.trace.api.featureflag.exposure.ExposuresRequest;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import okhttp3.Headers;
+import okhttp3.HttpUrl;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import org.jctools.queues.MpscBlockingConsumerArrayQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExposureWriterImpl implements ExposureWriter {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(ExposureWriterImpl.class);
+  private static final int DEFAULT_CAPACITY = 1 << 16; // 65536 elements
+  private static final int DEFAULT_FLUSH_INTERVAL_IN_SECONDS = 1;
+  private static final int FLUSH_THRESHOLD = 100;
+  private static final String EXPOSURES_API_PATH = "api/v2/exposures";
+  private static final String EVP_SUBDOMAIN_HEADER_NAME = "X-Datadog-EVP-Subdomain";
+  private static final String EVP_SUBDOMAIN_HEADER_VALUE = "event-platform-intake";
+  private static final HttpRetryPolicy.Factory RETRY_POLICY =
+      new HttpRetryPolicy.Factory(5, 100, 2.0, true);
+
+  private final MpscBlockingConsumerArrayQueue<ExposureEvent> queue;
+  private final Thread serializerThread;
+
+  public ExposureWriterImpl(final HttpUrl agentUrl, final Config config) {
+    this(DEFAULT_CAPACITY, DEFAULT_FLUSH_INTERVAL_IN_SECONDS, SECONDS, agentUrl, config);
+  }
+
+  ExposureWriterImpl(
+      final int capacity,
+      final long flushInterval,
+      final TimeUnit timeUnit,
+      final HttpUrl agentUrl,
+      final Config config) {
+    this.queue = new MpscBlockingConsumerArrayQueue<>(capacity);
+    final Headers headers = Headers.of(EVP_SUBDOMAIN_HEADER_NAME, EVP_SUBDOMAIN_HEADER_VALUE);
+    final HttpUrl url =
+        HttpUrl.get(
+            agentUrl.toString()
+                + DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT
+                + EXPOSURES_API_PATH);
+    final Map<String, String> context = new HashMap<>(4);
+    context.put("service", config.getServiceName() == null ? "unknown" : config.getServiceName());
+    if (config.getEnv() != null) {
+      context.put("env", config.getEnv());
+    }
+    if (config.getVersion() != null) {
+      context.put("version", config.getVersion());
+    }
+    final ExposureSerializingHandler serializer =
+        new ExposureSerializingHandler(queue, flushInterval, timeUnit, url, headers, context);
+    this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer);
+  }
+
+  @Override
+  public void init() {
+    FeatureFlaggingGateway.addExposureListener(this);
+    this.serializerThread.start();
+  }
+
+  @Override
+  public void close() {
+    FeatureFlaggingGateway.removeExposureListener(this);
+    this.serializerThread.interrupt();
+  }
+
+  @Override
+  public void accept(final ExposureEvent event) {
+    queue.offer(event);
+  }
+
+  private static class ExposureSerializingHandler implements Runnable {
+    private final MpscBlockingConsumerArrayQueue<ExposureEvent> queue;
+    private final long ticksRequiredToFlush;
+    private long lastTicks;
+
+    private final JsonAdapter<ExposuresRequest> jsonAdapter;
+    private final OkHttpClient httpClient;
+    private final HttpUrl submissionUrl;
+    private final Headers headers;
+
+    private final Map<String, String> context;
+    private final ExposureCache cache;
+
+    private final List<ExposureEvent> buffer = new ArrayList<>();
+
+    public ExposureSerializingHandler(
+        final MpscBlockingConsumerArrayQueue<ExposureEvent> queue,
+        final long flushInterval,
+        final TimeUnit timeUnit,
+        final HttpUrl submissionUrl,
+        final Headers headers,
+        final Map<String, String> context) {
+      this.queue = queue;
+      this.cache = new LRUExposureCache(queue.capacity());
+      this.jsonAdapter = new Moshi.Builder().build().adapter(ExposuresRequest.class);
+      this.httpClient = new OkHttpClient();
+      this.submissionUrl = submissionUrl;
+      this.headers = headers;
+      this.context = context;
+
+      this.lastTicks = System.nanoTime();
+      this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval);
+
+      LOGGER.debug("starting exposure serializer, url={}", submissionUrl);
+    }
+
+    @Override
+    public void run() {
+      try {
+        runDutyCycle();
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
+      LOGGER.debug(
+          "exposure processor worker exited. submitting exposures stopped. unsubmitted exposures left: {}",
+          !queue.isEmpty());
+    }
+
+    private void runDutyCycle() throws InterruptedException {
+      final Thread thread = Thread.currentThread();
+      while (!thread.isInterrupted()) {
+        ExposureEvent event;
+        while ((event = queue.poll(100, TimeUnit.MILLISECONDS)) != null) {
+          if (addToBuffer(event)) {
+            consumeBatch();
+            break;
+          }
+        }
+        flushIfNecessary();
+      }
+    }
+
+    private void consumeBatch() {
+      queue.drain(this::addToBuffer, queue.size());
+    }
+
+    /** 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)) {
+        buffer.add(event);
+        return true;
+      }
+      return false;
+    }
+
+    protected void flushIfNecessary() {
+      if (buffer.isEmpty()) {
+        return;
+      }
+      if (shouldFlush()) {
+        final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer);
+        final String reqBod = jsonAdapter.toJson(exposures);
+        final RequestBody requestBody =
+            RequestBody.create(okhttp3.MediaType.parse("application/json"), reqBod);
+        final Request request =
+            new Request.Builder().headers(headers).url(submissionUrl).post(requestBody).build();
+        try (okhttp3.Response response =
+            OkHttpUtils.sendWithRetries(httpClient, RETRY_POLICY, request)) {
+          if (response.isSuccessful()) {
+            LOGGER.debug(
+                "successfully flushed exposures request with {} evals", this.buffer.size());
+            this.buffer.clear();
+          } else {
+            LOGGER.error("Could not submit exposures (HTTP code {})", response.code());
+          }
+        } catch (Exception e) {
+          LOGGER.error("Could not submit exposures", e);
+        }
+      }
+    }
+
+    private boolean shouldFlush() {
+      long nanoTime = System.nanoTime();
+      long ticks = nanoTime - lastTicks;
+      if (ticks > ticksRequiredToFlush || queue.size() >= FLUSH_THRESHOLD) {
+        lastTicks = nanoTime;
+        return true;
+      }
+      return false;
+    }
+  }
+}