workqueue implementation for kube-style controller

Author: adohe

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

@@ -0,0 +1,49 @@
+package io.kubernetes.client.extended.wait;
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+
+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, Supplier<Boolean> condition) {
+    ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+    AtomicBoolean result = new AtomicBoolean(false);
+    long dueDate = System.currentTimeMillis() + timeout.toMillis();
+    ScheduledFuture<?> future =
+        executorService.scheduleAtFixedRate(
+            () -> {
+              try {
+                result.set(condition.get());
+              } catch (Exception e) {
+                result.set(false);
+              }
+            },
+            interval.toMillis(),
+            interval.toMillis(),
+            TimeUnit.MILLISECONDS);
+    try {
+      while (System.currentTimeMillis() < dueDate) {
+        if (result.get()) {
+          future.cancel(true);
+          return true;
+        }
+      }
+    } catch (Exception e) {
+      return result.get();
+    }
+    future.cancel(true);
+    return result.get();
+  }
+}

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

@@ -0,0 +1,95 @@
+package io.kubernetes.client.extended.workqueue;
+
+import com.google.common.primitives.Longs;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.Temporal;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+/** The default delaying queue implementation. */
+public class DefaultDelayingQueue<T> extends DefaultWorkQueue<T> implements DelayingQueue<T> {
+
+  private DelayQueue<WaitForEntry> delayQueue;
+  protected BlockingQueue<WaitForEntry> waitingForAddQueue;
+
+  public DefaultDelayingQueue(ExecutorService waitingWorker) {
+    this.delayQueue = new DelayQueue<>();
+    this.waitingForAddQueue = new LinkedBlockingQueue<>();
+    waitingWorker.submit(this::waitingLoop);
+  }
+
+  public DefaultDelayingQueue() {
+    this(Executors.newSingleThreadExecutor());
+  }
+
+  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 = duration.addTo(Instant.now());
+    this.waitingForAddQueue.add(entry);
+  }
+
+  private void waitingLoop() {
+    try {
+      while (true) {
+        if (super.shuttingDown()) {
+          return;
+        }
+        WaitForEntry entry = delayQueue.peek();
+        Duration nextReadyAt = Duration.ofMillis(10 * 1000L);
+        if (entry != null) {
+          Instant now = Instant.now();
+          if (!Duration.between(entry.readyAtMillis, now).isNegative()) {
+            delayQueue.remove(entry);
+            super.add((T) entry.data);
+          } else {
+            nextReadyAt = Duration.between(now, entry.readyAtMillis);
+          }
+        }
+
+        WaitForEntry waitForEntry =
+            waitingForAddQueue.poll(nextReadyAt.toMillis(), TimeUnit.MILLISECONDS);
+        if (waitForEntry != null
+            && Duration.between(waitForEntry.readyAtMillis, Instant.now()).isNegative()) {
+          this.delayQueue.offer(waitForEntry);
+        }
+      }
+    } catch (InterruptedException e) {
+      // empty block
+    }
+  }
+
+  // WaitForEntry holds the data to add and the time it should be added.
+  private static class WaitForEntry implements Delayed {
+
+    private Object data;
+    private Temporal readyAtMillis;
+
+    @Override
+    public long getDelay(TimeUnit unit) {
+      Duration duration = Duration.between(Instant.now(), readyAtMillis);
+      return unit.convert(duration.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(Delayed o) {
+      return Longs.compare(getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
+    }
+  }
+}

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,97 @@
+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 {@link WorkQueue} implementation that uses a doubly-linked list to store work items.
+ * This class ensures the added work items are, not in dirty set or currently processing set, before
+ * append them to the list.
+ *
+ * <p>Usage example, based on a typical producer-consumer scenario. </pre>
+ */
+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);
+}