HDFS-10645. Make block report size as a metric and add this metric to datanode web ui. Contributed by Yuanbo Liu.

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

@@ -312,6 +312,7 @@ Each metrics record contains tags such as SessionId and Hostname as additional i
 | `SendDataPacketBlockedOnNetworkNanosAvgTime` | Average waiting time of sending packets in nanoseconds |
 | `SendDataPacketTransferNanosNumOps` | Total number of sending packets |
 | `SendDataPacketTransferNanosAvgTime` | Average transfer time of sending packets in nanoseconds |
+| `BPServiceActorInfo` | The information about a block pool service actor |
 
 yarn context
 ============

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

@@ -26,10 +26,13 @@ import java.net.InetSocketAddress;
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -101,6 +104,9 @@ class BPServiceActor implements Runnable {
   private final DataNode dn;
   private final DNConf dnConf;
   private long prevBlockReportId;
+  private final SortedSet<Integer> blockReportSizes =
+      Collections.synchronizedSortedSet(new TreeSet<Integer>());
+  private final int maxDataLength;
 
   private final IncrementalBlockReportManager ibrManager;
 
@@ -122,6 +128,8 @@ class BPServiceActor implements Runnable {
     prevBlockReportId = ThreadLocalRandom.current().nextLong();
     scheduler = new Scheduler(dnConf.heartBeatInterval,
         dnConf.getLifelineIntervalMs(), dnConf.blockReportInterval);
+    // get the value of maxDataLength.
+    this.maxDataLength = dnConf.getMaxDataLength();
   }
 
   public DatanodeRegistration getBpRegistration() {
@@ -166,6 +174,8 @@ class BPServiceActor implements Runnable {
         String.valueOf(getScheduler().getLastHearbeatTime()));
     info.put("LastBlockReport",
         String.valueOf(getScheduler().getLastBlockReportTime()));
+    info.put("maxBlockReportSize", String.valueOf(getMaxBlockReportSize()));
+    info.put("maxDataLength", String.valueOf(maxDataLength));
     return info;
   }
 
@@ -305,6 +315,14 @@ class BPServiceActor implements Runnable {
     }
   }
 
