From 439938ae39704859125533aa6b587c2529c3f779 Mon Sep 17 00:00:00 2001 From: Jacek Lewandowski Date: Wed, 29 Apr 2015 22:26:16 +0200 Subject: [PATCH 1/4] SPARK-7169: Metrics can be additionally configured from Spark configuration (cherry picked from commit 71993fe7fae09eb797e97d447a2d5175a4519a96) Conflicts: core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala --- .../scala/org/apache/spark/SparkConf.scala | 13 ++++ .../apache/spark/metrics/MetricsConfig.scala | 44 +++++++----- .../apache/spark/metrics/MetricsSystem.scala | 3 +- .../spark/metrics/MetricsConfigSuite.scala | 71 +++++++++++++++++-- 4 files changed, 106 insertions(+), 25 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala index 46d72841dccce..aeced7c8876f7 100644 --- a/core/src/main/scala/org/apache/spark/SparkConf.scala +++ b/core/src/main/scala/org/apache/spark/SparkConf.scala @@ -133,6 +133,11 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging { set("spark.home", home) } + /** Set the metrics configuration property */ + def setMetricsProperty(name: String, value: String): SparkConf = { + set(s"spark.metrics.conf.$name", value) + } + /** Set multiple parameters together */ def setAll(settings: Traversable[(String, String)]): SparkConf = { settings.foreach { case (k, v) => set(k, v) } @@ -327,6 +332,14 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging { */ getAll.filter { case (k, _) => isAkkaConf(k) } + /** Get all metrics system properties. */ + def getMetricsConf: Seq[(String, String)] = { + val pattern = """^spark\.metrics\.conf\.(.*)$""".r + for ((pattern(key), value) <- getAll) yield { + (key, value) + } + } + /** * Returns the Spark application id, valid in the Driver after TaskScheduler registration and * from the start in the Executor. diff --git a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala index 8edf493780687..04f4fc9944c60 100644 --- a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala +++ b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala @@ -23,15 +23,16 @@ import java.util.Properties import scala.collection.mutable import scala.util.matching.Regex -import org.apache.spark.Logging import org.apache.spark.util.Utils +import org.apache.spark.{Logging, SparkConf} -private[spark] class MetricsConfig(val configFile: Option[String]) extends Logging { +private[spark] class MetricsConfig(conf: SparkConf) extends Logging { private val DEFAULT_PREFIX = "*" private val INSTANCE_REGEX = "^(\\*|[a-zA-Z]+)\\.(.+)".r private val DEFAULT_METRICS_CONF_FILENAME = "metrics.properties" + private[metrics] val configFile = conf.getOption("spark.metrics.conf") private[metrics] val properties = new Properties() private[metrics] var propertyCategories: mutable.HashMap[String, Properties] = null @@ -47,23 +48,10 @@ private[spark] class MetricsConfig(val configFile: Option[String]) extends Loggi setDefaultProperties(properties) // If spark.metrics.conf is not set, try to get file in class path - val isOpt: Option[InputStream] = configFile.map(new FileInputStream(_)).orElse { - try { - Option(Utils.getSparkClassLoader.getResourceAsStream(DEFAULT_METRICS_CONF_FILENAME)) - } catch { - case e: Exception => - logError("Error loading default configuration file", e) - None - } - } + loadPropertiesFromFile() - isOpt.foreach { is => - try { - properties.load(is) - } finally { - is.close() - } - } + // Also look for the properties in provided Spark configuration + conf.getMetricsConf map { case (key, value) => properties.setProperty(key, value) } propertyCategories = subProperties(properties, INSTANCE_REGEX) if (propertyCategories.contains(DEFAULT_PREFIX)) { @@ -97,5 +85,23 @@ private[spark] class MetricsConfig(val configFile: Option[String]) extends Loggi case None => propertyCategories.getOrElse(DEFAULT_PREFIX, new Properties) } } -} + private[this] def loadPropertiesFromFile(): Unit = { + var is: InputStream = null + try { + is = configFile match { + case Some(f) => new FileInputStream(f) + case None => Utils.getSparkClassLoader.getResourceAsStream(DEFAULT_METRICS_CONF_FILENAME) + } + + if (is != null) { + properties.load(is) + } + } catch { + case e: Exception => logError("Error loading configure file", e) + } finally { + if (is != null) is.close() + } + } + +} diff --git a/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala b/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala index 9150ad35712a1..ed5131c79fdc5 100644 --- a/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala +++ b/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala @@ -70,8 +70,7 @@ private[spark] class MetricsSystem private ( securityMgr: SecurityManager) extends Logging { - private[this] val confFile = conf.get("spark.metrics.conf", null) - private[this] val metricsConfig = new MetricsConfig(Option(confFile)) + private[this] val metricsConfig = new MetricsConfig(conf) private val sinks = new mutable.ArrayBuffer[Sink] private val sources = new mutable.ArrayBuffer[Source] diff --git a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala index a901a069d9bfe..dde2d01f6e764 100644 --- a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala +++ b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala @@ -17,6 +17,8 @@ package org.apache.spark.metrics +import org.apache.spark.SparkConf + import org.scalatest.BeforeAndAfter import org.apache.spark.SparkFunSuite @@ -29,7 +31,9 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { } test("MetricsConfig with default properties") { - val conf = new MetricsConfig(None) + val sparkConf = new SparkConf(loadDefaults = false) + sparkConf.set("spark.metrics.conf", "dummy-file") + val conf = new MetricsConfig(sparkConf) conf.initialize() assert(conf.properties.size() === 4) @@ -42,8 +46,10 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { assert(property.getProperty("sink.servlet.path") === "/metrics/json") } - test("MetricsConfig with properties set") { - val conf = new MetricsConfig(Option(filePath)) + test("MetricsConfig with properties set from a file") { + val sparkConf = new SparkConf(loadDefaults = false) + sparkConf.set("spark.metrics.conf", filePath) + val conf = new MetricsConfig(sparkConf) conf.initialize() val masterProp = conf.getInstance("master") @@ -67,8 +73,65 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json") } + test("MetricsConfig with properties set from a Spark configuration") { + val sparkConf = new SparkConf(loadDefaults = false) + sparkConf.setMetricsProperty("*.sink.console.period", "10") + sparkConf.setMetricsProperty("*.sink.console.unit", "seconds") + sparkConf.setMetricsProperty("*.source.jvm.class", "org.apache.spark.metrics.source.JvmSource") + sparkConf.setMetricsProperty("master.sink.console.period", "20") + sparkConf.setMetricsProperty("master.sink.console.unit", "minutes") + val conf = new MetricsConfig(sparkConf) + conf.initialize() + + val masterProp = conf.getInstance("master") + assert(masterProp.size() === 5) + assert(masterProp.getProperty("sink.console.period") === "20") + assert(masterProp.getProperty("sink.console.unit") === "minutes") + assert(masterProp.getProperty("source.jvm.class") === "org.apache.spark.metrics.source.JvmSource") + assert(masterProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(masterProp.getProperty("sink.servlet.path") === "/metrics/master/json") + + val workerProp = conf.getInstance("worker") + assert(workerProp.size() === 5) + assert(workerProp.getProperty("sink.console.period") === "10") + assert(workerProp.getProperty("sink.console.unit") === "seconds") + assert(workerProp.getProperty("source.jvm.class") === "org.apache.spark.metrics.source.JvmSource") + assert(workerProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json") + } + + test("MetricsConfig with properties set from a file and a Spark configuration") { + val sparkConf = new SparkConf(loadDefaults = false) + sparkConf.setMetricsProperty("*.sink.console.period", "10") + sparkConf.setMetricsProperty("*.sink.console.unit", "seconds") + sparkConf.setMetricsProperty("*.source.jvm.class", "org.apache.spark.SomeOtherSource") + sparkConf.setMetricsProperty("master.sink.console.period", "50") + sparkConf.setMetricsProperty("master.sink.console.unit", "seconds") + sparkConf.set("spark.metrics.conf", filePath) + val conf = new MetricsConfig(sparkConf) + conf.initialize() + + val masterProp = conf.getInstance("master") + assert(masterProp.size() === 5) + assert(masterProp.getProperty("sink.console.period") === "50") + assert(masterProp.getProperty("sink.console.unit") === "seconds") + assert(masterProp.getProperty("source.jvm.class") === "org.apache.spark.SomeOtherSource") + assert(masterProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(masterProp.getProperty("sink.servlet.path") === "/metrics/master/json") + + val workerProp = conf.getInstance("worker") + assert(workerProp.size() === 5) + assert(workerProp.getProperty("sink.console.period") === "10") + assert(workerProp.getProperty("sink.console.unit") === "seconds") + assert(workerProp.getProperty("source.jvm.class") === "org.apache.spark.SomeOtherSource") + assert(workerProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json") + } + test("MetricsConfig with subProperties") { - val conf = new MetricsConfig(Option(filePath)) + val sparkConf = new SparkConf(loadDefaults = false) + sparkConf.set("spark.metrics.conf", filePath) + val conf = new MetricsConfig(sparkConf) conf.initialize() val propCategories = conf.propertyCategories From ce66e7ee28458e37226109b5646a4885b4c00551 Mon Sep 17 00:00:00 2001 From: Marcelo Vanzin Date: Mon, 1 Jun 2015 13:36:48 -0700 Subject: [PATCH 2/4] Remove metrics config handling from SparkConf. --- .../scala/org/apache/spark/SparkConf.scala | 13 ---------- .../apache/spark/metrics/MetricsConfig.scala | 7 +++++- .../spark/metrics/MetricsConfigSuite.scala | 25 +++++++++++-------- 3 files changed, 21 insertions(+), 24 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala index aeced7c8876f7..46d72841dccce 100644 --- a/core/src/main/scala/org/apache/spark/SparkConf.scala +++ b/core/src/main/scala/org/apache/spark/SparkConf.scala @@ -133,11 +133,6 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging { set("spark.home", home) } - /** Set the metrics configuration property */ - def setMetricsProperty(name: String, value: String): SparkConf = { - set(s"spark.metrics.conf.$name", value) - } - /** Set multiple parameters together */ def setAll(settings: Traversable[(String, String)]): SparkConf = { settings.foreach { case (k, v) => set(k, v) } @@ -332,14 +327,6 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging { */ getAll.filter { case (k, _) => isAkkaConf(k) } - /** Get all metrics system properties. */ - def getMetricsConf: Seq[(String, String)] = { - val pattern = """^spark\.metrics\.conf\.(.*)$""".r - for ((pattern(key), value) <- getAll) yield { - (key, value) - } - } - /** * Returns the Spark application id, valid in the Driver after TaskScheduler registration and * from the start in the Executor. diff --git a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala index 04f4fc9944c60..e71b89f8a1edc 100644 --- a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala +++ b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala @@ -51,7 +51,12 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging { loadPropertiesFromFile() // Also look for the properties in provided Spark configuration - conf.getMetricsConf map { case (key, value) => properties.setProperty(key, value) } + val prefix = "spark.metrics.conf." + conf.getAll.foreach { + case (k, v) if k.startsWith(prefix) => + properties.setProperty(k.substring(prefix.length()), v) + case _ => + } propertyCategories = subProperties(properties, INSTANCE_REGEX) if (propertyCategories.contains(DEFAULT_PREFIX)) { diff --git a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala index dde2d01f6e764..d877182ebe08d 100644 --- a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala +++ b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala @@ -75,11 +75,11 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { test("MetricsConfig with properties set from a Spark configuration") { val sparkConf = new SparkConf(loadDefaults = false) - sparkConf.setMetricsProperty("*.sink.console.period", "10") - sparkConf.setMetricsProperty("*.sink.console.unit", "seconds") - sparkConf.setMetricsProperty("*.source.jvm.class", "org.apache.spark.metrics.source.JvmSource") - sparkConf.setMetricsProperty("master.sink.console.period", "20") - sparkConf.setMetricsProperty("master.sink.console.unit", "minutes") + setMetricsProperty(sparkConf, "*.sink.console.period", "10") + setMetricsProperty(sparkConf, "*.sink.console.unit", "seconds") + setMetricsProperty(sparkConf, "*.source.jvm.class", "org.apache.spark.metrics.source.JvmSource") + setMetricsProperty(sparkConf, "master.sink.console.period", "20") + setMetricsProperty(sparkConf, "master.sink.console.unit", "minutes") val conf = new MetricsConfig(sparkConf) conf.initialize() @@ -102,11 +102,11 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { test("MetricsConfig with properties set from a file and a Spark configuration") { val sparkConf = new SparkConf(loadDefaults = false) - sparkConf.setMetricsProperty("*.sink.console.period", "10") - sparkConf.setMetricsProperty("*.sink.console.unit", "seconds") - sparkConf.setMetricsProperty("*.source.jvm.class", "org.apache.spark.SomeOtherSource") - sparkConf.setMetricsProperty("master.sink.console.period", "50") - sparkConf.setMetricsProperty("master.sink.console.unit", "seconds") + setMetricsProperty(sparkConf, "*.sink.console.period", "10") + setMetricsProperty(sparkConf, "*.sink.console.unit", "seconds") + setMetricsProperty(sparkConf, "*.source.jvm.class", "org.apache.spark.SomeOtherSource") + setMetricsProperty(sparkConf, "master.sink.console.period", "50") + setMetricsProperty(sparkConf, "master.sink.console.unit", "seconds") sparkConf.set("spark.metrics.conf", filePath) val conf = new MetricsConfig(sparkConf) conf.initialize() @@ -153,4 +153,9 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { val servletProps = sinkProps("servlet") assert(servletProps.size() === 2) } + + private def setMetricsProperty(conf: SparkConf, name: String, value: String): Unit = { + conf.set(s"spark.metrics.conf.$name", value) + } + } From 702d5a3861d8bed714468ba866a58d9d19099a7a Mon Sep 17 00:00:00 2001 From: Marcelo Vanzin Date: Mon, 1 Jun 2015 14:58:36 -0700 Subject: [PATCH 3/4] Scalastyle. --- .../spark/metrics/MetricsConfigSuite.scala | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala index d877182ebe08d..41f2ff725a17b 100644 --- a/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala +++ b/core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala @@ -87,16 +87,20 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { assert(masterProp.size() === 5) assert(masterProp.getProperty("sink.console.period") === "20") assert(masterProp.getProperty("sink.console.unit") === "minutes") - assert(masterProp.getProperty("source.jvm.class") === "org.apache.spark.metrics.source.JvmSource") - assert(masterProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(masterProp.getProperty("source.jvm.class") === + "org.apache.spark.metrics.source.JvmSource") + assert(masterProp.getProperty("sink.servlet.class") === + "org.apache.spark.metrics.sink.MetricsServlet") assert(masterProp.getProperty("sink.servlet.path") === "/metrics/master/json") val workerProp = conf.getInstance("worker") assert(workerProp.size() === 5) assert(workerProp.getProperty("sink.console.period") === "10") assert(workerProp.getProperty("sink.console.unit") === "seconds") - assert(workerProp.getProperty("source.jvm.class") === "org.apache.spark.metrics.source.JvmSource") - assert(workerProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(workerProp.getProperty("source.jvm.class") === + "org.apache.spark.metrics.source.JvmSource") + assert(workerProp.getProperty("sink.servlet.class") === + "org.apache.spark.metrics.sink.MetricsServlet") assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json") } @@ -116,7 +120,8 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { assert(masterProp.getProperty("sink.console.period") === "50") assert(masterProp.getProperty("sink.console.unit") === "seconds") assert(masterProp.getProperty("source.jvm.class") === "org.apache.spark.SomeOtherSource") - assert(masterProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(masterProp.getProperty("sink.servlet.class") === + "org.apache.spark.metrics.sink.MetricsServlet") assert(masterProp.getProperty("sink.servlet.path") === "/metrics/master/json") val workerProp = conf.getInstance("worker") @@ -124,7 +129,8 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter { assert(workerProp.getProperty("sink.console.period") === "10") assert(workerProp.getProperty("sink.console.unit") === "seconds") assert(workerProp.getProperty("source.jvm.class") === "org.apache.spark.SomeOtherSource") - assert(workerProp.getProperty("sink.servlet.class") === "org.apache.spark.metrics.sink.MetricsServlet") + assert(workerProp.getProperty("sink.servlet.class") === + "org.apache.spark.metrics.sink.MetricsServlet") assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json") } From 737266f6ab74cf12bfb9bd6735b4df1430089794 Mon Sep 17 00:00:00 2001 From: Marcelo Vanzin Date: Fri, 5 Jun 2015 12:10:35 -0700 Subject: [PATCH 4/4] Feedback. --- .../apache/spark/metrics/MetricsConfig.scala | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala index e71b89f8a1edc..d7495551ad233 100644 --- a/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala +++ b/core/src/main/scala/org/apache/spark/metrics/MetricsConfig.scala @@ -32,7 +32,6 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging { private val INSTANCE_REGEX = "^(\\*|[a-zA-Z]+)\\.(.+)".r private val DEFAULT_METRICS_CONF_FILENAME = "metrics.properties" - private[metrics] val configFile = conf.getOption("spark.metrics.conf") private[metrics] val properties = new Properties() private[metrics] var propertyCategories: mutable.HashMap[String, Properties] = null @@ -47,8 +46,7 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging { // Add default properties in case there's no properties file setDefaultProperties(properties) - // If spark.metrics.conf is not set, try to get file in class path - loadPropertiesFromFile() + loadPropertiesFromFile(conf.getOption("spark.metrics.conf")) // Also look for the properties in provided Spark configuration val prefix = "spark.metrics.conf." @@ -91,10 +89,14 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging { } } - private[this] def loadPropertiesFromFile(): Unit = { + /** + * Loads configuration from a config file. If no config file is provided, try to get file + * in class path. + */ + private[this] def loadPropertiesFromFile(path: Option[String]): Unit = { var is: InputStream = null try { - is = configFile match { + is = path match { case Some(f) => new FileInputStream(f) case None => Utils.getSparkClassLoader.getResourceAsStream(DEFAULT_METRICS_CONF_FILENAME) } @@ -103,9 +105,13 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging { properties.load(is) } } catch { - case e: Exception => logError("Error loading configure file", e) + case e: Exception => + val file = path.getOrElse(DEFAULT_METRICS_CONF_FILENAME) + logError(s"Error loading configuration file $file", e) } finally { - if (is != null) is.close() + if (is != null) { + is.close() + } } }