Add token bucket rate limiter

Author: cizezsy

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/BucketRateLimiter.java

@@ -0,0 +1,84 @@
+package io.kubernetes.client.extended.workqueue.ratelimiter;
+
+import java.time.Duration;
+import java.time.Instant;
+
+/** A light-weight token bucket implementation for RateLimiter. */
+public class BucketRateLimiter<T> implements RateLimiter<T> {
+  public static final double INFINITE_LIMIT = Double.POSITIVE_INFINITY;
+
+  private long limit;
+
+  private long capacity;
+  private long tokens;
+
+  private Instant last;
+
+  /**
+   * @param limit Limit defines the maximum frequency of some events. Limit is represented as number
+   *     of events per second. A zero Limit allows no events.
+   * @param capacity Capacity is the maximum number of tokens can be consumed.
+   */
+  public BucketRateLimiter(long limit, long capacity) {
+    this.limit = limit;
+    this.capacity = capacity;
+
+    last = Instant.EPOCH;
+  }
+
+  @Override
+  public Duration when(T item) {
+    // if limit is infinite, it allows all event
+    if (limit == INFINITE_LIMIT) {
+      return Duration.ZERO;
+    }
+
+    // if capacity isn't positive, it will return the infinite delay
+    if (capacity <= 0) {
+      return Duration.ofSeconds(Long.MAX_VALUE);
+    }
+
+    Instant now = Instant.now();
+    if (now.isBefore(last)) {
+      last = now;
+    }
+
+    // Avoid making delta overflow below when last is very old.
+    Duration maxElapsed = durationFromTokens(capacity - tokens);
+    Duration elapsed = Duration.between(last, now);
+    if (elapsed.compareTo(maxElapsed) > 0) {
+      elapsed = maxElapsed;
+    }
+
+    // Calculate the new number of tokens, due to time that passed.
+    long delta = tokensFromDurations(elapsed);
+    tokens += delta;
+    if (tokens > capacity) {
+      tokens = capacity;
+    }
+
+    tokens -= 1;
+    if (tokens < 0) {
+      return durationFromTokens(-tokens);
+    }
+
+    return Duration.ZERO;
+  }
+
+  @Override
+  public void forget(T item) {}
+
+  @Override
+  public int numRequests(T item) {
+    return 0;
+  }
+
+  private Duration durationFromTokens(long tokens) {
+    long seconds = (tokens / limit);
+    return Duration.ofSeconds(seconds);
+  }
+
+  private long tokensFromDurations(Duration d) {
+    return (d.getSeconds() * limit);
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/DefaultControllerRateLimiter.java

@@ -0,0 +1,37 @@
+package io.kubernetes.client.extended.workqueue.ratelimiter;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * DefaultControllerRateLimiter is a default rate limiter for workqueue. It has both overall and
+ * per-item rate limiting. The overall is a token bucket and the per-item is exponential
+ */
+public class DefaultControllerRateLimiter<T> implements RateLimiter<T> {
+
+  private RateLimiter<T> internalRateLimiter;
+
+  public DefaultControllerRateLimiter() {
+    this.internalRateLimiter =
+        new MaxOfRateLimiter<>(
+            Arrays.asList(
+                new ItemExponentialFailureRateLimiter<>(
+                    Duration.ofMillis(5), Duration.ofSeconds(1000)),
+                new BucketRateLimiter<>(10, 100)));
+  }
+
+  @Override
+  public Duration when(T item) {
+    return internalRateLimiter.when(item);
+  }
+
+  @Override
+  public void forget(T item) {
+    internalRateLimiter.forget(item);
+  }
+
+  @Override
+  public int numRequests(T item) {
+    return internalRateLimiter.numRequests(item);
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/MaxOfRateLimiter.java

@@ -30,10 +30,7 @@ public Duration when(T item) {
 
   @Override
   public void forget(T item) {
-    rateLimiters.forEach(
-        r -> {
-          r.forget(item);
-        });
+    rateLimiters.forEach(r -> r.forget(item));
   }
 
   @Override

extended/src/test/java/io/kubernetes/client/extended/workqueue/DelayingWorkQueueTest.java

@@ -19,7 +19,7 @@ public void setUp() {
   public void testAddAfterWithinDelay() throws InterruptedException {
     String testStr = "test";
 
-    delayingWorkQueue.addAfter(testStr, Duration.ofSeconds(10));
+    delayingWorkQueue.addAfter(testStr, Duration.ofSeconds(5));
     assertEquals(0, delayingWorkQueue.size());
 
     Thread.sleep(2000);