+  private int getMaxBlockReportSize() {
+    int maxBlockReportSize = 0;
+    if (!blockReportSizes.isEmpty()) {
+      maxBlockReportSize = blockReportSizes.last();
+    }
+    return maxBlockReportSize;
+  }
+
   private long generateUniqueBlockReportId() {
     // Initialize the block report ID the first time through.
     // Note that 0 is used on the NN to indicate "uninitialized", so we should
@@ -353,12 +371,18 @@ class BPServiceActor implements Runnable {
     boolean success = false;
     long brSendStartTime = monotonicNow();
     long reportId = generateUniqueBlockReportId();
+    boolean useBlocksBuffer =
+        bpRegistration.getNamespaceInfo().isCapabilitySupported(
+            NamespaceInfo.Capability.STORAGE_BLOCK_REPORT_BUFFERS);
+    blockReportSizes.clear();
     try {
       if (totalBlockCount < dnConf.blockReportSplitThreshold) {
         // Below split threshold, send all reports in a single message.
         DatanodeCommand cmd = bpNamenode.blockReport(
             bpRegistration, bpos.getBlockPoolId(), reports,
               new BlockReportContext(1, 0, reportId, fullBrLeaseId));
+        blockReportSizes.add(
+            calculateBlockReportPBSize(useBlocksBuffer, reports));
         numRPCs = 1;
         numReportsSent = reports.length;
         if (cmd != null) {
@@ -372,6 +396,8 @@ class BPServiceActor implements Runnable {
               bpRegistration, bpos.getBlockPoolId(), singleReport,
               new BlockReportContext(reports.length, r, reportId,
                   fullBrLeaseId));
+          blockReportSizes.add(
+              calculateBlockReportPBSize(useBlocksBuffer, singleReport));
           numReportsSent++;
           numRPCs++;
           if (cmd != null) {
@@ -438,6 +464,21 @@ class BPServiceActor implements Runnable {
     return cmd;
   }
 
+  private int calculateBlockReportPBSize(
+      boolean useBlocksBuffer, StorageBlockReport[] reports) {
+    int reportSize = 0;
+
+    for (StorageBlockReport r : reports) {
+      if (useBlocksBuffer) {
+        reportSize += r.getBlocks().getBlocksBuffer().size();
+      } else {
+        // each block costs 10 bytes in PB because of uint64
+        reportSize += 10 * r.getBlocks().getBlockListAsLongs().length;
+      }
+    }
+    return reportSize;
+  }
+
   HeartbeatResponse sendHeartBeat(boolean requestBlockReportLease)
       throws IOException {
     scheduler.scheduleNextHeartbeat();

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

@@ -116,6 +116,7 @@ public class DNConf {
 
   private final int volFailuresTolerated;
   private final int volsConfigured;
+  private final int maxDataLength;
 
   public DNConf(Configuration conf) {
     this.conf = conf;
@@ -145,6 +146,8 @@ public class DNConf {
     readaheadLength = conf.getLong(
         HdfsClientConfigKeys.DFS_DATANODE_READAHEAD_BYTES_KEY,
         HdfsClientConfigKeys.DFS_DATANODE_READAHEAD_BYTES_DEFAULT);
+    maxDataLength = conf.getInt(DFSConfigKeys.IPC_MAXIMUM_DATA_LENGTH,
+        DFSConfigKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT);
     dropCacheBehindWrites = conf.getBoolean(
         DFSConfigKeys.DFS_DATANODE_DROP_CACHE_BEHIND_WRITES_KEY,
         DFSConfigKeys.DFS_DATANODE_DROP_CACHE_BEHIND_WRITES_DEFAULT);
@@ -354,4 +357,8 @@ public class DNConf {
   public int getVolsConfigured() {
     return volsConfigured;
   }
+
+  int getMaxDataLength() {
+    return maxDataLength;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/webapps/datanode/datanode.html

@@ -80,6 +80,7 @@
       <th>Actor State</th>
       <th>Last Heartbeat</th>
       <th>Last Block Report</th>
+      <th>Last Block Report Size (Max Size)</th>
     </tr>
   </thead>
   {#dn.BPServiceActorInfo}
@@ -89,6 +90,7 @@
       <td>{ActorState}</td>
       <td>{LastHeartbeat}s</td>
       <td>{#helper_relative_time value="{LastBlockReport}"/}</td>
+      <td>{maxBlockReportSize|fmt_bytes} ({maxDataLength|fmt_bytes})</td>
     </tr>
   {/dn.BPServiceActorInfo}
 </table>

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeMXBean.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.server.datanode;
 
 import java.lang.management.ManagementFactory;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -26,22 +27,31 @@ import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
 import com.google.common.base.Supplier;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
 import org.junit.Assert;
 import org.junit.Test;
 import org.mortbay.util.ajax.JSON;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Class for testing {@link DataNodeMXBean} implementation
  */
 public class TestDataNodeMXBean {
+
+  public static final Log LOG = LogFactory.getLog(TestDataNodeMXBean.class);
+
   @Test
   public void testDataNodeMXBean() throws Exception {
     Configuration conf = new Configuration();
@@ -93,6 +103,48 @@ public class TestDataNodeMXBean {
     return s.replaceAll("[0-9]+", "_DIGITS_");
   }
 
+  @Test
+  public void testDataNodeMXBeanBlockSize() throws Exception {
+    Configuration conf = new Configuration();
+
+    try(MiniDFSCluster cluster =
+        new MiniDFSCluster.Builder(conf).build()) {
+      DataNode dn = cluster.getDataNodes().get(0);
+      for (int i = 0; i < 100; i++) {
+        DFSTestUtil.writeFile(
+            cluster.getFileSystem(),
+            new Path("/foo" + String.valueOf(i) + ".txt"), "test content");
+      }
+      DataNodeTestUtils.triggerBlockReport(dn);
+      MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
+      ObjectName mxbeanName = new ObjectName(
+          "Hadoop:service=DataNode,name=DataNodeInfo");
+      String bpActorInfo = (String)mbs.getAttribute(mxbeanName,
+          "BPServiceActorInfo");
+      Assert.assertEquals(dn.getBPServiceActorInfo(), bpActorInfo);
+      LOG.info("bpActorInfo is " + bpActorInfo);
+      TypeReference<ArrayList<Map<String, String>>> typeRef
+          = new TypeReference<ArrayList<Map<String, String>>>() {};
+      ArrayList<Map<String, String>> bpActorInfoList =
+          new ObjectMapper().readValue(bpActorInfo, typeRef);
+      int maxDataLength =
+          Integer.valueOf(bpActorInfoList.get(0).get("maxDataLength"));
+      int confMaxDataLength = dn.getConf().getInt(
+          CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH,
+          CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT);
+      int maxBlockReportSize =
+          Integer.valueOf(bpActorInfoList.get(0).get("maxBlockReportSize"));
+      LOG.info("maxDataLength is " + maxDataLength);
+      LOG.info("maxBlockReportSize is " + maxBlockReportSize);
+      assertTrue("maxBlockReportSize should be greater than zero",
+          maxBlockReportSize > 0);
+      assertEquals("maxDataLength should be exactly "
+          + "the same value of ipc.maximum.data.length",
+          confMaxDataLength,
+          maxDataLength);
+    }
+  }
+
   @Test
   public void testDataNodeMXBeanBlockCount() throws Exception {
     Configuration conf = new Configuration();