[SPARK-11315] [YARN] Add YARN extension service to publish Spark events to YARN timeline service

pom.xml

@@ -214,6 +214,7 @@
     <PermGen>64m</PermGen>
     <MaxPermGen>512m</MaxPermGen>
     <CodeCacheSize>512m</CodeCacheSize>
+    <scalastyle.path>scalastyle-config.xml</scalastyle.path>
   </properties>
   <repositories>
     <repository>
@@ -2225,7 +2226,7 @@
           <failOnWarning>false</failOnWarning>
           <sourceDirectory>${basedir}/src/main/scala</sourceDirectory>
           <testSourceDirectory>${basedir}/src/test/scala</testSourceDirectory>
-          <configLocation>scalastyle-config.xml</configLocation>
+          <configLocation>${scalastyle.path}</configLocation>
           <outputFile>${basedir}/target/scalastyle-output.xml</outputFile>
           <inputEncoding>${project.build.sourceEncoding}</inputEncoding>
           <outputEncoding>${project.reporting.outputEncoding}</outputEncoding>

yarn/pom.xml

@@ -189,6 +189,106 @@
     </dependency>
   </dependencies>
 
+  <profiles>
+    <!-- Hadoop 2.6+ is needed to publish events to the YARN timeline service -->
+    <profile>
+      <id>hadoop-2.6</id>
+      <dependencies>
+        <dependency>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-core</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-json</artifactId>
+        </dependency>
+        <dependency>
+          <groupId>com.sun.jersey</groupId>
+          <artifactId>jersey-server</artifactId>
+        </dependency>
+      </dependencies>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.codehaus.mojo</groupId>
+            <artifactId>build-helper-maven-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>add-source</id>
+                <phase>generate-sources</phase>
+                <goals>
+                  <goal>add-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>src/history/main/scala</source>
+                  </sources>
+                </configuration>
+              </execution>
+              <execution>
+                <id>add-test-source</id>
+                <phase>generate-test-sources</phase>
+                <goals>
+                  <goal>add-test-source</goal>
+                </goals>
+                <configuration>
+                  <sources>
+                    <source>src/history/test/scala</source>
+                  </sources>
+                </configuration>
+              </execution>
+              <execution>
+                <id>add-test-resource</id>
+                <phase>generate-test-resources</phase>
+                <goals>
+                  <goal>add-test-resource</goal>
+                </goals>
+                <configuration>
+                  <resources>
+                    <resource>
+                      <directory>src/history/test/resources</directory>
+                    </resource>
+                  </resources>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+          <plugin>
+            <groupId>org.scalastyle</groupId>
+            <artifactId>scalastyle-maven-plugin</artifactId>
+            <configuration>
+              <verbose>false</verbose>
+              <failOnViolation>true</failOnViolation>
+              <!--<includeTestSourceDirectory>false</includeTestSourceDirectory>-->
+              <includeTestSourceDirectory>true</includeTestSourceDirectory>
+              <failOnWarning>false</failOnWarning>
+              <sourceDirectories>
+                <dir>${basedir}/src/main/scala</dir>
+                <dir>${basedir}/src/history/main/scala</dir>
+              </sourceDirectories>
+              <testSourceDirectories>
+                <dir>${basedir}/src/test/scala</dir>
+                <dir>${basedir}/src/history/test/scala</dir>
+              </testSourceDirectories>
+              <testSourceDirectory>${basedir}/src/test/scala</testSourceDirectory>
+              <configLocation>${scalastyle.path}</configLocation>
+              <outputFile>${basedir}/target/scalastyle-output.xml</outputFile>
+              <inputEncoding>${project.build.sourceEncoding}</inputEncoding>
+              <outputEncoding>${project.reporting.outputEncoding}</outputEncoding>
+            </configuration>
+            <executions>
+              <execution>
+                <goals>
+                  <goal>check</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
   <build>
     <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
     <testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>

yarn/src/history/main/scala/org/apache/spark/deploy/history/yarn/ExtendedMetricsSource.scala

@@ -0,0 +1,61 @@
+/*
+ * 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.deploy.history.yarn
+
+import com.codahale.metrics.{Metric, Timer}
+
+import org.apache.spark.metrics.source.Source
+
+/**
+ * An extended metrics source with some operations to build up the registry, and
+ * to time a closure.
+ */
+private[history] trait ExtendedMetricsSource extends Source {
+
+  /**
+   * A map to build up of all metrics to register and include in the string value
+   * @return
+   */
+  def metricsMap: Map[String, Metric]
+
+  protected def init(): Unit = {
+    metricsMap.foreach(elt => metricRegistry.register(elt._1, elt._2))
+  }
+
+  override def toString: String = {
+    def sb = new StringBuilder()
+    metricsMap.foreach(elt => sb.append(s" ${elt._1} = ${elt._2}\n"))
+    sb.toString()
+  }
+
+  /**
+   * Time a closure, returning its output.
+   * @param t timer
+   * @param f function
+   * @tparam T type of return value of the function
+   * @return the result of the function.
+   */
+  def time[T](t: Timer)(f: => T): T = {
+    val timeCtx = t.time()
+    try {
+      f
+    } finally {
+      timeCtx.close()
+    }
+  }
+}

yarn/src/history/main/scala/org/apache/spark/deploy/history/yarn/YarnEventListener.scala

@@ -0,0 +1,42 @@
+/*
+ * 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.deploy.history.yarn
+
+import org.apache.spark.{Logging, SparkContext, SparkFirehoseListener}
+import org.apache.spark.scheduler.SparkListenerEvent
+
+/**
+ * Spark listener which queues up all received events to the [[YarnHistoryService]] passed
+ * as a constructor. There's no attempt to filter event types at this point.
+ *
+ * @param sc context
+ * @param service service to forward events to
+ */
+private[spark] class YarnEventListener(sc: SparkContext, service: YarnHistoryService)
+  extends SparkFirehoseListener with Logging {
+
+  /**
+   * queue the event with the service, timestamped to the current time.
+   *
+   * @param event event to queue
+   */
+  override def onEvent(event: SparkListenerEvent): Unit = {
+    service.enqueue(event)
+  }
+
+}