Full implementation

Author: manuel-alvarez-alvarez

dd-java-agent/agent-featureflag/build.gradle

@@ -7,27 +7,46 @@ plugins {
 apply from: "$rootDir/gradle/java.gradle"
 apply from: "$rootDir/gradle/version.gradle"
 
-ext {
-  minJavaVersionForTests = JavaVersion.VERSION_11
-}
-
 java {
   sourceCompatibility = JavaVersion.VERSION_1_8
   targetCompatibility = JavaVersion.VERSION_1_8
 }
 
+excludedClassesCoverage += [
+  // Exposure POJO classes
+  'com.datadog.featureflag.exposure.Allocation',
+  'com.datadog.featureflag.exposure.ExposureEvent',
+  'com.datadog.featureflag.exposure.ExposuresRequest',
+  'com.datadog.featureflag.exposure.Flag',
+  'com.datadog.featureflag.exposure.Subject',
+  'com.datadog.featureflag.exposure.Variant',
+  // UFC v1 POJO classes
+  'com.datadog.featureflag.ufc.v1.Allocation',
+  'com.datadog.featureflag.ufc.v1.ConditionConfiguration',
+  'com.datadog.featureflag.ufc.v1.ConditionOperator',
+  'com.datadog.featureflag.ufc.v1.Environment',
+  'com.datadog.featureflag.ufc.v1.Flag',
+  'com.datadog.featureflag.ufc.v1.Rule',
+  'com.datadog.featureflag.ufc.v1.ServerConfiguration',
+  'com.datadog.featureflag.ufc.v1.Shard',
+  'com.datadog.featureflag.ufc.v1.ShardRange',
+  'com.datadog.featureflag.ufc.v1.Split',
+  'com.datadog.featureflag.ufc.v1.ValueType',
+  'com.datadog.featureflag.ufc.v1.Variant'
+]
+
 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-trace-api:openfeature')
-  testImplementation(libs.openfeature.sdk)
+  testImplementation project(':dd-java-agent:testing')
 }
 
 tasks.named("shadowJar", ShadowJar) {
@@ -38,9 +57,3 @@ tasks.named("jar", Jar) {
   archiveClassifier = 'unbundled'
 }
 
-[GroovyCompile].each {
-  tasks.withType(it).configureEach {
-    configureCompiler(it, 11, JavaVersion.VERSION_11)
-  }
-}
-

dd-java-agent/agent-featureflag/src/main/java/com/datadog/featureflag/ExposureEvaluatorAdapter.java

@@ -1,13 +1,16 @@
 package com.datadog.featureflag;
 
-import datadog.trace.api.featureflag.FeatureFlagEvaluator.Context;
-import datadog.trace.api.featureflag.FeatureFlagEvaluator.Resolution;
+import com.datadog.featureflag.exposure.ExposureEvent;
 
+/**
+ * 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 {
 
   void init();
 
   void close();
 
-  void write(String flag, Context context, Resolution<?> resolution);
+  void write(ExposureEvent event);
 }

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

@@ -0,0 +1,129 @@
+package com.datadog.featureflag;
+
+import com.datadog.featureflag.exposure.Allocation;
+import com.datadog.featureflag.exposure.ExposureEvent;
+import com.datadog.featureflag.exposure.Flag;
+import com.datadog.featureflag.exposure.Subject;
+import com.datadog.featureflag.exposure.Variant;
+import datadog.trace.api.featureflag.FeatureFlagEvaluator;
+import datadog.trace.core.util.LRUCache;
+import java.util.AbstractMap;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * This class processes evaluations from a {@link FeatureFlagEvaluator} and records the resulting
+ * exposures through a {@link ExposureWriter}. It uses an LRU cache to avoid sending duplicate
+ * exposure events.
+ */
+public class ExposureWriterEvaluatorAdapter implements FeatureFlagEvaluator {
+
+  private static final int DEFAULT_CACHE_CAPACITY = 50_000;
+
+  private final ExposureWriter writer;
+  private final FeatureFlagEvaluator delegate;
+  private final Set<ExposureEvent> cache;
+
+  public ExposureWriterEvaluatorAdapter(
+      final ExposureWriter writer, final FeatureFlagEvaluator delegate) {
+    this(DEFAULT_CACHE_CAPACITY, writer, delegate);
+  }
+
+  public ExposureWriterEvaluatorAdapter(
+      final int cacheCapacity, final ExposureWriter writer, final FeatureFlagEvaluator delegate) {
+    this.writer = writer;
+    this.delegate = delegate;
+    this.cache = Collections.newSetFromMap(new LRUCache<>(cacheCapacity));
+  }
+
+  @Override
+  public Resolution<Boolean> evaluate(
+      final String key, final Boolean defaultValue, final Context context) {
+    final Resolution<Boolean> resolution = delegate.evaluate(key, defaultValue, context);
+    writeResolution(context, resolution);
+    return resolution;
+  }
+
+  @Override
+  public Resolution<Integer> evaluate(
+      final String key, final Integer defaultValue, final Context context) {
+    final Resolution<Integer> resolution = delegate.evaluate(key, defaultValue, context);
+    writeResolution(context, resolution);
+    return resolution;
+  }
+
+  @Override
+  public Resolution<Double> evaluate(
+      final String key, final Double defaultValue, final Context context) {
+    final Resolution<Double> resolution = delegate.evaluate(key, defaultValue, context);
+    writeResolution(context, resolution);
+    return resolution;
+  }
+
+  @Override
+  public Resolution<String> evaluate(
+      final String key, final String defaultValue, final Context context) {
+    final Resolution<String> resolution = delegate.evaluate(key, defaultValue, context);
+    writeResolution(context, resolution);
+    return resolution;
+  }
+
+  @Override
+  public Resolution<Object> evaluate(
+      final String key, final Object defaultValue, final Context context) {
+    final Resolution<Object> resolution = delegate.evaluate(key, defaultValue, context);
+    writeResolution(context, resolution);
+    return resolution;
+  }
+
+  private <E> void writeResolution(final Context context, final Resolution<E> resolution) {
+    final String allocationKey = allocationKey(resolution);
+    final String variantKey = resolution.getVariant();
+    if (allocationKey == null || variantKey == null) {
+      return;
+    }
+    final ExposureEvent event =
+        new ExposureEvent(
+            System.currentTimeMillis(),
+            new Allocation(allocationKey),
+            new Flag(resolution.getFlagKey()),
+            new Variant(variantKey),
+            new Subject(context.getTargetingKey(), flattenContext(context)));
+
+    boolean writeEvent;
+    synchronized (cache) {
+      writeEvent = cache.add(event);
+    }
+    if (writeEvent) {
+      writer.write(event);
+    }
+  }
+
+  private AbstractMap<String, Object> flattenContext(final Context context) {
+    if (context == null) {
+      return null;
+    }
+    final Set<String> keys = context.keySet();
+    final HashMap<String, Object> result = new HashMap<>(keys.size());
+    for (final String key : keys) {
+      final Object value = context.getValue(key);
+      // TODO ignore nested elements for now as they are not supported by the backend
+      if (value instanceof Integer
+          || value instanceof Double
+          || value instanceof Boolean
+          || value instanceof String) {
+        result.put(key, value);
+      } else if (value == null) {
+        result.put(key, null);
+      }
+    }
+    return result;
+  }
+
+  private static String allocationKey(final Resolution<?> resolution) {
+    final Map<String, Object> meta = resolution.getFlagMetadata();
+    return meta == null ? null : (String) meta.get("allocationKey");
+  }
+}

dd-java-agent/agent-featureflag/src/main/java/com/datadog/featureflag/ExposureWriterEvaluatorAdapter.java

@@ -8,12 +8,9 @@
 import com.squareup.moshi.JsonAdapter;
 import com.squareup.moshi.Moshi;
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
-import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.communication.http.OkHttpUtils;
 import datadog.trace.api.Config;
-import datadog.trace.api.featureflag.FeatureFlagEvaluator.Context;
-import datadog.trace.api.featureflag.FeatureFlagEvaluator.Resolution;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -30,7 +27,7 @@
 
 public class ExposureWriterImpl implements ExposureWriter {
 
-  private static final String EXPOSURES_API_PATH = "/api/v2/exposures";
+  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";
 
@@ -43,13 +40,13 @@ public ExposureWriterImpl(
       final int capacity,
       final long flushInterval,
       final TimeUnit timeUnit,
-      final SharedCommunicationObjects sco,
-      Config config) {
+      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(
-            sco.agentUrl.toString()
+            agentUrl.toString()
                 + DDAgentFeaturesDiscovery.V2_EVP_PROXY_ENDPOINT
                 + EXPOSURES_API_PATH);
     final Map<String, String> context = new HashMap<>();
@@ -71,11 +68,13 @@ public void init() {
   }
 
   @Override
-  public void close() {}
+  public void close() {
+    this.serializerThread.interrupt();
+  }
 
   @Override
-  public void write(final String flag, final Context context, final Resolution<?> resolution) {
-    final long timestamp = System.currentTimeMillis();
+  public void write(final ExposureEvent event) {
+    queue.offer(event);
   }
 
   private static class ExposureSerializingHandler implements Runnable {
@@ -122,8 +121,8 @@ public void run() {
         Thread.currentThread().interrupt();
       }
       log.debug(
-          "exposure processor worker exited. submitting explosures stopped. unsubmitted evals left: "
-              + !queuesAreEmpty());
+          "exposure processor worker exited. submitting exposures stopped. unsubmitted exposures left: {}",
+          !queuesAreEmpty());
     }
 
     private void runDutyCycle() throws InterruptedException {