Update pauser to handle exception and run unpause as long as possible

lib/build.gradle

@@ -8,6 +8,7 @@ dependencies {
     testImplementation(platform("org.junit:junit-bom:${junitVersion}"))
     testImplementation 'org.junit.jupiter:junit-jupiter'
     testImplementation "org.mockito:mockito-core:${mockitoVersion}"
+    testImplementation "org.mockito:mockito-inline:${mockitoVersion}"
 }
 
 test {

lib/src/main/java/com/scalar/admin/kubernetes/PauseFailedException.java

@@ -0,0 +1,11 @@
+package com.scalar.admin.kubernetes;
+
+public class PauseFailedException extends PauserException {
+  public PauseFailedException(String message) {
+    super(message);
+  }
+
+  public PauseFailedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

lib/src/main/java/com/scalar/admin/kubernetes/Pauser.java

@@ -1,5 +1,6 @@
 package com.scalar.admin.kubernetes;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.scalar.admin.RequestCoordinator;
 import io.kubernetes.client.openapi.Configuration;
@@ -13,8 +14,6 @@
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * This class implements a pause operation for Scalar product pods in a Kubernetes cluster. The
@@ -34,10 +33,11 @@
 @NotThreadSafe
 public class Pauser {
 
-  private static final int MAX_UNPAUSE_RETRY_COUNT = 3;
+  @VisibleForTesting static final int MAX_UNPAUSE_RETRY_COUNT = 3;
 
-  private final Logger logger = LoggerFactory.getLogger(Pauser.class);
   private final TargetSelector targetSelector;
+  private Instant startTime;
+  private Instant endTime;
 
   /**
    * @param namespace The namespace where the pods are deployed.
@@ -77,67 +77,169 @@ public PausedDuration pause(int pauseDuration, @Nullable Long maxPauseWaitTime)
           "pauseDuration is required to be greater than 0 millisecond.");
     }
 
-    TargetSnapshot target;
+    // Get pods and deployment information before pause.
+    TargetSnapshot targetBeforePause;
     try {
-      target = targetSelector.select();
+      targetBeforePause = getTarget();
     } catch (Exception e) {
       throw new PauserException("Failed to find the target pods to pause.", e);
     }
 
-    RequestCoordinator coordinator = getRequestCoordinator(target);
-
-    coordinator.pause(true, maxPauseWaitTime);
-
-    Instant startTime = Instant.now();
-
-    Uninterruptibles.sleepUninterruptibly(pauseDuration, TimeUnit.MILLISECONDS);
+    // Get RequestCoordinator of Scalar Admin to pause.
+    RequestCoordinator requestCoordinator;
+    try {
+      requestCoordinator = getRequestCoordinator(targetBeforePause);
+    } catch (Exception e) {
+      throw new PauserException("Failed to initialize the request coordinator.", e);
+    }
 
-    Instant endTime = Instant.now();
+    // Run pause operation.
+    PauseFailedException pauseFailedException = null;
+    try {
+      pauseInternal(requestCoordinator, pauseDuration, maxPauseWaitTime);
+    } catch (Exception e) {
+      pauseFailedException = new PauseFailedException("Pause operation failed.", e);
+    }
 
-    unpauseWithRetry(coordinator, MAX_UNPAUSE_RETRY_COUNT, target);
+    // Run unpause operation.
+    UnpauseFailedException unpauseFailedException = null;
+    try {
+      unpauseWithRetry(requestCoordinator, MAX_UNPAUSE_RETRY_COUNT);
+    } catch (Exception e) {
+      unpauseFailedException = new UnpauseFailedException("Unpause operation failed.", e);
+    }
 
+    // Prepare error messages for each process.
+    String unpauseErrorMessage =
+        String.format(
+            "Unpause operation failed. Scalar products might still be in a paused state. You"
+                + " must restart related pods by using the `kubectl rollout restart deployment"
+                + " %s` command to unpause all pods. ",
+            targetBeforePause.getDeployment().getMetadata().getName());
+    String pauseErrorMessage =
+        "Pause operation failed. You cannot use the backup that was taken during this pause"
+            + " duration. You need to retry the pause operation from the beginning to"
+            + " take a backup. ";
+    String getTargetAfterPauseErrorMessage =
+        "Failed to find the target pods to examine if the targets pods were updated during"
+            + " paused. ";
+    String statusCheckErrorMessage =
+        "Status check failed. You cannot use the backup that was taken during this pause"
+            + " duration. You need to retry the pause operation from the beginning to"
+            + " take a backup. ";
+    String statusDifferentErrorMessage =
+        "The target pods were updated during the pause duration. You cannot use the backup that"
+            + " was taken during this pause duration. ";
+    String unpauseFailedButBackupOkErrorMessage =
+        String.format(
+            "Note that the pause operations for taking backup succeeded. You can use a backup that"
+                + " was taken during this pause duration: Start Time = %s, End Time = %s. ",
+            startTime, endTime);
+
+    // Get pods and deployment information after pause.
     TargetSnapshot targetAfterPause;
+    PauserException pauserException;
     try {
-      targetAfterPause = targetSelector.select();
+      targetAfterPause = getTarget();
     } catch (Exception e) {
-      throw new PauserException(
-          "Failed to find the target pods to examine if the targets pods were updated during"
-              + " paused.",
-          e);
+      pauserException = new PauserException(getTargetAfterPauseErrorMessage, e);
+      if (unpauseFailedException == null) {
+        throw pauserException;
+      } else {
+        throw new UnpauseFailedException(unpauseErrorMessage, pauserException);
+      }
     }
 
-    if (!target.getStatus().equals(targetAfterPause.getStatus())) {
-      throw new PauserException("The target pods were updated during paused. Please retry.");
+    // Check if pods and deployment information are the same between before pause and after pause.
+    boolean compareTargetSuccessful;
+    StatusCheckFailedException statusCheckFailedException;
+    try {
+      compareTargetSuccessful = compareTargetStatus(targetBeforePause, targetAfterPause);
+    } catch (Exception e) {
+      statusCheckFailedException = new StatusCheckFailedException(statusCheckErrorMessage, e);
+      if (unpauseFailedException == null) {
+        throw statusCheckFailedException;
+      } else {
+        throw new UnpauseFailedException(unpauseErrorMessage, statusCheckFailedException);
+      }
     }
 
-    return new PausedDuration(startTime, endTime);
+    // If both the pause operation and status check succeeded, you can use the backup that was taken
+    // during the pause duration.
+    boolean backupOk = (pauseFailedException == null) && compareTargetSuccessful;
+
+    // Create error message if any of the operations failed.
+    StringBuilder errorMessageBuilder = new StringBuilder();
+    if (unpauseFailedException != null) {
+      errorMessageBuilder.append(unpauseErrorMessage);
+      if (backupOk) {
+        errorMessageBuilder.append(unpauseFailedButBackupOkErrorMessage);
+      }
+    }
+    if (pauseFailedException != null) {
+      errorMessageBuilder.append(pauseErrorMessage);
+    }
+    if (!compareTargetSuccessful) {
+      errorMessageBuilder.append(statusDifferentErrorMessage);
+    }
+    String errorMessage = errorMessageBuilder.toString();
+
+    // Return the final result based on each process.
+    if (unpauseFailedException
+        != null) { // Unpause issue is the most critical because it might cause system down.
+      throw new UnpauseFailedException(errorMessage, unpauseFailedException);
+    } else if (pauseFailedException
+        != null) { // Pause Failed is second priority because pause issue might be caused by
+      // configuration error.
+      throw new PauseFailedException(errorMessage, pauseFailedException);
+    } else if (!compareTargetSuccessful) { // Status check failed is third priority because this
+      // issue might be caused by temporary issue, for example, pod restarts.
+      throw new PauseFailedException(errorMessage);
+    } else { // All operations succeeded.
+      return new PausedDuration(startTime, endTime);
+    }
   }
 
-  private void unpauseWithRetry(
-      RequestCoordinator coordinator, int maxRetryCount, TargetSnapshot target) {
+  @VisibleForTesting
+  void unpauseWithRetry(RequestCoordinator coordinator, int maxRetryCount) {
     int retryCounter = 0;
-
     while (true) {
       try {
         coordinator.unpause();
         return;
       } catch (Exception e) {
         if (++retryCounter >= maxRetryCount) {
-          logger.warn(
-              "Failed to unpause Scalar product. They are still in paused. You must restart related"
-                  + " pods by using the `kubectl rollout restart deployment {}`"
-                  + " command to unpase all pods.",
-              target.getDeployment().getMetadata().getName());
-          return;
+          throw e;
         }
       }
     }
   }
 
+  @VisibleForTesting
+  TargetSnapshot getTarget() throws PauserException {
+    return targetSelector.select();
+  }
+
+  @VisibleForTesting
   RequestCoordinator getRequestCoordinator(TargetSnapshot target) {
     return new RequestCoordinator(
         target.getPods().stream()
             .map(p -> new InetSocketAddress(p.getStatus().getPodIP(), target.getAdminPort()))
             .collect(Collectors.toList()));
   }
+
+  @VisibleForTesting
+  void pauseInternal(
+      RequestCoordinator requestCoordinator, int pauseDuration, @Nullable Long maxPauseWaitTime) {
+
+    requestCoordinator.pause(true, maxPauseWaitTime);
+    startTime = Instant.now();
+    Uninterruptibles.sleepUninterruptibly(pauseDuration, TimeUnit.MILLISECONDS);
+    endTime = Instant.now();
+  }
+
+  @VisibleForTesting
+  boolean compareTargetStatus(TargetSnapshot before, TargetSnapshot after) {
+    return before.getStatus().equals(after.getStatus());
+  }
 }

lib/src/main/java/com/scalar/admin/kubernetes/StatusCheckFailedException.java

@@ -0,0 +1,11 @@
+package com.scalar.admin.kubernetes;
+
+public class StatusCheckFailedException extends PauserException {
+  public StatusCheckFailedException(String message) {
+    super(message);
+  }
+
+  public StatusCheckFailedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

lib/src/main/java/com/scalar/admin/kubernetes/UnpauseFailedException.java

@@ -0,0 +1,11 @@
+package com.scalar.admin.kubernetes;
+
+public class UnpauseFailedException extends PauserException {
+  public UnpauseFailedException(String message) {
+    super(message);
+  }
+
+  public UnpauseFailedException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}