This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new 9b392d8a02 [CELEBORN-2359] Deduplicate # HELP/# TYPE lines in
PrometheusServlet output
9b392d8a02 is described below
commit 9b392d8a02c372b0fb9d90f59f26642c62048834
Author: afterincomparableyum
<[email protected]>
AuthorDate: Mon Oct 12 19:02:05 2026 +0800
[CELEBORN-2359] Deduplicate # HELP/# TYPE lines in PrometheusServlet output
### What changes were proposed in this pull request?
AbstractSource emits a `# HELP` and `# TYPE` line for every metric sample.
When a metric family has multiple label sets — like ActiveConnectionCount (per
application), RequestSlotsFailedCount (per status code), per-user/app resource
consumption, or FlushWorkingQueueSize (per mountpoint) — the same family is
exported with repeated `# HELP` and `# TYPE` lines. This violates the
Prometheus exposition format.
The fix is to override PrometheusServlet.getMetricsSnapshot to keep only
the first `# HELP` and `# TYPE` line per metric family while preserving all
sample lines, producing valid Prometheus output.
I also added a PrometheusServletSuite covering a multi-label-set counter
and a distinct gauge.
### Why are the changes needed?
To dedup the help and type lines to no longer violate prometheus exposition
format.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
CI/CD
Closes #3852 from afterincomparableyum/CELEBORN-2359.
Authored-by: afterincomparableyum
<[email protected]>
Signed-off-by: zhengtao <[email protected]>
Co-Authored-By: Codex <[email protected]>
(cherry picked from commit 63381a8593b6402965ffbd9dde7058bfc04e027b)
Signed-off-by: zhengtao <[email protected]>
AI-Model: gpt-6
AI-Contributed/Feature: 0/46
AI-Contributed/UT: 0/106
---
.../common/metrics/sink/PrometheusServlet.scala | 46 +++++++++
.../metrics/sink/PrometheusServletSuite.scala | 106 +++++++++++++++++++++
2 files changed, 152 insertions(+)
diff --git
a/service/src/main/scala/org/apache/celeborn/common/metrics/sink/PrometheusServlet.scala
b/service/src/main/scala/org/apache/celeborn/common/metrics/sink/PrometheusServlet.scala
index 27797c8fc9..5ce54db97d 100644
---
a/service/src/main/scala/org/apache/celeborn/common/metrics/sink/PrometheusServlet.scala
+++
b/service/src/main/scala/org/apache/celeborn/common/metrics/sink/PrometheusServlet.scala
@@ -19,9 +19,12 @@ package org.apache.celeborn.common.metrics.sink
import java.util.Properties
+import scala.collection.mutable
+
import com.codahale.metrics.MetricRegistry
import org.eclipse.jetty.servlet.ServletContextHandler
+import org.apache.celeborn.common.metrics.sink.PrometheusServlet.{HeaderLine,
MetricFamily}
import org.apache.celeborn.common.metrics.source.Source
import org.apache.celeborn.server.common.http.HttpUtils
import org.apache.celeborn.server.common.http.HttpUtils.ServletParams
@@ -37,4 +40,47 @@ class PrometheusServlet(
servletPath,
new ServletParams(_ => getMetricsSnapshot, "text/plain"))
}
+
+ /**
+ * Sources emit a # HELP / # TYPE pair before every sample and iterate
metrics in hash order,
+ * so samples of one metric family can be repeated and interleaved with
other families. The
+ * Prometheus exposition format requires a single # HELP / # TYPE per family
with all of its
+ * samples grouped together, so regroup the output by metric family.
+ */
+ override def getMetricsSnapshot: String = {
+ val raw = super.getMetricsSnapshot
+ val families = mutable.LinkedHashMap[String, MetricFamily]()
+
+ raw.linesIterator.foreach {
+ case line @ HeaderLine(kind, metric) =>
+ val family = families.getOrElseUpdate(metric, new MetricFamily)
+ if (kind == "HELP" && family.help == null) {
+ family.help = line
+ } else if (kind == "TYPE" && family.metricType == null) {
+ family.metricType = line
+ }
+ case line if line.nonEmpty =>
+ val metric = line.takeWhile(c => c != '{' && !c.isWhitespace)
+ families.getOrElseUpdate(metric, new MetricFamily).samples += line
+ case _ =>
+ }
+
+ val out = new StringBuilder(raw.length)
+ families.valuesIterator.foreach { family =>
+ if (family.help != null) out.append(family.help).append('\n')
+ if (family.metricType != null) out.append(family.metricType).append('\n')
+ family.samples.foreach(out.append(_).append('\n'))
+ }
+ out.result()
+ }
+}
+
+object PrometheusServlet {
+ private val HeaderLine = "^#\\s+(TYPE|HELP)\\s+(\\S+)\\b.*$".r
+
+ private class MetricFamily {
+ var help: String = _
+ var metricType: String = _
+ val samples = mutable.ArrayBuffer[String]()
+ }
}
diff --git
a/service/src/test/scala/org/apache/celeborn/common/metrics/sink/PrometheusServletSuite.scala
b/service/src/test/scala/org/apache/celeborn/common/metrics/sink/PrometheusServletSuite.scala
new file mode 100644
index 0000000000..dec14b2108
--- /dev/null
+++
b/service/src/test/scala/org/apache/celeborn/common/metrics/sink/PrometheusServletSuite.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.metrics.sink
+
+import java.util.Properties
+
+import com.codahale.metrics.MetricRegistry
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.metrics.source.{AbstractSource, Role}
+
+class PrometheusServletSuite extends CelebornFunSuite {
+
+ private def newServlet(sources: AbstractSource*): PrometheusServlet = {
+ new PrometheusServlet(
+ new Properties(),
+ new MetricRegistry(),
+ sources,
+ "/metrics/prometheus")
+ }
+
+ private def newSource(conf: CelebornConf): AbstractSource = {
+ new AbstractSource(conf, Role.WORKER) {
+ override def sourceName: String = "mockSource"
+ }
+ }
+
+ test("getMetricsSnapshot deduplicates # HELP and # TYPE lines per metric
family") {
+ val conf = new CelebornConf()
+ val source = newSource(conf)
+
+ val user1 = Map("user" -> "user1")
+ val user2 = Map("user" -> "user2")
+ source.addCounter("Counter", user1)
+ source.addCounter("Counter", user2)
+ source.incCounter("Counter", 1, user1)
+ source.incCounter("Counter", 2, user2)
+
+ source.addGauge("Gauge1") { () => 1000 }
+
+ val snapshot = newServlet(source).getMetricsSnapshot
+ val lines = snapshot.linesIterator.toList
+
+ val counterName = "metrics_Counter_Count"
+ val gaugeName = "metrics_Gauge1_Value"
+
+ assert(lines.count(_ == s"# HELP $counterName") == 1)
+ assert(lines.count(_ == s"# TYPE $counterName counter") == 1)
+ assert(lines.count(_.startsWith(s"$counterName{")) == 2)
+ assert(snapshot.contains("""user="user1""""))
+ assert(snapshot.contains("""user="user2""""))
+ assert(lines.count(_ == s"# HELP $gaugeName") == 1)
+ assert(lines.count(_ == s"# TYPE $gaugeName gauge") == 1)
+ val gaugeSamples = lines.filter(_.startsWith(gaugeName))
+ assert(gaugeSamples.size == 1)
+ assert(gaugeSamples.head.split(" ")(1) == "1000")
+ }
+
+ test("getMetricsSnapshot groups samples of a metric family that are
interleaved") {
+ val conf = new CelebornConf()
+ val user1 = Map("user" -> "user1")
+ val user2 = Map("user" -> "user2")
+ // Sources emit gauges before counters, so the raw snapshot is
+ // Counter{user1}, Gauge1, Counter{user2}.
+ val source1 = newSource(conf)
+ source1.addCounter("Counter", user1)
+ val source2 = newSource(conf)
+ source2.addGauge("Gauge1") { () => 1000 }
+ source2.addCounter("Counter", user2)
+
+ val servlet = newServlet(source1, source2)
+ val counterName = "metrics_Counter_Count"
+ val gaugeName = "metrics_Gauge1_Value"
+ val rawLines =
servlet.sources.map(_.getMetrics).mkString.linesIterator.toList
+ assert(rawLines.indexWhere(_.startsWith(gaugeName)) <
+ rawLines.lastIndexWhere(_.startsWith(s"$counterName{")))
+
+ val lines = servlet.getMetricsSnapshot.linesIterator.toList
+ val counterStart = lines.indexOf(s"# HELP $counterName")
+ assert(lines(counterStart + 1) == s"# TYPE $counterName counter")
+ val counterSamples = lines.slice(counterStart + 2, counterStart + 4)
+ assert(counterSamples.forall(_.startsWith(s"$counterName{")))
+ assert(counterSamples.exists(_.contains("user=\"user1\"")))
+ assert(counterSamples.exists(_.contains("user=\"user2\"")))
+ assert(lines.count(_.startsWith(s"$counterName{")) == 2)
+ val gaugeStart = lines.indexOf(s"# HELP $gaugeName")
+ assert(lines(gaugeStart + 1) == s"# TYPE $gaugeName gauge")
+ assert(lines(gaugeStart + 2).startsWith(s"$gaugeName{"))
+ }
+}
