Track state of upgrade to eliminate re-running upgrade code (#5357)

During an upgrade the UpgradeCoordinator will run Upgraders
for each version between the current version of the stored
data and the current version of the installed software. The
UpgradeCoordinator runs all of the necessary Upgraders on
ZooKeeper, then runs all of them on the root table, and then
finally runs all of them on the metadata table. This means
that when upgrading through multiple versions the version of
the data stored in ZooKeeper could be multiple versions ahead
of the root and metadata tables, until the Upgraders are run
on those tables.

When a failure occurs in the Upgrade the system is left in a
partially upgraded state. The user needs to find and fix any
issues before trying to start the system again. When the Manager
starts the next time it will attempt to do the upgrade from
the very beginning. An Upgrader implementation for one version
needs to be coded to handle the upgrade process running again,
so it needs to take into account the fact that the ZooKeeper
data has already been modified and skip that step, for example.
However, an Upgrader for one version can't account for the
changes in a future version.

This change creates a temporary object stored in ZooKeeper that
is created when an upgrade starts and deleted when the upgrade
finishes. It keeps track of the version of the ZooKeeper, root,
and metadata as the Upgraders are run against those objects and
the overall target version of the upgrade. TheUpgradeCoordinator
has been modified to use this object so that it does not re-run
Upgraders on those objects when they have already run successfully.

This change also fixes EasyMock usage in AccumuloTest unit test for
upgrades observed while adding the context to the
UpgradeCoordinator constructor and moves the PreUpgradeValidation
inside the UpgradeCoordinator (since they are related responsibilities).

Closes #5347


Co-authored-by: Christopher Tubbs <ctubbsii@apache.org>

Author: dlmarion

core/src/main/java/org/apache/accumulo/core/Constants.java

@@ -84,6 +84,8 @@ public class Constants {
   public static final String ZHDFS_RESERVATIONS = "/hdfs_reservations";
   public static final String ZRECOVERY = "/recovery";
 
+  public static final String ZUPGRADE_PROGRESS = "/upgrade_progress";
+
   /**
    * Base znode for storing secret keys that back delegation tokens
    */

server/manager/src/main/java/org/apache/accumulo/manager/Manager.java

@@ -116,7 +116,6 @@
 import org.apache.accumulo.manager.recovery.RecoveryManager;
 import org.apache.accumulo.manager.state.TableCounts;
 import org.apache.accumulo.manager.tableOps.TraceRepo;
-import org.apache.accumulo.manager.upgrade.PreUpgradeValidation;
 import org.apache.accumulo.manager.upgrade.UpgradeCoordinator;
 import org.apache.accumulo.server.AbstractServer;
 import org.apache.accumulo.server.HighlyAvailableService;
@@ -305,21 +304,22 @@ synchronized void setManagerState(final ManagerState newState) {
         break;
       case HAVE_LOCK:
         if (isUpgrading()) {
-          new PreUpgradeValidation().validate(getContext(), nextEvent);
-          upgradeCoordinator.upgradeZookeeper(getContext(), nextEvent);
+          upgradeCoordinator.preUpgradeValidation();
+          upgradeCoordinator.startOrContinueUpgrade();
+          upgradeCoordinator.upgradeZookeeper(nextEvent);
         }
         break;
       case NORMAL:
         if (isUpgrading()) {
-          upgradeMetadataFuture = upgradeCoordinator.upgradeMetadata(getContext(), nextEvent);
+          upgradeMetadataFuture = upgradeCoordinator.upgradeMetadata(nextEvent);
         }
         break;
       default:
         break;
     }
   }
 
-  private final UpgradeCoordinator upgradeCoordinator = new UpgradeCoordinator();
+  private final UpgradeCoordinator upgradeCoordinator;
 
   private Future<Void> upgradeMetadataFuture;
 
@@ -424,6 +424,7 @@ public static void main(String[] args) throws Exception {
   protected Manager(ConfigOpts opts, String[] args) throws IOException {
     super("manager", opts, args);
     ServerContext context = super.getContext();
+    upgradeCoordinator = new UpgradeCoordinator(context);
     balancerEnvironment = new BalancerEnvironmentImpl(context);
 
     AccumuloConfiguration aconf = context.getConfiguration();
@@ -1277,6 +1278,9 @@ public void run() {
     } catch (KeeperException | InterruptedException e) {
       throw new IllegalStateException("Exception getting manager lock", e);
     }
+    // Setting the Manager state to HAVE_LOCK has the side-effect of
+    // starting the upgrade process if necessary.
+    setManagerState(ManagerState.HAVE_LOCK);
 
     MetricsInfo metricsInfo = getContext().getMetricsInfo();
 
@@ -1632,7 +1636,6 @@ private ServiceLockData getManagerLock(final ServiceLockPath zManagerLoc)
       sleepUninterruptibly(TIME_TO_WAIT_BETWEEN_LOCK_CHECKS, MILLISECONDS);
     }
 
-    setManagerState(ManagerState.HAVE_LOCK);
     return sld;
   }
 

server/manager/src/main/java/org/apache/accumulo/manager/upgrade/PreUpgradeValidation.java

@@ -25,7 +25,6 @@
 
 import org.apache.accumulo.core.zookeeper.ZooSession;
 import org.apache.accumulo.core.zookeeper.ZooSession.ZKUtil;
-import org.apache.accumulo.manager.EventCoordinator;
 import org.apache.accumulo.server.AccumuloDataVersion;
 import org.apache.accumulo.server.ServerContext;
 import org.apache.zookeeper.KeeperException;
@@ -48,10 +47,11 @@ public class PreUpgradeValidation {
 
   private final static Logger log = LoggerFactory.getLogger(PreUpgradeValidation.class);
 
-  public void validate(final ServerContext context, final EventCoordinator eventCoordinator) {
-    int cv = AccumuloDataVersion.getCurrentVersion(context);
-    if (cv == AccumuloDataVersion.get()) {
-      log.debug("already at current data version: {}, skipping validation", cv);
+  public void validate(final ServerContext context) {
+    int storedVersion = AccumuloDataVersion.getCurrentVersion(context);
+    int currentVersion = AccumuloDataVersion.get();
+    if (storedVersion == currentVersion) {
+      log.debug("already at current data version: {}, skipping validation", currentVersion);
       return;
     }
     validateACLs(context);

server/manager/src/main/java/org/apache/accumulo/manager/upgrade/UpgradeCoordinator.java

@@ -132,12 +132,30 @@ public boolean isParentLevelUpgraded(KeyExtent extent) {
       Collections.unmodifiableMap(new TreeMap<>(Map.of(ROOT_TABLET_META_CHANGES,
           new Upgrader10to11(), REMOVE_DEPRECATIONS_FOR_VERSION_3, new Upgrader11to12())));
 
+  private final ServerContext context;
+  private final UpgradeProgressTracker progressTracker;
+  private final PreUpgradeValidation preUpgradeValidator;
+
   private volatile UpgradeStatus status;
 
-  public UpgradeCoordinator() {
+  public UpgradeCoordinator(ServerContext context) {
+    this.context = context;
+    progressTracker = new UpgradeProgressTracker(context);
+    preUpgradeValidator = new PreUpgradeValidation();
     status = UpgradeStatus.INITIAL;
   }
 
+  public void preUpgradeValidation() {
+    preUpgradeValidator.validate(context);
+  }
+
+  public void startOrContinueUpgrade() {
+    // The following check will fail if an upgrade is in progress
+    // but the target version is not the current version of the
+    // software.
+    progressTracker.startOrContinueUpgrade();
+  }
+
   private void setStatus(UpgradeStatus status, EventCoordinator eventCoordinator) {
     UpgradeStatus oldStatus = this.status;
     this.status = status;
@@ -155,8 +173,7 @@ private void handleFailure(Exception e) {
     System.exit(1);
   }
 
-  public synchronized void upgradeZookeeper(ServerContext context,
-      EventCoordinator eventCoordinator) {
+  public synchronized void upgradeZookeeper(EventCoordinator eventCoordinator) {
 
     Preconditions.checkState(status == UpgradeStatus.INITIAL,
         "Not currently in a suitable state to do zookeeper upgrade %s", status);
@@ -171,15 +188,24 @@ public synchronized void upgradeZookeeper(ServerContext context,
       }
 
       if (currentVersion < AccumuloDataVersion.get()) {
-        abortIfFateTransactions(context);
+        abortIfFateTransactions();
+
+        final UpgradeProgress progress = progressTracker.getProgress();
 
         for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+          if (progress.getZooKeeperVersion() >= currentVersion) {
+            log.info(
+                "ZooKeeper has already been upgraded to version {}, moving on to next upgrader",
+                currentVersion);
+            continue;
+          }
           log.info("Upgrading Zookeeper - current version {} as step towards target version {}", v,
               AccumuloDataVersion.get());
           var upgrader = upgraders.get(v);
           Objects.requireNonNull(upgrader,
               "upgrade ZooKeeper: failed to find upgrader for version " + currentVersion);
           upgrader.upgradeZookeeper(context);
+          progressTracker.updateZooKeeperVersion(v);
         }
       }
 
@@ -190,8 +216,7 @@ public synchronized void upgradeZookeeper(ServerContext context,
 
   }
 
-  public synchronized Future<Void> upgradeMetadata(ServerContext context,
-      EventCoordinator eventCoordinator) {
+  public synchronized Future<Void> upgradeMetadata(EventCoordinator eventCoordinator) {
     if (status == UpgradeStatus.COMPLETE) {
       return CompletableFuture.completedFuture(null);
     }
@@ -205,35 +230,51 @@ public synchronized Future<Void> upgradeMetadata(ServerContext context,
           .numMaxThreads(Integer.MAX_VALUE).withTimeOut(60L, SECONDS)
           .withQueue(new SynchronousQueue<>()).build().submit(() -> {
             try {
+              UpgradeProgress progress = progressTracker.getProgress();
               for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+                if (progress.getRootVersion() >= currentVersion) {
+                  log.info(
+                      "Root table has already been upgraded to version {}, moving on to next upgrader",
+                      currentVersion);
+                  continue;
+                }
                 log.info("Upgrading Root - current version {} as step towards target version {}", v,
                     AccumuloDataVersion.get());
                 var upgrader = upgraders.get(v);
                 Objects.requireNonNull(upgrader,
                     "upgrade root: failed to find root upgrader for version " + currentVersion);
                 upgraders.get(v).upgradeRoot(context);
+                progressTracker.updateRootVersion(v);
               }
               setStatus(UpgradeStatus.UPGRADED_ROOT, eventCoordinator);
 
               for (int v = currentVersion; v < AccumuloDataVersion.get(); v++) {
+                if (progress.getMetadataVersion() >= currentVersion) {
+                  log.info(
+                      "Metadata table has already been upgraded to version {}, moving on to next upgrader",
+                      currentVersion);
+                  continue;
+                }
                 log.info(
                     "Upgrading Metadata - current version {} as step towards target version {}", v,
                     AccumuloDataVersion.get());
                 var upgrader = upgraders.get(v);
                 Objects.requireNonNull(upgrader,
                     "upgrade metadata: failed to find upgrader for version " + currentVersion);
                 upgraders.get(v).upgradeMetadata(context);
+                progressTracker.updateMetadataVersion(v);
               }
               setStatus(UpgradeStatus.UPGRADED_METADATA, eventCoordinator);
 
               log.info("Validating configuration properties.");
-              validateProperties(context);
+              validateProperties();
 
               log.info("Updating persistent data version.");
               updateAccumuloVersion(context.getServerDirs(), context.getVolumeManager(),
                   currentVersion);
               log.info("Upgrade complete");
               setStatus(UpgradeStatus.COMPLETE, eventCoordinator);
+              progressTracker.upgradeComplete();
             } catch (Exception e) {
               handleFailure(e);
             }
@@ -244,7 +285,7 @@ public synchronized Future<Void> upgradeMetadata(ServerContext context,
     }
   }
 
-  private void validateProperties(ServerContext context) {
+  private void validateProperties() {
     ConfigCheckUtil.validate(context.getSiteConfiguration(), "site configuration");
     ConfigCheckUtil.validate(context.getConfiguration(), "system configuration");
     try {
@@ -311,7 +352,7 @@ public UpgradeStatus getStatus() {
    */
   @SuppressFBWarnings(value = "DM_EXIT",
       justification = "Want to immediately stop all manager threads on upgrade error")
-  private void abortIfFateTransactions(ServerContext context) {
+  private void abortIfFateTransactions() {
     try {
       final ReadOnlyTStore<UpgradeCoordinator> fate =
           new ZooStore<>(context.getZooKeeperRoot() + Constants.ZFATE, context.getZooSession());

server/manager/src/main/java/org/apache/accumulo/manager/upgrade/UpgradeProgress.java

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.manager.upgrade;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.accumulo.core.util.LazySingletons.GSON;
+
+/**
+ * Track upgrade progress for each component. The version stored is the most recent version for
+ * which an upgrade has been completed.
+ */
+public class UpgradeProgress {
+
+  private int zooKeeperVersion;
+  private int rootVersion;
+  private int metadataVersion;
+  private int upgradeTargetVersion;
+
+  public UpgradeProgress() {}
+
+  public UpgradeProgress(int currentVersion, int targetVersion) {
+    zooKeeperVersion = currentVersion;
+    rootVersion = currentVersion;
+    metadataVersion = currentVersion;
+    upgradeTargetVersion = targetVersion;
+  }
+
+  public void setZooKeeperVersion(int version) {
+    zooKeeperVersion = version;
+  }
+
+  public int getZooKeeperVersion() {
+    return zooKeeperVersion;
+  }
+
+  public void setRootVersion(int version) {
+    rootVersion = version;
+  }
+
+  public int getRootVersion() {
+    return rootVersion;
+  }
+
+  public void setMetadataVersion(int version) {
+    metadataVersion = version;
+  }
+
+  public int getMetadataVersion() {
+    return metadataVersion;
+  }
+
+  public int getUpgradeTargetVersion() {
+    return upgradeTargetVersion;
+  }
+
+  public byte[] toJsonBytes() {
+    return GSON.get().toJson(this).getBytes(UTF_8);
+  }
+
+  public static UpgradeProgress fromJsonBytes(byte[] jsonData) {
+    return GSON.get().fromJson(new String(jsonData, UTF_8), UpgradeProgress.class);
+  }
+
+}