This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 bf0a1e8fc2 [CELEBORN-2346] Add RequestSlots failure metrics
bf0a1e8fc2 is described below
commit bf0a1e8fc281d622fdf176076706d4dc4a52d80e
Author: Chao Sun <[email protected]>
AuthorDate: Wed Jun 3 17:38:19 2026 +0800
[CELEBORN-2346] Add RequestSlots failure metrics
### What changes were proposed in this pull request?
Export a new master `RequestSlotsFailed` counter with a bounded `status`
label for `SLOT_NOT_AVAILABLE` and `WORKER_EXCLUDED`.
This patch also makes metric-specific labels override configured
`celeborn.metrics.extraLabels`, while preserving reserved `role` and `instance`
labels. Configured extra labels are snapshotted when a metrics source is
constructed so labeled metric keys remain stable.
### Why are the changes needed?
`RequestSlots` placement failures directly affect shuffle registration, but
existing master metrics do not expose whether applications are receiving these
failure responses. The new counter provides a direct monitoring signal.
Without the label precedence change, an extra label such as `status=prod`
shadows the metric-specific failure reason and collapses both series.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
### How was this patch tested?
- `JAVA_HOME=$(brew --prefix openjdk17) dev/reformat`
- `JAVA_HOME=$(brew --prefix openjdk17) build/mvn --no-transfer-progress
-DskipTests
-DprotocPluginExecutable=/tmp/protoc-gen-grpc-java-noop-proto3-optional -pl
master -am install`
- `JAVA_HOME=$(brew --prefix openjdk17) build/mvn --no-transfer-progress
-DprotocPluginExecutable=/tmp/protoc-gen-grpc-java-noop-proto3-optional
-Dsuites=org.apache.celeborn.common.metrics.source.CelebornSourceSuite -pl
common test-compile scalatest:test`
- `JAVA_HOME=$(brew --prefix openjdk17) build/mvn --no-transfer-progress
-DprotocPluginExecutable=/tmp/protoc-gen-grpc-java-noop-proto3-optional
-Dsuites=org.apache.celeborn.service.deploy.master.MasterSourceSuite,org.apache.celeborn.service.deploy.master.MasterSuite
-pl master test-compile scalatest:test`
The local `protocPluginExecutable` override is an ARM workstation
workaround for the downloaded `protoc-gen-grpc-java` artifact.
Closes #3714 from sunchao/CELEBORN-2346-request-slots-failure-metrics.
Authored-by: Chao Sun <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
assets/grafana/celeborn-dashboard.json | 95 +++++++++++++++++++++-
.../common/metrics/source/AbstractSource.scala | 24 ++++--
.../metrics/source/CelebornSourceSuite.scala | 18 ++++
docs/monitoring.md | 1 +
.../celeborn/service/deploy/master/Master.scala | 2 +
.../service/deploy/master/MasterSource.scala | 16 ++++
.../service/deploy/master/MasterSourceSuite.scala | 43 ++++++++++
7 files changed, 191 insertions(+), 8 deletions(-)
diff --git a/assets/grafana/celeborn-dashboard.json
b/assets/grafana/celeborn-dashboard.json
index 65a984d677..fbbc1a05b5 100644
--- a/assets/grafana/celeborn-dashboard.json
+++ b/assets/grafana/celeborn-dashboard.json
@@ -2100,6 +2100,99 @@
],
"title": "metrics_RatisApplyCompletedIndexDiff_Value",
"type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "Number of failed RequestSlots responses over the
last 1 hour, broken down by status.",
+ "fieldConfig": {
+ "defaults": {
+ "color": {
+ "mode": "palette-classic"
+ },
+ "custom": {
+ "axisCenteredZero": false,
+ "axisColorMode": "text",
+ "axisLabel": "",
+ "axisPlacement": "auto",
+ "barAlignment": 0,
+ "drawStyle": "line",
+ "fillOpacity": 0,
+ "gradientMode": "none",
+ "hideFrom": {
+ "legend": false,
+ "tooltip": false,
+ "viz": false
+ },
+ "lineInterpolation": "linear",
+ "lineWidth": 1,
+ "pointSize": 5,
+ "scaleDistribution": {
+ "type": "linear"
+ },
+ "showPoints": "auto",
+ "spanNulls": false,
+ "stacking": {
+ "group": "A",
+ "mode": "none"
+ },
+ "thresholdsStyle": {
+ "mode": "off"
+ }
+ },
+ "mappings": [],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "green"
+ },
+ {
+ "color": "red",
+ "value": 1
+ }
+ ]
+ }
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 196
+ },
+ "id": 272,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "maxHeight": 600,
+ "mode": "single",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr":
"increase(metrics_RequestSlotsFailed_Count{role=\"Master\",
instance=~\"${instance}\"}[1h])",
+ "legendFormat": "{{statusCode}} ${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_RequestSlotsFailed_increase_1h",
+ "type": "timeseries"
}
],
"title": "Master",
@@ -18027,4 +18120,4 @@
"uid": "U_qgru_7z",
"version": 1,
"weekStart": ""
-}
\ No newline at end of file
+}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
index 6034dd9ca4..e7b71e5ec2 100644
---
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
@@ -68,6 +68,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
ThreadUtils.newDaemonSingleThreadScheduledExecutor("worker-metrics-cleaner")
val roleLabel: (String, String) = "role" -> role
+ val metricsExtraLabels: Map[String, String] = conf.metricsExtraLabels
val instanceLabel: Map[String, String] = role match {
case Role.MASTER =>
Map("instance" -> s"${Utils.localHostName(conf)}:${conf.masterHttpPort}")
@@ -75,7 +76,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
Map("instance" -> s"${Utils.localHostName(conf)}:${conf.workerHttpPort}")
case _ => Map.empty
}
- val staticLabels: Map[String, String] = conf.metricsExtraLabels + roleLabel
++ instanceLabel
+ val staticLabels: Map[String, String] = labelsWithCustomizedLabels(Map.empty)
val staticLabelsString: String = MetricLabels.labelString(staticLabels)
val applicationLabel = "applicationId"
@@ -110,7 +111,9 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
// filter out non-number type gauges
if (gauge.getValue.isInstanceOf[Number]) {
val metricNameWithLabel = metricNameWithCustomizedLabels(name, labels)
- namedGauges.putIfAbsent(metricNameWithLabel, NamedGauge(name, gauge,
labels ++ staticLabels))
+ namedGauges.putIfAbsent(
+ metricNameWithLabel,
+ NamedGauge(name, gauge, labelsWithCustomizedLabels(labels)))
metricRegistry.synchronized({
if (!metricRegistry.getMetrics.containsKey(metricNameWithLabel)) {
metricRegistry.register(metricNameWithLabel, gauge)
@@ -146,7 +149,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
meter: Meter): Unit = {
namedMeters.putIfAbsent(
metricNameWithCustomizedLabels(name, labels),
- NamedMeter(name, meter, labels ++ staticLabels))
+ NamedMeter(name, meter, labelsWithCustomizedLabels(labels)))
}
def addMeter(
@@ -177,7 +180,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
val namedTimer = NamedTimer(
name,
metricRegistry.timer(metricNameWithLabel, timerSupplier),
- labels ++ staticLabels)
+ labelsWithCustomizedLabels(labels))
val values = JavaUtils.newConcurrentHashMap[String, Long]()
(namedTimer, values)
})
@@ -189,7 +192,10 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
val metricNameWithLabel = metricNameWithCustomizedLabels(name, labels)
namedCounters.putIfAbsent(
metricNameWithLabel,
- NamedCounter(name, metricRegistry.counter(metricNameWithLabel), labels
++ staticLabels))
+ NamedCounter(
+ name,
+ metricRegistry.counter(metricNameWithLabel),
+ labelsWithCustomizedLabels(labels)))
}
def addHistogram(name: String): Unit = {
@@ -203,7 +209,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
NamedHistogram(
name,
metricRegistry.histogram(name, histogramSupplier),
- labels ++ staticLabels))
+ labelsWithCustomizedLabels(labels)))
}
def counters(): List[NamedCounter] = {
@@ -708,9 +714,13 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
if (labels.isEmpty) {
metricsName + staticLabelsString
} else {
- metricsName + MetricLabels.labelString(labels ++ staticLabels)
+ metricsName +
MetricLabels.labelString(labelsWithCustomizedLabels(labels))
}
}
+
+ protected def labelsWithCustomizedLabels(labels: Map[String, String]):
Map[String, String] = {
+ metricsExtraLabels ++ labels + roleLabel ++ instanceLabel
+ }
}
class TimerSupplier(val slidingWindowSize: Int)
diff --git
a/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
b/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
index 0f776f8df0..110d0b1ebe 100644
---
a/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
+++
b/common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala
@@ -115,4 +115,22 @@ class CelebornSourceSuite extends CelebornFunSuite {
createAbstractSourceAndCheck(conf, extraLabels3)
}
+
+ test("test customized labels override extra labels") {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.METRICS_EXTRA_LABELS.key,
"user=extra,role=extra,instance=extra")
+ val source = new AbstractSource(conf, Role.MASTER) {
+ override def sourceName: String = "mockSource"
+ }
+ val labels = Map("user" -> "metric")
+ source.addCounter("Counter", labels)
+ source.incCounter("Counter", 1, labels)
+ conf.set(CelebornConf.METRICS_EXTRA_LABELS.key, "user=changed")
+ source.incCounter("Counter", 1, labels)
+
+ val metrics = source.getMetrics
+ val instance = source.instanceLabel("instance")
+ assert(metrics.contains(
+
s"""metrics_Counter_Count{instance="$instance",role="Master",user="metric"}
2"""))
+ }
}
diff --git a/docs/monitoring.md b/docs/monitoring.md
index aa845640a0..5679ef548a 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -122,6 +122,7 @@ These metrics are exposed by Celeborn master.
| RatisApplyCompletedIndexDiff | The difference value of
ApplyCompletedIndex of the master nodes in HA mode. |
| PartitionSize | The size of estimated shuffle partition.
|
| OfferSlotsTime | The time for masters to handle
`RequestSlots` request when registering shuffle. |
+ | RequestSlotsFailed | The count of failed `RequestSlots`
requests, labeled by the failure status code. |
- namespace=CPU
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index 81e0e0076c..6b25552060 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -947,6 +947,7 @@ private[celeborn] class Master(
if (numAvailableWorkers == 0) {
logError(s"Offer slots for $shuffleKey failed due to all workers are
excluded!")
+ masterSource.incRequestSlotsFailed(StatusCode.WORKER_EXCLUDED)
context.reply(
RequestSlotsResponse(StatusCode.WORKER_EXCLUDED, new WorkerResource(),
requestSlots.packed))
return
@@ -1009,6 +1010,7 @@ private[celeborn] class Master(
// reply false if offer slots failed
if (slots == null || slots.isEmpty) {
logError(s"Offer slots for $numReducers reducers of $shuffleKey failed!")
+ masterSource.incRequestSlotsFailed(StatusCode.SLOT_NOT_AVAILABLE)
context.reply(RequestSlotsResponse(
StatusCode.SLOT_NOT_AVAILABLE,
new WorkerResource(),
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterSource.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterSource.scala
index aa28fd9005..3e22411ae1 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterSource.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterSource.scala
@@ -19,19 +19,31 @@ package org.apache.celeborn.service.deploy.master
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.metrics.source.{AbstractSource, Role}
+import org.apache.celeborn.common.protocol.message.StatusCode
class MasterSource(conf: CelebornConf) extends AbstractSource(conf,
Role.MASTER) {
override val sourceName = "master"
import MasterSource._
+ RequestSlotsFailureStatuses.foreach { status =>
+ addCounter(REQUEST_SLOTS_FAILED_COUNT, Map(STATUS_CODE_LABEL ->
status.name()))
+ }
// add timers
addTimer(OFFER_SLOTS_TIME)
addTimer(UPDATE_RESOURCE_CONSUMPTION_TIME)
// start cleaner
startCleaner()
+
+ def incRequestSlotsFailed(status: StatusCode): Unit = {
+ if (RequestSlotsFailureStatuses.contains(status)) {
+ incCounter(REQUEST_SLOTS_FAILED_COUNT, 1, Map(STATUS_CODE_LABEL ->
status.name()))
+ }
+ }
}
object MasterSource {
+ val STATUS_CODE_LABEL = "statusCode"
+
val WORKER_COUNT = "WorkerCount"
val LOST_WORKER_COUNT = "LostWorkerCount"
@@ -63,6 +75,7 @@ object MasterSource {
val ACTIVE_SHUFFLE_FILE_COUNT = "ActiveShuffleFileCount"
val OFFER_SLOTS_TIME = "OfferSlotsTime"
+ val REQUEST_SLOTS_FAILED_COUNT = "RequestSlotsFailed"
val RATIS_APPLY_COMPLETED_INDEX = "RatisApplyCompletedIndex"
@@ -71,4 +84,7 @@ object MasterSource {
val DEVICE_CELEBORN_TOTAL_CAPACITY = "DeviceCelebornTotalBytes"
val UPDATE_RESOURCE_CONSUMPTION_TIME = "UpdateResourceConsumptionTime"
+
+ private val RequestSlotsFailureStatuses =
+ Seq(StatusCode.SLOT_NOT_AVAILABLE, StatusCode.WORKER_EXCLUDED)
}
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSourceSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSourceSuite.scala
new file mode 100644
index 0000000000..cd591532f8
--- /dev/null
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSourceSuite.scala
@@ -0,0 +1,43 @@
+/*
+ * 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 org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.protocol.message.StatusCode
+
+class MasterSourceSuite extends CelebornFunSuite {
+
+ test("test request slots failed metrics") {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.METRICS_EXTRA_LABELS.key, "statusCode=prod")
+ val source = new MasterSource(conf)
+
+ source.incRequestSlotsFailed(StatusCode.SLOT_NOT_AVAILABLE)
+ source.incRequestSlotsFailed(StatusCode.WORKER_EXCLUDED)
+ source.incRequestSlotsFailed(StatusCode.SUCCESS)
+
+ val metrics = source.getMetrics
+ val instance = source.instanceLabel("instance")
+ assert(metrics.contains(
+
s"""metrics_RequestSlotsFailed_Count{instance="$instance",role="Master",statusCode="SLOT_NOT_AVAILABLE"}
1"""))
+ assert(metrics.contains(
+
s"""metrics_RequestSlotsFailed_Count{instance="$instance",role="Master",statusCode="WORKER_EXCLUDED"}
1"""))
+ assert(!metrics.contains("""statusCode="SUCCESS""""))
+ }
+}
