Implementation of the open feature SDK in the java tracer

.github/CODEOWNERS

@@ -139,3 +139,10 @@ 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
+/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
+/dd-java-agent/agent-feature-flagging/ @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,192 @@
+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.BackendApi;
+import datadog.communication.BackendApiFactory;
+import datadog.communication.ddagent.SharedCommunicationObjects;
+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 datadog.trace.api.intake.Intake;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+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 final MpscBlockingConsumerArrayQueue<ExposureEvent> queue;
+ private final Thread serializerThread;
+
+ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config config) {
+ this(DEFAULT_CAPACITY, DEFAULT_FLUSH_INTERVAL_IN_SECONDS, SECONDS, sco, config);
+ }
+
+ ExposureWriterImpl(
+ final int capacity,
+ final long flushInterval,
+ final TimeUnit timeUnit,
+ final SharedCommunicationObjects sco,
+ final Config config) {
+ this.queue = new MpscBlockingConsumerArrayQueue<>(capacity);
+ 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(
+ new BackendApiFactory(config, sco),
+ queue,
+ flushInterval,
+ timeUnit,
+ context,
+ this::close);
+ 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);
+ if (this.serializerThread.isAlive()) {
+ 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 BackendApiFactory backendApiFactory;
+ private BackendApi evp;
+
+ private final Map<String, String> context;
+ private final ExposureCache cache;
+
+ private final List<ExposureEvent> buffer = new ArrayList<>();
+ private final Runnable errorCallback;
+
+ public ExposureSerializingHandler(
+ final BackendApiFactory backendApiFactory,
+ final MpscBlockingConsumerArrayQueue<ExposureEvent> queue,
+ final long flushInterval,
+ final TimeUnit timeUnit,
+ final Map<String, String> context,
+ final Runnable errorCallback) {
+ this.queue = queue;
+ this.cache = new LRUExposureCache(queue.capacity());
+ this.jsonAdapter = new Moshi.Builder().build().adapter(ExposuresRequest.class);
+ this.backendApiFactory = backendApiFactory;
+ this.context = context;
+
+ this.lastTicks = System.nanoTime();
+ this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval);
+
+ this.errorCallback = errorCallback;
+
+ LOGGER.debug("starting exposure serializer");
+ }
+
+ @Override
+ public void run() {
+ evp = backendApiFactory.createBackendApi(Intake.EVENT_PLATFORM);
+ if (evp == null) {
+ errorCallback.run();
+ throw new IllegalArgumentException("EVP Proxy not available");
+ }
+ try {
+ runDutyCycle();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ LOGGER.debug("exposure processor worker exited. submitting exposures stopped.");
+ }
+
+ 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()) {
+ try {
+ 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);
+ evp.post("exposures", requestBody, stream -> null, null, false);
+ this.buffer.clear();
+ } 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;
+ }
+ }
+}