workqueue implementation for kube-style controller

Author: adohe

extended/src/main/java/io/kubernetes/client/extended/function/ConditionFunc.java

@@ -0,0 +1,6 @@
+package io.kubernetes.client.extended.function;
+
+@FunctionalInterface
+public interface ConditionFunc {
+  boolean cond() throws Exception;
+}

extended/src/main/java/io/kubernetes/client/extended/wait/Wait.java

@@ -0,0 +1,51 @@
+package io.kubernetes.client.extended.wait;
+
+import io.kubernetes.client.extended.function.ConditionFunc;
+import java.time.Duration;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class Wait {
+
+  /**
+   * Poll tries a condition func until it returns true, an exception, or the timeout is reached.
+   *
+   * @param interval the check interval
+   * @param timeout the timeout period
+   * @param condition the condition func
+   */
+  public static boolean poll(Duration interval, Duration timeout, ConditionFunc condition) {
+    ExecutorCompletionService<Boolean> completionService =
+        new ExecutorCompletionService<>(
+            Executors.newSingleThreadExecutor(), new LinkedBlockingQueue<>());
+    Timer timer = new Timer("poll-timer", true);
+    try {
+      timer.scheduleAtFixedRate(
+          new TimerTask() {
+            @Override
+            public void run() {
+              completionService.submit(condition::cond);
+            }
+          },
+          interval.toMillis(),
+          interval.toMillis());
+      Future<Boolean> future = completionService.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
+      if (future == null) {
+        return false;
+      }
+      if (future.isDone()) {
+        return true;
+      }
+    } catch (Exception e) {
+      return false;
+    } finally {
+      timer.cancel();
+    }
+    return false;
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueue.java

@@ -0,0 +1,65 @@
+package io.kubernetes.client.extended.workqueue;
+
+import com.google.common.primitives.Longs;
+import java.time.Duration;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/** The default delaying queue implementation. */
+public class DefaultDelayingQueue<T> extends DefaultWorkQueue<T> implements DelayingQueue<T> {
+
+  protected DelayQueue<WaitForEntry> delayQueue = new DelayQueue<>();
+
+  public DefaultDelayingQueue(ExecutorService waitingWorker) {
+    waitingWorker.submit(this::waitingLoop);
+  }
+
+  public void addAfter(T item, Duration duration) {
+    // don't add if we're already shutting down
+    if (super.shuttingDown()) {
+      return;
+    }
+
+    // immediately add things w/o delay
+    if (duration.isZero()) {
+      super.add(item);
+      return;
+    }
+    WaitForEntry entry = new WaitForEntry();
+    entry.data = item;
+    entry.readyAtMillis = System.currentTimeMillis() + duration.toMillis();
+    delayQueue.offer(entry);
+  }
+
+  private void waitingLoop() {
+    try {
+      while (true) {
+        if (super.shuttingDown()) {
+          return;
+        }
+        WaitForEntry entry = delayQueue.take();
+        super.add((T) entry.data);
+      }
+    } catch (InterruptedException e) {
+      // empty block
+    }
+  }
+
+  private static class WaitForEntry implements Delayed {
+
+    private Object data;
+    private long readyAtMillis;
+
+    @Override
+    public long getDelay(TimeUnit unit) {
+      return unit.convert(readyAtMillis - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(Delayed o) {
+      return Longs.compare(readyAtMillis, ((WaitForEntry) o).readyAtMillis);
+    }
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java

@@ -0,0 +1,80 @@
+package io.kubernetes.client.extended.workqueue;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+
+/** The default rate limiting queue implementation. */
+public class DefaultRateLimitingQueue<T> extends DefaultDelayingQueue<T>
+    implements RateLimitingQueue<T> {
+
+  private RateLimiter rateLimiter;
+
+  public DefaultRateLimitingQueue(ExecutorService waitingWorker) {
+    super(waitingWorker);
+    this.rateLimiter = new ExponentialRateLimiter();
+  }
+
+  public DefaultRateLimitingQueue(ExecutorService waitingWorker, RateLimiter rateLimiter) {
+    super(waitingWorker);
+    this.rateLimiter = rateLimiter;
+  }
+
+  @Override
+  public int numRequeues(T item) {
+    return rateLimiter.numRequeues(item);
+  }
+
+  @Override
+  public void forget(Object item) {
+    rateLimiter.forget(item);
+  }
+
+  @Override
+  public void addRateLimited(T item) {
+    super.addAfter(item, rateLimiter.when(item));
+  }
+
+  public static class ExponentialRateLimiter implements RateLimiter {
+
+    Duration baseDelay;
+    Duration maxDelay;
+
+    private Map<Object, Integer> failures = new ConcurrentHashMap<>();
+
+    public ExponentialRateLimiter() {
+      this.baseDelay = Duration.ofMillis(5);
+      this.maxDelay = Duration.ofSeconds(1000);
+    }
+
+    public ExponentialRateLimiter(Duration baseDelay, Duration maxDelay) {
+      this.baseDelay = baseDelay;
+      this.maxDelay = maxDelay;
+    }
+
+    @Override
+    public void forget(Object item) {
+      failures.remove(item);
+    }
+
+    @Override
+    public int numRequeues(Object item) {
+      return failures.get(item);
+    }
+
+    @Override
+    public Duration when(Object item) {
+      Integer exp = failures.getOrDefault(item, 0);
+      failures.put(item, exp + 1);
+      double backoff = baseDelay.toNanos() * Math.pow(2, exp);
+      if (backoff > Long.MAX_VALUE) {
+        return maxDelay;
+      }
+      if (backoff > maxDelay.toNanos()) {
+        return maxDelay;
+      }
+      return Duration.ofNanos(Double.valueOf(backoff).longValue());
+    }
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueue.java

@@ -0,0 +1,91 @@
+package io.kubernetes.client.extended.workqueue;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import java.util.LinkedList;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** The default workqueue implementation. */
+public class DefaultWorkQueue<T> implements WorkQueue<T> {
+
+  // queue defines the order in which we will work on items. Every element of queue
+  // should be in the dirty set and not in the processing set.
+  private LinkedList<T> queue;
+
+  // dirty defines all of the items that need to be processed.
+  private Set<T> dirty;
+
+  // Things that are currently being processed are in the processing set.
+  // These things may be simultaneously in the dirty set. When we finish
+  // processing something and remove it from this set, we'll check if
+  // it's in the dirty set, and if so, add it to the queue.
+  private Set<T> processing;
+
+  private AtomicBoolean shuttingDown = new AtomicBoolean(false);
+
+  public DefaultWorkQueue() {
+    this.queue = Lists.newLinkedList();
+    this.dirty = Sets.newHashSet();
+    this.processing = Sets.newHashSet();
+  }
+
+  @Override
+  public synchronized void add(T item) {
+    if (shuttingDown.get()) {
+      return;
+    }
+
+    if (this.dirty.contains(item)) {
+      return;
+    }
+
+    this.dirty.add(item);
+    if (this.processing.contains(item)) {
+      return;
+    }
+
+    this.queue.add(item);
+    this.notify();
+  }
+
+  @Override
+  public synchronized int length() {
+    return this.queue.size();
+  }
+
+  @Override
+  public synchronized T get() throws InterruptedException {
+    while (queue.size() == 0 && !shuttingDown.get()) {
+      this.wait();
+    }
+    if (queue.size() == 0) {
+      // We must be shutting down
+      return null;
+    }
+    T obj = this.queue.poll();
+    this.processing.add(obj);
+    this.dirty.remove(obj);
+    return obj;
+  }
+
+  @Override
+  public synchronized void done(T item) {
+    this.processing.remove(item);
+    if (this.dirty.contains(item)) {
+      this.queue.add(item);
+      this.notify();
+    }
+  }
+
+  @Override
+  public synchronized void shutDown() {
+    this.shuttingDown.compareAndSet(false, true);
+    this.notifyAll();
+  }
+
+  @Override
+  public synchronized boolean shuttingDown() {
+    return shuttingDown.get();
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/DelayingQueue.java

@@ -0,0 +1,18 @@
+package io.kubernetes.client.extended.workqueue;
+
+import java.time.Duration;
+
+/**
+ * DelayingQueue defines a queue that can Add an item at a later time. This makes it easier to
+ * requeue items after failures without ending up in a hot-loop.
+ */
+public interface DelayingQueue<T> extends WorkQueue<T> {
+
+  /**
+   * addAfter adds an item to the workqueue after the indicated duration has passed.
+   *
+   * @param item item to add
+   * @param duration specific duration
+   */
+  void addAfter(T item, Duration duration);
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/RateLimiter.java

@@ -0,0 +1,30 @@
+package io.kubernetes.client.extended.workqueue;
+
+import java.time.Duration;
+
+public interface RateLimiter {
+
+  /**
+   * when gets an item and gets to decide how long that item should wait.
+   *
+   * @param item specific item
+   * @return how long the item should wait
+   */
+  Duration when(Object item);
+
+  /**
+   * forget indicates that an item is finished being retried. Doesn't matter whether its for perm
+   * failing or for success, we'll stop tracking it
+   *
+   * @param item item that is finished being retried
+   */
+  void forget(Object item);
+
+  /**
+   * numRequeues returns back how many failures the item has had.
+   *
+   * @param item specific item
+   * @return how many failures the item has had
+   */
+  int numRequeues(Object item);
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/RateLimitingQueue.java

@@ -0,0 +1,29 @@
+package io.kubernetes.client.extended.workqueue;
+
+/** RateLimitingQueue defines a queue that rate limits items being added to the queue. */
+public interface RateLimitingQueue<T> extends DelayingQueue<T> {
+
+  /**
+   * addRateLimited adds an item to the workqueue after the rate limiter says its ok.
+   *
+   * @param item item to add
+   */
+  void addRateLimited(T item);
+
+  /**
+   * forget indicates that an item is finished being retried. Doesn't matter whether its for perm
+   * failing or for success, we'll stop the rate limiter from tracking it. This only clears the
+   * `rateLimiter`, you still have to call `Done` on the queue.
+   *
+   * @param item item which is finished being retried
+   */
+  void forget(T item);
+
+  /**
+   * numRequeues returns back how many times the item was requeued.
+   *
+   * @param item specific item
+   * @return times the item was requeued
+   */
+  int numRequeues(T item);
+}