Move job watcher cache to JobWatcher

Author: russgold

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

@@ -14,10 +14,6 @@
 
 public interface DomainProcessor {
 
-  public static DomainProcessor getInstance() {
-    return DomainProcessorImpl.INSTANCE;
-  }
-
   public void makeRightDomainPresence(
       DomainPresenceInfo info,
       boolean explicitRecheck,

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

@@ -57,7 +57,6 @@
 import oracle.kubernetes.weblogic.domain.model.DomainSpec;
 
 public class DomainProcessorImpl implements DomainProcessor {
-  static final DomainProcessor INSTANCE = new DomainProcessorImpl();
 
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
 
@@ -808,12 +807,7 @@ private static Step[] domainIntrospectionSteps(DomainPresenceInfo info, Step nex
     resources.add(
         JobHelper.deleteDomainIntrospectorJobStep(
             dom.getDomainUID(), dom.getMetadata().getNamespace(), null));
-    resources.add(
-        JobHelper.createDomainIntrospectorJobStep(
-            Main.tuningAndConfig.getWatchTuning(),
-            next,
-            jws,
-            Main.isNamespaceStopping(dom.getMetadata().getNamespace())));
+    resources.add(JobHelper.createDomainIntrospectorJobStep(next));
     return resources.toArray(new Step[0]);
   }
 

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

@@ -10,12 +10,14 @@
 import io.kubernetes.client.models.V1JobStatus;
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.util.Watch;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import oracle.kubernetes.operator.TuningParameters.WatchTuning;
 import oracle.kubernetes.operator.builders.WatchBuilder;
 import oracle.kubernetes.operator.builders.WatchI;
@@ -28,19 +30,36 @@
 import oracle.kubernetes.operator.work.NextAction;
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
+import oracle.kubernetes.weblogic.domain.model.Domain;
 
 /** Watches for Jobs to become Ready or leave Ready state. */
 public class JobWatcher extends Watcher<V1Job> implements WatchListener<V1Job> {
   private static final LoggingFacade LOGGER = LoggingFactory.getLogger("Operator", "Operator");
+  private static final Map<String, JobWatcher> JOB_WATCHERS = new HashMap<>();
+  private static JobWatcherFactory factory;
 
-  private final String ns;
+  private final String namespace;
 
   // Map of Pod name to Complete
   private final ConcurrentMap<String, Complete> completeCallbackRegistrations =
       new ConcurrentHashMap<>();
 
   /**
-   * Factory for JobWatcher.
+   * If a JobWatcher has already been created, returns it.
+   *
+   * @param domain the domain for which the job watcher is to be returned
+   * @return a cached jobwatcher, or null.
+   */
+  public static JobWatcher getOrCreateFor(Domain domain) {
+    return JOB_WATCHERS.computeIfAbsent(getNamespace(domain), n -> factory.createFor(domain));
+  }
+
+  static String getNamespace(Domain domain) {
+    return domain.getMetadata().getNamespace();
+  }
+
+  /**
+   * Creates a new JobWatcher and caches it by namespace.
    *
    * @param factory thread factory
    * @param ns Namespace
@@ -61,17 +80,27 @@ public static JobWatcher create(
   }
 
   private JobWatcher(
-      String ns, String initialResourceVersion, WatchTuning tuning, AtomicBoolean isStopping) {
+      String namespace,
+      String initialResourceVersion,
+      WatchTuning tuning,
+      AtomicBoolean isStopping) {
     super(initialResourceVersion, tuning, isStopping);
     setListener(this);
-    this.ns = ns;
+    this.namespace = namespace;
+  }
+
+  static void defineFactory(
+      ThreadFactory threadFactory,
+      WatchTuning tuning,
+      Function<String, AtomicBoolean> isNamespaceStopping) {
+    factory = new JobWatcherFactory(threadFactory, tuning, isNamespaceStopping);
   }
 
   @Override
   public WatchI<V1Job> initiateWatch(WatchBuilder watchBuilder) throws ApiException {
     return watchBuilder
         .withLabelSelectors(LabelConstants.DOMAINUID_LABEL, LabelConstants.CREATEDBYOPERATOR_LABEL)
-        .createJobWatch(ns);
+        .createJobWatch(namespace);
   }
 
   public void receivedResponse(Watch.Response<V1Job> item) {
@@ -222,6 +251,32 @@ public NextAction onSuccess(
     }
   }
 
+  static class JobWatcherFactory {
+    private ThreadFactory threadFactory;
+    private WatchTuning watchTuning;
+
+    private Function<String, AtomicBoolean> isNamespaceStopping;
+
+    JobWatcherFactory(
+        ThreadFactory threadFactory,
+        WatchTuning watchTuning,
+        Function<String, AtomicBoolean> isNamespaceStopping) {
+      this.threadFactory = threadFactory;
+      this.watchTuning = watchTuning;
+      this.isNamespaceStopping = isNamespaceStopping;
+    }
+
+    JobWatcher createFor(Domain domain) {
+      String namespace = getNamespace(domain);
+      return create(
+          threadFactory,
+          namespace,
+          domain.getMetadata().getResourceVersion(),
+          watchTuning,
+          isNamespaceStopping.apply(namespace));
+    }
+  }
+
   @FunctionalInterface
   private interface Complete {
     void isComplete(V1Job job);

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

@@ -127,7 +127,7 @@ public Thread newThread(Runnable r) {
   }
 
   static final Engine engine = new Engine(wrappedExecutorService);
-  private static final DomainProcessor processor = DomainProcessor.getInstance();
+  private static final DomainProcessor processor = new DomainProcessorImpl();
 
   static final ConcurrentMap<String, AtomicBoolean> isNamespaceStarted = new ConcurrentHashMap<>();
   static final ConcurrentMap<String, AtomicBoolean> isNamespaceStopping = new ConcurrentHashMap<>();
@@ -196,6 +196,8 @@ private static void begin() {
     principal = "system:serviceaccount:" + operatorNamespace + ":" + serviceAccountName;
 
     LOGGER.info(MessageKeys.OP_CONFIG_NAMESPACE, operatorNamespace);
+    JobWatcher.defineFactory(
+        threadFactory, tuningAndConfig.getWatchTuning(), Main::isNamespaceStopping);
 
     Collection<String> targetNamespaces = getTargetNamespaces();
     StringBuilder tns = new StringBuilder();
@@ -530,6 +532,15 @@ private static DomainWatcher createDomainWatcher(String ns, String initialResour
         isNamespaceStopping(ns));
   }
 
+  private static void createJobWatcher(String ns, String initialResourceVersion) {
+    JobWatcher.create(
+        threadFactory,
+        ns,
+        initialResourceVersion,
+        tuningAndConfig.getWatchTuning(),
+        isNamespaceStopping(ns));
+  }
+
   static String getOperatorNamespace() {
     String namespace = System.getenv("OPERATOR_NAMESPACE");
     if (namespace == null) {

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

@@ -12,14 +12,10 @@
 import io.kubernetes.client.models.V1Volume;
 import io.kubernetes.client.models.V1VolumeMount;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import javax.validation.constraints.NotNull;
 import oracle.kubernetes.operator.JobWatcher;
 import oracle.kubernetes.operator.LabelConstants;
 import oracle.kubernetes.operator.ProcessingConstants;
 import oracle.kubernetes.operator.TuningParameters;
-import oracle.kubernetes.operator.TuningParameters.WatchTuning;
 import oracle.kubernetes.operator.calls.CallResponse;
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
@@ -115,32 +111,18 @@ List<V1EnvVar> getConfiguredEnvVars(TuningParameters tuningParameters) {
   /**
    * Factory for {@link Step} that creates WebLogic domain introspector job.
    *
-   * @param tuning Watch tuning parameters
    * @param next Next processing step
-   * @param jws Map of JobWatcher objects, keyed by the string value of the name of a namespace
-   * @param isStopping Stop signal
    * @return Step for creating job
    */
-  public static Step createDomainIntrospectorJobStep(
-      WatchTuning tuning,
-      Step next,
-      @NotNull Map<String, JobWatcher> jws,
-      @NotNull AtomicBoolean isStopping) {
+  public static Step createDomainIntrospectorJobStep(Step next) {
 
-    return new DomainIntrospectorJobStep(tuning, next, jws, isStopping);
+    return new DomainIntrospectorJobStep(next);
   }
 
   static class DomainIntrospectorJobStep extends Step {
-    private final WatchTuning tuning;
-    private final Map<String, JobWatcher> jws;
-    private final AtomicBoolean isStopping;
 
-    DomainIntrospectorJobStep(
-        WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
+    DomainIntrospectorJobStep(Step next) {
       super(next);
-      this.tuning = tuning;
-      this.jws = jws;
-      this.isStopping = isStopping;
     }
 
     @Override
@@ -149,12 +131,12 @@ public NextAction apply(Packet packet) {
       if (runIntrospector(packet, info)) {
         JobStepContext context = new DomainIntrospectorJobStepContext(info, packet);
 
-        packet.putIfAbsent(START_TIME, Long.valueOf(System.currentTimeMillis()));
+        packet.putIfAbsent(START_TIME, System.currentTimeMillis());
 
         return doNext(
             context.createNewJob(
                 readDomainIntrospectorPodLogStep(
-                    tuning, ConfigMapHelper.createSitConfigMapStep(getNext()), jws, isStopping)),
+                    ConfigMapHelper.createSitConfigMapStep(getNext()))),
             packet);
       }
 
@@ -167,10 +149,7 @@ private static boolean runIntrospector(Packet packet, DomainPresenceInfo info) {
     LOGGER.fine("runIntrospector topology: " + config);
     LOGGER.fine("runningServersCount: " + runningServersCount(info));
     LOGGER.fine("creatingServers: " + creatingServers(info));
-    if (config == null || (runningServersCount(info) == 0 && creatingServers(info))) {
-      return true;
-    }
-    return false;
+    return config == null || (runningServersCount(info) == 0 && creatingServers(info));
   }
 
   private static int runningServersCount(DomainPresenceInfo info) {
@@ -180,7 +159,7 @@ private static int runningServersCount(DomainPresenceInfo info) {
   /**
    * TODO: Enhance determination of when we believe we're creating WLS managed server pods.
    *
-   * @param info
+   * @param info the domain presence info
    * @return True, if creating servers
    */
   static boolean creatingServers(DomainPresenceInfo info) {
@@ -262,43 +241,35 @@ String getJobDeletedMessageKey() {
       return MessageKeys.JOB_DELETED;
     }
 
-    protected void logJobDeleted(String domainUID, String namespace, String jobName) {
+    void logJobDeleted(String domainUID, String namespace, String jobName) {
       LOGGER.info(getJobDeletedMessageKey(), domainUID, namespace, jobName);
     }
 
     private Step deleteJob(Step next) {
       String jobName = JobHelper.createJobName(this.domainUID);
       logJobDeleted(this.domainUID, namespace, jobName);
-      Step step =
-          new CallBuilder()
-              .deleteJobAsync(
-                  jobName,
-                  this.namespace,
-                  new V1DeleteOptions().propagationPolicy("Foreground"),
-                  new DefaultResponseStep<>(next));
-      return step;
+      return new CallBuilder()
+          .deleteJobAsync(
+              jobName,
+              this.namespace,
+              new V1DeleteOptions().propagationPolicy("Foreground"),
+              new DefaultResponseStep<>(next));
     }
   }
 
-  private static Step createWatchDomainIntrospectorJobReadyStep(
-      WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
-    return new WatchDomainIntrospectorJobReadyStep(tuning, next, jws, isStopping);
+  private static Step createWatchDomainIntrospectorJobReadyStep(Step next) {
+    return new WatchDomainIntrospectorJobReadyStep(next);
   }
 
   /**
    * Factory for {@link Step} that reads WebLogic domain introspector job results from pod's log.
    *
-   * @param tuning Watch tuning parameters
    * @param next Next processing step
    * @return Step for reading WebLogic domain introspector pod log
    */
-  static Step readDomainIntrospectorPodLogStep(
-      WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
+  private static Step readDomainIntrospectorPodLogStep(Step next) {
     return createWatchDomainIntrospectorJobReadyStep(
-        tuning,
-        readDomainIntrospectorPodStep(new ReadDomainIntrospectorPodLogStep(next)),
-        jws,
-        isStopping);
+        readDomainIntrospectorPodStep(new ReadDomainIntrospectorPodLogStep(next)));
   }
 
   private static class ReadDomainIntrospectorPodLogStep extends Step {
@@ -318,16 +289,14 @@ public NextAction apply(Packet packet) {
     }
 
     private Step readDomainIntrospectorPodLog(String jobPodName, String namespace, Step next) {
-      Step step =
-          new CallBuilder()
-              .readPodLogAsync(
-                  jobPodName, namespace, new ReadDomainIntrospectorPodLogResponseStep(next));
-      return step;
+      return new CallBuilder()
+          .readPodLogAsync(
+              jobPodName, namespace, new ReadDomainIntrospectorPodLogResponseStep(next));
     }
   }
 
   private static class ReadDomainIntrospectorPodLogResponseStep extends ResponseStep<String> {
-    public ReadDomainIntrospectorPodLogResponseStep(Step nextStep) {
+    ReadDomainIntrospectorPodLogResponseStep(Step nextStep) {
       super(nextStep);
     }
 
@@ -376,7 +345,7 @@ private void cleanupJobArtifacts(Packet packet) {
    * @param next Next processing step
    * @return Step for reading WebLogic domain introspector pod
    */
-  public static Step readDomainIntrospectorPodStep(Step next) {
+  private static Step readDomainIntrospectorPodStep(Step next) {
     return new ReadDomainIntrospectorPodStep(next);
   }
 
@@ -396,24 +365,18 @@ public NextAction apply(Packet packet) {
     }
 
     private Step readDomainIntrospectorPod(String domainUID, String namespace, Step next) {
-
-      Step step =
-          new CallBuilder()
-              .withLabelSelectors(LabelConstants.JOBNAME_LABEL)
-              .listPodAsync(namespace, new PodListStep(domainUID, namespace, next));
-
-      return step;
+      return new CallBuilder()
+          .withLabelSelectors(LabelConstants.JOBNAME_LABEL)
+          .listPodAsync(namespace, new PodListStep(domainUID, namespace, next));
     }
   }
 
   private static class PodListStep extends ResponseStep<V1PodList> {
-    private final String ns;
     private final String domainUID;
 
     PodListStep(String domainUID, String ns, Step next) {
       super(next);
       this.domainUID = domainUID;
-      this.ns = ns;
     }
 
     @Override