perf: skip snapshot rebuild in mergeDuplicates when names are unique

prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java

@@ -49,6 +49,10 @@ private static final class Generation {
   // available processors. This is simpler than the striping used by LongAdder, so hot spots remain
   // possible when several recording threads resolve to the same stripe.
   private final AtomicLong[] stripedObservationCounts;
+  // Protected by appendLock. These are absolute per-stripe observation counts at activation, not
+  // the reset-adjusted count used by complete. Reused across generations to avoid scrape
+  // allocations.
+  private final long[] generationStartCounts;
   private final ReentrantLock observationLock = new ReentrantLock();
   private boolean reset;
   private long observationCountOffset;
@@ -76,33 +80,41 @@ private static final class Generation {
     this.maxBufferSize = maxBufferSize;
     this.beforeAppendLock = beforeAppendLock;
     stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()];
+    generationStartCounts = new long[stripedObservationCounts.length];
     for (int i = 0; i < stripedObservationCounts.length; i++) {
       stripedObservationCounts[i] = new AtomicLong();
     }
   }
 
   boolean append(double value) {
-    AtomicLong counter =
-        stripedObservationCounts[
-            stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)];
+    int stripe = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length);
+    AtomicLong counter = stripedObservationCounts[stripe];
     long count = counter.incrementAndGet();
     // The active bit is the exact handoff decision. An observation either increments its stripe
     // before the collector's getAndAdd(BUFFER_ACTIVE_BIT) and takes the direct path, or sees the
-    // active bit and is buffered in the current generation.
+    // active bit and may be buffered. The stripe ticket below also checks that it was not counted
+    // by a later collection that started before this thread read activeGeneration.
     if ((count & BUFFER_ACTIVE_BIT) == 0) {
       return false;
     }
+    // Allow tests to pause between allocating an observation ticket and reading the generation.
+    beforeAppendLock.run();
     Generation generation = activeGeneration;
     if (generation == null) {
       return false;
     }
