workqueue implementation for kube-style controller

Author: adohe

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

@@ -0,0 +1,68 @@
+package io.kubernetes.client.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> {
+
+    private 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/workqueue/DefaultRateLimitingQueue.java

@@ -0,0 +1,81 @@
+package io.kubernetes.client.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/workqueue/DefaultWorkQueue.java

@@ -0,0 +1,94 @@
+package io.kubernetes.client.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/workqueue/DelayingQueue.java

@@ -0,0 +1,18 @@
+package io.kubernetes.client.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/workqueue/RateLimiter.java

@@ -0,0 +1,30 @@
+package io.kubernetes.client.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/workqueue/RateLimitingQueue.java

@@ -0,0 +1,31 @@
+package io.kubernetes.client.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);
+}

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

@@ -0,0 +1,49 @@
+package io.kubernetes.client.workqueue;
+
+/**
+ * The workqueue interface defines the queue behavior.
+ */
+public interface WorkQueue<T> {
+
+    /**
+     * add marks item as needing processing.
+     *
+     * @param item item to add
+     */
+    void add(T item);
+
+    /**
+     * length returns the current queue length, for informational
+     * purposes only.
+     *
+     * @return current queue length
+     */
+    int length();
+
+    /**
+     * Get blocks until it can return an item to be processed. If shutdown = true,
+     * the caller should end their process.
+     *
+     * @return the object
+     */
+    T get() throws InterruptedException;
+
+    /**
+     * Done marks item as done processing, and if it has been marked as dirty again,
+     *
+     * @param item specific item
+     */
+    void done(T item);
+
+    /**
+     * ShutDown will cause q to ignore all new items added to it.
+     */
+    void shutDown();
+
+    /**
+     * returns whether the queue shutdown.
+     *
+     * @return returns {@code true} if the
+     */
+    boolean shuttingDown();
+}