HDDS-15766. Make Recon OM DB large tarball transfer reliable

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

@@ -127,6 +127,10 @@ public final class OzoneConsts {
       "flushBeforeCheckpoint";
   public static final String OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA =
       "includeSnapshotData";
+  // Opt-in request flag: transfer the active om.db in multiple size-bounded,
+  // resumable batches instead of a single tar. Default off (OM followers).
+  public static final String OZONE_DB_CHECKPOINT_CHUNK_ACTIVE_DB =
+      "chunkActiveDb";
   public static final String OZONE_DB_CHECKPOINT_REQUEST_TO_EXCLUDE_SST =
       "toExcludeSST";
 

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

@@ -126,7 +126,7 @@ public DBCheckpoint downloadDBSnapshotFromLeader(String leaderNodeID)
       LOG.info("Successfully untar the downloaded snapshot {} at {}.",
           targetFile, unTarredDb.toAbsolutePath());
       if (ratisSnapshotComplete(unTarredDb)) {
-        LOG.info("Ratis snapshot transfer is complete.");
+        LOG.info("DB snapshot transfer is complete.");
         return getCheckpointFromUntarredDb(unTarredDb);
       }
     }

hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManager.java

@@ -20,6 +20,7 @@
 import static java.net.HttpURLConnection.HTTP_CREATED;
 import static java.net.HttpURLConnection.HTTP_OK;
 import static org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_HTTP_ADDRESS_KEY;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY;
 import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_REQUEST_TIMEOUT;
 import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_REQUEST_TIMEOUT_DEFAULT;
 import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_CONNECTION_TIMEOUT;
