[CELEBORN-194] Introduce client side metrics for celeborn

assets/grafana/celeborn-dashboard.json

@@ -18057,6 +18057,296 @@
       ],
       "title": "Rpc",
       "type": "row"
+    },
+    {
+      "collapsed": true,
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 111
+      },
+      "id": 273,
+      "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": 274,
+          "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": 275,
+          "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": 276,
+          "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,

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
@@ -85,7 +104,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.")

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"
+}