HDDS-6961. [Snapshot] Bootstrapping slow followers/new followers.

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -141,6 +141,8 @@ public final class OzoneConsts {
   public static final String STORAGE_DIR_CHUNKS = "chunks";
   public static final String OZONE_DB_CHECKPOINT_REQUEST_FLUSH =
       "flushBeforeCheckpoint";
+  public static final String OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA =
+      "includeSnapshotData";
 
   public static final String RANGER_OZONE_SERVICE_VERSION_KEY =
       "#RANGEROZONESERVICEVERSION";
@@ -562,7 +564,13 @@ private OzoneConsts() {
   public static final int OZONE_MAXIMUM_ACCESS_ID_LENGTH = 100;
 
   public static final String OM_SNAPSHOT_NAME = "snapshotName";
+  public static final String OM_CHECKPOINT_DIR = "db.checkpoints";
   public static final String OM_SNAPSHOT_DIR = "db.snapshots";
+  public static final String OM_SNAPSHOT_CHECKPOINT_DIR = OM_SNAPSHOT_DIR
+      + OM_KEY_PREFIX + "checkpointState";
+  public static final String OM_SNAPSHOT_DIFF_DIR = OM_SNAPSHOT_DIR
+      + OM_KEY_PREFIX + "diffState";
+
   public static final String OM_SNAPSHOT_INDICATOR = ".snapshot";
   public static final String OM_SNAPSHOT_DIFF_DB_NAME = "db.snapdiff";
 

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

@@ -22,6 +22,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.nio.file.Path;
 import java.time.Duration;
 import java.time.Instant;
@@ -162,7 +163,7 @@ public void doGet(HttpServletRequest request, HttpServletResponse response) {
                file + ".tar\"");
 
       Instant start = Instant.now();
-      writeDBCheckpointToStream(checkpoint,
+      writeDbDataToStream(checkpoint, request,
           response.getOutputStream());
       Instant end = Instant.now();
 
@@ -188,4 +189,19 @@ public void doGet(HttpServletRequest request, HttpServletResponse response) {
     }
   }
 
+  /**
+   * Write checkpoint to the stream.
+   *
+   * @param checkpoint The checkpoint to be written.
+   * @param ignoredRequest The httpRequest which generated this checkpoint.
+   *        (Parameter is ignored in this class but used in child classes).
+   * @param destination The stream to write to.
+   */
+  public void writeDbDataToStream(DBCheckpoint checkpoint,
+                                  HttpServletRequest ignoredRequest,
+                                  OutputStream destination)
+      throws IOException, InterruptedException {
+    writeDBCheckpointToStream(checkpoint, destination);
+  }
+
 }

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

@@ -548,7 +548,7 @@ public static void writeDBCheckpointToStream(DBCheckpoint checkpoint,
     }
   }
 
