Merge pull request #1067 from oracle/cluster-rolling

Wait for other servers to become ready during a cluster rolling

Author: rjeberhard

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

@@ -36,8 +36,6 @@
 import oracle.kubernetes.operator.logging.MessageKeys;
 import oracle.kubernetes.operator.steps.DefaultResponseStep;
 import oracle.kubernetes.operator.wlsconfig.WlsDomainConfig;
-import oracle.kubernetes.operator.work.Fiber;
-import oracle.kubernetes.operator.work.Fiber.CompletionCallback;
 import oracle.kubernetes.operator.work.NextAction;
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
@@ -265,47 +263,15 @@ private Collection<String> getServerNames() {
    * @return Step
    */
   public static Step createProgressingStep(String reason, boolean isPreserveAvailable, Step next) {
-    return new ProgressingHookStep(reason, isPreserveAvailable, next);
-  }
-
-  private static class ProgressingHookStep extends Step {
-    private final String reason;
-    private final boolean isPreserveAvailable;
-
-    private ProgressingHookStep(String reason, boolean isPreserveAvailable, Step next) {
-      super(next);
-      this.reason = reason;
-      this.isPreserveAvailable = isPreserveAvailable;
-    }
-
-    @Override
-    public NextAction apply(Packet packet) {
-      Fiber f = Fiber.current().createChildFiber();
-      Packet p = new Packet();
-      p.getComponents().putAll(packet.getComponents());
-      f.start(
-          new ProgressingStep(reason, isPreserveAvailable),
-          p,
-          new CompletionCallback() {
-            @Override
-            public void onCompletion(Packet packet) {}
-
-            @Override
-            public void onThrowable(Packet packet, Throwable throwable) {
-              LOGGER.severe(MessageKeys.EXCEPTION, throwable);
-            }
-          });
-
-      return doNext(packet);
-    }
+    return new ProgressingStep(reason, isPreserveAvailable, next);
   }
 
   private static class ProgressingStep extends Step {
     private final String reason;
     private final boolean isPreserveAvailable;
 
-    private ProgressingStep(String reason, boolean isPreserveAvailable) {
-      super(null);
+    private ProgressingStep(String reason, boolean isPreserveAvailable, Step next) {
+      super(next);
       this.reason = reason;
       this.isPreserveAvailable = isPreserveAvailable;
     }
@@ -376,44 +342,14 @@ public NextAction apply(Packet packet) {
    * @return Step
    */
   public static Step createAvailableStep(String reason, Step next) {
-    return new AvailableHookStep(reason, next);
-  }
-
-  private static class AvailableHookStep extends Step {
-    private final String reason;
-
-    private AvailableHookStep(String reason, Step next) {
-      super(next);
-      this.reason = reason;
-    }
-
-    @Override
-    public NextAction apply(Packet packet) {
-      Fiber f = Fiber.current().createChildFiber();
-      Packet p = new Packet();
-      p.getComponents().putAll(packet.getComponents());
-      f.start(
-          new AvailableStep(reason),
-          p,
-          new CompletionCallback() {
-            @Override
-            public void onCompletion(Packet packet) {}
-
-            @Override
-            public void onThrowable(Packet packet, Throwable throwable) {
-              LOGGER.severe(MessageKeys.EXCEPTION, throwable);
-            }
-          });
-
-      return doNext(packet);
-    }
+    return new AvailableStep(reason, next);
   }
 
   private static class AvailableStep extends Step {
     private final String reason;
 
-    private AvailableStep(String reason) {
-      super(null);
+    private AvailableStep(String reason, Step next) {
+      super(next);
       this.reason = reason;
     }
 
@@ -499,44 +435,14 @@ public NextAction onSuccess(Packet packet, CallResponse<Domain> callResponse) {
    * @return Step
    */
   static Step createFailedStep(Throwable throwable, Step next) {
-    return new FailedHookStep(throwable, next);
-  }
-
-  private static class FailedHookStep extends Step {
-    private final Throwable throwable;
-
-    private FailedHookStep(Throwable throwable, Step next) {
-      super(next);
-      this.throwable = throwable;
-    }
-
-    @Override
-    public NextAction apply(Packet packet) {
-      Fiber f = Fiber.current().createChildFiber();
-      Packet p = new Packet();
-      p.getComponents().putAll(packet.getComponents());
-      f.start(
-          new FailedStep(throwable),
-          p,
-          new CompletionCallback() {
-            @Override
-            public void onCompletion(Packet packet) {}
-
-            @Override
-            public void onThrowable(Packet packet, Throwable throwable) {
-              LOGGER.severe(MessageKeys.EXCEPTION, throwable);
-            }
-          });
-
-      return doNext(packet);
-    }
+    return new FailedStep(throwable, next);
   }
 
   private static class FailedStep extends Step {
     private final Throwable throwable;
 
-    private FailedStep(Throwable throwable) {
-      super(null);
+    private FailedStep(Throwable throwable, Step next) {
+      super(next);
       this.throwable = throwable;
     }
 

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

@@ -87,11 +87,13 @@ public WatchI<V1Pod> initiateWatch(WatchBuilder watchBuilder) throws ApiExceptio
   public void receivedResponse(Watch.Response<V1Pod> item) {
     LOGGER.entering();
 
+    listener.receivedResponse(item);
+
     switch (item.type) {
       case "ADDED":
       case "MODIFIED":
         V1Pod pod = item.object;
-        Boolean isReady = PodHelper.isReady(pod);
+        Boolean isReady = !PodHelper.isDeleting(pod) && PodHelper.isReady(pod);
         String podName = pod.getMetadata().getName();
         if (isReady) {
           OnReady ready = readyCallbackRegistrations.remove(podName);
@@ -105,8 +107,6 @@ public void receivedResponse(Watch.Response<V1Pod> item) {
       default:
     }
 
-    listener.receivedResponse(item);
-
     LOGGER.exiting();
   }
 
@@ -131,7 +131,7 @@ private WaitForPodReadyStep(V1Pod pod, Step next) {
 
     @Override
     public NextAction apply(Packet packet) {
-      if (PodHelper.getReadyStatus(pod)) {
+      if (!PodHelper.isDeleting(pod) && PodHelper.getReadyStatus(pod)) {
         return doNext(packet);
       }
 
@@ -180,7 +180,9 @@ public NextAction onSuccess(
                                   V1Pod result,
                                   int statusCode,
                                   Map<String, List<String>> responseHeaders) {
-                                if (result != null && PodHelper.getReadyStatus(result)) {
+                                if (result != null
+                                    && !PodHelper.isDeleting(result)
+                                    && PodHelper.getReadyStatus(result)) {
                                   if (didResume.compareAndSet(false, true)) {
                                     readyCallbackRegistrations.remove(metadata.getName(), ready);
                                     fiber.resume(packet);

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

@@ -473,8 +473,7 @@ public NextAction onSuccess(Packet packet, CallResponse<V1Pod> callResponse) {
         setRecordedPod(newPod);
       }
 
-      PodAwaiterStepFactory pw = PodHelper.getPodAwaiterStepFactory(packet);
-      return doNext(pw.waitForReady(newPod, next), packet);
+      return doNext(next, packet);
     }
   }
 
@@ -504,8 +503,7 @@ public NextAction onSuccess(Packet packet, CallResponse<V1Pod> callResponse) {
         setRecordedPod(newPod);
       }
 
-      PodAwaiterStepFactory pw = PodHelper.getPodAwaiterStepFactory(packet);
-      return doNext(pw.waitForReady(newPod, next), packet);
+      return doNext(next, packet);
     }
   }