HDDS-13906. Reduce Bootstrap Write lock time on OM during bootstrapping execution.

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

@@ -37,7 +37,6 @@
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.anyCollection;
 import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anySet;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyBoolean;
@@ -237,19 +236,19 @@ public void write(int b) throws IOException {
     doCallRealMethod().when(omDbCheckpointServletMock)
         .writeDbDataToStream(any(), any(), any(), any(), any());
     doCallRealMethod().when(omDbCheckpointServletMock)
-        .writeDBToArchive(any(), any(), any(), any(), any(), any(), anyBoolean());
+        .collectFilesFromDir(any(), any(), any(), anyBoolean(), any());
 
     when(omDbCheckpointServletMock.getBootstrapStateLock())
         .thenReturn(lock);
     doCallRealMethod().when(omDbCheckpointServletMock).getCheckpoint(any(), anyBoolean());
     assertNull(doCallRealMethod().when(omDbCheckpointServletMock).getBootstrapTempData());
     doCallRealMethod().when(omDbCheckpointServletMock).
         processMetadataSnapshotRequest(any(), any(), anyBoolean(), anyBoolean());
-    doCallRealMethod().when(omDbCheckpointServletMock).writeDbDataToStream(any(), any(), any(), any());
+    doCallRealMethod().when(omDbCheckpointServletMock).writeDbDataToStream(any(), any(), any(), any(), any());
     doCallRealMethod().when(omDbCheckpointServletMock).getCompactionLogDir();
     doCallRealMethod().when(omDbCheckpointServletMock).getSstBackupDir();
     doCallRealMethod().when(omDbCheckpointServletMock)
-        .transferSnapshotData(anySet(), any(), anyCollection(), anyCollection(), any(), any(), anyMap());
+        .collectSnapshotData(anySet(),  anyCollection(), anyCollection(), any(), any());
     doCallRealMethod().when(omDbCheckpointServletMock).createAndPrepareCheckpoint(anyBoolean());
     doCallRealMethod().when(omDbCheckpointServletMock).getSnapshotDirsFromDB(any(), any(), any());
   }
