Publish timed metric max values over rolling 59-second windows

ebean-core/src/main/java/io/ebeaninternal/server/profile/DTimedMetric.java

@@ -3,8 +3,6 @@
 import io.ebean.meta.MetricVisitor;
 import io.ebean.metric.TimedMetric;
 
-import java.util.concurrent.atomic.LongAccumulator;
-
 /**
  * Used to collect timed execution statistics.
  * <p>
@@ -16,12 +14,17 @@ final class DTimedMetric implements TimedMetric {
   private final String name;
   private final ValueAdder count = new ValueAdder();
   private final ValueAdder total = new ValueAdder();
-  private final LongAccumulator max = new LongAccumulator(Math::max, 0);
+  private final ValueMax max;
   private boolean collected;
   private String reportName;
 
   DTimedMetric(String name) {
+    this(name, new ValueMax());
+  }
+
+  DTimedMetric(String name, ValueMax max) {
     this.name = name;
+    this.max = max;
   }
 
   @Override
@@ -31,7 +34,7 @@ public void addBatchSince(long startNanos, int batch) {
       final long mean = totalMicros / batch;
       count.add(batch);
       total.add(totalMicros);
-      max.accumulate(mean);
+      max.add(mean);
     }
   }
 
@@ -44,7 +47,7 @@ public void addSinceNanos(long startNanos) {
   public void add(long value) {
     count.add(1);
     total.add(value);
-    max.accumulate(value);
+    max.add(value);
   }
 
   @Override
@@ -76,6 +79,7 @@ public DTimeMetricStats collect(boolean reset) {
 
   @Override
   public DTimeMetricStats collect(MetricVisitor.Mode mode) {
+    final long maxValue = max.collect();
     final long countSum;
     switch (mode) {
       case RESET:
@@ -93,14 +97,14 @@ public DTimeMetricStats collect(MetricVisitor.Mode mode) {
     if (countSum == 0) {
       return null;
     } else {
-      return stats(mode, name, countSum);
+      return stats(mode, name, countSum, maxValue);
     }
   }
 
   /**
    * Return the current statistics resetting the internal values if reset is true.
    */
