HDDS-13184. Persist Block Size in Delete Transaction for SCM

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/upgrade/HDDSLayoutFeature.java

@@ -42,7 +42,8 @@ public enum HDDSLayoutFeature implements LayoutFeature {
                                      "to DatanodeDetails."),
   HBASE_SUPPORT(8, "Datanode RocksDB Schema Version 3 has an extra table " +
           "for the last chunk of blocks to support HBase.)"),
-  WITNESSED_CONTAINER_DB_PROTO_VALUE(9, "ContainerID table schema to use value type as proto");
+  WITNESSED_CONTAINER_DB_PROTO_VALUE(9, "ContainerID table schema to use value type as proto"),
+  STORAGE_SPACE_DISTRIBUTION(10, "Enhanced block deletion function for storage space distribution feature.");
 
   //////////////////////////////  //////////////////////////////
 

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java

@@ -29,6 +29,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionSummary;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ContainerBalancerStatusInfoResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.DecommissionScmResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartContainerBalancerResponseProto;
@@ -411,6 +412,12 @@ StartContainerBalancerResponseProto startContainerBalancer(
    */
   void transferLeadership(String newLeaderId) throws IOException;
 
+  /**
+   * Get deleted block summary.
+   * @throws IOException
+   */
+  DeletedBlocksTransactionSummary getDeletedBlockSummary() throws IOException;
+
   /**
    * Get usage information of datanode by address or uuid.
    *

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.hdds.scm.protocol;
 
+import jakarta.annotation.Nullable;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Collections;
@@ -31,6 +32,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionInfo;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionSummary;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ContainerBalancerStatusInfoResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.DecommissionScmResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartContainerBalancerResponseProto;
@@ -361,6 +363,14 @@ List<DeletedBlocksTransactionInfo> getFailedDeletedBlockTxn(int count,
   @Deprecated
   int resetDeletedBlockRetryCount(List<Long> txIDs) throws IOException;
 
+
+  /**
+   * Get deleted block summary.
+   * @throws IOException
+   */
+  @Nullable
+  DeletedBlocksTransactionSummary getDeletedBlockSummary() throws IOException;
+
   /**
    * Check if SCM is in safe mode.
    *

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java

@@ -24,6 +24,7 @@
 import com.google.common.base.Preconditions;
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
+import jakarta.annotation.Nullable;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -42,6 +43,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionInfo;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionSummary;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.GetScmInfoResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.TransferLeadershipRequestProto;
@@ -78,6 +80,8 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerWithPipelineRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainersOnDecomNodeRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainersOnDecomNodeResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetDeletedBlocksTxnSummaryRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetDeletedBlocksTxnSummaryResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetExistContainerWithPipelinesInBatchRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetMetricsRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetMetricsResponseProto;
@@ -801,6 +805,18 @@ public int resetDeletedBlockRetryCount(List<Long> txIDs)
     return 0;
   }
 
+  @Nullable
+  @Override
+  public DeletedBlocksTransactionSummary getDeletedBlockSummary() throws IOException {
+    GetDeletedBlocksTxnSummaryRequestProto request =
+        GetDeletedBlocksTxnSummaryRequestProto.newBuilder().build();
+    ScmContainerLocationResponse scmContainerLocationResponse = submitRequest(Type.GetDeletedBlocksTransactionSummary,
+        builder -> builder.setGetDeletedBlocksTxnSummaryRequest(request));
+    GetDeletedBlocksTxnSummaryResponseProto response =
+        scmContainerLocationResponse.getGetDeletedBlocksTxnSummaryResponse();
+    return response.hasSummary() ? response.getSummary() : null;
+  }
+
   /**
    * Check if SCM is in safe mode.
    *

hadoop-hdds/interface-admin/src/main/proto/ScmAdminProtocol.proto

@@ -86,6 +86,7 @@ message ScmContainerLocationRequest {
   optional GetMetricsRequestProto getMetricsRequest = 47;
   optional ContainerBalancerStatusInfoRequestProto containerBalancerStatusInfoRequest = 48;
   optional ReconcileContainerRequestProto reconcileContainerRequest = 49;
+  optional GetDeletedBlocksTxnSummaryRequestProto getDeletedBlocksTxnSummaryRequest = 50;
 }
 
 message ScmContainerLocationResponse {
@@ -143,6 +144,7 @@ message ScmContainerLocationResponse {
   optional GetMetricsResponseProto getMetricsResponse = 47;
   optional ContainerBalancerStatusInfoResponseProto containerBalancerStatusInfoResponse = 48;
   optional ReconcileContainerResponseProto reconcileContainerResponse = 49;
+  optional GetDeletedBlocksTxnSummaryResponseProto getDeletedBlocksTxnSummaryResponse = 50;
 
   enum Status {
     OK = 1;
@@ -199,6 +201,7 @@ enum Type {
   GetMetrics = 43;
   GetContainerBalancerStatusInfo = 44;
   ReconcileContainer = 45;
+  GetDeletedBlocksTransactionSummary = 46;
 }
 
 /**
@@ -545,6 +548,13 @@ message ResetDeletedBlockRetryCountResponseProto {
   required int32 resetCount = 1;
 }
 
+message GetDeletedBlocksTxnSummaryRequestProto {
+}
+
+message GetDeletedBlocksTxnSummaryResponseProto {
+  optional DeletedBlocksTransactionSummary summary = 1;
+}
+
 message FinalizeScmUpgradeRequestProto {
   required string upgradeClientId = 1;
 }

hadoop-hdds/interface-client/src/main/proto/hdds.proto

@@ -514,6 +514,13 @@ message DeletedBlocksTransactionInfo {
     optional int32 count = 4;
 }
 
+message DeletedBlocksTransactionSummary {
+    optional uint64 totalTransactionCount = 1;
+    optional uint64 totalBlockCount = 2;
+    optional uint64 totalBlockSize = 3;
+    optional uint64 totalBlockReplicatedSize = 4;
+}
+
 message CompactionFileInfoProto {
     optional string fileName = 1;
     optional string startKey = 2;

hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto

@@ -369,6 +369,8 @@ message DeletedBlocksTransaction {
   // the retry time of sending deleting command to datanode.
   // We don't have to store the retry count in DB.
   optional int32 count = 4 [deprecated=true];
+  optional uint64 totalBlockSize = 5;
+  optional uint64 totalBlockReplicatedSize = 6;
 }
 
 // ACK message datanode sent to SCM, contains the result of

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java

@@ -93,13 +93,13 @@ public BlockManagerImpl(final ConfigurationSource conf,
     this.writableContainerFactory = scm.getWritableContainerFactory();
 
     mxBean = MBeans.register("BlockManager", "BlockManagerImpl", this);
-    metrics = ScmBlockDeletingServiceMetrics.create();
+    metrics = ScmBlockDeletingServiceMetrics.create(this);
 
     // SCM block deleting transaction log and deleting service.
     deletedBlockLog = new DeletedBlockLogImpl(conf,
         scm,
         scm.getContainerManager(),
-        scm.getScmHAManager().getDBTransactionBuffer(),
+        scm.getScmHAManager().asSCMHADBTransactionBuffer(),
         metrics);
 
 
@@ -220,9 +220,7 @@ public void deleteBlocks(List<BlockGroup> keyBlocksInfoList)
       throw new SCMException("SafeModePrecheck failed for deleteBlocks",
           SCMException.ResultCodes.SAFE_MODE_EXCEPTION);
     }
-    Map<Long, List<Long>> containerBlocks = new HashMap<>();
-    // TODO: track the block size info so that we can reclaim the container
-    // TODO: used space when the block is deleted.
+    Map<Long, List<DeletedBlock>> containerBlocks = new HashMap<>();
     for (BlockGroup bg : keyBlocksInfoList) {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Deleting blocks {}",
@@ -232,10 +230,10 @@ public void deleteBlocks(List<BlockGroup> keyBlocksInfoList)
         BlockID block = deletedBlock.getBlockID();
         long containerID = block.getContainerID();
         if (containerBlocks.containsKey(containerID)) {
-          containerBlocks.get(containerID).add(block.getLocalID());
+          containerBlocks.get(containerID).add(deletedBlock);
         } else {
-          List<Long> item = new ArrayList<>();
-          item.add(block.getLocalID());
+          List<DeletedBlock> item = new ArrayList<>();
+          item.add(deletedBlock);
           containerBlocks.put(containerID, item);
         }
       }

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java

@@ -17,15 +17,19 @@
 
 package org.apache.hadoop.hdds.scm.block;
 
+import com.google.protobuf.ByteString;
+import jakarta.annotation.Nullable;
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeID;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DeletedBlocksTransactionSummary;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
 import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.common.DeletedBlock;
 import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
 
 /**
@@ -100,7 +104,7 @@ void recordTransactionCreated(
    * @param containerBlocksMap a map of containerBlocks.
    * @throws IOException
    */
-  void addTransactions(Map<Long, List<Long>> containerBlocksMap)
+  void addTransactions(Map<Long, List<DeletedBlock>> containerBlocksMap)
       throws IOException;
 
   /**
@@ -115,8 +119,13 @@ void addTransactions(Map<Long, List<Long>> containerBlocksMap)
   /**
    * Reinitialize the delete log from the db.
    * @param deletedBlocksTXTable delete transaction table
+   * @param statefulConfigTable stateful service config table
    */
-  void reinitialize(Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable);
+  void reinitialize(Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable,
+      Table<String, ByteString> statefulConfigTable) throws IOException;
 
   int getTransactionToDNsCommitMapSize();
+
+  @Nullable
+  DeletedBlocksTransactionSummary getTransactionSummary();
 }