HADOOP-2886. Track individual RPC metrics.

(girish vaitheeswaran via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@639359 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -126,6 +126,9 @@ Trunk (unreleased changes)
     HADOOP-2899. [HOD] Cleans up hdfs:///mapredsystem directory after
     deallocation. (Hemanth Yamijala via ddas) 
 
+    HADOOP-2886.  Track individual RPC metrics.
+    (girish vaitheeswaran via dhruba)
+
   OPTIMIZATIONS
 
     HADOOP-2790.  Fixed inefficient method hasSpeculativeTask by removing

src/java/org/apache/hadoop/ipc/RPC.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.io.*;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.conf.*;
+import org.apache.hadoop.metrics.util.MetricsTimeVaryingRate;
 
 /** A simple RPC mechanism.
  *
@@ -450,7 +451,18 @@ public class RPC {
             " procesingTime= " + processingTime);
         rpcMetrics.rpcQueueTime.inc(qTime);
         rpcMetrics.rpcProcessingTime.inc(processingTime);
-        
+
+	MetricsTimeVaryingRate m = rpcMetrics.metricsList.get(call.getMethodName());
+
+	if (m != null) {
+		m.inc(processingTime);
+	}
+	else {
+		rpcMetrics.metricsList.put(call.getMethodName(), new MetricsTimeVaryingRate(call.getMethodName()));
+		m = rpcMetrics.metricsList.get(call.getMethodName());
+		m.inc(processingTime);
+	}
+
         if (verbose) log("Return: "+value);
 
         return new ObjectWritable(method.getReturnType(), value);

src/java/org/apache/hadoop/ipc/metrics/RpcMetrics.java

@@ -27,6 +27,9 @@ import org.apache.hadoop.metrics.Updater;
 import org.apache.hadoop.metrics.jvm.JvmMetrics;
 import org.apache.hadoop.metrics.util.MetricsTimeVaryingRate;
 
+import java.util.HashMap;
+import java.util.*;
+
 /**
  * 
  * This class is for maintaining  the various RPC statistics
@@ -69,6 +72,9 @@ public class RpcMetrics implements Updater {
   public MetricsTimeVaryingRate rpcQueueTime = new MetricsTimeVaryingRate("RpcQueueTime");
   public MetricsTimeVaryingRate rpcProcessingTime = new MetricsTimeVaryingRate("RpcProcessingTime");
   public MetricsTimeVaryingRate rpcDiscardedOps = new MetricsTimeVaryingRate("RpcDiscardedOps");
+
+  public Map <String, MetricsTimeVaryingRate> metricsList = Collections.synchronizedMap(new HashMap<String, MetricsTimeVaryingRate>());
+
   
   
   /**
@@ -78,7 +84,20 @@ public class RpcMetrics implements Updater {
     rpcQueueTime.pushMetric(metricsRecord);
     rpcProcessingTime.pushMetric(metricsRecord);
     rpcDiscardedOps.pushMetric(metricsRecord);
-    
+
+    synchronized (metricsList) {
+	// Iterate through the rpcMetrics hashmap to propogate the different rpc metrics.
+	Set keys = metricsList.keySet();
+
+	Iterator keyIter = keys.iterator();
+
+	while (keyIter.hasNext()) {
+		Object key = keyIter.next();
+		MetricsTimeVaryingRate value = metricsList.get(key);
+
+		value.pushMetric(metricsRecord);
+	}
+    }
   }
 
   public void shutdown() {