HDFS-16993. Datanode supports configure TopN DatanodeNetworkCounts

Author: admin

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -468,6 +468,10 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final long DFS_DATANODE_PROCESS_COMMANDS_THRESHOLD_DEFAULT =
       TimeUnit.SECONDS.toMillis(2);
 
+  public static final String DFS_DATANODE_NETWORKERRORS_DISPLAY_TOPCOUNT =
+      "dfs.datanode.networkerrors.display.topcount";
+  public static final int DFS_DATANODE_NETWORKERRORS_DISPLAY_TOPCOUNT_DEFAULT = 10;
+
   public static final String DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_KEY = "dfs.namenode.datanode.registration.ip-hostname-check";
   public static final boolean DFS_NAMENODE_DATANODE_REGISTRATION_IP_HOSTNAME_CHECK_DEFAULT = true;
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -117,6 +117,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -128,6 +129,7 @@
 import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -2630,7 +2632,29 @@ public int getActiveTransferThreadCount() {
 
   @Override // DataNodeMXBean
   public Map<String, Map<String, Long>> getDatanodeNetworkCounts() {
-    return datanodeNetworkCounts.asMap();
+    int maxDisplay = getConf().getInt(DFSConfigKeys.DFS_DATANODE_NETWORKERRORS_DISPLAY_TOPCOUNT,
+        DFSConfigKeys.DFS_DATANODE_NETWORKERRORS_DISPLAY_TOPCOUNT_DEFAULT);
+    ConcurrentMap<String, Map<String, Long>> map = datanodeNetworkCounts.asMap();
+    Set<Map.Entry<String, Map<String, Long>>> entries = map.entrySet();
+    List<Map.Entry<String, Map<String, Long>>> list = new ArrayList<>(entries);
+    Collections.sort(list, new Comparator<Entry<String, Map<String, Long>>>() {
+      @Override
+      public int compare(Map.Entry<String, Map<String, Long>> o1,
+          Map.Entry<String, Map<String, Long>> o2) {
+        Map<String, Long> value1Map = o1.getValue();
+        Map<String, Long> value2Map = o2.getValue();
+        long compared = value2Map.getOrDefault(DataNode.NETWORK_ERRORS, 0L) -
+            value1Map.getOrDefault(DataNode.NETWORK_ERRORS, 0L);
+        return (int)compared;
+      }
+    });
+    Map<String, Map<String, Long>> resultMap = new ConcurrentHashMap<>();
+    maxDisplay = list.size() > maxDisplay ? maxDisplay : list.size();
+    for (int i = 0; i < maxDisplay; i++) {
+      resultMap.put(list.get(i).getKey(), list.get(i).getValue());
+    }
+    list.clear();
+    return resultMap;
   }
 
   void incrDatanodeNetworkErrors(String host) {