[SPARK-18364][YARN] Expose metrics for YarnShuffleService

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

@@ -176,7 +176,8 @@ private void checkAuth(TransportClient client, String appId) {
   /**
    * A simple class to wrap all shuffle service wrapper metrics
    */
-  private class ShuffleMetrics implements MetricSet {
+  @VisibleForTesting
+  public class ShuffleMetrics implements MetricSet {
     private final Map<String, Metric> allMetrics;
     // Time latency for open block request in ms
     private final Timer openBlockRequestLatencyMillis = new Timer();
@@ -185,7 +186,7 @@ private class ShuffleMetrics implements MetricSet {
     // Block transfer rate in byte per second
     private final Meter blockTransferRateBytes = new Meter();
 
-    private ShuffleMetrics() {
+    public ShuffleMetrics() {
       allMetrics = new HashMap<>();
       allMetrics.put("openBlockRequestLatencyMillis", openBlockRequestLatencyMillis);
       allMetrics.put("registerExecutorRequestLatencyMillis", registerExecutorRequestLatencyMillis);

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

@@ -19,8 +19,9 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
+import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.util.List;
 import java.util.Map;
@@ -35,9 +36,11 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.impl.MetricsSystemImpl;
+import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.server.api.*;
-import org.apache.spark.network.util.LevelDBProvider;
 import org.iq80.leveldb.DB;
 import org.iq80.leveldb.DBIterator;
 
@@ -50,6 +53,7 @@
 import org.apache.spark.network.server.TransportServer;
 import org.apache.spark.network.server.TransportServerBootstrap;
 import org.apache.spark.network.shuffle.ExternalShuffleBlockHandler;
+import org.apache.spark.network.util.LevelDBProvider;
 import org.apache.spark.network.util.TransportConf;
 import org.apache.spark.network.yarn.util.HadoopConfigProvider;
 
@@ -166,6 +170,23 @@ protected void serviceInit(Configuration conf) throws Exception {
       TransportConf transportConf = new TransportConf("shuffle", new HadoopConfigProvider(conf));
       blockHandler = new ExternalShuffleBlockHandler(transportConf, registeredExecutorFile);
 
+      // register metrics on the block handler into the Node Manager's metrics system.
+      try {
+        YarnShuffleServiceMetrics serviceMetrics = new YarnShuffleServiceMetrics(
+          blockHandler.getAllMetrics());
+        MetricsSystemImpl metricsSystem = (MetricsSystemImpl) DefaultMetricsSystem.instance();
+
+        Method registerSourceMethod = metricsSystem.getClass().getDeclaredMethod("registerSource",
+                String.class, String.class, MetricsSource.class);
+        registerSourceMethod.setAccessible(true);
+        registerSourceMethod.invoke(metricsSystem, "shuffleService", "Metrics on the Spark " +
+                "Shuffle Service", serviceMetrics);
+        logger.info("Registered metrics with Hadoop's DefaultMetricsSystem");
+      } catch (Exception e) {
+        logger.warn("Unable to register Spark Shuffle Service metrics with Node Manager; " +
+                "proceeding without metrics", e);
+      }
+
       // If authentication is enabled, set up the shuffle server to use a
       // special RPC handler that filters out unauthenticated fetch requests
       List<TransportServerBootstrap> bootstraps = Lists.newArrayList();

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

@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.yarn;
+
+import com.codahale.metrics.*;
+import com.google.common.annotations.VisibleForTesting;
+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;
+
+/**
+ * Modeled off of YARN's NodeManagerMetrics.
+ */
+public class YarnShuffleServiceMetrics implements MetricsSource {
+
+  private final MetricSet metricSet;
+
+  public YarnShuffleServiceMetrics(MetricSet metricSet) {
+    this.metricSet = metricSet;
+  }
+
+  /**
+   * Get metrics from the source
+   *
+   * @param collector to contain the resulting metrics snapshot
+   * @param all       if true, return all metrics even if unchanged.
+   */
+  @Override
+  public void getMetrics(MetricsCollector collector, boolean all) {
+    MetricsRecordBuilder metricsRecordBuilder = collector.addRecord("shuffleService");
+
+    for (Map.Entry<String, Metric> entry : metricSet.getMetrics().entrySet()) {
+      collectMetric(metricsRecordBuilder, entry.getKey(), entry.getValue());
+    }
+  }
+
+  @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;
+      metricsRecordBuilder
+        .addCounter(new ShuffleServiceMetricsInfo(name + "_count", "Count of timer " + name),
+          t.getCount())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate15", "15 minute rate of timer " + name),
+          t.getFifteenMinuteRate())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate5", "5 minute rate of timer " + name),
+          t.getFiveMinuteRate())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate1", "1 minute rate of timer " + name),
+          t.getOneMinuteRate())
+        .addGauge(new ShuffleServiceMetricsInfo(name + "_rateMean", "Mean rate of timer " + name),
+          t.getMeanRate());
+    } else if (metric instanceof Meter) {
+      Meter m = (Meter) metric;
+      metricsRecordBuilder
+        .addCounter(new ShuffleServiceMetricsInfo(name + "_count", "Count of meter " + name),
+          m.getCount())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate15", "15 minute rate of meter " + name),
+          m.getFifteenMinuteRate())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate5", "5 minute rate of meter " + name),
+          m.getFiveMinuteRate())
+        .addGauge(
+          new ShuffleServiceMetricsInfo(name + "_rate1", "1 minute rate of meter " + name),
+          m.getOneMinuteRate())
+        .addGauge(new ShuffleServiceMetricsInfo(name + "_rateMean", "Mean rate of meter " + name),
+          m.getMeanRate());
+    } else if (metric instanceof Gauge) {
+      Gauge m = (Gauge) metric;
+      Object gaugeValue = m.getValue();
+      if (gaugeValue instanceof Integer) {
+        Integer intValue = (Integer) gaugeValue;
+        metricsRecordBuilder
+          .addGauge(new ShuffleServiceMetricsInfo(name, "Integer value of " +
+            "gauge " + name), intValue.intValue());
+      }
+    }
+  }
+
+  private static class ShuffleServiceMetricsInfo implements MetricsInfo {
+
+    private final String name;
+    private final String description;
+
+    ShuffleServiceMetricsInfo(String name, String description) {
+      this.name = name;
+      this.description = description;
+    }
+
+    @Override
+    public String name() {
+      return name;
+    }
+
+    @Override
+    public String description() {
+      return description;
+    }
+  }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.network.yarn
+
+import org.apache.hadoop.metrics2.MetricsRecordBuilder
+import org.mockito.Matchers._
+import org.mockito.Mockito.{mock, times, verify, when}
+import org.scalatest.Matchers
+import scala.collection.JavaConverters._
+
+import org.apache.spark.network.server.OneForOneStreamManager
+import org.apache.spark.network.shuffle.{ExternalShuffleBlockHandler, ExternalShuffleBlockResolver}
+import org.apache.spark.SparkFunSuite
+
+class YarnShuffleServiceMetricsSuite extends SparkFunSuite with Matchers {
+
+  val streamManager = mock(classOf[OneForOneStreamManager])
+  val blockResolver = mock(classOf[ExternalShuffleBlockResolver])
+  when(blockResolver.getRegisteredExecutorsSize).thenReturn(42)
+
+  val metrics = new ExternalShuffleBlockHandler(streamManager, blockResolver).getAllMetrics
+
+  test("metrics named as expected") {
+    val allMetrics = Set(
+      "openBlockRequestLatencyMillis", "registerExecutorRequestLatencyMillis",
+      "blockTransferRateBytes", "registeredExecutorsSize")
+
+    metrics.getMetrics.keySet().asScala should be (allMetrics)
+  }
+
+  // these three metrics have the same effect on the collector
+  for (testname <- Seq("openBlockRequestLatencyMillis",
+      "registerExecutorRequestLatencyMillis",
+      "blockTransferRateBytes")) {
+    test(s"$testname - 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, testname,
+        metrics.getMetrics.get(testname))
+
+      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])
+    when(builder.addCounter(any(), anyLong())).thenReturn(builder)
+    when(builder.addGauge(any(), anyDouble())).thenReturn(builder)
+
+    YarnShuffleServiceMetrics.collectMetric(builder, "registeredExecutorsSize",
+      metrics.getMetrics.get("registeredExecutorsSize"))
+
+    // only one
+    verify(builder).addGauge(anyObject(), anyInt())
+  }
+}