Update metric source support for instrumentation

Author: jerryshao

conf/metrics.properties

@@ -1,9 +1,11 @@
-# syntax: [prefix].[sink].[instance].[options]
+# syntax: [prefix].[sink|source].[instance].[options]
 
 *.sink.console.period=10
 
 *.sink.console.unit=second
 
+master.source.jvm.class=spark.metrics.source.JvmSource
+
 master.sink.console.period=10
 
 master.sink.console.unit=second

core/src/main/scala/spark/deploy/master/MasterInstrumentation.scala

@@ -15,12 +15,15 @@ private[spark] trait MasterInstrumentation extends AbstractInstrumentation {
   def initialize(master: Master) {
     masterInst = Some(master)
 
+    // Register all the sources
+    registerSources()
+    
     // Register and start all the sinks
-    registerSinks
+    registerSinks()
   }
 
   def uninitialize() {
-    unregisterSinks
+    unregisterSinks()
   }
 
   // Gauge for worker numbers in cluster

core/src/main/scala/spark/deploy/worker/WorkerInstrumentation.scala

@@ -15,6 +15,9 @@ private[spark] trait WorkerInstrumentation extends AbstractInstrumentation {
   def initialize(worker: Worker) {
     workerInst = Some(worker)
 
+    // Register all the sources
+    registerSources()
+    
     // Register and start all the sinks
     registerSinks()
   }
@@ -36,7 +39,7 @@ private[spark] trait WorkerInstrumentation extends AbstractInstrumentation {
   })
 
   // Gauge for memory used of this worker
-  metricRegistry.register(MetricRegistry.name(classOf[Worker], "mem_used", "Mbytes"), 
+  metricRegistry.register(MetricRegistry.name(classOf[Worker], "mem_used", "MBytes"), 
     new Gauge[Int] {
       override def getValue: Int = workerInst.map(_.memoryUsed).getOrElse(0)
   })

core/src/main/scala/spark/metrics/AbstractInstrumentation.scala

@@ -9,17 +9,39 @@ import java.util.concurrent.TimeUnit
 
 import spark.Logging
 import spark.metrics.sink._
+import spark.metrics.source._
 
-trait AbstractInstrumentation extends Logging {
+private [spark] trait AbstractInstrumentation extends Logging {
   initLogging()
 
+  // Get MetricRegistry handler
   def registryHandler: MetricRegistry
+  // Get the instance name
   def instance: String
 
   val confFile = System.getProperty("spark.metrics.conf.file", MetricsConfig.DEFAULT_CONFIG_FILE)
   val metricsConfig = new MetricsConfig(confFile)
 
   val sinks = new mutable.ArrayBuffer[Sink]
+  val sources = new mutable.ArrayBuffer[Source]
+  
+  def registerSources() {
+    val instConfig = metricsConfig.getInstance(instance)
+    val sourceConfigs = MetricsConfig.subProperties(instConfig, AbstractInstrumentation.SOURCE_REGEX)
+    
+    // Register all the sources
+    sourceConfigs.foreach { kv =>
+      val classPath = kv._2.getProperty("class")
+      try {
+        val source = Class.forName(classPath).getConstructor(classOf[MetricRegistry])
+          .newInstance(registryHandler)
+        sources += source.asInstanceOf[Source]
+      } catch {
+        case e: Exception => logError("source class " + classPath + " cannot be instantialized", e)
+      }
+    }
+    sources.foreach(_.registerSource)
+  }
 
   def registerSinks() {
     val instConfig = metricsConfig.getInstance(instance)
@@ -33,17 +55,17 @@ trait AbstractInstrumentation extends Logging {
       val classPath = if (AbstractInstrumentation.DEFAULT_SINKS.contains(kv._1)) {
         AbstractInstrumentation.DEFAULT_SINKS(kv._1)
       } else {
+        // For non-default sink, a property class should be set and create using reflection
         kv._2.getProperty("class")
       }
       try {
         val sink = Class.forName(classPath).getConstructor(classOf[Properties], classOf[MetricRegistry])
           .newInstance(kv._2, registryHandler)
         sinks += sink.asInstanceOf[Sink]
       } catch {
-        case e: Exception => logError("class " + classPath + "cannot be instantialize", e)
+        case e: Exception => logError("sink class " + classPath + " cannot be instantialized", e)
       }
     }
-    
     sinks.foreach(_.registerSink)
   }
 
@@ -58,6 +80,7 @@ object AbstractInstrumentation {
       "csv" -> "spark.metrics.sink.CsvSink")
 
   val SINK_REGEX = "^sink\\.(.+)\\.(.+)".r
+  val SOURCE_REGEX = "^source\\.(.+)\\.(.+)".r
 
   val timeUnits = Map(
       "millisecond" -> TimeUnit.MILLISECONDS,

core/src/main/scala/spark/metrics/MetricsConfig.scala

@@ -6,7 +6,7 @@ import java.io.FileInputStream
 import scala.collection.mutable
 import scala.util.matching.Regex
 
-class MetricsConfig(val configFile: String) {
+private [spark] class MetricsConfig(val configFile: String) {
   val properties = new Properties()
   var fis: FileInputStream = _
 
@@ -36,7 +36,7 @@ class MetricsConfig(val configFile: String) {
 }
 
 object MetricsConfig {
-  val DEFAULT_CONFIG_FILE = "/home/jerryshao/project/sotc_cloud-spark/conf/metrics.properties"
+  val DEFAULT_CONFIG_FILE = "conf/metrics.properties"
   val DEFAULT_PREFIX = "*"
   val INSTANCE_REGEX = "^(\\*|[a-zA-Z]+)\\.(.+)".r
 
@@ -45,9 +45,11 @@ object MetricsConfig {
 
     import scala.collection.JavaConversions._
     prop.foreach { kv => 
-      val regex(a, b) = kv._1
-      subProperties.getOrElseUpdate(a, new Properties).setProperty(b, kv._2)
-      println(">>>>>subProperties added  " + a + " " + b + " " + kv._2)
+      if (regex.findPrefixOf(kv._1) != None) {
+        val regex(prefix, suffix) = kv._1
+        subProperties.getOrElseUpdate(prefix, new Properties).setProperty(suffix, kv._2)
+        println(">>>>>subProperties added  " + prefix + " " + suffix + " " + kv._2)
+      }
     }
 
     subProperties

core/src/main/scala/spark/metrics/sink/CsvSink.scala

@@ -46,8 +46,8 @@ object CsvSink {
   val CSV_KEY_UNIT = "unit"
   val CSV_KEY_DIR = "directory"
 
-  val CSV_DEFAULT_PERIOD = "1"
-  val CSV_DEFAULT_UNIT = "minute"
+  val CSV_DEFAULT_PERIOD = "10"
+  val CSV_DEFAULT_UNIT = "second"
   val CSV_DEFAULT_DIR = "/tmp/"
 }
 

core/src/main/scala/spark/metrics/sink/Sink.scala

@@ -2,5 +2,6 @@ package spark.metrics.sink
 
 trait Sink {
   def registerSink: Unit
+  
   def unregisterSink: Unit
 }

core/src/main/scala/spark/metrics/source/JvmSource.scala

@@ -0,0 +1,17 @@
+package spark.metrics.source
+
+import com.codahale.metrics.MetricRegistry
+import com.codahale.metrics.jvm.{MemoryUsageGaugeSet, GarbageCollectorMetricSet}
+
+class JvmSource(registry: MetricRegistry) extends Source {
+  // Initialize memory usage gauge for jvm
+  val memUsageMetricSet = new MemoryUsageGaugeSet
+  
+  // Initialize garbage collection usage gauge for jvm
+  val gcMetricSet = new GarbageCollectorMetricSet
+  
+  override def registerSource() {
+    registry.registerAll(memUsageMetricSet)
+    registry.registerAll(gcMetricSet)
+  }
+}

core/src/main/scala/spark/metrics/source/Source.scala

@@ -0,0 +1,5 @@
+package spark.metrics.source
+
+trait Source {
+  def registerSource: Unit
+}