-  private static void includeFile(File file, String entryName,
+  public static void includeFile(File file, String entryName,
                                  ArchiveOutputStream archiveOutputStream)
       throws IOException {
     ArchiveEntry archiveEntry =

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

@@ -20,20 +20,16 @@
 package org.apache.hadoop.hdds.utils.db;
 
 import java.io.Closeable;
-import java.io.File;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.time.Instant;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.utils.db.RocksDatabase.RocksCheckpoint;
-import org.awaitility.core.ConditionTimeoutException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.awaitility.Awaitility.with;
-
 /**
  * RocksDB Checkpoint Manager, used to create and cleanup checkpoints.
  */
@@ -44,9 +40,6 @@ public class RDBCheckpointManager implements Closeable {
   private static final Logger LOG =
       LoggerFactory.getLogger(RDBCheckpointManager.class);
   private final String checkpointNamePrefix;
-  private static final Duration POLL_DELAY_DURATION = Duration.ZERO;
-  private static final Duration POLL_INTERVAL_DURATION = Duration.ofMillis(100);
-  private static final Duration POLL_MAX_DURATION = Duration.ofSeconds(5);
 
   /**
    * Create a checkpoint manager with a prefix to be added to the
@@ -96,7 +89,8 @@ public RocksDBCheckpoint createCheckpoint(String parentDir, String name) {
       LOG.info("Created checkpoint in rocksDB at {} in {} milliseconds",
               checkpointPath, duration);
 
-      waitForCheckpointDirectoryExist(checkpointPath.toFile());
+      RDBCheckpointUtils.waitForCheckpointDirectoryExist(
+          checkpointPath.toFile());
 
       return new RocksDBCheckpoint(
           checkpointPath,
@@ -109,29 +103,6 @@ public RocksDBCheckpoint createCheckpoint(String parentDir, String name) {
     return null;
   }
 
-  /**
-   * Wait for checkpoint directory to be created for 5 secs with 100 millis
-   * poll interval.
-   */
-  public static void waitForCheckpointDirectoryExist(File file)
-      throws IOException {
-    Instant start = Instant.now();
-    try {
-      with().atMost(POLL_MAX_DURATION)
-          .pollDelay(POLL_DELAY_DURATION)
-          .pollInterval(POLL_INTERVAL_DURATION)
-          .await()
-          .until(file::exists);
-      LOG.info("Waited for {} milliseconds for checkpoint directory {}" +
-              " availability.",
-          Duration.between(start, Instant.now()).toMillis(),
-          file.getAbsoluteFile());
-    } catch (ConditionTimeoutException exception) {
-      LOG.info("Checkpoint directory: {} didn't get created in 5 secs.",
-          file.getAbsolutePath());
-    }
-  }
-
   /**
    * Create RocksDB snapshot by saving a checkpoint to a directory.
    *

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

@@ -0,0 +1,70 @@
+/*
+ * 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
+ *
+ *      http://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.hadoop.hdds.utils.db;
+
+import org.awaitility.core.ConditionTimeoutException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Duration;
+import java.time.Instant;
+
+import static org.awaitility.Awaitility.with;
+
+/**
+ * RocksDB Checkpoint Utilities.
+ */
+public final class RDBCheckpointUtils {
+  static final Logger LOG =
+      LoggerFactory.getLogger(RDBCheckpointUtils.class);
+  private static final Duration POLL_DELAY_DURATION = Duration.ZERO;
+  private static final Duration POLL_INTERVAL_DURATION = Duration.ofMillis(100);
+  private static final Duration POLL_MAX_DURATION = Duration.ofSeconds(5);
+
+  private RDBCheckpointUtils() { }
+
+  /**
+   * Wait for checkpoint directory to be created for 5 secs with 100 millis
+   * poll interval.
+   * @param file Checkpoint directory.
+   * @return true if found.
+   */
+  public static boolean waitForCheckpointDirectoryExist(File file)
+      throws IOException {
+    Instant start = Instant.now();
+    try {
+      with().atMost(POLL_MAX_DURATION)
+          .pollDelay(POLL_DELAY_DURATION)
+          .pollInterval(POLL_INTERVAL_DURATION)
+          .await()
+          .until(file::exists);
+      LOG.info("Waited for {} milliseconds for checkpoint directory {}" +
+              " availability.",
+          Duration.between(start, Instant.now()).toMillis(),
+          file.getAbsoluteFile());
+      return true;
+    } catch (ConditionTimeoutException exception) {
+      LOG.info("Checkpoint directory: {} didn't get created in 5 secs.",
+          file.getAbsolutePath());
+      return false;
+    }
+  }
+}

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

@@ -50,9 +50,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.ozone.OzoneConsts.OM_CHECKPOINT_DIR;
+import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_CHECKPOINT_DIR;
+import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIFF_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.DB_COMPACTION_LOG_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.DB_COMPACTION_SST_BACKUP_DIR;
-import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.SNAPSHOT_INFO_TABLE;
 
 /**
@@ -107,8 +110,9 @@ public RDBStore(File dbFile, ManagedDBOptions dbOptions,
     try {
       if (enableCompactionLog) {
         rocksDBCheckpointDiffer = new RocksDBCheckpointDiffer(
-            dbLocation.getParent(), DB_COMPACTION_SST_BACKUP_DIR,
-            DB_COMPACTION_LOG_DIR, dbLocation.toString(),
+            dbLocation.getParent() + OM_KEY_PREFIX + OM_SNAPSHOT_DIFF_DIR,
+            DB_COMPACTION_SST_BACKUP_DIR, DB_COMPACTION_LOG_DIR,
+            dbLocation.toString(),
             maxTimeAllowedForSnapshotInDag, compactionDagDaemonInterval);
         rocksDBCheckpointDiffer.setRocksDBForCompactionTracking(dbOptions);
       } else {
@@ -136,7 +140,7 @@ public RDBStore(File dbFile, ManagedDBOptions dbOptions,
 
       //create checkpoints directory if not exists.
       checkpointsParentDir =
-          Paths.get(dbLocation.getParent(), "db.checkpoints").toString();
+          dbLocation.getParent() + OM_KEY_PREFIX + OM_CHECKPOINT_DIR;
       File checkpointsDir = new File(checkpointsParentDir);
       if (!checkpointsDir.exists()) {
         boolean success = checkpointsDir.mkdir();
@@ -147,15 +151,15 @@ public RDBStore(File dbFile, ManagedDBOptions dbOptions,
         }
       }
 
-      //create snapshot directory if does not exist.
+      //create snapshot checkpoint directory if does not exist.
       snapshotsParentDir = Paths.get(dbLocation.getParent(),
-          OM_SNAPSHOT_DIR).toString();
+          OM_SNAPSHOT_CHECKPOINT_DIR).toString();
       File snapshotsDir = new File(snapshotsParentDir);
       if (!snapshotsDir.exists()) {
-        boolean success = snapshotsDir.mkdir();
+        boolean success = snapshotsDir.mkdirs();
         if (!success) {
           throw new IOException(
-              "Unable to create RocksDB snapshot directory: " +
+              "Unable to create RocksDB snapshot checkpoint directory: " +
               snapshotsParentDir);
         }
       }

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdiff/RocksDBCheckpointDiffer.java

@@ -230,7 +230,7 @@ private String createCompactionLogDir(String metadataDir,
 
     final File parentDir = new File(metadataDir);
     if (!parentDir.exists()) {
-      if (!parentDir.mkdir()) {
+      if (!parentDir.mkdirs()) {
         LOG.error("Error creating compaction log parent dir.");
         return null;
       }

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

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_REQUEST_FLUSH;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
@@ -164,13 +165,15 @@ public String getOMDBCheckpointEnpointUrl(boolean isHttpPolicy) {
       if (StringUtils.isNotEmpty(getHttpAddress())) {
         return "http://" + getHttpAddress() +
             OZONE_DB_CHECKPOINT_HTTP_ENDPOINT +
-            "?" + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "=true";
+            "?" + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "=true&" +
+            OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA + "=true";
       }
     } else {
       if (StringUtils.isNotEmpty(getHttpsAddress())) {
         return "https://" + getHttpsAddress() +
             OZONE_DB_CHECKPOINT_HTTP_ENDPOINT +
-            "?" + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "=true";
+            "?" + OZONE_DB_CHECKPOINT_REQUEST_FLUSH + "=true&" +
+            OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA + "=true";
       }
     }
     return null;

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsSnapshot.java

@@ -29,7 +29,6 @@
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
-import org.apache.hadoop.ozone.om.OMStorage;
 import org.apache.hadoop.ozone.om.OzoneManager;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.util.ToolRunner;
@@ -45,11 +44,10 @@
 import org.junit.jupiter.params.provider.ValueSource;
 
 import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY;
-import static org.apache.hadoop.ozone.OzoneConsts.OM_DB_NAME;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
-import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_OFS_URI_SCHEME;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR;
+import static org.apache.hadoop.ozone.om.OmSnapshotManager.getSnapshotPath;
 
 /**
  * Test client-side CRUD snapshot operations with Ozone Manager.
@@ -332,16 +330,14 @@ private String createSnapshot() throws Exception {
     // Asserts that create request succeeded
     Assertions.assertEquals(0, res);
 
-    File metaDir = OMStorage
-        .getOmDbDir(ozoneManager.getConfiguration());
+    OzoneConfiguration conf = ozoneManager.getConfiguration();
 
     // wait till the snapshot directory exists
     SnapshotInfo snapshotInfo = ozoneManager.getMetadataManager()
         .getSnapshotInfoTable()
         .get(SnapshotInfo.getTableKey(VOLUME, BUCKET, snapshotName));
-    String snapshotDirName = metaDir + OM_KEY_PREFIX +
-        OM_SNAPSHOT_DIR + OM_KEY_PREFIX + OM_DB_NAME +
-        snapshotInfo.getCheckpointDirName() + OM_KEY_PREFIX + "CURRENT";
+    String snapshotDirName = getSnapshotPath(conf, snapshotInfo) +
+        OM_KEY_PREFIX + "CURRENT";
     GenericTestUtils.waitFor(() -> new File(snapshotDirName).exists(),
         1000, 100000);
 

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMDbCheckpointServlet.java

@@ -29,6 +29,7 @@
 import java.util.Collections;
 import java.util.UUID;
 
+import org.apache.commons.compress.compressors.CompressorException;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics;
 import org.apache.hadoop.hdds.scm.server.SCMDBCheckpointServlet;
@@ -46,6 +47,8 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.mockito.Matchers;
+
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
@@ -99,7 +102,9 @@ public void shutdown() {
   }
 
   @Test
-  public void testDoGet() throws ServletException, IOException {
+  public void testDoGet()
+      throws ServletException, IOException, CompressorException,
+      InterruptedException {
 
     File tempFile = null;
     try {
@@ -114,6 +119,8 @@ public void testDoGet() throws ServletException, IOException {
           Collections.emptyList(),
           Collections.emptyList(),
           false);
+      doCallRealMethod().when(scmDbCheckpointServletMock)
+         .writeDbDataToStream(any(), any(), any());
 
       HttpServletRequest requestMock = mock(HttpServletRequest.class);
       HttpServletResponse responseMock = mock(HttpServletResponse.class);