HADOOP-18710. Add RPC metrics for response time (#5545). Contributed by liuguanghua.

Reviewed-by: Inigo Goiri <inigoiri@apache.org>
Signed-off-by: Ayush Saxena <ayushsaxena@apache.org>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -627,8 +627,11 @@ public abstract class Server {
         details.get(Timing.PROCESSING, rpcMetrics.getMetricsTimeUnit());
     long waitTime =
         details.get(Timing.LOCKWAIT, rpcMetrics.getMetricsTimeUnit());
+    long responseTime =
+        details.get(Timing.RESPONSE, rpcMetrics.getMetricsTimeUnit());
     rpcMetrics.addRpcLockWaitTime(waitTime);
     rpcMetrics.addRpcProcessingTime(processingTime);
+    rpcMetrics.addRpcResponseTime(responseTime);
     // don't include lock wait for detailed metrics.
     processingTime -= waitTime;
     String name = call.getDetailedMetricsName();

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcMetrics.java

@@ -75,6 +75,8 @@ public class RpcMetrics {
           new MutableQuantiles[intervals.length];
       rpcProcessingTimeQuantiles =
           new MutableQuantiles[intervals.length];
+      rpcResponseTimeQuantiles =
+          new MutableQuantiles[intervals.length];
       deferredRpcProcessingTimeQuantiles =
           new MutableQuantiles[intervals.length];
       for (int i = 0; i < intervals.length; i++) {
@@ -90,6 +92,10 @@ public class RpcMetrics {
             "rpcProcessingTime" + interval + "s",
             "rpc processing time in " + metricsTimeUnit, "ops",
             "latency", interval);
+        rpcResponseTimeQuantiles[i] = registry.newQuantiles(
+            "rpcResponseTime" + interval + "s",
+            "rpc response time in " + metricsTimeUnit, "ops",
+            "latency", interval);
         deferredRpcProcessingTimeQuantiles[i] = registry.newQuantiles(
             "deferredRpcProcessingTime" + interval + "s",
             "deferred rpc processing time in " + metricsTimeUnit, "ops",
@@ -114,6 +120,8 @@ public class RpcMetrics {
   MutableQuantiles[] rpcLockWaitTimeQuantiles;
   @Metric("Processing time") MutableRate rpcProcessingTime;
   MutableQuantiles[] rpcProcessingTimeQuantiles;
+  @Metric("Response time") MutableRate rpcResponseTime;
+  MutableQuantiles[] rpcResponseTimeQuantiles;
   @Metric("Deferred Processing time") MutableRate deferredRpcProcessingTime;
   MutableQuantiles[] deferredRpcProcessingTimeQuantiles;
   @Metric("Number of authentication failures")
@@ -282,6 +290,15 @@ public class RpcMetrics {
     }
   }
 
+  public void addRpcResponseTime(long responseTime) {
+    rpcResponseTime.add(responseTime);
+    if (rpcQuantileEnable) {
+      for (MutableQuantiles q : rpcResponseTimeQuantiles) {
+        q.add(responseTime);
+      }
+    }
+  }
+
   public void addDeferredRpcProcessingTime(long processingTime) {
     deferredRpcProcessingTime.add(processingTime);
     if (rpcQuantileEnable) {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPC.java

@@ -1336,12 +1336,16 @@ public class TestRPC extends TestRpcBase {
           3000, getLongCounter("RpcProcessingTimeNumOps", rpcMetrics));
       assertEquals("Expected correct rpc lock wait count",
           3000, getLongCounter("RpcLockWaitTimeNumOps", rpcMetrics));
+      assertEquals("Expected correct rpc response count",
+          3000, getLongCounter("RpcResponseTimeNumOps", rpcMetrics));
       assertEquals("Expected zero rpc lock wait time",
           0, getDoubleGauge("RpcLockWaitTimeAvgTime", rpcMetrics), 0.001);
       MetricsAsserts.assertQuantileGauges("RpcQueueTime" + interval + "s",
           rpcMetrics);
       MetricsAsserts.assertQuantileGauges("RpcProcessingTime" + interval + "s",
           rpcMetrics);
+      MetricsAsserts.assertQuantileGauges("RpcResponseTime" + interval + "s",
+          rpcMetrics);
       String actualUserVsCon = MetricsAsserts
           .getStringMetric("NumOpenConnectionsPerUser", rpcMetrics);
       String proxyUser =