HDDS-13967. OM bootstrap lock should acquire a lock on SnapshotLocalDataManager before copying the local data files (#9335)

Author: swamirishi

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

@@ -35,11 +35,13 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 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;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.eq;
@@ -109,6 +111,7 @@
 import org.apache.hadoop.ozone.om.lock.IOzoneManagerLock;
 import org.apache.hadoop.ozone.om.lock.OMLockDetails;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils;
+import org.apache.hadoop.ozone.om.snapshot.SnapshotCache;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.ozone.test.GenericTestUtils;
@@ -169,7 +172,8 @@ private void setupCluster() throws Exception {
     conf.set(OZONE_ADMINISTRATORS, OZONE_ADMINISTRATORS_WILDCARD);
     cluster.waitForClusterToBeReady();
     client = cluster.newClient();
-    om = cluster.getOzoneManager();
+    OzoneManager normalOm = cluster.getOzoneManager();
+    om = spy(normalOm);
   }
 
   private void setupMocks() throws Exception {
@@ -244,7 +248,7 @@ public void write(int b) throws IOException {
     doCallRealMethod().when(omDbCheckpointServletMock).getCompactionLogDir();
     doCallRealMethod().when(omDbCheckpointServletMock).getSstBackupDir();
     doCallRealMethod().when(omDbCheckpointServletMock)
-        .transferSnapshotData(anySet(), any(), anySet(), any(), any(), anyMap());
+        .transferSnapshotData(anySet(), any(), anyCollection(), anyCollection(), any(), any(), anyMap());
     doCallRealMethod().when(omDbCheckpointServletMock).createAndPrepareCheckpoint(anyBoolean());
     doCallRealMethod().when(omDbCheckpointServletMock).getSnapshotDirsFromDB(any(), any(), any());
   }
@@ -767,27 +771,33 @@ public void testCheckpointIncludesSnapshotsFromFrozenState() throws Exception {
         .filter(snap -> snap.getName().equals("snapshot1"))
         .findFirst()
         .orElseThrow(() -> new RuntimeException("snapshot1 not found"));
-
     // Setup servlet mocks for checkpoint processing
     setupMocks();
     when(requestMock.getParameter(OZONE_DB_CHECKPOINT_INCLUDE_SNAPSHOT_DATA)).thenReturn("true");
 
     // Create a checkpoint that captures current state (S1)
-    DBStore dbStore = om.getMetadataManager().getStore();
-    DBStore spyDbStore = spy(dbStore);
-    AtomicReference<DBCheckpoint> capturedCheckpoint = new AtomicReference<>();
+    DBStore spyDbStore = spy(om.getMetadataManager().getStore());
 
-    when(spyDbStore.getCheckpoint(true)).thenAnswer(invocation -> {
-      // Purge snapshot2 before checkpoint
+    AtomicReference<DBCheckpoint> capturedCheckpoint = new AtomicReference<>();
+    SnapshotCache spySnapshotCache = spy(om.getOmSnapshotManager().getSnapshotCache());
+    OmSnapshotManager spySnapshotManager = spy(om.getOmSnapshotManager());
+    when(om.getOmSnapshotManager()).thenReturn(spySnapshotManager);
+    when(spySnapshotManager.getSnapshotCache()).thenReturn(spySnapshotCache);
+    // Mock the snapshot cache to create a snapshot2 just after taking a snapshot cache lock.
+    doAnswer(invocationOnMock -> {
+      Object ret = invocationOnMock.callRealMethod();
       // create snapshot 3 before checkpoint
       client.getObjectStore().createSnapshot(volumeName, bucketName, "snapshot2");
       // Also wait for double buffer to flush to ensure all transactions are committed
       om.awaitDoubleBufferFlush();
-      DBCheckpoint checkpoint = spy(dbStore.getCheckpoint(true));
+      return ret;
+    }).when(spySnapshotCache).lock();
+    doAnswer(invocation -> {
+      DBCheckpoint checkpoint = (DBCheckpoint) spy(invocation.callRealMethod());
       doNothing().when(checkpoint).cleanupCheckpoint(); // Don't cleanup for verification
       capturedCheckpoint.set(checkpoint);
       return checkpoint;
-    });
+    }).when(spyDbStore).getCheckpoint(eq(true));
 
     // Initialize servlet
     doCallRealMethod().when(omDbCheckpointServletMock).initialize(any(), any(),

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

@@ -53,7 +53,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
@@ -70,6 +72,7 @@
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.lock.BootstrapStateHandler;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
+import org.apache.hadoop.ozone.om.lock.HierarchicalResourceLockManager.HierarchicalResourceLock;
 import org.apache.hadoop.ozone.om.lock.OMLockDetails;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotUtils;
@@ -208,7 +211,6 @@ Path getCompactionLogDir() {
    * @param tmpdir            Temporary directory for staging files during archiving.
    * @throws IOException if an I/O error occurs during processing or streaming.
    */
-
   public void writeDbDataToStream(HttpServletRequest request, OutputStream destination,
       Set<String> sstFilesToExclude, Path tmpdir) throws IOException {
     DBCheckpoint checkpoint = null;
@@ -219,12 +221,12 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
     AtomicLong maxTotalSstSize = new AtomicLong(getConf().getLong(OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_KEY,
         OZONE_OM_RATIS_SNAPSHOT_MAX_TOTAL_SST_SIZE_DEFAULT));
 
-    Set<Path> snapshotPaths = Collections.emptySet();
+    Collection<Path> snapshotPaths = Collections.emptySet();
 
     if (!includeSnapshotData) {
       maxTotalSstSize.set(Long.MAX_VALUE);
     } else {
-      snapshotPaths = getSnapshotDirsFromDB(omMetadataManager, omMetadataManager, snapshotLocalDataManager);
+      snapshotPaths = getSnapshotDirsFromDB(omMetadataManager, omMetadataManager, snapshotLocalDataManager).values();
     }
 
     if (sstFilesToExclude.isEmpty()) {
@@ -260,6 +262,7 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
         // this is the last step where we transfer the active om.db contents
         Map<String, String> hardLinkFileMap = new HashMap<>();
         SnapshotCache snapshotCache = om.getOmSnapshotManager().getSnapshotCache();
+        OmSnapshotLocalDataManager localDataManager = om.getOmSnapshotManager().getSnapshotLocalDataManager();
         /*
          * Acquire snapshot cache lock when includeSnapshotData is true to prevent race conditions
          * between checkpoint operations and snapshot purge operations. Without this lock, a purge
@@ -272,7 +275,9 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
          * In this case, the try-with-resources block does not call close() on any resource,
          * which is intentional because snapshot consistency is not required.
          */
-        try (UncheckedAutoCloseableSupplier<OMLockDetails> lock = includeSnapshotData ? snapshotCache.lock() : null) {
+        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
@@ -283,18 +288,21 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
           if (includeSnapshotData) {
             List<Path> sstBackupFiles = extractSSTFilesFromCompactionLog(checkpoint);
             // get the list of snapshots from the checkpoint
+            Map<UUID, Path> snapshotInCheckpoint;
             try (OmMetadataManagerImpl checkpointMetadataManager = OmMetadataManagerImpl
                     .createCheckpointMetadataManager(om.getConfiguration(), checkpoint)) {
-              snapshotPaths = getSnapshotDirsFromDB(omMetadataManager, checkpointMetadataManager,
+              snapshotInCheckpoint = getSnapshotDirsFromDB(omMetadataManager, checkpointMetadataManager,
                   snapshotLocalDataManager);
             }
             writeDBToArchive(sstFilesToExclude, getCompactionLogDir(), maxTotalSstSize, archiveOutputStream, tmpdir,
                 hardLinkFileMap, false);
             writeDBToArchive(sstFilesToExclude, sstBackupFiles.stream(), maxTotalSstSize, archiveOutputStream, tmpdir,
                 hardLinkFileMap, false);
+            Collection<Path> snapshotLocalPropertyFiles = getSnapshotLocalDataPaths(localDataManager,
+                snapshotInCheckpoint.keySet());
             // This is done to ensure all data to be copied correctly is flushed in the snapshot DB
-            transferSnapshotData(sstFilesToExclude, tmpdir, snapshotPaths, maxTotalSstSize, archiveOutputStream,
-                hardLinkFileMap);
+            transferSnapshotData(sstFilesToExclude, tmpdir, snapshotInCheckpoint.values(), snapshotLocalPropertyFiles,
+                maxTotalSstSize, archiveOutputStream, hardLinkFileMap);
           }
         }
         writeHardlinkFile(getConf(), hardLinkFileMap, archiveOutputStream);
@@ -309,6 +317,36 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
     }
   }
 
+  /**
+   * Retrieves the paths to the local property YAML files for the specified snapshot IDs.
+   * This method resolves the chain of previous snapshot references for each snapshot ID
+   * and gathers their corresponding local property YAML file paths.
+   *
+   * @param localDataManager The OmSnapshotLocalDataManager instance responsible for managing
+   *                         snapshot data and metadata.
+   * @param snapshotIds      A set of snapshot IDs for which the local property YAML file
+   *                         paths should be resolved.
+   * @return A collection of paths to the local property YAML files for the specified
+   *         snapshot IDs.
+   */
+  private Collection<Path> getSnapshotLocalDataPaths(OmSnapshotLocalDataManager localDataManager,
+      Set<UUID> snapshotIds) {
+    Set<UUID> snapshotLocalDataIds = new HashSet<>();
+    Map<UUID, OmSnapshotLocalDataManager.SnapshotVersionsMeta> versionNodeMap =
+        localDataManager.getVersionNodeMapUnmodifiable();
+    for (UUID snapshot : snapshotIds) {
+      UUID id = snapshot;
+      // Get the previous snapshot id for the current snapshot id until we reach null or the first snapshot id which
+      // is already in the snapshotLocalDataIds set.
+      while (id != null && !snapshotLocalDataIds.contains(id)) {
+        snapshotLocalDataIds.add(id);
+        id = versionNodeMap.get(id).getPreviousSnapshotId();
+      }
+    }
+    return snapshotLocalDataIds.stream().map(localDataManager::getSnapshotLocalPropertyYamlPath)
+        .map(Paths::get).collect(Collectors.toList());
+  }
+
   /**
    * Transfers the snapshot data from the specified snapshot directories into the archive output stream,
    * handling deduplication and managing resource locking.
@@ -321,19 +359,18 @@ public void writeDbDataToStream(HttpServletRequest request, OutputStream destina
    * @param hardLinkFileMap     Map of hardlink file paths to their unique identifiers for deduplication.
    * @throws IOException if an I/O error occurs during processing.
    */
-  void transferSnapshotData(Set<String> sstFilesToExclude, Path tmpdir, Set<Path> snapshotPaths,
-      AtomicLong maxTotalSstSize, ArchiveOutputStream<TarArchiveEntry> archiveOutputStream,
-      Map<String, String> hardLinkFileMap) throws IOException {
+  void transferSnapshotData(Set<String> sstFilesToExclude, Path tmpdir, Collection<Path> snapshotPaths,
+      Collection<Path> snapshotLocalPropertyFiles, AtomicLong maxTotalSstSize,
+      ArchiveOutputStream<TarArchiveEntry> archiveOutputStream, Map<String, String> hardLinkFileMap)
+      throws IOException {
     for (Path snapshotDir : snapshotPaths) {
       writeDBToArchive(sstFilesToExclude, snapshotDir, maxTotalSstSize, archiveOutputStream, tmpdir, hardLinkFileMap,
           false);
-      Path snapshotLocalPropertyYaml = Paths.get(
-          OmSnapshotLocalDataManager.getSnapshotLocalPropertyYamlPath(snapshotDir));
-      if (Files.exists(snapshotLocalPropertyYaml)) {
-        File yamlFile = snapshotLocalPropertyYaml.toFile();
-        hardLinkFileMap.put(yamlFile.getAbsolutePath(), yamlFile.getName());
-        linkAndIncludeFile(yamlFile, yamlFile.getName(), archiveOutputStream, tmpdir);
-      }
+    }
+    for (Path snapshotLocalPropertyYaml : snapshotLocalPropertyFiles) {
+      File yamlFile = snapshotLocalPropertyYaml.toFile();
+      hardLinkFileMap.put(yamlFile.getAbsolutePath(), yamlFile.getName());
+      linkAndIncludeFile(yamlFile, yamlFile.getName(), archiveOutputStream, tmpdir);
     }
   }
 
@@ -399,12 +436,12 @@ private OzoneConfiguration getConf() {
    * Collects paths to all snapshot databases from the OM DB.
    *
    * @param activeOMMetadataManager OMMetadataManager instance
-   * @return Set of paths to snapshot databases
+   * @return Map of paths to snapshot databases with snapshot IDs as keys.
    * @throws IOException if an I/O error occurs
    */
-  Set<Path> getSnapshotDirsFromDB(OMMetadataManager activeOMMetadataManager, OMMetadataManager omMetadataManager,
+  Map<UUID, Path> getSnapshotDirsFromDB(OMMetadataManager activeOMMetadataManager, OMMetadataManager omMetadataManager,
       OmSnapshotLocalDataManager localDataManager) throws IOException {
-    Set<Path> snapshotPaths = new HashSet<>();
+    Map<UUID, Path> snapshotPaths = new HashMap<>();
     try (TableIterator<String, ? extends Table.KeyValue<String, SnapshotInfo>> iter =
         omMetadataManager.getSnapshotInfoTable().iterator()) {
       while (iter.hasNext()) {
@@ -414,7 +451,7 @@ Set<Path> getSnapshotDirsFromDB(OMMetadataManager activeOMMetadataManager, OMMet
                  localDataManager.getOmSnapshotLocalDataMeta(snapshotInfo.getSnapshotId())) {
           Path snapshotDir = getSnapshotPath(activeOMMetadataManager, snapshotInfo.getSnapshotId(),
               snapLocalMeta.getMeta().getVersion());
-          snapshotPaths.add(snapshotDir);
+          snapshotPaths.put(snapshotInfo.getSnapshotId(), snapshotDir);
         }
       }
     }

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

@@ -23,7 +23,7 @@
 
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Set;
+import java.util.Collection;
 import javax.servlet.http.HttpServletRequest;
 import org.apache.commons.io.IOCase;
 import org.apache.commons.io.file.Counters;
@@ -55,7 +55,7 @@ public static boolean includeSnapshotData(HttpServletRequest request) {
     return Boolean.parseBoolean(includeParam);
   }
 
-  public static void logEstimatedTarballSize(Path dbLocation, Set<Path> snapshotPaths) {
+  public static void logEstimatedTarballSize(Path dbLocation, Collection<Path> snapshotPaths) {
     try {
       Counters.PathCounters counters = Counters.longPathCounters();
       CountingPathVisitor visitor = new CountingPathVisitor(

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

@@ -69,13 +69,11 @@
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.ozone.om.lock.HierarchicalResourceLockManager;
 import org.apache.hadoop.ozone.om.lock.HierarchicalResourceLockManager.HierarchicalResourceLock;
-import org.apache.hadoop.ozone.om.lock.OMLockDetails;
 import org.apache.hadoop.ozone.om.upgrade.OMLayoutVersionManager;
 import org.apache.hadoop.ozone.util.ObjectSerializer;
 import org.apache.hadoop.ozone.util.YamlSerializer;
 import org.apache.ratis.util.function.CheckedFunction;
 import org.apache.ratis.util.function.CheckedSupplier;
-import org.apache.ratis.util.function.UncheckedAutoCloseableSupplier;
 import org.rocksdb.LiveFileMetaData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -111,8 +109,6 @@ public class OmSnapshotLocalDataManager implements AutoCloseable {
   private final MutableGraph<LocalDataVersionNode> localDataGraph;
   private final Map<UUID, SnapshotVersionsMeta> versionNodeMap;
   private final OMMetadataManager omMetadataManager;
-  // Used for acquiring locks on the entire data structure.
-  private final ReadWriteLock fullLock;
   // Used for taking a lock on internal data structure Map and Graph to ensure thread safety;
   private final ReadWriteLock internalLock;
   // Locks should be always acquired by iterating through the snapshot chain to avoid deadlocks.
@@ -136,11 +132,14 @@ public void computeAndSetChecksum(Yaml yaml, OmSnapshotLocalData data) throws IO
       }
     };
     this.versionNodeMap = new ConcurrentHashMap<>();
-    this.fullLock = new ReentrantReadWriteLock();
     this.internalLock = new ReentrantReadWriteLock();
     init(configuration, snapshotChainManager, omLayoutVersionManager, defaultSnapProvider);
   }
 
+  public Map<UUID, SnapshotVersionsMeta> getVersionNodeMapUnmodifiable() {
+    return Collections.unmodifiableMap(versionNodeMap);
+  }
+
   @VisibleForTesting
   Map<UUID, SnapshotVersionsMeta> getVersionNodeMap() {
     return versionNodeMap;
@@ -224,8 +223,7 @@ public WritableOmSnapshotLocalDataProvider getWritableOmSnapshotLocalData(UUID s
     return new WritableOmSnapshotLocalDataProvider(snapshotId, previousSnapshotId);
   }
 
-  public WritableOmSnapshotLocalDataProvider getWritableOmSnapshotLocalData(UUID snapshotId)
-      throws IOException {
+  public WritableOmSnapshotLocalDataProvider getWritableOmSnapshotLocalData(UUID snapshotId) throws IOException {
     return new WritableOmSnapshotLocalDataProvider(snapshotId);
   }
 
@@ -448,19 +446,8 @@ void checkOrphanSnapshotVersions(OMMetadataManager metadataManager, SnapshotChai
    * @return an instance of {@code UncheckedAutoCloseableSupplier<OMLockDetails>} representing
    *         the acquired lock details, where the lock will automatically be released on close.
    */
-  public UncheckedAutoCloseableSupplier<OMLockDetails> lock() {
-    this.fullLock.writeLock().lock();
-    return new UncheckedAutoCloseableSupplier<OMLockDetails>() {
-      @Override
-      public OMLockDetails get() {
-        return OMLockDetails.EMPTY_DETAILS_LOCK_ACQUIRED;
-      }
-
-      @Override
-      public void close() {
-        fullLock.writeLock().unlock();
-      }
-    };
+  public HierarchicalResourceLock lock() throws IOException {
+    return locks.acquireResourceWriteLock(SNAPSHOT_LOCAL_DATA_LOCK);
   }
 
   private void validateVersionRemoval(UUID snapshotId, int version) throws IOException {
@@ -831,18 +818,15 @@ public final class WritableOmSnapshotLocalDataProvider extends ReadableOmSnapsho
 
     private WritableOmSnapshotLocalDataProvider(UUID snapshotId) throws IOException {
       super(snapshotId, false);
-      fullLock.readLock().lock();
     }
 
     private WritableOmSnapshotLocalDataProvider(UUID snapshotId, UUID snapshotIdToBeResolved) throws IOException {
       super(snapshotId, false, null, snapshotIdToBeResolved, true);
-      fullLock.readLock().lock();
     }
 
     private WritableOmSnapshotLocalDataProvider(UUID snapshotId,
         CheckedSupplier<Pair<OmSnapshotLocalData, File>, IOException> snapshotLocalDataSupplier) throws IOException {
       super(snapshotId, false, snapshotLocalDataSupplier, null, false);
-      fullLock.readLock().lock();
     }
 
     private SnapshotVersionsMeta validateModification(OmSnapshotLocalData snapshotLocalData)
@@ -1005,7 +989,6 @@ private boolean isDirty() {
     @Override
     public void close() throws IOException {
       super.close();
-      fullLock.readLock().unlock();
     }
   }