HDDS-13009. Background snapshot defrag service

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

@@ -145,6 +145,12 @@ <KEY, VALUE> TypedTable<KEY, VALUE> getTable(
    */
   Map<Integer, String> getTableNames();
 
+  /**
+   * Drop the specific table.
+   * @param tableName - Name of the table to truncate.
+   */
+  void dropTable(String tableName) throws RocksDatabaseException;
+
   /**
    * Get data written to DB since a specific sequence number.
    */
@@ -162,4 +168,6 @@ DBUpdatesWrapper getUpdatesSince(long sequenceNumber, long limitCount)
    * @return true if the DB is closed.
    */
   boolean isClosed();
+
+  String getSnapshotsParentDir();
 }

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

@@ -189,6 +189,7 @@ public String getSnapshotMetadataDir() {
     return dbLocation.getParent() + OM_KEY_PREFIX + OM_SNAPSHOT_DIFF_DIR;
   }
 
+  @Override
   public String getSnapshotsParentDir() {
     return snapshotsParentDir;
   }
@@ -332,6 +333,29 @@ public Map<Integer, String> getTableNames() {
     return db.getColumnFamilyNames();
   }
 
+  /**
+  /**
+   * Drops a table from the database by removing its associated column family.
+   * <p>
+   * <b>Warning:</b> This operation should be used with extreme caution. If the table needs to be used again,
+   * it is recommended to reinitialize the entire DB store, as the column family will be permanently
+   * removed from the database. This method is suitable for truncating a RocksDB column family in a single operation.
+   *
+   * @param tableName the name of the table to be dropped
+   * @throws RocksDatabaseException if an error occurs while attempting to drop the table
+   */
+  @Override
+  public void dropTable(String tableName) throws RocksDatabaseException {
+    ColumnFamily columnFamily = db.getColumnFamily(tableName);
+    if (columnFamily != null) {
+      try {
+        db.getManagedRocksDb().get().dropColumnFamily(columnFamily.getHandle());
+      } catch (RocksDBException e) {
+        throw new RocksDatabaseException("Failed to drop " + tableName, e);
+      }  
+    }
+  }
+
   public Collection<ColumnFamily> getColumnFamilies() {
     return db.getExtraColumnFamilies();
   }

hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java

@@ -21,6 +21,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import java.io.IOException;
+import java.nio.file.Path;
 import java.time.Duration;
 import java.util.Iterator;
 import java.util.List;
