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 cecbe7e749 [CELEBORN-194] Introduce client side metrics for celeborn
cecbe7e749 is described below
commit cecbe7e749063c07701e94f1e90a66d300eb96c0
Author: AmandeepSingh285 <[email protected]>
AuthorDate: Wed Oct 21 10:51:24 2026 +0800
[CELEBORN-194] Introduce client side metrics for celeborn
### What changes were proposed in this pull request?
Adding client side metrics for Celeborn via heartbeat to master. Adding
gauge metrics.
### Why are the changes needed?
These changes help increase observability for Celeborn clients.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
Local setup and unit tests
<img width="1151" height="651" alt="image"
src="https://github.com/user-attachments/assets/daac4dbf-23ef-4f17-a7d7-cce9a839ba06";
/>
Closes #3740 from AmandeepSingh285/adding-client-side-metrics.
Lead-authored-by: AmandeepSingh285 <[email protected]>
Co-authored-by: amandeeps.28 <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
assets/grafana/celeborn-dashboard.json | 290 ++++++++++++++++
.../celeborn/client/ApplicationHeartbeater.scala | 26 +-
.../celeborn/client/CelebornClientSource.scala | 47 +++
.../apache/celeborn/client/LifecycleManager.scala | 22 +-
.../client/CelebornClientSourceSuite.scala | 42 +++
.../celeborn/client/WorkerStatusTrackerSuite.scala | 31 ++
common/src/main/proto/TransportMessages.proto | 13 +
.../org/apache/celeborn/common/CelebornConf.scala | 63 ++++
.../celeborn/common/metrics/ClientMetric.scala | 25 ++
.../common/metrics/source/AbstractSource.scala | 111 +++++-
.../common/protocol/message/ControlMessages.scala | 32 +-
.../metrics/source/CelebornSourceSuite.scala | 94 +++++
.../apache/celeborn/common/util/UtilsSuite.scala | 31 +-
docs/configuration/client.md | 2 +
docs/configuration/master.md | 3 +
docs/configuration/metrics.md | 5 +
docs/monitoring.md | 11 +
.../master/clustermeta/AbstractMetaManager.java | 32 +-
.../master/clustermeta/IMetadataHandler.java | 2 +
.../clustermeta/SingleMasterMetaManager.java | 16 +-
.../master/clustermeta/ha/HAMasterMetaManager.java | 14 +
.../deploy/master/clustermeta/ha/MetaHandler.java | 4 +-
master/src/main/proto/Resource.proto | 2 +
.../deploy/master/ApplicationMetricsSource.scala | 133 +++++++
.../celeborn/service/deploy/master/Master.scala | 33 +-
.../clustermeta/DefaultMetaSystemSuiteJ.java | 305 +++++++++++++++-
.../clustermeta/ha/MasterStateMachineSuiteJ.java | 29 ++
.../ha/RatisMasterStatusSystemSuiteJ.java | 382 ++++++++++++++++++++-
.../master/ApplicationMetricsSourceSuite.scala | 292 ++++++++++++++++
29 files changed, 2057 insertions(+), 35 deletions(-)
diff --git a/assets/grafana/celeborn-dashboard.json
b/assets/grafana/celeborn-dashboard.json
index a3c1ae80fd..0a505040e9 100644
--- a/assets/grafana/celeborn-dashboard.json
+++ b/assets/grafana/celeborn-dashboard.json
@@ -18256,6 +18256,296 @@
],
"title": "Rpc",
"type": "row"
+ },
+ {
+ "collapsed": true,
+ "gridPos": {
+ "h": 1,
+ "w": 24,
+ "x": 0,
+ "y": 111
+ },
+ "id": 277,
+ "panels": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "The number of active shuffles reported by the
client.",
+ "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": 80
+ }
+ ]
+ }
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 1
+ },
+ "id": 278,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "single",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr":
"metrics_ClientActiveShuffleCount_Value{instance=~\"${instance}\"}",
+ "legendFormat": "${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_ClientActiveShuffleCount_Value",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "The number of workers excluded by the client.",
+ "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": 80
+ }
+ ]
+ }
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 12,
+ "y": 1
+ },
+ "id": 279,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "single",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr":
"metrics_ClientExcludedWorkerCount_Value{instance=~\"${instance}\"}",
+ "legendFormat": "${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_ClientExcludedWorkerCount_Value",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "The number of workers in shutting-down state as seen
by the client.",
+ "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": 80
+ }
+ ]
+ }
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 9
+ },
+ "id": 280,
+ "options": {
+ "legend": {
+ "calcs": [],
+ "displayMode": "list",
+ "placement": "bottom",
+ "showLegend": true
+ },
+ "tooltip": {
+ "mode": "single",
+ "sort": "none"
+ }
+ },
+ "targets": [
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "editorMode": "code",
+ "expr":
"metrics_ClientShuttingWorkerCount_Value{instance=~\"${instance}\"}",
+ "legendFormat": "${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_ClientShuttingWorkerCount_Value",
+ "type": "timeseries"
+ }
+ ],
+ "title": "Client Side Metrics",
+ "type": "row"
}
],
"preload": false,
diff --git
a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
index 84f4d07253..8797cfdfb7 100644
---
a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
+++
b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
@@ -28,6 +28,7 @@ import org.apache.commons.lang3.StringUtils
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.client.MasterClient
import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.metrics.ClientMetric
import org.apache.celeborn.common.protocol.PbReviseLostShufflesResponse
import
org.apache.celeborn.common.protocol.message.ControlMessages.{ApplicationLost,
ApplicationLostResponse, CheckQuotaResponse, HeartbeatFromApplication,
HeartbeatFromApplicationResponse, ReviseLostShuffles, ZERO_UUID}
import org.apache.celeborn.common.protocol.message.StatusCode
@@ -42,10 +43,28 @@ class ApplicationHeartbeater(
(Long, Long, Map[String, java.lang.Long], Map[String,
java.lang.Long])),
workerStatusTracker: WorkerStatusTracker,
registeredShuffles: ConcurrentHashMap.KeySetView[Int, java.lang.Boolean],
- cancelAllActiveStages: String => Unit) extends Logging {
+ cancelAllActiveStages: String => Unit,
+ clientMetrics: () => util.Map[String, ClientMetric] =
+ () => java.util.Collections.emptyMap[String, ClientMetric]()) extends
Logging {
private var stopped = false
private val reviseLostShuffles = conf.reviseLostShufflesEnabled
+ private val appMetricLabels: util.Map[String, String] =
+ conf.clientMetricsAppLabels.asJava
+
+ if (conf.metricsSystemEnable && conf.clientMetricsEnabled &&
appMetricLabels.isEmpty) {
+ logWarning(
+ s"Client metrics are enabled
('${CelebornConf.CLIENT_METRICS_ENABLED.key}'=true) but " +
+ s"'${CelebornConf.CLIENT_METRICS_APP_LABELS.key}' is empty. Client
metrics are only " +
+ "emitted in application heartbeats when at least one app label is set,
so no client " +
+ "metrics will reach the master. Set app labels (e.g. env=prod) to
enable emission.")
+ } else if (!conf.metricsSystemEnable && conf.clientMetricsEnabled) {
+ logWarning(
+ s"Client metrics are enabled
('${CelebornConf.CLIENT_METRICS_ENABLED.key}'=true) but " +
+ s"the metrics system is disabled
('${CelebornConf.METRICS_ENABLED.key}'=false), so no " +
+ "client metrics will reach the master. Set " +
+ s"'${CelebornConf.METRICS_ENABLED.key}'=true to enable emission.")
+ }
// Use independent app heartbeat threads to avoid being blocked by other
operations.
private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs
@@ -89,7 +108,10 @@ 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.")
diff --git
a/client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala
b/client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala
new file mode 100644
index 0000000000..7d78152747
--- /dev/null
+++
b/client/src/main/scala/org/apache/celeborn/client/CelebornClientSource.scala
@@ -0,0 +1,47 @@
+/*
+ * 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"
+
+ def getMetricsSnapshot(): Map[String, ClientMetric] = {
+ val snapshot = Map.newBuilder[String, ClientMetric]
+ gauges().foreach { g =>
+ snapshot += g.name -> ClientMetric(
+ g.gauge.getValue.asInstanceOf[Number].longValue(),
+ MetricType.Gauge)
+ }
+ snapshot.result()
+ }
+
+ def stop(): Unit = metricsCleaner.shutdown()
+}
+
+object CelebornClientSource {
+ val EXCLUDED_WORKER_COUNT = "ClientExcludedWorkerCount"
+ val SHUTTING_WORKER_COUNT = "ClientShuttingWorkerCount"
+ val ACTIVE_SHUFFLE_COUNT = "ClientActiveShuffleCount"
+}
diff --git
a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
index 8e90ddd9ad..3b4ea83ef7 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -45,6 +45,7 @@ import
org.apache.celeborn.common.client.{ApplicationInfoProvider, MasterClient}
import org.apache.celeborn.common.identity.{IdentityProvider, UserIdentifier}
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{ApplicationMeta,
ShufflePartitionLocationInfo, WorkerInfo}
+import org.apache.celeborn.common.metrics.ClientMetric
import org.apache.celeborn.common.metrics.source.Role
import org.apache.celeborn.common.network.protocol.{SerdeVersion,
TransportMessagesHelper}
import org.apache.celeborn.common.network.sasl.registration.RegistrationInfo
@@ -226,8 +227,23 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
}
private val masterClient = new MasterClient(masterRpcEnvInUse, conf, false)
+ private val clientMetricsEnabled = conf.metricsSystemEnable &&
conf.clientMetricsEnabled
+ val clientSource: Option[CelebornClientSource] =
+ if (clientMetricsEnabled) Some(new CelebornClientSource(conf)) else None
+
val commitManager = new CommitManager(appUniqueId, conf, this)
val workerStatusTracker = new WorkerStatusTracker(conf, this)
+ clientSource.foreach { source =>
+ source.addGauge(CelebornClientSource.ACTIVE_SHUFFLE_COUNT) { () =>
+ registeredShuffle.size
+ }
+ source.addGauge(CelebornClientSource.EXCLUDED_WORKER_COUNT) { () =>
+ workerStatusTracker.excludedWorkers.size
+ }
+ source.addGauge(CelebornClientSource.SHUTTING_WORKER_COUNT) { () =>
+ workerStatusTracker.shuttingWorkers.size
+ }
+ }
private val heartbeater =
new ApplicationHeartbeater(
appUniqueId,
@@ -240,7 +256,10 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
},
workerStatusTracker,
registeredShuffle,
- reason => cancelAllActiveStages(reason))
+ reason => cancelAllActiveStages(reason),
+ () =>
+ clientSource.map(_.getMetricsSnapshot().asJava)
+ .getOrElse(java.util.Collections.emptyMap[String, ClientMetric]()))
private def resetFallbackCounts(counts: ConcurrentHashMap[String,
java.lang.Long])
: Map[String, java.lang.Long] = {
val fallbackCounts = new util.HashMap[String, java.lang.Long]()
@@ -2183,6 +2202,7 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
*/
override def stop(): Unit = {
heartbeater.stop()
+ clientSource.foreach(_.stop())
super.stop()
}
diff --git
a/client/src/test/scala/org/apache/celeborn/client/CelebornClientSourceSuite.scala
b/client/src/test/scala/org/apache/celeborn/client/CelebornClientSourceSuite.scala
new file mode 100644
index 0000000000..86b785c5bf
--- /dev/null
+++
b/client/src/test/scala/org/apache/celeborn/client/CelebornClientSourceSuite.scala
@@ -0,0 +1,42 @@
+/*
+ * 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("gauges registered on the source are reflected in metrics and
snapshot") {
+ val source = new CelebornClientSource(new CelebornConf())
+ val excluded = new AtomicInteger(0)
+
+ source.addGauge(CelebornClientSource.EXCLUDED_WORKER_COUNT) { () =>
excluded.get() }
+
+
assert(source.getMetricsSnapshot()(CelebornClientSource.EXCLUDED_WORKER_COUNT).value
== 0)
+
+ excluded.set(5)
+ val metrics = source.getMetrics
+ assert(metrics.contains("metrics_ClientExcludedWorkerCount_Value"))
+ val snapshot = source.getMetricsSnapshot()
+ assert(snapshot(CelebornClientSource.EXCLUDED_WORKER_COUNT).value == 5)
+ assert(snapshot(CelebornClientSource.EXCLUDED_WORKER_COUNT).metricType ==
MetricType.Gauge)
+ }
+}
diff --git
a/client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala
b/client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala
index 43cbefd45e..96281a5045 100644
---
a/client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala
+++
b/client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala
@@ -27,6 +27,7 @@ import scala.collection.mutable.ArrayBuffer
import org.junit.Assert
import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.client.LifecycleManager.ShuffleFailedWorkers
import org.apache.celeborn.common.CelebornConf
import
org.apache.celeborn.common.CelebornConf.{CLIENT_EXCLUDED_WORKER_EXPIRE_TIMEOUT,
CLIENT_SHUFFLE_DYNAMIC_RESOURCE_ENABLED}
import org.apache.celeborn.common.meta.WorkerInfo
@@ -184,6 +185,36 @@ class WorkerStatusTrackerSuite extends CelebornFunSuite {
errors.get())
}
+ test("recordWorkerFailure updates client worker-excluded gauge") {
+ val celebornConf = new CelebornConf()
+ // Wire a bare WorkerStatusTracker + CelebornClientSource exactly like
LifecycleManager does
+ // for the excluded-workers gauge, without standing up an RpcEnv or
background threads.
+ val statusTracker = new WorkerStatusTracker(celebornConf, null)
+ val source = new CelebornClientSource(celebornConf)
+ source.addGauge(CelebornClientSource.EXCLUDED_WORKER_COUNT) { () =>
+ statusTracker.excludedWorkers.size
+ }
+ try {
+ 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)
+
+ Assert.assertEquals(
+ 2L,
+
source.getMetricsSnapshot()(CelebornClientSource.EXCLUDED_WORKER_COUNT).value)
+
+ // re-recording already-excluded workers does not change the gauge
+ statusTracker.recordWorkerFailure(failed)
+ Assert.assertEquals(
+ 2L,
+
source.getMetricsSnapshot()(CelebornClientSource.EXCLUDED_WORKER_COUNT).value)
+ } finally {
+ source.stop()
+ }
+ }
+
private def buildResponse(
excludedWorkerHosts: Array[String],
unknownWorkerHosts: Array[String],
diff --git a/common/src/main/proto/TransportMessages.proto
b/common/src/main/proto/TransportMessages.proto
index e57055391e..e136d3c0c7 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -518,6 +518,17 @@ message PbHeartbeatFromApplication {
int64 applicationCount = 9;
map<string, int64> applicationFallbackCounts = 10;
PbHeartbeatInfo heartbeatInfo = 11;
+ map<string, PbClientMetric> clientMetrics = 12;
+ map<string, string> metricLabels = 13;
+}
+
+enum PbMetricType {
+ GAUGE = 0;
+}
+
+message PbClientMetric {
+ int64 value = 1;
+ PbMetricType type = 2;
}
message PbHeartbeatFromApplicationResponse {
@@ -1055,6 +1066,8 @@ message PbMetaAppHeartbeatRequest {
PbHeartbeatInfo heartbeatInfo = 7;
int64 applicationCount = 8;
map<string, int64> applicationFallbackCounts = 9;
+ map<string, int64> clientGauges = 10;
+ map<string, string> clientMetricLabels = 11;
}
message PbMetaWorkerExcludeRequest {
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 3b4e5f99ca..32f120d4a1 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -917,6 +917,12 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
// //////////////////////////////////////////////////////
def metricsConf: Option[String] = get(METRICS_CONF)
def metricsSystemEnable: Boolean = get(METRICS_ENABLED)
+ def clientMetricsEnabled: Boolean = get(CLIENT_METRICS_ENABLED)
+ def masterClientMetricsEnabled: Boolean = get(MASTER_CLIENT_METRICS_ENABLED)
+ def masterClientMetricsRemovedAppRetentionMs: Long =
+ get(MASTER_CLIENT_METRICS_REMOVED_APP_RETENTION)
+ def masterClientMetricsSeriesCardinalityWarnThreshold: Int =
+ get(MASTER_CLIENT_METRICS_SERIES_CARDINALITY_WARN_THRESHOLD)
def metricsSampleRate: Double = get(METRICS_SAMPLE_RATE)
def metricsPrefix: String = get(METRICS_PREFIX)
def metricsSlidingWindowSize: Int = get(METRICS_SLIDING_WINDOW_SIZE)
@@ -924,6 +930,8 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable
with Logging with Se
def metricsCapacity: Int = get(METRICS_CAPACITY)
def metricsExtraLabels: Map[String, String] =
get(METRICS_EXTRA_LABELS).map(Utils.parseKeyValuePair).toMap
+ def clientMetricsAppLabels: Map[String, String] =
+ get(CLIENT_METRICS_APP_LABELS).map(Utils.parseKeyValuePair).toMap
def metricsWorkerAppTopResourceConsumptionCount: Int =
get(METRICS_WORKER_APP_TOP_RESOURCE_CONSUMPTION_COUNT)
def metricsWorkerAppTopResourceConsumptionBytesWrittenThreshold: Long =
@@ -6095,6 +6103,47 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
+ val CLIENT_METRICS_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.metrics.enabled")
+ .categories("client", "metrics")
+ .doc("When true, the LifecycleManager collects client-side metrics. " +
+ "Requires `celeborn.metrics.enabled` to also be true. Note that client
metrics are only " +
+ "emitted in application heartbeats when
`celeborn.client.metrics.appLabels` is set.")
+ .version("0.7.0")
+ .booleanConf
+ .createWithDefault(false)
+
+ val MASTER_CLIENT_METRICS_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.metrics.master.clientMetrics.enabled")
+ .categories("master", "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.retention")
+ .categories("master", "metrics")
+ .doc("How long to retain removed application IDs in the client metrics
source to " +
+ "reject late heartbeats after an application is lost. Entries older
than this are " +
+ "periodically evicted.")
+ .version("0.7.0")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefaultString("120s")
+
+ val MASTER_CLIENT_METRICS_SERIES_CARDINALITY_WARN_THRESHOLD:
ConfigEntry[Int] =
+
buildConf("celeborn.metrics.master.clientMetrics.seriesCardinality.warnThreshold")
+ .categories("master", "metrics")
+ .doc("Client metric series are keyed only by their (low-cardinality)
label set and are " +
+ "only reclaimed when an application is lost, so a high-cardinality " +
+ "`celeborn.client.metrics.appLabels` configuration can grow the number
of distinct " +
+ "series without bound. If the number of tracked series exceeds this
threshold, the " +
+ "master logs a one-time warning.")
+ .version("0.7.0")
+ .intConf
+ .createWithDefault(1000)
+
val METRICS_SAMPLE_RATE: ConfigEntry[Double] =
buildConf("celeborn.metrics.sample.rate")
.categories("metrics")
@@ -6141,6 +6190,20 @@ object CelebornConf extends Logging {
"Allowed pattern is:
`<label1_key>=<label1_value>[,<label2_key>=<label2_value>]*`")
.createWithDefault(Seq.empty)
+ val CLIENT_METRICS_APP_LABELS: ConfigEntry[Seq[String]] =
+ buildConf("celeborn.client.metrics.appLabels")
+ .categories("client", "metrics")
+ .doc("Custom metric labels sent from the client in each application
heartbeat and applied " +
+ "to client metrics exposed on the master's Prometheus endpoint. " +
+ "Labels' pattern is:
`<label1_key>=<label1_value>[,<label2_key>=<label2_value>]*`; e.g.
`env=prod,version=1`")
+ .version("0.7.0")
+ .stringConf
+ .toSequence
+ .checkValue(
+ labels => labels.map(_ =>
Try(Utils.parseKeyValuePair(_))).forall(_.isSuccess),
+ "Allowed pattern is:
`<label1_key>=<label1_value>[,<label2_key>=<label2_value>]*`")
+ .createWithDefault(Seq.empty)
+
val METRICS_WORKER_APP_TOP_RESOURCE_CONSUMPTION_COUNT: ConfigEntry[Int] =
buildConf("celeborn.metrics.worker.app.topResourceConsumption.count")
.categories("metrics")
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/ClientMetric.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/ClientMetric.scala
new file mode 100644
index 0000000000..a00d60f7b5
--- /dev/null
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/ClientMetric.scala
@@ -0,0 +1,25 @@
+/*
+ * 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.common.metrics
+
+sealed trait MetricType
+object MetricType {
+ case object Gauge extends MetricType
+}
+
+case class ClientMetric(value: Long, metricType: MetricType)
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 2047a7531b..58a3a0ad0c 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
@@ -19,6 +19,7 @@ package org.apache.celeborn.common.metrics.source
import java.util.{Map => JMap}
import java.util.concurrent.{ConcurrentHashMap, ConcurrentLinkedQueue,
ScheduledExecutorService, TimeUnit}
+import java.util.concurrent.atomic.AtomicLong
import scala.collection.JavaConverters._
import scala.collection.mutable
@@ -48,6 +49,25 @@ case class NamedHistogram(name: String, histogram:
Histogram, labels: Map[String
case class NamedTimer(name: String, timer: Timer, labels: Map[String, String])
extends MetricLabels
+case class TrackedGauge(
+ namedGauge: NamedGauge[Long],
+ sum: AtomicLong,
+ perAppValues: ConcurrentHashMap[String, java.lang.Long]) {
+
+ def updateAppValue(appId: String, newValue: java.lang.Long): Unit = {
+ val oldVal =
+ if (newValue != null) {
+ perAppValues.put(appId, newValue)
+ } else {
+ perAppValues.remove(appId)
+ }
+ val delta =
+ (if (newValue != null) newValue.longValue() else 0L) -
+ (if (oldVal != null) oldVal.longValue() else 0L)
+ if (delta != 0L) sum.addAndGet(delta)
+ }
+}
+
abstract class AbstractSource(conf: CelebornConf, role: String)
extends Source with Logging {
override val metricRegistry = new MetricRegistry()
@@ -76,6 +96,8 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
Map("instance" -> s"${Utils.localHostName(conf)}:${conf.masterHttpPort}")
case Role.WORKER =>
Map("instance" -> s"${Utils.localHostName(conf)}:${conf.workerHttpPort}")
+ case Role.CLIENT =>
+ Map("instance" -> Utils.localHostName(conf))
case _ => Map.empty
}
val staticLabels: Map[String, String] = labelsWithCustomizedLabels(Map.empty)
@@ -101,6 +123,9 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
protected val namedHistogram: ConcurrentHashMap[String, NamedHistogram] =
JavaUtils.newConcurrentHashMap[String, NamedHistogram]()
+ protected val namedGaugesWithDetails: ConcurrentHashMap[String,
TrackedGauge] =
+ JavaUtils.newConcurrentHashMap[String, TrackedGauge]()
+
def addTimerMetrics(namedTimer: NamedTimer): Unit = {
val timerMetricsString = getTimerMetrics(namedTimer)
timerMetrics.add(timerMetricsString)
@@ -214,12 +239,58 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
labelsWithCustomizedLabels(labels)))
}
+ protected def isAppRemoved(appId: String): Boolean = false
+
+ protected def addOrUpdateGaugeForApp(
+ name: String,
+ labels: Map[String, String],
+ appId: String,
+ value: Long): Unit = {
+ val key = metricNameWithCustomizedLabels(name, labels)
+ namedGaugesWithDetails.compute(
+ key,
+ new java.util.function.BiFunction[String, TrackedGauge, TrackedGauge] {
+ override def apply(k: String, existing: TrackedGauge): TrackedGauge = {
+ if (isAppRemoved(appId)) {
+ if (existing != null) {
+ existing.updateAppValue(appId, null)
+ if (existing.perAppValues.isEmpty) {
+ metricRegistry.remove(key)
+ return null
+ }
+ }
+ return existing
+ }
+ val tracked =
+ if (existing != null) {
+ existing
+ } else {
+ val sum = new AtomicLong()
+ val gauge = metricRegistry.gauge(
+ key,
+ new GaugeSupplier[Long](() =>
sum.get())).asInstanceOf[Gauge[Long]]
+ TrackedGauge(
+ NamedGauge(name, gauge, labelsWithCustomizedLabels(labels)),
+ sum,
+ JavaUtils.newConcurrentHashMap[String, java.lang.Long]())
+ }
+ tracked.updateAppValue(appId, value)
+ tracked
+ }
+ })
+
+ if (isAppRemoved(appId)) {
+ removeAppFromGauge(key, appId)
+ }
+ }
+
def counters(): List[NamedCounter] = {
namedCounters.values().asScala.toList
}
def gauges(): List[NamedGauge[_]] = {
- namedGauges.values().asScala.toList
+ namedGauges.values().asScala.toList ++
+ namedGaugesWithDetails.values().asScala.toList.map(_.namedGauge)
}
def meters(): List[NamedMeter] = {
@@ -247,7 +318,8 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
}
def gaugeExists(name: String, labels: Map[String, String]): Boolean = {
- namedGauges.containsKey(metricNameWithCustomizedLabels(name, labels))
+ val key = metricNameWithCustomizedLabels(name, labels)
+ namedGauges.containsKey(key) || namedGaugesWithDetails.containsKey(key)
}
def needSample(): Boolean = {
@@ -279,6 +351,39 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
metricNameWithLabel
}
+ protected def removeAppFromMetrics(appId: String): Unit = {
+
namedGaugesWithDetails.keySet().asScala.toList.foreach(removeAppFromGauge(_,
appId))
+ }
+
+ protected def removeAllAppGauges(): Unit = {
+ namedGaugesWithDetails.keySet().asScala.toList.foreach { key =>
+ namedGaugesWithDetails.computeIfPresent(
+ key,
+ new java.util.function.BiFunction[String, TrackedGauge, TrackedGauge] {
+ override def apply(k: String, tracked: TrackedGauge): TrackedGauge =
{
+ metricRegistry.remove(key)
+ null
+ }
+ })
+ }
+ }
+
+ private def removeAppFromGauge(key: String, appId: String): Unit = {
+ namedGaugesWithDetails.computeIfPresent(
+ key,
+ new java.util.function.BiFunction[String, TrackedGauge, TrackedGauge] {
+ override def apply(k: String, tracked: TrackedGauge): TrackedGauge = {
+ tracked.updateAppValue(appId, null)
+ if (tracked.perAppValues.isEmpty) {
+ metricRegistry.remove(key)
+ null
+ } else {
+ tracked
+ }
+ }
+ })
+ }
+
override def sample[T](metricsName: String, key: String)(f: => T): T = {
sample(metricsName, key, Map.empty[String, String])(f)
}
@@ -640,6 +745,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
namedTimers.size() +
namedMeters.size() +
namedGauges.size() +
+ namedGaugesWithDetails.size() +
namedCounters.size()
sum
}
@@ -693,6 +799,7 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
metricsCleaner.shutdown()
namedCounters.clear()
namedGauges.clear()
+ namedGaugesWithDetails.clear()
namedMeters.clear()
namedTimers.clear()
timerMetrics.clear()
diff --git
a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
index 580c986ae6..580291d34b 100644
---
a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala
@@ -29,6 +29,7 @@ import org.roaringbitmap.RoaringBitmap
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DiskInfo, WorkerInfo, WorkerStatus}
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
import org.apache.celeborn.common.network.protocol.{SerdeVersion,
TransportMessage}
import org.apache.celeborn.common.protocol._
import org.apache.celeborn.common.protocol.MessageType._
@@ -389,7 +390,11 @@ object ControlMessages extends Logging {
applicationFallbackCounts: util.Map[String, java.lang.Long],
needCheckedWorkerList: util.List[WorkerInfo],
override var requestId: String = ZERO_UUID,
- shouldResponse: Boolean = false) extends MasterRequestMessage
+ shouldResponse: Boolean = false,
+ clientMetrics: util.Map[String, ClientMetric] =
+ java.util.Collections.emptyMap[String, ClientMetric](),
+ metricLabels: util.Map[String, String] =
java.util.Collections.emptyMap[String, String]())
+ extends MasterRequestMessage
case class HeartbeatFromApplicationResponse(
statusCode: StatusCode,
@@ -874,7 +879,9 @@ object ControlMessages extends Logging {
applicationFallbackCounts,
needCheckedWorkerList,
requestId,
- shouldResponse) =>
+ shouldResponse,
+ clientMetrics,
+ metricLabels) =>
val payload = PbHeartbeatFromApplication.newBuilder()
.setAppId(appId)
.setRequestId(requestId)
@@ -887,6 +894,13 @@ object ControlMessages extends Logging {
.addAllNeedCheckedWorkerList(needCheckedWorkerList.asScala.map(
PbSerDeUtils.toPbWorkerInfo(_, true, true)).toList.asJava)
.setShouldResponse(shouldResponse)
+ .putAllClientMetrics(clientMetrics.asScala.map { case (name, metric) =>
+ val pbType = metric.metricType match {
+ case MetricType.Gauge => PbMetricType.GAUGE
+ }
+ name ->
PbClientMetric.newBuilder().setValue(metric.value).setType(pbType).build()
+ }.asJava)
+ .putAllMetricLabels(metricLabels)
.build().toByteArray
new TransportMessage(MessageType.HEARTBEAT_FROM_APPLICATION, payload)
@@ -1377,7 +1391,19 @@ 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.flatMap {
+ case (name, pbMetric) =>
+ pbMetric.getType match {
+ case PbMetricType.GAUGE =>
+ Some(name -> ClientMetric(pbMetric.getValue,
MetricType.Gauge))
+ case unknown =>
+ logWarning(s"Unknown PbMetricType $unknown for metric
$name, skipping")
+ None
+ }
+ }.asJava),
+ new util.HashMap[String,
String](pbHeartbeatFromApplication.getMetricLabelsMap))
case HEARTBEAT_FROM_APPLICATION_RESPONSE_VALUE =>
val pbHeartbeatFromApplicationResponse =
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 c6eeda1956..d148c26d6a 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
@@ -17,11 +17,41 @@
package org.apache.celeborn.common.metrics.source
+import java.util.concurrent.{CyclicBarrier, TimeUnit}
+
import org.apache.celeborn.CelebornFunSuite
import org.apache.celeborn.common.CelebornConf
class CelebornSourceSuite extends CelebornFunSuite {
+ private class TestSource extends AbstractSource(new CelebornConf(),
Role.MASTER) {
+ override def sourceName: String = "testSource"
+
+ def updateGauge(
+ name: String,
+ labels: Map[String, String],
+ appId: String,
+ value: Long): Unit =
+ addOrUpdateGaugeForApp(name, labels, appId, value)
+
+ def removeApp(appId: String): Unit = removeAppFromMetrics(appId)
+
+ def trackedGaugeCount: Int = namedGaugesWithDetails.size()
+ }
+
+ private def hasLabels(
+ actual: Map[String, String],
+ expected: Map[String, String]): Boolean =
+ expected.forall { case (key, value) => actual.get(key).contains(value) }
+
+ private def gaugeValue(
+ source: AbstractSource,
+ labels: Map[String, String],
+ name: String): Option[Long] =
+ source.gauges()
+ .find(g => g.name == name && hasLabels(g.labels, labels))
+ .map(_.gauge.getValue.asInstanceOf[Number].longValue())
+
test("test histogram") {
val conf = new CelebornConf()
@@ -164,4 +194,68 @@ class CelebornSourceSuite extends CelebornFunSuite {
assert(metrics.contains(
s"""metrics_Counter_Count{instance="$instance",role="Master",user="metric"}
2"""))
}
+
+ test("dynamic app gauge reuses existing gauge registration and updates
value") {
+ val source = new TestSource()
+ val labels = Map("user" -> "metric")
+
+ source.updateGauge("DynamicGauge", labels, "app-1", 1L)
+ source.updateGauge("DynamicGauge", labels, "app-1", 42L)
+
+ assert(source.trackedGaugeCount == 1)
+ assert(source.gauges().count(g => g.name == "DynamicGauge" &&
hasLabels(g.labels, labels)) == 1)
+ assert(gaugeValue(source, labels, "DynamicGauge").contains(42L))
+ assert(source.getMetrics.contains("""metrics_DynamicGauge_Value"""))
+ assert(source.getMetrics.contains("""user="metric""""))
+ }
+
+ test("dynamic app gauges are removed only after all contributing apps are
removed") {
+ val source = new TestSource()
+ val labels = Map("user" -> "metric")
+
+ source.updateGauge("DynamicGauge", labels, "app-1", 1L)
+ source.updateGauge("DynamicGauge", labels, "app-2", 2L)
+
+ source.removeApp("app-1")
+
+ assert(source.trackedGaugeCount == 1)
+ assert(gaugeValue(source, labels, "DynamicGauge").contains(2L))
+
+ source.removeApp("app-2")
+
+ assert(source.trackedGaugeCount == 0)
+ assert(gaugeValue(source, labels, "DynamicGauge").isEmpty)
+ assert(!source.gaugeExists("DynamicGauge", labels))
+ }
+
+ test("concurrent add and remove do not orphan a live app's gauge value") {
+ for (_ <- 1 to 50) {
+ val source = new TestSource()
+ val labels = Map("user" -> "metric")
+ val barrier = new CyclicBarrier(2)
+
+ source.updateGauge("G", labels, "app-A", 10L)
+
+ val remover = new Thread(() => {
+ barrier.await(5, TimeUnit.SECONDS)
+ source.removeApp("app-A")
+ })
+ val adder = new Thread(() => {
+ barrier.await(5, TimeUnit.SECONDS)
+ source.updateGauge("G", labels, "app-B", 20L)
+ })
+
+ remover.start()
+ adder.start()
+ remover.join(5000)
+ adder.join(5000)
+
+ assert(
+ source.trackedGaugeCount == 1,
+ "app-B is live so the gauge must still be tracked")
+ assert(
+ gaugeValue(source, labels, "G").contains(20L),
+ "app-B's value must be visible in the gauge")
+ }
+ }
}
diff --git
a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
index 7b115f044d..7f1725383a 100644
--- a/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala
@@ -32,9 +32,10 @@ import
org.apache.celeborn.common.client.{MasterEndpointResolver, StaticMasterEn
import org.apache.celeborn.common.exception.CelebornException
import org.apache.celeborn.common.identity.{DefaultIdentityProvider,
UserIdentifier}
import org.apache.celeborn.common.meta.WorkerInfo
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricType}
import org.apache.celeborn.common.network.protocol.SerdeVersion
import org.apache.celeborn.common.protocol.{PartitionLocation,
PbRequestWorkers, PbRequestWorkersResponse, PbReviseLostShuffles,
PbReviseLostShufflesResponse, StorageInfo, TransportModuleConstants}
-import
org.apache.celeborn.common.protocol.message.ControlMessages.{CommitFilesResponse,
GetReducerFileGroupResponse, MapperEnd, ReviseLostShuffles,
ReviseLostShufflesResponse}
+import
org.apache.celeborn.common.protocol.message.ControlMessages.{CommitFilesResponse,
GetReducerFileGroupResponse, HeartbeatFromApplication, MapperEnd,
ReviseLostShuffles, ReviseLostShufflesResponse}
import org.apache.celeborn.common.protocol.message.StatusCode
class UtilsSuite extends CelebornFunSuite {
@@ -339,6 +340,34 @@ class UtilsSuite extends CelebornFunSuite {
assert(concurrent.size() == 1)
}
+ test("HeartbeatFromApplication carries client metrics through pb serde") {
+ val clientMetrics = new util.HashMap[String, ClientMetric]()
+ clientMetrics.put("ClientExcludedWorkerCount", ClientMetric(2L,
MetricType.Gauge))
+
+ val metricLabels = new util.HashMap[String, String]()
+ metricLabels.put("env", "prod")
+ metricLabels.put("team", "data-eng")
+
+ 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,
+ metricLabels = metricLabels)
+
+ val heartbeatTrans =
Utils.fromTransportMessage(Utils.toTransportMessage(heartbeat))
+ .asInstanceOf[HeartbeatFromApplication]
+
+ assert(heartbeatTrans.clientMetrics == clientMetrics)
+ assert(heartbeatTrans.metricLabels == metricLabels)
+ }
+
test("validate number of client/server netty threads") {
val celebornConf = new CelebornConf()
celebornConf.set("celeborn.io.maxDefaultNettyThreads", "100")
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index c5957be652..04aa511b67 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -50,6 +50,8 @@ license: |
| celeborn.client.flink.resultPartition.supportFloatingBuffer | true | false |
Whether to support floating buffer for result partitions. | 0.3.0 |
remote-shuffle.job.support-floating-buffer-per-output-gate |
| celeborn.client.flink.shuffle.fallback.policy | AUTO | false | Celeborn
supports the following kind of fallback policies. 1. ALWAYS: always use flink
built-in shuffle implementation; 2. AUTO: prefer to use celeborn shuffle
implementation, and fallback to use flink built-in shuffle implementation based
on certain factors, e.g. availability of enough workers and quota; 3. NEVER:
always use celeborn shuffle implementation, and fail fast when it it is
concluded that fallback is required ba [...]
| celeborn.client.inputStream.creation.window | 16 | false | Window size that
CelebornShuffleReader pre-creates CelebornInputStreams, for coalesced scenario
where multiple Partitions are read | 0.5.1 | |
+| celeborn.client.metrics.appLabels | | false | Custom metric labels sent
from the client in each application heartbeat and applied to client metrics
exposed on the master's Prometheus endpoint. Labels' pattern is:
`<label1_key>=<label1_value>[,<label2_key>=<label2_value>]*`; e.g.
`env=prod,version=1` | 0.7.0 | |
+| celeborn.client.metrics.enabled | false | false | When true, the
LifecycleManager collects client-side metrics. Requires
`celeborn.metrics.enabled` to also be true. Note that client metrics are only
emitted in application heartbeats when `celeborn.client.metrics.appLabels` is
set. | 0.7.0 | |
| celeborn.client.mr.pushData.max | 32m | false | Max size for a push data
sent from mr client. | 0.4.0 | |
| celeborn.client.partition.reader.checkpoint.enabled | false | false |
Whether or not checkpoint reads when re-creating a partition reader. Setting to
true minimizes the amount of unnecessary reads during partition read retries |
0.6.0 | |
| celeborn.client.partition.reader.waitLog.threshold | 60s | false | The
threshold in milliseconds for logging partition read wait time. Log messages
will be generated when wait time exceeds multiples of this threshold. | 0.6.2 |
|
diff --git a/docs/configuration/master.md b/docs/configuration/master.md
index 8e5b27c3f0..e22d9ac969 100644
--- a/docs/configuration/master.md
+++ b/docs/configuration/master.md
@@ -89,6 +89,9 @@ license: |
| celeborn.master.userResourceConsumption.metrics.enabled | false | false |
Whether to enable resource consumption metrics. | 0.6.0 | |
| celeborn.master.userResourceConsumption.update.interval | 30s | false | Time
length for a window about compute user resource consumption. | 0.3.0 | |
| celeborn.master.workerUnavailableInfo.expireTimeout | 1800s | false | Worker
unavailable info would be cleared when the retention period is expired. Set -1
to disable the expiration. | 0.3.1 | |
+| celeborn.metrics.master.clientMetrics.enabled | false | false | When true,
the master exposes client-side metrics forwarded in application heartbeats on
its Prometheus endpoint. | 0.7.0 | |
+| celeborn.metrics.master.clientMetrics.removedApp.retention | 120s | false |
How long to retain removed application IDs in the client metrics source to
reject late heartbeats after an application is lost. Entries older than this
are periodically evicted. | 0.7.0 | |
+| celeborn.metrics.master.clientMetrics.seriesCardinality.warnThreshold | 1000
| false | Client metric series are keyed only by their (low-cardinality) label
set and are only reclaimed when an application is lost, so a high-cardinality
`celeborn.client.metrics.appLabels` configuration can grow the number of
distinct series without bound. If the number of tracked series exceeds this
threshold, the master logs a one-time warning. | 0.7.0 | |
| celeborn.quota.cluster.enabled | true | false | Whether to enable
cluster-level quota. | 0.6.0 | |
| celeborn.quota.enabled | true | false | When Master side sets to true, the
master will enable to check the quota via QuotaManager. When Client side sets
to true, LifecycleManager will request Master side to check whether the current
user has enough quota before registration of shuffle. Fallback to the default
shuffle service when Master side checks that there is no enough quota for
current user. | 0.2.0 | |
| celeborn.quota.tenant.enabled | true | false | Whether to enable
tenant-level quota. | 0.6.0 | |
diff --git a/docs/configuration/metrics.md b/docs/configuration/metrics.md
index 31cd4007d5..93d5c2cbea 100644
--- a/docs/configuration/metrics.md
+++ b/docs/configuration/metrics.md
@@ -19,6 +19,8 @@ license: |
<!--begin-include-->
| Key | Default | isDynamic | Description | Since | Deprecated |
| --- | ------- | --------- | ----------- | ----- | ---------- |
+| celeborn.client.metrics.appLabels | | false | Custom metric labels sent
from the client in each application heartbeat and applied to client metrics
exposed on the master's Prometheus endpoint. Labels' pattern is:
`<label1_key>=<label1_value>[,<label2_key>=<label2_value>]*`; e.g.
`env=prod,version=1` | 0.7.0 | |
+| celeborn.client.metrics.enabled | false | false | When true, the
LifecycleManager collects client-side metrics. Requires
`celeborn.metrics.enabled` to also be true. Note that client metrics are only
emitted in application heartbeats when `celeborn.client.metrics.appLabels` is
set. | 0.7.0 | |
| celeborn.metrics.capacity | 4096 | false | The maximum number of metrics
which a source can use to generate output strings. | 0.2.0 | |
| celeborn.metrics.collectPerfCritical.enabled | false | false | It controls
whether to collect metrics which may affect performance. When enable, Celeborn
collects them. | 0.2.0 | |
| celeborn.metrics.conf | <undefined> | false | Custom metrics
configuration file path. Default use `metrics.properties` in classpath. | 0.3.0
| |
@@ -28,6 +30,9 @@ license: |
| celeborn.metrics.json.pretty.enabled | true | false | When true, view
metrics in json pretty format | 0.4.0 | |
| celeborn.metrics.loggerSink.output.enabled | false | false | Whether to
output scraped metrics to the logger. This config will have effect if you
enabled logger sink.If you will not scrape metrics periodically, do add
`org.apache.celeborn.common.metrics.sink.LoggerSink` to metrics.properties. |
0.6.0 | |
| celeborn.metrics.loggerSink.scrape.interval | 30min | false | The interval
of logger sink to scrape its own metrics. This config will have effect if you
enabled logger sink. If you will not scrape metrics periodically, do add
`org.apache.celeborn.common.metrics.sink.LoggerSink` to metrics.properties. |
0.6.0 | |
+| celeborn.metrics.master.clientMetrics.enabled | false | false | When true,
the master exposes client-side metrics forwarded in application heartbeats on
its Prometheus endpoint. | 0.7.0 | |
+| celeborn.metrics.master.clientMetrics.removedApp.retention | 120s | false |
How long to retain removed application IDs in the client metrics source to
reject late heartbeats after an application is lost. Entries older than this
are periodically evicted. | 0.7.0 | |
+| celeborn.metrics.master.clientMetrics.seriesCardinality.warnThreshold | 1000
| false | Client metric series are keyed only by their (low-cardinality) label
set and are only reclaimed when an application is lost, so a high-cardinality
`celeborn.client.metrics.appLabels` configuration can grow the number of
distinct series without bound. If the number of tracked series exceeds this
threshold, the master logs a one-time warning. | 0.7.0 | |
| celeborn.metrics.prefix | metrics | false | Prefix metrics with this value.
It must be a valid Prometheus metric name, matching the regex
`[a-zA-Z_][a-zA-Z0-9_]*`. | 1.0.0 | |
| celeborn.metrics.prometheus.path | /metrics/prometheus | false | URI context
path of prometheus metrics HTTP server. | 0.4.0 | |
| celeborn.metrics.sample.rate | 1.0 | false | It controls if Celeborn collect
timer metrics for some operations. Its value should be in [0.0, 1.0]. | 0.2.0 |
|
diff --git a/docs/monitoring.md b/docs/monitoring.md
index bbdb90e3fd..22e8f1062d 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -175,6 +175,17 @@ These metrics are exposed by Celeborn master.
| thread_is_terminated_count | The terminated thread count of current
thread group.
|
| thread_is_shutdown_count | The shutdown thread count of current
thread group.
|
+ - namespace=application
+ - **notes:**
+ - These are client-side metrics forwarded to the master via
application heartbeats. They are exposed on the master's Prometheus endpoint
when `celeborn.metrics.master.clientMetrics.enabled` is true and
`celeborn.client.metrics.appLabels` is set on the client.
+ - Gauge values are aggregated (summed) across all reporting
applications that share the same label set.
+
+ | Metric Name | Description
|
+
|------------------------------|--------------------------------------------------------------------------------|
+ | ClientActiveShuffleCount | The number of active shuffles on the
client. |
+ | ClientExcludedWorkerCount | The number of workers excluded by the
client. |
+ | ClientShuttingWorkerCount | The number of workers in shutting-down
state as seen by the client. |
+
#### Worker
These metrics are exposed by Celeborn worker.
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
index f726c28919..e7a2506155 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java
@@ -56,6 +56,7 @@ import org.apache.celeborn.common.util.JavaUtils;
import org.apache.celeborn.common.util.PbSerDeUtils;
import org.apache.celeborn.common.util.Utils;
import org.apache.celeborn.common.util.WorkerStatusUtils;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
/**
* Note: Do not update the worker collections directly from outside the
metadata manager, especially
@@ -106,6 +107,16 @@ public abstract class AbstractMetaManager implements
IMetadataHandler {
public final ConcurrentHashMap<String, ApplicationMeta> applicationMetas =
JavaUtils.newConcurrentHashMap();
+ private final ApplicationMetricsSource applicationMetricsSource;
+
+ protected AbstractMetaManager(ApplicationMetricsSource
applicationMetricsSource) {
+ this.applicationMetricsSource = applicationMetricsSource;
+ }
+
+ public ApplicationMetricsSource applicationMetricsSource() {
+ return applicationMetricsSource;
+ }
+
public void updateApplicationInfo(
String appId, UserIdentifier userIdentifier, Map<String, String>
extraInfo) {
applicationInfos.putIfAbsent(
@@ -169,7 +180,9 @@ public abstract class AbstractMetaManager implements
IMetadataHandler {
long shuffleCount,
long applicationCount,
Map<String, Long> shuffleFallbackCounts,
- Map<String, Long> applicationFallbackCounts) {
+ Map<String, Long> applicationFallbackCounts,
+ Map<String, Long> clientGauges,
+ Map<String, String> clientMetricLabels) {
appHeartbeatTime.put(appId, time);
partitionTotalWritten.add(totalWritten);
partitionTotalFileCount.add(fileCount);
@@ -177,6 +190,13 @@ public abstract class AbstractMetaManager implements
IMetadataHandler {
applicationTotalCount.add(applicationCount);
addFallbackCounts(this.shuffleFallbackCounts, shuffleFallbackCounts);
addFallbackCounts(this.applicationFallbackCounts,
applicationFallbackCounts);
+ if (!clientMetricLabels.isEmpty()) {
+ try {
+ applicationMetricsSource.updateApplicationMetrics(appId,
clientMetricLabels, clientGauges);
+ } catch (Throwable t) {
+ LOG.warn("Failed to update client metrics for app {}", appId, t);
+ }
+ }
}
public void updateAppLostMeta(String appId) {
@@ -184,6 +204,11 @@ public abstract class AbstractMetaManager implements
IMetadataHandler {
appHeartbeatTime.remove(appId);
applicationMetas.remove(appId);
applicationInfos.remove(appId);
+ try {
+ applicationMetricsSource.removeApplicationMetrics(appId);
+ } catch (Throwable t) {
+ LOG.warn("Failed to remove client metrics for app {}", appId, t);
+ }
}
@VisibleForTesting
@@ -566,6 +591,11 @@ public abstract class AbstractMetaManager implements
IMetadataHandler {
workerEventInfos.clear();
applicationMetas.clear();
applicationInfos.clear();
+ try {
+ applicationMetricsSource.clearApplicationMetrics();
+ } catch (Throwable t) {
+ LOG.warn("Failed to clear client metrics while restoring meta", t);
+ }
}
public void updateMetaByReportWorkerUnavailable(List<WorkerInfo>
failedWorkers) {
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/IMetadataHandler.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/IMetadataHandler.java
index e3e063348f..07e440252b 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/IMetadataHandler.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/IMetadataHandler.java
@@ -50,6 +50,8 @@ public interface IMetadataHandler {
Map<String, Long> shuffleFallbackCounts,
Map<String, Long> applicationFallbackCounts,
long time,
+ Map<String, Long> clientGauges,
+ Map<String, String> clientMetricLabels,
String requestId);
void handleAppLost(String appId, String requestId);
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
index c4b2e843f2..2f64d7ae53 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
@@ -32,6 +32,7 @@ import org.apache.celeborn.common.meta.WorkerStatus;
import org.apache.celeborn.common.network.CelebornRackResolver;
import org.apache.celeborn.common.quota.ResourceConsumption;
import org.apache.celeborn.common.rpc.RpcEnv;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
public class SingleMasterMetaManager extends AbstractMetaManager {
private static final Logger LOG =
LoggerFactory.getLogger(SingleMasterMetaManager.class);
@@ -42,6 +43,15 @@ public class SingleMasterMetaManager extends
AbstractMetaManager {
public SingleMasterMetaManager(
RpcEnv rpcEnv, CelebornConf conf, CelebornRackResolver rackResolver) {
+ this(rpcEnv, conf, rackResolver, new ApplicationMetricsSource(conf));
+ }
+
+ public SingleMasterMetaManager(
+ RpcEnv rpcEnv,
+ CelebornConf conf,
+ CelebornRackResolver rackResolver,
+ ApplicationMetricsSource applicationMetricsSource) {
+ super(applicationMetricsSource);
this.rpcEnv = rpcEnv;
this.conf = conf;
this.initialEstimatedPartitionSize = conf.initialEstimatedPartitionSize();
@@ -92,6 +102,8 @@ public class SingleMasterMetaManager extends
AbstractMetaManager {
Map<String, Long> shuffleFallbackCounts,
Map<String, Long> applicationFallbackCounts,
long time,
+ Map<String, Long> clientGauges,
+ Map<String, String> clientMetricLabels,
String requestId) {
updateAppHeartbeatMeta(
appId,
@@ -101,7 +113,9 @@ public class SingleMasterMetaManager extends
AbstractMetaManager {
shuffleCount,
applicationCount,
shuffleFallbackCounts,
- applicationFallbackCounts);
+ applicationFallbackCounts,
+ clientGauges,
+ clientMetricLabels);
}
@Override
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
index 3372143aa2..55f717d16f 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
@@ -35,6 +35,7 @@ import org.apache.celeborn.common.meta.WorkerStatus;
import org.apache.celeborn.common.network.CelebornRackResolver;
import org.apache.celeborn.common.quota.ResourceConsumption;
import org.apache.celeborn.common.rpc.RpcEnv;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
import
org.apache.celeborn.service.deploy.master.clustermeta.AbstractMetaManager;
import org.apache.celeborn.service.deploy.master.clustermeta.MetaUtil;
import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos;
@@ -51,6 +52,15 @@ public class HAMasterMetaManager extends AbstractMetaManager
{
}
public HAMasterMetaManager(RpcEnv rpcEnv, CelebornConf conf,
CelebornRackResolver rackResolver) {
+ this(rpcEnv, conf, rackResolver, new ApplicationMetricsSource(conf));
+ }
+
+ public HAMasterMetaManager(
+ RpcEnv rpcEnv,
+ CelebornConf conf,
+ CelebornRackResolver rackResolver,
+ ApplicationMetricsSource applicationMetricsSource) {
+ super(applicationMetricsSource);
this.rpcEnv = rpcEnv;
this.conf = conf;
this.initialEstimatedPartitionSize = conf.initialEstimatedPartitionSize();
@@ -167,6 +177,8 @@ public class HAMasterMetaManager extends
AbstractMetaManager {
Map<String, Long> shuffleFallbackCounts,
Map<String, Long> applicationFallbackCounts,
long time,
+ Map<String, Long> clientGauges,
+ Map<String, String> clientMetricLabels,
String requestId) {
try {
ratisServer.submitRequest(
@@ -183,6 +195,8 @@ public class HAMasterMetaManager extends
AbstractMetaManager {
.setApplicationCount(applicationCount)
.putAllShuffleFallbackCounts(shuffleFallbackCounts)
.putAllApplicationFallbackCounts(applicationFallbackCounts)
+ .putAllClientGauges(clientGauges)
+ .putAllClientMetricLabels(clientMetricLabels)
.build())
.build());
} catch (CelebornRuntimeException e) {
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
index 86097117e8..834b201086 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MetaHandler.java
@@ -181,7 +181,9 @@ public class MetaHandler {
shuffleCount,
applicationCount,
shuffleFallbackCounts,
- applicationFallbackCounts);
+ applicationFallbackCounts,
+ request.getAppHeartbeatRequest().getClientGaugesMap(),
+ request.getAppHeartbeatRequest().getClientMetricLabelsMap());
break;
case AppLost:
diff --git a/master/src/main/proto/Resource.proto
b/master/src/main/proto/Resource.proto
index d8fceee156..51eab97715 100644
--- a/master/src/main/proto/Resource.proto
+++ b/master/src/main/proto/Resource.proto
@@ -124,6 +124,8 @@ message AppHeartbeatRequest {
HeartbeatInfo heartbeatInfo = 7;
optional int64 applicationCount = 8;
map<string, int64> applicationFallbackCounts = 9;
+ map<string, int64> clientGauges = 10;
+ map<string, string> clientMetricLabels = 11;
}
message AppLostRequest {
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala
new file mode 100644
index 0000000000..5224a63645
--- /dev/null
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSource.scala
@@ -0,0 +1,133 @@
+/*
+ * 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.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 ValidLabelName = "^[a-zA-Z_][a-zA-Z0-9_]*$".r.pattern
+ private val UnsafeLabelValue = """["\\\n\r]""".r
+
+ 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: JMap[String, String],
+ gauges: JMap[String, java.lang.Long]): Unit = {
+ if (!masterClientMetricsEnabled) {
+ return
+ }
+ if (removedAppIds.containsKey(appId)) {
+ removeAppFromMetrics(appId)
+ return
+ }
+
+ val labels = metricLabels.asScala.toMap
+ if (!validateLabels(labels)) {
+ logWarning(s"Ignoring client metrics from $appId: labels contain invalid
Prometheus " +
+ "label names or unsafe values (quotes, backslashes, or newlines)")
+ return
+ }
+
+ gauges.asScala.foreach { case (name, value) =>
+ addOrUpdateGaugeForApp(name, labels, appId, value)
+ }
+
+ warnIfSeriesCardinalityHigh()
+ }
+
+ override protected def isAppRemoved(appId: String): Boolean =
+ removedAppIds.containsKey(appId)
+
+ def removeApplicationMetrics(appId: String): Unit = {
+ if (masterClientMetricsEnabled) {
+ removedAppIds.put(appId, System.currentTimeMillis())
+ }
+ removeAppFromMetrics(appId)
+ }
+
+ def clearApplicationMetrics(): Unit = {
+ removeAllAppGauges()
+ }
+
+ private def validateLabels(labels: Map[String, String]): Boolean = {
+ labels.forall { case (key, value) =>
+ ValidLabelName.matcher(key).matches() &&
UnsafeLabelValue.findFirstIn(value).isEmpty
+ }
+ }
+
+ private def warnIfSeriesCardinalityHigh(): Unit = {
+ if (seriesCardinalityWarned.get()) {
+ return
+ }
+ val trackedSeries = namedGauges.size() + namedGaugesWithDetails.size() +
namedCounters.size()
+ if (trackedSeries > seriesCardinalityWarnThreshold &&
seriesCardinalityWarned.compareAndSet(
+ false,
+ true)) {
+ logWarning(
+ s"Client metrics are tracking $trackedSeries distinct series,
exceeding " +
+ s"$seriesCardinalityWarnThreshold. Client metric series are keyed by
" +
+ s"'${CelebornConf.CLIENT_METRICS_APP_LABELS.key}' and are only
reclaimed when an " +
+ "application is lost, so high-cardinality labels can grow memory
without bound. " +
+ "Ensure these labels are low-cardinality (e.g. env/team), not
per-application values.")
+ }
+ }
+}
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 61a444b7ca..de2540e867 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
@@ -41,7 +41,7 @@ import org.apache.celeborn.common.exception.CelebornException
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DiskInfo, WorkerInfo, WorkerStatus}
-import org.apache.celeborn.common.metrics.MetricsSystem
+import org.apache.celeborn.common.metrics.{ClientMetric, MetricsSystem,
MetricType}
import org.apache.celeborn.common.metrics.source.{JVMCPUSource, JVMSource,
ResourceConsumptionSource, Role, SystemMiscSource, ThreadPoolSource}
import org.apache.celeborn.common.network.CelebornRackResolver
import org.apache.celeborn.common.network.protocol.{TransportMessage,
TransportMessagesHelper}
@@ -176,6 +176,7 @@ private[celeborn] class Master(
} else {
new SingleMasterMetaManager(internalRpcEnvInUse, conf, rackResolver)
}
+ metricsSystem.registerSource(statusSystem.applicationMetricsSource())
secretRegistry.setMetadataHandler(statusSystem)
// Threads
@@ -196,6 +197,7 @@ private[celeborn] class Master(
private val dfsExpireDirsTimeoutMS = conf.dfsExpireDirsTimeoutMS
private val remoteStorageDirs = conf.remoteStorageDirs
+ private val masterClientMetricsEnabled = conf.masterClientMetricsEnabled
private val quotaManager = new QuotaManager(
statusSystem,
@@ -477,7 +479,9 @@ private[celeborn] class Master(
applicationFallbackCounts,
needCheckedWorkerList,
requestId,
- shouldResponse) =>
+ shouldResponse,
+ clientMetrics,
+ metricLabels) =>
logDebug(s"Received heartbeat from app $appId")
checkAuth(context, appId)
executeWithLeaderChecker(
@@ -493,7 +497,9 @@ private[celeborn] class Master(
applicationFallbackCounts,
needCheckedWorkerList,
requestId,
- shouldResponse))
+ shouldResponse,
+ clientMetrics,
+ metricLabels))
case pbRegisterWorker: PbRegisterWorker =>
val requestId = pbRegisterWorker.getRequestId
@@ -1296,7 +1302,23 @@ private[celeborn] class Master(
applicationFallbackCounts: util.Map[String, java.lang.Long],
needCheckedWorkerList: util.List[WorkerInfo],
requestId: String,
- shouldResponse: Boolean): Unit = {
+ shouldResponse: Boolean,
+ clientMetrics: util.Map[String, ClientMetric],
+ metricLabels: util.Map[String, String]): Unit = {
+ val (clientGauges, clientMetricLabels) =
+ if (masterClientMetricsEnabled && !metricLabels.isEmpty) {
+ val gauges = new util.HashMap[String, java.lang.Long]()
+ clientMetrics.asScala.foreach { case (name, metric) =>
+ metric.metricType match {
+ case MetricType.Gauge => gauges.put(name, metric.value)
+ }
+ }
+ (gauges, metricLabels)
+ } else {
+ (
+ Collections.emptyMap[String, java.lang.Long](),
+ Collections.emptyMap[String, String]())
+ }
statusSystem.handleAppHeartbeat(
appId,
totalWritten,
@@ -1306,7 +1328,10 @@ private[celeborn] class Master(
shuffleFallbackCounts,
applicationFallbackCounts,
System.currentTimeMillis(),
+ clientGauges,
+ clientMetricLabels,
requestId)
+
gaugeShuffleFallbackCounts()
val unknownWorkers = needCheckedWorkerList.asScala.filterNot(w =>
statusSystem.workersMap.containsKey(w.toUniqueId)).asJava
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/DefaultMetaSystemSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/DefaultMetaSystemSuiteJ.java
index 898e7686e6..ed6eedcf91 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/DefaultMetaSystemSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/DefaultMetaSystemSuiteJ.java
@@ -18,12 +18,22 @@
package org.apache.celeborn.service.deploy.master.clustermeta;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.io.File;
+import java.nio.file.Files;
import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.junit.After;
@@ -38,11 +48,13 @@ import org.apache.celeborn.common.meta.ApplicationMeta;
import org.apache.celeborn.common.meta.DiskInfo;
import org.apache.celeborn.common.meta.WorkerInfo;
import org.apache.celeborn.common.meta.WorkerStatus;
+import org.apache.celeborn.common.network.CelebornRackResolver;
import org.apache.celeborn.common.quota.ResourceConsumption;
import org.apache.celeborn.common.rpc.RpcEndpointAddress;
import org.apache.celeborn.common.rpc.RpcEndpointRef;
import org.apache.celeborn.common.rpc.RpcEnv;
import org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
public class DefaultMetaSystemSuiteJ {
@@ -644,12 +656,32 @@ public class DefaultMetaSystemSuiteJ {
public void testHandleAppHeartbeat() {
Long dummy = 1235L;
statusSystem.handleAppHeartbeat(
- APPID1, 1, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
assertEquals(dummy, statusSystem.appHeartbeatTime.get(APPID1));
String appId2 = "app02";
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
assertEquals(dummy, statusSystem.appHeartbeatTime.get(appId2));
assertEquals(2, statusSystem.appHeartbeatTime.size());
@@ -997,28 +1029,88 @@ public class DefaultMetaSystemSuiteJ {
Long dummy = 1235L;
statusSystem.handleAppHeartbeat(
- APPID1, 10000000000l, 1, 1, 1, new HashMap<>(), new HashMap<>(),
dummy, getNewReqeustId());
+ APPID1,
+ 10000000000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
String appId2 = "app02";
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Max size
statusSystem.handleUpdatePartitionSize();
Assert.assertEquals(statusSystem.estimatedPartitionSize,
conf.maxPartitionSizeToEstimate());
statusSystem.handleAppHeartbeat(
- APPID1, 1000000000l, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1000000000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Size between minEstimateSize -> maxEstimateSize
statusSystem.handleUpdatePartitionSize();
Assert.assertEquals(500000000, statusSystem.estimatedPartitionSize);
statusSystem.handleAppHeartbeat(
- APPID1, 1000l, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
statusSystem.handleAppHeartbeat(
- appId2, 1000l, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1000l,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Min size
statusSystem.handleUpdatePartitionSize();
@@ -1111,6 +1203,8 @@ public class DefaultMetaSystemSuiteJ {
shuffleFallbackCounts,
applicationFallbackCounts,
dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
getNewReqeustId());
shuffleFallbackCounts.put(POLICY1, 1L);
shuffleFallbackCounts.put(POLICY2, 2L);
@@ -1125,6 +1219,8 @@ public class DefaultMetaSystemSuiteJ {
shuffleFallbackCounts,
applicationFallbackCounts,
dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
getNewReqeustId());
assertEquals(5, statusSystem.shuffleTotalCount.longValue());
@@ -1147,4 +1243,197 @@ public class DefaultMetaSystemSuiteJ {
assertEquals(statusSystem.applicationInfos.get(appId).userIdentifier(),
userIdentifier);
assertEquals(statusSystem.applicationInfos.get(appId).extraInfo(),
extraInfo);
}
+
+ private static final String CLIENT_GAUGE = "ClientActiveShuffleCount";
+ private static final Map<String, String> CLIENT_LABELS =
+ Collections.singletonMap("team", "data-eng");
+
+ private static CelebornConf clientMetricsConf() {
+ CelebornConf metricsConf = new CelebornConf();
+ metricsConf.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED().key(),
"true");
+ return metricsConf;
+ }
+
+ private AbstractMetaManager statusSystemWithMetrics(ApplicationMetricsSource
source) {
+ return new SingleMasterMetaManager(mockRpcEnv, conf, new
CelebornRackResolver(conf), source);
+ }
+
+ private void heartbeatWithGauge(AbstractMetaManager system, String appId,
long value) {
+ system.handleAppHeartbeat(
+ appId,
+ 1,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ System.currentTimeMillis(),
+ Collections.singletonMap(CLIENT_GAUGE, value),
+ CLIENT_LABELS,
+ getNewReqeustId());
+ }
+
+ /** Sum of the client gauge across apps as exported by the source, or null
if not exported. */
+ private static Long exportedGauge(ApplicationMetricsSource source) {
+ if (source.gauges().isEmpty()) {
+ return null;
+ }
+ return ((Number) source.gauges().head().gauge().getValue()).longValue();
+ }
+
+ /** A metrics source that fails on every call, to exercise the apply-path
guardrails. */
+ private static class FailingMetricsSource extends ApplicationMetricsSource {
+ FailingMetricsSource(CelebornConf conf) {
+ super(conf);
+ }
+
+ @Override
+ public void updateApplicationMetrics(
+ String appId, Map<String, String> metricLabels, Map<String, Long>
gauges) {
+ throw new IllegalArgumentException("injected update failure");
+ }
+
+ @Override
+ public void removeApplicationMetrics(String appId) {
+ throw new IllegalStateException("injected remove failure");
+ }
+
+ @Override
+ public void clearApplicationMetrics() {
+ throw new IllegalStateException("injected clear failure");
+ }
+ }
+
+ @Test
+ public void testDefaultConstructorCreatesApplicationMetricsSource() {
+ AbstractMetaManager system = new SingleMasterMetaManager(mockRpcEnv,
clientMetricsConf());
+ ApplicationMetricsSource source = system.applicationMetricsSource();
+ try {
+ assertNotNull(source);
+ assertNull(exportedGauge(source));
+
+ heartbeatWithGauge(system, APPID1, 7);
+ assertEquals(Long.valueOf(7), exportedGauge(source));
+
+ system.handleAppLost(APPID1, getNewReqeustId());
+ assertNull(exportedGauge(source));
+ } finally {
+ source.destroy();
+ }
+ }
+
+ @Test
+ public void testInjectedApplicationMetricsSourceIsExposed() {
+ ApplicationMetricsSource source = new
ApplicationMetricsSource(clientMetricsConf());
+ try {
+ assertSame(source,
statusSystemWithMetrics(source).applicationMetricsSource());
+ } finally {
+ source.destroy();
+ }
+ }
+
+ @Test
+ public void testClientMetricsFailureDoesNotFailMetaUpdates() throws
Exception {
+ FailingMetricsSource source = new
FailingMetricsSource(clientMetricsConf());
+ File snapshot = Files.createTempFile("celeborn-meta",
".snapshot").toFile();
+ try {
+ statusSystem = statusSystemWithMetrics(source);
+
+ // In HA mode these run inside the Ratis apply, where an exception
terminates the master.
+ heartbeatWithGauge(statusSystem, APPID1, 7);
+ assertTrue(statusSystem.appHeartbeatTime.containsKey(APPID1));
+
+ statusSystem.writeMetaInfoToFile(snapshot);
+ statusSystem.restoreMetaFromFile(snapshot);
+ assertTrue(statusSystem.appHeartbeatTime.containsKey(APPID1));
+
+ statusSystem.handleAppLost(APPID1, getNewReqeustId());
+ Assert.assertFalse(statusSystem.appHeartbeatTime.containsKey(APPID1));
+ } finally {
+ source.destroy();
+ snapshot.delete();
+ }
+ }
+
+ @Test
+ public void testRestoreMetaFromSnapshotClearsClientMetrics() throws
Exception {
+ ApplicationMetricsSource source = new
ApplicationMetricsSource(clientMetricsConf());
+ File snapshot = Files.createTempFile("celeborn-meta",
".snapshot").toFile();
+ String appId2 = "app02";
+ String appId3 = "app03";
+ try {
+ // The snapshot only knows appId2: APPID1's AppLost is covered by it, so
a master
+ // installing it never applies that AppLost.
+ SingleMasterMetaManager snapshotSource = new
SingleMasterMetaManager(mockRpcEnv, conf);
+ heartbeatWithGauge(snapshotSource, appId2, 1);
+ snapshotSource.writeMetaInfoToFile(snapshot);
+
+ statusSystem = statusSystemWithMetrics(source);
+ heartbeatWithGauge(statusSystem, APPID1, 7);
+ heartbeatWithGauge(statusSystem, appId3, 2);
+ statusSystem.handleAppLost(appId3, getNewReqeustId());
+ assertEquals(Long.valueOf(7), exportedGauge(source));
+
+ statusSystem.restoreMetaFromFile(snapshot);
+
+ // Nothing from before the install survives, including APPID1 which
would otherwise never
+ // be removed.
+ Assert.assertNull(exportedGauge(source));
+ Assert.assertFalse(statusSystem.appHeartbeatTime.containsKey(APPID1));
+
+ // Live apps repopulate on their next heartbeat; an app lost before the
install stays
+ // ignored.
+ heartbeatWithGauge(statusSystem, appId2, 5);
+ heartbeatWithGauge(statusSystem, appId3, 9);
+ assertEquals(Long.valueOf(5), exportedGauge(source));
+ } finally {
+ source.destroy();
+ snapshot.delete();
+ }
+ }
+
+ @Test
+ public void testConcurrentHeartbeatsAndAppLostLeaveNoClientMetrics() throws
Exception {
+ // Single-master mode applies heartbeats on RPC threads and AppLost on
another thread, so
+ // unlike HA they are not serialized. Heartbeats still in flight when the
app is lost must
+ // not leave a series behind.
+ ApplicationMetricsSource source = new
ApplicationMetricsSource(clientMetricsConf());
+ int heartbeatThreads = 4;
+ ExecutorService pool = Executors.newFixedThreadPool(heartbeatThreads + 1);
+ try {
+ statusSystem = statusSystemWithMetrics(source);
+ for (int round = 0; round < 200; round++) {
+ String appId = "race-app-" + round;
+ CountDownLatch start = new CountDownLatch(1);
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < heartbeatThreads; t++) {
+ long value = t + 1;
+ futures.add(
+ pool.submit(
+ () -> {
+ start.await();
+ for (int i = 0; i < 20; i++) {
+ heartbeatWithGauge(statusSystem, appId, value);
+ }
+ return null;
+ }));
+ }
+ futures.add(
+ pool.submit(
+ () -> {
+ start.await();
+ statusSystem.handleAppLost(appId, getNewReqeustId());
+ return null;
+ }));
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ Assert.assertNull("series leaked for " + appId, exportedGauge(source));
+ }
+ } finally {
+ pool.shutdownNow();
+ source.destroy();
+ }
+ }
}
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
index 46a9c68e61..f272153664 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
@@ -50,6 +50,7 @@ import org.apache.celeborn.common.quota.ResourceConsumption;
import org.apache.celeborn.common.rpc.RpcEnv;
import org.apache.celeborn.common.util.JavaUtils;
import org.apache.celeborn.common.util.Utils;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos;
public class MasterStateMachineSuiteJ extends RatisBaseSuiteJ {
@@ -124,6 +125,34 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Assert.assertEquals(1, latest.getFiles().size());
}
+ @Test
+ public void testHAMetaManagerCreatesItsOwnApplicationMetricsSource() {
+ CelebornConf conf = new CelebornConf();
+ conf.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED().key(), "true");
+ HAMasterMetaManager system = new HAMasterMetaManager(null, conf);
+ ApplicationMetricsSource source = system.applicationMetricsSource();
+ try {
+ Assert.assertNotNull(source);
+ system.updateAppHeartbeatMeta(
+ "app-1",
+ System.currentTimeMillis(),
+ 1,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ Collections.singletonMap("ClientActiveShuffleCount", 5L),
+ Collections.singletonMap("team", "data-eng"));
+ Assert.assertFalse(source.gauges().isEmpty());
+
+ system.updateAppLostMeta("app-1");
+ Assert.assertTrue(source.gauges().isEmpty());
+ } finally {
+ source.destroy();
+ }
+ }
+
@Test
public void testObjSerde() throws IOException, InterruptedException {
CelebornConf conf = new CelebornConf();
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
index 7ca8a83cf9..0656844c2a 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/RatisMasterStatusSystemSuiteJ.java
@@ -27,10 +27,13 @@ import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
import scala.Tuple2;
import com.google.common.collect.Lists;
+import org.apache.ratis.protocol.TransferLeadershipRequest;
+import org.apache.ratis.rpc.CallId;
import org.apache.ratis.util.LifeCycle;
import org.junit.*;
import org.mockito.Mockito;
@@ -43,6 +46,7 @@ import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.common.meta.DiskInfo;
import org.apache.celeborn.common.meta.WorkerInfo;
import org.apache.celeborn.common.meta.WorkerStatus;
+import org.apache.celeborn.common.network.CelebornRackResolver;
import org.apache.celeborn.common.network.sasl.SaslCredentials;
import org.apache.celeborn.common.network.sasl.SaslTestBase;
import org.apache.celeborn.common.network.sasl.SecretRegistry;
@@ -59,6 +63,7 @@ import org.apache.celeborn.common.rpc.RpcEnv;
import org.apache.celeborn.common.rpc.netty.NettyRpcEndpointRef;
import org.apache.celeborn.common.util.Utils;
import org.apache.celeborn.common.util.Utils$;
+import org.apache.celeborn.service.deploy.master.ApplicationMetricsSource;
import org.apache.celeborn.service.deploy.master.MasterSecretRegistryImpl;
import
org.apache.celeborn.service.deploy.master.clustermeta.AbstractMetaManager;
import org.apache.celeborn.service.deploy.master.clustermeta.ResourceProtos;
@@ -110,6 +115,16 @@ public class RatisMasterStatusSystemSuiteJ {
public static void resetRaftServer(
CelebornConf conf1, CelebornConf conf2, CelebornConf conf3, boolean
sslEnabled)
throws IOException, InterruptedException {
+ resetRaftServer(conf1, conf2, conf3, sslEnabled, new
ApplicationMetricsSource[3]);
+ }
+
+ public static void resetRaftServer(
+ CelebornConf conf1,
+ CelebornConf conf2,
+ CelebornConf conf3,
+ boolean sslEnabled,
+ ApplicationMetricsSource[] metricsSources)
+ throws IOException, InterruptedException {
Mockito.when(mockRpcEnv.setupEndpointRef(Mockito.any(), Mockito.any()))
.thenReturn(mockRpcEndpoint);
when(mockRpcEnv.setupEndpointRef(any(), any())).thenReturn(dummyRef);
@@ -134,9 +149,9 @@ public class RatisMasterStatusSystemSuiteJ {
configureServerConf(conf3, 3);
}
- STATUSSYSTEM1 = new HAMasterMetaManager(mockRpcEnv, conf1);
- STATUSSYSTEM2 = new HAMasterMetaManager(mockRpcEnv, conf2);
- STATUSSYSTEM3 = new HAMasterMetaManager(mockRpcEnv, conf3);
+ STATUSSYSTEM1 = newStatusSystem(conf1, metricsSources[0]);
+ STATUSSYSTEM2 = newStatusSystem(conf2, metricsSources[1]);
+ STATUSSYSTEM3 = newStatusSystem(conf3, metricsSources[2]);
MetaHandler handler1 = new MetaHandler(STATUSSYSTEM1);
MetaHandler handler2 = new MetaHandler(STATUSSYSTEM2);
@@ -209,6 +224,11 @@ public class RatisMasterStatusSystemSuiteJ {
}
}
+ private static HAMasterMetaManager newStatusSystem(
+ CelebornConf conf, ApplicationMetricsSource metricsSource) {
+ return new HAMasterMetaManager(mockRpcEnv, conf, new
CelebornRackResolver(conf), metricsSource);
+ }
+
@Test
public void testLeaderAvailable() {
boolean hasLeader =
@@ -809,6 +829,262 @@ public class RatisMasterStatusSystemSuiteJ {
Assert.assertTrue(STATUSSYSTEM3.registeredAppAndShuffles.isEmpty());
}
+ private static void waitFor(BooleanSupplier condition, String description)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 30_000L;
+ while (!condition.getAsBoolean()) {
+ if (System.currentTimeMillis() > deadline) {
+ fail("Timed out waiting for " + description);
+ }
+ Thread.sleep(100L);
+ }
+ }
+
+ private static void transferLeadership(HARaftServer from, HARaftServer to)
throws Exception {
+ from.getServer()
+ .transferLeadership(
+ new TransferLeadershipRequest(
+ from.getClientId(),
+ from.getServer().getId(),
+ from.getGroupId(),
+ CallId.getAndIncrement(),
+ to.getServer().getId(),
+ 30_000L));
+ waitFor(to::isLeader, "leadership to transfer");
+ // isLeader() trusts a cached LEADER role, so force a refresh on the old
leader to make sure
+ // it has observed the step-down (and fired its leadership listeners)
before asserting.
+ waitFor(
+ () -> {
+ from.updateServerRole();
+ return !from.isLeader();
+ },
+ "old leader to observe step-down");
+ }
+
+ /** A freshly elected leader rejects writes until it has committed an entry
in its term. */
+ private static void submitWhenLeaderReady(Runnable submit) throws
InterruptedException {
+ waitFor(
+ () -> {
+ try {
+ submit.run();
+ return true;
+ } catch (CelebornRuntimeException e) {
+ return false;
+ }
+ },
+ "new leader to accept writes");
+ }
+
+ private static final Map<String, String> CLIENT_LABELS =
+ Collections.singletonMap("team", "data-eng");
+
+ private void heartbeatWithGauge(HAMasterMetaManager leader, String appId,
long value)
+ throws InterruptedException {
+ Map<String, Long> gauges =
Collections.singletonMap("ClientActiveShuffleCount", value);
+ submitWhenLeaderReady(
+ () ->
+ leader.handleAppHeartbeat(
+ appId,
+ 1,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ System.currentTimeMillis(),
+ gauges,
+ CLIENT_LABELS,
+ getNewReqeustId()));
+ }
+
+ /** Sum of the client gauge across apps as exported by the source, or null
if not exported. */
+ private static Long exportedGauge(ApplicationMetricsSource source) {
+ if (source.gauges().isEmpty()) {
+ return null;
+ }
+ return ((Number) source.gauges().head().gauge().getValue()).longValue();
+ }
+
+ private static void waitForAllMasters(ApplicationMetricsSource[] sources,
Long expected)
+ throws InterruptedException {
+ waitFor(
+ () -> Arrays.stream(sources).allMatch(s ->
Objects.equals(exportedGauge(s), expected)),
+ "all masters to export " + expected);
+ }
+
+ @Test
+ public void testClientMetricsReplicatedAcrossLeadershipChanges() throws
Exception {
+ ApplicationMetricsSource[] sources = new ApplicationMetricsSource[3];
+ try {
+ CelebornConf metricsConf = new CelebornConf();
+ metricsConf.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED().key(),
"true");
+ for (int i = 0; i < 3; i++) {
+ sources[i] = new ApplicationMetricsSource(metricsConf);
+ }
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false,
+ sources);
+ HARaftServer[] servers = {RATISSERVER1, RATISSERVER2, RATISSERVER3};
+ HAMasterMetaManager[] systems = {STATUSSYSTEM1, STATUSSYSTEM2,
STATUSSYSTEM3};
+
+ int a = RATISSERVER1.isLeader() ? 0 : (RATISSERVER2.isLeader() ? 1 : 2);
+ int b = (a + 1) % 3;
+
+ // A heartbeat reaching leader A is applied on every master.
+ heartbeatWithGauge(systems[a], APPID1, 7);
+ waitForAllMasters(sources, 7L);
+
+ // Leadership moves to B; B's heartbeats keep every master, including A,
up to date.
+ transferLeadership(servers[a], servers[b]);
+ heartbeatWithGauge(systems[b], APPID1, 5);
+ waitForAllMasters(sources, 5L);
+
+ // The app terminates on B: the replicated AppLost removes it from every
master.
+ submitWhenLeaderReady(() -> systems[b].handleAppLost(APPID1,
getNewReqeustId()));
+ waitForAllMasters(sources, null);
+
+ // A becomes leader again: a late heartbeat from the terminated app
stays ignored on every
+ // master, while live apps are tracked.
+ transferLeadership(servers[b], servers[a]);
+ heartbeatWithGauge(systems[a], APPID1, 9);
+ heartbeatWithGauge(systems[a], "appId2", 4);
+ waitForAllMasters(sources, 4L);
+ } finally {
+ for (ApplicationMetricsSource source : sources) {
+ if (source != null) {
+ source.destroy();
+ }
+ }
+ // Restart the shared cluster without the metrics sources.
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false);
+ }
+ }
+
+ @Test
+ public void testHeartbeatDelayedPastRetentionIsClearedByNextAppLost() throws
Exception {
+ ApplicationMetricsSource[] sources = new ApplicationMetricsSource[3];
+ try {
+ CelebornConf metricsConf = new CelebornConf();
+ metricsConf.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED().key(),
"true");
+
metricsConf.set(CelebornConf.MASTER_CLIENT_METRICS_REMOVED_APP_RETENTION().key(),
"500ms");
+ for (int i = 0; i < 3; i++) {
+ sources[i] = new ApplicationMetricsSource(metricsConf);
+ }
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false,
+ sources);
+ HAMasterMetaManager[] systems = {STATUSSYSTEM1, STATUSSYSTEM2,
STATUSSYSTEM3};
+ HAMasterMetaManager leader = pickLeaderStatusSystem();
+
+ heartbeatWithGauge(leader, APPID1, 7);
+ waitForAllMasters(sources, 7L);
+ submitWhenLeaderReady(() -> leader.handleAppLost(APPID1,
getNewReqeustId()));
+ waitForAllMasters(sources, null);
+
+ // Once the removed-app retention has passed, a heartbeat that was
delayed in flight is
+ // no longer rejected and re-creates the series on every master.
+ Thread.sleep(1_500L);
+ heartbeatWithGauge(leader, APPID1, 3);
+ waitForAllMasters(sources, 3L);
+
+ // It cannot leak: the same entry puts the app back into
appHeartbeatTime on every master,
+ // so the leader's timeout checker will lose it again, which clears it
everywhere.
+ Assert.assertTrue(
+ Arrays.stream(systems).allMatch(s ->
s.appHeartbeatTime.containsKey(APPID1)));
+ submitWhenLeaderReady(() -> leader.handleAppLost(APPID1,
getNewReqeustId()));
+ waitForAllMasters(sources, null);
+ } finally {
+ for (ApplicationMetricsSource source : sources) {
+ if (source != null) {
+ source.destroy();
+ }
+ }
+ // Restart the shared cluster without the metrics sources.
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false);
+ }
+ }
+
+ /** A metrics source that fails on every call, to exercise the apply-path
guardrails. */
+ private static class FailingMetricsSource extends ApplicationMetricsSource {
+ FailingMetricsSource(CelebornConf conf) {
+ super(conf);
+ }
+
+ @Override
+ public void updateApplicationMetrics(
+ String appId, Map<String, String> metricLabels, Map<String, Long>
gauges) {
+ throw new IllegalArgumentException("injected update failure");
+ }
+
+ @Override
+ public void removeApplicationMetrics(String appId) {
+ throw new IllegalStateException("injected remove failure");
+ }
+ }
+
+ @Test
+ public void testClientMetricsFailureDoesNotBreakRatisApply() throws
Exception {
+ ApplicationMetricsSource[] sources = new ApplicationMetricsSource[3];
+ try {
+ CelebornConf metricsConf = new CelebornConf();
+ metricsConf.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED().key(),
"true");
+ for (int i = 0; i < 3; i++) {
+ sources[i] = new FailingMetricsSource(metricsConf);
+ }
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false,
+ sources);
+ HAMasterMetaManager[] systems = {STATUSSYSTEM1, STATUSSYSTEM2,
STATUSSYSTEM3};
+ HAMasterMetaManager leader = pickLeaderStatusSystem();
+
+ // An exception escaping the apply would terminate every master on the
same entry.
+ heartbeatWithGauge(leader, APPID1, 7);
+ waitFor(
+ () -> Arrays.stream(systems).allMatch(s ->
s.appHeartbeatTime.containsKey(APPID1)),
+ "heartbeat to be applied on every master");
+
+ submitWhenLeaderReady(() -> leader.handleAppLost(APPID1,
getNewReqeustId()));
+ waitFor(
+ () -> Arrays.stream(systems).noneMatch(s ->
s.appHeartbeatTime.containsKey(APPID1)),
+ "AppLost to be applied on every master");
+
+ // The state machine keeps applying later entries.
+ heartbeatWithGauge(leader, "appId2", 4);
+ waitFor(
+ () -> Arrays.stream(systems).allMatch(s ->
s.appHeartbeatTime.containsKey("appId2")),
+ "later heartbeat to be applied on every master");
+ } finally {
+ for (ApplicationMetricsSource source : sources) {
+ if (source != null) {
+ source.destroy();
+ }
+ }
+ // Restart the shared cluster without the metrics sources.
+ resetRaftServer(
+ configureServerConf(new CelebornConf(), 1),
+ configureServerConf(new CelebornConf(), 2),
+ configureServerConf(new CelebornConf(), 3),
+ false);
+ }
+ }
+
@Test
public void testHandleUnRegisterShuffle() throws InterruptedException {
AbstractMetaManager statusSystem = pickLeaderStatusSystem();
@@ -993,7 +1269,17 @@ public class RatisMasterStatusSystemSuiteJ {
long dummy = 1235L;
statusSystem.handleAppHeartbeat(
- APPID1, 1, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
Thread.sleep(3000L);
Assert.assertEquals(Long.valueOf(dummy),
STATUSSYSTEM1.appHeartbeatTime.get(APPID1));
Assert.assertEquals(Long.valueOf(dummy),
STATUSSYSTEM2.appHeartbeatTime.get(APPID1));
@@ -1001,7 +1287,17 @@ public class RatisMasterStatusSystemSuiteJ {
String appId2 = "app02";
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
Thread.sleep(3000L);
Assert.assertEquals(Long.valueOf(dummy),
STATUSSYSTEM1.appHeartbeatTime.get(appId2));
@@ -1683,28 +1979,88 @@ public class RatisMasterStatusSystemSuiteJ {
Long dummy = 1235L;
statusSystem.handleAppHeartbeat(
- APPID1, 10000000000l, 1, 1, 1, new HashMap<>(), new HashMap<>(),
dummy, getNewReqeustId());
+ APPID1,
+ 10000000000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
String appId2 = "app02";
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Max size
statusSystem.handleUpdatePartitionSize();
Assert.assertEquals(statusSystem.estimatedPartitionSize,
conf.maxPartitionSizeToEstimate());
statusSystem.handleAppHeartbeat(
- APPID1, 1000000000l, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1000000000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
statusSystem.handleAppHeartbeat(
- appId2, 1, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Size between minEstimateSize -> maxEstimateSize
statusSystem.handleUpdatePartitionSize();
Assert.assertEquals(statusSystem.estimatedPartitionSize, 500000000);
statusSystem.handleAppHeartbeat(
- APPID1, 1000l, 1, 1, 1, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ APPID1,
+ 1000l,
+ 1,
+ 1,
+ 1,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
statusSystem.handleAppHeartbeat(
- appId2, 1000l, 1, 2, 2, new HashMap<>(), new HashMap<>(), dummy,
getNewReqeustId());
+ appId2,
+ 1000l,
+ 1,
+ 2,
+ 2,
+ new HashMap<>(),
+ new HashMap<>(),
+ dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ getNewReqeustId());
// Min size
statusSystem.handleUpdatePartitionSize();
@@ -2109,6 +2465,8 @@ public class RatisMasterStatusSystemSuiteJ {
shuffleFallbackCounts,
applicationFallbackCounts,
dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
getNewReqeustId());
shuffleFallbackCounts.put(POLICY1, 1L);
shuffleFallbackCounts.put(POLICY2, 2L);
@@ -2123,6 +2481,8 @@ public class RatisMasterStatusSystemSuiteJ {
shuffleFallbackCounts,
applicationFallbackCounts,
dummy,
+ Collections.emptyMap(),
+ Collections.emptyMap(),
getNewReqeustId());
assertEquals(statusSystem.shuffleTotalCount.longValue(), 5);
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSourceSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSourceSuite.scala
new file mode 100644
index 0000000000..f7a7c1d661
--- /dev/null
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/ApplicationMetricsSourceSuite.scala
@@ -0,0 +1,292 @@
+/*
+ * 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 java.util.concurrent.{CountDownLatch, Delayed,
ScheduledThreadPoolExecutor, TimeUnit}
+import java.util.concurrent.atomic.AtomicBoolean
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+
+class ApplicationMetricsSourceSuite extends CelebornFunSuite {
+
+ // Track every source created by a test so its metricsCleaner daemon thread
can be shut down
+ // in afterEach, instead of leaking a scheduled thread per source across the
suite.
+ private val createdSources = ArrayBuffer[ApplicationMetricsSource]()
+
+ private def newSource(conf: CelebornConf): ApplicationMetricsSource = {
+ val source = new ApplicationMetricsSource(conf)
+ createdSources += source
+ source
+ }
+
+ override def afterEach(): Unit = {
+ createdSources.foreach(_.destroy())
+ createdSources.clear()
+ super.afterEach()
+ }
+
+ /**
+ * A source whose first isAppRemoved call (made inside compute, before a new
series is
+ * published) blocks until the test releases it, so cleanup can be run
deterministically in
+ * the window between the removal check and the map insertion.
+ */
+ private class PausingSource(conf: CelebornConf) extends
ApplicationMetricsSource(conf) {
+ val paused = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ private val firstCall = new AtomicBoolean(true)
+
+ override protected def isAppRemoved(appId: String): Boolean = {
+ val removed = super.isAppRemoved(appId)
+ if (firstCall.compareAndSet(true, false)) {
+ paused.countDown()
+ assert(release.await(10, TimeUnit.SECONDS))
+ }
+ removed
+ }
+ }
+
+ /** Runs the first heartbeat of an absent series, firing `cleanup`
mid-insertion. */
+ private def raceFirstInsertion(
+ source: PausingSource,
+ labels: Map[String, String])(cleanup: => Unit): Unit = {
+ val heartbeat = new Thread(() => update(source, gaugeMetrics(7), labels))
+ heartbeat.start()
+ assert(source.paused.await(10, TimeUnit.SECONDS))
+ cleanup
+ source.release.countDown()
+ heartbeat.join(10000)
+ assert(!heartbeat.isAlive)
+ }
+
+ private def enabledConf(): CelebornConf = {
+ val c = new CelebornConf()
+ c.set(CelebornConf.MASTER_CLIENT_METRICS_ENABLED, true)
+ c
+ }
+
+ private def scheduledTaskCount(source: ApplicationMetricsSource): Int =
+
source.metricsCleaner.asInstanceOf[ScheduledThreadPoolExecutor].getQueue.size()
+
+ private def gaugeMetrics(value: Long): JHashMap[String, java.lang.Long] = {
+ val map = new JHashMap[String, java.lang.Long]()
+ map.put("ClientActiveShuffleCount", value)
+ map
+ }
+
+ private def update(
+ source: ApplicationMetricsSource,
+ metrics: JHashMap[String, java.lang.Long],
+ labels: Map[String, String] = Map.empty,
+ appId: String = "app-1"): Unit =
+ source.updateApplicationMetrics(appId, labels.asJava, metrics)
+
+ private def gaugeValue(
+ source: ApplicationMetricsSource,
+ labels: Map[String, String],
+ name: String = "ClientActiveShuffleCount"): Option[Long] =
+ source.gauges()
+ .find(g => g.name == name && hasLabels(g.labels, labels))
+ .map(_.gauge.getValue.asInstanceOf[Number].longValue())
+
+ private def hasLabels(
+ actual: Map[String, String],
+ expected: Map[String, String]): Boolean =
+ expected.forall { case (key, value) => actual.get(key).contains(value) }
+
+ test("masterClientMetrics disabled: removed app cleaner is not scheduled") {
+ val source = newSource(new CelebornConf())
+
+ assert(scheduledTaskCount(source) == 0)
+ }
+
+ test("removed app cleaner uses configured retention as schedule interval") {
+ val conf = enabledConf()
+ conf.set(CelebornConf.MASTER_CLIENT_METRICS_REMOVED_APP_RETENTION.key,
"2s")
+ val retentionMs = conf.masterClientMetricsRemovedAppRetentionMs
+ assert(retentionMs == 2000L)
+ val source = newSource(conf)
+
+ val scheduledTasks = source.metricsCleaner
+ .asInstanceOf[ScheduledThreadPoolExecutor]
+ .getQueue
+ assert(scheduledTasks.size() == 1)
+ val delayMs =
scheduledTasks.peek().asInstanceOf[Delayed].getDelay(TimeUnit.MILLISECONDS)
+ assert(delayMs > 0)
+ assert(delayMs <= retentionMs)
+ }
+
+ test("client labels are used as metric labels") {
+ val source = newSource(enabledConf())
+
+ update(source, gaugeMetrics(5), Map("team" -> "data-eng"))
+
+ val metrics = source.getMetrics
+ assert(metrics.contains("""team="data-eng""""))
+ }
+
+ test("gauge is updated to the latest reported value") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(1), labels)
+ update(source, gaugeMetrics(42), labels)
+
+ assert(gaugeValue(source, labels).contains(42L))
+ }
+
+ test("gauge for a label set aggregates values across apps by sum") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(3), labels, "app-1")
+ update(source, gaugeMetrics(7), labels, "app-2")
+
+ assert(gaugeValue(source, labels).contains(10L))
+ }
+
+ test("masterClientMetrics disabled: replicated client metrics are ignored") {
+ val source = newSource(new CelebornConf())
+
+ update(source, gaugeMetrics(3), Map("team" -> "data-eng"))
+
+ assert(source.gauges().isEmpty)
+ }
+
+ test("removing one app updates gauge sum while another app still
contributes") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(3), labels, "app-1")
+ update(source, gaugeMetrics(7), labels, "app-2")
+
+ source.removeApplicationMetrics("app-1")
+
+ assert(gaugeValue(source, labels).contains(7L))
+ }
+
+ test("removing the last contributing app deregisters tracked metrics") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(3), labels, "app-1")
+ update(source, gaugeMetrics(7), labels, "app-2")
+
+ source.removeApplicationMetrics("app-1")
+
+ assert(gaugeValue(source, labels).contains(7L))
+
+ source.removeApplicationMetrics("app-2")
+
+ assert(gaugeValue(source, labels).isEmpty)
+ }
+
+ test("late heartbeats for removed apps are ignored") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(3), labels, "app-1")
+ source.removeApplicationMetrics("app-1")
+ update(source, gaugeMetrics(9), labels, "app-1")
+
+ assert(gaugeValue(source, labels).isEmpty)
+ }
+
+ test("labels with invalid Prometheus key names are rejected") {
+ val source = newSource(enabledConf())
+
+ update(source, gaugeMetrics(5), Map("invalid-key" -> "ok"))
+ assert(source.gauges().isEmpty)
+
+ update(source, gaugeMetrics(5), Map("123start" -> "ok"))
+ assert(source.gauges().isEmpty)
+
+ update(source, gaugeMetrics(5), Map("valid_key" -> "ok"))
+ assert(source.gauges().nonEmpty)
+ }
+
+ test("labels with unsafe values (quotes, backslashes, newlines) are
rejected") {
+ val source = newSource(enabledConf())
+
+ update(source, gaugeMetrics(5), Map("team" -> """val"ue"""))
+ assert(source.gauges().isEmpty)
+
+ update(source, gaugeMetrics(5), Map("team" -> "val\\ue"))
+ assert(source.gauges().isEmpty)
+
+ update(source, gaugeMetrics(5), Map("team" -> "val\nue"))
+ assert(source.gauges().isEmpty)
+
+ update(source, gaugeMetrics(5), Map("team" -> "safe_value"))
+ assert(source.gauges().nonEmpty)
+ }
+
+ test("gauge values from multiple apps are summed, not last-writer-wins") {
+ val source = newSource(enabledConf())
+ val labels = Map("team" -> "data-eng")
+
+ update(source, gaugeMetrics(3), labels, "app-1")
+ update(source, gaugeMetrics(7), labels, "app-2")
+
+ assert(gaugeValue(source, labels).contains(10L))
+
+ update(source, gaugeMetrics(5), labels, "app-1")
+ assert(gaugeValue(source, labels).contains(12L))
+ }
+
+ test("clearApplicationMetrics drops all series but keeps removed apps
ignored") {
+ val source = newSource(enabledConf())
+ val engLabels = Map("team" -> "data-eng")
+ val mlLabels = Map("team" -> "ml")
+ update(source, gaugeMetrics(3), engLabels, "app-1")
+ update(source, gaugeMetrics(4), engLabels, "app-2")
+ update(source, gaugeMetrics(5), mlLabels, "app-3")
+ source.removeApplicationMetrics("app-3")
+ assert(source.gauges().size == 1)
+
+ source.clearApplicationMetrics()
+
+ assert(source.gauges().isEmpty)
+ assert(!source.gaugeExists("ClientActiveShuffleCount", engLabels))
+
+ // Live apps repopulate from their next heartbeat, starting from a clean
sum.
+ update(source, gaugeMetrics(6), engLabels, "app-1")
+ assert(gaugeValue(source, engLabels).contains(6L))
+
+ // An app removed before the clear stays ignored.
+ update(source, gaugeMetrics(5), mlLabels, "app-3")
+ assert(gaugeValue(source, mlLabels).isEmpty)
+ }
+
+ test("removal racing the first insertion of an absent series leaves no
gauge") {
+ val source = new PausingSource(enabledConf())
+ createdSources += source
+ val labels = Map("team" -> "data-eng")
+
+ raceFirstInsertion(source, labels) {
+ source.removeApplicationMetrics("app-1")
+ }
+
+ assert(gaugeValue(source, labels).isEmpty)
+ assert(source.gauges().isEmpty)
+ }
+}
