[SPARK-20863] Add metrics/instrumentation to LiveListenerBus

core/src/main/scala/org/apache/spark/scheduler/LiveListenerBus.scala

@@ -23,6 +23,7 @@ import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 import scala.util.DynamicVariable
 
 import com.codahale.metrics.{Counter, Gauge, MetricRegistry, Timer}
+import com.google.common.cache.{CacheBuilder, CacheLoader, LoadingCache}
 
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.internal.config._
@@ -111,6 +112,15 @@ private[spark] class LiveListenerBus(conf: SparkConf) extends SparkListenerBus {
     }
   }
 
+  override protected def createTimer(listener: SparkListenerInterface): Option[Timer] = {
+    if (listener.getClass.getName.startsWith("org.apache.spark")) {
+      metrics.perListenerTimers.size()
+      Some(metrics.perListenerTimers(listener.getClass.getSimpleName))
+    } else {
+      None
+    }
+  }
+
   /**
    * Start sending events to attached listeners.
    *
@@ -270,5 +280,16 @@ private[spark] class LiveListenerBusMetrics(queue: LinkedBlockingQueue[_]) exten
       override def getValue: Int = queue.size()
     })
   }
+
+  /**
+   * Mapping from fully-qualified listener class name to a timer tracking the processing time of
+   * events processed by that listener.
+   */
+  val perListenerTimers: LoadingCache[String, Timer] =
+    CacheBuilder.newBuilder().build[String, Timer](new CacheLoader[String, Timer] {
+      override def load(listenerName: String): Timer = {
+        metricRegistry.timer(MetricRegistry.name("listenerProcessingTime", listenerName))
+      }
+    })
 }
 

core/src/main/scala/org/apache/spark/util/ListenerBus.scala

@@ -23,29 +23,41 @@ import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 import scala.util.control.NonFatal
 
+import com.codahale.metrics.Timer
+
 import org.apache.spark.internal.Logging
 
 /**
  * An event bus which posts events to its listeners.
  */
 private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging {
 
+  private[this] val listenersPlusTimers = new CopyOnWriteArrayList[(L, Timer)]
+
   // Marked `private[spark]` for access in tests.
-  private[spark] val listeners = new CopyOnWriteArrayList[L]
+  private[spark] def listeners = listenersPlusTimers.asScala.map(_._1).asJava
+
+  /**
+   * Returns a CodaHale metrics Timer for measuring the listener's event processing time.
+   * This method is intended to be overridden by subclasses.
+   */
+  protected def createTimer(listener: L): Option[Timer] = None
 
   /**
    * Add a listener to listen events. This method is thread-safe and can be called in any thread.
    */
   final def addListener(listener: L): Unit = {
-    listeners.add(listener)
+    listenersPlusTimers.add((listener, createTimer(listener).orNull))
   }
 
   /**
    * Remove a listener and it won't receive any events. This method is thread-safe and can be called
    * in any thread.
    */
   final def removeListener(listener: L): Unit = {
-    listeners.remove(listener)
+    listenersPlusTimers.asScala.find(_._1 eq listener).foreach { listenerAndTimer =>
+      listenersPlusTimers.remove(listenerAndTimer)
+    }
   }
 
   /**
@@ -56,14 +68,25 @@ private[spark] trait ListenerBus[L <: AnyRef, E] extends Logging {
     // JavaConverters can create a JIterableWrapper if we use asScala.
     // However, this method will be called frequently. To avoid the wrapper cost, here we use
     // Java Iterator directly.
-    val iter = listeners.iterator
+    val iter = listenersPlusTimers.iterator
     while (iter.hasNext) {
-      val listener = iter.next()
+      val listenerAndMaybeTimer = iter.next()
+      val listener = listenerAndMaybeTimer._1
+      val maybeTimer = listenerAndMaybeTimer._2
+      var maybeTimerContext = if (maybeTimer != null) {
+        maybeTimer.time()
+      } else {
+        null
+      }
       try {
         doPostEvent(listener, event)
       } catch {
         case NonFatal(e) =>
           logError(s"Listener ${Utils.getFormattedClassName(listener)} threw an exception", e)
+      } finally {
+        if (maybeTimerContext != null) {
+          maybeTimerContext.stop()
+        }
       }
     }
   }