@@ -337,6 +338,10 @@ public void testOmDBSyncWithSeqNumberMismatch() throws Exception {
     // to fallback on full snapshot. Full snapshot sync should normalize the
     // OM active DB lastSequence number and Recon's OM snapshot DB lastSequence
     // number.
+    // Force the full-snapshot fallback to transfer the active DB in small,
+    // resumable batches so the chunked path is exercised end to end.
+    cluster.getOzoneManager().getConfiguration()
+        .setLong(OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY, 1024);
     impl.syncDataFromOM();
     assertTrue(logs.getOutput()
         .contains("Requested sequence not yet written in the db"));
@@ -347,6 +352,10 @@ public void testOmDBSyncWithSeqNumberMismatch() throws Exception {
             .getLatestSequenceNumber();
     assertEquals(0, metrics.getSequenceNumberLag());
     assertEquals(omLatestSeqNumber, reconLatestSeqNumber);
+    // The chunked full snapshot assembled a complete DB (seq numbers match
+    // above) and recorded at least one transfer batch.
+    assertTrue(impl.getReconSyncMetrics().getLastSnapshotBatchCount() >= 1,
+        "chunked full snapshot should record at least one batch");
     reconLatestSeqNumber =
         ((RDBStore) reconMetadataManagerInstance.getStore()).getDb()
             .getLatestSequenceNumber();

hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestReconWithOzoneManagerHA.java

@@ -18,7 +18,6 @@
 package org.apache.hadoop.ozone.recon;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
 import static org.apache.hadoop.ozone.recon.ReconOmMetaManagerTestUtils.waitForEventBufferEmpty;
 import static org.apache.hadoop.ozone.recon.ReconOmMetaManagerTestUtils.waitUntilReconKeyCounts;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -120,14 +119,6 @@ public void testReconGetsSnapshotFromLeader() throws Exception {
     OzoneManagerServiceProviderImpl impl = (OzoneManagerServiceProviderImpl)
         recon.getReconServer().getOzoneManagerServiceProvider();
 
-    String hostname =
-        ozoneManager.get().getHttpServer().getHttpAddress().getHostName();
-    String expectedUrl = "http://" +
-        (hostname.equals("0.0.0.0") ? "localhost" : hostname) + ":" +
-        ozoneManager.get().getHttpServer().getHttpAddress().getPort() +
-        OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
-    String snapshotUrl = impl.getOzoneManagerSnapshotUrl();
-    assertEquals(expectedUrl, snapshotUrl);
     // Write some data
     String keyPrefix = "ratis";
     OzoneOutputStream key = objectStore.getVolume(VOL_NAME)

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.om;
 
 import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE;
+import static org.apache.hadoop.hdds.utils.HddsServerUtil.OZONE_RATIS_SNAPSHOT_COMPLETE_FLAG_NAME;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_ENABLED;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ADMINISTRATORS_WILDCARD;
@@ -26,6 +27,7 @@
 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_CHECKPOINT_DIR;
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_CHUNK_ACTIVE_DB;
 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.om.OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY;
@@ -287,6 +289,61 @@ public void testTarballBatching(boolean includeSnapshots) throws Exception {
     }
   }
 
+  @Test
+  public void testChunkActiveDbProducesIncompleteBatch() throws Exception {
+    String volumeName = "vol" + RandomStringUtils.secure().nextNumeric(5);
+    String bucketName = "buck" + RandomStringUtils.secure().nextNumeric(5);
+    AtomicReference<DBCheckpoint> realCheckpoint = new AtomicReference<>();
+    // No snapshot data (Recon's mode), but ask to chunk the active DB.
+    setupClusterAndMocks(volumeName, bucketName, realCheckpoint, false);
+    when(requestMock.getParameter(OZONE_DB_CHECKPOINT_CHUNK_ACTIVE_DB))
+        .thenReturn("true");
+    // Tiny cap forces the active DB (many SST files) across multiple batches.
+    om.getConfiguration().setLong(OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY, 1);
+
+    omDbCheckpointServletMock.doGet(requestMock, responseMock);
+
+    String testDirName = folder.resolve("testDir").toString();
+    String newDbDirName = testDirName + OM_KEY_PREFIX + OM_DB_NAME;
+    File newDbDir = new File(newDbDirName);
+    assertTrue(newDbDir.mkdirs());
+    FileUtil.unTar(tempFile, newDbDir);
+
+    File[] batchFiles = newDbDir.listFiles();
+    assertNotNull(batchFiles);
+    // Progress guarantee: at least one file is sent even under a tiny cap.
+    assertTrue(batchFiles.length >= 1);
+    // The completion marker and hardLinkFile are sent only in the final batch,
+    // so an incomplete chunked batch must not carry them.
+    assertFalse(new File(newDbDir, OZONE_RATIS_SNAPSHOT_COMPLETE_FLAG_NAME).exists(),
+        "an incomplete chunked batch must not carry the completion marker");
+    assertFalse(
+        new File(newDbDir, InodeMetadataRocksDBCheckpoint.OM_HARDLINK_FILE).exists(),
+        "an incomplete chunked batch must not carry the hardLinkFile");
+  }
+
+  @Test
+  public void testCollectFilesFromDirAlwaysSendsAtLeastOneOversizedFile()
+      throws Exception {
+    OMDBCheckpointServletInodeBasedXfer servlet =
+        new OMDBCheckpointServletInodeBasedXfer();
+    Path dbDir = Files.createTempDirectory(folder, "dbdir-");
+    // Two SST files, each larger than the per-batch budget below.
+    Files.write(dbDir.resolve("000001.sst"), new byte[8192]);
+    Files.write(dbDir.resolve("000002.sst"), new byte[8192]);
+    OMDBArchiver archiver = new OMDBArchiver();
+    archiver.setTmpDir(folder);
+
+    boolean shouldContinue = servlet.collectFilesFromDir(new HashSet<>(), dbDir,
+        new AtomicLong(4096), true, archiver);
+
+    // The first (oversized) file is still sent so the transfer makes progress,
+    // then the batch stops before the second file.
+    assertFalse(shouldContinue, "batch should stop once the budget is exceeded");
+    assertEquals(1, archiver.getFilesToWriteIntoTarball().size(),
+        "an oversized file must still be sent so the transfer never stalls");
+  }
+
   @Test
   public void testWriteDBToArchiveClosesFilesListStream() throws Exception {
     OMDBCheckpointServletInodeBasedXfer servlet = new OMDBCheckpointServletInodeBasedXfer();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMDBCheckpointServletInodeBasedXfer.java

@@ -24,6 +24,7 @@
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY;
 import static org.apache.hadoop.ozone.om.OmSnapshotManager.getSnapshotPath;
+import static org.apache.hadoop.ozone.om.snapshot.OMDBCheckpointUtils.chunkActiveDb;
 import static org.apache.hadoop.ozone.om.snapshot.OMDBCheckpointUtils.includeSnapshotData;
 import static org.apache.hadoop.ozone.om.snapshot.OMDBCheckpointUtils.logEstimatedTarballSize;
 import static org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils.DATA_PREFIX;
@@ -243,13 +244,19 @@ public void collectDbDataToTransfer(HttpServletRequest request,
     OMMetadataManager omMetadataManager = om.getMetadataManager();
     OmSnapshotLocalDataManager snapshotLocalDataManager = om.getOmSnapshotManager().getSnapshotLocalDataManager();
     boolean includeSnapshotData = includeSnapshotData(request);
+    // When set, the active om.db is transferred in size-bounded, resumable
+    // batches instead of a single batch. Recon opts in; OM followers do not.
+    boolean chunkActiveDb = chunkActiveDb(request);
     AtomicLong maxTotalSstSize = new AtomicLong(getConf().getLong(OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY,
         OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_DEFAULT));
 
     Collection<Path> snapshotPaths = Collections.emptySet();
 
     if (!includeSnapshotData) {
-      maxTotalSstSize.set(Long.MAX_VALUE);
+      // Keep the active DB in a single batch unless the caller asked to chunk it.
+      if (!chunkActiveDb) {
+        maxTotalSstSize.set(Long.MAX_VALUE);
+      }
     } else {
       snapshotPaths = getSnapshotDirsFromDB(omMetadataManager, omMetadataManager, snapshotLocalDataManager).values();
     }
@@ -299,16 +306,28 @@ public void collectDbDataToTransfer(HttpServletRequest request,
          * In this case, the try-with-resources block does not call close() on any resource,
          * which is intentional because snapshot consistency is not required.
          */
+        boolean activeDbComplete = true;
         try (UncheckedAutoCloseableSupplier<OMLockDetails> snapshotDBLock =
                  includeSnapshotData ? snapshotCache.lock() : null;
              HierarchicalResourceLock snapshotLocalDataLock = includeSnapshotData ? localDataManager.lock() : null) {
           // get the list of sst files of the checkpoint.
           checkpoint = createAndPrepareCheckpoint(true);
-          // unlimited files as we want the Active DB contents to be transferred in a single batch
-          maxTotalSstSize.set(Long.MAX_VALUE);
           Path checkpointDir = checkpoint.getCheckpointLocation();
-          collectFilesFromDir(sstFilesToExclude, checkpointDir, maxTotalSstSize, false, omdbArchiver);
-          if (includeSnapshotData) {
+          if (chunkActiveDb) {
+            // Send active-DB SST files under the size cap; if some are left for a
+            // later batch, stop before the completion marker is written.
+            activeDbComplete = collectFilesFromDir(sstFilesToExclude, checkpointDir,
+                maxTotalSstSize, true, omdbArchiver);
+            LOG.info("Chunked active DB transfer: sent an SST batch, activeDbComplete={}.",
+                activeDbComplete);
+          }
+          if (activeDbComplete) {
+            // Active DB in one batch (default), or the final chunked batch: send
+            // the remaining files unbounded (any SSTs sent above are excluded).
+            maxTotalSstSize.set(Long.MAX_VALUE);
+            collectFilesFromDir(sstFilesToExclude, checkpointDir, maxTotalSstSize, false, omdbArchiver);
+          }
+          if (activeDbComplete && includeSnapshotData) {
             List<Path> sstBackupFiles = extractSSTFilesFromCompactionLog(checkpoint);
             // get the list of snapshots from the checkpoint
             Map<UUID, Path> snapshotInCheckpoint;
@@ -330,7 +349,11 @@ public void collectDbDataToTransfer(HttpServletRequest request,
                 maxTotalSstSize, omdbArchiver);
           }
         }
-        omdbArchiver.setCompleted(true);
+        // A chunked transfer only completes once the whole active DB has been
+        // sent; a partial batch omits the completion marker so the client loops.
+        if (activeDbComplete) {
+          omdbArchiver.setCompleted(true);
+        }
       }
 
     } catch (IOException ioe) {
@@ -549,7 +572,10 @@ boolean collectFilesFromDir(Set<String> sstFilesToExclude, Stream<Path> files,
         if (!sstFilesToExclude.contains(fileId)) {
           try {
             long fileSize = Files.size(dbFile);
-            if (maxTotalSstSize.get() - fileSize <= 0) {
+            // Always send at least one file per batch: if nothing has been sent
+            // yet, send this file even when it exceeds the budget, so a single
+            // file larger than the cap can never stall the transfer.
+            if (filesWritten > 0 && maxTotalSstSize.get() - fileSize <= 0) {
               return false;
             }
             bytesRecorded += omdbArchiver.recordFileEntry(dbFile.toFile(), fileId);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/OMDBCheckpointUtils.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.om.snapshot;
 
 import static org.apache.commons.io.filefilter.TrueFileFilter.TRUE;
+import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_CHUNK_ACTIVE_DB;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA;
 import static org.apache.hadoop.ozone.OzoneConsts.ROCKSDB_SST_SUFFIX;
 
@@ -55,6 +56,17 @@ public static boolean includeSnapshotData(HttpServletRequest request) {
     return Boolean.parseBoolean(includeParam);
   }
 
+  /**
+   * When true, the active om.db is transferred in multiple size-bounded,
+   * resumable batches instead of a single batch. Defaults to false so OM
+   * follower bootstrap keeps its single-batch active-DB behavior.
+   */
+  public static boolean chunkActiveDb(HttpServletRequest request) {
+    String chunkParam =
+        request.getParameter(OZONE_DB_CHECKPOINT_CHUNK_ACTIVE_DB);
+    return Boolean.parseBoolean(chunkParam);
+  }
+
   public static void logEstimatedTarballSize(Path dbLocation, Collection<Path> snapshotPaths) {
     try {
       Counters.PathCounters counters = Counters.longPathCounters();

hadoop-ozone/recon/pom.xml

@@ -138,6 +138,10 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-hdfs-client</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.ozone</groupId>
       <artifactId>hdds-cli-common</artifactId>

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/metrics/ReconSyncMetrics.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.metrics2.annotation.Metrics;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.metrics2.lib.MutableCounterLong;
+import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
 import org.apache.hadoop.metrics2.lib.MutableRate;
 import org.apache.hadoop.ozone.OzoneConsts;
 
@@ -73,6 +74,9 @@ public final class ReconSyncMetrics {
   @Metric(about = "Count of failed snapshot downloads")
   private MutableCounterLong snapshotDownloadFailures;
 
+  @Metric(about = "Number of batches in the most recent full DB snapshot download")
+  private MutableGaugeLong lastSnapshotBatchCount;
+
   private ReconSyncMetrics() {
   }
 
@@ -135,6 +139,10 @@ public void incrSnapshotDownloadFailures() {
     this.snapshotDownloadFailures.incr();
   }
 
+  public void setLastSnapshotBatchCount(long batches) {
+    this.lastSnapshotBatchCount.set(batches);
+  }
+
   // Getters for testing
   public long getDeltaFetchSuccess() {
     return deltaFetchSuccess.value();
@@ -176,6 +184,10 @@ public long getSnapshotDownloadFailures() {
     return snapshotDownloadFailures.value();
   }
 
+  public long getLastSnapshotBatchCount() {
+    return lastSnapshotBatchCount.value();
+  }
+
   public MutableRate getFullDBRequestLatency() {
     return fullDBRequestLatency;
   }