HDDS-8208. [SNAPSHOT] Allow om snapshot bootstrap tarballs to be incremental.

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java

@@ -54,7 +54,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.io.FilenameFilter;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -66,6 +65,8 @@
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CA_LIST_RETRY_INTERVAL;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CA_LIST_RETRY_INTERVAL_DEFAULT;
@@ -357,26 +358,26 @@ public static File getMetaDir(DBDefinition definition,
   }
 
   /**
-   * Scan the DB dir and return the existing SST files.
+   * Scan the DB dir and return the existing SST files,
+   * including omSnapshot sst files.
    * SSTs could be used for avoiding repeated download.
    *
    * @param db the file representing the DB to be scanned
    * @return the list of SST file name. If db not exist, will return empty list
    */
-  public static List<String> getExistingSstFiles(File db) {
+  public static List<String> getExistingSstFiles(File db) throws IOException {
     List<String> sstList = new ArrayList<>();
     if (!db.exists()) {
       return sstList;
     }
-    FilenameFilter filter = new FilenameFilter() {
-      @Override
-      public boolean accept(File dir, String name) {
-        return name.endsWith(ROCKSDB_SST_SUFFIX);
-      }
-    };
-    String[] tempArray = db.list(filter);
-    if (tempArray != null) {
-      sstList = Arrays.asList(tempArray);
+
+    int truncateLength = db.toString().length() + 1;
+    // Walk the db dir and get all sst files including omSnapshot files.
+    try (Stream<Path> files = Files.walk(db.toPath())) {
+      sstList =
+          files.filter(path -> path.toString().endsWith(ROCKSDB_SST_SUFFIX)).
+              map(p -> p.toString().substring(truncateLength)).
+              collect(Collectors.toList());
       if (LOG.isDebugEnabled()) {
         LOG.debug("Scanned SST files {} in {}.", sstList, db.getAbsolutePath());
       }

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/RDBSnapshotProvider.java

@@ -60,6 +60,7 @@ public abstract class RDBSnapshotProvider implements Closeable {
   private final AtomicReference<String> lastLeaderRef;
   private final AtomicLong numDownloaded;
   private FaultInjector injector;
+  private final AtomicLong initCount;
 
   public RDBSnapshotProvider(File snapshotDir, String dbName) {
     this.snapshotDir = snapshotDir;
@@ -68,6 +69,7 @@ public RDBSnapshotProvider(File snapshotDir, String dbName) {
     this.injector = null;
     this.lastLeaderRef = new AtomicReference<>(null);
     this.numDownloaded = new AtomicLong();
+    this.initCount = new AtomicLong();
     init();
   }
 
@@ -91,6 +93,7 @@ public synchronized void init() {
 
     // reset leader info
     lastLeaderRef.set(null);
+    initCount.incrementAndGet();
   }
 
   /**
@@ -112,13 +115,14 @@ public DBCheckpoint downloadDBSnapshotFromLeader(String leaderNodeID)
     LOG.info("Successfully download the latest snapshot {} from leader OM: {}",
         targetFile, leaderNodeID);
 
+    numDownloaded.incrementAndGet();
+    injectPause();
+
     RocksDBCheckpoint checkpoint = getCheckpointFromSnapshotFile(targetFile,
         candidateDir, true);
     LOG.info("Successfully untar the downloaded snapshot {} at {}.", targetFile,
         checkpoint.getCheckpointLocation());
 
-    numDownloaded.incrementAndGet();
-    injectPause();
     return checkpoint;
   }
 
@@ -131,7 +135,8 @@ public DBCheckpoint downloadDBSnapshotFromLeader(String leaderNodeID)
    *
    * @param currentLeader the ID of leader node
    */
-  private void checkLeaderConsistent(String currentLeader) {
+  @VisibleForTesting
+  void checkLeaderConsistent(String currentLeader) throws IOException {
     String lastLeader = lastLeaderRef.get();
     if (lastLeader != null) {
       if (!lastLeader.equals(currentLeader)) {
@@ -230,4 +235,9 @@ public void setInjector(FaultInjector injector) {
   public long getNumDownloaded() {
     return numDownloaded.get();
   }
+
+  @VisibleForTesting
+  public long getInitCount() {
+    return initCount.get();
+  }
 }

hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/utils/TestRDBSnapshotProvider.java

@@ -46,6 +46,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -58,6 +59,7 @@
 import static org.apache.hadoop.hdds.utils.db.TestRDBStore.newRDBStore;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -98,7 +100,7 @@ public void init(@TempDir File tempDir) throws Exception {
         MAX_DB_UPDATES_SIZE_THRESHOLD);
     rdbSnapshotProvider = new RDBSnapshotProvider(testDir, "test.db") {
       @Override
-      public void close() throws IOException {
+      public void close() {
       }
 
       @Override
@@ -236,4 +238,28 @@ public void insertRandomData(RDBStore dbStore, int familyIndex)
       throw new IOException(e);
     }
   }
+
+  @Test
+  public void testCheckLeaderConsistent() throws IOException {
+    // Leader initialized to null at startup.
+    assertEquals(1, rdbSnapshotProvider.getInitCount());
+    File dummyFile = new File(rdbSnapshotProvider.getCandidateDir(),
+        "file1.sst");
+    Files.write(dummyFile.toPath(),
+        "dummyData".getBytes(StandardCharsets.UTF_8));
+    assertTrue(dummyFile.exists());
+
+    // Set the leader.
+    rdbSnapshotProvider.checkLeaderConsistent("node1");
+    assertEquals(2, rdbSnapshotProvider.getInitCount());
+    assertFalse(dummyFile.exists());
+
+    // Confirm setting the same leader doesn't reinitialize.
+    rdbSnapshotProvider.checkLeaderConsistent("node1");
+    assertEquals(2, rdbSnapshotProvider.getInitCount());
+
+    // Confirm setting different leader does reinitialize.
+    rdbSnapshotProvider.checkLeaderConsistent("node2");
+    assertEquals(3, rdbSnapshotProvider.getInitCount());
+  }
 }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -233,7 +233,7 @@ private OMConfigKeys() {
       "ozone.om.snapshot.provider.request.timeout";
   public static final TimeDuration
       OZONE_OM_SNAPSHOT_PROVIDER_REQUEST_TIMEOUT_DEFAULT =
-      TimeDuration.valueOf(5000, TimeUnit.MILLISECONDS);
+      TimeDuration.valueOf(300000, TimeUnit.MILLISECONDS);
 
   public static final String OZONE_OM_FS_SNAPSHOT_MAX_LIMIT =
       "ozone.om.fs.snapshot.max.limit";

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOMDbCheckpointServlet.java

@@ -121,6 +121,7 @@ public class TestOMDbCheckpointServlet {
   private String snapshotDirName2;
   private Path compactionDirPath;
   private DBCheckpoint dbCheckpoint;
+  private static final String FABRICATED_FILE_NAME = "fabricatedFile.sst";
 
   @Rule
   public Timeout timeout = Timeout.seconds(240);
@@ -322,7 +323,7 @@ public void testWriteDbDataToStream() throws Exception {
     String newDbDirName = testDirName + OM_KEY_PREFIX + OM_DB_NAME;
     int newDbDirLength = newDbDirName.length() + 1;
     File newDbDir = new File(newDbDirName);
-    newDbDir.mkdirs();
+    Assert.assertTrue(newDbDir.mkdirs());
     FileUtil.unTar(tempFile, newDbDir);
 
     // Move snapshot dir to correct location.
@@ -361,7 +362,7 @@ public void testWriteDbDataToStream() throws Exception {
       for (String line : lines.collect(Collectors.toList())) {
         Assert.assertFalse("CURRENT file is not a hard link",
             line.contains("CURRENT"));
-        if (line.contains("fabricatedFile")) {
+        if (line.contains(FABRICATED_FILE_NAME)) {
           fabricatedLinkLines.add(line);
         } else {
           checkLine(shortSnapshotLocation, shortSnapshotLocation2, line);
@@ -477,15 +478,15 @@ private void prepSnapshotData() throws Exception {
     Path fabricatedSnapshot  = Paths.get(
         new File(snapshotDirName).getParent(),
         "fabricatedSnapshot");
-    fabricatedSnapshot.toFile().mkdirs();
-    Assert.assertTrue(Paths.get(fabricatedSnapshot.toString(), "fabricatedFile")
-        .toFile().createNewFile());
+    Assert.assertTrue(fabricatedSnapshot.toFile().mkdirs());
+    Assert.assertTrue(Paths.get(fabricatedSnapshot.toString(),
+        FABRICATED_FILE_NAME).toFile().createNewFile());
 
     // Create fabricated links to snapshot dirs
     // to confirm that links are recognized even if
-    // they are don't point to the checkpoint directory.
-    Path fabricatedFile = Paths.get(snapshotDirName, "fabricatedFile");
-    Path fabricatedLink = Paths.get(snapshotDirName2, "fabricatedFile");
+    // they don't point to the checkpoint directory.
+    Path fabricatedFile = Paths.get(snapshotDirName, FABRICATED_FILE_NAME);
+    Path fabricatedLink = Paths.get(snapshotDirName2, FABRICATED_FILE_NAME);
 
     Files.write(fabricatedFile,
         "fabricatedData".getBytes(StandardCharsets.UTF_8));
@@ -495,7 +496,7 @@ private void prepSnapshotData() throws Exception {
     compactionDirPath = Paths.get(metaDir.toString(),
         OM_SNAPSHOT_DIFF_DIR, DB_COMPACTION_SST_BACKUP_DIR);
     Path fabricatedLink2 = Paths.get(compactionDirPath.toString(),
-        "fabricatedFile");
+        FABRICATED_FILE_NAME);
     Files.createLink(fabricatedLink2, fabricatedFile);
     Path currentFile = Paths.get(metaDir.toString(),
                                     OM_DB_NAME, "CURRENT");
@@ -565,7 +566,7 @@ private void checkFabricatedLines(Set<String> directories, List<String> lines,
     // find the real file
     String realDir = null;
     for (String dir: directories) {
-      if (Paths.get(testDirName, dir, "fabricatedFile").toFile().exists()) {
+      if (Paths.get(testDirName, dir, FABRICATED_FILE_NAME).toFile().exists()) {
         Assert.assertNull(
             "Exactly one copy of the fabricated file exists in the tarball",
             realDir);
@@ -589,8 +590,8 @@ private void checkFabricatedLines(Set<String> directories, List<String> lines,
       Path path0 = Paths.get(files[0]);
       Path path1 = Paths.get(files[1]);
       Assert.assertTrue("fabricated entries contains correct file name",
-          path0.getFileName().toString().equals("fabricatedFile") &&
-              path1.getFileName().toString().equals("fabricatedFile"));
+          path0.getFileName().toString().equals(FABRICATED_FILE_NAME) &&
+              path1.getFileName().toString().equals(FABRICATED_FILE_NAME));
     }
   }