Copilot commented on code in PR #3740:
URL: https://github.com/apache/celeborn/pull/3740#discussion_r3451456775
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -222,8 +223,21 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
}
private val masterClient = new MasterClient(masterRpcEnvInUse, conf, false)
+ val clientSource = new CelebornClientSource(conf)
+ private[client] val clientMetricsEnabled = conf.metricsSystemEnable &&
conf.clientMetricsEnabled
val commitManager = new CommitManager(appUniqueId, conf, this)
val workerStatusTracker = new WorkerStatusTracker(conf, this)
+ if (clientMetricsEnabled) {
+ clientSource.addGauge(CelebornClientSource.ACTIVE_SHUFFLE_COUNT) { () =>
+ registeredShuffle.size
+ }
+ clientSource.addGauge(CelebornClientSource.EXCLUDED_WORKER_COUNT) { () =>
+ workerStatusTracker.excludedWorkers.size
+ }
+ clientSource.addGauge(CelebornClientSource.SHUTTING_WORKER_COUNT) { () =>
+ workerStatusTracker.shuttingWorkers.size
+ }
+ }
Review Comment:
`CelebornClientSource` is instantiated unconditionally even when
`clientMetricsEnabled` is false. Since the source currently starts its cleaner
thread in its constructor, this can create unnecessary background
threads/overhead for every client. Prefer creating/starting the source only
when metrics are enabled (e.g., lazy/Option + guard), or defer
`startCleaner()`/registration until `clientMetricsEnabled` is true.
##########
client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala:
##########
@@ -0,0 +1,72 @@
+/*
+ * 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.client
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
+import org.apache.celeborn.common.metrics.source.{AbstractSource, Role}
+
+/**
+ * Metrics source for the Celeborn client
+ */
+class CelebornClientSource(conf: CelebornConf) extends AbstractSource(conf,
Role.CLIENT) {
+ override val sourceName = "client"
+
+ import CelebornClientSource._
+
+ addCounter(REGISTER_SHUFFLE_COUNT)
+ addCounter(REGISTER_SHUFFLE_FAIL_COUNT)
+ addCounter(UNREGISTER_SHUFFLE_COUNT)
+ addCounter(REVIVE_REQUEST_COUNT)
+ addCounter(REVIVE_FAIL_COUNT)
+ addCounter(SLOT_RESERVATION_FAIL_COUNT)
+ addCounter(SHUFFLE_FETCH_FAILURE_COUNT)
+ addCounter(SHUFFLE_DATA_LOST_COUNT)
+
+ def getMetricsSnapshot(): Map[String, ClientMetric] = {
+ val counterMetrics = counters().map(c =>
+ c.name -> ClientMetric(c.counter.getCount, MetricType.Counter))
+ val gaugeMetrics = gauges().map(g =>
+ g.name ->
ClientMetric(g.gauge.getValue.asInstanceOf[Number].longValue(),
MetricType.Gauge))
+ (counterMetrics ++ gaugeMetrics).toMap
+ }
+
+ // start cleaner thread
+ startCleaner()
Review Comment:
`startCleaner()` is invoked unconditionally when `CelebornClientSource` is
constructed, which can spawn a cleanup thread even for clients that never
enable or use client metrics (see `LifecycleManager` creating the source
regardless of config). Consider moving cleaner startup behind the enablement
check, or providing a constructor/flag that avoids starting background
maintenance when metrics are disabled.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.service.deploy.master
+
+import java.util.{Map => JMap}
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
+import org.apache.celeborn.common.metrics.source.{AbstractSource, Role}
+import org.apache.celeborn.common.util.JavaUtils
+
+/**
+ * Holds the client-side metrics that applications report in their heartbeat
and re-exposes them on
+ * the master's Prometheus endpoint, labeled by `applicationId`. Both the
registrations and any
+ * cached state are dropped when the application is terminated.
+ */
+class ApplicationMetricsSource(conf: CelebornConf)
+ extends AbstractSource(conf, Role.MASTER) with Logging {
+ override val sourceName = "application"
+
+ // applicationId -> (metricName -> latest reported gauge value)
+ private val appGaugeCache =
+ JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
java.lang.Long]]()
+
+ // applicationId -> (metricName -> last reported counter value, used to
compute deltas)
+ private val appCounterPrev =
+ JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
java.lang.Long]]()
+
+ startCleaner()
+
+ def updateApplicationMetrics(appId: String, metrics: JMap[String,
ClientMetric]): Unit = {
+ if (metrics.isEmpty) return
+ metrics.asScala.foreach { case (name, metric) =>
+ val labels = Map(applicationLabel -> appId)
Review Comment:
Returning early on an empty `metrics` map means previously reported metrics
for `appId` will remain exposed indefinitely (until `removeApplicationMetrics`
on app-lost), even if the client stops reporting metrics (e.g., metrics toggled
off) or intentionally sends an empty set. If an empty map is intended to mean
“no metrics”, consider clearing existing per-app metric state (e.g.,
`removeApplicationMetrics(appId)` or clearing caches) when `metrics.isEmpty`.
##########
master/src/test/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSourceSuite.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.service.deploy.master
+
+import java.util.{HashMap => JHashMap}
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
+
+class ApplicationMetricsSourceSuite extends CelebornFunSuite {
+
+ private def metricsOf(app: String, value: Long): JHashMap[String,
ClientMetric] = {
Review Comment:
The `app` parameter is unused inside `metricsOf`, which can introduce
warnings and confusion in this new test file. Consider removing the unused
parameter (or using it if the helper is meant to be app-specific).
##########
client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala:
##########
@@ -159,6 +160,31 @@ class WorkerStatusTrackerSuite extends CelebornFunSuite {
errors.get())
}
+ test("recordWorkerFailure increments client worker-excluded counter and
gauge") {
Review Comment:
The test name claims both a counter and a gauge are incremented, but the
assertions in this test only validate the excluded-worker gauge value (and
`CelebornClientSource` defines `ClientExcludedWorkerCount` as a gauge). Either
update the test name to match what is verified, or extend the test to assert
the counter behavior if a counter is actually expected.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.service.deploy.master
+
+import java.util.{Map => JMap}
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
+import org.apache.celeborn.common.metrics.source.{AbstractSource, Role}
+import org.apache.celeborn.common.util.JavaUtils
+
+/**
+ * Holds the client-side metrics that applications report in their heartbeat
and re-exposes them on
+ * the master's Prometheus endpoint, labeled by `applicationId`. Both the
registrations and any
+ * cached state are dropped when the application is terminated.
+ */
+class ApplicationMetricsSource(conf: CelebornConf)
+ extends AbstractSource(conf, Role.MASTER) with Logging {
+ override val sourceName = "application"
+
+ // applicationId -> (metricName -> latest reported gauge value)
+ private val appGaugeCache =
+ JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
java.lang.Long]]()
+
+ // applicationId -> (metricName -> last reported counter value, used to
compute deltas)
+ private val appCounterPrev =
+ JavaUtils.newConcurrentHashMap[String, ConcurrentHashMap[String,
java.lang.Long]]()
+
+ startCleaner()
+
+ def updateApplicationMetrics(appId: String, metrics: JMap[String,
ClientMetric]): Unit = {
+ if (metrics.isEmpty) return
+ metrics.asScala.foreach { case (name, metric) =>
+ val labels = Map(applicationLabel -> appId)
+ metric.metricType match {
+ case MetricType.Gauge => updateGauge(appId, name, labels, metric.value)
+ case MetricType.Counter => updateCounter(appId, name, labels,
metric.value)
+ }
+ }
+ }
+
+ private def updateGauge(
+ appId: String,
+ name: String,
+ labels: Map[String, String],
+ value: Long): Unit = {
+ val cache = appGaugeCache.computeIfAbsent(appId, _ =>
JavaUtils.newConcurrentHashMap())
+ cache.put(name, value)
+ if (!gaugeExists(name, labels)) {
+ addGauge(name, labels) { () =>
+ Option(appGaugeCache.get(appId))
+ .flatMap(m => Option(m.get(name)))
+ .map(_.longValue())
+ .getOrElse(0L)
+ }
+ }
+ }
+
+ private def updateCounter(
+ appId: String,
+ name: String,
+ labels: Map[String, String],
+ newValue: Long): Unit = {
+ val prev = appCounterPrev.computeIfAbsent(appId, _ =>
JavaUtils.newConcurrentHashMap())
+ if (!counterExists(name, labels)) {
+ addCounter(name, labels)
+ }
+ val prevValue = prev.getOrDefault(name, 0L)
+ val delta = newValue - prevValue
+ if (delta > 0) {
+ incCounter(name, delta, labels)
+ }
+ prev.put(name, newValue)
+ }
Review Comment:
If the client-side counter resets (process restart, overflow, metric reset),
`delta` becomes negative and the master drops the increment, which can make the
exported counter permanently undercount relative to what the client reports. A
common approach is to treat `newValue < prevValue` as a reset and either (a)
increment by `newValue` (reset semantics) or (b) reset the exported counter (if
supported) and then set prev accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
