[CELEBORN-2357] Add Metrics Prefix

common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

@@ -911,6 +911,7 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
   def metricsConf: Option[String] = get(METRICS_CONF)
   def metricsSystemEnable: Boolean = get(METRICS_ENABLED)
   def metricsSampleRate: Double = get(METRICS_SAMPLE_RATE)
+  def metricsPrefix: String = get(METRICS_PREFIX)
   def metricsSlidingWindowSize: Int = get(METRICS_SLIDING_WINDOW_SIZE)
   def metricsCollectCriticalEnabled: Boolean = get(METRICS_COLLECT_CRITICAL_ENABLED)
   def metricsCapacity: Int = get(METRICS_CAPACITY)
@@ -5901,6 +5902,14 @@ object CelebornConf extends Logging {
       .stringConf
       .createOptional
 
+  val METRICS_PREFIX: ConfigEntry[String] =
+    buildConf("celeborn.metrics.prefix")
+      .categories("metrics")
+      .doc("Prefix metrics with this value.")
+      .version("0.7.0")
+      .stringConf
+      .createWithDefault("metrics")
+
   val METRICS_ENABLED: ConfigEntry[Boolean] =
     buildConf("celeborn.metrics.enabled")
       .categories("metrics")

common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala

@@ -60,6 +60,8 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
 
   val metricsCapacity: Int = conf.metricsCapacity
 
+  val metricsPrefix: String = conf.metricsPrefix
+
   val timerSupplier = new TimerSupplier(metricsSlidingWindowSize)
 
   val histogramSupplier = new HistogramSupplier(metricsSlidingWindowSize)
@@ -701,7 +703,7 @@ abstract class AbstractSource(conf: CelebornConf, role: String)
   }
 
   protected def normalizeKey(key: String): String = {
-    s"metrics_${key.replaceAll("[^a-zA-Z0-9]", "_")}_"
+    s"${metricsPrefix}_${key.replaceAll("[^a-zA-Z0-9]", "_")}_"
   }
 
   def reportNanosAsMills(value: Double): Double = {

common/src/test/scala/org/apache/celeborn/common/metrics/source/CelebornSourceSuite.scala

@@ -38,6 +38,24 @@ class CelebornSourceSuite extends CelebornFunSuite {
     assert(res.contains("metrics_abc_Count"))
   }
 
+  test("test customized metrics prefix") {
+    val conf = new CelebornConf()
+    conf.set(CelebornConf.METRICS_PREFIX.key, "celeborn")
+
+    val mockSource = new AbstractSource(conf, Role.WORKER) {
+      override def sourceName: String = "mockSource"
+    }
+    val histogram = "abc"
+    mockSource.addHistogram(histogram)
+    for (i <- 1 to 100) {
+      mockSource.updateHistogram(histogram, 10)
+    }
+    val res = mockSource.getMetrics
+
+    assert(res.contains("celeborn_abc_Count"))
+    assert(!res.contains("metrics_abc_Count"))
+  }
+
   test("test getMetrics with customized label") {
     val conf = new CelebornConf()
     createAbstractSourceAndCheck(conf, "", Role.MASTER)

docs/configuration/metrics.md

@@ -28,6 +28,7 @@ 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.prefix | metrics | false | Prefix metrics with this value. | 0.7.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 |  | 
 | celeborn.metrics.timer.slidingWindow.size | 4096 | false | The sliding window size of timer metric. | 0.2.0 |  |