Merge branch 'develop' of https://github.com/oracle/weblogic-kubernetes-operator into develop-acao

pull

Author: alan-cao

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

@@ -147,12 +147,14 @@ public void dispatchPodWatch(Watch.Response<V1Pod> item) {
         if (info != null) {
           switch (item.type) {
             case "ADDED":
+              info.setServerPodBeingDeleted(serverName, Boolean.FALSE);
+              // fall through
             case "MODIFIED":
               info.setServerPodFromEvent(serverName, pod);
               break;
             case "DELETED":
               boolean removed = info.deleteServerPodFromEvent(serverName, pod);
-              if (removed && info.isNotDeleting()) {
+              if (removed && info.isNotDeleting() && !info.isServerPodBeingDeleted(serverName)) {
                 LOGGER.info(
                     MessageKeys.POD_DELETED, domainUID, metadata.getNamespace(), serverName);
                 makeRightDomainPresence(info, true, false, true);

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

@@ -8,10 +8,11 @@
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.models.V1Pod;
 import io.kubernetes.client.util.Watch;
+import java.util.ArrayList;
+import java.util.Collection;
+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 oracle.kubernetes.operator.TuningParameters.WatchTuning;
@@ -39,8 +40,24 @@ public class PodWatcher extends Watcher<V1Pod>
   private final WatchListener<V1Pod> listener;
 
   // Map of Pod name to OnReady
-  private final ConcurrentMap<String, OnReady> readyCallbackRegistrations =
-      new ConcurrentHashMap<>();
+  private final Map<String, Collection<OnReady>> readyCallbackRegistrations = new HashMap<>();
+
+  private void registerOnReady(String podName, OnReady onReady) {
+    synchronized (readyCallbackRegistrations) {
+      Collection<OnReady> col = readyCallbackRegistrations.get(podName);
+      if (col == null) {
+        col = new ArrayList<>();
+        readyCallbackRegistrations.put(podName, col);
+      }
+      col.add(onReady);
+    }
+  }
+
+  private Collection<OnReady> retrieveOnReady(String podName) {
+    synchronized (readyCallbackRegistrations) {
+      return readyCallbackRegistrations.remove(podName);
+    }
+  }
 
   /**
    * Factory for PodWatcher.
@@ -96,9 +113,11 @@ public void receivedResponse(Watch.Response<V1Pod> item) {
         Boolean isReady = !PodHelper.isDeleting(pod) && PodHelper.isReady(pod);
         String podName = pod.getMetadata().getName();
         if (isReady) {
-          OnReady ready = readyCallbackRegistrations.remove(podName);
-          if (ready != null) {
-            ready.onReady();
+          Collection<OnReady> col = retrieveOnReady(podName);
+          if (col != null) {
+            for (OnReady ready : col) {
+              ready.onReady();
+            }
           }
         }
         break;
@@ -148,7 +167,7 @@ public NextAction apply(Packet packet) {
                     fiber.resume(packet);
                   }
                 };
-            readyCallbackRegistrations.put(metadata.getName(), ready);
+            registerOnReady(metadata.getName(), ready);
 
             // Timing window -- pod may have come ready before registration for callback
             CallBuilderFactory factory =

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

@@ -143,7 +143,7 @@ private void watchForEvents() {
             new WatchBuilder()
                 .withResourceVersion(resourceVersion.toString())
                 .withTimeoutSeconds(tuning.watchLifetime))) {
-      while (watch.hasNext()) {
+      while (hasNext(watch)) {
         Watch.Response<T> item = watch.next();
 
         if (isStopping()) {
@@ -164,6 +164,15 @@ private void watchForEvents() {
     }
   }
 
+  private boolean hasNext(WatchI<T> watch) {
+    try {
+      return watch.hasNext();
+    } catch (Throwable ex) {
+      // no-op on exception during hasNext
+    }
+    return false;
+  }
+
   /**
    * Initiates a watch by using the watch builder to request any updates for the specified watcher.
    *

operator/src/main/java/oracle/kubernetes/operator/calls/AsyncRequestStep.java

@@ -190,7 +190,7 @@ public void onSuccess(
                         try {
                           cc.cancel();
                         } finally {
-                          LOGGER.info(
+                          LOGGER.fine(
                               MessageKeys.ASYNC_TIMEOUT,
                               requestParams.call,
                               requestParams.namespace,

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

@@ -122,6 +122,18 @@ private V1Pod getPod(ServerKubernetesObjects sko) {
     return sko.getPod().get();
   }
 
+  private Boolean getPodIsBeingDeleted(ServerKubernetesObjects sko) {
+    return sko.isPodBeingDeleted().get();
+  }
+
+  public Boolean isServerPodBeingDeleted(String serverName) {
+    return getPodIsBeingDeleted(getSko(serverName));
+  }
+
+  public void setServerPodBeingDeleted(String serverName, Boolean isBeingDeleted) {
+    getSko(serverName).isPodBeingDeleted().set(isBeingDeleted);
+  }
+
   /**
    * Returns a collection of all servers defined.
    *

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

@@ -103,7 +103,21 @@ private Step getConflictStep() {
 
   abstract Map<String, String> getPodAnnotations();
 
-  private class ConflictStep extends Step {
+  private abstract class BaseStep extends Step {
+    BaseStep() {
+      this(null);
+    }
+
+    BaseStep(Step next) {
+      super(next);
+    }
+
+    protected String getDetail() {
+      return getServerName();
+    }
+  }
+
+  private class ConflictStep extends BaseStep {
 
     @Override
     public NextAction apply(Packet packet) {
@@ -225,8 +239,12 @@ private List<V1ContainerPort> getContainerPorts() {
   // ----------------------- step methods ------------------------------
 
   // Prevent the watcher from recreating pod with old spec
-  private void clearRecord() {
-    setRecordedPod(null);
+  private void markBeingDeleted() {
+    info.setServerPodBeingDeleted(getServerName(), Boolean.TRUE);
+  }
+
+  private void clearBeingDeleted() {
+    info.setServerPodBeingDeleted(getServerName(), Boolean.FALSE);
   }
 
   private void setRecordedPod(V1Pod pod) {
@@ -269,6 +287,7 @@ private Step deletePod(Step next) {
    * @return a step to be scheduled.
    */
   Step createPod(Step next) {
+    clearBeingDeleted();
     return createPodAsync(createResponse(next));
   }
 
@@ -337,15 +356,15 @@ Step createCyclePodStep(Step next) {
     return new CyclePodStep(next);
   }
 
-  private class CyclePodStep extends Step {
+  private class CyclePodStep extends BaseStep {
 
     CyclePodStep(Step next) {
       super(next);
     }
 
     @Override
     public NextAction apply(Packet packet) {
-      clearRecord();
+      markBeingDeleted();
       return doNext(deletePod(getNext()), packet);
     }
   }
@@ -373,7 +392,7 @@ private String getReasonToRecycle(V1Pod currentPod) {
     return compatibility.getIncompatibility();
   }
 
-  private class VerifyPodStep extends Step {
+  private class VerifyPodStep extends BaseStep {
 
     VerifyPodStep(Step next) {
       super(next);
@@ -399,11 +418,21 @@ public NextAction apply(Packet packet) {
     }
   }
 
+  private abstract class BaseResponseStep extends ResponseStep<V1Pod> {
+    BaseResponseStep(Step next) {
+      super(next);
+    }
+
+    protected String getDetail() {
+      return getServerName();
+    }
+  }
+
   private ResponseStep<V1Pod> createResponse(Step next) {
     return new CreateResponseStep(next);
   }
 
-  private class CreateResponseStep extends ResponseStep<V1Pod> {
+  private class CreateResponseStep extends BaseResponseStep {
     CreateResponseStep(Step next) {
       super(next);
     }
@@ -433,6 +462,10 @@ private class DeleteResponseStep extends ResponseStep<V1Status> {
       super(next);
     }
 
+    protected String getDetail() {
+      return getServerName();
+    }
+
     @Override
     public NextAction onFailure(Packet packet, CallResponse<V1Status> callResponses) {
       if (callResponses.getStatusCode() == CallBuilder.NOT_FOUND) {
@@ -451,12 +484,10 @@ private ResponseStep<V1Pod> replaceResponse(Step next) {
     return new ReplacePodResponseStep(next);
   }
 
-  private class ReplacePodResponseStep extends ResponseStep<V1Pod> {
-    private final Step next;
+  private class ReplacePodResponseStep extends BaseResponseStep {
 
     ReplacePodResponseStep(Step next) {
       super(next);
-      this.next = next;
     }
 
     @Override
@@ -473,15 +504,16 @@ public NextAction onSuccess(Packet packet, CallResponse<V1Pod> callResponse) {
         setRecordedPod(newPod);
       }
 
-      return doNext(next, packet);
+      PodAwaiterStepFactory pw = packet.getSPI(PodAwaiterStepFactory.class);
+      return doNext(pw.waitForReady(newPod, getNext()), packet);
     }
   }
 
   private ResponseStep<V1Pod> patchResponse(Step next) {
     return new PatchPodResponseStep(next);
   }
 
-  private class PatchPodResponseStep extends ResponseStep<V1Pod> {
+  private class PatchPodResponseStep extends BaseResponseStep {
     private final Step next;
 
     PatchPodResponseStep(Step next) {