HDFS-16582. Expose aggregate latency of slow node as perceived by the reporting node

Author: virajjasani

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

@@ -1901,8 +1901,9 @@ public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
         if (LOG.isDebugEnabled()) {
           LOG.debug("DataNode " + nodeReg + " reported slow peers: " + slowPeersMap);
         }
-        for (String slowNodeId : slowPeersMap.keySet()) {
-          slowPeerTracker.addReport(slowNodeId, nodeReg.getIpcAddr(false));
+        for (Map.Entry<String, Double> slowNodeId : slowPeersMap.entrySet()) {
+          slowPeerTracker.addReport(slowNodeId.getKey(), nodeReg.getIpcAddr(false),
+              slowNodeId.getValue());
         }
       }
     }

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

@@ -58,20 +58,20 @@ public boolean isSlowPeerTrackerEnabled() {
   }
 
   @Override
-  public void addReport(String slowNode, String reportingNode) {
+  public void addReport(String slowNode, String reportingNode, Double slowNodeLatency) {
     LOG.trace("Adding slow peer report is disabled. To enable it, please enable config {}.",
         DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY);
   }
 
   @Override
-  public Set<String> getReportsForNode(String slowNode) {
+  public Set<SlowPeerLatencyWithReportingNode> getReportsForNode(String slowNode) {
     LOG.trace("Retrieval of slow peer report is disabled. To enable it, please enable config {}.",
         DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY);
     return ImmutableSet.of();
   }
 
   @Override
-  public Map<String, SortedSet<String>> getReportsForAllDataNodes() {
+  public Map<String, SortedSet<SlowPeerLatencyWithReportingNode>> getReportsForAllDataNodes() {
     LOG.trace("Retrieval of slow peer report for all nodes is disabled. "
             + "To enable it, please enable config {}.",
         DFSConfigKeys.DFS_DATANODE_PEER_STATS_ENABLED_KEY);

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

@@ -0,0 +1,84 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hdfs.server.blockmanagement;
+
+import java.util.SortedSet;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * This structure is a thin wrapper over slow peer reports to make Json
+ * [de]serialization easy.
+ */
+@InterfaceAudience.Private
+final class SlowPeerJsonReport {
+
+  @JsonProperty("SlowNode")
+  private final String slowNode;
+
+  @JsonProperty("SlowPeerLatencyWithReportingNodes")
+  private final SortedSet<SlowPeerLatencyWithReportingNode> slowPeerLatencyWithReportingNodes;
+
+  public SlowPeerJsonReport(
+      @JsonProperty("SlowNode")
+          String slowNode,
+      @JsonProperty("SlowPeerLatencyWithReportingNodes")
+          SortedSet<SlowPeerLatencyWithReportingNode> slowPeerLatencyWithReportingNodes) {
+    this.slowNode = slowNode;
+    this.slowPeerLatencyWithReportingNodes = slowPeerLatencyWithReportingNodes;
+  }
+
+  public String getSlowNode() {
+    return slowNode;
+  }
+
+  public SortedSet<SlowPeerLatencyWithReportingNode> getSlowPeerLatencyWithReportingNodes() {
+    return slowPeerLatencyWithReportingNodes;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    SlowPeerJsonReport that = (SlowPeerJsonReport) o;
+
+    return new EqualsBuilder()
+        .append(slowNode, that.slowNode)
+        .append(slowPeerLatencyWithReportingNodes, that.slowPeerLatencyWithReportingNodes)
+        .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+        .append(slowNode)
+        .append(slowPeerLatencyWithReportingNodes)
+        .toHashCode();
+  }
+}

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

@@ -0,0 +1,88 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.hdfs.server.blockmanagement;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * This class represents the reporting node and the slow node's latency as observed by the
+ * reporting node. This class is used by SlowPeerJsonReport class.
+ */
+@InterfaceAudience.Private
+final class SlowPeerLatencyWithReportingNode
+    implements Comparable<SlowPeerLatencyWithReportingNode> {
+
+  @JsonProperty("ReportingNode")
+  private final String reportingNode;
+
+  @JsonProperty("ReportedLatency")
+  private final Double reportedLatency;
+
+  public SlowPeerLatencyWithReportingNode(
+      @JsonProperty("ReportingNode")
+          String reportingNode,
+      @JsonProperty("ReportedLatency")
+          Double reportedLatency) {
+    this.reportingNode = reportingNode;
+    this.reportedLatency = reportedLatency;
+  }
+
+  public String getReportingNode() {
+    return reportingNode;
+  }
+
+  public Double getReportedLatency() {
+    return reportedLatency;
+  }
+
+  @Override
+  public int compareTo(SlowPeerLatencyWithReportingNode o) {
+    return this.reportingNode.compareTo(o.getReportingNode());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    SlowPeerLatencyWithReportingNode that = (SlowPeerLatencyWithReportingNode) o;
+
+    return new EqualsBuilder()
+        .append(reportingNode, that.reportingNode)
+        .append(reportedLatency, that.reportedLatency)
+        .isEquals();
+  }
+
+  @Override
+  public int hashCode() {
+    return new HashCodeBuilder(17, 37)
+        .append(reportingNode)
+        .append(reportedLatency)
+        .toHashCode();
+  }
+}