From bc2244d635598813d0845900c4d4db94895ca224 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Fabian=20St=C3=A4ber?= <fabian@fstab.de>
Date: Thu, 9 Nov 2023 17:33:05 +0100
Subject: [PATCH] Fix sliding window rotation #894 (#896)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Signed-off-by: Fabian Stäber <fabian@fstab.de>
---
 .../metrics/core/metrics/SlidingWindow.java   |  9 ++-
 .../core/metrics/SlidingWindowTest.java       | 77 +++++++++++++++++++
 2 files changed, 84 insertions(+), 2 deletions(-)
 create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/SlidingWindowTest.java

diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/SlidingWindow.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/SlidingWindow.java
index b789e809c..62164a48b 100644
--- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/SlidingWindow.java
+++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/SlidingWindow.java
@@ -2,6 +2,7 @@
 
 import java.lang.reflect.Array;
 import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
 import java.util.function.ObjDoubleConsumer;
 import java.util.function.Supplier;
 
@@ -22,6 +23,7 @@ public class SlidingWindow<T> {
     private int currentBucket;
     private long lastRotateTimestampMillis;
     private final long durationBetweenRotatesMillis;
+    LongSupplier currentTimeMillis = System::currentTimeMillis; // to be replaced in unit tests
 
     /**
      * Example: If the {@code maxAgeSeconds} is 60 and {@code ageBuckets} is 3, then 3 instances of {@code T}
@@ -56,11 +58,14 @@ public synchronized T current() {
      * Observe a value.
      */
     public synchronized void observe(double value) {
-        observeFunction.accept(rotate(), value);
+        rotate();
+        for (T t : ringBuffer) {
+            observeFunction.accept(t, value);
+        }
     }
 
     private T rotate() {
-        long timeSinceLastRotateMillis = System.currentTimeMillis() - lastRotateTimestampMillis;
+        long timeSinceLastRotateMillis = currentTimeMillis.getAsLong() - lastRotateTimestampMillis;
         while (timeSinceLastRotateMillis > durationBetweenRotatesMillis) {
             ringBuffer[currentBucket] = constructor.get();
             if (++currentBucket >= ringBuffer.length) {
diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/SlidingWindowTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/SlidingWindowTest.java
new file mode 100644
index 000000000..3b57220f8
--- /dev/null
+++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/SlidingWindowTest.java
@@ -0,0 +1,77 @@
+package io.prometheus.metrics.core.metrics;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class SlidingWindowTest {
+
+    static class Observer {
+
+        List<Double> values = new ArrayList<>();
+
+        public void observe(double value) {
+            values.add(value);
+        }
+
+        void assertValues(double... expectedValues) {
+            ArrayList<Double> expectedList = new ArrayList<>();
+            for (double expectedValue : expectedValues) {
+                expectedList.add(expectedValue);
+            }
+            Assert.assertEquals(expectedList, values);
+        }
+    }
+
+    private final AtomicLong currentTimeMillis = new AtomicLong();
+    private SlidingWindow<Observer> ringBuffer;
+    private final long maxAgeSeconds = 30;
+    private final int ageBuckets = 5;
+    private final long timeBetweenRotateMillis = maxAgeSeconds * 1000 / ageBuckets + 2;
+
+    @Before
+    public void setUp() {
+        currentTimeMillis.set(System.currentTimeMillis());
+        ringBuffer = new SlidingWindow<>(Observer.class, Observer::new, Observer::observe, maxAgeSeconds, ageBuckets);
+        ringBuffer.currentTimeMillis = currentTimeMillis::get;
+    }
+
+    @Test
+    public void testRotate() {
+        for (int i=0; i<ageBuckets; i++) {
+            currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+            ringBuffer.observe(1.0);
+        }
+        ringBuffer.current().assertValues(1.0, 1.0, 1.0, 1.0, 1.0);
+        currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+        ringBuffer.current().assertValues(1.0, 1.0, 1.0, 1.0);
+        currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+        ringBuffer.current().assertValues(1.0, 1.0, 1.0);
+        currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+        ringBuffer.current().assertValues(1.0, 1.0);
+        currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+        ringBuffer.current().assertValues(1.0);
+        currentTimeMillis.addAndGet(timeBetweenRotateMillis);
+        ringBuffer.current().assertValues();
+    }
+
+    @Test
+    public void testMultiRotate() {
+        ringBuffer.observe(1.0);
+        currentTimeMillis.addAndGet(2 * timeBetweenRotateMillis); // 2/5 of max aqe
+        ringBuffer.observe(2.0);
+        ringBuffer.current().assertValues(1.0, 2.0);
+        currentTimeMillis.addAndGet(3 * timeBetweenRotateMillis); // 5/5 of max age -> first observation evicted
+        ringBuffer.current().assertValues(2.0);
+        ringBuffer.observe(3.0);
+        ringBuffer.current().assertValues(2.0, 3.0);
+        currentTimeMillis.addAndGet(2 * timeBetweenRotateMillis); // 7/5 of max age
+        ringBuffer.current().assertValues(3.0);
+        currentTimeMillis.addAndGet(3 * timeBetweenRotateMillis); // 10/5 of max age
+        ringBuffer.current().assertValues(); // empty
+    }
+}