Merge pull request #1026 from yue9944882/feat/exception-handler

Reflector and leader-elector exception handlers

Author: k8s-ci-robot

extended/src/main/java/io/kubernetes/client/extended/leaderelection/LeaderElector.java

@@ -9,6 +9,7 @@
 import java.util.List;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -23,6 +24,7 @@ public class LeaderElector {
   // internal bookkeeping
   private LeaderElectionRecord observedRecord;
   private long observedTimeMilliSeconds;
+  private final Consumer<Throwable> exceptionHandler;
 
   private ScheduledExecutorService scheduledWorkers =
       Executors.newSingleThreadScheduledExecutor(
@@ -33,6 +35,14 @@ public class LeaderElector {
       Executors.newSingleThreadExecutor(makeThreadFactory("leader-elector-hook-worker-%d"));
 
   public LeaderElector(LeaderElectionConfig config) {
+    this(
+        config,
+        (t) -> {
+          log.error("Unexpected error on acquiring or renewing the lease", t);
+        });
+  }
+
+  public LeaderElector(LeaderElectionConfig config, Consumer<Throwable> exceptionHandler) {
     if (config == null) {
       throw new IllegalArgumentException("Config must be provided.");
     }
@@ -68,6 +78,7 @@ public LeaderElector(LeaderElectionConfig config) {
       throw new IllegalArgumentException(String.join(",", errors));
     }
     this.config = config;
+    this.exceptionHandler = exceptionHandler;
   }
 
   public void run(Runnable startLeadingHook, Runnable stopLeadingHook) {
@@ -110,7 +121,7 @@ private boolean acquire() {
               } catch (CancellationException e) {
                 log.info("Processing tryAcquireOrRenew successfully canceled");
               } catch (Throwable t) {
-                log.error("Error processing tryAcquireOrRenew as {}", t.getMessage());
+                this.exceptionHandler.accept(t);
                 future.cancel(true);
               }
             },

extended/src/test/java/io/kubernetes/client/extended/leaderelection/LeaderElectionTest.java

@@ -1,19 +1,27 @@
 package io.kubernetes.client.extended.leaderelection;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
+import io.kubernetes.client.openapi.ApiException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 
+@RunWith(MockitoJUnitRunner.class)
 public class LeaderElectionTest {
 
   @Before
@@ -22,6 +30,8 @@ public void setUp() {
     MockResourceLock.leaderRecord = null;
   }
 
+  @Mock private Lock lock;
+
   @Test
   public void testSimpleLeaderElection() throws InterruptedException {
     List<String> electionHistory = new ArrayList<>();
@@ -250,6 +260,28 @@ private void assertHistory(List<String> history, String... expected) {
     }
   }
 
+  @Test
+  public void testLeaderElectionCaptureException() throws ApiException, InterruptedException {
+    RuntimeException expectedException = new RuntimeException("noxu");
+    AtomicReference<Throwable> actualException = new AtomicReference<>();
+    when(lock.get()).thenThrow(expectedException);
+    LeaderElectionConfig leaderElectionConfig = new LeaderElectionConfig();
+    leaderElectionConfig.setLock(lock);
+    leaderElectionConfig.setLeaseDuration(Duration.ofMillis(1000));
+    leaderElectionConfig.setRetryPeriod(Duration.ofMillis(200));
+    leaderElectionConfig.setRenewDeadline(Duration.ofMillis(700));
+    LeaderElector leaderElector =
+        new LeaderElector(leaderElectionConfig, (t) -> actualException.set(t));
+
+    ExecutorService leaderElectionWorker = Executors.newFixedThreadPool(1);
+    leaderElectionWorker.submit(
+        () -> {
+          leaderElector.run(() -> {}, () -> {});
+        });
+    Thread.sleep(Duration.ofSeconds(2).toMillis());
+    assertEquals(expectedException, actualException.get().getCause());
+  }
+
   public static class MockResourceLock implements Lock {
 
     public static ReentrantLock lock;

util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java

@@ -6,11 +6,13 @@
 import io.kubernetes.client.informer.ListerWatcher;
 import io.kubernetes.client.openapi.models.V1ListMeta;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
-import io.kubernetes.client.util.*;
+import io.kubernetes.client.util.CallGeneratorParams;
+import io.kubernetes.client.util.Watchable;
 import java.net.ConnectException;
 import java.time.Duration;
-import java.util.*;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -31,21 +33,32 @@ public class ReflectorRunnable<
 
   private AtomicBoolean isActive = new AtomicBoolean(true);
 
+  private final BiConsumer<Class<ApiType>, Throwable> exceptionHandler;
+
   public ReflectorRunnable(
       Class<ApiType> apiTypeClass, ListerWatcher listerWatcher, DeltaFIFO store) {
+    this(apiTypeClass, listerWatcher, store, ReflectorRunnable::defaultWatchErrorHandler);
+  }
+
+  public ReflectorRunnable(
+      Class<ApiType> apiTypeClass,
+      ListerWatcher listerWatcher,
+      DeltaFIFO store,
+      BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
     this.listerWatcher = listerWatcher;
     this.store = store;
     this.apiTypeClass = apiTypeClass;
+    this.exceptionHandler = exceptionHandler;
   }
 
   /**
    * run first lists all items and get the resource version at the moment of call, and then use the
    * resource version to watch.
    */
   public void run() {
-    try {
-      log.info("{}#Start listing and watching...", apiTypeClass);
+    log.info("{}#Start listing and watching...", apiTypeClass);
 
+    try {
       ApiListType list = listerWatcher.list(new CallGeneratorParams(Boolean.FALSE, null, null));
 
       V1ListMeta listMeta = list.getMetadata();
@@ -82,25 +95,27 @@ public void run() {
                       Long.valueOf(Duration.ofMinutes(5).toMillis()).intValue()));
           watchHandler(watch);
         } catch (Throwable t) {
-          log.info("{}#Watch connection get exception {}", apiTypeClass, t.getMessage());
-          Throwable cause = t.getCause();
-          // If this is "connection refused" error, it means that most likely apiserver is not
-          // responsive.
-          // It doesn't make sense to re-list all objects because most likely we will be able to
-          // restart
-          // watch where we ended.
-          // If that's the case wait and resend watch request.
-          if (cause != null && (cause instanceof ConnectException)) {
-            log.info("{}#Watch get connect exception, retry watch", apiTypeClass);
-            Thread.sleep(1000L);
+          if (isConnectException(t)) {
+            // If this is "connection refused" error, it means that most likely apiserver is not
+            // responsive.
+            // It doesn't make sense to re-list all objects because most likely we will be able to
+            // restart
+            // watch where we ended.
+            // If that's the case wait and resend watch request.
+            log.info("{}#Watch get connect exception, retry watch", this.apiTypeClass);
+            try {
+              Thread.sleep(1000L);
+            } catch (InterruptedException e) {
+              // no-op
+            }
             continue;
           }
           if ((t instanceof RuntimeException)
               && t.getMessage().contains("IO Exception during hasNext")) {
-            log.info("{}#Read timeout retry list and watch", apiTypeClass);
+            log.info("{}#Read timeout retry list and watch", this.apiTypeClass);
             return;
           }
-          log.error("{}#Watch failed as {} unexpected", apiTypeClass, t.getMessage(), t);
+          this.exceptionHandler.accept(apiTypeClass, t);
           return;
         } finally {
           if (watch != null) {
@@ -110,7 +125,7 @@ public void run() {
         }
       }
     } catch (Throwable t) {
-      log.error("Failed to list-watch", t);
+      this.exceptionHandler.accept(apiTypeClass, t);
     }
   }
 
@@ -168,4 +183,18 @@ private void watchHandler(Watchable<ApiType> watch) {
       }
     }
   }
+
+  private static <ApiType extends KubernetesObject> void defaultWatchErrorHandler(
+      Class<ApiType> watchingApiTypeClass, Throwable t) {
+    log.error(String.format("%s#Reflector loop failed unexpectedly", watchingApiTypeClass), t);
+  }
+
+  private boolean isConnectException(Throwable t) {
+    if (t instanceof ConnectException) {
+      return true;
+    }
+    // ApiException can nest a ConnectException
+    Throwable cause = t.getCause();
+    return cause instanceof ConnectException;
+  }
 }

util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java

@@ -6,10 +6,14 @@
 import io.kubernetes.client.informer.EventType;
 import io.kubernetes.client.informer.ListerWatcher;
 import io.kubernetes.client.openapi.ApiException;
-import io.kubernetes.client.openapi.models.*;
+import io.kubernetes.client.openapi.models.V1ListMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+import io.kubernetes.client.openapi.models.V1Status;
 import io.kubernetes.client.util.CallGeneratorParams;
 import io.kubernetes.client.util.Watch;
 import io.kubernetes.client.util.Watchable;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.EnvironmentVariables;
@@ -90,4 +94,55 @@ public Watchable<V1Pod> watch(CallGeneratorParams params) throws ApiException {
       reflectorRunnable.stop();
     }
   }
+
+  @Test
+  public void testReflectorRunnableCaptureListException()
+      throws ApiException, InterruptedException {
+    RuntimeException expectedException = new RuntimeException("noxu");
+    AtomicReference<Throwable> actualException = new AtomicReference<>();
+    when(listerWatcher.list(any())).thenThrow(expectedException);
+    ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
+        new ReflectorRunnable<>(
+            V1Pod.class,
+            listerWatcher,
+            deltaFIFO,
+            (apiType, t) -> {
+              actualException.set(t);
+            });
+    try {
+      Thread thread = new Thread(reflectorRunnable::run);
+      thread.setDaemon(true);
+      thread.start();
+      Thread.sleep(1000);
+    } finally {
+      reflectorRunnable.stop();
+    }
+    assertEquals(expectedException, actualException.get());
+  }
+
+  @Test
+  public void testReflectorRunnableCaptureWatchException()
+      throws ApiException, InterruptedException {
+    RuntimeException expectedException = new RuntimeException("noxu");
+    AtomicReference<Throwable> actualException = new AtomicReference<>();
+    when(listerWatcher.list(any())).thenReturn(new V1PodList().metadata(new V1ListMeta()));
+    when(listerWatcher.watch(any())).thenThrow(expectedException);
+    ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
+        new ReflectorRunnable<>(
+            V1Pod.class,
+            listerWatcher,
+            deltaFIFO,
+            (apiType, t) -> {
+              actualException.set(t);
+            });
+    try {
+      Thread thread = new Thread(reflectorRunnable::run);
+      thread.setDaemon(true);
+      thread.start();
+      Thread.sleep(1000);
+    } finally {
+      reflectorRunnable.stop();
+    }
+    assertEquals(expectedException, actualException.get());
+  }
 }