Reuse aggregation handles for delta temporality

open-telemetry · Feb 3, 2023 · 7c88492 · 7c88492
1 parent 249d097
commit 7c88492
Showing 1 changed file with 16 additions and 2 deletions.
diff --git a/...ain/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java b/...ain/java/io/opentelemetry/sdk/metrics/internal/state/DefaultSynchronousMetricStorage.java
@@ -25,6 +25,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -49,6 +50,7 @@ public final class DefaultSynchronousMetricStorage<T extends PointData, U extend
   private final ConcurrentHashMap<Attributes, AggregatorHandle<T, U>> aggregatorHandles =
       new ConcurrentHashMap<>();
   private final AttributesProcessor attributesProcessor;
+  private final ConcurrentLinkedQueue<AggregatorHandle<T, U>> pool = new ConcurrentLinkedQueue<>();
 
   DefaultSynchronousMetricStorage(
       RegisteredReader registeredReader,
@@ -99,7 +101,11 @@ private AggregatorHandle<T, U> getAggregatorHandle(Attributes attributes, Contex
               + ").");
       return null;
     }
-    AggregatorHandle<T, U> newHandle = aggregator.createHandle();
+    // Get handle from pool if available, else create a new one.
+    AggregatorHandle<T, U> newHandle = pool.poll();
+    if (newHandle == null) {
+      newHandle = aggregator.createHandle();
+    }
     handle = aggregatorHandles.putIfAbsent(attributes, newHandle);
     return handle != null ? handle : newHandle;
   }
@@ -119,16 +125,24 @@ public MetricData collect(
     // Grab aggregated points.
     List<T> points = new ArrayList<>(aggregatorHandles.size());
     for (Map.Entry<Attributes, AggregatorHandle<T, U>> entry : aggregatorHandles.entrySet()) {
+      T point = entry.getValue().aggregateThenMaybeReset(start, epochNanos, entry.getKey(), reset);
       if (reset) {
         aggregatorHandles.remove(entry.getKey(), entry.getValue());
+        // Return the aggregator to the pool.
+        pool.offer(entry.getValue());
       }
-      T point = entry.getValue().aggregateThenMaybeReset(start, epochNanos, entry.getKey(), reset);
       if (point == null) {
         continue;
       }
       points.add(point);
     }
 
+    // Trim pool down if needed
+    int toDelete = pool.size() - MAX_CARDINALITY;
+    for (int i = 0; i < toDelete; i++) {
+      pool.poll();
+    }
+
     if (points.isEmpty()) {
       return EmptyMetricData.getInstance();
     }