Update modules

Author: manuel-alvarez-alvarez

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

@@ -12,29 +12,6 @@ java {
   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

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

@@ -1,16 +1,14 @@
 package com.datadog.featureflag;
 
-import com.datadog.featureflag.exposure.ExposureEvent;
+import datadog.trace.api.featureflag.FeatureFlagGateway;
 
 /**
  * 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 {
+public interface ExposureWriter extends AutoCloseable, FeatureFlagGateway.ExposureListener {
 
   void init();
 
   void close();
-
-  void write(ExposureEvent event);
 }

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

@@ -1,20 +1,25 @@
 package com.datadog.featureflag;
 
-import static datadog.trace.util.AgentThreadFactory.AgentThread.LLMOBS_EVALS_PROCESSOR;
+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.datadog.featureflag.exposure.ExposureEvent;
-import com.datadog.featureflag.exposure.ExposuresRequest;
 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.FeatureFlagGateway;
+import datadog.trace.api.featureflag.exposure.ExposureEvent;
+import datadog.trace.api.featureflag.exposure.ExposuresRequest;
+import datadog.trace.core.util.LRUCache;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import okhttp3.Headers;
 import okhttp3.HttpUrl;
@@ -27,16 +32,24 @@
 
 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 = 50;
   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 Logger log = LoggerFactory.getLogger(ExposureWriterImpl.class);
+  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(
+  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,
@@ -59,27 +72,27 @@ public ExposureWriterImpl(
     }
     final ExposureSerializingHandler serializer =
         new ExposureSerializingHandler(queue, flushInterval, timeUnit, url, headers, context);
-    this.serializerThread = newAgentThread(LLMOBS_EVALS_PROCESSOR, serializer);
+    this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer);
   }
 
   @Override
   public void init() {
+    FeatureFlagGateway.addExposureListener(this);
     this.serializerThread.start();
   }
 
   @Override
   public void close() {
+    FeatureFlagGateway.removeExposureListener(this);
     this.serializerThread.interrupt();
   }
 
   @Override
-  public void write(final ExposureEvent event) {
+  public void accept(final ExposureEvent event) {
     queue.offer(event);
   }
 
   private static class ExposureSerializingHandler implements Runnable {
-    private static final int FLUSH_THRESHOLD = 50;
-
     private final MpscBlockingConsumerArrayQueue<ExposureEvent> queue;
     private final long ticksRequiredToFlush;
     private long lastTicks;
@@ -90,6 +103,7 @@ private static class ExposureSerializingHandler implements Runnable {
     private final Headers headers;
 
     private final Map<String, String> context;
+    private final Set<ExposureEvent> cache;
 
     private final List<ExposureEvent> buffer = new ArrayList<>();
 
@@ -101,6 +115,7 @@ public ExposureSerializingHandler(
         final Headers headers,
         final Map<String, String> context) {
       this.queue = queue;
+      this.cache = Collections.newSetFromMap(new LRUCache<>(queue.capacity()));
       this.jsonAdapter = new Moshi.Builder().build().adapter(ExposuresRequest.class);
       this.httpClient = new OkHttpClient();
       this.submissionUrl = submissionUrl;
@@ -110,7 +125,7 @@ public ExposureSerializingHandler(
       this.lastTicks = System.nanoTime();
       this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval);
 
-      log.debug("starting exposure serializer, url={}", submissionUrl);
+      LOGGER.debug("starting exposure serializer, url={}", submissionUrl);
     }
 
     @Override
@@ -120,25 +135,36 @@ public void run() {
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
       }
-      log.debug(
+      LOGGER.debug(
           "exposure processor worker exited. submitting exposures stopped. unsubmitted exposures left: {}",
-          !queuesAreEmpty());
+          !queue.isEmpty());
     }
 
     private void runDutyCycle() throws InterruptedException {
       final Thread thread = Thread.currentThread();
       while (!thread.isInterrupted()) {
-        final ExposureEvent event = queue.poll(100, TimeUnit.MILLISECONDS);
-        if (event != null) {
-          buffer.add(event);
-          consumeBatch();
+        ExposureEvent event;
+        while ((event = queue.poll(100, TimeUnit.MILLISECONDS)) != null) {
+          if (addToBuffer(event)) {
+            consumeBatch();
+            break;
+          }
         }
         flushIfNecessary();
       }
     }
 
     private void consumeBatch() {
-      queue.drain(buffer::add, queue.size());
+      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(ExposureEvent event) {
+      if (cache.add(event)) {
+        buffer.add(event);
+        return true;
+      }
+      return false;
     }
 
     protected void flushIfNecessary() {
@@ -147,27 +173,22 @@ protected void flushIfNecessary() {
       }
       if (shouldFlush()) {
         final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer);
-        final HttpRetryPolicy.Factory retryPolicyFactory =
-            new HttpRetryPolicy.Factory(5, 100, 2.0, true);
         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, retryPolicyFactory, request)) {
-
+            OkHttpUtils.sendWithRetries(httpClient, RETRY_POLICY, request)) {
           if (response.isSuccessful()) {
-            log.debug("successfully flushed exposures request with {} evals", this.buffer.size());
+            LOGGER.debug(
+                "successfully flushed exposures request with {} evals", this.buffer.size());
             this.buffer.clear();
           } else {
-            log.error(
-                "Could not submit exposures (HTTP code {}) {}",
-                response.code(),
-                response.body() != null ? response.body().string() : "");
+            LOGGER.error("Could not submit exposures (HTTP code {})", response.code());
           }
         } catch (Exception e) {
-          log.error("Could not submit exposures", e);
+          LOGGER.error("Could not submit exposures", e);
         }
       }
     }
@@ -181,9 +202,5 @@ private boolean shouldFlush() {
       }
       return false;
     }
-
-    protected boolean queuesAreEmpty() {
-      return queue.isEmpty();
-    }
   }
 }