Modified to add application id to metrics name

Author: sarutak

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -186,8 +186,6 @@ class SparkContext(config: SparkConf) extends Logging {
 
   val master = conf.get("spark.master")
   val appName = conf.get("spark.app.name")
-  val uniqueAppName = appName + "-" + System.currentTimeMillis()
-  conf.set("spark.unique.app.name", uniqueAppName)
 
   // Generate the random name for a temp folder in Tachyon
   // Add a timestamp as the suffix here to make it more safe
@@ -312,6 +310,14 @@ class SparkContext(config: SparkConf) extends Logging {
   // constructor
   taskScheduler.start()
 
+  val appId = taskScheduler.applicationId().getOrElse(System.currentTimeMillis().toString)
+  conf.set("spark.app.id", appId)
+
+  val metricsSystem = env.metricsSystem
+  metricsSystem.registerSources()
+  metricsSystem.registerSinks()
+  metricsSystem.start()
+
   private[spark] val cleaner: Option[ContextCleaner] = {
     if (conf.getBoolean("spark.cleaner.referenceTracking", true)) {
       Some(new ContextCleaner(this))

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -261,9 +261,13 @@ object SparkEnv extends Logging {
     val metricsSystem = if (isDriver) {
       MetricsSystem.createMetricsSystem("driver", conf, securityManager)
     } else {
-      MetricsSystem.createMetricsSystem("executor", conf, securityManager)
+      val ms = MetricsSystem.createMetricsSystem("executor", conf, securityManager)
+      ms.registerSources()
+      ms.registerSinks()
+      ms.start()
+      ms
     }
-    metricsSystem.start()
+
 
     // Set the sparkFiles directory, used when downloading dependencies.  In local mode,
     // this is a temporary directory; in distributed mode, this is the executor's current working

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -106,7 +106,8 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       executorId: String,
       hostname: String,
       cores: Int,
-      workerUrl: Option[String]) {
+      workerUrl: Option[String],
+      appId: String) {
 
     SignalLogger.register(log)
 
@@ -122,7 +123,8 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       val driver = fetcher.actorSelection(driverUrl)
       val timeout = AkkaUtils.askTimeout(executorConf)
       val fut = Patterns.ask(driver, RetrieveSparkProps, timeout)
-      val props = Await.result(fut, timeout).asInstanceOf[Seq[(String, String)]]
+      val props = Await.result(fut, timeout).asInstanceOf[Seq[(String, String)]] ++
+        Seq[(String, String)](("spark.app.id", appId))
       fetcher.shutdown()
 
       // Create a new ActorSystem using driver's Spark properties to run the backend.
@@ -144,16 +146,16 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
 
   def main(args: Array[String]) {
     args.length match {
-      case x if x < 4 =>
+      case x if x < 5 =>
         System.err.println(
           // Worker url is used in spark standalone mode to enforce fate-sharing with worker
           "Usage: CoarseGrainedExecutorBackend <driverUrl> <executorId> <hostname> " +
           "<cores> [<workerUrl>]")
         System.exit(1)
-      case 4 =>
-        run(args(0), args(1), args(2), args(3).toInt, None)
-      case x if x > 4 =>
-        run(args(0), args(1), args(2), args(3).toInt, Some(args(4)))
+      case 5 =>
+        run(args(0), args(1), args(2), args(3).toInt, None, args(4))
+      case x if x > 5 =>
+        run(args(0), args(1), args(2), args(3).toInt, Some(args(4)), args(5))
     }
   }
 }

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

@@ -80,8 +80,6 @@ private[spark] class MetricsSystem private (val instance: String,
   def getServletHandlers = metricsServlet.map(_.getHandlers).getOrElse(Array())
 
   metricsConfig.initialize()
-  registerSources()
-  registerSinks()
 
   def start() {
     sinks.foreach(_.start)
@@ -96,11 +94,14 @@ private[spark] class MetricsSystem private (val instance: String,
   }
 
   def buildRegistryName(source: Source) = {
-    val appNameOpt = conf.getOption("spark.unique.app.name")
+    val appNameOpt = conf.getOption("spark.app.name")
+    val appIdOpt = conf.getOption("spark.app.id")
     val executorIdOpt = conf.getOption("spark.executor.id")
     val registryName = {
-      if (appNameOpt.isDefined && executorIdOpt.isDefined) {
-        MetricRegistry.name(appNameOpt.get, executorIdOpt.get, source.sourceName)
+      if (appNameOpt.isDefined && appIdOpt.isDefined &&
+        executorIdOpt.isDefined) {
+        MetricRegistry.name(appIdOpt.get, appNameOpt.get,
+          executorIdOpt.get, source.sourceName)
       } else {
         MetricRegistry.name(source.sourceName)
       }

yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala

@@ -47,6 +47,7 @@ class ExecutorRunnable(
     hostname: String,
     executorMemory: Int,
     executorCores: Int,
+    appAttemptId: String,
     securityMgr: SecurityManager)
   extends Runnable with ExecutorRunnableUtil with Logging {
 
@@ -83,7 +84,7 @@ class ExecutorRunnable(
     ctx.setContainerTokens(ByteBuffer.wrap(dob.getData()))
 
     val commands = prepareCommand(masterAddress, slaveId, hostname, executorMemory, executorCores,
-      localResources)
+      appAttemptId, localResources)
     logInfo("Setting up executor with commands: " + commands)
     ctx.setCommands(commands)
 

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnableUtil.scala

@@ -44,6 +44,7 @@ trait ExecutorRunnableUtil extends Logging {
       hostname: String,
       executorMemory: Int,
       executorCores: Int,
+      appId: String,
       localResources: HashMap[String, LocalResource]): List[String] = {
     // Extra options for the JVM
     val javaOpts = ListBuffer[String]()
@@ -112,6 +113,7 @@ trait ExecutorRunnableUtil extends Logging {
       slaveId.toString,
       hostname.toString,
       executorCores.toString,
+      appId,
       "1>", ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout",
       "2>", ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr")
 

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -56,6 +56,7 @@ object AllocationType extends Enumeration {
 private[yarn] abstract class YarnAllocator(
     conf: Configuration,
     sparkConf: SparkConf,
+    appAttemptId: ApplicationAttemptId,
     args: ApplicationMasterArguments,
     preferredNodes: collection.Map[String, collection.Set[SplitInfo]],
     securityMgr: SecurityManager)
@@ -295,6 +296,7 @@ private[yarn] abstract class YarnAllocator(
             executorHostname,
             executorMemory,
             executorCores,
+            appAttemptId.getApplicationId.toString,
             securityMgr)
           launcherPool.execute(executorRunnable)
         }

yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala

@@ -47,6 +47,7 @@ class ExecutorRunnable(
     hostname: String,
     executorMemory: Int,
     executorCores: Int,
+    appId: String,
     securityMgr: SecurityManager)
   extends Runnable with ExecutorRunnableUtil with Logging {
 
@@ -80,7 +81,7 @@ class ExecutorRunnable(
     ctx.setTokens(ByteBuffer.wrap(dob.getData()))
 
     val commands = prepareCommand(masterAddress, slaveId, hostname, executorMemory, executorCores,
-      localResources)
+      appId, localResources)
 
     logInfo(s"Setting up executor with environment: $env")
     logInfo("Setting up executor with commands: " + commands)

yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala

@@ -41,7 +41,7 @@ private[yarn] class YarnAllocationHandler(
     args: ApplicationMasterArguments,
     preferredNodes: collection.Map[String, collection.Set[SplitInfo]], 
     securityMgr: SecurityManager)
-  extends YarnAllocator(conf, sparkConf, args, preferredNodes, securityMgr) {
+  extends YarnAllocator(conf, sparkConf, appAttemptId, args, preferredNodes, securityMgr) {
 
   override protected def releaseContainer(container: Container) = {
     amClient.releaseAssignedContainer(container.getId())