allow configuring of MetricsSystem namespace

previously always namespaced metrics with app ID; now metrics-
config-file can specify a different Spark conf param to use, e.g.
app name.

Author: ryan-williams

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

@@ -36,6 +36,8 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging {
   private[metrics] val properties = new Properties()
   private[metrics] var perInstanceProperties: mutable.HashMap[String, Properties] = null
 
+  private[metrics] var metricsNamespaceConfParam: String = null
+
   private def setDefaultProperties(prop: Properties) {
     prop.setProperty("*.sink.servlet.class", "org.apache.spark.metrics.sink.MetricsServlet")
     prop.setProperty("*.sink.servlet.path", "/metrics/json")
@@ -65,6 +67,8 @@ private[spark] class MetricsConfig(conf: SparkConf) extends Logging {
         prop.put(k, v)
       }
     }
+
+    metricsNamespaceConfParam = properties.getProperty("namespace-conf-param", "spark.app.id")
   }
 
   def subProperties(prop: Properties, regex: Regex): mutable.HashMap[String, Properties] = {

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

@@ -124,7 +124,7 @@ private[spark] class MetricsSystem private (
    *         application, executor/driver and metric source.
    */
   private[spark] def buildRegistryName(source: Source): String = {
-    val appId = conf.getOption("spark.app.id")
+    val appId = conf.getOption(metricsConfig.metricsNamespaceConfParam)
     val executorId = conf.getOption("spark.executor.id")
     val defaultName = MetricRegistry.name(source.sourceName)
 
@@ -135,8 +135,12 @@ private[spark] class MetricsSystem private (
         // Only Driver and Executor set spark.app.id and spark.executor.id.
         // Other instance types, e.g. Master and Worker, are not related to a specific application.
         val warningMsg = s"Using default name $defaultName for source because %s is not set."
-        if (appId.isEmpty) { logWarning(warningMsg.format("spark.app.id")) }
-        if (executorId.isEmpty) { logWarning(warningMsg.format("spark.executor.id")) }
+        if (appId.isEmpty) {
+          logWarning(warningMsg.format(metricsConfig.metricsNamespaceConfParam))
+        }
+        if (executorId.isEmpty) {
+          logWarning(warningMsg.format("spark.executor.id"))
+        }
         defaultName
       }
     } else { defaultName }

core/src/test/resources/test_metrics_config_namespace.properties

@@ -0,0 +1,25 @@
+#
+# 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.
+#
+
+namespace-conf-param = spark.app.name
+
+*.sink.console.period = 10
+*.sink.console.unit = seconds
+*.source.jvm.class = org.apache.spark.metrics.source.JvmSource
+master.sink.console.period = 20
+master.sink.console.unit = minutes
+

core/src/test/resources/test_metrics_system_namespace.properties

@@ -0,0 +1,24 @@
+#
+# 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.
+#
+
+namespace-conf-param = spark.app.name
+
+*.sink.console.period = 10
+*.sink.console.unit = seconds
+test.sink.console.class = org.apache.spark.metrics.sink.ConsoleSink
+test.sink.console.period = 20
+test.sink.console.unit = minutes

core/src/test/scala/org/apache/spark/metrics/MetricsConfigSuite.scala

@@ -82,6 +82,7 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter {
     setMetricsProperty(sparkConf, "master.sink.console.unit", "minutes")
     val conf = new MetricsConfig(sparkConf)
     conf.initialize()
+    assert(conf.metricsNamespaceConfParam == "spark.app.id")
 
     val masterProp = conf.getInstance("master")
     assert(masterProp.size() === 5)
@@ -134,6 +135,17 @@ class MetricsConfigSuite extends SparkFunSuite with BeforeAndAfter {
     assert(workerProp.getProperty("sink.servlet.path") === "/metrics/json")
   }
 
+  test("MetricsConfig with alternate namespace set") {
+    val sparkConf = new SparkConf
+    sparkConf.set(
+      "spark.metrics.conf",
+      getClass.getClassLoader.getResource("test_metrics_config_namespace.properties").getFile()
+    )
+    val conf = new MetricsConfig(sparkConf)
+    conf.initialize()
+    assert(conf.metricsNamespaceConfParam == "spark.app.name")
+  }
+
   test("MetricsConfig with subProperties") {
     val sparkConf = new SparkConf(loadDefaults = false)
     sparkConf.set("spark.metrics.conf", filePath)

core/src/test/scala/org/apache/spark/metrics/MetricsSystemSuite.scala

@@ -114,6 +114,32 @@ class MetricsSystemSuite extends SparkFunSuite with BeforeAndAfter with PrivateM
     assert(metricName === source.sourceName)
   }
 
+  test("MetricsSystem with Driver instance and alternate namespace set") {
+    val source = new Source {
+      override val sourceName = "dummySource"
+      override val metricRegistry = new MetricRegistry()
+    }
+
+    val appId = "testId"
+    conf.set("spark.app.id", appId)
+
+    val appName = "testName"
+    conf.set("spark.app.name", appName)
+
+    val executorId = "driver"
+    conf.set("spark.executor.id", executorId)
+
+    val alternateNamespaceFile =
+      getClass.getClassLoader.getResource("test_metrics_system_namespace.properties").getFile
+    conf.set("spark.metrics.conf", alternateNamespaceFile)
+
+    val instanceName = "driver"
+    val driverMetricsSystem = MetricsSystem.createMetricsSystem(instanceName, conf, securityMgr)
+
+    val metricName = driverMetricsSystem.buildRegistryName(source)
+    assert(metricName === s"$appName.$executorId.${source.sourceName}")
+  }
+
   test("MetricsSystem with Executor instance") {
     val source = new Source {
       override val sourceName = "dummySource"