HDDS-12852. Implement a sliding window counter utility

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/utils/SlidingWindow.java

@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.utils;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import org.apache.hadoop.util.Time;
+
+/**
+ * A time-based sliding window implementation that tracks event timestamps.
+ */
+public class SlidingWindow {
+  private final Object lock = new Object();
+  private final int windowSize;
+  private final Deque<Long> timestamps;
+  private final long expiryDurationMillis;
+  private final Clock clock;
+
+  /**
+   * Default constructor that uses system clock.
+   *
+   * @param windowSize     the maximum number of events that are tracked
+   * @param expiryDuration the duration after which an entry in the window expires
+   */
+  public SlidingWindow(int windowSize, Duration expiryDuration) {
+    this(windowSize, expiryDuration, new SystemClock());
+  }
+
+  /**
+   * Constructor with custom clock for testing.
+   *
+   * @param windowSize     the maximum number of events that are tracked
+   * @param expiryDuration the duration after which an entry in the window expires
+   * @param clock          the clock to use for time measurements
+   */
+  public SlidingWindow(int windowSize, Duration expiryDuration, Clock clock) {
+    if (windowSize <= 0) {
+      throw new IllegalArgumentException("Window size must be greater than 0");
+    }
+    if (expiryDuration.isNegative() || expiryDuration.isZero()) {
+      throw new IllegalArgumentException("Expiry duration must be greater than 0");
+    }
+    this.windowSize = windowSize;
+    this.expiryDurationMillis = expiryDuration.toMillis();
+    this.clock = clock;
+    // We limit the initial queue size to 100 to control the memory usage
+    this.timestamps = new ArrayDeque<>(Math.min(windowSize + 1, 100));
+  }
+
+  public void add() {
+    synchronized (lock) {
+      if (isFull()) {
+        timestamps.remove();
+      }
+
+      timestamps.add(getCurrentTime());
+    }
+  }
+
+  /**
+   * Checks if the sliding window has exceeded its maximum size.
+   * This is useful to track if we have encountered more events than the window's defined limit.
+   * @return true if the number of tracked timestamps in the sliding window
+   *         exceeds the specified window size, false otherwise.
+   */
+  public boolean isFull() {
+    synchronized (lock) {
+      removeExpired();
+      return timestamps.size() > windowSize;
+    }
+  }
+
+  private void removeExpired() {
+    synchronized (lock) {
+      long currentTime = getCurrentTime();
+      long expirationThreshold = currentTime - expiryDurationMillis;
+
+      while (!timestamps.isEmpty() && timestamps.peek() < expirationThreshold) {
+        timestamps.remove();
+      }
+    }
+  }
+
+  public int getWindowSize() {
+    return windowSize;
+  }
+
+  private long getCurrentTime() {
+    return clock.millis();
+  }
+
+  /**
+   * Implementation of Clock that uses Time.monotonicNow() for real usage.
+   */
+  private static final class SystemClock extends Clock {
+    @Override
+    public long millis() {
+      return Time.monotonicNow();
+    }
+
+    @Override
+    public java.time.Instant instant() {
+      return java.time.Instant.ofEpochMilli(millis());
+    }
+
+    @Override
+    public java.time.ZoneId getZone() {
+      return java.time.ZoneOffset.UTC;
+    }
+
+    @Override
+    public Clock withZone(java.time.ZoneId zone) {
+      return this; // Ignore zone for monotonic clock
+    }
+  }
+}

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/utils/TestSlidingWindow.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.container.common.utils;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import org.apache.ozone.test.TestClock;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link SlidingWindow} class.
+ */
+class TestSlidingWindow {
+
+  private SlidingWindow slidingWindow;
+  private TestClock testClock;
+
+  @Test
+  void testConstructorValidation() {
+    // Test invalid window size
+    assertThrows(IllegalArgumentException.class, () -> new SlidingWindow(0, Duration.ofMillis(100)));
+    assertThrows(IllegalArgumentException.class, () -> new SlidingWindow(-1, Duration.ofMillis(100)));
+
+    // Test invalid expiry duration
+    assertThrows(IllegalArgumentException.class, () -> new SlidingWindow(1, Duration.ofMillis(0)));
+    assertThrows(IllegalArgumentException.class, () -> new SlidingWindow(1, Duration.ofMillis(-1)));
+  }
+
+  @Test
+  void testAdd() {
+    testClock = TestClock.newInstance();
+    slidingWindow = new SlidingWindow(3, Duration.ofSeconds(5), testClock);
+    for (int i = 0; i < slidingWindow.getWindowSize(); i++) {
+      slidingWindow.add();
+      assertFalse(slidingWindow.isFull());
+    }
+
+    slidingWindow.add();
+    assertTrue(slidingWindow.isFull());
+  }
+
+  @Test
+  void testEventExpiration() {
+    testClock = TestClock.newInstance();
+    slidingWindow = new SlidingWindow(2, Duration.ofMillis(500), testClock);
+
+    // Add events to reach threshold
+    slidingWindow.add();
+    slidingWindow.add();
+    slidingWindow.add();
+    assertTrue(slidingWindow.isFull());
+
+    // Fast forward time to expire events
+    testClock.fastForward(600);
+
+    assertFalse(slidingWindow.isFull());
+
+    // Add one more event - should not be enough to mark as full
+    slidingWindow.add();
+    assertFalse(slidingWindow.isFull());
+  }
+
+  @Test
+  void testPartialExpiration() {
+    testClock = TestClock.newInstance();
+    slidingWindow = new SlidingWindow(3, Duration.ofSeconds(1), testClock);
+
+    slidingWindow.add();
+    slidingWindow.add();
+    slidingWindow.add();
+    slidingWindow.add();
+    assertTrue(slidingWindow.isFull());
+
+    testClock.fastForward(600);
+    slidingWindow.add(); // this will remove the oldest event as the window is full
+
+    // Fast forward time to expire the oldest events
+    testClock.fastForward(500);
+    assertFalse(slidingWindow.isFull());
+  }
+}

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/utils/package-info.java

@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/** Tests for Common container utils. */
+package org.apache.hadoop.ozone.container.common.utils;