HDDS-15715. Add helpful CLI flags to finalization CLIs

hadoop-hdds/interface-client/src/main/proto/hdds.proto

@@ -434,6 +434,9 @@ message UpgradeStatus {
   optional int32 numDatanodesFinalized = 2;
   optional int32 numDatanodesTotal = 3;
   optional bool shouldFinalize = 4;
+  optional uint32 scmApparentVersion = 5;
+  optional uint32 minDatanodeApparentVersion = 6;
+  optional uint32 maxDatanodeApparentVersion = 7;
 }
 
 /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java

@@ -152,6 +152,8 @@ default int getAllNodeCount() {
   default DatanodeFinalizationCounts getDatanodeFinalizationCounts() {
     int finalizedNodes = 0;
     int totalHealthyNodes = 0;
+    int minApparentVersion = Integer.MAX_VALUE;
+    int maxApparentVersion = 0;
 
     for (DatanodeInfo dn : getAllNodes()) {
       try {
@@ -173,6 +175,10 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() {
         ComponentVersion dnApparentVersion = dn.getLastKnownApparentVersion();
         ComponentVersion dnSoftwareVersion = dn.getLastKnownSoftwareVersion();
 
+        int dnApparentVersionInt = dnApparentVersion.serialize();
+        minApparentVersion = Math.min(minApparentVersion, dnApparentVersionInt);
+        maxApparentVersion = Math.max(maxApparentVersion, dnApparentVersionInt);
+
         if (!dnApparentVersion.equals(dnSoftwareVersion)) {
           // Datanode has not yet finalized
           LOG.debug("Datanode {} has not yet finalized: apparent version={}, software version={}",
@@ -187,7 +193,17 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() {
       }
     }
 
-    return new DatanodeFinalizationCounts(finalizedNodes, totalHealthyNodes);
+    if (minApparentVersion == Integer.MAX_VALUE) {
+      // No healthy datanode with version info was found
+      minApparentVersion = 0;
+    }
+
+    return DatanodeFinalizationCounts.newBuilder()
+        .setNumFinalizedDatanodes(finalizedNodes)
+        .setTotalHealthyDatanodes(totalHealthyNodes)
+        .setMinApparentVersion(minApparentVersion)
+        .setMaxApparentVersion(maxApparentVersion)
+        .build();
   }
 
   /**
@@ -498,11 +514,18 @@ default void removeNode(DatanodeDetails datanodeDetails) throws NodeNotFoundExce
   final class DatanodeFinalizationCounts {
     private final int numFinalizedDatanodes;
     private final int totalHealthyDatanodes;
+    private final int minApparentVersion;
+    private final int maxApparentVersion;
+
+    private DatanodeFinalizationCounts(Builder b) {
+      this.numFinalizedDatanodes = b.numFinalizedDatanodes;
+      this.totalHealthyDatanodes = b.totalHealthyDatanodes;
+      this.minApparentVersion = b.minApparentVersion;
+      this.maxApparentVersion = b.maxApparentVersion;
+    }
 
-    public DatanodeFinalizationCounts(int numFinalizedDatanodes,
-                                      int totalHealthyDatanodes) {
-      this.numFinalizedDatanodes = numFinalizedDatanodes;
-      this.totalHealthyDatanodes = totalHealthyDatanodes;
+    public static Builder newBuilder() {
+      return new Builder();
     }
 
     public int getNumFinalizedDatanodes() {
@@ -516,5 +539,47 @@ public int getTotalHealthyDatanodes() {
     public boolean allNodesFinalized() {
       return numFinalizedDatanodes == totalHealthyDatanodes;
     }
+
+    public int getMinApparentVersion() {
+      return minApparentVersion;
+    }
+
+    public int getMaxApparentVersion() {
+      return maxApparentVersion;
+    }
+
+    /**
+     * Builder for {@link DatanodeFinalizationCounts}.
+     */
+    public static final class Builder {
+      private int numFinalizedDatanodes;
+      private int totalHealthyDatanodes;
+      private int minApparentVersion;
+      private int maxApparentVersion;
+
+      public Builder setNumFinalizedDatanodes(int value) {
+        this.numFinalizedDatanodes = value;
+        return this;
+      }
+
+      public Builder setTotalHealthyDatanodes(int value) {
+        this.totalHealthyDatanodes = value;
+        return this;
+      }
+
+      public Builder setMinApparentVersion(int value) {
+        this.minApparentVersion = value;
+        return this;
+      }
+
+      public Builder setMaxApparentVersion(int value) {
+        this.maxApparentVersion = value;
+        return this;
+      }
+
+      public DatanodeFinalizationCounts build() {
+        return new DatanodeFinalizationCounts(this);
+      }
+    }
   }
 }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java

@@ -1214,18 +1214,26 @@ public HddsProtos.UpgradeStatus queryUpgradeStatus() throws IOException {
     try {
       getScm().checkAdminAccess(getRemoteUser(), true);
 
+      if (scm.getScmContext().isInSafeMode()) {
+        throw new SCMException("Cannot query upgrade status while SCM is in safe mode. Wait until SCM exits "
+            + "safe mode and try again.", ResultCodes.SAFE_MODE_EXCEPTION);
+      }
+
       boolean scmFinalized = !scm.getVersionManager().needsFinalization();
       NodeManager.DatanodeFinalizationCounts datanodeFinalizationCounts =
           scm.getScmNodeManager().getDatanodeFinalizationCounts();
       int finalizedDatanodes = datanodeFinalizationCounts.getNumFinalizedDatanodes();
       int healthyDatanodes = datanodeFinalizationCounts.getTotalHealthyDatanodes();
-      boolean shouldFinalize = scmFinalized && datanodeFinalizationCounts.allNodesFinalized() && !scm.isInSafeMode();
+      boolean shouldFinalize = scmFinalized && datanodeFinalizationCounts.allNodesFinalized();
 
       HddsProtos.UpgradeStatus result = HddsProtos.UpgradeStatus.newBuilder()
           .setScmFinalized(scmFinalized)
           .setNumDatanodesFinalized(finalizedDatanodes)
           .setNumDatanodesTotal(healthyDatanodes)
           .setShouldFinalize(shouldFinalize)
+          .setScmApparentVersion(scm.getVersionManager().getApparentVersion().serialize())
+          .setMinDatanodeApparentVersion(datanodeFinalizationCounts.getMinApparentVersion())
+          .setMaxDatanodeApparentVersion(datanodeFinalizationCounts.getMaxApparentVersion())
           .build();
 
       AUDIT.logReadSuccess(buildAuditMessageForSuccess(SCMAction.QUERY_UPGRADE_STATUS, null));

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java

@@ -684,6 +684,36 @@ public void testDatanodeFinalizedCounterTracksRegistrationAndRemoveNode()
     }
   }
 
+  @Test
+  public void testDatanodeFinalizationCountsTracksApparentVersionRange()
+      throws IOException, AuthenticationException {
+    try (SCMNodeManager nodeManager = createNodeManager(getConf())) {
+      // Two healthy datanodes reporting different apparent versions.
+      registerWithCapacity(nodeManager,
+          toVersionProto(HDDSLayoutFeature.INITIAL_VERSION, HDDSVersion.SOFTWARE_VERSION), success);
+      registerWithCapacity(nodeManager, defaultVersionProto(), success);
+
+      NodeManager.DatanodeFinalizationCounts counts = nodeManager.getDatanodeFinalizationCounts();
+      assertEquals(2, counts.getTotalHealthyDatanodes());
+      assertEquals(HDDSLayoutFeature.INITIAL_VERSION.serialize(), counts.getMinApparentVersion(),
+          "Min apparent version should reflect the least-finalized datanode");
+      assertEquals(HDDSVersion.SOFTWARE_VERSION.serialize(), counts.getMaxApparentVersion(),
+          "Max apparent version should reflect the most-finalized datanode");
+    }
+  }
+
+  @Test
+  public void testDatanodeFinalizationCountsWithNoHealthyDatanodes()
+      throws IOException, AuthenticationException {
+    try (SCMNodeManager nodeManager = createNodeManager(getConf())) {
+      NodeManager.DatanodeFinalizationCounts counts = nodeManager.getDatanodeFinalizationCounts();
+      assertEquals(0, counts.getTotalHealthyDatanodes());
+      // The MAX_VALUE sentinel used to find the min must be normalized to 0 when there is nothing to count.
+      assertEquals(0, counts.getMinApparentVersion());
+      assertEquals(0, counts.getMaxApparentVersion());
+    }
+  }
+
   private static Stream<Arguments> ineligibleHealthStates() {
     return Stream.of(
         Arguments.of(NodeStatus.inServiceStale()),

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/TestSCMClientProtocolServer.java

@@ -21,7 +21,6 @@
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_READONLY_ADMINISTRATORS;
 import static org.apache.hadoop.ozone.upgrade.UpgradeFinalization.Status.ALREADY_FINALIZED;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
@@ -50,12 +49,14 @@
 import org.apache.hadoop.hdds.scm.HddsTestUtils;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerManagerImpl;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
 import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
 import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
 import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocolServerSideTranslatorPB;
 import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager;
+import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager.SafeModeStatus;
 import org.apache.hadoop.hdds.scm.server.upgrade.FinalizationManager;
 import org.apache.hadoop.hdds.scm.server.upgrade.ScmVersionManager;
 import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
@@ -288,20 +289,19 @@ public void testQueryUpgradeStatus() throws Exception {
   }
 
   @Test
-  public void testQueryUpgradeStatusInSafemode() throws Exception {
-    // mockSafeModeManager defaults to returning true for getInSafeMode()
-    when(mockSafeModeManager.getInSafeMode()).thenReturn(true);
-    assertTrue(scm.isInSafeMode());
-
-    HddsProtos.UpgradeStatus status = server.queryUpgradeStatus();
+  public void testQueryUpgradeStatusInSafemode() {
+    // Put SCM into safe mode via the context the server consults.
+    scm.getScmContext().updateSafeModeStatus(SafeModeStatus.INITIAL);
+    try {
+      assertTrue(scm.getScmContext().isInSafeMode());
 
-    // SCM starts already finalized in tests
-    assertTrue(status.getScmFinalized());
-    // No datanodes registered
-    assertEquals(0, status.getNumDatanodesFinalized());
-    assertEquals(0, status.getNumDatanodesTotal());
-    // shouldFinalize is false because SCM is in safe mode
-    assertFalse(status.getShouldFinalize());
+      // Querying upgrade status is blocked while SCM is in safe mode.
+      SCMException ex = assertThrows(SCMException.class, () -> server.queryUpgradeStatus());
+      assertEquals(SCMException.ResultCodes.SAFE_MODE_EXCEPTION, ex.getResult());
+    } finally {
+      // Restore for other tests sharing the static SCM instance.
+      scm.getScmContext().updateSafeModeStatus(SafeModeStatus.OUT_OF_SAFE_MODE);
+    }
   }
 
   private ContainerInfo newContainerWithLastUsedTime(long containerId,

hadoop-ozone/cli-admin/src/main/java/org/apache/hadoop/ozone/admin/upgrade/FinalizeSubCommand.java

@@ -17,29 +17,42 @@
 
 package org.apache.hadoop.ozone.admin.upgrade;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.hdds.cli.AbstractSubcommand;
 import org.apache.hadoop.hdds.cli.HddsVersionProvider;
 import org.apache.hadoop.ozone.OzoneManagerVersion;
 import org.apache.hadoop.ozone.admin.om.OmAddressOptions;
 import org.apache.hadoop.ozone.client.rpc.RpcClient;
 import org.apache.hadoop.ozone.om.protocol.OzoneManagerProtocol;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.QueryUpgradeStatusResponse;
 import picocli.CommandLine;
 
 /**
  * Handler for the ozone admin upgrade finalize command.
  */
 @CommandLine.Command(
     name = "finalize",
-    description = "Initiates the the process to finalize a cluster upgrade.",
+    description = "Initiates the process to finalize a cluster upgrade. This command is idempotent.",
     mixinStandardHelpOptions = true,
     versionProvider = HddsVersionProvider.class
 )
 public class FinalizeSubCommand extends AbstractSubcommand implements Callable<Integer> {
 
+  /** Poll cadence used by {@code --wait}. Overridable from tests via {@link #setPollIntervalMillis(long)}. */
+  private long pollIntervalMillis = TimeUnit.SECONDS.toMillis(5);
+
   @CommandLine.Mixin
   private OmAddressOptions.OptionalServiceIdOrHostMixin omAddressOptions;
 
+  @CommandLine.Option(names = {"--wait"},
+      defaultValue = "false",
+      description = "After initiating finalization, poll the cluster status until the entire cluster (OM, SCM, "
+          + "and all healthy datanodes) is finalized. Interrupt with Ctrl-C to stop waiting; finalization "
+          + "continues on the server.")
+  private boolean wait;
+
   @Override
   public Integer call() throws Exception {
     try (OzoneManagerProtocol client = getClient()) {
@@ -50,12 +63,63 @@ public Integer call() throws Exception {
         return 1;
       }
       client.finalizeUpgrade();
+      if (wait) {
+        out().println("Cluster finalization has been started. Waiting for the cluster to finalize; "
+            + "interrupt with Ctrl-C to stop waiting (finalization continues on the server).");
+        return waitForFinalization(client);
+      }
       out().println("Cluster finalization has been started. Monitor progress with `ozone admin upgrade status`");
     }
     return 0;
   }
 
+  /**
+   * Polls the cluster status until OM, SCM and all healthy datanodes report finalized, or the operator
+   * interrupts the command. A failed status query is reported to stderr and retried on the next poll,
+   * so transient RPC errors do not abort the wait. {@code --wait} is safe to re-run: if the cluster is
+   * already finalized, the first poll returns done and the command exits 0.
+   */
+  private int waitForFinalization(OzoneManagerProtocol client) {
+    while (true) {
+      QueryUpgradeStatusResponse status = null;
+      try {
+        status = client.queryUpgradeStatus();
+      } catch (Exception e) {
+        err().println("Failed to query upgrade status: " + e.getMessage()
+            + ". Retrying, or use `ozone admin upgrade status` to monitor progress.");
+      }
+
+      if (status != null) {
+        // Finalization checks before sleeping, so an already-finalized cluster returns without waiting.
+        if (status.getClusterFinalized()) {
+          out().println("Finalization complete.");
+          return 0;
+        }
+
+        if (isVerbose()) {
+          StatusSubCommand.printVerbose(status, out());
+        } else {
+          StatusSubCommand.printBasic(status, out());
+        }
+        out().flush();
+      }
+
+      try {
+        Thread.sleep(pollIntervalMillis);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        out().println("Waiting interrupted. Use `ozone admin upgrade status` to monitor progress.");
+        return 0;
+      }
+    }
+  }
+
   protected OzoneManagerProtocol getClient() throws Exception {
     return omAddressOptions.newClient();
   }
+
+  @VisibleForTesting
+  void setPollIntervalMillis(long millis) {
+    this.pollIntervalMillis = millis;
+  }
 }