SPARK-18364 Add relevant measurements for Timer metrics (apache#236)

Author: ash211

common/network-yarn/src/main/java/org/apache/spark/network/yarn/YarnShuffleServiceMetrics.java

@@ -19,18 +19,27 @@
 
 import com.codahale.metrics.*;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 import org.apache.hadoop.metrics2.MetricsCollector;
 import org.apache.hadoop.metrics2.MetricsInfo;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
 import org.apache.hadoop.metrics2.MetricsSource;
 
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Modeled off of YARN's NodeManagerMetrics.
  */
 public class YarnShuffleServiceMetrics implements MetricsSource {
 
+  // Converting from the dropwizard-metrics default of nanoseconds into milliseconds to match how
+  // MetricsServlet serializes times (to milliseconds) configured via the MetricsModule passed into
+  // its Jackson ObjectMapper. Without this rate factor applied, the Timer metrics from
+  // ExternalShuffleBlockManager#ShuffleMetrics with "Millis" suffixes are misleading, as they
+  // would otherwise contain values in nanoseconds units
+  private static final double rateFactor = (double) TimeUnit.MILLISECONDS.toNanos(1L);
+
   private final MetricSet metricSet;
 
   public YarnShuffleServiceMetrics(MetricSet metricSet) {
@@ -52,13 +61,50 @@ public void getMetrics(MetricsCollector collector, boolean all) {
     }
   }
 
+  private static void addSnapshotToMetricRecordBuilder(Snapshot snapshot,
+                                                       MetricsRecordBuilder builder,
+                                                       String name,
+                                                       String metricType) {
+
+    ImmutableMap<String, Double> doubleValues = ImmutableMap.<String, Double>builder()
+      .put("median", snapshot.getMedian())
+      .put("mean", snapshot.getMean())
+      .put("75th", snapshot.get75thPercentile())
+      .put("95th", snapshot.get95thPercentile())
+      .put("98th", snapshot.get98thPercentile())
+      .put("99th", snapshot.get99thPercentile())
+      .put("999th", snapshot.get999thPercentile())
+      .build();
+
+    ImmutableMap<String, Long> longValues = ImmutableMap.<String, Long>builder()
+      .put("min", snapshot.getMin())
+      .put("max", snapshot.getMax())
+      .build();
+
+    for (Map.Entry<String, Double> entry : doubleValues.entrySet()) {
+      builder.addGauge(
+        new ShuffleServiceMetricsInfo(name + "_" + entry.getKey(),
+          entry.getKey() + " of " + metricType + " " + name),
+        entry.getValue() / rateFactor);
+    }
+
+    for (Map.Entry<String, Long> entry : longValues.entrySet()) {
+      builder.addGauge(
+        new ShuffleServiceMetricsInfo(name + "_" + entry.getKey(),
+          entry.getKey() + " of " + metricType + " " + name),
+        entry.getValue() / rateFactor);
+    }
+
+  }
+
   @VisibleForTesting
   public static void collectMetric(
           MetricsRecordBuilder metricsRecordBuilder, String name, Metric metric) {
 
     // The metric types used in ExternalShuffleBlockHandler.ShuffleMetrics
     if (metric instanceof Timer) {
       Timer t = (Timer) metric;
+      Snapshot snapshot = t.getSnapshot();
       metricsRecordBuilder
         .addCounter(new ShuffleServiceMetricsInfo(name + "_count", "Count of timer " + name),
           t.getCount())
@@ -73,6 +119,7 @@ public static void collectMetric(
           t.getOneMinuteRate())
         .addGauge(new ShuffleServiceMetricsInfo(name + "_rateMean", "Mean rate of timer " + name),
           t.getMeanRate());
+      addSnapshotToMetricRecordBuilder(snapshot, metricsRecordBuilder, name, "timer");
     } else if (metric instanceof Meter) {
       Meter m = (Meter) metric;
       metricsRecordBuilder

resource-managers/yarn/src/test/scala/org/apache/spark/network/yarn/YarnShuffleServiceMetricsSuite.scala

@@ -16,7 +16,11 @@
  */
 package org.apache.spark.network.yarn
 
-import org.apache.hadoop.metrics2.MetricsRecordBuilder
+import java.util.concurrent.TimeUnit
+
+import com.codahale.metrics.Timer
+import org.apache.hadoop.metrics2.{MetricsInfo, MetricsRecordBuilder}
+import org.mockito.ArgumentCaptor
 import org.mockito.Matchers._
 import org.mockito.Mockito.{mock, times, verify, when}
 import org.scalatest.Matchers
@@ -44,8 +48,7 @@ class YarnShuffleServiceMetricsSuite extends SparkFunSuite with Matchers {
 
   // these three metrics have the same effect on the collector
   for (testname <- Seq("openBlockRequestLatencyMillis",
-      "registerExecutorRequestLatencyMillis",
-      "blockTransferRateBytes")) {
+      "registerExecutorRequestLatencyMillis")) {
     test(s"$testname - collector receives correct types") {
       val builder = mock(classOf[MetricsRecordBuilder])
       when(builder.addCounter(any(), anyLong())).thenReturn(builder)
@@ -55,10 +58,22 @@ class YarnShuffleServiceMetricsSuite extends SparkFunSuite with Matchers {
         metrics.getMetrics.get(testname))
 
       verify(builder).addCounter(anyObject(), anyLong())
-      verify(builder, times(4)).addGauge(anyObject(), anyDouble())
+      verify(builder, times(13)).addGauge(anyObject(), anyDouble())
     }
   }
 
+  test(s"blockTransferRateBytes - collector receives correct types") {
+    val builder = mock(classOf[MetricsRecordBuilder])
+    when(builder.addCounter(any(), anyLong())).thenReturn(builder)
+    when(builder.addGauge(any(), anyDouble())).thenReturn(builder)
+
+    YarnShuffleServiceMetrics.collectMetric(builder, "blockTransferRateBytes",
+      metrics.getMetrics.get("blockTransferRateBytes"))
+
+    verify(builder).addCounter(anyObject(), anyLong())
+    verify(builder, times(4)).addGauge(anyObject(), anyDouble())
+  }
+
   // this metric writes only one gauge to the collector
   test("registeredExecutorsSize - collector receives correct types") {
     val builder = mock(classOf[MetricsRecordBuilder])
@@ -71,4 +86,28 @@ class YarnShuffleServiceMetricsSuite extends SparkFunSuite with Matchers {
     // only one
     verify(builder).addGauge(anyObject(), anyInt())
   }
+
+  test("openBlockRequestLatencyMillis has correct units") {
+    val builder = mock(classOf[MetricsRecordBuilder])
+    when(builder.addCounter(any(), anyLong())).thenReturn(builder)
+    when(builder.addGauge(any(), anyDouble())).thenReturn(builder)
+
+    metrics.getMetrics.get("openBlockRequestLatencyMillis").asInstanceOf[Timer]
+      .update(1, TimeUnit.MILLISECONDS)
+
+    YarnShuffleServiceMetrics.collectMetric(builder, "openBlockRequestLatencyMillis",
+      metrics.getMetrics.get("openBlockRequestLatencyMillis"))
+
+    val descriptorCaptor = ArgumentCaptor.forClass(classOf[MetricsInfo])
+    val doubleCaptor = ArgumentCaptor.forClass(classOf[java.lang.Double])
+    verify(builder, times(13)).addGauge(descriptorCaptor.capture(), doubleCaptor.capture())
+
+    descriptorCaptor.getAllValues.asScala.zip(doubleCaptor.getAllValues.asScala).foreach {
+      case (arg: MetricsInfo, double: java.lang.Double) =>
+        arg.name() match {
+          case s if !s.contains("rate") => double shouldEqual 1.0 // still in milliseconds
+          case _ => None
+        }
+    }
+  }
 }