Allow configuring the maximum age of Kubernetes Watch (#6117)

Motivation:

Currently, the KubernetesEndpointGroup watch is designed to stay active
for as long as possible and automatically restart when an error occurs.
However, in practice, maintaining a long-running watch session
inevitably leads to errors, causing it to terminate and restart. From
the perspective of monitoring observers, these errors could be
false-positive noise.

To help prevent these errors, I would suggest periodically closing the
watch when it reaches a certain lifespan and creating a new one.

Modifications:

- Add `maxWatchAge()` option to `KubernetesEndpointGroupBuilder`
  - If unset, the maximum age of a `Watch` defaults to 10 minutes.
- Refactor `KubernetesEndpointGroup` to start a new watch service when
the maximum age is reached.

Result:

You can now set the maximum age of a `Watch` used in
`KubernetesEndpointGroup`.

Author: ikhoon

kubernetes/src/main/java/com/linecorp/armeria/client/kubernetes/endpoints/KubernetesEndpointGroup.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Predicate;
 
 import org.jctools.maps.NonBlockingHashMap;
@@ -215,7 +216,11 @@ public static KubernetesEndpointGroupBuilder builder(Config kubeConfig) {
     @Nullable
     private final String portName;
     private final Predicate<? super NodeAddress> nodeAddressFilter;
+    private final long maxWatchAgeMillis;
 
+    // watchLock may acquire the lock for a long time, so we need to use a separate lock.
+    @SuppressWarnings("PreferReentrantShortLock")
+    private final ReentrantLock watchLock = new ReentrantLock();
     @Nullable
     private volatile Watch nodeWatch;
     @Nullable
@@ -230,10 +235,19 @@ public static KubernetesEndpointGroupBuilder builder(Config kubeConfig) {
     @Nullable
     private volatile Integer nodePort;
 
-    private final ReentrantShortLock lock = new ReentrantShortLock();
-    @GuardedBy("lock")
+    private final ReentrantShortLock schedulerLock = new ReentrantShortLock();
+    @GuardedBy("schedulerLock")
     @Nullable
-    private ScheduledFuture<?> scheduledFuture;
+    private ScheduledFuture<?> updateScheduledFuture;
+    @GuardedBy("schedulerLock")
+    @Nullable
+    private ScheduledFuture<?> serviceScheduledFuture;
+    @GuardedBy("schedulerLock")
+    @Nullable
+    private ScheduledFuture<?> nodeScheduledFuture;
+    @GuardedBy("schedulerLock")
+    @Nullable
+    private ScheduledFuture<?> podScheduledFuture;
 
     private volatile boolean closed;
     private volatile int numServiceFailures;
@@ -243,40 +257,52 @@ public static KubernetesEndpointGroupBuilder builder(Config kubeConfig) {
     KubernetesEndpointGroup(KubernetesClient client, @Nullable String namespace, String serviceName,
                             @Nullable String portName, Predicate<? super NodeAddress> nodeAddressFilter,
                             boolean autoClose, EndpointSelectionStrategy selectionStrategy,
-                            boolean allowEmptyEndpoints, long selectionTimeoutMillis) {
+                            boolean allowEmptyEndpoints, long selectionTimeoutMillis, long maxWatchAgeMillis) {
         super(selectionStrategy, allowEmptyEndpoints, selectionTimeoutMillis);
         this.client = client;
         this.namespace = namespace;
         this.serviceName = serviceName;
         this.portName = portName;
         this.nodeAddressFilter = nodeAddressFilter;
         this.autoClose = autoClose;
+        this.maxWatchAgeMillis = maxWatchAgeMillis == Long.MAX_VALUE ? 0 : maxWatchAgeMillis;
         watchJob(this::watchNode);
         watchJob(this::watchService);
     }
 
     private void watchService() {
-        final Watch oldServiceWatch = serviceWatch;
-        if (oldServiceWatch != null) {
-            oldServiceWatch.close();
-        }
-
-        if (closed) {
-            return;
-        }
-        final Watch newServiceWatch;
+        watchLock.lock();
         try {
-            newServiceWatch = doWatchService();
-        } catch (Exception e) {
-            logger.warn("[{}/{}] Failed to start the service watcher.", namespace, serviceName, e);
-            return;
-        }
-        // Recheck the closed flag because the doWatchService() method may take a while.
-        if (closed) {
-            newServiceWatch.close();
-        } else {
+            final Watch oldServiceWatch = serviceWatch;
+            if (oldServiceWatch != null) {
+                oldServiceWatch.close();
+            }
+            if (closed) {
+                return;
+            }
+            final Watch newServiceWatch;
+            logger.info("[{}/{}] Start the service watcher...", namespace, serviceName);
+            try {
+                newServiceWatch = doWatchService();
+            } catch (Exception e) {
+                logger.warn("[{}/{}] Failed to start the service watcher.", namespace, serviceName, e);
+                return;
+            }
+            // Recheck the closed flag because the doWatchService() method may take a while.
+            if (closed) {
+                newServiceWatch.close();
+                return;
+            }
             serviceWatch = newServiceWatch;
             logger.info("[{}/{}] Service watcher is started.", namespace, serviceName);
+        } finally {
+            watchLock.unlock();
+        }
+
+        if (maxWatchAgeMillis > 0) {
+            logger.debug("[{}/{}] Schedule a new Service watcher to start in {} ms.", namespace,
+                         serviceName, maxWatchAgeMillis);
+            scheduleWatchService(maxWatchAgeMillis);
         }
     }
 
@@ -333,14 +359,17 @@ public void eventReceived(Action action, Service service0) {
 
             @Override
             public void onClose(WatcherException cause) {
+                // Note: As per the JavaDoc of Watcher.onClose(), the method should not be implemented in a
+                //       blocking way.
                 if (closed) {
                     return;
                 }
                 logger.warn("[{}/{}] Service watcher is closed.", namespace, serviceName, cause);
                 logger.info("[{}/{}] Reconnecting the service watcher...", namespace, serviceName);
 
                 // Immediately retry on the first failure.
-                watchJob(() -> watchService(), numServiceFailures++);
+                final long delayMillis = delayMillis(numServiceFailures++);
+                scheduleWatchService(delayMillis);
             }
 
             @Override
@@ -357,27 +386,33 @@ public void onClose() {
     }
 
     private void watchPod() {
-        final Watch oldPodWatch = podWatch;
-        if (oldPodWatch != null) {
-            oldPodWatch.close();
-        }
-
-        if (closed) {
-            return;
-        }
-        final Watch newPodwatch;
+        watchLock.lock();
         try {
-            newPodwatch = doWatchPod();
-        } catch (Exception e) {
-            logger.warn("[{}/{}] Failed to start the pod watcher.", namespace, serviceName, e);
-            return;
-        }
-        // Recheck the closed flag because the doWatchPod() method may take a while.
-        if (closed) {
-            newPodwatch.close();
-        } else {
-            podWatch = newPodwatch;
-            logger.info("[{}/{}] Pod watcher is started.", namespace, serviceName);
+            final Watch oldPodWatch = podWatch;
+            if (oldPodWatch != null) {
+                oldPodWatch.close();
+            }
+
+            if (closed) {
+                return;
+            }
+            final Watch newPodwatch;
+            logger.info("[{}/{}] Start the pod watcher...", namespace, serviceName);
+            try {
+                newPodwatch = doWatchPod();
+            } catch (Exception e) {
+                logger.warn("[{}/{}] Failed to start the pod watcher.", namespace, serviceName, e);
+                return;
+            }
+            // Recheck the closed flag because the doWatchPod() method may take a while.
+            if (closed) {
+                newPodwatch.close();
+            } else {
+                podWatch = newPodwatch;
+                logger.info("[{}/{}] Pod watcher is started.", namespace, serviceName);
+            }
+        } finally {
+            watchLock.unlock();
         }
     }
 
@@ -428,7 +463,8 @@ public void onClose(WatcherException cause) {
                 logger.warn("[{}/{}] Pod watcher is closed.", namespace, serviceName, cause);
                 logger.info("[{}/{}] Reconnecting the pod watcher...", namespace, serviceName);
 
-                watchJob(() -> watchPod(), numPodFailures++);
+                final long delayMillis = delayMillis(numPodFailures++);
+                scheduleWatchPod(delayMillis);
             }
 
             @Override
@@ -449,21 +485,33 @@ public void onClose() {
     }
 
     private void watchNode() {
-        final Watch oldNodeWatch = nodeWatch;
-        if (oldNodeWatch != null) {
-            oldNodeWatch.close();
-        }
+        watchLock.lock();
+        try {
+            final Watch oldNodeWatch = nodeWatch;
+            if (oldNodeWatch != null) {
+                oldNodeWatch.close();
+            }
 
-        if (closed) {
-            return;
-        }
-        final Watch newNodeWatch = doWatchNode();
-        // Recheck the closed flag because the doWatchNode() method may take a while.
-        if (closed) {
-            newNodeWatch.close();
-        } else {
+            if (closed) {
+                return;
+            }
+            logger.info("[{}/{}] Start the node watcher...", namespace, serviceName);
+            final Watch newNodeWatch = doWatchNode();
+            // Recheck the closed flag because the doWatchNode() method may take a while.
+            if (closed) {
+                newNodeWatch.close();
+                return;
+            }
             nodeWatch = newNodeWatch;
             logger.info("[{}/{}] Node watcher is started.", namespace, serviceName);
+        } finally {
+            watchLock.unlock();
+        }
+
+        if (maxWatchAgeMillis > 0) {
+            logger.debug("[{}/{}] Schedule a new Node watcher to start in {} ms.",
+                         namespace, serviceName, maxWatchAgeMillis);
+            scheduleWatchNode(maxWatchAgeMillis);
         }
     }
 
@@ -516,7 +564,8 @@ public void onClose(WatcherException cause) {
                 }
                 logger.warn("[{}/{}] Node watcher is closed.", namespace, serviceName, cause);
                 logger.info("[{}/{}] Reconnecting the node watcher...", namespace, serviceName);
-                watchJob(() -> watchNode(), numNodeFailures++);
+                final long delayMillis = Backoff.ofDefault().nextDelayMillis(numNodeFailures++);
+                scheduleWatchNode(delayMillis);
             }
 
             @Override
@@ -529,23 +578,81 @@ public void onClose() {
     }
 
     private void watchJob(Runnable job) {
-        watchJob(job, 0);
+        final Runnable safeRunnable = safeRunnable(job);
+        worker.execute(safeRunnable);
+    }
+
+    private ScheduledFuture<?> scheduleJob(Runnable job, long delayMillis) {
+        return worker.schedule(safeRunnable(job), delayMillis, TimeUnit.MILLISECONDS);
+    }
+
+    private void scheduleWatchService(long delayMillis) {
+        schedulerLock.lock();
+        try {
+            final ScheduledFuture<?> serviceScheduledFuture = this.serviceScheduledFuture;
+            if (serviceScheduledFuture != null) {
+                serviceScheduledFuture.cancel(false);
+            }
+
+            if (delayMillis == 0) {
+                watchJob(this::watchService);
+                return;
+            }
+            this.serviceScheduledFuture = scheduleJob(this::watchService, delayMillis);
+        } finally {
+            schedulerLock.unlock();
+        }
+    }
+
+    private void scheduleWatchNode(long delayMillis) {
+        schedulerLock.lock();
+        try {
+            final ScheduledFuture<?> nodeScheduledFuture = this.nodeScheduledFuture;
+            if (nodeScheduledFuture != null) {
+                nodeScheduledFuture.cancel(false);
+            }
+            if (delayMillis == 0) {
+                watchJob(this::watchNode);
+                return;
+            }
+            this.nodeScheduledFuture = scheduleJob(this::watchNode, delayMillis);
+        } finally {
+            schedulerLock.unlock();
+        }
     }
 
-    private void watchJob(Runnable job, int numAttempts) {
-        final Runnable safeRunnable = () -> {
+    private void scheduleWatchPod(long delayMillis) {
+        schedulerLock.lock();
+        try {
+            final ScheduledFuture<?> podScheduledFuture = this.podScheduledFuture;
+            if (podScheduledFuture != null) {
+                podScheduledFuture.cancel(false);
+            }
+            if (delayMillis == 0) {
+                watchJob(this::watchPod);
+                return;
+            }
+            this.podScheduledFuture = scheduleJob(this::watchPod, delayMillis);
+        } finally {
+            schedulerLock.unlock();
+        }
+    }
+
+    private Runnable safeRunnable(Runnable job) {
+        return () -> {
             try {
                 job.run();
             } catch (Exception e) {
                 logger.warn("[{}/{}] Failed to run a watch job.", namespace, serviceName, e);
             }
         };
+    }
+
+    private static long delayMillis(int numAttempts) {
         if (numAttempts == 0) {
-            worker.execute(safeRunnable);
-        } else {
-            worker.schedule(safeRunnable, Backoff.ofDefault().nextDelayMillis(numAttempts),
-                            TimeUnit.MILLISECONDS);
+            return 0;
         }
+        return Backoff.ofDefault().nextDelayMillis(numAttempts);
     }
 
     private void maybeUpdateEndpoints(boolean scheduledJob) {
@@ -567,22 +674,22 @@ private void maybeUpdateEndpoints(boolean scheduledJob) {
             return;
         }
 
-        lock.lock();
+        schedulerLock.lock();
         try {
             if (scheduledJob) {
-                scheduledFuture = null;
+                updateScheduledFuture = null;
             } else {
-                if (scheduledFuture != null) {
+                if (updateScheduledFuture != null) {
                     // A scheduled job is already scheduled.
                     return;
                 }
                 // Schedule a job to debounce the update of the endpoints.
-                scheduledFuture = worker.schedule(() -> maybeUpdateEndpoints(true),
-                                                  DEBOUNCE_MILLIS, TimeUnit.MILLISECONDS);
+                updateScheduledFuture = worker.schedule(() -> maybeUpdateEndpoints(true),
+                                                        DEBOUNCE_MILLIS, TimeUnit.MILLISECONDS);
                 return;
             }
         } finally {
-            lock.unlock();
+            schedulerLock.unlock();
         }
 
         assert nodePort != null;