HDFS-16917 Add transfer rate quantile metrics for DataNode reads

Author: Ravindra Dingankar

hadoop-common-project/hadoop-common/src/site/markdown/Metrics.md

@@ -370,6 +370,7 @@ Each metrics record contains tags such as SessionId and Hostname as additional i
 |:---- |:---- |
 | `BytesWritten` | Total number of bytes written to DataNode |
 | `BytesRead` | Total number of bytes read from DataNode |
+| `BytesReadTransferRate`*num*`s(50/75/90/95/99)thPercentileTransferRate` | The 50/75/90/95/99th percentile of the transfer rate of bytes read from the Data Node. The transfer rate is measured in bytes per millisecond. |
 | `BlocksWritten` | Total number of blocks written to DataNode |
 | `BlocksRead` | Total number of blocks read from DataNode |
 | `BlocksReplicated` | Total number of blocks replicated |

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

@@ -632,6 +632,9 @@ public void readBlock(final ExtendedBlock block,
       datanode.metrics.incrBytesRead((int) read);
       datanode.metrics.incrBlocksRead();
       datanode.metrics.incrTotalReadTime(duration);
+      if (duration != 0) {
+          datanode.metrics.addBytesReadTransferRate(read / duration);
+      }
     } catch ( SocketException ignored ) {
       LOG.trace("{}:Ignoring exception while serving {} to {}",
           dnR, block, remoteAddress, ignored);
@@ -1122,6 +1125,9 @@ public void copyBlock(final ExtendedBlock block,
       datanode.metrics.incrBytesRead((int) read);
       datanode.metrics.incrBlocksRead();
       datanode.metrics.incrTotalReadTime(duration);
+      if (duration != 0) {
+          datanode.metrics.addBytesReadTransferRate(read / duration);
+      }
 
       LOG.info("Copied {} to {}", block, peer.getRemoteAddressString());
     } catch (IOException ioe) {

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

@@ -61,6 +61,8 @@ public class DataNodeMetrics {
   @Metric MutableCounterLong bytesRead;
   @Metric("Milliseconds spent reading")
   MutableCounterLong totalReadTime;
+  @Metric MutableRate bytesReadTransferRate;
+  final MutableQuantiles[] bytesReadTransferRateQuantiles;
   @Metric MutableCounterLong blocksWritten;
   @Metric MutableCounterLong blocksRead;
   @Metric MutableCounterLong blocksReplicated;
@@ -227,6 +229,7 @@ public DataNodeMetrics(String name, String sessionId, int[] intervals,
     sendDataPacketTransferNanosQuantiles = new MutableQuantiles[len];
     ramDiskBlocksEvictionWindowMsQuantiles = new MutableQuantiles[len];
     ramDiskBlocksLazyPersistWindowMsQuantiles = new MutableQuantiles[len];
+    bytesReadTransferRateQuantiles = new MutableQuantiles[len];
 
     for (int i = 0; i < len; i++) {
       int interval = intervals[i];
@@ -255,6 +258,10 @@ public DataNodeMetrics(String name, String sessionId, int[] intervals,
           "ramDiskBlocksLazyPersistWindows" + interval + "s",
           "Time between the RamDisk block write and disk persist in ms",
           "ops", "latency", interval);
+      bytesReadTransferRateQuantiles[i] = registry.newQuantiles(
+                "bytesReadTransferRate" + interval + "s",
+                "Rate at which bytes are read from Data Node",
+                "ops", "transfer rate", interval);
     }
   }
 
@@ -315,6 +322,13 @@ public void addIncrementalBlockReport(long latency,
       nnRpcLatency.add("IncrementalBlockReportsFor" + rpcMetricSuffix, latency);
     }
   }
+  
+  public void addBytesReadTransferRate(long latencyMs) {
+      bytesReadTransferRate.add(latencyMs);
+      for (MutableQuantiles q : bytesReadTransferRateQuantiles) {
+          q.add(latencyMs);
+      }
+  }
 
   public void addCacheReport(long latency) {
     cacheReports.add(latency);