You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by fe...@apache.org on 2025/05/13 04:05:53 UTC
(celeborn) branch main updated: [CELEBORN-1974] ApplicationId as metrics label should be behind a config flag
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new a547cdaef [CELEBORN-1974] ApplicationId as metrics label should be behind a config flag
a547cdaef is described below
commit a547cdaeff852d1050b97e452eb2f0eec98fbd3a
Author: Sanskar Modi <sa...@gmail.com>
AuthorDate: Mon May 12 21:05:45 2025 -0700
[CELEBORN-1974] ApplicationId as metrics label should be behind a config flag
### What changes were proposed in this pull request?
Push applicationId as metrics label only if `celeborn.metrics.worker.appLevel.enabled` is true.
### Why are the changes needed?
At Uber, We use m3 for monitoring, it tries to make a new series using all the present metrics label. Having applicationId as a metrics introduces too much cardinality in `activeconnectioncount` and we are unable to use it, while it is an useful metric with/without applicationId as label. Similarly for resourceConsumption, userIdentifier alone can be used.
### Does this PR introduce _any_ user-facing change?
Yes, changed the default config value.
### How was this patch tested?
NA
Closes #3221 from s0nskar/application_tag.
Lead-authored-by: Sanskar Modi <sa...@gmail.com>
Co-authored-by: Wang, Fei <fw...@ebay.com>
Signed-off-by: Wang, Fei <fw...@ebay.com>
---
.../org/apache/celeborn/common/CelebornConf.scala | 17 ++++++++-
docs/configuration/metrics.md | 3 +-
.../service/deploy/worker/WorkerSource.scala | 43 +++++++++++++---------
.../service/deploy/worker/WorkerSuite.scala | 1 +
4 files changed, 44 insertions(+), 20 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 6455e4051..d33028a9a 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -894,6 +894,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def metricsWorkerForceAppendPauseSpentTimeThreshold: Int =
get(METRICS_WORKER_PAUSE_SPENT_TIME_FORCE_APPEND_THRESHOLD)
def metricsJsonPrettyEnabled: Boolean = get(METRICS_JSON_PRETTY_ENABLED)
+ def metricsWorkerAppLevelEnabled: Boolean = get(METRICS_WORKER_APP_LEVEL_ENABLED)
// //////////////////////////////////////////////////////
// Quota //
@@ -5520,10 +5521,12 @@ object CelebornConf extends Logging {
.categories("metrics")
.doc("Size for top items about top resource consumption applications list of worker. " +
"The top resource consumption is determined by sum of diskBytesWritten and hdfsBytesWritten. " +
- "The top resource consumption count prevents the total number of metrics from exceeding the metrics capacity.")
+ "The top resource consumption count prevents the total number of metrics from exceeding the metrics capacity. " +
+ "Note: This will add applicationId as label which is considered as a high cardinality label, " +
+ "be careful enabling it on metrics systems that are not optimized for high cardinality columns.")
.version("0.6.0")
.intConf
- .createWithDefault(50)
+ .createWithDefault(0)
val METRICS_WORKER_APP_TOP_RESOURCE_CONSUMPTION_BYTES_WRITTEN_THRESHOLD: ConfigEntry[Long] =
buildConf("celeborn.metrics.worker.app.topResourceConsumption.bytesWrittenThreshold")
@@ -5569,6 +5572,16 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
+ val METRICS_WORKER_APP_LEVEL_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.metrics.worker.appLevel.enabled")
+ .categories("metrics")
+ .doc("When true, enable worker application level metrics. Note: applicationId is " +
+ "considered as a high cardinality label, be careful enabling it on metrics systems " +
+ "that are not optimized for high cardinality columns.")
+ .version("0.6.0")
+ .booleanConf
+ .createWithDefault(true)
+
val IDENTITY_PROVIDER: ConfigEntry[String] =
buildConf("celeborn.identity.provider")
.withAlternative("celeborn.quota.identity.provider")
diff --git a/docs/configuration/metrics.md b/docs/configuration/metrics.md
index b590f5628..77c7d6d59 100644
--- a/docs/configuration/metrics.md
+++ b/docs/configuration/metrics.md
@@ -30,6 +30,7 @@ license: |
| celeborn.metrics.sample.rate | 1.0 | false | It controls if Celeborn collect timer metrics for some operations. Its value should be in [0.0, 1.0]. | 0.2.0 | |
| celeborn.metrics.timer.slidingWindow.size | 4096 | false | The sliding window size of timer metric. | 0.2.0 | |
| celeborn.metrics.worker.app.topResourceConsumption.bytesWrittenThreshold | 0b | false | Threshold of bytes written for top resource consumption applications list of worker. The application which has bytes written less than this threshold will not be included in the top resource consumption list, including diskBytesWritten and hdfsBytesWritten. | 0.6.0 | |
-| celeborn.metrics.worker.app.topResourceConsumption.count | 50 | false | Size for top items about top resource consumption applications list of worker. The top resource consumption is determined by sum of diskBytesWritten and hdfsBytesWritten. The top resource consumption count prevents the total number of metrics from exceeding the metrics capacity. | 0.6.0 | |
+| celeborn.metrics.worker.app.topResourceConsumption.count | 0 | false | Size for top items about top resource consumption applications list of worker. The top resource consumption is determined by sum of diskBytesWritten and hdfsBytesWritten. The top resource consumption count prevents the total number of metrics from exceeding the metrics capacity. Note: This will add applicationId as label which is considered as a high cardinality label, be careful enabling it on metrics systems that [...]
+| celeborn.metrics.worker.appLevel.enabled | true | false | When true, enable worker application level metrics. Note: applicationId is considered as a high cardinality label, be careful enabling it on metrics systems that are not optimized for high cardinality columns. | 0.6.0 | |
| celeborn.metrics.worker.pauseSpentTime.forceAppend.threshold | 10 | false | Force append worker pause spent time even if worker still in pause serving state. Help user can find worker pause spent time increase, when worker always been pause state. | | |
<!--end-include-->
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
index c4a82225c..01fd138b5 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
@@ -34,6 +34,7 @@ class WorkerSource(conf: CelebornConf) extends AbstractSource(conf, Role.WORKER)
val appActiveConnections: ConcurrentHashMap[String, util.Set[String]] =
JavaUtils.newConcurrentHashMap[String, util.Set[String]]
+ private val metricsAppLevelEnabled = conf.metricsWorkerAppLevelEnabled
import WorkerSource._
// add counters
@@ -92,36 +93,44 @@ class WorkerSource(conf: CelebornConf) extends AbstractSource(conf, Role.WORKER)
}
def connectionActive(client: TransportClient): Unit = {
- appActiveConnections.putIfAbsent(
- client.getChannel.id().asLongText(),
- Sets.newConcurrentHashSet[String]())
+ if (metricsAppLevelEnabled) {
+ appActiveConnections.putIfAbsent(
+ client.getChannel.id().asLongText(),
+ Sets.newConcurrentHashSet[String]())
+ }
incCounter(ACTIVE_CONNECTION_COUNT, 1)
}
def connectionInactive(client: TransportClient): Unit = {
- val applicationIds = appActiveConnections.remove(client.getChannel.id().asLongText())
incCounter(ACTIVE_CONNECTION_COUNT, -1)
- if (null != applicationIds) {
- applicationIds.asScala.foreach(applicationId =>
- incCounter(ACTIVE_CONNECTION_COUNT, -1, Map(applicationLabel -> applicationId)))
+ if (metricsAppLevelEnabled) {
+ val applicationIds = appActiveConnections.remove(client.getChannel.id().asLongText())
+ if (null != applicationIds) {
+ applicationIds.asScala.foreach(applicationId =>
+ incCounter(ACTIVE_CONNECTION_COUNT, -1, Map(applicationLabel -> applicationId)))
+ }
}
}
def recordAppActiveConnection(client: TransportClient, shuffleKey: String): Unit = {
- val applicationIds = appActiveConnections.get(client.getChannel.id().asLongText())
- val applicationId = Utils.splitShuffleKey(shuffleKey)._1
- if (applicationIds != null && !applicationIds.contains(applicationId)) {
- addCounter(ACTIVE_CONNECTION_COUNT, Map(applicationLabel -> applicationId))
- incCounter(ACTIVE_CONNECTION_COUNT, 1, Map(applicationLabel -> applicationId))
- applicationIds.add(applicationId)
+ if (metricsAppLevelEnabled) {
+ val applicationIds = appActiveConnections.get(client.getChannel.id().asLongText())
+ val applicationId = Utils.splitShuffleKey(shuffleKey)._1
+ if (applicationIds != null && !applicationIds.contains(applicationId)) {
+ addCounter(ACTIVE_CONNECTION_COUNT, Map(applicationLabel -> applicationId))
+ incCounter(ACTIVE_CONNECTION_COUNT, 1, Map(applicationLabel -> applicationId))
+ applicationIds.add(applicationId)
+ }
}
}
def removeAppActiveConnection(applicationIds: util.Set[String]): Unit = {
- applicationIds.asScala.foreach(applicationId =>
- removeCounter(ACTIVE_CONNECTION_COUNT, Map(applicationLabel -> applicationId)))
- appActiveConnections.values().asScala.foreach(connectionAppIds =>
- applicationIds.asScala.foreach(applicationId => connectionAppIds.remove(applicationId)))
+ if (metricsAppLevelEnabled) {
+ applicationIds.asScala.foreach(applicationId =>
+ removeCounter(ACTIVE_CONNECTION_COUNT, Map(applicationLabel -> applicationId)))
+ appActiveConnections.values().asScala.foreach(connectionAppIds =>
+ applicationIds.asScala.foreach(applicationId => connectionAppIds.remove(applicationId)))
+ }
}
// start cleaner thread
diff --git a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/WorkerSuite.scala b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/WorkerSuite.scala
index 491f6f5b2..2e13ef1d6 100644
--- a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/WorkerSuite.scala
+++ b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/WorkerSuite.scala
@@ -131,6 +131,7 @@ class WorkerSuite extends AnyFunSuite with BeforeAndAfterEach {
test("handle top resource consumption") {
conf.set(CelebornConf.WORKER_STORAGE_DIRS.key, "/tmp")
+ conf.set(CelebornConf.METRICS_WORKER_APP_TOP_RESOURCE_CONSUMPTION_COUNT, 5)
worker = new Worker(conf, workerArgs)
val userIdentifier = new UserIdentifier("default", "celeborn")
worker.handleTopAppResourceConsumption(Map(userIdentifier -> ResourceConsumption(