Rolling concurrency

Author: rjeberhard

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/helpers/RollingHelper.java

@@ -176,84 +176,87 @@ public String getDetail() {
 
     @Override
     public NextAction apply(Packet packet) {
-      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);
+      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);
+                  }
                 }
               }
             }
           }
-        }
 
-        // 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;
+          // 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;
+            }
           }
-        }
 
-        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()));
-            }
+          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();
+              // Wait for at least one of the not-yet-ready servers to become ready
+              return doForkAtLeastOne(this, packet, waitForUnreadyServers);
+            } else {
+              throw new IllegalStateException();
+            }
           }
-        }
 
-        readyServers.removeAll(servers);
-        LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
+          readyServers.removeAll(servers);
+          LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
 
-        return doNext(new ServersThatCanRestartNowStep(serversThatCanRestartNow, this), packet);
+          return doNext(new ServersThatCanRestartNowStep(serversThatCanRestartNow, this), packet);
+        }
       }
 
       return doNext(packet);