@@ -424,9 +423,10 @@ public void testWriteDBToArchive(boolean expectOnlySstFiles) throws Exception {
     Files.write(nonSstFile, "log content".getBytes(StandardCharsets.UTF_8));
     Set<String> sstFilesToExclude = new HashSet<>();
     AtomicLong maxTotalSstSize = new AtomicLong(1000000); // Sufficient size
-    Map<String, String> hardLinkFileMap = new java.util.HashMap<>();
+    OMDBArchiver omdbArchiver = new OMDBArchiver();
     Path tmpDir = folder.resolve("tmp");
     Files.createDirectories(tmpDir);
+    omdbArchiver.setTmpDir(tmpDir);
     TarArchiveOutputStream mockArchiveOutputStream = mock(TarArchiveOutputStream.class);
     List<String> fileNames = new ArrayList<>();
     try (MockedStatic<Archiver> archiverMock = mockStatic(Archiver.class)) {
@@ -441,9 +441,8 @@ public void testWriteDBToArchive(boolean expectOnlySstFiles) throws Exception {
         aos.closeArchiveEntry();
         return 100L;
       });
-      boolean success = omDbCheckpointServletMock.writeDBToArchive(
-          sstFilesToExclude, dbDir, maxTotalSstSize, mockArchiveOutputStream,
-              tmpDir, hardLinkFileMap, expectOnlySstFiles);
+      boolean success = omDbCheckpointServletMock.collectFilesFromDir(
+          sstFilesToExclude, dbDir, maxTotalSstSize, expectOnlySstFiles, omdbArchiver);
       assertTrue(success);
       verify(mockArchiveOutputStream, times(fileNames.size())).putArchiveEntry(any());
       verify(mockArchiveOutputStream, times(fileNames.size())).closeArchiveEntry();

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

@@ -0,0 +1,130 @@
+/*
+ * 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.ozone.om;
+
+import static org.apache.hadoop.hdds.utils.Archiver.includeFile;
+import static org.apache.hadoop.hdds.utils.Archiver.tar;
+import static org.apache.hadoop.hdds.utils.HddsServerUtil.includeRatisSnapshotCompleteFlag;
+import static org.apache.hadoop.ozone.om.OMDBCheckpointServletInodeBasedXfer.writeHardlinkFile;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.compress.archivers.ArchiveOutputStream;
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.util.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class for handling operations relevant to archiving the OM DB tarball.
+ * Mainly maintains a map for recording the files collected from reading
+ * the checkpoint and snapshot DB's. It temporarily creates hardlinks and stores
+ * the link data in the map to release the bootstrap lock quickly
+ * and do the actual write at the end outside the lock.
+ */
+public class OMDBArchiver {
+
+  private Path tmpDir;
+  private Map<String, File> filesToWriteIntoTarball;
+  private Map<String, String> hardLinkFileMap;
+  private static final Logger LOG = LoggerFactory.getLogger(OMDBArchiver.class);
+
+  public OMDBArchiver() {
+    this.tmpDir = null;
+    this.filesToWriteIntoTarball = new HashMap<>();
+    hardLinkFileMap = null;
+  }
+
+  public void setTmpDir(Path tmpDir) {
+    this.tmpDir = tmpDir;
+  }
+
+  public Map<String, String> getHardLinkFileMap() {
+    return hardLinkFileMap;
+  }
+
+  public void setHardLinkFileMap(Map<String, String> hardLinkFileMap) {
+    this.hardLinkFileMap = hardLinkFileMap;
+  }
+
+  /**
+   * @param file the file to create a hardlink and record into the map
+   * @param entryName name of the entry corresponding to file
+   * @return the file size
+   * @throws IOException in case of hardlink failure
+   *
+   * Records the given file entry into the map after taking a hardlink.
+   */
+  public long recordFileEntry(File file, String entryName) throws IOException {
+    File link = tmpDir.resolve(entryName).toFile();
+    long bytes = 0;
+    try {
+      Files.createLink(link.toPath(), file.toPath());
+      filesToWriteIntoTarball.put(entryName, link);
+      bytes = file.length();
+    } catch (IOException ioe) {
+      LOG.error("Couldn't create hardlink for file {} while including it in tarball.",
+          file.getAbsolutePath(), ioe);
+      throw ioe;
+    }
+    return bytes;
+  }
+
+  /**
+   * @param conf the configuration object to obtain metadata paths
+   * @param outputStream the tarball archive output stream
+   * @throws IOException in case of write failure to the archive
+   *
+   * Writes all the files captured by the map into the archive and
+   * also includes the hardlinkFile and the completion marker file.
+   */
+  public void writeToArchive(OzoneConfiguration conf, OutputStream outputStream)
+      throws IOException {
+    long bytesWritten = 0;
+    long lastLoggedTime = Time.now();
+    long filesWritten = 0;
+    try (ArchiveOutputStream<TarArchiveEntry> archiveOutput = tar(outputStream)) {
+      for (Map.Entry<String, File> kv : filesToWriteIntoTarball.entrySet()) {
+        String entryName = kv.getKey();
+        File link = kv.getValue();
+        try {
+          bytesWritten += includeFile(link, entryName, archiveOutput);
+          if (Time.monotonicNow() - lastLoggedTime >= 30000) {
+            LOG.info("Transferred {} KB, #files {} to checkpoint tarball stream...",
+                bytesWritten / (1024), filesWritten);
+            lastLoggedTime = Time.monotonicNow();
+          }
+        } catch (IOException ioe) {
+          LOG.error("Couldn't create hardlink for file {} while including it in tarball.",
+              link.getAbsolutePath(), ioe);
+          throw ioe;
+        } finally {
+          Files.deleteIfExists(link.toPath());
+        }
+      }
+      writeHardlinkFile(conf, hardLinkFileMap, archiveOutput);
+      includeRatisSnapshotCompleteFlag(archiveOutput);
+    }
+  }
+}