Merge pull request #860 from oracle/owls-71734

WIP: cache the JobWatcher by namespace

Author: rjeberhard

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

@@ -15,6 +15,7 @@
 import io.kubernetes.client.util.Watch;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -656,6 +657,8 @@ public void makeRightDomainPresence(
     }
   }
 
+  private static final Map<String, JobWatcher> jws = new ConcurrentHashMap<>();
+
   private void internalMakeRightDomainPresence(
       @Nullable DomainPresenceInfo info, boolean isDeleting, boolean isWillInterrupt) {
     String ns = info.getNamespace();
@@ -999,7 +1002,11 @@ private static Step[] domainIntrospectionSteps(DomainPresenceInfo info, Step nex
         JobHelper.deleteDomainIntrospectorJobStep(
             dom.getDomainUID(), dom.getMetadata().getNamespace(), null));
     resources.add(
-        JobHelper.createDomainIntrospectorJobStep(Main.tuningAndConfig.getWatchTuning(), next));
+        JobHelper.createDomainIntrospectorJobStep(
+            Main.tuningAndConfig.getWatchTuning(),
+            next,
+            jws,
+            Main.isNamespaceStopping(dom.getMetadata().getNamespace())));
     return resources.toArray(new Step[resources.size()]);
   }
 

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

@@ -13,6 +13,9 @@
 import io.kubernetes.client.models.V1VolumeMount;
 import java.util.ArrayList;
 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;
