Merge pull request #525 from adohe/master

add workqueue implementation for kube-style controller

Author: k8s-ci-robot

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,136 @@
+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.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+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> {
+
+  public static Duration heartBeatInterval = Duration.ofSeconds(10);
+
+  private DelayQueue<WaitForEntry<T>> delayQueue;
+  private ConcurrentMap<T, WaitForEntry<T>> waitingEntryByData;
+  protected BlockingQueue<WaitForEntry<T>> waitingForAddQueue;
+
+  public DefaultDelayingQueue(ExecutorService waitingWorker) {
+    this.delayQueue = new DelayQueue<>();
+    this.waitingEntryByData = new ConcurrentHashMap<>();
+    this.waitingForAddQueue = new LinkedBlockingQueue<>(1000);
+    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.isShuttingDown()) {
+      return;
+    }
+
+    // immediately add things w/o delay
+    if (duration.isZero()) {
+      super.add(item);
+      return;
+    }
+    WaitForEntry<T> entry = new WaitForEntry<>(item, duration.addTo(Instant.now()));
+    this.waitingForAddQueue.offer(entry);
+  }
+
+  private void waitingLoop() {
+    try {
+      while (true) {
+        // underlying work-queue is shutting down, quit the loop.
+        if (super.isShuttingDown()) {
+          return;
+        }
+        // peek the first item from the delay queue
+        WaitForEntry<T> entry = delayQueue.peek();
+        // default next ready-at time to "never"
+        Duration nextReadyAt = heartBeatInterval;
+        if (entry != null) {
+          // the delay-queue isn't empty, so we deal with the item in the following logic:
+          // 1. check if the item is ready to fire
+          //   a. if ready, remove it from the delay-queue and push it into underlying work-queue
+          //   b. if not, refresh the next ready-at time.
+          Instant now = Instant.now();
+          if (!Duration.between(entry.readyAtMillis, now).isNegative()) {
+            delayQueue.remove(entry);
+            super.add(entry.data);
+            this.waitingEntryByData.remove(entry.data);
+            continue;
+          } else {
+            nextReadyAt = Duration.between(now, entry.readyAtMillis);
+          }
+        }
+
+        WaitForEntry<T> waitForEntry =
+            waitingForAddQueue.poll(nextReadyAt.toMillis(), TimeUnit.MILLISECONDS);
+        if (waitForEntry != null) {
+          if (Duration.between(waitForEntry.readyAtMillis, Instant.now()).isNegative()) {
+            // the item is not yet ready, insert it to the delay-queue
+            insert(this.delayQueue, this.waitingEntryByData, waitForEntry);
+          } else {
+            // the item is ready as soon as received, fire it to the work-queue directly
+            super.add(waitForEntry.data);
+          }
+        }
+      }
+    } catch (InterruptedException e) {
+      // empty block
+    }
+  }
+
+  private void insert(
+      DelayQueue<WaitForEntry<T>> q, Map<T, WaitForEntry<T>> knownEntries, WaitForEntry entry) {
+    WaitForEntry existing = knownEntries.get((T) entry.data);
+    if (existing != null) {
+      if (Duration.between(existing.readyAtMillis, entry.readyAtMillis).isNegative()) {
+        q.remove(existing);
+        existing.readyAtMillis = entry.readyAtMillis;
+        q.add(existing);
+      }
+
+      return;
+    }
+
+    q.offer(entry);
+    knownEntries.put((T) entry.data, entry);
+  }
+
+  // WaitForEntry holds the data to add and the time it should be added.
+  private class WaitForEntry<T> implements Delayed {
+
+    private WaitForEntry(T data, Temporal readyAtMillis) {
+      this.data = data;
+      this.readyAtMillis = readyAtMillis;
+    }
+
+    private T 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,96 @@
+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;
+
+/**
+ * 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 boolean shuttingDown = false;
+
+  public DefaultWorkQueue() {
+    this.queue = Lists.newLinkedList();
+    this.dirty = Sets.newHashSet();
+    this.processing = Sets.newHashSet();
+  }
+
+  @Override
+  public synchronized void add(T item) {
+    if (shuttingDown) {
+      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) {
+      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 = true;
+    this.notifyAll();
+  }
+
+  @Override
+  public synchronized boolean isShuttingDown() {
+    return shuttingDown;
+  }
+}

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);
+}