-  private DTimeMetricStats stats(MetricVisitor.Mode mode, String name, long countSum) {
+  private DTimeMetricStats stats(MetricVisitor.Mode mode, String name, long countSum, long maxValue) {
     try {
       final long totalSum;
       switch (mode) {
@@ -116,7 +120,7 @@ private DTimeMetricStats stats(MetricVisitor.Mode mode, String name, long countS
         default:
           throw new IllegalStateException("Unknown metric collection mode");
       }
-      return new DTimeMetricStats(name, collected, countSum, totalSum, max.getThenReset());
+      return new DTimeMetricStats(name, collected, countSum, totalSum, maxValue);
     } finally {
       collected = true;
     }

ebean-core/src/main/java/io/ebeaninternal/server/profile/ValueMax.java

@@ -0,0 +1,48 @@
+package io.ebeaninternal.server.profile;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAccumulator;
+import java.util.function.LongSupplier;
+
+/**
+ * Accumulates a maximum value and publishes it in rolling 59-second windows.
+ */
+final class ValueMax {
+
+  private static final long WINDOW_NANOS = TimeUnit.SECONDS.toNanos(59);
+
+  private final LongSupplier nanoTime;
+  private final LongAccumulator value;
+  private volatile long published;
+  private long lastResetNanos;
+
+  ValueMax() {
+    this(System::nanoTime);
+  }
+
+  ValueMax(LongSupplier nanoTime) {
+    this.nanoTime = nanoTime;
+    this.value = new LongAccumulator(Math::max, 0);
+    this.lastResetNanos = nanoTime.getAsLong() - 2 * WINDOW_NANOS;
+  }
+
+  void add(long amount) {
+    value.accumulate(amount);
+  }
+
+  synchronized long collect() {
+    long now = nanoTime.getAsLong();
+    if (now - lastResetNanos >= WINDOW_NANOS) {
+      published = value.getThenReset();
+      lastResetNanos = now;
+    }
+    return published;
+  }
+
+  synchronized void reset() {
+    value.reset();
+    published = 0;
+    lastResetNanos = nanoTime.getAsLong() - 2 * WINDOW_NANOS;
+  }
+
+}

ebean-core/src/test/java/io/ebeaninternal/server/profile/DQueryPlanMetricTest.java

@@ -5,19 +5,22 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 class DQueryPlanMetricTest {
 
+  private final AtomicLong nanoTime = new AtomicLong();
   Function<String, String> naming = (String name) -> "prefix[" + name.replace('.', '-') + "]";
 
   @Test
   void visit() {
 
     DQueryPlanMeta meta = new DQueryPlanMeta(Object.class, "dto.Object.lab", "lab", null, "sql", "hash");
-    DTimedMetric metric = new DTimedMetric("org.timed.plan");
+    DTimedMetric metric = new DTimedMetric("org.timed.plan", new ValueMax(nanoTime::get));
     DQueryPlanMetric planMetric = new DQueryPlanMetric(meta, metric);
 
     metric.add(560);
@@ -46,10 +49,10 @@ void visit() {
   }
 
   @Test
-  void visitCumulativeResetsMax() {
+  void visitCumulativePublishesSharedMax() {
 
     DQueryPlanMeta meta = new DQueryPlanMeta(Object.class, "dto.Object.lab", "lab", null, "sql", "hash");
-    DTimedMetric metric = new DTimedMetric("org.timed.plan");
+    DTimedMetric metric = new DTimedMetric("org.timed.plan", new ValueMax(nanoTime::get));
     DQueryPlanMetric planMetric = new DQueryPlanMetric(meta, metric);
 
     metric.add(560);
@@ -74,7 +77,7 @@ void visitCumulativeResetsMax() {
       assertThat(result.get(0).name()).isEqualTo("prefix[dto-Object-lab]");
       assertThat(result.get(0).count()).isEqualTo(2);
       assertThat(result.get(0).total()).isEqualTo(820);
-      assertThat(result.get(0).max()).isEqualTo(0);
+      assertThat(result.get(0).max()).isEqualTo(560);
     }
 
     metric.add(410);
@@ -87,7 +90,14 @@ void visitCumulativeResetsMax() {
       assertThat(result.get(0).name()).isEqualTo("prefix[dto-Object-lab]");
       assertThat(result.get(0).count()).isEqualTo(3);
       assertThat(result.get(0).total()).isEqualTo(1230);
-      assertThat(result.get(0).max()).isEqualTo(410);
+      assertThat(result.get(0).max()).isEqualTo(560);
     }
+
+    nanoTime.addAndGet(TimeUnit.SECONDS.toNanos(59));
+    BasicMetricVisitor visitor = new BasicMetricVisitor("v", naming, false, true, true, true);
+    planMetric.visit(visitor);
+    List<MetaQueryMetric> result = visitor.queryMetrics();
+    assertThat(result).hasSize(1);
+    assertThat(result.get(0).max()).isEqualTo(410);
   }
 }

ebean-core/src/test/java/io/ebeaninternal/server/profile/DTimedMetricTest.java

@@ -5,16 +5,20 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 public class DTimedMetricTest {
 
+  private final AtomicLong nanoTime = new AtomicLong();
+
   @Test
   public void addSinceNanos() throws InterruptedException {
 
-    DTimedMetric metric = new DTimedMetric("addSinceNanos");
+    DTimedMetric metric = new DTimedMetric("addSinceNanos", new ValueMax(nanoTime::get));
 
     long start = System.nanoTime();
     Thread.sleep(11);
@@ -28,6 +32,7 @@ public void addSinceNanos() throws InterruptedException {
 
     metric.addSinceNanos(start);
 
+    nanoTime.addAndGet(TimeUnit.SECONDS.toNanos(59));
     stats = metric.collect(true);
     assertThat(stats.count()).isEqualTo(1);
     assertThat(stats.total()).isGreaterThan(10);
@@ -37,7 +42,7 @@ public void addSinceNanos() throws InterruptedException {
   @Test
   public void addBatchSince() throws InterruptedException {
 
-    DTimedMetric metric = new DTimedMetric("addSinceNanos");
+    DTimedMetric metric = new DTimedMetric("addSinceNanos", new ValueMax(nanoTime::get));
 
     long start = System.nanoTime();
     Thread.sleep(11);
@@ -52,6 +57,7 @@ public void addBatchSince() throws InterruptedException {
 
     metric.addBatchSince(start, 2);
 
+    nanoTime.addAndGet(TimeUnit.SECONDS.toNanos(59));
     stats = metric.collect(true);
     assertThat(stats.count()).isEqualTo(2);
     assertThat(stats.total()).isGreaterThan(10000);
@@ -92,8 +98,8 @@ void visit() {
   }
 
   @Test
-  void collectCumulativeResetsMax() {
-    DTimedMetric metric = new DTimedMetric("org.timed");
+  void collectCumulativePublishesSharedMax() {
+    DTimedMetric metric = new DTimedMetric("org.timed", new ValueMax(nanoTime::get));
     metric.add(560);
     metric.add(500);
 
@@ -105,7 +111,7 @@ void collectCumulativeResetsMax() {
     stats = metric.collect(false);
     assertThat(stats.count()).isEqualTo(2);
     assertThat(stats.total()).isEqualTo(1060);
-    assertThat(stats.max()).isEqualTo(0);
+    assertThat(stats.max()).isEqualTo(560);
 
     metric.add(160);
     metric.add(100);
@@ -114,12 +120,16 @@ void collectCumulativeResetsMax() {
     stats = metric.collect(false);
     assertThat(stats.count()).isEqualTo(5);
     assertThat(stats.total()).isEqualTo(1470);
+    assertThat(stats.max()).isEqualTo(560);
+
+    nanoTime.addAndGet(TimeUnit.SECONDS.toNanos(59));
+    stats = metric.collect(false);
     assertThat(stats.max()).isEqualTo(160);
   }
 
   @Test
   void cumulativeAndDeltaAreIndependent() {
-    DTimedMetric metric = new DTimedMetric("org.timed");
+    DTimedMetric metric = new DTimedMetric("org.timed", new ValueMax(nanoTime::get));
     metric.add(560);
     metric.add(500);
 
@@ -132,6 +142,7 @@ void cumulativeAndDeltaAreIndependent() {
     DTimeMetricStats delta = metric.collect(true);
     assertThat(delta.count()).isEqualTo(3);
     assertThat(delta.total()).isEqualTo(1220);
+    assertThat(delta.max()).isEqualTo(560);
 
     cumulative = metric.collect(false);
     assertThat(cumulative).isNull();