Add metrics to mesos cluster scheduler.

Author: tnachen

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/DriverQueue.scala

@@ -44,6 +44,8 @@ private[mesos] class DriverQueue(state: MesosClusterPersistenceEngine, capacity:
 
   def isFull = count >= capacity
 
+  def size: Int = count
+
   def contains(submissionId: String): Boolean = {
     queue.exists(s => s.submissionId.equals(submissionId))
   }

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/LaunchedDrivers.scala

@@ -40,6 +40,8 @@ private[mesos] class LaunchedDrivers(state: MesosClusterPersistenceEngine) {
     }
   }
 
+  def size: Int = drivers.size
+
   def get(submissionId: String): MesosClusterTaskState = drivers(submissionId)
 
   def states: Iterable[MesosClusterTaskState] = {

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala

@@ -23,18 +23,20 @@ import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.locks.ReentrantLock
 import java.util.{Collections, Date, List => JList}
 
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
 import org.apache.mesos.Protos.Environment.Variable
 import org.apache.mesos.Protos.TaskStatus.Reason
 import org.apache.mesos.Protos.{TaskState => MesosTaskState, _}
 import org.apache.mesos.{Scheduler, SchedulerDriver}
+
 import org.apache.spark.deploy.mesos.MesosDriverDescription
 import org.apache.spark.deploy.rest.{CreateSubmissionResponse, KillSubmissionResponse, SubmissionStatusResponse}
+import org.apache.spark.metrics.MetricsSystem
 import org.apache.spark.util.Utils
-import org.apache.spark.{SparkConf, SparkException, TaskState}
-
-import scala.collection.JavaConversions._
-import scala.collection.mutable
-import scala.collection.mutable.ArrayBuffer
+import org.apache.spark.{SparkConf, SparkException, SecurityManager, TaskState}
 
 
 /**
@@ -111,6 +113,8 @@ private[spark] class MesosClusterSchedulerDriver(
 
   var frameworkUrl: String = _
 
+  private val metricsSystem =
+    MetricsSystem.createMetricsSystem("mesos_cluster", conf, new SecurityManager(conf))
   private val master = conf.get("spark.master")
   private val appName = conf.get("spark.app.name")
   private val queuedCapacity = conf.getInt("spark.deploy.mesos.queuedDrivers", 200)
@@ -123,13 +127,14 @@ private[spark] class MesosClusterSchedulerDriver(
   private var frameworkId: String = null
 
   // Stores all the launched and running drivers' states.
-  private var launchedDrivers: LaunchedDrivers = _
+  var launchedDrivers: LaunchedDrivers = _
 
   // A queue that stores all the submitted drivers that hasn't been launched.
-  private var queue: DriverQueue = _
+  var queue: DriverQueue = _
 
   // All supervised drivers that are waiting to retry after termination.
-  private var superviseRetryList: SuperviseRetryList = _
+  var superviseRetryList: SuperviseRetryList = _
+
   private var masterInfo: MasterInfo = _
 
   private def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")  // For application IDs
@@ -223,11 +228,15 @@ private[spark] class MesosClusterSchedulerDriver(
       frameworkId = id
     }
     recoverState
+    metricsSystem.registerSource(new MesosClusterSchedulerSource(this))
+    metricsSystem.start()
     startScheduler(
       "MesosClusterScheduler", master, MesosClusterSchedulerDriver.this, builder.build())
   }
 
   def stop(): Unit = {
+    metricsSystem.report()
+    metricsSystem.stop()
     driver.stop(true)
   }
 

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSource.scala

@@ -0,0 +1,40 @@
+/*
+ * 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.scheduler.cluster.mesos
+
+import com.codahale.metrics.{Gauge, MetricRegistry}
+
+import org.apache.spark.metrics.source.Source
+
+private[mesos] class MesosClusterSchedulerSource(scheduler: MesosClusterSchedulerDriver) extends Source {
+  override def sourceName: String = "mesos_cluster"
+
+  override def metricRegistry: MetricRegistry = new MetricRegistry()
+
+  metricRegistry.register(MetricRegistry.name("waitingDrivers"), new Gauge[Int] {
+    override def getValue: Int = scheduler.queue.size
+  })
+
+  metricRegistry.register(MetricRegistry.name("launchedDrivers"), new Gauge[Int] {
+    override def getValue: Int = scheduler.launchedDrivers.size
+  })
+
+  metricRegistry.register(MetricRegistry.name("retryDrivers"), new Gauge[Int] {
+    override def getValue: Int = scheduler.superviseRetryList.size
+  })
+}

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/SuperviseRetryList.scala

@@ -55,6 +55,8 @@ private[mesos] class SuperviseRetryList(state: MesosClusterPersistenceEngine) {
     state.fetchAll[RetryState]().foreach(drivers.+=)
   }
 
+  def size: Int = drivers.size
+
   def contains(submissionId: String): Boolean =
     drivers.exists(d => d.submission.submissionId.equals(submissionId))