perf(feature-flagging): bound exposure delivery memory

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -28,7 +28,13 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake, boolean responseCompression) {
-    HttpRetryPolicy.Factory retryPolicyFactory = new HttpRetryPolicy.Factory(5, 100, 2.0, true);
+    return createBackendApi(
+        intake, responseCompression, new HttpRetryPolicy.Factory(5, 100, 2.0, true));
+  }
+
+  /** Creates a backend API with the retry policy required by the calling product. */
+  public @Nullable BackendApi createBackendApi(
+      Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) {
 
     if (intake.isAgentlessEnabled(config)) {
       HttpUrl agentlessUrl = HttpUrl.get(intake.getAgentlessUrl(config));

products/feature-flagging/feature-flagging-lib/src/main/java/com/datadog/featureflag/ExposurePayloads.java

@@ -0,0 +1,169 @@
+package com.datadog.featureflag;
+
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.Moshi;
+import com.squareup.moshi.Types;
+import datadog.trace.api.featureflag.exposure.ExposureEvent;
+import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+final class ExposurePayloads {
+
+  private static final byte[] PAYLOAD_SUFFIX = FeatureFlagEvpPublisher.utf8Bytes("]}");
+  private static final byte[] JSON_COMMA = FeatureFlagEvpPublisher.utf8Bytes(",");
+
+  private static final JsonAdapter<ExposureEvent> EVENT_JSON_ADAPTER;
+  private static final JsonAdapter<Map<String, String>> CONTEXT_JSON_ADAPTER;
+
+  static {
+    final Moshi moshi = new Moshi.Builder().build();
+    EVENT_JSON_ADAPTER = moshi.adapter(ExposureEvent.class);
+    final Type contextType = Types.newParameterizedType(Map.class, String.class, String.class);
+    CONTEXT_JSON_ADAPTER = moshi.adapter(contextType);
+  }
+
+  private ExposurePayloads() {}
+
+  static EncodedPayloads buildPayloadsForTest(
+      final List<ExposureEvent> events,
+      final Map<String, String> context,
+      final int payloadSizeLimitBytes) {
+    final List<EncodedPayload> payloads = new ArrayList<>();
+    final EncodingResult result =
+        writePayloads(events, context, payloadSizeLimitBytes, payloads::add);
+    return new EncodedPayloads(payloads, result.droppedPayloadLimit, result.droppedSerialization);
+  }
+
+  static EncodingResult writePayloads(
+      final List<ExposureEvent> events,
+      final Map<String, String> context,
+      final int payloadSizeLimitBytes,
+      final Consumer<EncodedPayload> payloadConsumer) {
+    final byte[] prefix = payloadPrefix(context);
+    EncodedPayloadBuilder current = new EncodedPayloadBuilder(prefix);
+    long droppedPayloadLimit = 0;
+    long droppedSerialization = 0;
+
+    for (final ExposureEvent event : events) {
+      final byte[] eventBytes;
+      try {
+        eventBytes = encodeEvent(event);
+      } catch (final RuntimeException ignored) {
+        droppedSerialization++;
+        continue;
+      }
+      if (!current.canAdd(eventBytes, payloadSizeLimitBytes) && !current.isEmpty()) {
+        payloadConsumer.accept(current.toPayload());
+        current = new EncodedPayloadBuilder(prefix);
+      }
+      if (current.canAdd(eventBytes, payloadSizeLimitBytes)) {
+        current.add(eventBytes);
+      } else {
+        droppedPayloadLimit++;
+      }
+    }
+
+    if (!current.isEmpty()) {
+      payloadConsumer.accept(current.toPayload());
+    }
+    return new EncodingResult(droppedPayloadLimit, droppedSerialization);
+  }
+
+  private static byte[] payloadPrefix(final Map<String, String> context) {
+    return FeatureFlagEvpPublisher.utf8Bytes(
+        "{\"context\":" + CONTEXT_JSON_ADAPTER.toJson(context) + ",\"exposures\":[");
+  }
+
+  private static byte[] encodeEvent(final ExposureEvent event) {
+    return FeatureFlagEvpPublisher.utf8Bytes(EVENT_JSON_ADAPTER.toJson(event));
+  }
+
+  static final class EncodedPayloads {
+    final List<EncodedPayload> payloads;
+    final long droppedPayloadLimit;
+    final long droppedSerialization;
+
+    private EncodedPayloads(
+        final List<EncodedPayload> payloads,
+        final long droppedPayloadLimit,
+        final long droppedSerialization) {
+      this.payloads = payloads;
+      this.droppedPayloadLimit = droppedPayloadLimit;
+      this.droppedSerialization = droppedSerialization;
+    }
+  }
+
+  static final class EncodingResult {
+    final long droppedPayloadLimit;
+    final long droppedSerialization;
+
+    private EncodingResult(final long droppedPayloadLimit, final long droppedSerialization) {
+      this.droppedPayloadLimit = droppedPayloadLimit;
+      this.droppedSerialization = droppedSerialization;
+    }
+  }
+
+  static final class EncodedPayload {
+    final byte[] body;
+    final int eventCount;
+
+    private EncodedPayload(final byte[] body, final int eventCount) {
+      this.body = body;
+      this.eventCount = eventCount;
+    }
+  }
+
+  private static final class EncodedPayloadBuilder {
+    private final byte[] prefix;
+    private final List<byte[]> events = new ArrayList<>();
+    private int eventBytes;
+
+    private EncodedPayloadBuilder(final byte[] prefix) {
+      this.prefix = prefix;
+    }
+
+    private boolean isEmpty() {
+      return events.isEmpty();
+    }
+
+    private boolean canAdd(final byte[] event, final int payloadSizeLimitBytes) {
+      return sizeWith(event) <= payloadSizeLimitBytes;
+    }
+
+    private long sizeWith(final byte[] event) {
+      return (long) prefix.length
+          + PAYLOAD_SUFFIX.length
+          + eventBytes
+          + event.length
+          + events.size();
+    }
+
+    private void add(final byte[] event) {
+      events.add(event);
+      eventBytes += event.length;
+    }
+
+    private EncodedPayload toPayload() {
+      final int size =
+          prefix.length + PAYLOAD_SUFFIX.length + eventBytes + Math.max(0, events.size() - 1);
+      final byte[] body = new byte[size];
+      int offset = 0;
+      System.arraycopy(prefix, 0, body, offset, prefix.length);
+      offset += prefix.length;
+      for (int index = 0; index < events.size(); index++) {
+        if (index > 0) {
+          System.arraycopy(JSON_COMMA, 0, body, offset, JSON_COMMA.length);
+          offset += JSON_COMMA.length;
+        }
+        final byte[] event = events.get(index);
+        System.arraycopy(event, 0, body, offset, event.length);
+        offset += event.length;
+      }
+      System.arraycopy(PAYLOAD_SUFFIX, 0, body, offset, PAYLOAD_SUFFIX.length);
+      return new EncodedPayload(body, events.size());
+    }
+  }
+}

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

@@ -8,16 +8,20 @@
 import datadog.common.queue.MessagePassingBlockingQueue;
 import datadog.common.queue.Queues;
 import datadog.communication.BackendApiFactory;
+import datadog.communication.EvpProxy;
 import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.communication.http.HttpRetryPolicy;
 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.internal.VisibleForTesting;
+import datadog.trace.api.telemetry.CoreMetricCollector;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -27,11 +31,25 @@ public class ExposureWriterImpl implements ExposureWriter {
   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;
+  static final int MAX_BATCH_EVENTS = 1_000;
+  static final int EXPOSURE_PAYLOAD_SIZE_LIMIT_BYTES = EvpProxy.PAYLOAD_SIZE_LIMIT_BYTES;
+  static final String EXPOSURE_DROPPED_METRIC = "exposures.events.dropped";
+  static final String DROP_REASON_QUEUE_OVERFLOW = "queue_overflow";
+  static final String DROP_REASON_PAYLOAD_LIMIT = "payload_limit";
+  static final String DROP_REASON_SERIALIZATION = "serialization";
+  static final String DROP_REASON_DELIVERY_FAILURE = "delivery_failure";
   private static final String EXPOSURES_ROUTE = "exposures";
+  private static final CoreMetricCollector CORE_METRICS = CoreMetricCollector.getInstance();
 
   private final MessagePassingBlockingQueue<ExposureEvent> queue;
+  private final AtomicLong droppedQueueOverflow = new AtomicLong();
+  private final ExposureSerializingHandler serializer;
   private final Thread serializerThread;
 
+  private static void countDropped(final long value, final String reason) {
+    CORE_METRICS.count(EXPOSURE_DROPPED_METRIC, value, "reason:" + reason);
+  }
+
   public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config config) {
     this(DEFAULT_CAPACITY, DEFAULT_FLUSH_INTERVAL_IN_SECONDS, SECONDS, sco, config);
   }
@@ -43,13 +61,14 @@ public ExposureWriterImpl(final SharedCommunicationObjects sco, final Config con
       final SharedCommunicationObjects sco,
       final Config config) {
     this.queue = Queues.mpscBlockingConsumerArrayQueue(capacity);
-    final ExposureSerializingHandler serializer =
+    this.serializer =
         new ExposureSerializingHandler(
             new BackendApiFactory(config, sco),
             queue,
             flushInterval,
             timeUnit,
             FeatureFlagEvpContext.from(config),
+            droppedQueueOverflow,
             this::close);
     this.serializerThread = newAgentThread(FEATURE_FLAG_EXPOSURE_PROCESSOR, serializer);
   }
@@ -70,7 +89,9 @@ public void close() {
 
   @Override
   public void accept(final ExposureEvent event) {
-    queue.offer(event);
+    if (!queue.offer(event)) {
+      droppedQueueOverflow.incrementAndGet();
+    }
   }
 
   @VisibleForTesting
@@ -83,6 +104,16 @@ int queueSize() {
     return queue.size();
   }
 
+  @VisibleForTesting
+  long droppedQueueOverflow() {
+    return droppedQueueOverflow.get();
+  }
+
+  @VisibleForTesting
+  void flushForTest() {
+    serializer.flushIfNecessary();
+  }
+
   private static class ExposureSerializingHandler implements Runnable {
     private final MessagePassingBlockingQueue<ExposureEvent> queue;
     private final long ticksRequiredToFlush;
@@ -92,7 +123,8 @@ private static class ExposureSerializingHandler implements Runnable {
     private final Map<String, String> context;
     private final ExposureCache cache;
 
-    private final List<ExposureEvent> buffer = new ArrayList<>();
+    private final List<ExposureEvent> buffer = new ArrayList<>(MAX_BATCH_EVENTS);
+    private final AtomicLong droppedQueueOverflow;
     private final Runnable errorCallback;
 
     public ExposureSerializingHandler(
@@ -101,11 +133,15 @@ public ExposureSerializingHandler(
         final long flushInterval,
         final TimeUnit timeUnit,
         final Map<String, String> context,
+        final AtomicLong droppedQueueOverflow,
         final Runnable errorCallback) {
       this.queue = queue;
       this.cache = new LRUExposureCache(queue.capacity());
-      this.evpPublisher = new FeatureFlagEvpPublisher<>(backendApiFactory, ExposuresRequest.class);
+      this.evpPublisher =
+          new FeatureFlagEvpPublisher<>(
+              backendApiFactory, ExposuresRequest.class, true, HttpRetryPolicy.Factory.NEVER_RETRY);
       this.context = context;
+      this.droppedQueueOverflow = droppedQueueOverflow;
 
       this.lastTicks = System.nanoTime();
       this.ticksRequiredToFlush = timeUnit.toNanos(flushInterval);
@@ -144,7 +180,8 @@ private void runDutyCycle() throws InterruptedException {
     }
 
     private void consumeBatch() {
-      queue.drain(this::addToBuffer, queue.size());
+      final int remainingCapacity = MAX_BATCH_EVENTS - buffer.size();
+      queue.drain(this::addToBuffer, Math.min(queue.size(), remainingCapacity));
     }
 
     /** Adds an element to the buffer taking care of duplicated exposures thanks to the LRU cache */
@@ -157,32 +194,61 @@ private boolean addToBuffer(final ExposureEvent event) {
     }
 
     protected void flushIfNecessary() {
+      reportQueueDrops();
       if (buffer.isEmpty()) {
         return;
       }
       if (shouldFlush()) {
-        final byte[] payload;
+        final ExposurePayloads.EncodingResult result;
         try {
-          final ExposuresRequest exposures = new ExposuresRequest(this.context, this.buffer);
-          payload = evpPublisher.serialize(exposures);
-        } catch (RuntimeException e) {
-          LOGGER.error(EXCLUDE_TELEMETRY, "Could not serialize exposures; dropping batch", e);
-          this.buffer.clear();
-          return;
+          result =
+              ExposurePayloads.writePayloads(
+                  buffer, context, EXPOSURE_PAYLOAD_SIZE_LIMIT_BYTES, this::submitPayload);
+        } finally {
+          buffer.clear();
         }
-        try {
-          evpPublisher.post(EXPOSURES_ROUTE, payload);
-          this.buffer.clear();
-        } catch (Exception e) {
-          LOGGER.debug("Could not submit exposures", e);
+        if (result.droppedSerialization > 0) {
+          countDropped(result.droppedSerialization, DROP_REASON_SERIALIZATION);
+          LOGGER.error(
+              EXCLUDE_TELEMETRY,
+              "Could not serialize {} exposure event(s); dropping events",
+              result.droppedSerialization);
+        }
+        if (result.droppedPayloadLimit > 0) {
+          countDropped(result.droppedPayloadLimit, DROP_REASON_PAYLOAD_LIMIT);
+          LOGGER.warn(
+              "Exposure payload limit dropped {} event(s) (best-effort telemetry)",
+              result.droppedPayloadLimit);
         }
       }
     }
 
+    private void submitPayload(final ExposurePayloads.EncodedPayload payload) {
+      try {
+        evpPublisher.post(EXPOSURES_ROUTE, payload.body);
+      } catch (Exception e) {
+        countDropped(payload.eventCount, DROP_REASON_DELIVERY_FAILURE);
+        LOGGER.debug("Could not submit exposures; dropping attempted batch", e);
+      }
+    }
+
+    private void reportQueueDrops() {
+      final long dropped = droppedQueueOverflow.getAndSet(0);
+      if (dropped > 0) {
+        countDropped(dropped, DROP_REASON_QUEUE_OVERFLOW);
+        LOGGER.warn(
+            "Exposure queue full - dropped {} event(s) under backpressure"
+                + " (best-effort telemetry)",
+            dropped);
+      }
+    }
+
     private boolean shouldFlush() {
       long nanoTime = System.nanoTime();
       long ticks = nanoTime - lastTicks;
-      if (ticks > ticksRequiredToFlush || queue.size() >= FLUSH_THRESHOLD) {
+      if (ticks > ticksRequiredToFlush
+          || buffer.size() >= MAX_BATCH_EVENTS
+          || queue.size() >= FLUSH_THRESHOLD) {
         lastTicks = nanoTime;
         return true;
       }