You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2013/10/09 20:09:04 UTC
[1/7] git commit: SPARK-900 Use coarser grained naming for metrics
Updated Branches:
refs/heads/master 3218fa795 -> 19d445d37
SPARK-900 Use coarser grained naming for metrics
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/4b68be5f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/4b68be5f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/4b68be5f
Branch: refs/heads/master
Commit: 4b68be5f3c0a251453c184b233b3ca490812dafd
Parents: 714fdab
Author: Grace Huang <ji...@intel.com>
Authored: Fri Sep 27 14:47:38 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Fri Sep 27 14:47:38 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 5 +-
.../spark/deploy/master/MasterSource.scala | 7 ++-
.../spark/deploy/worker/WorkerSource.scala | 11 ++--
.../apache/spark/executor/ExecutorSource.scala | 19 +++---
.../spark/scheduler/DAGSchedulerSource.scala | 11 ++--
.../spark/storage/BlockManagerSource.scala | 9 +--
.../apache/spark/util/NamingConventions.scala | 62 ++++++++++++++++++++
7 files changed, 96 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index 5a24042..c72322e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -20,6 +20,7 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
+import org.apache.spark.util.NamingConventions
class ApplicationSource(val application: ApplicationInfo) extends Source {
val metricRegistry = new MetricRegistry()
@@ -30,11 +31,11 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(MetricRegistry.name("runtime_ms"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name("cores", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index 23d1cb7..de39398 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -20,23 +20,24 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
+import org.apache.spark.util.NamingConventions
private[spark] class MasterSource(val master: Master) extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name("workers","number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name("apps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name("waitingApps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index df269fd..fc4f4ae 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -20,32 +20,33 @@ package org.apache.spark.deploy.worker
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
+import org.apache.spark.util.NamingConventions
private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name("executors", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name("coresUsed", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name("memUsed", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name("coresFree", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name("memFree", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 18c9dc1..6cbd154 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -26,6 +26,7 @@ import org.apache.hadoop.fs.LocalFileSystem
import scala.collection.JavaConversions._
import org.apache.spark.metrics.source.Source
+import org.apache.spark.util.NamingConventions
class ExecutorSource(val executor: Executor, executorId: String) extends Source {
private def fileStats(scheme: String) : Option[FileSystem.Statistics] =
@@ -43,31 +44,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", "activeTask", "count"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", "completeTask", "count"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", "currentPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getPoolSize()
})
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", "maxPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getMaximumPoolSize()
})
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, "bytesRead", _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, "bytesWritten", _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, "readOps", _.getReadOps(), 0)
- registerFileSystemStat(scheme, "largeReadOps", _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, "writeOps", _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 446d490..9e90a08 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -21,29 +21,30 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
+import org.apache.spark.util.NamingConventions
private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: SparkContext)
extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", "failedStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", "runningStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", "waitingStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", "allJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", "activeJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index acc3951..4312250 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -21,6 +21,7 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
+import org.apache.spark.util.NamingConventions
private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: SparkContext)
@@ -28,7 +29,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", "maxMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -36,7 +37,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "remainingMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -44,7 +45,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "memUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -53,7 +54,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4b68be5f/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/NamingConventions.scala b/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
new file mode 100644
index 0000000..7263361
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
@@ -0,0 +1,62 @@
+/*
+ * 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.util
+
+/**
+ * all utilities related to naming conventions
+ */
+private[spark] object NamingConventions {
+
+ /**
+ * Lower camelCase which convert the phrases into camelCase style with the first letter lowercase
+ */
+ def lowerCamelCaseConversion(phrases: Seq[String]): Seq[String] = {
+ var first = true
+
+ for (elem <- phrases) yield {
+ if (first) {
+ first = false
+ elem
+ }
+ else {
+ elem.capitalize
+ }
+ }
+ }
+
+ /**
+ * The standard camelCase style
+ */
+ def camelCaseConversion(phrases: Seq[String]): Seq[String] = {
+ phrases.map(_.capitalize)
+ }
+
+ def noConversion = { x: Seq[String] => x }
+
+ /**
+ * Concatenate the words using certain naming style.
+ * The default style is lowerCamelCase with empty connector.
+ */
+ def makeIdentifier(phrases: Seq[String], namingConversion: (Seq[String]) => Seq[String] = lowerCamelCaseConversion) (implicit connector: String = "" ): String = {
+ namingConversion(phrases.filter(_.nonEmpty)).mkString(connector)
+ }
+
+ def makeMetricName(phrases: String*): String = {
+ makeIdentifier(phrases, noConversion)("_")
+ }
+}
\ No newline at end of file
[6/7] git commit: remove those futile suffixes like number/count
Posted by pw...@apache.org.
remove those futile suffixes like number/count
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/f7628e40
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/f7628e40
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/f7628e40
Branch: refs/heads/master
Commit: f7628e40330a1721e3d69ccb2390dfa75f839768
Parents: 22bed59
Author: Grace Huang <ji...@intel.com>
Authored: Wed Oct 9 08:36:41 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Wed Oct 9 08:36:41 2013 +0800
----------------------------------------------------------------------
.../apache/spark/deploy/master/ApplicationSource.scala | 2 +-
.../org/apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../org/apache/spark/deploy/worker/WorkerSource.scala | 6 +++---
.../scala/org/apache/spark/executor/ExecutorSource.scala | 4 ++--
.../org/apache/spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
5 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index 01dbb3d..c87b66f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -34,7 +34,7 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name("cores_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index a0f6fc4..36c1b87 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -26,17 +26,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name("workers_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name("apps_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name("waitingApps_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index 5c2d3ee..b24e936 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -25,12 +25,12 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name("executors_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name("coresUsed_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
@@ -40,7 +40,7 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name("coresFree_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 2123904..34ed9c8 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -43,12 +43,12 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", "activeTask_count"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTasks"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", "completeTask_count"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTasks"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 474f3e8..1515148 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -27,23 +27,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", "failedStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", "runningStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", "waitingStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", "allJobs_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", "activeJobs_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
[7/7] git commit: Merge pull request #22 from GraceH/metrics-naming
Posted by pw...@apache.org.
Merge pull request #22 from GraceH/metrics-naming
SPARK-900 Use coarser grained naming for metrics
see SPARK-900 Use coarser grained naming for metrics.
Now the new metric name is formatted as {XXX.YYY.ZZZ.COUNTER_UNIT}, XXX.YYY.ZZZ represents the group name, which can group several metrics under the same Ganglia view.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/19d445d3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/19d445d3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/19d445d3
Branch: refs/heads/master
Commit: 19d445d37c38a35cc68da91f53a8780b89a1f8c9
Parents: 3218fa7 f7628e4
Author: Patrick Wendell <pw...@gmail.com>
Authored: Wed Oct 9 11:08:34 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Wed Oct 9 11:08:34 2013 -0700
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../apache/spark/deploy/worker/WorkerSource.scala | 10 +++++-----
.../apache/spark/executor/ExecutorSource.scala | 18 +++++++++---------
.../spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
.../apache/spark/storage/BlockManagerSource.scala | 8 ++++----
6 files changed, 27 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
[3/7] git commit: Revert "remedy the line-wrap while exceeding 100
chars"
Posted by pw...@apache.org.
Revert "remedy the line-wrap while exceeding 100 chars"
This reverts commit 892fb8ffa85016a63d7d00dd6f1abc58ccf034a2.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a2af6b54
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a2af6b54
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a2af6b54
Branch: refs/heads/master
Commit: a2af6b543a0a70d94a451c9022deea181d04f8e8
Parents: 892fb8f
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 17:44:56 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 17:44:56 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 13 ++--
.../spark/deploy/master/MasterSource.scala | 19 +++---
.../spark/deploy/worker/WorkerSource.scala | 31 +++++----
.../apache/spark/executor/ExecutorSource.scala | 45 +++++--------
.../spark/scheduler/DAGSchedulerSource.scala | 31 +++++----
.../spark/storage/BlockManagerSource.scala | 66 +++++++++-----------
6 files changed, 91 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index f0b1f77..c72322e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -31,13 +31,12 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")),
- new Gauge[Long] { override def getValue: Long = application.duration })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
+ override def getValue: Long = application.duration
+ })
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")),
- new Gauge[Int] { override def getValue: Int = application.coresGranted })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
+ override def getValue: Int = application.coresGranted
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index 8a88fef..de39398 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -27,18 +27,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("workers","number")),
- new Gauge[Int] { override def getValue: Int = master.workers.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
+ override def getValue: Int = master.workers.size
+ })
// Gauge for application numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")),
- new Gauge[Int] { override def getValue: Int = master.apps.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
+ override def getValue: Int = master.apps.size
+ })
// Gauge for waiting application numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")),
- new Gauge[Int] { override def getValue: Int = master.waitingApps.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
+ override def getValue: Int = master.waitingApps.size
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index 0596f14..fc4f4ae 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -26,28 +26,27 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")),
- new Gauge[Int] { override def getValue: Int = worker.executors.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.executors.size
+ })
// Gauge for cores used of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")),
- new Gauge[Int] { override def getValue: Int = worker.coresUsed })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.coresUsed
+ })
// Gauge for memory used of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")),
- new Gauge[Int] { override def getValue: Int = worker.memoryUsed })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
+ override def getValue: Int = worker.memoryUsed
+ })
// Gauge for cores free of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")),
- new Gauge[Int] { override def getValue: Int = worker.coresFree })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.coresFree
+ })
// Gauge for memory free of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")),
- new Gauge[Int] { override def getValue: Int = worker.memoryFree })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
+ override def getValue: Int = worker.memoryFree
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index d063e4a..6cbd154 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -44,42 +44,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getActiveCount() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getActiveCount()
+ })
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")),
- new Gauge[Long] { override def getValue: Long = executor.threadPool.getCompletedTaskCount() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
+ override def getValue: Long = executor.threadPool.getCompletedTaskCount()
+ })
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getPoolSize() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getPoolSize()
+ })
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getMaximumPoolSize() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getMaximumPoolSize()
+ })
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("read", "bytes"),
- _.getBytesRead(), 0L)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("write", "bytes"),
- _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("read", "ops"),
- _.getReadOps(), 0)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("largeRead", "ops"),
- _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("write", "ops"),
- _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
}
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 02fb807..9e90a08 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -28,24 +28,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.failed.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.failed.size
+ })
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.running.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.running.size
+ })
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.waiting.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.waiting.size
+ })
- metricRegistry.register(
- MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.nextJobId.get() })
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.nextJobId.get()
+ })
- metricRegistry.register(
- MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.activeJobs.size })
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.activeJobs.size
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index fcf9da4..4312250 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -29,48 +29,40 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- maxMem / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ maxMem / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- remainingMem / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ remainingMem / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- (maxMem - remainingMem) / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ (maxMem - remainingMem) / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val diskSpaceUsed = storageStatusList
- .flatMap(_.blocks.values.map(_.diskSize))
- .reduceOption(_ + _)
- .getOrElse(0L)
-
- diskSpaceUsed / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val diskSpaceUsed = storageStatusList
+ .flatMap(_.blocks.values.map(_.diskSize))
+ .reduceOption(_ + _)
+ .getOrElse(0L)
+
+ diskSpaceUsed / 1024 / 1024
+ }
})
}
-
[2/7] git commit: remedy the line-wrap while exceeding 100 chars
Posted by pw...@apache.org.
remedy the line-wrap while exceeding 100 chars
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/892fb8ff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/892fb8ff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/892fb8ff
Branch: refs/heads/master
Commit: 892fb8ffa85016a63d7d00dd6f1abc58ccf034a2
Parents: 4b68be5
Author: Grace Huang <ji...@intel.com>
Authored: Mon Sep 30 20:12:55 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Mon Sep 30 20:12:55 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 13 ++--
.../spark/deploy/master/MasterSource.scala | 19 +++---
.../spark/deploy/worker/WorkerSource.scala | 31 ++++-----
.../apache/spark/executor/ExecutorSource.scala | 45 ++++++++-----
.../spark/scheduler/DAGSchedulerSource.scala | 31 ++++-----
.../spark/storage/BlockManagerSource.scala | 66 +++++++++++---------
6 files changed, 114 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index c72322e..f0b1f77 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -31,12 +31,13 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
- override def getValue: Long = application.duration
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")),
+ new Gauge[Long] { override def getValue: Long = application.duration })
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
- override def getValue: Int = application.coresGranted
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")),
+ new Gauge[Int] { override def getValue: Int = application.coresGranted })
}
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index de39398..8a88fef 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -27,17 +27,18 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
- override def getValue: Int = master.workers.size
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("workers","number")),
+ new Gauge[Int] { override def getValue: Int = master.workers.size })
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
- override def getValue: Int = master.apps.size
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")),
+ new Gauge[Int] { override def getValue: Int = master.apps.size })
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
- override def getValue: Int = master.waitingApps.size
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")),
+ new Gauge[Int] { override def getValue: Int = master.waitingApps.size })
}
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index fc4f4ae..0596f14 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -26,27 +26,28 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
- override def getValue: Int = worker.executors.size
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")),
+ new Gauge[Int] { override def getValue: Int = worker.executors.size })
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
- override def getValue: Int = worker.coresUsed
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")),
+ new Gauge[Int] { override def getValue: Int = worker.coresUsed })
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
- override def getValue: Int = worker.memoryUsed
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")),
+ new Gauge[Int] { override def getValue: Int = worker.memoryUsed })
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
- override def getValue: Int = worker.coresFree
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")),
+ new Gauge[Int] { override def getValue: Int = worker.coresFree })
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
- override def getValue: Int = worker.memoryFree
- })
+ metricRegistry.register(
+ MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")),
+ new Gauge[Int] { override def getValue: Int = worker.memoryFree })
}
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 6cbd154..d063e4a 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -44,31 +44,42 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
- override def getValue: Int = executor.threadPool.getActiveCount()
- })
+ metricRegistry.register(
+ MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")),
+ new Gauge[Int] { override def getValue: Int = executor.threadPool.getActiveCount() })
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
- override def getValue: Long = executor.threadPool.getCompletedTaskCount()
- })
+ metricRegistry.register(
+ MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")),
+ new Gauge[Long] { override def getValue: Long = executor.threadPool.getCompletedTaskCount() })
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
- override def getValue: Int = executor.threadPool.getPoolSize()
- })
+ metricRegistry.register(
+ MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")),
+ new Gauge[Int] { override def getValue: Int = executor.threadPool.getPoolSize() })
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
- override def getValue: Int = executor.threadPool.getMaximumPoolSize()
- })
+ metricRegistry.register(
+ MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")),
+ new Gauge[Int] { override def getValue: Int = executor.threadPool.getMaximumPoolSize() })
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
+ registerFileSystemStat(scheme,
+ NamingConventions.makeMetricName("read", "bytes"),
+ _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme,
+ NamingConventions.makeMetricName("write", "bytes"),
+ _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme,
+ NamingConventions.makeMetricName("read", "ops"),
+ _.getReadOps(), 0)
+ registerFileSystemStat(scheme,
+ NamingConventions.makeMetricName("largeRead", "ops"),
+ _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme,
+ NamingConventions.makeMetricName("write", "ops"),
+ _.getWriteOps(), 0)
}
}
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 9e90a08..02fb807 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -28,23 +28,24 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
- override def getValue: Int = dagScheduler.failed.size
- })
+ metricRegistry.register(
+ MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")),
+ new Gauge[Int] { override def getValue: Int = dagScheduler.failed.size })
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
- override def getValue: Int = dagScheduler.running.size
- })
+ metricRegistry.register(
+ MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")),
+ new Gauge[Int] { override def getValue: Int = dagScheduler.running.size })
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
- override def getValue: Int = dagScheduler.waiting.size
- })
+ metricRegistry.register(
+ MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")),
+ new Gauge[Int] { override def getValue: Int = dagScheduler.waiting.size })
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
- override def getValue: Int = dagScheduler.nextJobId.get()
- })
+ metricRegistry.register(
+ MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")),
+ new Gauge[Int] { override def getValue: Int = dagScheduler.nextJobId.get() })
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
- override def getValue: Int = dagScheduler.activeJobs.size
- })
+ metricRegistry.register(
+ MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")),
+ new Gauge[Int] { override def getValue: Int = dagScheduler.activeJobs.size })
}
+
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/892fb8ff/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index 4312250..fcf9da4 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -29,40 +29,48 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- maxMem / 1024 / 1024
- }
+ metricRegistry.register(
+ MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")),
+ new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ maxMem / 1024 / 1024
+ }
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- remainingMem / 1024 / 1024
- }
+ metricRegistry.register(
+ MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")),
+ new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ remainingMem / 1024 / 1024
+ }
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- (maxMem - remainingMem) / 1024 / 1024
- }
+ metricRegistry.register(
+ MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")),
+ new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ (maxMem - remainingMem) / 1024 / 1024
+ }
})
- metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val diskSpaceUsed = storageStatusList
- .flatMap(_.blocks.values.map(_.diskSize))
- .reduceOption(_ + _)
- .getOrElse(0L)
-
- diskSpaceUsed / 1024 / 1024
- }
+ metricRegistry.register(
+ MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val diskSpaceUsed = storageStatusList
+ .flatMap(_.blocks.values.map(_.diskSize))
+ .reduceOption(_ + _)
+ .getOrElse(0L)
+
+ diskSpaceUsed / 1024 / 1024
+ }
})
}
+
[5/7] git commit: create metrics name manually.
Posted by pw...@apache.org.
create metrics name manually.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/22bed59d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/22bed59d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/22bed59d
Branch: refs/heads/master
Commit: 22bed59d2de15d89ab5a778bea7439711e3cb13b
Parents: 188abbf
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 18:01:11 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 18:01:11 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../apache/spark/deploy/worker/WorkerSource.scala | 10 +++++-----
.../apache/spark/executor/ExecutorSource.scala | 18 +++++++++---------
.../spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
.../apache/spark/storage/BlockManagerSource.scala | 8 ++++----
6 files changed, 27 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index 5a24042..01dbb3d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -34,7 +34,7 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name("cores", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores_number"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index 23d1cb7..a0f6fc4 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -26,17 +26,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name("workers","number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers_number"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name("apps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps_number"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name("waitingApps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps_number"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index df269fd..5c2d3ee 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -25,27 +25,27 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name("executors", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors_number"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name("coresUsed", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed_number"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name("memUsed", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memUsed_MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name("coresFree", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree_number"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name("memFree", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memFree_MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 18c9dc1..2123904 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -43,31 +43,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", "activeTask", "count"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTask_count"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", "completeTask", "count"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTask_count"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", "currentPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "currentPool_size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getPoolSize()
})
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", "maxPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "maxPool_size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getMaximumPoolSize()
})
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, "bytesRead", _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, "bytesWritten", _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, "readOps", _.getReadOps(), 0)
- registerFileSystemStat(scheme, "largeReadOps", _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, "writeOps", _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, "read_bytes", _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, "write_bytes", _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, "read_ops", _.getReadOps(), 0)
+ registerFileSystemStat(scheme, "largeRead_ops", _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, "write_ops", _.getWriteOps(), 0)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 446d490..474f3e8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -27,23 +27,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", "failedStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", "runningStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", "waitingStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", "allJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", "activeJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index acc3951..e5068d5 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -28,7 +28,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", "maxMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "maxMem_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -36,7 +36,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "remainingMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "remainingMem_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -44,7 +44,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "memUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "memUsed_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -53,7 +53,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
[4/7] git commit: Revert "SPARK-900 Use coarser grained naming for
metrics"
Posted by pw...@apache.org.
Revert "SPARK-900 Use coarser grained naming for metrics"
This reverts commit 4b68be5f3c0a251453c184b233b3ca490812dafd.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/188abbf8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/188abbf8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/188abbf8
Branch: refs/heads/master
Commit: 188abbf8f1765d3f52eff8b852309352630ab8fb
Parents: a2af6b5
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 17:45:14 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 17:45:14 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 5 +-
.../spark/deploy/master/MasterSource.scala | 7 +--
.../spark/deploy/worker/WorkerSource.scala | 11 ++--
.../apache/spark/executor/ExecutorSource.scala | 19 +++---
.../spark/scheduler/DAGSchedulerSource.scala | 11 ++--
.../spark/storage/BlockManagerSource.scala | 9 ++-
.../apache/spark/util/NamingConventions.scala | 62 --------------------
7 files changed, 28 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index c72322e..5a24042 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -20,7 +20,6 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
class ApplicationSource(val application: ApplicationInfo) extends Source {
val metricRegistry = new MetricRegistry()
@@ -31,11 +30,11 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("runtime_ms"), new Gauge[Long] {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores", "number"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index de39398..23d1cb7 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -20,24 +20,23 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
private[spark] class MasterSource(val master: Master) extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers","number"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps", "number"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps", "number"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index fc4f4ae..df269fd 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -20,33 +20,32 @@ package org.apache.spark.deploy.worker
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors", "number"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed", "number"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memUsed", "MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree", "number"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memFree", "MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 6cbd154..18c9dc1 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -26,7 +26,6 @@ import org.apache.hadoop.fs.LocalFileSystem
import scala.collection.JavaConversions._
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
class ExecutorSource(val executor: Executor, executorId: String) extends Source {
private def fileStats(scheme: String) : Option[FileSystem.Statistics] =
@@ -44,31 +43,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTask", "count"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTask", "count"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "currentPool", "size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getPoolSize()
})
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "maxPool", "size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getMaximumPoolSize()
})
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, "bytesRead", _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, "bytesWritten", _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, "readOps", _.getReadOps(), 0)
+ registerFileSystemStat(scheme, "largeReadOps", _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, "writeOps", _.getWriteOps(), 0)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 9e90a08..446d490 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -21,30 +21,29 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
-import org.apache.spark.util.NamingConventions
private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: SparkContext)
extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index 4312250..acc3951 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -21,7 +21,6 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
-import org.apache.spark.util.NamingConventions
private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: SparkContext)
@@ -29,7 +28,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "maxMem", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -37,7 +36,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "remainingMem", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -45,7 +44,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "memUsed", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -54,7 +53,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/NamingConventions.scala b/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
deleted file mode 100644
index 7263361..0000000
--- a/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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.util
-
-/**
- * all utilities related to naming conventions
- */
-private[spark] object NamingConventions {
-
- /**
- * Lower camelCase which convert the phrases into camelCase style with the first letter lowercase
- */
- def lowerCamelCaseConversion(phrases: Seq[String]): Seq[String] = {
- var first = true
-
- for (elem <- phrases) yield {
- if (first) {
- first = false
- elem
- }
- else {
- elem.capitalize
- }
- }
- }
-
- /**
- * The standard camelCase style
- */
- def camelCaseConversion(phrases: Seq[String]): Seq[String] = {
- phrases.map(_.capitalize)
- }
-
- def noConversion = { x: Seq[String] => x }
-
- /**
- * Concatenate the words using certain naming style.
- * The default style is lowerCamelCase with empty connector.
- */
- def makeIdentifier(phrases: Seq[String], namingConversion: (Seq[String]) => Seq[String] = lowerCamelCaseConversion) (implicit connector: String = "" ): String = {
- namingConversion(phrases.filter(_.nonEmpty)).mkString(connector)
- }
-
- def makeMetricName(phrases: String*): String = {
- makeIdentifier(phrases, noConversion)("_")
- }
-}
\ No newline at end of file