@@ -85,6 +86,13 @@ public interface OMMetadataManager extends DBStoreHAManager, AutoCloseable {
   @VisibleForTesting
   DBStore getStore();
 
+  /**
+   * Retrieves the parent directory of all the snapshots in the system.
+   *
+   * @return a Path object representing the parent directory of the snapshot.
+   */
+  Path getSnapshotParentDir();
+
   /**
    * Returns the OzoneManagerLock used on Metadata DB.
    *

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

@@ -421,10 +421,14 @@ public void startSnapshotDefragService(OzoneConfiguration conf) {
           OZONE_SNAPSHOT_DEFRAG_SERVICE_TIMEOUT_DEFAULT,
           TimeUnit.MILLISECONDS);
 
-      snapshotDefragService =
-          new SnapshotDefragService(serviceInterval, TimeUnit.MILLISECONDS,
-              serviceTimeout, ozoneManager, conf);
-      snapshotDefragService.start();
+      try {
+        snapshotDefragService =
+            new SnapshotDefragService(serviceInterval, TimeUnit.MILLISECONDS,
+                serviceTimeout, ozoneManager, conf);
+        snapshotDefragService.start();
+      } catch (IOException e) {
+        LOG.error("Error starting Snapshot Defrag Service", e);
+      }
     } else {
       LOG.info("SnapshotDefragService is disabled. Snapshot defragmentation will not run periodically.");
     }

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

@@ -234,6 +234,11 @@ protected OmMetadataManagerImpl() {
 
   public static OmMetadataManagerImpl createCheckpointMetadataManager(
       OzoneConfiguration conf, DBCheckpoint checkpoint) throws IOException {
+    return createCheckpointMetadataManager(conf, checkpoint, true);
+  }
+
+  public static OmMetadataManagerImpl createCheckpointMetadataManager(
+      OzoneConfiguration conf, DBCheckpoint checkpoint, boolean readOnly) throws IOException {
     Path path = checkpoint.getCheckpointLocation();
     Path parent = path.getParent();
     if (parent == null) {
@@ -246,7 +251,11 @@ public static OmMetadataManagerImpl createCheckpointMetadataManager(
       throw new IllegalStateException("DB checkpoint dir name should not "
           + "have been null. Checkpoint path is " + path);
     }
-    return new OmMetadataManagerImpl(conf, dir, name.toString());
+    return new OmMetadataManagerImpl(conf, dir, name.toString(), readOnly);
+  }
+
+  protected OmMetadataManagerImpl(OzoneConfiguration conf, File dir, String name) throws IOException {
+    this(conf, dir, name, true);
   }
 
   /**
@@ -257,15 +266,15 @@ public static OmMetadataManagerImpl createCheckpointMetadataManager(
    * @param name - Checkpoint directory name.
    * @throws IOException
    */
-  protected OmMetadataManagerImpl(OzoneConfiguration conf, File dir, String name)
+  protected OmMetadataManagerImpl(OzoneConfiguration conf, File dir, String name, boolean readOnly)
       throws IOException {
     lock = new OmReadOnlyLock();
     hierarchicalLockManager = new ReadOnlyHierarchicalResourceLockManager();
     omEpoch = 0;
     int maxOpenFiles = conf.getInt(OZONE_OM_SNAPSHOT_DB_MAX_OPEN_FILES, OZONE_OM_SNAPSHOT_DB_MAX_OPEN_FILES_DEFAULT);
 
     this.store = newDBStoreBuilder(conf, name, dir)
-        .setOpenReadOnly(true)
+        .setOpenReadOnly(readOnly)
         .disableDefaultCFAutoCompaction(true)
         .setMaxNumberOfOpenFiles(maxOpenFiles)
         .setEnableCompactionDag(false)
@@ -519,6 +528,11 @@ public DBStore getStore() {
     return store;
   }
 
+  @Override
+  public Path getSnapshotParentDir() {
+    return Paths.get(store.getSnapshotsParentDir());
+  }
+
   /**
    * Given a volume return the corresponding DB key.
    *

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

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.om;
 
+import static org.apache.commons.io.file.PathUtils.deleteDirectory;
 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.hadoop.hdds.StringUtils.getLexicographicallyHigherString;
 import static org.apache.hadoop.hdds.utils.db.DBStoreBuilder.DEFAULT_COLUMN_FAMILY_NAME;
@@ -97,6 +98,7 @@
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.SnapshotDiffJob;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
+import org.apache.hadoop.ozone.om.lock.OMLockDetails;
 import org.apache.hadoop.ozone.om.service.SnapshotDiffCleanupService;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager;
 import org.apache.hadoop.ozone.om.snapshot.SnapshotCache;
@@ -512,6 +514,64 @@ public void invalidateCacheEntry(UUID key) {
     }
   }
 
+  /**
+   * Deletes the snapshot checkpoint directories for a given snapshot ID up to
+   * the specified maximum version. This method ensures that all directories
+   * containing checkpoint data for the specified snapshot ID up to the max
+   * version are removed in a controlled manner.
+   *
+   * @param snapshotId The unique identifier of the snapshot whose checkpoint
+   *                   directories are to be deleted.
+   * @param maxVersion The maximum version of checkpoint directories to delete.
+   *                   If a value less than 0 is provided, it defaults to the
+   *                   current maximum version of the snapshot.
+   * @throws IOException If there is a failure acquiring the snapshot database
+   *                     lock or while deleting directories.
+   * @throws IllegalArgumentException If the specified maxVersion is greater
+   *                                  than the current maximum version of the
+   *                                  snapshot.
+   */
+  public void deleteSnapshotCheckpointDirectories(UUID snapshotId, int maxVersion) throws IOException {
+    // Acquire Snapshot DBHandle lock before removing the older version to ensure all readers are done with the
+    // snapshot db use.
+    try (UncheckedAutoCloseableSupplier<OMLockDetails> lock = getSnapshotCache().lock(snapshotId)) {
+      if (!lock.get().isLockAcquired()) {
+        throw new IOException("Failed to acquire dbHandlelock on snapshot: " + snapshotId);
+      }
+      try (OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataMetaProvider snapMetaProvider =
+               snapshotLocalDataManager.getOmSnapshotLocalDataMeta(snapshotId)) {
+        if (maxVersion < 0) {
+          maxVersion = snapMetaProvider.getMeta().getVersion();
+        }
+        if (maxVersion > snapMetaProvider.getMeta().getVersion()) {
+          throw new IllegalArgumentException(
+              String.format("Max Version to be deleted can never be greater than the existing " +
+                  "version of the snapshot. Argument passed : %d and snapshotMaxVersion : %d", maxVersion,
+                  snapMetaProvider.getMeta().getVersion()));
+        }
+        // Binary search the smallest existing version and delete the older versions starting from the smallest version.
+        // This is to ensure efficient crash recovery.
+        int smallestExistingVersion = 0;
+        int largestExistingVersion = maxVersion;
+        while (smallestExistingVersion <= largestExistingVersion) {
+          int midVersion = smallestExistingVersion + (largestExistingVersion - smallestExistingVersion) / 2;
+          Path path = OmSnapshotManager.getSnapshotPath(ozoneManager.getMetadataManager(), snapshotId, midVersion);
+          if (path.toFile().exists()) {
+            largestExistingVersion = midVersion - 1;
+          } else {
+            smallestExistingVersion = midVersion + 1;
+          }
+        }
+        // Delete the older version directories. Always starting deletes from smallest version to largest version to
+        // ensure binary search works correctly on a later basis.
+        for (int version = smallestExistingVersion; version <= maxVersion; version++) {
+          Path path = OmSnapshotManager.getSnapshotPath(ozoneManager.getMetadataManager(), snapshotId, version);
+          deleteDirectory(path);
+        }
+      }
+    }
+  }
+
   /**
    * Creates snapshot checkpoint that corresponds to snapshotInfo.
    * @param omMetadataManager the metadata manager

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

@@ -18,22 +18,18 @@
 package org.apache.hadoop.ozone.om.response.snapshot;
 
 import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.SNAPSHOT_INFO_TABLE;
-import static org.apache.hadoop.ozone.om.lock.FlatResource.SNAPSHOT_DB_LOCK;
 
 import com.google.common.annotations.VisibleForTesting;
 import jakarta.annotation.Nonnull;
 import java.io.IOException;
-import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
-import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.BatchOperation;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
 import org.apache.hadoop.ozone.om.OmSnapshotManager;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
-import org.apache.hadoop.ozone.om.lock.OMLockDetails;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
 import org.apache.hadoop.ozone.om.snapshot.OmSnapshotLocalDataManager;
@@ -95,10 +91,9 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
       if (snapshotInfo == null) {
         continue;
       }
-
+      OmSnapshotManager omSnapshotManager = metadataManager.getOzoneManager().getOmSnapshotManager();
       // Remove and close snapshot's RocksDB instance from SnapshotCache.
-      ((OmMetadataManagerImpl) omMetadataManager).getOzoneManager().getOmSnapshotManager()
-          .invalidateCacheEntry(snapshotInfo.getSnapshotId());
+      omSnapshotManager.invalidateCacheEntry(snapshotInfo.getSnapshotId());
       // Remove the snapshot from snapshotId to snapshotTableKey map.
       ((OmMetadataManagerImpl) omMetadataManager).getSnapshotChainManager()
           .removeFromSnapshotIdToTable(snapshotInfo.getSnapshotId());
@@ -109,7 +104,8 @@ protected void addToDBBatch(OMMetadataManager omMetadataManager,
       // snapshot purged txn is flushed to rocksdb.
       updateLocalData(snapshotLocalDataManager, snapshotInfo);
       // Delete Snapshot checkpoint directory.
-      deleteCheckpointDirectory(snapshotLocalDataManager, omMetadataManager, snapshotInfo);
+
+      omSnapshotManager.deleteSnapshotCheckpointDirectories(snapshotInfo.getSnapshotId(), -1);
       // Delete snapshotInfo from the table.
       omMetadataManager.getSnapshotInfoTable().deleteWithBatch(batchOperation, dbKey);
     }
@@ -133,34 +129,6 @@ private void updateLocalData(OmSnapshotLocalDataManager localDataManager, Snapsh
     }
   }
 
-  /**
-   * Deletes the checkpoint directory for a snapshot.
-   */
-  private void deleteCheckpointDirectory(OmSnapshotLocalDataManager snapshotLocalDataManager,
-      OMMetadataManager omMetadataManager, SnapshotInfo snapshotInfo) throws IOException {
-    // Acquiring write lock to avoid race condition with sst filtering service which creates a sst filtered file
-    // inside the snapshot directory. Any operation apart which doesn't create/delete files under this snapshot
-    // directory can run in parallel along with this operation.
-    OMLockDetails omLockDetails = omMetadataManager.getLock()
-        .acquireWriteLock(SNAPSHOT_DB_LOCK, snapshotInfo.getSnapshotId().toString());
-    boolean acquiredSnapshotLock = omLockDetails.isLockAcquired();
-    if (acquiredSnapshotLock) {
-      try (OmSnapshotLocalDataManager.ReadableOmSnapshotLocalDataMetaProvider snapMetaProvider =
-               snapshotLocalDataManager.getOmSnapshotLocalDataMeta(snapshotInfo)) {
-        Path snapshotDirPath = OmSnapshotManager.getSnapshotPath(omMetadataManager, snapshotInfo,
-            snapMetaProvider.getMeta().getVersion());
-        try {
-          FileUtils.deleteDirectory(snapshotDirPath.toFile());
-        } catch (IOException ex) {
-          LOG.error("Failed to delete snapshot directory {} for snapshot {}",
-              snapshotDirPath, snapshotInfo.getTableKey(), ex);
-        } finally {
-          omMetadataManager.getLock().releaseWriteLock(SNAPSHOT_DB_LOCK, snapshotInfo.getSnapshotId().toString());
-        }
-      }
-    }
-  }
-
   @VisibleForTesting
   public Map<String, SnapshotInfo> getUpdatedSnapInfos() {
     return updatedSnapInfos;

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

@@ -291,7 +291,10 @@ public void release(UUID key) {
    */
   public UncheckedAutoCloseableSupplier<OMLockDetails> lock() {
     return lock(() -> lock.acquireResourceWriteLock(SNAPSHOT_DB_LOCK),
-        () -> lock.releaseResourceWriteLock(SNAPSHOT_DB_LOCK), () -> cleanup(true));
+        () -> lock.releaseResourceWriteLock(SNAPSHOT_DB_LOCK), () -> {
+        cleanup(true);
+        return dbMap.isEmpty();
+      });
   }
 
   /**
@@ -303,22 +306,24 @@ public UncheckedAutoCloseableSupplier<OMLockDetails> lock() {
   public UncheckedAutoCloseableSupplier<OMLockDetails> lock(UUID snapshotId) {
     return lock(() -> lock.acquireWriteLock(SNAPSHOT_DB_LOCK, snapshotId.toString()),
         () -> lock.releaseWriteLock(SNAPSHOT_DB_LOCK, snapshotId.toString()),
-        () -> cleanup(snapshotId));
+        () -> {
+        cleanup(snapshotId, false);
+        return !dbMap.containsKey(snapshotId);
+      });
   }
 
   private OMLockDetails getEmptyOmLockDetails(OMLockDetails lockDetails) {
     return lockDetails.isLockAcquired() ? EMPTY_DETAILS_LOCK_ACQUIRED : EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
   }
 
   private UncheckedAutoCloseableSupplier<OMLockDetails> lock(Supplier<OMLockDetails> lockFunction,
-      Supplier<OMLockDetails> unlockFunction, Supplier<Void> cleanupFunction) {
+      Supplier<OMLockDetails> unlockFunction, Supplier<Boolean> cleanupFunction) {
     Supplier<OMLockDetails> emptyLockFunction = () -> getEmptyOmLockDetails(lockFunction.get());
     Supplier<OMLockDetails> emptyUnlockFunction = () -> getEmptyOmLockDetails(unlockFunction.get());
 
     AtomicReference<OMLockDetails> lockDetails = new AtomicReference<>(emptyLockFunction.get());
     if (lockDetails.get().isLockAcquired()) {
-      cleanupFunction.get();
-      if (!dbMap.isEmpty()) {
+      if (!cleanupFunction.get()) {
         lockDetails.set(emptyUnlockFunction.get());
       }
     }
@@ -349,14 +354,19 @@ public OMLockDetails get() {
   private synchronized Void cleanup(boolean force) {
     if (force || dbMap.size() > cacheSizeLimit) {
       for (UUID evictionKey : pendingEvictionQueue) {
-        cleanup(evictionKey);
+        cleanup(evictionKey, true);
       }
     }
     return null;
   }
 
-  private synchronized Void cleanup(UUID evictionKey) {
+  private synchronized Void cleanup(UUID evictionKey, boolean expectKeyToBePresent) {
     ReferenceCounted<OmSnapshot> snapshot = dbMap.get(evictionKey);
+
+    if (!expectKeyToBePresent && snapshot == null) {
+      return null;
+    }
+
     if (snapshot != null && snapshot.getTotalRefCount() == 0) {
       try {
         compactSnapshotDB(snapshot.get());