@@ -115,21 +118,30 @@ List<V1EnvVar> getConfiguredEnvVars(TuningParameters tuningParameters) {
    *
    * @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
    * @return Step for creating job
    */
-  public static Step createDomainIntrospectorJobStep(WatchTuning tuning, Step next) {
+  public static Step createDomainIntrospectorJobStep(
+      WatchTuning tuning,
+      Step next,
+      @NotNull Map<String, JobWatcher> jws,
+      @NotNull AtomicBoolean isStopping) {
 
-    // return new DomainIntrospectorJobStep(
-    //    readDomainIntrospectorPodLogStep(ConfigMapHelper.createSitConfigMapStep(next)));
-    return new DomainIntrospectorJobStep(tuning, next);
+    return new DomainIntrospectorJobStep(tuning, next, jws, isStopping);
   }
 
   static class DomainIntrospectorJobStep extends Step {
     private final WatchTuning tuning;
+    private final Map<String, JobWatcher> jws;
+    private final AtomicBoolean isStopping;
 
-    public DomainIntrospectorJobStep(WatchTuning tuning, Step next) {
+    DomainIntrospectorJobStep(
+        WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
       super(next);
       this.tuning = tuning;
+      this.jws = jws;
+      this.isStopping = isStopping;
     }
 
     @Override
@@ -143,7 +155,7 @@ public NextAction apply(Packet packet) {
         return doNext(
             context.createNewJob(
                 readDomainIntrospectorPodLogStep(
-                    tuning, ConfigMapHelper.createSitConfigMapStep(getNext()))),
+                    tuning, ConfigMapHelper.createSitConfigMapStep(getNext()), jws, isStopping)),
             packet);
       }
 
@@ -269,8 +281,9 @@ private Step deleteJob(Step next) {
     }
   }
 
-  private static Step createWatchDomainIntrospectorJobReadyStep(WatchTuning tuning, Step next) {
-    return new WatchDomainIntrospectorJobReadyStep(tuning, next);
+  private static Step createWatchDomainIntrospectorJobReadyStep(
+      WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
+    return new WatchDomainIntrospectorJobReadyStep(tuning, next, jws, isStopping);
   }
 
   /**
@@ -280,9 +293,13 @@ private static Step createWatchDomainIntrospectorJobReadyStep(WatchTuning tuning
    * @param next Next processing step
    * @return Step for reading WebLogic domain introspector pod log
    */
-  public static Step readDomainIntrospectorPodLogStep(WatchTuning tuning, Step next) {
+  static Step readDomainIntrospectorPodLogStep(
+      WatchTuning tuning, Step next, Map<String, JobWatcher> jws, AtomicBoolean isStopping) {
     return createWatchDomainIntrospectorJobReadyStep(
-        tuning, readDomainIntrospectorPodStep(new ReadDomainIntrospectorPodLogStep(next)));
+        tuning,
+        readDomainIntrospectorPodStep(new ReadDomainIntrospectorPodLogStep(next)),
+        jws,
+        isStopping);
   }
 
   private static class ReadDomainIntrospectorPodLogStep extends Step {

operator/src/main/java/oracle/kubernetes/operator/steps/WatchDomainIntrospectorJobReadyStep.java

@@ -1,11 +1,13 @@
-// Copyright 2017, 2018, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2017, 2019, 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.steps;
 
 import io.kubernetes.client.models.V1Job;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import javax.validation.constraints.NotNull;
 import oracle.kubernetes.operator.JobWatcher;
 import oracle.kubernetes.operator.ProcessingConstants;
 import oracle.kubernetes.operator.TuningParameters.WatchTuning;
@@ -17,10 +19,18 @@
 
 public class WatchDomainIntrospectorJobReadyStep extends Step {
   private final WatchTuning tuning;
+  private final Map<String, JobWatcher> jws;
+  private final AtomicBoolean isStopping;
 
-  public WatchDomainIntrospectorJobReadyStep(WatchTuning tuning, Step next) {
+  public WatchDomainIntrospectorJobReadyStep(
+      WatchTuning tuning,
+      Step next,
+      @NotNull Map<String, JobWatcher> jws,
+      @NotNull AtomicBoolean isStopping) {
     super(next);
     this.tuning = tuning;
+    this.jws = jws;
+    this.isStopping = isStopping;
   }
 
   @Override
@@ -33,15 +43,20 @@ public NextAction apply(Packet packet) {
 
     // No need to spawn a watcher if the job is already complete
     if (domainIntrospectorJob != null && !JobWatcher.isComplete(domainIntrospectorJob)) {
-      JobWatcher jw =
-          JobWatcher.create(
-              ThreadFactorySingleton.getInstance(),
-              namespace,
-              initialResourceVersion,
-              tuning,
-              new AtomicBoolean(false));
-
-      return doNext(jw.waitForReady(domainIntrospectorJob, getNext()), packet);
+      JobWatcher jw = jws.get(namespace);
+      if (jw == null) {
+        jw =
+            JobWatcher.create(
+                ThreadFactorySingleton.getInstance(),
+                namespace,
+                initialResourceVersion,
+                tuning,
+                isStopping);
+        jws.put(namespace, jw);
+      }
+      NextAction retVal = doNext(jw.waitForReady(domainIntrospectorJob, getNext()), packet);
+
+      return retVal;
     }
 
     return doNext(packet);

operator/src/test/java/oracle/kubernetes/operator/helpers/DomainIntrospectorJobTest.java

@@ -1,4 +1,4 @@
-// Copyright 2018, Oracle Corporation and/or its affiliates.  All rights reserved.
+// Copyright 2018, 2019, 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.
 
@@ -19,9 +19,12 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.LogRecord;
 import oracle.kubernetes.TestUtils;
+import oracle.kubernetes.operator.JobWatcher;
 import oracle.kubernetes.operator.KubernetesConstants;
 import oracle.kubernetes.operator.LabelConstants;
 import oracle.kubernetes.operator.TuningParameters;
@@ -290,7 +293,12 @@ static String getJobName() {
   }
 
   FiberTestSupport.StepFactory getStepFactory() {
-    return next -> JobHelper.createDomainIntrospectorJobStep(new WatchTuning(30, 0), next);
+    return next ->
+        JobHelper.createDomainIntrospectorJobStep(
+            new WatchTuning(30, 0),
+            next,
+            new ConcurrentHashMap<String, JobWatcher>(),
+            new AtomicBoolean(false));
   }
 
   V1PodList createListPods() {