Refactor away from ConfigHolder, ConfigHelper

Author: rjeberhard

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

@@ -9,6 +9,7 @@
 import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.watcher.WatchListener;
 
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -18,9 +19,9 @@
 public class ConfigMapWatcher extends Watcher<V1ConfigMap> {
   private final String ns;
 
-  public static ConfigMapWatcher create(String ns, String initialResourceVersion, WatchListener<V1ConfigMap> listener, AtomicBoolean isStopping) {
+  public static ConfigMapWatcher create(ScheduledExecutorService threadPool, String ns, String initialResourceVersion, WatchListener<V1ConfigMap> listener, AtomicBoolean isStopping) {
     ConfigMapWatcher watcher = new ConfigMapWatcher(ns, initialResourceVersion, listener, isStopping);
-    watcher.start("Thread-ConfigMapWatcher-" + ns);
+    watcher.start(threadPool);
     return watcher;
   }
 

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

@@ -24,6 +24,7 @@
 import oracle.kubernetes.weblogic.domain.v1.ServerStartup;
 import oracle.kubernetes.weblogic.domain.v1.ServerStatus;
 import oracle.kubernetes.operator.helpers.CallBuilder;
+import oracle.kubernetes.operator.helpers.CallBuilderFactory;
 import oracle.kubernetes.operator.helpers.DomainPresenceInfo;
 import oracle.kubernetes.operator.helpers.DomainPresenceInfo.ServerStartupInfo;
 import oracle.kubernetes.operator.helpers.ResponseStep;
@@ -34,6 +35,7 @@
 import oracle.kubernetes.operator.wlsconfig.WlsClusterConfig;
 import oracle.kubernetes.operator.wlsconfig.WlsDomainConfig;
 import oracle.kubernetes.operator.wlsconfig.WlsServerConfig;
+import oracle.kubernetes.operator.work.ContainerResolver;
 import oracle.kubernetes.operator.work.Fiber;
 import oracle.kubernetes.operator.work.Fiber.CompletionCallback;
 import oracle.kubernetes.operator.work.NextAction;
@@ -610,7 +612,8 @@ public NextAction apply(Packet packet) {
   private static NextAction doDomainUpdate(Domain dom, DomainPresenceInfo info, Packet packet, Step conflictStep, Step next) {
     V1ObjectMeta meta = dom.getMetadata();
     NextAction na = new NextAction();
-    na.invoke(CallBuilder.create().replaceDomainAsync(meta.getName(), meta.getNamespace(), dom, new ResponseStep<Domain>(next) {
+    CallBuilderFactory factory = ContainerResolver.getInstance().getContainer().getSPI(CallBuilderFactory.class);
+    na.invoke(factory.create().replaceDomainAsync(meta.getName(), meta.getNamespace(), dom, new ResponseStep<Domain>(next) {
       @Override
       public NextAction onFailure(Packet packet, ApiException e, int statusCode,
           Map<String, List<String>> responseHeaders) {

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

@@ -9,6 +9,7 @@
 import oracle.kubernetes.operator.watcher.WatchListener;
 import oracle.kubernetes.weblogic.domain.v1.Domain;
 
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -18,9 +19,9 @@
 public class DomainWatcher extends Watcher<Domain> {
   private final String ns;
 
-  public static DomainWatcher create(String ns, String initialResourceVersion, WatchListener<Domain> listener, AtomicBoolean isStopping) {
+  public static DomainWatcher create(ScheduledExecutorService threadPool, String ns, String initialResourceVersion, WatchListener<Domain> listener, AtomicBoolean isStopping) {
     DomainWatcher watcher = new DomainWatcher(ns, initialResourceVersion, listener, isStopping);
-    watcher.start("Thread-DomainWatcher-" + ns);
+    watcher.start(threadPool);
     return watcher;
   }
 

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

@@ -11,6 +11,7 @@
 import oracle.kubernetes.operator.watcher.WatchListener;
 
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -20,9 +21,9 @@
 public class IngressWatcher extends Watcher<V1beta1Ingress> {
   private final String ns;
 
-  public static IngressWatcher create(String ns, String initialResourceVersion, WatchListener<V1beta1Ingress> listener, AtomicBoolean isStopping) {
+  public static IngressWatcher create(ScheduledExecutorService threadPool, String ns, String initialResourceVersion, WatchListener<V1beta1Ingress> listener, AtomicBoolean isStopping) {
     IngressWatcher watcher = new IngressWatcher(ns, initialResourceVersion, listener, isStopping);
-    watcher.start("Thread-IngressWatcher-" + ns);
+    watcher.start(threadPool);
     return watcher;
   }
 

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

@@ -12,11 +12,13 @@
 import oracle.kubernetes.operator.builders.WatchBuilder;
 import oracle.kubernetes.operator.builders.WatchI;
 import oracle.kubernetes.operator.helpers.CallBuilder;
+import oracle.kubernetes.operator.helpers.CallBuilderFactory;
 import oracle.kubernetes.operator.helpers.ResponseStep;
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
 import oracle.kubernetes.operator.logging.MessageKeys;
 import oracle.kubernetes.operator.watcher.WatchListener;
+import oracle.kubernetes.operator.work.ContainerResolver;
 import oracle.kubernetes.operator.work.NextAction;
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
@@ -25,6 +27,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -42,15 +45,16 @@ public class PodWatcher extends Watcher<V1Pod> implements WatchListener<V1Pod> {
 
   /**
    * Factory for PodWatcher
+   * @param theradPool executor
    * @param ns Namespace
    * @param initialResourceVersion Initial resource version or empty string
    * @param listener Callback for watch events
    * @param isStopping Stop signal
    * @return Pod watcher for the namespace
    */
-  public static PodWatcher create(String ns, String initialResourceVersion, WatchListener<V1Pod> listener, AtomicBoolean isStopping) {
+  public static PodWatcher create(ScheduledExecutorService threadPool, String ns, String initialResourceVersion, WatchListener<V1Pod> listener, AtomicBoolean isStopping) {
     PodWatcher watcher = new PodWatcher(ns, initialResourceVersion, listener, isStopping);
-    watcher.start("Thread-PodWatcher-" + ns);
+    watcher.start(threadPool);
     return watcher;
   }
 
@@ -188,7 +192,8 @@ public NextAction apply(Packet packet) {
         readyCallbackRegistrations.put(metadata.getName(), ready);
 
         // Timing window -- pod may have come ready before registration for callback
-        fiber.createChildFiber().start(CallBuilder.create().readPodAsync(
+        CallBuilderFactory factory = ContainerResolver.getInstance().getContainer().getSPI(CallBuilderFactory.class);
+        fiber.createChildFiber().start(factory.create().readPodAsync(
             metadata.getName(), metadata.getNamespace(), new ResponseStep<V1Pod>(null) {
               @Override
               public NextAction onFailure(Packet packet, ApiException e, int statusCode,

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

@@ -9,6 +9,7 @@
  */
 public interface ProcessingConstants {
 
+  public static final String MAIN_COMPONENT_NAME = "main";
   public static final String DOMAIN_COMPONENT_NAME = "domain";
   public static final String FIBER_COMPONENT_NAME = "fiber";
 

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

@@ -19,11 +19,11 @@
 import com.google.common.base.Charsets;
 import com.google.common.io.CharStreams;
 
+import io.kubernetes.client.ApiClient;
 import io.kubernetes.client.ApiException;
 import io.kubernetes.client.Exec;
 import io.kubernetes.client.models.V1Pod;
-import oracle.kubernetes.operator.helpers.ClientHelper;
-import oracle.kubernetes.operator.helpers.ClientHolder;
+import oracle.kubernetes.operator.helpers.ClientPool;
 import oracle.kubernetes.operator.helpers.DomainPresenceInfo;
 import oracle.kubernetes.operator.helpers.ServerKubernetesObjects;
 import oracle.kubernetes.operator.logging.LoggingFacade;
@@ -131,13 +131,12 @@ public NextAction apply(Packet packet) {
       final boolean tty = true;
 
       return doSuspend(fiber -> {
-        ClientHelper helper = ClientHelper.getInstance();
-        ClientHolder holder = helper.take();
-        Exec exec = new Exec(holder.getApiClient());
         Process proc = null;
         String state = null;
+        ClientPool helper = ClientPool.getInstance();
+        ApiClient client = helper.take();
         try {
-          proc = exec.exec(pod,
+          proc = new Exec(client).exec(pod,
               new String[] { "/weblogic-operator/scripts/readState.sh" },
               KubernetesConstants.CONTAINER_NAME, stdin, tty);
 
@@ -150,7 +149,7 @@ public NextAction apply(Packet packet) {
         } catch (IOException | ApiException | InterruptedException e) {
           LOGGER.warning(MessageKeys.EXCEPTION, e);
         } finally {
-          helper.recycle(holder);
+          helper.recycle(client);
           if (proc != null) {
             proc.destroy();
           }

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

@@ -11,6 +11,7 @@
 import oracle.kubernetes.operator.watcher.WatchListener;
 
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -20,9 +21,9 @@
 public class ServiceWatcher extends Watcher<V1Service> {
   private final String ns;
 
-  public static ServiceWatcher create(String ns, String initialResourceVersion, WatchListener<V1Service> listener, AtomicBoolean isStopping) {
+  public static ServiceWatcher create(ScheduledExecutorService threadPool, String ns, String initialResourceVersion, WatchListener<V1Service> listener, AtomicBoolean isStopping) {
     ServiceWatcher watcher = new ServiceWatcher(ns, initialResourceVersion, listener, isStopping);
-    watcher.start("Thread-ServiceWatcher-" + ns);
+    watcher.start(threadPool);
     return watcher;
   }
 

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

@@ -0,0 +1,171 @@
+// Copyright 2017, 2018, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Licensed under the Universal Permissive License v 1.0 as shown at http://oss.oracle.com/licenses/upl.
+
+package oracle.kubernetes.operator;
+
+import java.io.IOException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import oracle.kubernetes.operator.helpers.ConfigMapConsumer;
+import oracle.kubernetes.operator.logging.LoggingFacade;
+import oracle.kubernetes.operator.logging.LoggingFactory;
+import oracle.kubernetes.operator.logging.MessageKeys;
+
+public class TuningParameters extends ConfigMapConsumer {
+  private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
+  private static TuningParameters INSTANCE = null;
+  
+  private final ReadWriteLock lock = new ReentrantReadWriteLock();
+  private MainTuning main = null;
+  private CallBuilderTuning callBuilder = null;
+  private WatchTuning watch = null;
+  private PodTuning pod = null;
+  
+  public static class MainTuning {
+    public final int statusUpdateTimeoutSeconds;
+    public final int unchangedCountToDelayStatusRecheck; 
+    public final long initialShortDelay; 
+    public final long eventualLongDelay;
+    
+    public MainTuning(int statusUpdateTimeoutSeconds, int unchangedCountToDelayStatusRecheck, 
+        long initialShortDelay, long eventualLongDelay) {
+      this.statusUpdateTimeoutSeconds = statusUpdateTimeoutSeconds;
+      this.unchangedCountToDelayStatusRecheck = unchangedCountToDelayStatusRecheck;
+      this.initialShortDelay = initialShortDelay;
+      this.eventualLongDelay = eventualLongDelay;
+    }
+  }
+  
+  public static class CallBuilderTuning {
+    public final int callRequestLimit;
+    public final int callMaxRetryCount;
+    public final int callTimeoutSeconds;
+    
+    public CallBuilderTuning(int callRequestLimit, int callMaxRetryCount, int callTimeoutSeconds) {
+      this.callRequestLimit = callRequestLimit;
+      this.callMaxRetryCount = callMaxRetryCount;
+      this.callTimeoutSeconds = callTimeoutSeconds;
+    }
+  }
+  
+  public static class WatchTuning {
+    public final int watchLifetime;
+    
+    public WatchTuning(int watchLifetime) {
+      this.watchLifetime = watchLifetime;
+    }
+  }
+  
+  public static class PodTuning {
+    public final int readinessProbeInitialDelaySeconds;
+    public final int readinessProbeTimeoutSeconds;
+    public final int readinessProbePeriodSeconds;
+    public final int livenessProbeInitialDelaySeconds;
+    public final int livenessProbeTimeoutSeconds;
+    public final int livenessProbePeriodSeconds;
+
+    public PodTuning(int readinessProbeInitialDelaySeconds, int readinessProbeTimeoutSeconds, 
+        int readinessProbePeriodSeconds, int livenessProbeInitialDelaySeconds, 
+        int livenessProbeTimeoutSeconds, int livenessProbePeriodSeconds) {
+      this.readinessProbeInitialDelaySeconds = readinessProbeInitialDelaySeconds;
+      this.readinessProbeTimeoutSeconds = readinessProbeTimeoutSeconds;
+      this.readinessProbePeriodSeconds = readinessProbePeriodSeconds;
+      this.livenessProbeInitialDelaySeconds = livenessProbeInitialDelaySeconds;
+      this.livenessProbeTimeoutSeconds = livenessProbeTimeoutSeconds;
+      this.livenessProbePeriodSeconds = livenessProbePeriodSeconds;
+    }
+  }
+
+  public synchronized static TuningParameters initializeInstance(
+      ScheduledExecutorService threadPool, String mountPoint) throws IOException {
+    if (INSTANCE == null) {
+      INSTANCE = new TuningParameters(threadPool, mountPoint);
+      return INSTANCE;
+    }
+    throw new IllegalStateException();
+  }
+  
+  private TuningParameters(ScheduledExecutorService threadPool, String mountPoint) throws IOException {
+    super(threadPool, mountPoint, () -> {
+      updateTuningParameters();
+    });
+  }
+
+  public static void updateTuningParameters() {
+    INSTANCE.update();
+  }
+  
+  private void update() {
+    LOGGER.info(MessageKeys.TUNING_PARAMETERS);
+    
+    MainTuning main = new MainTuning(
+        (int) readTuningParameter("statusUpdateTimeoutSeconds", 10),
+        (int) readTuningParameter("statueUpdateUnchangedCountToDelayStatusRecheck", 10),
+        readTuningParameter("statusUpdateInitialShortDelay", 3),
+        readTuningParameter("statusUpdateEventualLongDelay", 30));
+
+    CallBuilderTuning callBuilder = new CallBuilderTuning(
+        (int) readTuningParameter("callRequestLimit", 500),
+        (int) readTuningParameter("callMaxRetryCount", 5),
+        (int) readTuningParameter("callTimeoutSeconds", 10));
+    
+    WatchTuning watch = new WatchTuning(
+        (int) readTuningParameter("watchLifetime", 45));
+    
+    PodTuning pod = new PodTuning(
+        (int) readTuningParameter("readinessProbeInitialDelaySeconds", 2),
+        (int) readTuningParameter("readinessProbeTimeoutSeconds", 5),
+        (int) readTuningParameter("readinessProbeTimeoutSeconds", 5),
+        (int) readTuningParameter("livenessProbeInitialDelaySeconds", 10),
+        (int) readTuningParameter("livenessProbeTimeoutSeconds", 5),
+        (int) readTuningParameter("livenessProbePeriodSeconds", 10));
+
+    lock.writeLock().lock();
+    try {
+      this.main = main;
+      this.callBuilder = callBuilder;
+      this.watch = watch;
+      this.pod = pod;
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+  
+  public MainTuning getMainTuning() {
+    lock.readLock().lock();
+    try {
+      return main;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+  
+  public CallBuilderTuning getCallBuilderTuning() {
+    lock.readLock().lock();
+    try {
+      return callBuilder;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+  
+  public WatchTuning getWatchTuning() {
+    lock.readLock().lock();
+    try {
+      return watch;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+  
+  public PodTuning getPodTuning() {
+    lock.readLock().lock();
+    try {
+      return pod;
+    } finally {
+      lock.readLock().unlock();
+    }
+  }
+}