HDDS-7974. [Snapshot] KeyDeletingService to be aware of Ozone snapshots

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1200,6 +1200,7 @@ message DeletedKeys {
 
 message PurgeKeysRequest {
     repeated DeletedKeys deletedKeys = 1;
+    optional string fromSnapshot = 2;
 }
 
 message PurgeKeysResponse {

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

@@ -544,4 +544,10 @@ String getMultipartKey(long volumeId, long bucketId,
    */
   long getBucketId(String volume, String bucket) throws IOException;
 
+  /**
+   * Returns List<{@link BlockGroup}> for a key in the deletedTable.
+   * @param deletedKey - key to be purged from the deletedTable
+   * @return {@link BlockGroup}
+   */
+  List<BlockGroup> getBlocksForKeyDelete(String deletedKey) throws IOException;
 }

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

@@ -282,7 +282,8 @@ public void start(OzoneConfiguration configuration) {
           TimeUnit.MILLISECONDS);
       try {
         snapshotDeletingService = new SnapshotDeletingService(
-            snapshotServiceInterval, snapshotServiceTimeout, ozoneManager);
+            snapshotServiceInterval, snapshotServiceTimeout,
+            ozoneManager, scmClient.getBlockClient());
         snapshotDeletingService.start();
       } catch (IOException e) {
         LOG.error("Error starting Snapshot Deleting Service", e);

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

@@ -344,7 +344,7 @@ protected OmMetadataManagerImpl() {
       File checkpoint = Paths.get(metaDir.toPath().toString(), dbName).toFile();
       RDBCheckpointManager.waitForCheckpointDirectoryExist(checkpoint);
     }
-    setStore(loadDB(conf, metaDir, dbName, true,
+    setStore(loadDB(conf, metaDir, dbName, false,
             java.util.Optional.of(Boolean.TRUE)));
     initializeOmTables(false);
   }
@@ -1787,4 +1787,30 @@ public long getBucketId(String volume, String bucket) throws IOException {
     }
     return omBucketInfo.getObjectID();
   }
+
+  @Override
+  public List<BlockGroup> getBlocksForKeyDelete(String deletedKey)
+      throws IOException {
+    RepeatedOmKeyInfo omKeyInfo = getDeletedTable().get(deletedKey);
+    if (omKeyInfo == null) {
+      return null;
+    }
+
+    List<BlockGroup> result = new ArrayList<>();
+    // Add all blocks from all versions of the key to the deletion list
+    for (OmKeyInfo info : omKeyInfo.cloneOmKeyInfoList()) {
+      for (OmKeyLocationInfoGroup keyLocations :
+          info.getKeyLocationVersions()) {
+        List<BlockID> item = keyLocations.getLocationList().stream()
+            .map(b -> new BlockID(b.getContainerID(), b.getLocalID()))
+            .collect(Collectors.toList());
+        BlockGroup keyBlocks = BlockGroup.newBuilder()
+            .setKeyName(deletedKey)
+            .addAllBlockIDs(item)
+            .build();
+        result.add(keyBlocks);
+      }
+    }
+    return result;
+  }
 }

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

@@ -68,6 +68,7 @@ public class OmSnapshot implements IOmMetadataReader, Closeable {
   private final String bucketName;
   private final String snapshotName;
   private final OMMetadataManager omMetadataManager;
+  private final KeyManager keyManager;
 
   public OmSnapshot(KeyManager keyManager,
                     PrefixManager prefixManager,
@@ -82,6 +83,7 @@ public OmSnapshot(KeyManager keyManager,
     this.bucketName = bucketName;
     this.volumeName = volumeName;
     this.omMetadataManager = keyManager.getMetadataManager();
+    this.keyManager = keyManager;
   }
 
 
@@ -262,4 +264,8 @@ public void close() throws IOException {
   public OMMetadataManager getMetadataManager() {
     return omMetadataManager;
   }
+
+  public KeyManager getKeyManager() {
+    return keyManager;
+  }
 }

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

@@ -119,7 +119,7 @@ public final class OmSnapshotManager implements AutoCloseable {
   // TODO: [SNAPSHOT] create config for max allowed page size.
   private final int maxPageSize = 1000;
 
-  OmSnapshotManager(OzoneManager ozoneManager) {
+  public OmSnapshotManager(OzoneManager ozoneManager) {
     this.options = new ManagedDBOptions();
     this.options.setCreateIfMissing(true);
     this.columnFamilyOptions = new ManagedColumnFamilyOptions();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyPurgeRequest.java

@@ -18,8 +18,13 @@
 
 package org.apache.hadoop.ozone.om.request.key;
 
+import java.io.IOException;
 import java.util.ArrayList;
+
+import org.apache.hadoop.ozone.om.OmSnapshot;
+import org.apache.hadoop.ozone.om.OmSnapshotManager;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
 import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
 import org.apache.hadoop.ozone.om.response.OMClientResponse;
@@ -33,6 +38,8 @@
 
 import java.util.List;
 
+import static org.apache.hadoop.ozone.om.OmSnapshotManager.getSnapshotPrefix;
+
 /**
  * Handles purging of keys from OM DB.
  */
@@ -51,6 +58,9 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     PurgeKeysRequest purgeKeysRequest = getOmRequest().getPurgeKeysRequest();
     List<DeletedKeys> bucketDeletedKeysList = purgeKeysRequest
         .getDeletedKeysList();
+    OmSnapshotManager omSnapshotManager = ozoneManager.getOmSnapshotManager();
+    String fromSnapshot = purgeKeysRequest.hasFromSnapshot() ?
+        purgeKeysRequest.getFromSnapshot() : null;
     List<String> keysToBePurgedList = new ArrayList<>();
 
     OMResponse.Builder omResponse = OmResponseUtil.getOMResponseBuilder(
@@ -64,10 +74,27 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       }
     }
 
-    omClientResponse = new OMKeyPurgeResponse(omResponse.build(),
-          keysToBePurgedList);
-    addResponseToDoubleBuffer(trxnLogIndex, omClientResponse,
-        omDoubleBufferHelper);
+    try {
+      OmSnapshot omFromSnapshot = null;
+      if (fromSnapshot != null) {
+        SnapshotInfo snapshotInfo =
+            ozoneManager.getMetadataManager().getSnapshotInfoTable()
+                .get(fromSnapshot);
+        omFromSnapshot = (OmSnapshot) omSnapshotManager
+            .checkForSnapshot(snapshotInfo.getVolumeName(),
+                snapshotInfo.getBucketName(),
+                getSnapshotPrefix(snapshotInfo.getName()));
+      }
+
+      omClientResponse = new OMKeyPurgeResponse(omResponse.build(),
+          keysToBePurgedList, omFromSnapshot);
+    }  catch (IOException ex) {
+      omClientResponse = new OMKeyPurgeResponse(
+          createErrorOMResponse(omResponse, ex));
+    } finally {
+      addResponseToDoubleBuffer(trxnLogIndex, omClientResponse,
+          omDoubleBufferHelper);
+    }
 
     return omClientResponse;
   }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/key/OMKeyPurgeResponse.java

@@ -18,7 +18,9 @@
 
 package org.apache.hadoop.ozone.om.response.key;
 
+import org.apache.hadoop.hdds.utils.db.DBStore;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
+import org.apache.hadoop.ozone.om.OmSnapshot;
 import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
 import org.apache.hadoop.ozone.om.request.key.OMKeyPurgeRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
@@ -36,19 +38,44 @@
 @CleanupTableInfo(cleanupTables = {DELETED_TABLE})
 public class OMKeyPurgeResponse extends OmKeyResponse {
   private List<String> purgeKeyList;
+  private OmSnapshot fromSnapshot;
 
   public OMKeyPurgeResponse(@Nonnull OMResponse omResponse,
-      @Nonnull List<String> keyList) {
+      @Nonnull List<String> keyList, OmSnapshot fromSnapshot) {
     super(omResponse);
     this.purgeKeyList = keyList;
+    this.fromSnapshot = fromSnapshot;
+  }
+
+  /**
+   * For when the request is not successful.
+   * For a successful request, the other constructor should be used.
+   */
+  public OMKeyPurgeResponse(@Nonnull OMResponse omResponse) {
+    super(omResponse);
+    checkStatusNotOK();
   }
 
   @Override
   public void addToDBBatch(OMMetadataManager omMetadataManager,
       BatchOperation batchOperation) throws IOException {
 
+    if (fromSnapshot != null) {
+      DBStore fromSnapshotStore = fromSnapshot.getMetadataManager().getStore();
+      // Init Batch Operation for snapshot db.
+      try (BatchOperation writeBatch = fromSnapshotStore.initBatchOperation()) {
+        processKeys(writeBatch, fromSnapshot.getMetadataManager());
+        fromSnapshotStore.commitBatchOperation(writeBatch);
+      }
+    } else {
+      processKeys(batchOperation, omMetadataManager);
+    }
+  }
+
+  public void processKeys(BatchOperation batchOp,
+      OMMetadataManager metadataManager) throws IOException {
     for (String key : purgeKeyList) {
-      omMetadataManager.getDeletedTable().deleteWithBatch(batchOperation,
+      metadataManager.getDeletedTable().deleteWithBatch(batchOp,
           key);
     }
   }