tie operator liveness more closely to main work queue

Author: rjeberhard

operator/src/main/java/oracle/kubernetes/operator/Main.java

@@ -79,12 +79,14 @@ public Thread newThread(Runnable r) {
   }
 
   private static final ThreadFactory threadFactory = new WrappedThreadFactory();
+  private static final ScheduledExecutorService wrappedExecutorService =
+      Engine.wrappedExecutorService("operator", container);
 
   static final TuningParameters tuningAndConfig;
 
   static {
     try {
-      TuningParameters.initializeInstance(threadFactory, "/operator/config");
+      TuningParameters.initializeInstance(wrappedExecutorService, "/operator/config");
       tuningAndConfig = TuningParameters.getInstance();
     } catch (IOException e) {
       LOGGER.warning(MessageKeys.EXCEPTION, e);
@@ -94,9 +96,6 @@ public Thread newThread(Runnable r) {
 
   private static final CallBuilderFactory callBuilderFactory = new CallBuilderFactory();
 
-  private static final ScheduledExecutorService wrappedExecutorService =
-      Engine.wrappedExecutorService("operator", container);
-
   static {
     container
         .getComponents()
@@ -435,9 +434,8 @@ private static void stopRestServer() {
 
   private static void startLivenessThread() {
     LOGGER.info(MessageKeys.STARTING_LIVENESS_THREAD);
-    livenessThread = new OperatorLiveness();
-    livenessThread.setDaemon(true);
-    livenessThread.start();
+    // every five seconds we need to update the last modified time on the liveness file
+    wrappedExecutorService.scheduleWithFixedDelay(new OperatorLiveness(), 5, 5, TimeUnit.SECONDS);
   }
 
   private static void waitForDeath() {

operator/src/main/java/oracle/kubernetes/operator/OperatorLiveness.java

@@ -12,33 +12,23 @@
 import oracle.kubernetes.operator.logging.MessageKeys;
 
 /**
- * This thread maintains the "liveness" indicator so that Kubernetes knows the Operator is still
+ * This task maintains the "liveness" indicator so that Kubernetes knows the Operator is still
  * alive.
  */
-public class OperatorLiveness extends Thread {
+public class OperatorLiveness implements Runnable {
 
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
   private final File livenessFile = new File("/operator/.alive");
 
   @Override
   public void run() {
-    // every five seconds we need to update the last modified time on the liveness file
-
-    while (true) {
-
-      if (!livenessFile.exists()) {
-        try {
-          livenessFile.createNewFile();
-        } catch (IOException ioe) {
-          LOGGER.info(MessageKeys.COULD_NOT_CREATE_LIVENESS_FILE);
-        }
-      }
-      livenessFile.setLastModified(new Date().getTime());
-
+    if (!livenessFile.exists()) {
       try {
-        Thread.sleep(5000);
-      } catch (InterruptedException ignore) {
+        livenessFile.createNewFile();
+      } catch (IOException ioe) {
+        LOGGER.info(MessageKeys.COULD_NOT_CREATE_LIVENESS_FILE);
       }
     }
+    livenessFile.setLastModified(new Date().getTime());
   }
 }

operator/src/main/java/oracle/kubernetes/operator/TuningParameters.java

@@ -6,13 +6,13 @@
 
 import java.io.IOException;
 import java.util.Map;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ScheduledExecutorService;
 
 public interface TuningParameters extends Map<String, String> {
 
-  static TuningParameters initializeInstance(ThreadFactory factory, String mountPoint)
-      throws IOException {
-    return TuningParametersImpl.initializeInstance(factory, mountPoint);
+  static TuningParameters initializeInstance(
+      ScheduledExecutorService executorService, String mountPoint) throws IOException {
+    return TuningParametersImpl.initializeInstance(executorService, mountPoint);
   }
 
   public static TuningParameters getInstance() {

operator/src/main/java/oracle/kubernetes/operator/TuningParametersImpl.java

@@ -5,7 +5,7 @@
 package oracle.kubernetes.operator;
 
 import java.io.IOException;
-import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import oracle.kubernetes.operator.helpers.ConfigMapConsumer;
@@ -23,10 +23,10 @@ public class TuningParametersImpl extends ConfigMapConsumer implements TuningPar
   private WatchTuning watch = null;
   private PodTuning pod = null;
 
-  static synchronized TuningParameters initializeInstance(ThreadFactory factory, String mountPoint)
-      throws IOException {
+  static synchronized TuningParameters initializeInstance(
+      ScheduledExecutorService executorService, String mountPoint) throws IOException {
     if (INSTANCE == null) {
-      INSTANCE = new TuningParametersImpl(factory, mountPoint);
+      INSTANCE = new TuningParametersImpl(executorService, mountPoint);
       return INSTANCE;
     }
     throw new IllegalStateException();
@@ -36,8 +36,9 @@ public static synchronized TuningParameters getInstance() {
     return INSTANCE;
   }
 
-  private TuningParametersImpl(ThreadFactory factory, String mountPoint) throws IOException {
-    super(factory, mountPoint, TuningParametersImpl::updateTuningParameters);
+  private TuningParametersImpl(ScheduledExecutorService executorService, String mountPoint)
+      throws IOException {
+    super(executorService, mountPoint, TuningParametersImpl::updateTuningParameters);
     update();
   }
 

operator/src/main/java/oracle/kubernetes/operator/helpers/ConfigMapConsumer.java

@@ -18,10 +18,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import oracle.kubernetes.operator.logging.LoggingFacade;
@@ -42,9 +40,10 @@ public class ConfigMapConsumer implements Map<String, String> {
   private final AtomicReference<ScheduledFuture<?>> future = new AtomicReference<>(null);
   private final Runnable onUpdate;
 
-  public ConfigMapConsumer(ThreadFactory factory, String mountPoint, Runnable onUpdate)
+  public ConfigMapConsumer(
+      ScheduledExecutorService executorService, String mountPoint, Runnable onUpdate)
       throws IOException {
-    this.threadPool = Executors.newScheduledThreadPool(2, factory);
+    this.threadPool = executorService;
     this.mountPointDir = new File(mountPoint);
     this.watcher = FileSystems.getDefault().newWatchService();
     this.onUpdate = onUpdate;

operator/src/main/java/oracle/kubernetes/operator/work/Engine.java

@@ -5,8 +5,8 @@
 package oracle.kubernetes.operator.work;
 
 import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -15,9 +15,10 @@ public class Engine {
   private static final int DEFAULT_THREAD_COUNT = 10;
 
   public static ScheduledExecutorService wrappedExecutorService(String id, Container container) {
-    return wrap(
-        container,
-        Executors.newScheduledThreadPool(DEFAULT_THREAD_COUNT, new DaemonThreadFactory(id)));
+    ScheduledThreadPoolExecutor threadPool =
+        new ScheduledThreadPoolExecutor(DEFAULT_THREAD_COUNT, new DaemonThreadFactory(id));
+    threadPool.setRemoveOnCancelPolicy(true);
+    return wrap(container, threadPool);
   }
 
   private volatile ScheduledExecutorService threadPool;