[FFM-10816] - Remove metrics flush on map overflow (#183)

* [FFM-10816] - Remove metrics flush on map overflow

What
When too many evaluations are processed and with enough targets the internal metrics map we use will hit its limit very quickly. This causes the metrics to be flushed to the server.

Why
Flushing metrics is unbound and can result in more pressure on the server than necessary

Testing
Manual + unit

Author: andybharness

settings.gradle

@@ -4,7 +4,7 @@ dependencyResolutionManagement {
     versionCatalogs {
         libs {
             // main sdk version
-            version('sdk', '1.5.1');
+            version('sdk', '1.5.2');
 
             // sdk deps
             version('okhttp3', '4.12.0')

src/main/java/io/harness/cf/client/api/BaseConfig.java

@@ -27,7 +27,7 @@ public class BaseConfig {
   @Getter(AccessLevel.NONE)
   private final int frequency = 60; // unit: second
 
-  @Builder.Default private final int bufferSize = 2048;
+  @Builder.Default private final int bufferSize = 5000;
 
   // Flag to set all attributes as private
   @Deprecated @Builder.Default private final boolean allAttributesPrivate = false;

src/main/java/io/harness/cf/client/api/MetricsProcessor.java

@@ -1,5 +1,6 @@
 package io.harness.cf.client.api;
 
+import static io.harness.cf.client.common.SdkCodes.warnMetricsBufferFull;
 import static io.harness.cf.client.common.Utils.shutdownExecutorService;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -70,6 +71,10 @@ private void transferValueIntoMapAtomicallyAndUpdateTo(
             return newValue;
           });
     }
+
+    public boolean containsKey(K key) {
+      return freqMap.containsKey(key);
+    }
   }
 
   private static final String FEATURE_NAME_ATTRIBUTE = "featureName";
@@ -91,55 +96,66 @@ private void transferValueIntoMapAtomicallyAndUpdateTo(
   private static final Target globalTarget =
       Target.builder().name(GLOBAL_TARGET_NAME).identifier(GLOBAL_TARGET).build();
 
+  private static final int MAX_SENT_TARGETS_TO_RETAIN = 100_000;
+  private static final int MAX_FREQ_MAP_TO_RETAIN = 10_000;
+
+  private static final LongAdder evalCounter = new LongAdder();
+  private static final LongAdder metricsEvalsDropped = new LongAdder();
+  private static final LongAdder targetsSeenDropped = new LongAdder();
   private final Connector connector;
   private final BaseConfig config;
   private final FrequencyMap<MetricEvent> frequencyMap;
-  private final Set<Target> uniqueTargetSet;
+  private final Set<Target> targetsSeen;
 
   private ScheduledFuture<?> runningTask = null;
   private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
 
   private final LongAdder metricsSent = new LongAdder();
+  private final int maxFreqMapSize;
 
   public MetricsProcessor(
       @NonNull Connector connector, @NonNull BaseConfig config, @NonNull MetricsCallback callback) {
     this.connector = connector;
     this.config = config;
     this.frequencyMap = new FrequencyMap<>();
-    this.uniqueTargetSet = ConcurrentHashMap.newKeySet();
+    this.targetsSeen = ConcurrentHashMap.newKeySet();
+    this.maxFreqMapSize = clamp(config.getBufferSize(), 2048, MAX_FREQ_MAP_TO_RETAIN);
     callback.onMetricsReady();
   }
 
+  private int clamp(int value, int lower, int higher) {
+    return Math.max(lower, Math.min(higher, value));
+  }
+
   @Deprecated /* The name of this method no longer makes sense since we moved to a map, kept for source compatibility */
   public void pushToQueue(Target target, String featureName, Variation variation) {
     registerEvaluation(target, featureName, variation);
   }
 
   void registerEvaluation(Target target, String featureName, Variation variation) {
 
-    final int freqMapSize = frequencyMap.size();
+    Target metricTarget = globalTarget;
 
-    if (freqMapSize > config.getBufferSize()) {
-      if (log.isWarnEnabled()) {
-        log.warn(
-            "Metric frequency map exceeded buffer size ({} > {}), force flushing",
-            freqMapSize,
-            config.getBufferSize());
+    if (target != null) {
+      if (!targetsSeen.contains(target) && targetsSeen.size() + 1 > MAX_SENT_TARGETS_TO_RETAIN) {
+        targetsSeenDropped.increment();
+      } else {
+        targetsSeen.add(target);
+        if (!config.isGlobalTargetEnabled()) {
+          metricTarget = target;
+        }
       }
-      // If the map is starting to grow too much then push the events now and reset the counters
-      scheduler.schedule(this::runOneIteration, 0, SECONDS);
     }
 
-    Target metricTarget = globalTarget;
+    final MetricEvent metricsEvent = new MetricEvent(featureName, metricTarget, variation);
 
-    if (target != null) {
-      uniqueTargetSet.add(target);
-      if (!config.isGlobalTargetEnabled()) {
-        metricTarget = target;
-      }
+    if (!frequencyMap.containsKey(metricsEvent) && frequencyMap.size() + 1 > maxFreqMapSize) {
+      metricsEvalsDropped.increment();
+    } else {
+      frequencyMap.increment(metricsEvent);
     }
 
-    frequencyMap.increment(new MetricEvent(featureName, metricTarget, variation));
+    evalCounter.increment();
   }
 
   /** This method sends the metrics data to the analytics server and resets the cache */
@@ -259,15 +275,23 @@ private void addTargetData(Metrics metrics, Target target) {
 
   void runOneIteration() {
     Thread.currentThread().setName("MetricsThread");
+
+    final long droppedEvals = metricsEvalsDropped.sumThenReset();
+    final long droppedTargets = targetsSeenDropped.sumThenReset();
+
+    if (droppedEvals > 0 || droppedTargets > 0) {
+      warnMetricsBufferFull(droppedEvals, droppedTargets);
+    }
+
     if (log.isDebugEnabled()) {
       log.debug(
           "Drain metrics queue : frequencyMap size={} uniqueTargetSet size={}",
           frequencyMap.size(),
-          uniqueTargetSet.size());
+          targetsSeen.size());
     }
-    sendDataAndResetCache(frequencyMap.drainToMap(), new HashSet<>(uniqueTargetSet));
+    sendDataAndResetCache(frequencyMap.drainToMap(), new HashSet<>(targetsSeen));
 
-    uniqueTargetSet.clear();
+    targetsSeen.clear();
   }
 
   public void start() {
@@ -328,7 +352,15 @@ long getQueueSize() {
   }
 
   long getTargetSetSize() {
-    return uniqueTargetSet.size();
+    return targetsSeen.size();
+  }
+
+  long getMetricsEvalsDropped() {
+    return metricsEvalsDropped.sum();
+  }
+
+  long getTargetsSeenDropped() {
+    return targetsSeenDropped.sum();
   }
 
   void reset() {

src/main/java/io/harness/cf/client/common/SdkCodes.java

@@ -65,6 +65,14 @@ public static void warnPostMetricsFailed(String reason) {
     log.warn(sdkErrMsg(7002, Optional.ofNullable(reason)));
   }
 
+  public static void warnMetricsBufferFull(long droppedEvals, long droppedTargets) {
+    log.warn(
+        sdkErrMsg(
+            7008,
+            Optional.of(
+                "- evals dropped: " + droppedEvals + " targets dropped: " + droppedTargets)));
+  }
+
   public static void warnDefaultVariationServed(String identifier, Target target, String def) {
     String targetId = (target == null) ? "null" : target.getIdentifier();
     String msg = String.format("identifier=%s, target=%s, default=%s", identifier, targetId, def);
@@ -111,7 +119,8 @@ public static void warnBucketByAttributeNotFound(String bucketBy, String usingVa
                 // SDK_METRICS_7xxx
                 {"7000", "Metrics thread started, intervalMs:"},
                 {"7001", "Metrics thread exited"},
-                {"7002", "Posting metrics failed, reason:"}
+                {"7002", "Posting metrics failed, reason:"},
+                {"7008", "Metrics buffer is full and metrics will be discarded"}
               })
           .collect(Collectors.toMap(entry -> Integer.parseInt(entry[0]), entry -> entry[1]));
 

src/test/java/io/harness/cf/client/api/MetricsProcessorTest.java

@@ -34,8 +34,7 @@ public void setup() {
     MockitoAnnotations.openMocks(this);
     metricsProcessor =
         Mockito.spy(
-            new MetricsProcessor(
-                connector, BaseConfig.builder().bufferSize(BUFFER_SIZE).build(), this));
+            new MetricsProcessor(connector, BaseConfig.builder().bufferSize(10_001).build(), this));
 
     metricsProcessor.reset();
   }
@@ -79,7 +78,7 @@ public void shouldNotThrowOutOfMemoryErrorWhenCreatingThreads() throws Interrupt
           Variation variation =
               Variation.builder().identifier("true" + v).name("name" + v).value("true").build();
 
-          metricsProcessor.pushToQueue(target, feature.getFeature(), variation);
+          metricsProcessor.registerEvaluation(target, feature.getFeature(), variation);
 
           maxQueueMapSize = Math.max(maxQueueMapSize, metricsProcessor.getQueueSize());
           maxUniqueTargetSetSize =
@@ -89,13 +88,14 @@ public void shouldNotThrowOutOfMemoryErrorWhenCreatingThreads() throws Interrupt
 
       if (t % 10 == 0) {
         log.info(
-            "Metrics frequency map (cur: {} max: {}) Unique targets (cur: {} max: {}) Events sent ({}) Events pending ({})",
+            "Metrics frequency map (cur: {} max: {}) Unique targets (cur: {} max: {}) Events sent ({}) Events pending ({}) Dropped ({})",
             metricsProcessor.getQueueSize(),
             maxQueueMapSize,
             metricsProcessor.getTargetSetSize(),
             maxUniqueTargetSetSize,
             metricsProcessor.getMetricsSent(),
-            metricsProcessor.getPendingMetricsToBeSent());
+            metricsProcessor.getPendingMetricsToBeSent(),
+            metricsProcessor.getMetricsEvalsDropped());
 
         metricsProcessor.flushQueue(); // mimic scheduled job
       }