[FLINK-30090] Support timespan for TimerGauges

Author: gyfora

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/TimerGauge.java

@@ -32,9 +32,21 @@
  * happen in a couple of hours, the returned value will account for this ongoing measurement.
  */
 public class TimerGauge implements Gauge<Long>, View {
+
+    private static final int DEFAULT_TIME_SPAN_IN_SECONDS = 60;
+
     private final Clock clock;
 
-    private long previousCount;
+    /** The time-span over which the average is calculated. */
+    private final int timeSpanInSeconds;
+    /** Circular array containing the history of values. */
+    private final long[] values;
+    /** The index in the array for the current time. */
+    private int idx = 0;
+
+    private boolean fullWindow = false;
+
+    private long currentValue;
     private long currentCount;
     private long currentMeasurementStartTS;
     /**
@@ -50,11 +62,24 @@ public class TimerGauge implements Gauge<Long>, View {
     private long accumulatedCount;
 
     public TimerGauge() {
-        this(SystemClock.getInstance());
+        this(DEFAULT_TIME_SPAN_IN_SECONDS);
+    }
+
+    public TimerGauge(int timeSpanInSeconds) {
+        this(SystemClock.getInstance(), timeSpanInSeconds);
     }
 
     public TimerGauge(Clock clock) {
+        this(clock, DEFAULT_TIME_SPAN_IN_SECONDS);
+    }
+
+    public TimerGauge(Clock clock, int timeSpanInSeconds) {
         this.clock = clock;
+        this.timeSpanInSeconds =
+                Math.max(
+                        timeSpanInSeconds - (timeSpanInSeconds % UPDATE_INTERVAL_SECONDS),
+                        UPDATE_INTERVAL_SECONDS);
+        this.values = new long[this.timeSpanInSeconds / UPDATE_INTERVAL_SECONDS];
     }
 
     public synchronized void markStart() {
@@ -89,15 +114,32 @@ public synchronized void update() {
             currentMaxSingleMeasurement =
                     Math.max(currentMaxSingleMeasurement, now - currentMeasurementStartTS);
         }
-        previousCount = Math.max(Math.min(currentCount / UPDATE_INTERVAL_SECONDS, 1000), 0);
+        updateCurrentValue();
         previousMaxSingleMeasurement = currentMaxSingleMeasurement;
         currentCount = 0;
         currentMaxSingleMeasurement = 0;
     }
 
+    private void updateCurrentValue() {
+        if (idx == values.length - 1) {
+            fullWindow = true;
+        }
+        values[idx] = currentCount;
+        idx = (idx + 1) % values.length;
+
+        int maxIndex = fullWindow ? values.length : idx;
+        long totalTime = 0;
+        for (int i = 0; i < maxIndex; i++) {
+            totalTime += values[i];
+        }
+
+        currentValue =
+                Math.max(Math.min(totalTime / (UPDATE_INTERVAL_SECONDS * maxIndex), 1000), 0);
+    }
+
     @Override
     public synchronized Long getValue() {
-        return previousCount;
+        return currentValue;
     }
 
     /**

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/TimerGaugeTest.java

@@ -45,7 +45,7 @@ public class TimerGaugeTest {
     @Test
     public void testBasicUsage() {
         ManualClock clock = new ManualClock(42_000_000);
-        TimerGauge gauge = new TimerGauge(clock);
+        TimerGauge gauge = new TimerGauge(clock, View.UPDATE_INTERVAL_SECONDS);
 
         gauge.update();
         assertThat(gauge.getValue(), is(0L));
@@ -74,7 +74,7 @@ public void testBasicUsage() {
     @Test
     public void testUpdateWithoutMarkingEnd() {
         ManualClock clock = new ManualClock(42_000_000);
-        TimerGauge gauge = new TimerGauge(clock);
+        TimerGauge gauge = new TimerGauge(clock, View.UPDATE_INTERVAL_SECONDS);
 
         gauge.markStart();
         clock.advanceTime(SLEEP, TimeUnit.MILLISECONDS);
@@ -107,4 +107,48 @@ public void testGetWithoutUpdate() {
         assertThat(gauge.getValue(), is(0L));
         assertThat(gauge.getMaxSingleMeasurement(), is(0L));
     }
+
+    @Test
+    public void testLargerTimespan() {
+        ManualClock clock = new ManualClock(42_000_000);
+        TimerGauge gauge = new TimerGauge(clock, 2 * View.UPDATE_INTERVAL_SECONDS);
+
+        gauge.markStart();
+        clock.advanceTime(SLEEP, TimeUnit.MILLISECONDS);
+        gauge.markEnd();
+        gauge.update();
+
+        assertThat(gauge.getValue(), is(SLEEP / View.UPDATE_INTERVAL_SECONDS));
+        assertThat(gauge.getMaxSingleMeasurement(), is(SLEEP));
+        assertEquals(gauge.getAccumulatedCount(), SLEEP);
+
+        gauge.update();
+        // One sleep in 2 intervals
+        assertThat(gauge.getValue(), is(SLEEP / (View.UPDATE_INTERVAL_SECONDS * 2)));
+        assertThat(gauge.getMaxSingleMeasurement(), is(0L));
+        assertEquals(gauge.getAccumulatedCount(), SLEEP);
+
+        // One sleep in each interval
+
+        gauge.markStart();
+        clock.advanceTime(SLEEP, TimeUnit.MILLISECONDS);
+        gauge.markEnd();
+        gauge.update();
+
+        gauge.markStart();
+        clock.advanceTime(SLEEP, TimeUnit.MILLISECONDS);
+        gauge.markEnd();
+        gauge.update();
+
+        assertThat(gauge.getValue(), is(SLEEP / (View.UPDATE_INTERVAL_SECONDS)));
+
+        // Check that the getMaxSingleMeasurement can go down after an update
+        gauge.markStart();
+        clock.advanceTime(SLEEP / 2, TimeUnit.MILLISECONDS);
+        gauge.markEnd();
+        gauge.update();
+
+        assertThat(gauge.getMaxSingleMeasurement(), is(SLEEP / 2));
+        assertEquals(gauge.getAccumulatedCount(), 3 * SLEEP + SLEEP / 2);
+    }
 }