HDDS-11881. Simplify DatanodeAdminMonitorImpl Code Structure.

Author: fanshilun

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

@@ -286,7 +286,7 @@ public String toString() {
    *
    * When considering inflight operations, it is assumed any operation will
    * fail. However, to consider the worst case and avoid data loss, we always
-   * assume a delete will succeed and and add will fail. In this way, we will
+   * assume a delete will succeed and add will fail. In this way, we will
    * avoid scheduling too many deletes which could result in dataloss.
    *
    * Decisions around over-replication are made only on healthy replicas,
@@ -320,7 +320,7 @@ public String toString() {
    * For under replicated containers we do consider inflight add and delete to
    * avoid scheduling more adds than needed. There is additional logic around
    * containers with maintenance replica to ensure minHealthyForMaintenance
-   * replia are maintained.
+   * replica are maintained.
    *
    * @return Delta of replicas needed. Negative indicates over replication and
    *         containers should be removed. Positive indicates over replication
@@ -621,7 +621,7 @@ public int getExcessRedundancy(boolean includePendingDelete) {
    * considering inflight add and deletes.
    * @return zero if perfectly replicated, a negative value for over replication
    *         and a positive value for under replication. The magnitude of the
-   *         return value indicates how many replias the container is over or
+   *         return value indicates how many replies the container is over or
    *         under replicated by.
    */
   private int redundancyDelta(boolean includePendingDelete,
@@ -650,7 +650,7 @@ boolean insufficientDueToOutOfService() {
 
   /**
    * How many more replicas can be lost before the container is
-   * unreadable, assuming any infligh deletes will complete. For containers
+   * unreadable, assuming any inflight deletes will complete. For containers
    * which are under-replicated due to decommission
    * or maintenance only, the remaining redundancy will include those
    * decommissioning or maintenance replicas, as they are technically still

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

@@ -145,7 +145,7 @@ public int processAndSendCommands(
 
     if (targetDatanodes.size() < replicaCount.additionalReplicaNeeded()) {
       // The placement policy failed to find enough targets to satisfy fix
-      // the under replication. There fore even though some commands were sent,
+      // the under replication. Therefore even though some commands were sent,
       // we throw an exception to indicate that the container is still under
       // replicated and should be re-queued for another attempt later.
       LOG.debug("Placement policy failed to find enough targets to satisfy " +

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

@@ -92,7 +92,7 @@ public void processAll(ReplicationQueue queue) {
     List<HealthResult> failedOnes = new LinkedList<>();
     // Getting the limit requires iterating over all nodes registered in
     // NodeManager and counting the healthy ones. This is somewhat expensive
-    // so we get get the count once per iteration as it should not change too
+    // so we get the count once per iteration as it should not change too
     // often.
     long inflightLimit = replicationManager.getReplicationInFlightLimit();
     while (true) {

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeAdminMonitor.java

@@ -33,7 +33,7 @@ public interface DatanodeAdminMonitor extends Runnable {
 
   void startMonitoring(DatanodeDetails dn);
   void stopMonitoring(DatanodeDetails dn);
-  Set<DatanodeAdminMonitorImpl.TrackedNode> getTrackedNodes();
+  Set<TrackedNode> getTrackedNodes();
   void setMetrics(NodeDecommissionMetrics metrics);
   Map<String, List<ContainerID>> getContainersPendingReplication(DatanodeDetails dn)
       throws NodeNotFoundException;

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeAdminMonitorImpl.java

@@ -25,7 +25,6 @@
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
-import org.apache.hadoop.hdds.scm.container.ContainerReplica;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.replication.ContainerReplicaCount;
 import org.apache.hadoop.hdds.scm.container.replication.ReplicationManager;
@@ -39,8 +38,6 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayDeque;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -49,12 +46,14 @@
 import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
+
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.DELETED;
+import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState.DELETING;
 
 /**
  * Monitor thread which watches for nodes to be decommissioned, recommissioned
  * or placed into maintenance. Newly added nodes are queued in pendingNodes
- * and recommissoned nodes are queued in cancelled nodes. On each monitor
+ * and recommissioned nodes are queued in cancelled nodes. On each monitor
  * 'tick', the cancelled nodes are processed and removed from the monitor.
  * Then any pending nodes are added to the trackedNodes set, where they stay
  * until decommission or maintenance has ended.
@@ -89,50 +88,6 @@ public class DatanodeAdminMonitorImpl implements DatanodeAdminMonitor {
   private long unClosedContainers = 0;
   private long underReplicatedContainers = 0;
 
-  /**
-   * Inner class for snapshot of Datanode ContainerState in
-   * Decommissioning and Maintenance mode workflow.
-   */
-  public static final class TrackedNode {
-
-    private DatanodeDetails datanodeDetails;
-    private long startTime = 0L;
-    private Map<String, List<ContainerID>> containersReplicatedOnNode = new ConcurrentHashMap<>();
-
-    public TrackedNode(DatanodeDetails datanodeDetails, long startTime) {
-      this.datanodeDetails = datanodeDetails;
-      this.startTime = startTime;
-    }
-
-    @Override
-    public int hashCode() {
-      return datanodeDetails.hashCode();
-    }
-
-    @Override
-    public boolean equals(Object obj) {
-      return obj instanceof TrackedNode &&
-          datanodeDetails.equals(((TrackedNode) obj).datanodeDetails);
-    }
-
-    public DatanodeDetails getDatanodeDetails() {
-      return datanodeDetails;
-    }
-
-    public long getStartTime() {
-      return startTime;
-    }
-
-    public Map<String, List<ContainerID>> getContainersReplicatedOnNode() {
-      return containersReplicatedOnNode;
-    }
-
-    public void setContainersReplicatedOnNode(List<ContainerID> underReplicated, List<ContainerID> unClosed) {
-      this.containersReplicatedOnNode.put("UnderReplicated", Collections.unmodifiableList(underReplicated));
-      this.containersReplicatedOnNode.put("UnClosed", Collections.unmodifiableList(unClosed));
-    }
-  }
-
   private Map<String, ContainerStateInWorkflow> containerStateByHost;
 
   private static final Logger LOG =
@@ -407,91 +362,96 @@ private boolean checkPipelinesClosedOnNode(TrackedNode dn)
 
   private boolean checkContainersReplicatedOnNode(TrackedNode dn)
       throws NodeNotFoundException {
-    int sufficientlyReplicated = 0;
-    int deleting = 0;
-    int underReplicated = 0;
-    int unclosed = 0;
-    List<ContainerID> underReplicatedIDs = new ArrayList<>();
-    List<ContainerID> unClosedIDs = new ArrayList<>();
-    Set<ContainerID> containers =
-        nodeManager.getContainers(dn.getDatanodeDetails());
-    for (ContainerID cid : containers) {
+    TrackedNodeContainers tnc =
+        new TrackedNodeContainers(containerDetailsLoggingLimit);
+    DatanodeDetails details = dn.getDatanodeDetails();
+    Set<ContainerID> containers = nodeManager.getContainers(details);
+
+    handContainersReplicatedOnNode(details, containers, tnc);
+
+    LOG.info("Dn {} has {} sufficientlyReplicated, {} deleting, " +
+        "{} underReplicated and {} unclosed containers", dn,
+        tnc.getSufficientlyReplicated(),
+        tnc.getDeleting(),
+        tnc.getUnderReplicated(),
+        tnc.getUnclosed());
+
+    containerStateByHost.put(details.getHostName(),
+        new ContainerStateInWorkflow(details.getHostName(),
+        tnc.getSufficientlyReplicated(),
+        tnc.getUnderReplicated(),
+        tnc.getUnclosed(),
+        0L, dn.getStartTime()));
+
+    sufficientlyReplicatedContainers += tnc.getSufficientlyReplicated();
+    underReplicatedContainers += tnc.getUnderReplicated();
+    unClosedContainers += tnc.getUnclosed();
+    dn.setContainersReplicatedOnNode(tnc);
+    return tnc.getUnderReplicated() == 0 && tnc.getUnclosed() == 0;
+  }
+
+  private void handContainersReplicatedOnNode(DatanodeDetails details,
+      Set<ContainerID> containers, TrackedNodeContainers tnc) {
+
+    for (ContainerID cid :  containers) {
       try {
         ContainerReplicaCount replicaSet =
             replicationManager.getContainerReplicaCount(cid);
 
         // If a container is deleted or deleting, and we have a replica on this
         // datanode, just ignore it. It should not block decommission.
-        HddsProtos.LifeCycleState containerState
-            = replicaSet.getContainer().getState();
-        if (containerState == HddsProtos.LifeCycleState.DELETED
-            || containerState == HddsProtos.LifeCycleState.DELETING) {
-          deleting++;
+        HddsProtos.LifeCycleState containerState = replicaSet.getContainer().getState();
+        if (containerState == DELETED || containerState == DELETING) {
+          tnc.incrDeleting();
           continue;
         }
 
+        // If the container is unhealthy, we need to add it to the unClosed list.
         boolean isHealthy = replicaSet.isHealthyEnoughForOffline();
         if (!isHealthy) {
-          unClosedIDs.add(cid);
-          if (unclosed < containerDetailsLoggingLimit
-              || LOG.isDebugEnabled()) {
-            LOG.info("Unclosed Container {} {}; {}", cid, replicaSet, replicaDetails(replicaSet.getReplicas()));
+          tnc.addUnClosedIDs(cid);
+          if (tnc.isLogUnClosedContainers() || LOG.isDebugEnabled()) {
+            LOG.info("Unclosed Container {}, ReplicaSet {}, ReplicaDetails {}.",
+                cid, replicaSet, tnc.replicaDetails(replicaSet));
           }
-          unclosed++;
           continue;
         }
 
         // If we get here, the container is closed or quasi-closed and all the replicas match that
         // state, except for any which are unhealthy. As the container is closed, we can check
         // if it is sufficiently replicated using replicationManager, but this only works if the
         // legacy RM is not enabled.
-        boolean legacyEnabled = conf.getBoolean("hdds.scm.replication.enable" +
-            ".legacy", false);
+        boolean legacyEnabled = conf.getBoolean("hdds.scm.replication.enable.legacy", false);
         boolean replicatedOK;
         if (legacyEnabled) {
-          replicatedOK = replicaSet.isSufficientlyReplicatedForOffline(dn.getDatanodeDetails(), nodeManager);
+          replicatedOK = replicaSet.isSufficientlyReplicatedForOffline(details, nodeManager);
         } else {
           ReplicationManagerReport report = new ReplicationManagerReport();
           replicationManager.checkContainerStatus(replicaSet.getContainer(), report);
           replicatedOK = report.getStat(ReplicationManagerReport.HealthState.UNDER_REPLICATED) == 0;
         }
+
         if (replicatedOK) {
-          sufficientlyReplicated++;
+          tnc.incrSufficientlyReplicated();
         } else {
-          underReplicatedIDs.add(cid);
-          if (underReplicated < containerDetailsLoggingLimit || LOG.isDebugEnabled()) {
-            LOG.info("Under Replicated Container {} {}; {}", cid, replicaSet, replicaDetails(replicaSet.getReplicas()));
+          tnc.addUnderReplicated(cid);
+          if (tnc.isLogUnderReplicatedContainers() || LOG.isDebugEnabled()) {
+            LOG.info("Under Replicated Container {}, ReplicaSet {},  ReplicaDetails {}.",
+                cid, replicaSet, tnc.replicaDetails(replicaSet));
           }
-          underReplicated++;
         }
       } catch (ContainerNotFoundException e) {
-        LOG.warn("ContainerID {} present in node list for {} but not found in containerManager", cid,
-            dn.getDatanodeDetails());
+        LOG.warn("ContainerID {} present in node list for {} but not found in containerManager.",
+            cid, details);
       }
     }
-    LOG.info("{} has {} sufficientlyReplicated, {} deleting, {} " +
-            "underReplicated and {} unclosed containers",
-        dn, sufficientlyReplicated, deleting, underReplicated, unclosed);
-    containerStateByHost.put(dn.getDatanodeDetails().getHostName(),
-        new ContainerStateInWorkflow(dn.getDatanodeDetails().getHostName(),
-            sufficientlyReplicated,
-            underReplicated,
-            unclosed,
-            0L, dn.getStartTime()));
-    sufficientlyReplicatedContainers += sufficientlyReplicated;
-    underReplicatedContainers += underReplicated;
-    unClosedContainers += unclosed;
-    if (LOG.isDebugEnabled() && underReplicatedIDs.size() < 10000 &&
-        unClosedIDs.size() < 10000) {
-      LOG.debug("{} has {} underReplicated [{}] and {} unclosed [{}] " +
-              "containers", dn, underReplicated,
-          underReplicatedIDs.stream().map(
-              Object::toString).collect(Collectors.joining(", ")),
-          unclosed, unClosedIDs.stream().map(
-              Object::toString).collect(Collectors.joining(", ")));
+
+    if (LOG.isDebugEnabled() ||
+        (tnc.isLogUnderReplicatedContainers() && tnc.isLogUnClosedContainers())) {
+      LOG.debug("DN {} has {} underReplicated [{}] and {} unclosed [{}] containers", details.getHostName(),
+          tnc.getUnderReplicated(), tnc.getFormatUnderReplicatedIDs(),
+          tnc.getUnclosed(), tnc.getFormatUnClosedIDs());
     }
-    dn.setContainersReplicatedOnNode(underReplicatedIDs, unClosedIDs);
-    return underReplicated == 0 && unclosed == 0;
   }
 
   @Override
@@ -506,16 +466,6 @@ public Map<String, List<ContainerID>> getContainersPendingReplication(DatanodeDe
     return new HashMap<>();
   }
 
-  private String replicaDetails(Collection<ContainerReplica> replicas) {
-    StringBuilder sb = new StringBuilder();
-    sb.append("Replicas{");
-    sb.append(replicas.stream()
-        .map(Object::toString)
-        .collect(Collectors.joining(",")));
-    sb.append("}");
-    return sb.toString();
-  }
-
   private void completeDecommission(DatanodeDetails dn)
       throws NodeNotFoundException {
     setNodeOpState(dn, NodeOperationalState.DECOMMISSIONED);

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/TrackedNode.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.hadoop.hdds.scm.node;
+
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * This class is used to track DataNodes that are
+ * being decommissioned or are in maintenance mode.
+ */
+public class TrackedNode {
+  private DatanodeDetails datanodeDetails;
+  private long startTime;
+  private Map<String, List<ContainerID>> containersReplicatedOnNode = new ConcurrentHashMap<>();
+
+  public TrackedNode(DatanodeDetails datanodeDetails, long startTime) {
+    this.datanodeDetails = datanodeDetails;
+    this.startTime = startTime;
+  }
+
+  @Override
+  public int hashCode() {
+    return datanodeDetails.hashCode();
+  }
+
+  @Override
+  public boolean equals(Object obj) {
+    return obj instanceof TrackedNode &&
+        datanodeDetails.equals(((TrackedNode) obj).getDatanodeDetails());
+  }
+
+  public DatanodeDetails getDatanodeDetails() {
+    return datanodeDetails;
+  }
+
+  public long getStartTime() {
+    return startTime;
+  }
+
+  public Map<String, List<ContainerID>> getContainersReplicatedOnNode() {
+    return containersReplicatedOnNode;
+  }
+
+  public void setContainersReplicatedOnNode(TrackedNodeContainers containers) {
+    List<ContainerID> underReplicated = containers.getUnderReplicatedIDs();
+    List<ContainerID> unClosed = containers.getUnClosedIDs();
+    this.containersReplicatedOnNode.put("UnderReplicated",
+        Collections.unmodifiableList(underReplicated));
+    this.containersReplicatedOnNode.put("UnClosed",
+        Collections.unmodifiableList(unClosed));
+  }
+}