-    beforeAppendLock.run();
     appendLock.lock();
     try {
       Generation current = activeGeneration;
       if (current != generation || !generation.active) {
         return false;
       }
+      if ((count & ~BUFFER_ACTIVE_BIT) <= generationStartCounts[stripe]) {
+        // This observation incremented its stripe in an earlier generation. The current collector
+        // already includes it in expectedCount, so buffering it here would make the collector wait
+        // for an observation that is only replayed after that same wait finishes.
+        return false;
+      }
       while (generation.size >= maxBufferSize && generation.active) {
         try {
           bufferSpaceAvailable.await();
@@ -179,8 +191,10 @@ <T extends DataPointSnapshot> T run(
       try {
         activeGeneration = generation;
         long total = 0;
-        for (AtomicLong counter : stripedObservationCounts) {
-          total += counter.getAndAdd(BUFFER_ACTIVE_BIT);
+        for (int i = 0; i < stripedObservationCounts.length; i++) {
+          long count = stripedObservationCounts[i].getAndAdd(BUFFER_ACTIVE_BIT);
+          generationStartCounts[i] = count;
+          total += count;
         }
         expectedCount = total - observationCountOffset;
       } finally {

prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java

@@ -8,6 +8,9 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -180,83 +183,123 @@ void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException {
   }
 
   @Test
-  void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException {
-    CountDownLatch firstRunStarted = new CountDownLatch(1);
-    CountDownLatch firstRunMayFinish = new CountDownLatch(1);
-    CountDownLatch stalled = new CountDownLatch(1);
-    CountDownLatch release = new CountDownLatch(1);
+  void lateAppenderCountedByNextGenerationMustNotBeBufferedAgain() throws Exception {
+    assertLateAppenderHandoff(false);
+  }
+
+  @Test
+  void lateAppenderHandoffUsesAbsoluteStripeCountsAfterReset() throws Exception {
+    assertLateAppenderHandoff(true);
+  }
+
+  private static void assertLateAppenderHandoff(boolean reset) throws Exception {
+    CountDownLatch firstSnapshotStarted = new CountDownLatch(1);
+    CountDownLatch finishFirstSnapshot = new CountDownLatch(1);
+    CountDownLatch observationCounted = new CountDownLatch(1);
+    CountDownLatch readGeneration = new CountDownLatch(1);
     CountDownLatch secondRunStarted = new CountDownLatch(1);
-    AtomicBoolean appended = new AtomicBoolean();
     AtomicLong completedObservations = new AtomicLong();
+    AtomicLong secondExpectedCount = new AtomicLong();
+    AtomicBoolean pauseFirstAppender = new AtomicBoolean(true);
     Buffer buffer =
         new Buffer(
-            TimeUnit.SECONDS.toNanos(1),
+            TimeUnit.SECONDS.toNanos(5),
             16,
             () -> {
-              stalled.countDown();
-              try {
-                release.await();
-              } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
+              if (pauseFirstAppender.compareAndSet(true, false)) {
+                observationCounted.countDown();
+                awaitLatch(readGeneration);
               }
             });
-    Thread firstRun =
-        new Thread(
-            () ->
-                buffer.run(
-                    ignored -> {
-                      firstRunStarted.countDown();
-                      return firstRunMayFinish.getCount() == 0;
-                    },
-                    () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0),
-                    ignored -> {}),
-            "buffer-first-runner");
-    firstRun.setDaemon(true);
-    firstRun.start();
-    assertThat(firstRunStarted.await(5, TimeUnit.SECONDS)).isTrue();
+    if (reset) {
+      assertThat(buffer.append(1.0)).isFalse();
+      buffer.observeDirect(completedObservations::incrementAndGet);
+    }
+    ExecutorService executor = Executors.newFixedThreadPool(2);
+    try {
+      Future<CounterSnapshot.CounterDataPointSnapshot> firstRun =
+          executor.submit(
+              () ->
+                  buffer.run(
+                      expectedCount -> completedObservations.get() == expectedCount,
+                      () -> {
+                        firstSnapshotStarted.countDown();
+                        awaitLatch(finishFirstSnapshot);
+                        CounterSnapshot.CounterDataPointSnapshot snapshot =
+                            new CounterSnapshot.CounterDataPointSnapshot(
+                                completedObservations.get(), Labels.EMPTY, null, 0);
+                        if (reset) {
+                          completedObservations.set(0);
+                          buffer.reset();
+                        }
+                        return snapshot;
+                      },
+                      ignored -> completedObservations.incrementAndGet()));
+      awaitLatch(firstSnapshotStarted);
 
-    Thread appender =
-        new Thread(
-            () -> {
-              appended.set(buffer.append(1.0));
-              if (!appended.get()) {
-                buffer.observeDirect(
-                    () -> {
-                      completedObservations.incrementAndGet();
-                      return null;
-                    });
-              }
-            },
-            "buffer-late-appender");
-    appender.setDaemon(true);
-    appender.start();
-    assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue();
+      // Increment while generation A is active, but do not read activeGeneration yet.
+      Future<Boolean> appender =
+          executor.submit(
+              () -> {
+                boolean appended = buffer.append(1.0);
+                if (!appended) {
+                  buffer.observeDirect(completedObservations::incrementAndGet);
+                }
+                return appended;
+              });
+      awaitLatch(observationCounted);
+      finishFirstSnapshot.countDown();
+      assertThat(firstRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(reset ? 1 : 0);
 
-    firstRunMayFinish.countDown();
-    firstRun.join(5_000);
-    assertThat(firstRun.isAlive()).isFalse();
+      Future<CounterSnapshot.CounterDataPointSnapshot> secondRun =
+          executor.submit(
+              () ->
+                  buffer.run(
+                      expectedCount -> {
+                        secondExpectedCount.set(expectedCount);
+                        secondRunStarted.countDown();
+                        return completedObservations.get() == expectedCount;
+                      },
+                      () ->
+                          new CounterSnapshot.CounterDataPointSnapshot(
+                              completedObservations.get(), Labels.EMPTY, null, 0),
+                      ignored -> completedObservations.incrementAndGet()));
+      awaitLatch(secondRunStarted);
+      assertThat(secondExpectedCount).hasValue(1);
+      // An observation arriving after B's activation still belongs in B's buffer. It must not
+      // appear in B's snapshot and must be replayed exactly once before the following collection.
+      assertThat(buffer.append(1.0)).isTrue();
 
-    Thread secondRun =
-        new Thread(
-            () ->
-                buffer.run(
-                    expectedCount -> {
-                      secondRunStarted.countDown();
-                      return completedObservations.get() == expectedCount;
-                    },
-                    () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0),
-                    ignored -> {}),
-            "buffer-second-runner");
-    secondRun.setDaemon(true);
-    secondRun.start();
-    assertThat(secondRunStarted.await(5, TimeUnit.SECONDS)).isTrue();
-    release.countDown();
-    appender.join(5_000);
-    secondRun.join(5_000);
+      // B includes the paused observation in expectedCount. Buffering it in B would make B wait
+      // until its own timeout/replay; it must instead complete via the direct observation path.
+      readGeneration.countDown();
+      assertThat(secondRun.get(10, TimeUnit.SECONDS).getValue()).isEqualTo(1);
+      assertThat(appender.get(10, TimeUnit.SECONDS)).isFalse();
+      assertThat(completedObservations).hasValue(2);
+      assertThat(
+              buffer
+                  .run(
+                      expectedCount -> completedObservations.get() == expectedCount,
+                      () ->
+                          new CounterSnapshot.CounterDataPointSnapshot(
+                              completedObservations.get(), Labels.EMPTY, null, 0),
+                      ignored -> completedObservations.incrementAndGet())
+                  .getValue())
+          .isEqualTo(2);
+    } finally {
+      finishFirstSnapshot.countDown();
+      readGeneration.countDown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+    }
+  }
 
-    assertThat(appender.isAlive()).isFalse();
-    assertThat(secondRun.isAlive()).isFalse();
-    assertThat(appended).isFalse();
-    assertThat(completedObservations).hasValue(1);
+  private static void awaitLatch(CountDownLatch latch) {
+    try {
+      assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException(e);
+    }
   }
 }

prometheus-metrics-exposition-textformats/src/main/java/io/prometheus/metrics/expositionformats/TextFormatUtil.java

@@ -41,6 +41,24 @@ public static MetricSnapshots mergeDuplicates(MetricSnapshots metricSnapshots) {
       return metricSnapshots;
     }
 
+    // MetricSnapshots is sorted by prometheus name, so any duplicates are adjacent. Detect them in
+    // a single allocation-free pass; when there are none (the common case) return the input as-is
+    // rather than rebuilding it through a map, a list per group and a new MetricSnapshots.
+    boolean hasDuplicates = false;
+    for (int i = 1; i < metricSnapshots.size(); i++) {
+      if (metricSnapshots
+          .get(i)
+          .getMetadata()
+          .getPrometheusName()
+          .equals(metricSnapshots.get(i - 1).getMetadata().getPrometheusName())) {
+        hasDuplicates = true;
+        break;
+      }
+    }
+    if (!hasDuplicates) {
+      return metricSnapshots;
+    }
+
     Map<String, List<MetricSnapshot>> grouped = new LinkedHashMap<>();
 
     for (MetricSnapshot snapshot : metricSnapshots) {

prometheus-metrics-exposition-textformats/src/test/java/io/prometheus/metrics/expositionformats/TextFormatUtilTest.java

@@ -178,4 +178,65 @@ void testMergeDuplicates_histogramSameGaugeFlag_preservesGaugeHistogram() {
     assertThat(merged.isGaugeHistogram()).isTrue();
     assertThat(merged.getDataPoints()).hasSize(2);
   }
+
+  @Test
+  void testMergeDuplicates_uniqueNames_returnsSameInstance() {
+    CounterSnapshot counter1 =
+        CounterSnapshot.builder()
+            .name("api_responses")
+            .dataPoint(CounterSnapshot.CounterDataPointSnapshot.builder().value(1).build())
+            .build();
+    CounterSnapshot counter2 =
+        CounterSnapshot.builder()
+            .name("api_errors")
+            .dataPoint(CounterSnapshot.CounterDataPointSnapshot.builder().value(2).build())
+            .build();
+
+    MetricSnapshots snapshots = new MetricSnapshots(counter1, counter2);
+
+    // No duplicate prometheus names, so the fast path returns the input unchanged rather
+    // than rebuilding it. isSameAs pins that: an inverted condition would rebuild and fail here.
+    assertThat(TextFormatUtil.mergeDuplicates(snapshots)).isSameAs(snapshots);
+  }
+
+  @Test
+  void testMergeDuplicates_duplicateNotAtStart_merges() {
+    CounterSnapshot a =
+        CounterSnapshot.builder()
+            .name("a")
+            .dataPoint(CounterSnapshot.CounterDataPointSnapshot.builder().value(1).build())
+            .build();
+    CounterSnapshot m =
+        CounterSnapshot.builder()
+            .name("m")
+            .dataPoint(CounterSnapshot.CounterDataPointSnapshot.builder().value(1).build())
+            .build();
+    CounterSnapshot z1 =
+        CounterSnapshot.builder()
+            .name("z")
+            .dataPoint(
+                CounterSnapshot.CounterDataPointSnapshot.builder()
+                    .labels(Labels.of("outcome", "SUCCESS"))
+                    .value(1)
+                    .build())
+            .build();
+    CounterSnapshot z2 =
+        CounterSnapshot.builder()
+            .name("z")
+            .dataPoint(
+                CounterSnapshot.CounterDataPointSnapshot.builder()
+                    .labels(Labels.of("outcome", "FAILURE"))
+                    .value(2)
+                    .build())
+            .build();
+
+    // Sorted by prometheus name the duplicate "z" pair is last (indices 2 and 3), so
+    // detection must scan the whole range, not just the first pair.
+    MetricSnapshots snapshots = new MetricSnapshots(a, m, z1, z2);
+    MetricSnapshots result = TextFormatUtil.mergeDuplicates(snapshots);
+
+    assertThat(result).hasSize(3);
+    assertThat(result.get(2).getMetadata().getName()).isEqualTo("z");
+    assertThat(result.get(2).getDataPoints()).hasSize(2);
+  }
 }

prometheus-metrics-model/src/main/java/io/prometheus/metrics/model/snapshots/MetricSnapshots.java

@@ -36,6 +36,9 @@ public MetricSnapshots(MetricSnapshot... snapshots) {
    */
   public MetricSnapshots(Collection<MetricSnapshot> snapshots) {
     List<MetricSnapshot> list = new ArrayList<>(snapshots);
+    // Sort by prometheus name so snapshots that share a prometheus name (i.e. the same exposed
+    // family) are adjacent. TextFormatUtil.mergeDuplicates depends on this ordering to detect
+    // duplicates in a single linear pass; if this sort key changes, update that fast path too.
     list.sort(comparing(s -> s.getMetadata().getPrometheusName()));
 
     // Validate no conflicting metric types