[SPARK-11373] [CORE] Add metrics to the History Server and FsHistoryProvider

core/src/main/scala/org/apache/spark/deploy/history/ApplicationCache.scala

@@ -24,12 +24,11 @@ import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
 import scala.collection.JavaConverters._
 import scala.util.control.NonFatal
 
-import com.codahale.metrics.{Counter, MetricRegistry, Timer}
+import com.codahale.metrics.{Counter, Counting, Metric, Timer}
 import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache, RemovalListener, RemovalNotification}
 import org.eclipse.jetty.servlet.FilterHolder
 
 import org.apache.spark.internal.Logging
-import org.apache.spark.metrics.source.Source
 import org.apache.spark.ui.SparkUI
 import org.apache.spark.util.Clock
 
@@ -388,7 +387,7 @@ private[history] final case class CacheKey(appId: String, attemptId: Option[Stri
  * Metrics of the cache
  * @param prefix prefix to register all entries under
  */
-private[history] class CacheMetrics(prefix: String) extends Source {
+private[history] class CacheMetrics(prefix: String) extends HistoryMetricSource(prefix) {
 
   /* metrics: counters and timers */
   val lookupCount = new Counter()
@@ -410,34 +409,25 @@ private[history] class CacheMetrics(prefix: String) extends Source {
     ("update.triggered.count", updateTriggeredCount))
 
   /** all metrics, including timers */
-  private val allMetrics = counters ++ Seq(
+  private val allMetrics: Seq[(String, Metric with Counting)] = counters ++ Seq(
     ("load.timer", loadTimer),
     ("update.probe.timer", updateProbeTimer))
 
   /**
    * Name of metric source
    */
-  override val sourceName = "ApplicationCache"
-
-  override val metricRegistry: MetricRegistry = new MetricRegistry
+  override val sourceName = "application.cache"
 
   /**
    * Startup actions.
    * This includes registering metrics with [[metricRegistry]]
    */
   private def init(): Unit = {
-    allMetrics.foreach { case (name, metric) =>
-      metricRegistry.register(MetricRegistry.name(prefix, name), metric)
-    }
+    register(allMetrics)
   }
 
-  override def toString: String = {
-    val sb = new StringBuilder()
-    counters.foreach { case (name, counter) =>
-      sb.append(name).append(" = ").append(counter.getCount).append('\n')
-    }
-    sb.toString()
-  }
+  init()
+
 }
 
 /**

core/src/main/scala/org/apache/spark/deploy/history/ApplicationHistoryProvider.scala

@@ -17,11 +17,14 @@
 
 package org.apache.spark.deploy.history
 
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.zip.ZipOutputStream
 
 import scala.xml.Node
 
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkException, SparkFirehoseListener}
+import org.apache.spark.metrics.source.Source
+import org.apache.spark.scheduler._
 import org.apache.spark.ui.SparkUI
 
 private[spark] case class ApplicationAttemptInfo(
@@ -74,6 +77,8 @@ private[history] case class LoadedAppUI(
 
 private[history] abstract class ApplicationHistoryProvider {
 
+  private val started = new AtomicBoolean(false)
+
   /**
    * Returns the count of application event logs that the provider is currently still processing.
    * History Server UI can use this to indicate to a user that the application listing on the UI
@@ -98,6 +103,19 @@ private[history] abstract class ApplicationHistoryProvider {
     return 0;
   }
 
+  /**
+   * Bind to the History Server: threads should be started here; exceptions may be raised
+   * Start the provider: threads should be started here; exceptions may be raised
+   * if the history provider cannot be started.
+   * The base implementation contains a re-entrancy check and should
+   * be invoked first.
+   * @return the metric information for registration
+   */
+  def start(): Option[Source] = {
+    require(!started.getAndSet(true), "History provider already started")
+    None
+  }
+
   /**
    * Returns a list of applications available for the history server to show.
    *
@@ -145,3 +163,15 @@ private[history] abstract class ApplicationHistoryProvider {
    */
   def getEmptyListingHtml(): Seq[Node] = Seq.empty
 }
+
+/**
+ * A simple counter of events.
+ * There is no concurrency support here: all events must come in sequentially.
+ */
+private[history] class EventCountListener extends SparkFirehoseListener {
+  var eventCount = 0L
+
+  override def onEvent(event: SparkListenerEvent): Unit = {
+    eventCount += 1
+  }
+}