Terminate is not delete

Author: rjeberhard

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/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) {

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

@@ -5,13 +5,8 @@
 package oracle.kubernetes.operator.helpers;
 
 import io.kubernetes.client.models.V1Pod;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import oracle.kubernetes.operator.PodAwaiterStepFactory;
+import java.util.*;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import oracle.kubernetes.operator.ProcessingConstants;
 import oracle.kubernetes.operator.logging.LoggingFacade;
 import oracle.kubernetes.operator.logging.LoggingFactory;
@@ -74,7 +69,7 @@ public NextAction apply(Packet packet) {
       List<String> availableServers = getReadyServers(info);
 
       Collection<StepAndPacket> serversThatCanRestartNow = new ArrayList<>();
-      Map<String, Collection<StepAndPacket>> clusteredRestarts = new HashMap<>();
+      Map<String, Queue<StepAndPacket>> clusteredRestarts = new HashMap<>();
 
       List<String> servers = new ArrayList<>();
       for (Map.Entry<String, StepAndPacket> entry : rolling.entrySet()) {
@@ -96,9 +91,9 @@ public NextAction apply(Packet packet) {
         }
 
         // clustered server
-        Collection<StepAndPacket> cr = clusteredRestarts.get(clusterName);
+        Queue<StepAndPacket> cr = clusteredRestarts.get(clusterName);
         if (cr == null) {
-          cr = new ArrayList<>();
+          cr = new ConcurrentLinkedQueue<>();
           clusteredRestarts.put(clusterName, cr);
         }
         cr.add(entry.getValue());
@@ -116,7 +111,7 @@ public NextAction apply(Packet packet) {
       }
 
       if (!clusteredRestarts.isEmpty()) {
-        for (Map.Entry<String, Collection<StepAndPacket>> entry : clusteredRestarts.entrySet()) {
+        for (Map.Entry<String, Queue<StepAndPacket>> entry : clusteredRestarts.entrySet()) {
           work.add(
               new StepAndPacket(
                   new RollSpecificClusterStep(entry.getKey(), entry.getValue(), null), packet));
@@ -160,13 +155,13 @@ private static List<String> getReadyServers(DomainPresenceInfo info) {
 
   private static class RollSpecificClusterStep extends Step {
     private final String clusterName;
-    private final Iterator<StepAndPacket> it;
+    private final Queue<StepAndPacket> servers;
 
     public RollSpecificClusterStep(
-        String clusterName, Collection<StepAndPacket> clusteredServerRestarts, Step next) {
+        String clusterName, Queue<StepAndPacket> clusteredServerRestarts, Step next) {
       super(next);
       this.clusterName = clusterName;
-      it = clusteredServerRestarts.iterator();
+      servers = clusteredServerRestarts;
     }
 
     @Override
@@ -176,89 +171,58 @@ public String getDetail() {
 
     @Override
     public NextAction apply(Packet packet) {
-      synchronized (it) {
-        if (it.hasNext()) {
-          DomainPresenceInfo info = packet.getSPI(DomainPresenceInfo.class);
-          WlsDomainConfig config =
-              (WlsDomainConfig) packet.get(ProcessingConstants.DOMAIN_TOPOLOGY);
-
-          // Refresh as this is constantly changing
-          Domain dom = info.getDomain();
-          // These are presently Ready servers
-          List<String> availableServers = getReadyServers(info);
-
-          List<String> servers = new ArrayList<>();
-          List<String> readyServers = new ArrayList<>();
-          List<V1Pod> notReadyServers = new ArrayList<>();
-
-          Collection<StepAndPacket> serversThatCanRestartNow = new ArrayList<>();
-
-          int countReady = 0;
-          WlsClusterConfig cluster = config != null ? config.getClusterConfig(clusterName) : null;
-          if (cluster != null) {
-            List<WlsServerConfig> serversConfigs = cluster.getServerConfigs();
-            if (serversConfigs != null) {
-              for (WlsServerConfig s : serversConfigs) {
-                // figure out how many servers are currently ready
-                String name = s.getName();
-                if (availableServers.contains(name)) {
-                  readyServers.add(s.getName());
-                  countReady++;
-                } else {
-                  V1Pod pod = info.getServerPod(name);
-                  if (pod != null) {
-                    notReadyServers.add(pod);
-                  }
-                }
-              }
-            }
-          }
+      DomainPresenceInfo info = packet.getSPI(DomainPresenceInfo.class);
+      WlsDomainConfig config = (WlsDomainConfig) packet.get(ProcessingConstants.DOMAIN_TOPOLOGY);
 
-          // then add as many as possible next() entries leaving at least minimum cluster
-          // availability
-          while (countReady-- > dom.getMinAvailable(clusterName)) {
-            StepAndPacket current = it.next();
-            WlsServerConfig serverConfig =
-                (WlsServerConfig) current.packet.get(ProcessingConstants.SERVER_SCAN);
-            String serverName = null;
-            if (serverConfig != null) {
-              serverName = serverConfig.getName();
-            } else if (config != null) {
-              serverName = config.getAdminServerName();
-            }
-            if (serverName != null) {
-              servers.add(serverName);
-            }
-            serversThatCanRestartNow.add(current);
-            if (!it.hasNext()) {
-              break;
-            }
-          }
+      // Refresh as this is constantly changing
+      Domain dom = info.getDomain();
+      // These are presently Ready servers
+      List<String> availableServers = getReadyServers(info);
 
-          if (serversThatCanRestartNow.isEmpty()) {
-            // Not enough servers are ready to let us restart a server now
-            if (!notReadyServers.isEmpty()) {
-              PodAwaiterStepFactory pw = PodHelper.getPodAwaiterStepFactory(packet);
-              Collection<StepAndPacket> waitForUnreadyServers = new ArrayList<>();
-              for (V1Pod pod : notReadyServers) {
-                waitForUnreadyServers.add(
-                    new StepAndPacket(pw.waitForReady(pod, null), packet.clone()));
-              }
-
-              // Wait for at least one of the not-yet-ready servers to become ready
-              return doForkAtLeastOne(this, packet, waitForUnreadyServers);
-            } else {
-              throw new IllegalStateException();
+      List<String> readyServers = new ArrayList<>();
+
+      int countReady = 0;
+      WlsClusterConfig cluster = config != null ? config.getClusterConfig(clusterName) : null;
+      if (cluster != null) {
+        List<WlsServerConfig> serversConfigs = cluster.getServerConfigs();
+        if (serversConfigs != null) {
+          for (WlsServerConfig s : serversConfigs) {
+            // figure out how many servers are currently ready
+            String name = s.getName();
+            if (availableServers.contains(name)) {
+              readyServers.add(s.getName());
+              countReady++;
             }
           }
+        }
+      }
 
-          readyServers.removeAll(servers);
-          LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
+      LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
 
-          return doNext(new ServersThatCanRestartNowStep(serversThatCanRestartNow, this), packet);
-        }
+      int countToRestartNow = Math.max(1, countReady - dom.getMinAvailable(clusterName));
+      Collection<StepAndPacket> restarts = new ArrayList<>();
+      for (int i = 0; i < countToRestartNow; i++) {
+        restarts.add(new StepAndPacket(new RestartOneClusteredServerStep(servers, null), packet));
       }
+      return doForkJoin(getNext(), packet, restarts);
+    }
+  }
+
+  private static class RestartOneClusteredServerStep extends Step {
+    private final Queue<StepAndPacket> servers;
 
+    public RestartOneClusteredServerStep(Queue<StepAndPacket> servers, Step next) {
+      super(next);
+      this.servers = servers;
+    }
+
+    @Override
+    public NextAction apply(Packet packet) {
+      StepAndPacket serverToRestart = servers.poll();
+      if (serverToRestart != null) {
+        Collection<StepAndPacket> col = Collections.singleton(serverToRestart);
+        return doForkJoin(this, packet, col);
+      }
       return doNext(packet);
     }
   }