SteNicholas commented on code in PR #3740: URL: https://github.com/apache/celeborn/pull/3740#discussion_r3487564511
########## 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 = { Review Comment: `Master` extends plain `RpcEndpoint` (not `ThreadSafeRpcEndpoint`), so its Inbox sets `enableConcurrent = true` and app heartbeats are dispatched **concurrently**. If a heartbeat for `appId` is in-flight/queued when `handleAppLost` → `removeApplicationMetrics(appId)` runs (e.g. a false timeout from a GC pause, or the last heartbeat racing `ApplicationLost`), this method then re-`computeIfAbsent`s the caches and re-`addGauge`/`addCounter`s the app's metrics. Since `removeApplicationMetrics` only ever fires once per app, those `applicationId=<dead app>` series leak permanently and the counter re-emits its full cumulative value (prev reset to 0 → delta = full value). Needs a liveness check against the live-app set, or removal coordinated so a later heartbeat can't resurrect. ########## 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) Review Comment: `clientSource` is created unconditionally (before the `clientMetricsEnabled` guard), and `AbstractSource`'s constructor spawns a daemon `worker-metrics-cleaner` scheduled executor; `CelebornClientSource` also calls `startCleaner()` and registers 8 counters in its ctor. But `LifecycleManager.stop()` only calls `heartbeater.stop()` + `super.stop()` — it never calls `clientSource.destroy()`. So every LifecycleManager (one per Spark app driver) leaks a daemon thread + a MetricRegistry, **even when `celeborn.client.metrics.enabled=false`** (the default). On multi-tenant/long-lived drivers (Spark Connect, Kyuubi, notebooks) and in `WorkerStatusTrackerSuite`'s new test (which only calls `stop()`), these accumulate. Suggest gating creation on `clientMetricsEnabled` and calling `clientSource.destroy()` in `stop()`. (The cleaner is also a no-op here — `clearOldValues` only scans `namedTimers`, and this source has none.) ########## 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 Review Comment: Two issues here. (a) **Non-atomic** read-modify-write: `getOrDefault` → `delta` → `incCounter` → `prev.put` is not atomic, and with `enableConcurrent=true` two heartbeats for the same app over-count (both read the same `prev`) or lose updates. Use `ConcurrentHashMap.merge`/`compute` to make delta+store atomic. (b) **Fragile absolute→delta**: the client reports an absolute cumulative value; `if (delta > 0)` then `prev.put(name, newValue)` (unconditional, line 92) means a same-appId restart (counter resets) or a reordered/retried heartbeat rewinds `prev` and either stalls the counter or over-counts on the next tick. Consider exposing the client's absolute value directly as a gauge (Prometheus `rate()`/`increase()` already tolerate counter resets) — that removes `appCounterPrev` and this whole bug class. ########## master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala: ########## @@ -75,12 +75,14 @@ private[celeborn] class Master( new ResourceConsumptionSource(conf, Role.MASTER) private val threadPoolSource = ThreadPoolSource(conf, Role.MASTER) private val masterSource = new MasterSource(conf) + private val applicationMetricsSource = new ApplicationMetricsSource(conf) private val jvmSource = new JVMSource(conf, Role.MASTER) private val jvmCpuSource = new JVMCPUSource(conf, Role.MASTER) private val systemMiscSource = new SystemMiscSource(conf, Role.MASTER) metricsSystem.registerSource(resourceConsumptionSource) metricsSystem.registerSource(masterSource) + metricsSystem.registerSource(applicationMetricsSource) Review Comment: `ApplicationMetricsSource` re-exposes ~11 metrics per heartbeating app labeled by `applicationId`, with no top-N cap and gated only by the *client*-side flag (no master-side enable). All share one `metricsCapacity` (default 4096); `AbstractSource.getMetrics` silently truncates beyond that (logWarning only) and emits counters **last**, so app counters (e.g. `ClientShuffleDataLostCount`) are dropped first at ~370 concurrent apps. The codebase already handles this exact cardinality problem for worker per-app metrics via `celeborn.metrics.worker.app.topResourceConsumption.count` (default 0 = off, top-N capped, documented as high-cardinality). Worth following that precedent (master-side enable + top-N), and note the source/cleaner are constructed even when master `metricsSystemEnable=false`. ########## 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) + } + + def removeApplicationMetrics(appId: String): Unit = { + val labels = Map(applicationLabel -> appId) + val gaugeCache = appGaugeCache.remove(appId) Review Comment: `removeApplicationMetrics` drops the `appGaugeCache` entry (line 97) **before** unregistering the gauges (line 99). The registered gauge closure reads `Option(appGaugeCache.get(appId))...getOrElse(0L)`, so a concurrent Prometheus scrape landing in that window reports `0` for the about-to-be-removed gauge — a transient flap-to-0. Unregister the gauges first, then drop the cache (or guard the closure). ########## common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala: ########## @@ -1359,7 +1370,15 @@ object ControlMessages extends Logging { pbHeartbeatFromApplication.getNeedCheckedWorkerListList.asScala .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava), pbHeartbeatFromApplication.getRequestId, - pbHeartbeatFromApplication.getShouldResponse) + pbHeartbeatFromApplication.getShouldResponse, + new util.HashMap[String, ClientMetric]( + pbHeartbeatFromApplication.getClientMetricsMap.asScala.map { case (name, pbMetric) => + val metricType = pbMetric.getType match { + case PbMetricType.COUNTER => MetricType.Counter Review Comment: `fromPb` uses `case PbMetricType.COUNTER => Counter; case _ => Gauge`, silently coercing `UNRECOGNIZED`/any future enum value to `Gauge`. With version skew, a newer client's counter would be decoded on an older master as a gauge → routed to `updateGauge` (last-value) instead of `updateCounter` (delta), i.e. silently wrong semantics. At minimum `logWarning` on the default branch; better, handle `GAUGE`/`UNRECOGNIZED` explicitly. ########## client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala: ########## @@ -282,6 +282,11 @@ class ChangePartitionManager( None, lifecycleManager.workerStatusTracker.workerAvailableByLocation(req.oldPartition)))) } + if (lifecycleManager.clientMetricsEnabled) { + lifecycleManager.clientSource.incCounter( + CelebornClientSource.REVIVE_FAIL_COUNT, Review Comment: `REVIVE_FAIL_COUNT` is incremented here by `changePartitions.size` (per-partition), but in `LifecycleManager.handleRevive` it's incremented by +1 per batch on the unregistered/stage-ended paths, while `REVIVE_REQUEST_COUNT` is incremented by `partitionIds.size`. Mixing per-partition and per-batch units in one series makes a fail-rate uninterpretable. Separately, `SHUFFLE_DATA_LOST_COUNT` is incremented both in `LifecycleManager.handleMapPartitionEnd` and in `ReducePartitionCommitHandler.stageEnd` — please confirm those are disjoint events and not double-counting the same lost shuffle. Pick one unit per metric. -- 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]
