HDDS-7740. [Snapshot] Implement SnapshotDeletingService

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

@@ -232,6 +232,16 @@ public final class OzoneConfigKeys {
       OZONE_SNAPSHOT_SST_FILTERING_SERVICE_TIMEOUT_DEFAULT = "300s";
       // 300s for default
 
+  public static final String OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL =
+      "ozone.snapshot.deleting.service.interval";
+  public static final String
+      OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL_DEFAULT = "30s";
+
+  public static final String OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT =
+      "ozone.snapshot.deleting.service.timeout";
+  public static final String
+      OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT_DEFAULT = "300s";
+
   public static final String OZONE_BLOCK_DELETING_SERVICE_WORKERS =
       "ozone.block.deleting.service.workers";
   public static final int OZONE_BLOCK_DELETING_SERVICE_WORKERS_DEFAULT

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

@@ -237,6 +237,11 @@ public enum Units { TB, GB, MB, KB, B }
 
   public static final int INVALID_PORT = -1;
 
+  /**
+   * Object ID to identify reclaimable uncommitted blocks.
+   */
+  public static final long OBJECT_ID_RECLAIM_BLOCKS = 0L;
+
 
   /**
    * Default SCM Datanode ID file name.

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -3132,6 +3132,14 @@
       sst filtering service per time interval.
     </description>
   </property>
+  <property>
+    <name>ozone.snapshot.deleting.limit.per.task</name>
+    <value>10</value>
+    <tag>OZONE, PERFORMANCE, OM</tag>
+    <description>The maximum number of snapshots that would be reclaimed by
+      Snapshot Deleting Service per run.
+    </description>
+  </property>
   <property>
     <name>ozone.snapshot.filtering.service.interval</name>
     <value>1m</value>
@@ -3147,6 +3155,25 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.snapshot.deleting.service.timeout</name>
+    <value>300s</value>
+    <tag>OZONE, PERFORMANCE, OM</tag>
+    <description>
+      Timeout value for SnapshotDeletingService.
+    </description>
+  </property>
+
+  <property>
+    <name>ozone.snapshot.deleting.service.interval</name>
+    <value>30s</value>
+    <tag>OZONE, PERFORMANCE, OM</tag>
+    <description>
+      The time interval between successive SnapshotDeletingService
+      thread run.
+    </description>
+  </property>
+
   <property>
     <name>ozone.scm.event.ContainerReport.thread.pool.size</name>
     <value>10</value>

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -308,6 +308,7 @@ public static boolean isReadOnly(
     case SetRangerServiceVersion:
     case CreateSnapshot:
     case DeleteSnapshot:
+    case SnapshotMoveDeletedKeys:
       return false;
     default:
       LOG.error("CmdType {} is not categorized as readOnly or not.", cmdType);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/audit/OMAction.java

@@ -93,8 +93,9 @@ public enum OMAction implements AuditAction {
   TENANT_LIST_USER,
 
   CREATE_SNAPSHOT,
+  LIST_SNAPSHOT,
   DELETE_SNAPSHOT,
-  LIST_SNAPSHOT;
+  SNAPSHOT_MOVE_DELETED_KEYS;
 
   @Override
   public String getAction() {

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OMConfigKeys.java

@@ -324,6 +324,10 @@ private OMConfigKeys() {
       "ozone.snapshot.filtering.limit.per.task";
   public static final int SNAPSHOT_SST_DELETING_LIMIT_PER_TASK_DEFAULT = 2;
 
+  public static final String SNAPSHOT_DELETING_LIMIT_PER_TASK =
+      "ozone.snapshot.deleting.limit.per.task";
+  public static final int SNAPSHOT_DELETING_LIMIT_PER_TASK_DEFAULT = 10;
+
   public static final String OZONE_SNAPSHOT_SST_FILTERING_SERVICE_INTERVAL =
       "ozone.snapshot.filtering.service.interval";
   public static final String

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/SnapshotInfo.java

@@ -368,10 +368,17 @@ public OzoneManagerProtocolProtos.SnapshotInfo getProtobuf() {
         .setBucketName(bucketName)
         .setSnapshotStatus(snapshotStatus.toProto())
         .setCreationTime(creationTime)
-        .setDeletionTime(deletionTime)
-        .setPathPreviousSnapshotID(pathPreviousSnapshotID)
-        .setGlobalPreviousSnapshotID(globalPreviousSnapshotID)
-        .setSnapshotPath(snapshotPath)
+        .setDeletionTime(deletionTime);
+
+    if (pathPreviousSnapshotID != null) {
+      sib.setPathPreviousSnapshotID(pathPreviousSnapshotID);
+    }
+
+    if (globalPreviousSnapshotID != null) {
+      sib.setGlobalPreviousSnapshotID(globalPreviousSnapshotID);
+    }
+
+    sib.setSnapshotPath(snapshotPath)
         .setCheckpointDir(checkpointDir)
         .setDbTxSequenceNumber(dbTxSequenceNumber);
     return sib.build();
@@ -392,12 +399,19 @@ public static SnapshotInfo getFromProtobuf(
         .setSnapshotStatus(SnapshotStatus.valueOf(snapshotInfoProto
             .getSnapshotStatus()))
         .setCreationTime(snapshotInfoProto.getCreationTime())
-        .setDeletionTime(snapshotInfoProto.getDeletionTime())
-        .setPathPreviousSnapshotID(snapshotInfoProto.
-            getPathPreviousSnapshotID())
-        .setGlobalPreviousSnapshotID(snapshotInfoProto.
-            getGlobalPreviousSnapshotID())
-        .setSnapshotPath(snapshotInfoProto.getSnapshotPath())
+        .setDeletionTime(snapshotInfoProto.getDeletionTime());
+
+    if (snapshotInfoProto.hasPathPreviousSnapshotID()) {
+      osib.setPathPreviousSnapshotID(snapshotInfoProto.
+          getPathPreviousSnapshotID());
+    }
+
+    if (snapshotInfoProto.hasGlobalPreviousSnapshotID()) {
+      osib.setGlobalPreviousSnapshotID(snapshotInfoProto.
+          getGlobalPreviousSnapshotID());
+    }
+
+    osib.setSnapshotPath(snapshotInfoProto.getSnapshotPath())
         .setCheckpointDir(snapshotInfoProto.getCheckpointDir())
         .setDbTxSequenceNumber(snapshotInfoProto.getDbTxSequenceNumber());
 
@@ -499,8 +513,9 @@ public boolean equals(Object o) {
         name.equals(that.name) && volumeName.equals(that.volumeName) &&
         bucketName.equals(that.bucketName) &&
         snapshotStatus == that.snapshotStatus &&
-        pathPreviousSnapshotID.equals(that.pathPreviousSnapshotID) &&
-        globalPreviousSnapshotID.equals(that.globalPreviousSnapshotID) &&
+        Objects.equals(pathPreviousSnapshotID, that.pathPreviousSnapshotID) &&
+        Objects.equals(
+            globalPreviousSnapshotID, that.globalPreviousSnapshotID) &&
         snapshotPath.equals(that.snapshotPath) &&
         checkpointDir.equals(that.checkpointDir);
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/WithObjectID.java

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.ozone.om.helpers;
 
+import static org.apache.hadoop.ozone.OzoneConsts.OBJECT_ID_RECLAIM_BLOCKS;
+
 /**
  * Mixin class to handle ObjectID and UpdateID.
  */
@@ -52,7 +54,7 @@ public long getUpdateID() {
   }
 
   /**
-   * Set the Object ID. If this value is already set then this function throws.
+   * Set the Object ID.
    * There is a reason why we cannot use the final here. The object
    * ({@link OmVolumeArgs}/ {@link OmBucketInfo}/ {@link OmKeyInfo}) is
    * deserialized from the protobuf in many places in code. We need to set
@@ -61,7 +63,7 @@ public long getUpdateID() {
    * @param obId - long
    */
   public void setObjectID(long obId) {
-    if (this.objectID != 0) {
+    if (this.objectID != 0 && obId != OBJECT_ID_RECLAIM_BLOCKS) {
       throw new UnsupportedOperationException("Attempt to modify object ID " +
           "which is not zero. Current Object ID is " + this.objectID);
     }

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

@@ -130,8 +130,9 @@ enum Type {
   ListSnapshot = 113;
   SnapshotDiff = 114;
   DeleteSnapshot = 115;
+  SnapshotMoveDeletedKeys = 116;
 
-  TransferLeadership = 116;
+  TransferLeadership = 117;
 }
 
 message OMRequest {
@@ -244,8 +245,9 @@ message OMRequest {
   optional ListSnapshotRequest              ListSnapshotRequest            = 113;
   optional SnapshotDiffRequest              snapshotDiffRequest            = 114;
   optional DeleteSnapshotRequest            DeleteSnapshotRequest          = 115;
+  optional SnapshotMoveDeletedKeysRequest   SnapshotMoveDeletedKeysRequest = 116;
 
-  optional hdds.TransferLeadershipRequestProto      TransferOmLeadershipRequest    = 116;
+  optional hdds.TransferLeadershipRequestProto      TransferOmLeadershipRequest    = 117;
 
 }
 
@@ -352,8 +354,9 @@ message OMResponse {
   optional ListSnapshotResponse              ListSnapshotResponse          = 113;
   optional SnapshotDiffResponse              snapshotDiffResponse          = 114;
   optional DeleteSnapshotResponse            DeleteSnapshotResponse        = 115;
+  optional SnapshotMoveDeletedKeysResponse   SnapshotMoveDeletedKeysResponse = 116;
 
-  optional hdds.TransferLeadershipResponseProto   TransferOmLeadershipResponse  = 116;
+  optional hdds.TransferLeadershipResponseProto   TransferOmLeadershipResponse  = 117;
 }
 
 enum Status {
@@ -1703,6 +1706,17 @@ message DeleteSnapshotRequest {
   optional uint64 deletionTime = 4;
 }
 
+message SnapshotMoveDeletedKeysRequest {
+  optional SnapshotInfo fromSnapshot = 1;
+  repeated SnapshotMoveKeyInfos nextDBKeys = 2;
+  repeated SnapshotMoveKeyInfos reclaimKeys = 3;
+}
+
+message SnapshotMoveKeyInfos {
+  optional string key = 1;
+  repeated KeyInfo keyInfos = 2;
+}
+
 message DeleteTenantRequest {
     optional string tenantId = 1;
 }
@@ -1765,6 +1779,10 @@ message DeleteSnapshotResponse {
 
 }
 
+message SnapshotMoveDeletedKeysResponse {
+
+}
+
 message SnapshotDiffReportProto {
   optional string volumeName = 1;
   optional string bucketName = 2;

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

@@ -247,4 +247,10 @@ List<OmKeyInfo> getPendingDeletionSubFiles(long volumeId,
    * @return Background service.
    */
   BackgroundService getSnapshotSstFilteringService();
+
+  /**
+   * Returns the instance of Snapshot Deleting service.
+   * @return Background service.
+   */
+  BackgroundService getSnapshotDeletingService();
 }

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

@@ -84,6 +84,7 @@
 import org.apache.hadoop.ozone.om.service.DirectoryDeletingService;
 import org.apache.hadoop.ozone.om.service.KeyDeletingService;
 import org.apache.hadoop.ozone.om.service.OpenKeyCleanupService;
+import org.apache.hadoop.ozone.om.service.SnapshotDeletingService;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo;
 import org.apache.hadoop.hdds.security.token.OzoneBlockTokenSecretManager;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
@@ -110,6 +111,10 @@
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_LIST_TRASH_KEYS_MAX_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL_DEFAULT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_SST_FILTERING_SERVICE_TIMEOUT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_SNAPSHOT_SST_FILTERING_SERVICE_TIMEOUT_DEFAULT;
 import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_DELIMITER;
@@ -159,6 +164,7 @@ public class KeyManagerImpl implements KeyManager {
   private BackgroundService keyDeletingService;
 
   private BackgroundService snapshotSstFilteringService;
+  private BackgroundService snapshotDeletingService;
 
   private final KeyProviderCryptoExtension kmsProvider;
   private final boolean enableFileSystemPaths;
@@ -261,6 +267,24 @@ public void start(OzoneConfiguration configuration) {
         snapshotSstFilteringService.start();
       }
     }
+
+    if (snapshotDeletingService == null) {
+      long snapshotServiceInterval = configuration.getTimeDuration(
+          OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL,
+          OZONE_SNAPSHOT_DELETING_SERVICE_INTERVAL_DEFAULT,
+          TimeUnit.MILLISECONDS);
+      long snapshotServiceTimeout = configuration.getTimeDuration(
+          OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT,
+          OZONE_SNAPSHOT_DELETING_SERVICE_TIMEOUT_DEFAULT,
+          TimeUnit.MILLISECONDS);
+      try {
+        snapshotDeletingService = new SnapshotDeletingService(
+            snapshotServiceInterval, snapshotServiceTimeout, ozoneManager);
+        snapshotDeletingService.start();
+      } catch (IOException e) {
+        LOG.error("Error starting Snapshot Deleting Service", e);
+      }
+    }
   }
 
   KeyProviderCryptoExtension getKMSProvider() {
@@ -285,6 +309,10 @@ public void stop() throws IOException {
       snapshotSstFilteringService.shutdown();
       snapshotSstFilteringService = null;
     }
+    if (snapshotDeletingService != null) {
+      snapshotDeletingService.shutdown();
+      snapshotDeletingService = null;
+    }
   }
 
   private OmBucketInfo getBucketInfo(String volumeName, String bucketName)
@@ -621,6 +649,10 @@ public BackgroundService getSnapshotSstFilteringService() {
     return snapshotSstFilteringService;
   }
 
+  public BackgroundService getSnapshotDeletingService() {
+    return snapshotDeletingService;
+  }
+
   @Override
   public OmMultipartUploadList listMultipartUploads(String volumeName,
       String bucketName, String prefix) throws OMException {

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

@@ -230,7 +230,7 @@ public SnapshotInfo getSnapshotInfo(String volumeName,
         bucketName, snapshotName));
   }
 
-  private SnapshotInfo getSnapshotInfo(String key) throws IOException {
+  public SnapshotInfo getSnapshotInfo(String key) throws IOException {
     SnapshotInfo snapshotInfo;
     try {
       snapshotInfo = ozoneManager.getMetadataManager()

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

@@ -452,6 +452,7 @@ private enum State {
   // This metadata reader points to the active filesystem
   private OmMetadataReader omMetadataReader;
   private OmSnapshotManager omSnapshotManager;
+  private SnapshotChainManager snapshotChainManager;
 
   /** A list of property that are reconfigurable at runtime. */
   private final SortedSet<String> reconfigurableProperties =
@@ -769,6 +770,7 @@ private void instantiateServices(boolean withNewSnapshot) throws IOException {
     omMetadataReader = new OmMetadataReader(keyManager, prefixManager,
         this, LOG, AUDIT, metrics);
     omSnapshotManager = new OmSnapshotManager(this);
+    snapshotChainManager = new SnapshotChainManager(metadataManager);
 
     // Snapshot metrics
     updateActiveSnapshotMetrics();
@@ -1508,6 +1510,15 @@ public OmSnapshotManager getOmSnapshotManager() {
     return omSnapshotManager;
   }
 
+  /**
+   * Get Snapshot Chain Manager.
+   *
+   * @return SnapshotChainManager.
+   */
+  public SnapshotChainManager getSnapshotChainManager() {
+    return snapshotChainManager;
+  }
+
   /**
    * Get metadata manager.
    *