HDDS-4131. Container report should update container key count and bytes used if they differ in SCM

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/AbstractContainerReportHandler.java

@@ -20,6 +20,7 @@
 
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleEvent;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
@@ -28,6 +29,9 @@
 import org.slf4j.Logger;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
 import java.util.UUID;
 import java.util.function.Supplier;
 
@@ -76,7 +80,7 @@ protected void processContainerReplica(final DatanodeDetails datanodeDetails,
     // Synchronized block should be replaced by container lock,
     // once we have introduced lock inside ContainerInfo.
     synchronized (containerManager.getContainer(containerId)) {
-      updateContainerStats(containerId, replicaProto);
+      updateContainerStats(datanodeDetails, containerId, replicaProto);
       updateContainerState(datanodeDetails, containerId, replicaProto);
       updateContainerReplica(datanodeDetails, containerId, replicaProto);
     }
@@ -90,7 +94,8 @@ protected void processContainerReplica(final DatanodeDetails datanodeDetails,
    * @param replicaProto Container Replica information
    * @throws ContainerNotFoundException If the container is not present
    */
-  private void updateContainerStats(final ContainerID containerId,
+  private void updateContainerStats(final DatanodeDetails datanodeDetails,
+                                    final ContainerID containerId,
                                     final ContainerReplicaProto replicaProto)
       throws ContainerNotFoundException {
 
@@ -103,14 +108,44 @@ private void updateContainerStats(final ContainerID containerId,
         containerInfo.updateSequenceId(
             replicaProto.getBlockCommitSequenceId());
       }
+      List<ContainerReplica> otherReplicas =
+          getOtherReplicas(containerId, datanodeDetails);
+      long usedBytes = replicaProto.getUsed();
+      long keyCount = replicaProto.getKeyCount();
+      for (ContainerReplica r : otherReplicas) {
+        // Open containers are generally growing in key count and size, the
+        // overall size should be the min of all reported replicas.
+        if (containerInfo.getState().equals(HddsProtos.LifeCycleState.OPEN)) {
+          usedBytes = Math.min(usedBytes, r.getBytesUsed());
+          keyCount = Math.min(keyCount, r.getKeyCount());
+        } else {
+          // Containers which are not open can only shrink in size, so use the
+          // largest values reported.
+          usedBytes = Math.max(usedBytes, r.getBytesUsed());
+          keyCount = Math.max(keyCount, r.getKeyCount());
+        }
+      }
 
-      if (containerInfo.getUsedBytes() < replicaProto.getUsed()) {
-        containerInfo.setUsedBytes(replicaProto.getUsed());
+      if (containerInfo.getUsedBytes() != usedBytes) {
+        containerInfo.setUsedBytes(usedBytes);
+      }
+      if (containerInfo.getNumberOfKeys() != keyCount) {
+        containerInfo.setNumberOfKeys(keyCount);
       }
-      if (containerInfo.getNumberOfKeys() < replicaProto.getKeyCount()) {
-        containerInfo.setNumberOfKeys(replicaProto.getKeyCount());
+    }
+  }
+
+  private List<ContainerReplica> getOtherReplicas(ContainerID containerId,
+      DatanodeDetails exclude) throws ContainerNotFoundException {
+    List<ContainerReplica> filteredReplicas = new ArrayList<>();
+    Set<ContainerReplica> replicas
+        = containerManager.getContainerReplicas(containerId);
+    for (ContainerReplica r : replicas) {
+      if (!r.getDatanodeDetails().equals(exclude)) {
+        filteredReplicas.add(r);
       }
     }
+    return filteredReplicas;
   }
 
   /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java

@@ -312,14 +312,6 @@ private void processContainer(ContainerID id) {
           action -> replicas.stream()
               .noneMatch(r -> r.getDatanodeDetails().equals(action.datanode)));
 
-      /*
-       * If the container is in CLOSED state, check and update it's key count
-       * and bytes used statistics if needed.
-       */
-      if (state == LifeCycleState.CLOSED) {
-        checkAndUpdateContainerInfo(container, replicas);
-      }
-
       /*
        * We don't have to take any action if the container is healthy.
        *
@@ -773,32 +765,6 @@ private void handleUnstableContainer(final ContainerInfo container,
 
   }
 
-  /**
-   * Check and update Container key count and used bytes based on it's replica's
-   * data.
-   */
-  private void checkAndUpdateContainerInfo(final ContainerInfo container,
-      final Set<ContainerReplica> replicas) {
-    // check container key count and bytes used
-    long maxUsedBytes = 0;
-    long maxKeyCount = 0;
-    ContainerReplica[] rps = replicas.toArray(new ContainerReplica[0]);
-    for (int i = 0; i < rps.length; i++) {
-      maxUsedBytes = Math.max(maxUsedBytes, rps[i].getBytesUsed());
-      maxKeyCount = Math.max(maxKeyCount, rps[i].getKeyCount());
-    }
-    if (maxKeyCount < container.getNumberOfKeys()) {
-      LOG.debug("Container {} key count changed from {} to {}",
-          container.containerID(), container.getNumberOfKeys(), maxKeyCount);
-      container.setNumberOfKeys(maxKeyCount);
-    }
-    if (maxUsedBytes < container.getUsedBytes()) {
-      LOG.debug("Container {} used bytes changed from {} to {}",
-          container.containerID(), container.getUsedBytes(), maxUsedBytes);
-      container.setUsedBytes(maxUsedBytes);
-    }
-  }
-
   /**
    * Sends close container command for the given container to the given
    * datanode.

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java

@@ -398,8 +398,6 @@ public void updateDeleteTransactionId(Map<Long, Long> deleteTransactionMap)
               SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
         }
         containerInfo.updateDeleteTransactionId(entry.getValue());
-        containerInfo.setNumberOfKeys(containerInfoInMem.getNumberOfKeys());
-        containerInfo.setUsedBytes(containerInfoInMem.getUsedBytes());
         containerStore
             .putWithBatch(batchOperation, containerIdObject, containerInfo);
       }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestBlockDeletion.java

@@ -270,11 +270,8 @@ public void testContainerStatistics() throws Exception {
     });
 
     om.deleteKey(keyArgs);
-    // Want for blocks to be deleted
+    // Wait for blocks to be deleted and container reports to be processed
     Thread.sleep(5000);
-    scm.getReplicationManager().processContainersNow();
-    // Wait for container statistics change
-    Thread.sleep(1000);
     containerInfos = scm.getContainerManager().getContainers();
     containerInfos.stream().forEach(container -> {
       Assert.assertEquals(0, container.getUsedBytes());