HDFS-17709. [ARR] Add async responder performance metrics.

Author: hfutatzhanghb

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/protocolPB/AsyncRpcProtocolPBUtil.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.hdfs.protocolPB;
 
+import org.apache.hadoop.hdfs.server.federation.metrics.FederationRPCMetrics;
 import org.apache.hadoop.hdfs.server.federation.router.ThreadLocalContext;
 import org.apache.hadoop.hdfs.server.federation.router.async.utils.ApplyFunction;
 import org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil;
@@ -28,6 +29,7 @@
 import org.apache.hadoop.ipc.ProtobufRpcEngineCallback2;
 import org.apache.hadoop.ipc.internal.ShadedProtobufHelper;
 import org.apache.hadoop.thirdparty.protobuf.Message;
+import org.apache.hadoop.util.Time;
 import org.apache.hadoop.util.concurrent.AsyncGet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -87,6 +89,7 @@ public static <T, R> R asyncIpcClient(
     // transfer thread local context to worker threads of executor.
     ThreadLocalContext threadLocalContext = new ThreadLocalContext();
     asyncCompleteWith(responseFuture.handleAsync((result, e) -> {
+      FederationRPCMetrics.ASYNC_RESPONDER_START_TIME.set(Time.monotonicNow());
       threadLocalContext.transfer();
       if (e != null) {
         throw warpCompletionException(e);
@@ -136,6 +139,7 @@ public static <T> void asyncRouterServer(ServerReq<T> req, ServerRes<T> res) {
       } else {
         callback.error(e.getCause());
       }
+      FederationRPCMetrics.addAsyncResponderThreadTime();
       return null;
     });
   }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCMetrics.java

@@ -30,6 +30,7 @@
 import org.apache.hadoop.metrics2.lib.MetricsRegistry;
 import org.apache.hadoop.metrics2.lib.MutableCounterLong;
 import org.apache.hadoop.metrics2.lib.MutableRate;
+import org.apache.hadoop.util.Time;
 
 /**
  * Implementation of the RPC metrics collector.
@@ -41,9 +42,15 @@ public class FederationRPCMetrics implements FederationRPCMBean {
   private final MetricsRegistry registry = new MetricsRegistry("router");
 
   private RouterRpcServer rpcServer;
+  public static final ThreadLocal<Long> ASYNC_RESPONDER_START_TIME =
+      ThreadLocal.withInitial(() -> -1L);
+  public static final ThreadLocal<Long> ASYNC_RESPONDER_END_TIME =
+      ThreadLocal.withInitial(() -> -1L);
 
   @Metric("Time for the router to process an operation internally")
   private MutableRate processing;
+  @Metric("Time for the router async responder to process an operation internally")
+  private static MutableRate asyncResponderProcessing;
   @Metric("Number of operations the Router processed internally")
   private MutableCounterLong processingOp;
   @Metric("Time for the Router to proxy an operation to the Namenodes")
@@ -302,6 +309,20 @@ public void addProcessingTime(long time) {
     processingOp.incr();
   }
 
+  public static void addAsyncResponderThreadTime() {
+    ASYNC_RESPONDER_END_TIME.set(Time.monotonicNow());
+    long duration = getAsyncResponderProcessingTime();
+    asyncResponderProcessing.add(duration);
+  }
+
+  public static long getAsyncResponderProcessingTime() {
+    if (ASYNC_RESPONDER_START_TIME.get() != null && ASYNC_RESPONDER_START_TIME.get() > 0 &&
+        ASYNC_RESPONDER_END_TIME.get() != null && ASYNC_RESPONDER_END_TIME.get() > 0) {
+      return ASYNC_RESPONDER_END_TIME.get() - ASYNC_RESPONDER_START_TIME.get();
+    }
+    return -1;
+  }
+
   @Override
   public double getProcessingAvg() {
     return processing.lastStat().mean();