HDFS-17292. Show the number of times the slowPeerCollectorDaemon thread has collected SlowNodes.

Author: huangzhaobo99

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/NamenodeBeanMetrics.java

@@ -889,6 +889,11 @@ public float getReconstructionQueuesInitProgress() {
     return 0;
   }
 
+  @Override
+  public String getCollectSlowNodesIpAddrFrequencyMap() {
+    return "N/A";
+  }
+
   private Router getRouter() throws IOException {
     if (this.router == null) {
       throw new IOException("Router is not initialized");

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java

@@ -69,6 +69,7 @@
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
@@ -214,6 +215,8 @@ public class DatanodeManager {
   private final long slowPeerCollectionInterval;
   private volatile int maxSlowPeerReportNodes;
 
+  private final Map<String, Integer> collectSlowNodesIpAddrFrequencyMap = new ConcurrentHashMap<>();
+
   @Nullable
   private final SlowDiskTracker slowDiskTracker;
 
@@ -421,6 +424,7 @@ public void stopSlowPeerCollector() {
     } finally {
       slowPeerCollectorDaemon = null;
       slowNodesUuidSet.clear();
+      collectSlowNodesIpAddrFrequencyMap.clear();
     }
   }
 
@@ -2193,8 +2197,23 @@ public Set<String> getSlowPeersUuidSet() {
     Preconditions.checkNotNull(slowPeerTracker, "slowPeerTracker should not be un-assigned");
     slowNodes = slowPeerTracker.getSlowNodes(maxSlowPeerReportNodes);
     List<DatanodeDescriptor> datanodeDescriptors = getDnDescriptorsFromIpAddr(slowNodes);
-    datanodeDescriptors.forEach(
-        datanodeDescriptor -> slowPeersUuidSet.add(datanodeDescriptor.getDatanodeUuid()));
+    datanodeDescriptors.forEach(datanodeDescriptor -> {
+      slowPeersUuidSet.add(datanodeDescriptor.getDatanodeUuid());
+      String ipAddr = datanodeDescriptor.getIpAddr();
+      if (!collectSlowNodesIpAddrFrequencyMap.containsKey(ipAddr)) {
+        collectSlowNodesIpAddrFrequencyMap.put(ipAddr, 1);
+      } else {
+        collectSlowNodesIpAddrFrequencyMap.put(ipAddr,
+            collectSlowNodesIpAddrFrequencyMap.get(ipAddr) + 1);
+      }
+    });
+    // Clean up nodes that are not in the host2DatanodeMap
+    for (String ipAddr : collectSlowNodesIpAddrFrequencyMap.keySet()) {
+      DatanodeDescriptor datanodeByHost = host2DatanodeMap.getDatanodeByHost(ipAddr);
+      if (datanodeByHost == null) {
+        collectSlowNodesIpAddrFrequencyMap.remove(ipAddr);
+      }
+    }
     return slowPeersUuidSet;
   }
 
@@ -2222,6 +2241,10 @@ public static Set<String> getSlowNodesUuidSet() {
     return slowNodesUuidSet;
   }
 
+  public Map<String, Integer> getCollectSlowNodesIpAddrFrequencyMap() {
+    return collectSlowNodesIpAddrFrequencyMap;
+  }
+
   /**
    * Use only for testing.
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -4920,6 +4920,14 @@ public float getReconstructionQueuesInitProgress() {
     return blockManager.getReconstructionQueuesInitProgress();
   }
 
+  @Override // FSNamesystemMBean
+  @Metric
+  public String getCollectSlowNodesIpAddrFrequencyMap() {
+    Map<String, Integer> recordSlowNodesIpAddr =
+        getBlockManager().getDatanodeManager().getCollectSlowNodesIpAddrFrequencyMap();
+    return JSON.toString(recordSlowNodesIpAddr);
+  }
+
   /**
    * Returns the length of the wait Queue for the FSNameSystemLock.
    *

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/metrics/FSNamesystemMBean.java

@@ -268,4 +268,13 @@ public interface FSNamesystemMBean {
    * @return Returns values between 0 and 1 for the progress.
    */
   float getReconstructionQueuesInitProgress();
+
+
+  /**
+   * Returns a nested JSON object listing the collect slowNodesIpAddr frequency Map,
+   * e.g. {"1.1.1.1":4,"2.2.2.2":3}
+   *
+   * @return JSON string.
+   */
+  String getCollectSlowNodesIpAddrFrequencyMap();
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/blockmanagement/TestReplicationPolicyExcludeSlowNodes.java

@@ -21,22 +21,31 @@
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.TestBlockStoragePolicy;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.protocol.OutlierMetrics;
 
 import org.apache.hadoop.test.GenericTestUtils;
+import org.eclipse.jetty.util.ajax.JSON;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import java.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
 
 @RunWith(Parameterized.class)
@@ -176,4 +185,58 @@ public void testSlowPeerTrackerEnabledClearSlowNodes() throws Exception {
     }
   }
 
+  /**
+   * Dependent on the SlowNode related config, therefore placing
+   * 'testCollectSlowNodesIpAddrFrequencyMetrics' unit test in the
+   * TestReplicationPolicyExcludeSlowNodes class.
+   * <br>
+   * Test metrics associated with CollectSlowNodesIpAddrFrequency.
+   */
+  @Test
+  public void testCollectSlowNodesIpAddrFrequencyMetrics() throws Exception {
+    namenode.getNamesystem().writeLock();
+    try {
+      FSNamesystem fsNamesystem = namenode.getNamesystem();
+      assertEquals("{}", fsNamesystem.getCollectSlowNodesIpAddrFrequencyMap(), "{}");
+      MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
+      ObjectName mxBeanName = new ObjectName("Hadoop:service=NameNode,name=FSNamesystemState");
+      String ipAddrFrequency =
+          (String) mBeanServer.getAttribute(mxBeanName, "CollectSlowNodesIpAddrFrequencyMap");
+      assertEquals("{}", ipAddrFrequency, "{}");
+
+      // add nodes
+      for (DatanodeDescriptor dataNode : dataNodes) {
+        dnManager.addDatanode(dataNode);
+      }
+
+      // mock slow nodes
+      SlowPeerTracker tracker = dnManager.getSlowPeerTracker();
+      Assert.assertNotNull(tracker);
+      OutlierMetrics outlierMetrics = new OutlierMetrics(0.0, 0.0, 0.0, 5.0);
+      tracker.addReport(dataNodes[0].getInfoAddr(), dataNodes[2].getInfoAddr(), outlierMetrics);
+      tracker.addReport(dataNodes[1].getInfoAddr(), dataNodes[2].getInfoAddr(), outlierMetrics);
+
+      // waiting for slow nodes collector run and collect at least 2 times
+      Thread.sleep(3000);
+      assertNotEquals("{}", fsNamesystem.getCollectSlowNodesIpAddrFrequencyMap(), "{}");
+
+      // check jmx data
+      ipAddrFrequency =
+          (String) mBeanServer.getAttribute(mxBeanName, "CollectSlowNodesIpAddrFrequencyMap");
+      assertNotEquals("{}", ipAddrFrequency, "{}");
+      Map<String, Long> ipAddrFrequencyMap = (HashMap) JSON.parse(ipAddrFrequency);
+      for (Map.Entry<String, Long> entry : ipAddrFrequencyMap.entrySet()) {
+        assertTrue(dataNodes[0].getIpAddr().equals(entry.getKey())
+            || dataNodes[1].getIpAddr().equals(entry.getKey()));
+        assertTrue(entry.getValue() > 1);
+      }
+
+      // check reconfig
+      namenode.reconfigureProperty(DFS_DATANODE_PEER_STATS_ENABLED_KEY, "false");
+      assertEquals("{}", dnManager.getCollectSlowNodesIpAddrFrequencyMap().toString());
+    } finally {
+      namenode.getNamesystem().writeUnlock();
+    }
+  }
+
 }