This is an automated email from the ASF dual-hosted git repository.
zaynt4606 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 cbcbeeb8ed [CELEBORN-2403] Add worker metrics for disk status and
unhealthy disk count
cbcbeeb8ed is described below
commit cbcbeeb8ed4a4ccf770028cd912ac4ed90cb73f6
Author: yew1eb <[email protected]>
AuthorDate: Tue Oct 13 18:10:40 2026 +0800
[CELEBORN-2403] Add worker metrics for disk status and unhealthy disk count
### What changes were proposed in this pull request?
Add two gauges on the worker to expose disk health status, registered in
`LocalDeviceMonitor.init()` and evaluated lazily from the live
`DiskInfo.status` at scrape time:
1. `UnhealthyDiskCount` — the number of local disks whose status is not
`HEALTHY`, intended for alerting (e.g. `> 0`).
2. `DiskStatus` with a `mountpoint` label — the current `DiskStatus` value
(0: HEALTHY, 1: READ_OR_WRITE_FAILURE, 2: IO_HANG, 3: HIGH_DISK_USAGE, 4:
CRITICAL_ERROR) of each disk, for locating the exact unhealthy disk.
Also included: two corresponding panels in the Grafana dashboard
(`assets/grafana/celeborn-dashboard.json`) and the new metrics in
`docs/monitoring.md`.
### Why are the changes needed?
`DeviceMonitor` already maintains `DiskInfo.status` for every local disk,
but none of the existing metrics expose it —
`DeviceOSFreeBytes`/`DeviceCelebornFreeBytes` report capacity, and
`Device_<name>_<Status>_Count` only counts non-critical error events. To notice
a broken disk, operators have to grep worker logs, which makes alerting on disk
failures impossible.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
Two new worker metrics are exposed: `UnhealthyDiskCount` and `DiskStatus`
(with a `mountpoint` label).
### How was this patch tested?
- Added a new test case `monitor disk status metrics` in
`DeviceMonitorSuite` covering the multi-disk scenario: all disks healthy
initially, then one disk marked `READ_OR_WRITE_FAILURE` (aggregate count
becomes 1 and only that disk's gauge changes), then recovered to `HEALTHY`.
- `./build/mvn test -pl worker` passes.
- `python3 dev/lint_grafana.py assets/grafana/celeborn-dashboard.json`
passes.
- `./build/mvn spotless:check -pl worker` passes.
Closes #3782 from yew1eb/worker-disk-status-metrics.
Authored-by: yew1eb <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/213
AI-Contributed/UT: 0/45
---
assets/grafana/celeborn-dashboard.json | 201 ++++++++++++++++++++-
docs/monitoring.md | 2 +
.../service/deploy/worker/WorkerSource.scala | 2 +
.../deploy/worker/storage/DeviceMonitor.scala | 8 +
.../deploy/worker/storage/DeviceMonitorSuite.scala | 45 +++++
5 files changed, 257 insertions(+), 1 deletion(-)
diff --git a/assets/grafana/celeborn-dashboard.json
b/assets/grafana/celeborn-dashboard.json
index fbbc1a05b5..a3c1ae80fd 100644
--- a/assets/grafana/celeborn-dashboard.json
+++ b/assets/grafana/celeborn-dashboard.json
@@ -16923,6 +16923,205 @@
"title": "metrics_DeviceCelebornFreeBytes_Value",
"type": "timeseries"
},
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "",
+ "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"
+ }
+ ]
+ },
+ "unit": "none"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 0,
+ "y": 1161
+ },
+ "id": 273,
+ "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_UnhealthyDiskCount_Value{instance=~\"${instance}\"}",
+ "legendFormat": "${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_UnhealthyDiskCount_Value",
+ "type": "timeseries"
+ },
+ {
+ "datasource": {
+ "type": "prometheus",
+ "uid": "${DS_PROMETHEUS}"
+ },
+ "description": "",
+ "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": [
+ {
+ "options": {
+ "0": {
+ "text": "HEALTHY"
+ },
+ "1": {
+ "text": "READ_OR_WRITE_FAILURE"
+ },
+ "2": {
+ "text": "IO_HANG"
+ },
+ "3": {
+ "text": "HIGH_DISK_USAGE"
+ },
+ "4": {
+ "text": "CRITICAL_ERROR"
+ }
+ },
+ "type": "value"
+ }
+ ],
+ "thresholds": {
+ "mode": "absolute",
+ "steps": [
+ {
+ "color": "green"
+ }
+ ]
+ },
+ "unit": "none"
+ },
+ "overrides": []
+ },
+ "gridPos": {
+ "h": 8,
+ "w": 12,
+ "x": 12,
+ "y": 1161
+ },
+ "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_DiskStatus_Value{instance=~\"${instance}\"}",
+ "legendFormat": "{{mountpoint}} ${baseLegend}",
+ "range": true,
+ "refId": "A"
+ }
+ ],
+ "title": "metrics_DiskStatus_Value",
+ "type": "timeseries"
+ },
{
"datasource": {
"type": "prometheus",
@@ -16985,7 +17184,7 @@
"h": 8,
"w": 12,
"x": 0,
- "y": 1161
+ "y": 1169
},
"id": 127,
"options": {
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 58c4d7ad32..bbdb90e3fd 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -261,6 +261,8 @@ These metrics are exposed by Celeborn worker.
| DeviceOSTotalBytes | The total usable space of OS
for device monitor.
|
| DeviceCelebornFreeBytes | The actual usable space of
Celeborn for device.
|
| DeviceCelebornTotalBytes | The total space of Celeborn for
device.
|
+ | DiskStatus | The status of the disk on the
worker. 0:HEALTHY 1:READ_OR_WRITE_FAILURE 2:IO_HANG 3:HIGH_DISK_USAGE
4:CRITICAL_ERROR. Label: mountpoint |
+ | UnhealthyDiskCount | The number of disks whose
status is not HEALTHY on the worker.
|
| PotentialConsumeSpeed | The speed of potential
consumption for congestion control.
|
| UserProduceSpeed | The speed of user production
for congestion control.
|
| WorkerConsumeSpeed | The speed of worker consumption
for congestion control.
|
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
index f938580514..20c7cf6426 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala
@@ -278,6 +278,8 @@ object WorkerSource {
val DEVICE_CELEBORN_FREE_CAPACITY = "DeviceCelebornFreeBytes"
val DEVICE_CELEBORN_TOTAL_CAPACITY = "DeviceCelebornTotalBytes"
val PARTITION_FILE_SIZE = "PartitionFileSizeBytes"
+ val DISK_STATUS = "DiskStatus"
+ val UNHEALTHY_DISK_COUNT = "UnhealthyDiskCount"
// congestion control
val POTENTIAL_CONSUME_SPEED = "PotentialConsumeSpeed"
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala
index 01af7468a8..946c9c3775 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitor.scala
@@ -107,6 +107,14 @@ class LocalDeviceMonitor(
diskInfos.map(_.actualUsableSpace).sum
}
}
+ workerSource.addGauge(WorkerSource.UNHEALTHY_DISK_COUNT) { () =>
+ diskInfos.values().asScala.count(_.status != DiskStatus.HEALTHY)
+ }
+ diskInfos.asScala.foreach { case (mountPoint, diskInfo) =>
+ workerSource.addGauge(WorkerSource.DISK_STATUS, Map("mountpoint" ->
mountPoint)) { () =>
+ diskInfo.status.getValue
+ }
+ }
}
override def startCheck(): Unit = {
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitorSuite.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitorSuite.scala
index 27dbadd78e..5a84ea99a6 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitorSuite.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/DeviceMonitorSuite.scala
@@ -468,4 +468,49 @@ class DeviceMonitorSuite extends AnyFunSuite {
assertEquals(1264867868672L, metrics2.head.gauge.getValue)
}
}
+
+ test("monitor disk status metrics") {
+
withObjectMocked[org.apache.celeborn.service.deploy.worker.storage.DeviceMonitor.type]
{
+
when(DeviceMonitor.getDiskUsageInfos(diskInfos2.get("/mnt/disk1"))).thenReturn(
+ dfBOut1DiskUsageInfo)
+
when(DeviceMonitor.getDiskUsageInfos(diskInfos2.get("/mnt/disk2"))).thenReturn(
+ dfBOut2DiskUsageInfo)
+
when(DeviceMonitor.getDiskUsageInfos(diskInfos2.get("/mnt/disk3"))).thenReturn(
+ dfBOut3DiskUsageInfo)
+
when(DeviceMonitor.getDiskUsageInfos(diskInfos2.get("/mnt/disk4"))).thenReturn(
+ dfBOut4DiskUsageInfo)
+
when(DeviceMonitor.getDiskUsageInfos(diskInfos2.get("/mnt/disk5"))).thenReturn(
+ dfBOut5DiskUsageInfo)
+
+ deviceMonitor2.init()
+
+ val unhealthyDiskCountMetrics =
+ workerSource2.gauges().filter(_.name ==
WorkerSource.UNHEALTHY_DISK_COUNT)
+ assertEquals(1, unhealthyDiskCountMetrics.size)
+ assertEquals(0, unhealthyDiskCountMetrics.head.gauge.getValue)
+
+ val diskStatusMetrics = workerSource2.gauges()
+ .filter(_.name == WorkerSource.DISK_STATUS)
+ .sortBy(_.labels("mountpoint"))
+ assertEquals(5, diskStatusMetrics.size)
+ diskStatusMetrics.foreach { m =>
+ assertEquals(DiskStatus.HEALTHY.getValue, m.gauge.getValue)
+ }
+
+ // mark one disk unhealthy, only its own gauge and the aggregate count
should change
+ diskInfos2.get("/mnt/disk3").setStatus(DiskStatus.READ_OR_WRITE_FAILURE)
+ assertEquals(1, unhealthyDiskCountMetrics.head.gauge.getValue)
+ diskStatusMetrics.foreach { m =>
+ if (m.labels("mountpoint") == "/mnt/disk3") {
+ assertEquals(DiskStatus.READ_OR_WRITE_FAILURE.getValue,
m.gauge.getValue)
+ } else {
+ assertEquals(DiskStatus.HEALTHY.getValue, m.gauge.getValue)
+ }
+ }
+
+ // recover, the aggregate count should drop back to 0
+ diskInfos2.get("/mnt/disk3").setStatus(DiskStatus.HEALTHY)
+ assertEquals(0, unhealthyDiskCountMetrics.head.gauge.getValue)
+ }
+ }
}
