You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/01/28 08:03:03 UTC
[incubator-celeborn] branch main updated: [CELEBORN-229][FOLLOWUP] Support collect metrics with customized labels (#1174)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new d6d537df [CELEBORN-229][FOLLOWUP] Support collect metrics with customized labels (#1174)
d6d537df is described below
commit d6d537df93cc44492c86fc06ca867ca38e56661a
Author: nafiy <30...@users.noreply.github.com>
AuthorDate: Sat Jan 28 16:02:58 2023 +0800
[CELEBORN-229][FOLLOWUP] Support collect metrics with customized labels (#1174)
---
.../celeborn/common/metrics/MetricLabels.scala | 36 ++++++++++
.../common/metrics/source/AbstractSource.scala | 77 +++++++++++++++-------
.../metrics/source/CelebornSourceSuite.scala | 26 +++++---
3 files changed, 108 insertions(+), 31 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/metrics/MetricLabels.scala b/common/src/main/scala/org/apache/celeborn/common/metrics/MetricLabels.scala
new file mode 100644
index 00000000..b7fa05cb
--- /dev/null
+++ b/common/src/main/scala/org/apache/celeborn/common/metrics/MetricLabels.scala
@@ -0,0 +1,36 @@
+/*
+ * 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.celeborn.common.metrics
+
+private[metrics] trait MetricLabels {
+ val labels: Map[String, String]
+
+ def labelString: String = MetricLabels.labelString(labels)
+}
+
+object MetricLabels {
+ def labelString(labels: Map[String, String]): String = {
+ "{" +
+ labels
+ .map { case (key: String, value: String) => s"""$key="$value"""" }
+ .toList
+ .sorted
+ .mkString(" ") +
+ "}"
+ }
+}
diff --git a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
index 6bdf85b8..27681c36 100644
--- a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
@@ -27,17 +27,9 @@ import com.codahale.metrics._
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.metrics.{ResettableSlidingWindowReservoir, RssHistogram, RssTimer}
+import org.apache.celeborn.common.metrics.{MetricLabels, ResettableSlidingWindowReservoir, RssHistogram, RssTimer}
import org.apache.celeborn.common.util.{ThreadUtils, Utils}
-private[source] trait MetricLabels {
- val labels: Map[String, String]
- def labelString: String = {
- "{" + labels.map { case (key: String, value: String) => s"""$key="$value"""" }.mkString(
- " ") + "}"
- }
-}
-
case class NamedCounter(name: String, counter: Counter, labels: Map[String, String])
extends MetricLabels
@@ -98,7 +90,9 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
def addTimer(name: String, labels: Map[String, String]): Unit = {
val namedTimer = NamedTimer(name, metricRegistry.timer(name, timerSupplier), labels + roleLabel)
- namedTimers.putIfAbsent(name, (namedTimer, new ConcurrentHashMap[String, Long]()))
+ namedTimers.putIfAbsent(
+ metricNameWithLabels(name, labels + roleLabel),
+ (namedTimer, new ConcurrentHashMap[String, Long]()))
}
protected val namedCounters: ConcurrentHashMap[String, NamedCounter] =
@@ -106,8 +100,10 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
def addCounter(name: String): Unit = addCounter(name, Map.empty[String, String])
- def addCounter(name: String, labels: Map[String, String] = Map.empty[String, String]): Unit = {
- namedCounters.put(name, NamedCounter(name, metricRegistry.counter(name), labels + roleLabel))
+ def addCounter(name: String, labels: Map[String, String]): Unit = {
+ namedCounters.put(
+ metricNameWithLabels(name, labels + roleLabel),
+ NamedCounter(name, metricRegistry.counter(name), labels + roleLabel))
}
protected def counters(): List[NamedCounter] = {
@@ -137,43 +133,65 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
}
override def sample[T](metricsName: String, key: String)(f: => T): T = {
+ sample(metricsName, key, Map.empty[String, String])(f)
+ }
+
+ def sample[T](metricsName: String, key: String, labels: Map[String, String])(f: => T): T = {
val sample = needSample()
var r: Any = null
try {
if (sample) {
- doStartTimer(metricsName, key)
+ doStartTimer(metricsName, key, labels)
}
r = f
} finally {
if (sample) {
- doStopTimer(metricsName, key)
+ doStopTimer(metricsName, key, labels)
}
}
r.asInstanceOf[T]
}
override def startTimer(metricsName: String, key: String): Unit = {
+ startTimer(metricsName, key, Map.empty[String, String])
+ }
+
+ def startTimer(metricsName: String, key: String, labels: Map[String, String]): Unit = {
if (needSample()) {
- doStartTimer(metricsName, key)
+ doStartTimer(metricsName, key, labels)
}
}
override def stopTimer(metricsName: String, key: String): Unit = {
- doStopTimer(metricsName, key)
+ stopTimer(metricsName, key, Map.empty[String, String])
+ }
+
+ def stopTimer(metricsName: String, key: String, labels: Map[String, String]): Unit = {
+ doStopTimer(metricsName, key, labels)
}
def doStartTimer(metricsName: String, key: String): Unit = {
- val pair = namedTimers.get(metricsName)
+ doStartTimer(metricsName, key, Map.empty[String, String])
+ }
+
+ def doStartTimer(metricsName: String, key: String, labels: Map[String, String]): Unit = {
+ val name = metricNameWithLabels(metricsName, labels + roleLabel)
+ val pair = namedTimers.get(name)
if (pair != null) {
pair._2.put(key, System.nanoTime())
} else {
- logWarning(s"Metric $metricsName not found!")
+ logWarning(s"Metric $name not found!")
}
}
protected def doStopTimer(metricsName: String, key: String): Unit = {
+ doStopTimer(metricsName, key, Map.empty[String, String])
+ }
+
+ protected def doStopTimer(metricsName: String, key: String, labels: Map[String, String]): Unit = {
+ val name = metricNameWithLabels(metricsName, labels + roleLabel)
try {
- val (namedTimer, map) = namedTimers.get(metricsName)
+ val (namedTimer, map) = namedTimers.get(name)
val startTime = Option(map.remove(key))
startTime match {
case Some(t) =>
@@ -185,16 +203,25 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
}
} catch {
case e: Exception =>
- logWarning("Exception encountered in Metrics StopTimer", e)
+ logWarning(s"Exception encountered during stop timer of metric $name", e)
}
}
- override def incCounter(metricsName: String, incV: Long = 1): Unit = {
- val counter = namedCounters.get(metricsName)
+ def incCounter(metricsName: String): Unit = {
+ incCounter(metricsName, 1)
+ }
+
+ override def incCounter(metricsName: String, incV: Long): Unit = {
+ incCounter(metricsName, incV, Map.empty[String, String])
+ }
+
+ def incCounter(metricsName: String, incV: Long, labels: Map[String, String]): Unit = {
+ val name = metricNameWithLabels(metricsName, labels + roleLabel)
+ val counter = namedCounters.get(name)
if (counter != null) {
counter.counter.inc(incV)
} else {
- logWarning(s"Metric $metricsName not found!")
+ logWarning(s"Metric $name not found!")
}
}
@@ -328,6 +355,10 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
protected def reportNanosAsMills(value: Double): Double = {
BigDecimal(value / 1000000).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
}
+
+ protected def metricNameWithLabels(metricsName: String, labels: Map[String, String]): String = {
+ metricsName + MetricLabels.labelString(labels)
+ }
}
class TimerSupplier(val slidingWindowSize: Int)
diff --git a/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala b/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
index 1cbec78d..211792ca 100644
--- a/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
@@ -26,22 +26,32 @@ class CelebornSourceSuite extends CelebornFunSuite {
val mockSource = new AbstractSource(new CelebornConf(), "mock") {
override def sourceName: String = "mockSource"
}
+ val user1 = Map("user" -> "user1")
+ val user2 = Map("user" -> "user2")
+ val user3 = Map("user" -> "user3")
mockSource.addGauge("Gauge1", _ => 1000)
- mockSource.addGauge("Gauge2", _ => 2000, Map("user" -> "user1"))
+ mockSource.addGauge("Gauge2", _ => 2000, user1)
mockSource.addCounter("Counter1")
+ mockSource.addCounter("Counter2", user2)
+ // test operation with and without label
mockSource.incCounter("Counter1", 3000)
- mockSource.addCounter("Counter2", Map("user" -> "user2"))
- mockSource.incCounter("Counter2", 4000)
+ mockSource.incCounter("Counter2", 4000, user2)
mockSource.addTimer("Timer1")
- mockSource.addTimer("Timer2", Map("user" -> "user3"))
+ mockSource.addTimer("Timer2", user3)
+ // ditto
+ mockSource.startTimer("Timer1", "key1")
+ mockSource.startTimer("Timer2", "key2", user3)
+ Thread.sleep(10)
+ mockSource.stopTimer("Timer1", "key1")
+ mockSource.stopTimer("Timer2", "key2", user3)
val res = mockSource.getMetrics()
val exp1 = "metrics_Gauge1_Value{role=\"mock\"} 1000"
- val exp2 = "metrics_Gauge2_Value{user=\"user1\" role=\"mock\"} 2000"
+ val exp2 = "metrics_Gauge2_Value{role=\"mock\" user=\"user1\"} 2000"
val exp3 = "metrics_Counter1_Count{role=\"mock\"} 3000"
- val exp4 = "metrics_Counter2_Count{user=\"user2\" role=\"mock\"} 4000"
- val exp5 = "metrics_Timer1_Count{role=\"mock\"} 0"
- val exp6 = "metrics_Timer2_Count{user=\"user3\" role=\"mock\"} 0"
+ val exp4 = "metrics_Counter2_Count{role=\"mock\" user=\"user2\"} 4000"
+ val exp5 = "metrics_Timer1_Count{role=\"mock\"} 1"
+ val exp6 = "metrics_Timer2_Count{role=\"mock\" user=\"user3\"} 1"
assert(res.contains(exp1))
assert(res.contains(exp2))