SteNicholas commented on code in PR #3740:
URL: https://github.com/apache/celeborn/pull/3740#discussion_r3642624778
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -85,10 +99,14 @@ class ApplicationHeartbeater(
tmpApplicationFallbackCounts.asJava,
workerStatusTracker.getNeedCheckedWorkers().toList.asJava,
ZERO_UUID,
- true)
+ true,
+ if (appMetricLabels.isEmpty)
java.util.Collections.emptyMap[String, ClientMetric]()
+ else clientMetrics(),
+ appMetricLabels)
val response = requestHeartbeat(appHeartbeat)
if (response.statusCode == StatusCode.SUCCESS) {
logDebug("Successfully send app heartbeat.")
+ commitClientMetrics()
Review Comment:
**[P1] Counter deltas can be double-counted after a lost heartbeat
response.** `requestHeartbeat` converts a timeout/transport exception into
`REQUEST_FAILED`, so this commit is skipped. But the master may already have
applied `updateApplicationMetrics` before its response was lost; the next
heartbeat then carries the same unacknowledged delta again and the master
increments it twice. Since these heartbeats also use `ZERO_UUID`, there is no
snapshot identity to deduplicate. Please retain a sequence/request ID with the
pending snapshot and have the master ignore an already-applied snapshot (or use
another protocol that is idempotent across ambiguous timeouts).
##########
client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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 java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.JavaConverters._
+
+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._
+
+ // Tracks the counter baselines that have been acknowledged by the master,
so we can send
+ // deltas rather than cumulative counts.
+ private val counterPrev = new ConcurrentHashMap[String, java.lang.Long]()
+
+ // Counter values captured by the most recent getMetricsSnapshot() that have
not yet been
+ // acknowledged.
+ private val pendingCounterValues = new ConcurrentHashMap[String,
java.lang.Long]()
+
+ 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().flatMap { c =>
+ val current = c.counter.getCount
+ val prev =
Option(counterPrev.get(c.name)).map(_.longValue()).getOrElse(0L)
+ val delta = current - prev
+ pendingCounterValues.put(c.name, current)
+ if (delta > 0) Some(c.name -> ClientMetric(delta, MetricType.Counter))
+ else None
+ }
+ // Gauges: send the latest value as-is.
+ val gaugeMetrics = gauges().map(g =>
+ g.name ->
ClientMetric(g.gauge.getValue.asInstanceOf[Number].longValue(),
MetricType.Gauge))
+ (counterMetrics ++ gaugeMetrics).toMap
+ }
+
+ def commitSnapshot(): Unit = {
+ pendingCounterValues.entrySet().asScala.foreach { entry =>
+ counterPrev.put(entry.getKey, entry.getValue)
+ }
+ pendingCounterValues.clear()
+ }
+
+ def start(): Unit = startCleaner()
Review Comment:
**[P2] This starts a cleaner thread that can never clean anything for this
source.** `startCleaner()` only scans `namedTimers`, while
`CelebornClientSource` defines counters and gauges but no timers. Because
`LifecycleManager` invokes `source.start()`, every metrics-enabled client gets
a scheduled daemon thread that wakes forever to scan an empty timer map. Client
metric snapshots do not require it, so please remove this scheduling path
(while retaining lifecycle cleanup for any executor state that remains).
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+
+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, Utils}
+
+class ApplicationMetricsSource(conf: CelebornConf)
+ extends AbstractSource(conf, Role.MASTER) with Logging {
+ override val sourceName = "application"
+
+ private val masterClientMetricsEnabled = conf.masterClientMetricsEnabled
+ private val removedAppRetentionMs =
conf.masterClientMetricsRemovedAppRetentionMs
+
+ // Tracking applications that have been terminated
+ private val removedAppIds =
+ JavaUtils.newConcurrentHashMap[String, java.lang.Long]()
+
+ private val seriesCardinalityWarnThreshold =
+ conf.masterClientMetricsSeriesCardinalityWarnThreshold
+ private val seriesCardinalityWarned = new AtomicBoolean(false)
+
+ if (masterClientMetricsEnabled) {
+ startRemovedAppCleaner()
+ }
+
+ private def startRemovedAppCleaner(): Unit = {
+ val cleanTask: Runnable = new Runnable {
+ override def run(): Unit = Utils.tryLogNonFatalError {
+ val cutoff = System.currentTimeMillis() - removedAppRetentionMs
+ removedAppIds.entrySet().asScala.foreach { entry =>
+ if (entry.getValue < cutoff) {
+ removedAppIds.remove(entry.getKey, entry.getValue)
+ }
+ }
+ }
+ }
+ metricsCleaner.scheduleWithFixedDelay(
+ cleanTask,
+ removedAppRetentionMs,
+ removedAppRetentionMs,
+ TimeUnit.MILLISECONDS)
+ }
+
+ def updateApplicationMetrics(
+ appId: String,
+ metricLabels: Map[String, String],
+ metrics: JMap[String, ClientMetric]): Unit = {
+ if (!masterClientMetricsEnabled || metricLabels.isEmpty) {
+ return
+ }
+
+ if (removedAppIds.containsKey(appId)) {
+ return
+ }
+
+ metrics.asScala.foreach { case (name, metric) =>
+ metric.metricType match {
+ case MetricType.Gauge =>
+ addOrUpdateGaugeForApp(name, metricLabels, appId, metric.value)
+ case MetricType.Counter =>
+ addOrUpdateCounterForApp(name, metricLabels, appId, metric.value)
+ }
+ }
+
+ if (removedAppIds.containsKey(appId)) {
+ removeAppFromMetrics(appId)
+ }
+
+ warnIfSeriesCardinalityHigh()
+ }
+
+ def removeApplicationMetrics(appId: String): Unit = {
+ if (masterClientMetricsEnabled) {
+ removedAppIds.put(appId, System.currentTimeMillis())
+ }
+ removeAppFromMetrics(appId)
+ }
+
+ private def warnIfSeriesCardinalityHigh(): Unit = {
+ val trackedSeries = gauges().size + counters().size
Review Comment:
**[P2] The one-time cardinality warning still scans and allocates every
series on every heartbeat.** `gauges()` and `counters()` each convert the
entire map to a Scala `List`, and this count is computed before
`seriesCardinalityWarned` gates the log. Once the threshold is crossed—the
exact high-cardinality case this protects—every subsequent heartbeat remains
O(number of series) and allocates both lists. Return immediately when
`seriesCardinalityWarned.get()` is true and use the underlying tracked-map
sizes instead of materializing metric lists.
--
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]
