SteNicholas commented on code in PR #3740:
URL: https://github.com/apache/celeborn/pull/3740#discussion_r3548677037
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -85,7 +90,10 @@ class ApplicationHeartbeater(
tmpApplicationFallbackCounts.asJava,
workerStatusTracker.getNeedCheckedWorkers().toList.asJava,
ZERO_UUID,
- true)
+ true,
+ if (appMetricLabels.isEmpty)
java.util.Collections.emptyMap[String, ClientMetric]()
Review Comment:
**Silent no-op unless `celeborn.client.metrics.appLabels` is set.** The
client only sends metrics when `appMetricLabels` is non-empty, and the master's
`updateApplicationMetrics` early-returns when `metricLabels.isEmpty`. So an
operator who turns on the two obvious switches —
`celeborn.client.metrics.enabled=true` and
`celeborn.metrics.master.clientMetrics.enabled=true` — but leaves `appLabels`
at its default (empty) gets **zero** metrics and **zero** log output. Neither
enable-flag's doc mentions that `appLabels` is mandatory.
At minimum, log a one-time `warn` on the client when metrics are enabled but
`appLabels` is empty. Better: don't gate emission on labels at all — emit with
just `role`/`instance` so the default enablement path produces something.
##########
common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala:
##########
@@ -277,6 +335,26 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
metricNameWithLabel
}
+ protected def removeAppFromMetrics(appId: String): Unit = {
+ removeAppFromTracked(namedGaugesWithDetails, appId)(d =>
removeGauge(d.name, d.originalLabels))
+ removeAppFromTracked(namedCountersWithDetails, appId)(d =>
+ removeCounter(d.name, d.originalLabels))
+ }
+
+ private def removeAppFromTracked[T](
+ tracked: ConcurrentHashMap[String, AppMetricDetails[T]],
+ appId: String)(deregister: AppMetricDetails[T] => Unit): Unit = {
+ val iter = tracked.entrySet().iterator()
+ while (iter.hasNext) {
+ val details = iter.next().getValue
+ details.additionalDetails.remove(appId)
+ if (details.additionalDetails.isEmpty) {
Review Comment:
**Lost increment + transient deregistration under concurrency.** For two
apps sharing one label set, `removeAppFromTracked`
(`additionalDetails.remove(appId)` → `isEmpty` → `iter.remove()` →
`deregister`) is not synchronized with
`addOrUpdateCounterForApp`/`addOrUpdateGaugeForApp` (`computeIfAbsent` →
`additionalDetails.add(appId)` → `handle.inc/set`).
Interleaving: key `K` has `additionalDetails = {app-1}`. Thread A
(heartbeat, app-2's first contribution to `K`) gets the existing `details` from
`computeIfAbsent`. Thread B (app-1 lost) sees the set as `{app-1}` → empty →
`iter.remove()` + `removeCounter(K)`. Thread A resumes:
`additionalDetails.add("app-2")` + `handle.inc(delta)` on the now-orphaned
handle. Result: app-2's delta is permanently lost and the metric disappears
from `/metrics` until app-2's next heartbeat recreates it from 0. Since
`Master` is a concurrent endpoint and app-lost fires from
`timeoutDeadApplications`, this window is real.
(Minor, same struct: `AppMetricDetails.additionalDetails` actually holds
contributing appIds — a name like `appIds`/`contributingAppIds` would read far
more clearly, since this set is the refcount that drives deregistration.)
##########
common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala:
##########
@@ -212,6 +227,45 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
labelsWithCustomizedLabels(labels)))
}
+ protected def addOrUpdateGaugeForApp(
+ name: String,
+ labels: Map[String, String],
+ appId: String,
+ value: Long): Unit = {
+ val details = namedGaugesWithDetails.computeIfAbsent(
+ metricNameWithCustomizedLabels(name, labels),
+ (_: String) => {
+ val holder = new AtomicLong()
+ addGauge(name, labels)(() => holder.get())
+ AppMetricDetails(name, holder, labels,
ConcurrentHashMap.newKeySet[String]())
+ })
+ details.additionalDetails.add(appId)
+ details.handle.set(value)
+ }
+
+ protected def addOrUpdateCounterForApp(
+ name: String,
+ labels: Map[String, String],
+ appId: String,
+ delta: Long): Unit = {
+ if (delta <= 0) {
+ return
+ }
+ val metricKey = metricNameWithCustomizedLabels(name, labels)
+ val details = namedCountersWithDetails.computeIfAbsent(
+ metricKey,
+ (_: String) => {
+ addCounter(name, labels)
+ AppMetricDetails(
+ name,
+ namedCounters.get(metricKey).counter,
Review Comment:
**NPE under the app-lost/heartbeat race.** Inside `computeIfAbsent`, this
does `addCounter(name, labels)` and then
`namedCounters.get(metricKey).counter`, assuming the just-added counter is
still present. But `namedCounters` is a *different* map, not covered by the
`namedCountersWithDetails` bin lock. Because the `Master` endpoint dispatches
concurrently (see the review summary), a concurrent `removeAppFromTracked`
deregister for the same label key can run `removeCounter(metricKey)` (→
`namedCounters.remove`) in the window between `addCounter`'s `putIfAbsent` (a
no-op if the counter already existed) and this `.get`, making
`namedCounters.get(metricKey)` return `null` → NPE thrown out of
`updateApplicationMetrics` into `handleHeartbeatFromApplication`.
Consider having `addCounter` return the `NamedCounter` (or holding a single
lock around the tracked-metric mutations) so this never observes a null.
##########
client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala:
##########
@@ -159,6 +160,31 @@ class WorkerStatusTrackerSuite extends CelebornFunSuite {
errors.get())
}
+ test("recordWorkerFailure updates client worker-excluded gauge") {
+ val celebornConf = new CelebornConf()
+ celebornConf.set(CelebornConf.METRICS_ENABLED.key, "true")
+ celebornConf.set(CelebornConf.CLIENT_METRICS_ENABLED.key, "true")
+ val lifecycleManager = new LifecycleManager("app-metrics-test",
celebornConf)
+ try {
+ val statusTracker = lifecycleManager.workerStatusTracker
+ val source = lifecycleManager.clientSource.get
+
+ val failed = new ShuffleFailedWorkers()
+ val now = System.currentTimeMillis()
+ failed.put(mock("host1"), (StatusCode.WORKER_UNRESPONSIVE, now))
+ failed.put(mock("host2"), (StatusCode.WORKER_UNRESPONSIVE, now))
+ statusTracker.recordWorkerFailure(failed)
+
+ val snapshot = source.getMetricsSnapshot()
+ Assert.assertEquals(2L,
snapshot(CelebornClientSource.EXCLUDED_WORKER_COUNT).value)
+
+ // re-recording already-excluded workers does not change the gauge
+ statusTracker.recordWorkerFailure(failed)
Review Comment:
**Dead comment / missing assertion, and this is an integration test in a
unit suite.** The `// re-recording already-excluded workers does not change the
gauge` behavior is never checked — there's no `assertEquals` after this second
`recordWorkerFailure(failed)`, so a regression that double-counted (gauge → 4)
would still pass. Add the assertion.
Separately, `new LifecycleManager("app-metrics-test", celebornConf)` (line
167) is heavy for a unit test: the constructor creates an `RpcEnv` (binds an
ephemeral socket), runs `initialize()` → `registerApplicationInfo()` against a
non-existent master, and starts the heartbeater/commit/changePartition/release
background threads that then spin retrying connections. A bare
`WorkerStatusTracker` + `CelebornClientSource` (wired to the excluded-workers
gauge) would cover this gauge deterministically without the port bind and
connection noise.
##########
common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala:
##########
@@ -259,6 +260,29 @@ class UtilsSuite extends CelebornFunSuite {
assert(set.size == 0)
}
+ test("HeartbeatFromApplication carries client metrics through pb serde") {
+ val clientMetrics = new util.HashMap[String, ClientMetric]()
+ clientMetrics.put("ClientRegisterShuffleCount", ClientMetric(5L,
MetricType.Counter))
+ clientMetrics.put("ClientExcludedWorkerCount", ClientMetric(2L,
MetricType.Gauge))
+
+ val heartbeat = HeartbeatFromApplication(
+ "app-1",
+ 100L,
+ 10L,
+ 3L,
+ 1L,
+ new util.HashMap[String, java.lang.Long](),
+ new util.HashMap[String, java.lang.Long](),
+ new util.ArrayList(),
+ shouldResponse = true,
+ clientMetrics = clientMetrics)
+
+ val heartbeatTrans =
Utils.fromTransportMessage(Utils.toTransportMessage(heartbeat))
+ .asInstanceOf[HeartbeatFromApplication]
+
+ assert(heartbeatTrans.clientMetrics == clientMetrics)
Review Comment:
The roundtrip assertion is sound for `clientMetrics` (Java `HashMap.equals`
+ `ClientMetric`/`MetricType` case-class equality), but `metricLabels` (new
proto field 13) is left at the default empty map and never round-tripped. A bug
that dropped or mis-copied `metricLabels` on the wire would ship undetected.
Construct the heartbeat with a non-empty `metricLabels` and assert it survives
serde too. (Also worth noting: the master-side `ApplicationMetricsSourceSuite`
creates ~13 enabled sources whose `metricsCleaner` daemon thread is never shut
down — an `afterEach`/`destroy()` would avoid leaking scheduled threads across
the suite.)
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -5965,6 +5971,34 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
+ val CLIENT_METRICS_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.metrics.enabled")
+ .categories("metrics")
+ .doc("When true, the LifecycleManager collects client-side metrics. " +
+ "Requires `celeborn.metrics.enabled` to also be true.")
+ .version("0.7.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val MASTER_CLIENT_METRICS_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.metrics.master.clientMetrics.enabled")
+ .categories("metrics")
+ .doc("When true, the master exposes client-side metrics forwarded in
application " +
+ "heartbeats on its Prometheus endpoint.")
+ .version("0.7.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val MASTER_CLIENT_METRICS_REMOVED_APP_RETENTION: ConfigEntry[Long] =
+ buildConf("celeborn.metrics.master.clientMetrics.removedApp.retentionMs")
Review Comment:
Minor config-naming nit:
`celeborn.metrics.master.clientMetrics.removedApp.retentionMs` embeds the unit
(`Ms`) in the key of a `timeConf`, which reads self-contradictorily against its
own default (`retentionMs = 5min`) and invites passing a raw millisecond
number. No other `timeConf` in the repo suffixes its unit (siblings use
`.timeout`, `.interval`, `.expireTimeout`, `.retention`). Suggest
`...removedApp.retention` (or `.expireTimeout`). While here:
`CLIENT_METRICS_ENABLED` uses `.categories("metrics")` but its sibling
`CLIENT_METRICS_APP_LABELS` uses `.categories("client", "metrics")`, so the
enable flag is missing from `docs/configuration/client.md` — consider
`.categories("client", "metrics")` for consistency.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 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]()
+
+ 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 ||
removedAppIds.containsKey(appId)) {
Review Comment:
**TOCTOU between this gate and `removeApplicationMetrics` → resurrected,
permanently-leaked metrics.** This checks `removedAppIds.containsKey(appId)`
and then (in `addOrUpdate*ForApp`) mutates the metric maps in separate,
non-atomic steps relative to `removeApplicationMetrics` (which does
`removedAppIds.put` + `removeAppFromMetrics`). On the concurrent `Master`
endpoint:
1. Thread A (heartbeat for `app-1`) reads `containsKey("app-1") == false`.
2. Thread B (`timeoutDeadApplications` → `handleApplicationLost`) runs
`removeApplicationMetrics("app-1")` fully.
3. Thread A resumes and re-registers `app-1` into `additionalDetails` and
the registry.
`app-1` is now gated in `removedAppIds` (so `removeApplicationMetrics` won't
run again) yet its metric series stays registered; after the retention cleaner
evicts `app-1` from `removedAppIds`, nothing ever deregisters that series → a
dead app's series leaks indefinitely.
Relatedly, series are only reclaimed via `handleAppLost`; on a master
failover `removedAppIds` (in-memory) is lost, and with high-cardinality
operator `appLabels` there's no upper bound on distinct series. Worth a
cap/warn and/or documenting that labels must be low-cardinality.
##########
common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala:
##########
@@ -212,6 +227,45 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
labelsWithCustomizedLabels(labels)))
}
+ protected def addOrUpdateGaugeForApp(
+ name: String,
+ labels: Map[String, String],
+ appId: String,
+ value: Long): Unit = {
+ val details = namedGaugesWithDetails.computeIfAbsent(
+ metricNameWithCustomizedLabels(name, labels),
+ (_: String) => {
+ val holder = new AtomicLong()
+ addGauge(name, labels)(() => holder.get())
+ AppMetricDetails(name, holder, labels,
ConcurrentHashMap.newKeySet[String]())
+ })
+ details.additionalDetails.add(appId)
+ details.handle.set(value)
Review Comment:
**Gauge aggregation is last-writer-wins across apps sharing a label set
(counters sum, gauges clobber).** `appId` is intentionally kept only in
`additionalDetails`, never in the metric key, so N apps that share a label set
(e.g. coarse `env=prod`) collapse to one series and this `handle.set(value)`
makes the exported gauge flap to whichever app heartbeated last — not a sum or
max. A dashboard reading `ClientActiveShuffleCount` then sees an arbitrary
single app's value. The PR's own test asserts this, so it looks intended, but
the counter-vs-gauge asymmetry is surprising; please at least document it (and
consider `max`/`sum` semantics, or requiring a per-app label for gauges).
##########
client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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 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 previous counter values so we can send deltas to the master.
+ private val counterPrev = 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] = {
+ // Counters: compute delta since last snapshot
+ val counterMetrics = counters().flatMap { c =>
+ val current = c.counter.getCount
+ val prev = Option(counterPrev.put(c.name,
current)).map(_.longValue()).getOrElse(0L)
Review Comment:
**Counter deltas are lost when a heartbeat fails to send.**
`counterPrev.put(c.name, current)` advances `prev` to `current` while the
heartbeat message is being *built*; if `requestHeartbeat(...)` then fails
(master unreachable/timeout) there's no retry and no rollback, so the emitted
delta is discarded. The master's counter permanently under-counts by every
failed heartbeat's delta.
(The symmetric hazard — advancing only on success — would double-count the
timeout-but-actually-applied case, so a fully correct fix needs an
ack/idempotency key. At least worth a comment acknowledging the at-most-once
semantics, or carrying-forward the un-acked delta into the next snapshot.)
##########
client/src/test/scala/org/apache/celeborn/client/CelebornClientSourceSuite.scala:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.atomic.AtomicInteger
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.metrics.MetricType
+
+class CelebornClientSourceSuite extends CelebornFunSuite {
+
+ test("counters are declared, increment, and emit with role=Client") {
+ val source = new CelebornClientSource(new CelebornConf())
+
+ source.incCounter(CelebornClientSource.REGISTER_SHUFFLE_COUNT)
+ source.incCounter(CelebornClientSource.REGISTER_SHUFFLE_COUNT)
+ source.incCounter(CelebornClientSource.REGISTER_SHUFFLE_FAIL_COUNT)
+ source.incCounter(CelebornClientSource.UNREGISTER_SHUFFLE_COUNT, 3)
+ source.incCounter(CelebornClientSource.REVIVE_REQUEST_COUNT, 5)
+ source.incCounter(CelebornClientSource.REVIVE_FAIL_COUNT, 2)
+ source.incCounter(CelebornClientSource.SLOT_RESERVATION_FAIL_COUNT)
+ source.incCounter(CelebornClientSource.SHUFFLE_FETCH_FAILURE_COUNT)
+ source.incCounter(CelebornClientSource.SHUFFLE_DATA_LOST_COUNT)
+
+ val metrics = source.getMetrics
+ assert(metrics.contains("""metrics_ClientRegisterShuffleCount_Count"""))
+ assert(metrics.contains("""role="Client""""))
+
+ val snapshot = source.getMetricsSnapshot()
Review Comment:
**The delta semantics — the whole point of `counterPrev` — is never
exercised.** Every test calls `getMetricsSnapshot()` only once, so it only ever
sees the first (full-value) delta. If the delta code regressed to emit
cumulative counts (dropping `counterPrev`) or to keep emitting unchanged
counters, the master would double-count every heartbeat, yet all three tests
here would still pass. Please add a test that increments, snapshots, increments
again, snapshots again, and asserts the second snapshot returns only the new
delta and omits unchanged counters.
--
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]
