Skip to content

[CELEBORN-2469] Add worker stats collection and reporting for autoscaling - #3858

Open
Kalvin2077 wants to merge 5 commits into
apache:mainfrom
Kalvin2077:cip13/worker-metric
Open

Kalvin2077 wants to merge 5 commits into
apache:mainfrom
Kalvin2077:cip13/worker-metric

Conversation

@Kalvin2077

@Kalvin2077 Kalvin2077 commented Sep 24, 2026 •

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

Implement the Worker metrics portion of CIP-13:

  • Add a configurable background collector for memory, system load and local disk metrics.
  • Report its cached snapshot through the existing Worker heartbeat using map<string, double> stats = 3 in PbWorkerStatus.
  • Retain the latest snapshot per registered Worker in a local Master cache, with registration and Worker-loss cleanup. Metrics remain outside lifecycle metadata and Raft state.
  • Document collector configuration and extension behavior.

The numeric map is an explicit difference from CIP-13’s proposed string map and remains part of the design discussion.

Why are the changes needed?

Future scaling policies need Worker resource observations. This PR supplies the collection and reporting path; scaling decisions and platform operations are follow-up work.

Does this PR resolve a correctness bug?

  • Yes

Does this PR introduce any user-facing change?

  • Yes

Adds celeborn.scale.metricCollectorClassName, celeborn.scale.metricInterval (default 5s) and celeborn.scale.metricWindowSize (default 6).

How was this patch tested?

21 selected tests passed on JDK 17, covering protobuf compatibility, collection, cache cleanup, failure isolation and Worker-to-Master RPCs. Module builds, Spotless and Apache Rat passed.

Performance impact

A local TCP benchmark compared heartbeats without metrics against the four default metrics. Each case ran twice for 10 seconds after warmup, at workerCount / 30 heartbeats/s. All 14,672 measured responses in this comparison succeeded.

Logical Workers Change in mean run p99 Change in whole-JVM heap after setup GC
1,000 +0.192 ms (+2.96%) +1.00 MiB
10,000 +0.0095 ms (+0.49%) +9.62 MiB

The four metrics add 128 protobuf bytes per heartbeat, approximately 42.67 kB/s at 10,000 Workers with a 30-second period, excluding transport framing. Heartbeats read cached measurements; collection adds no extra RPC.

These are observed differences between two-run averages. CPU ranges overlapped. Client and non-HA Master shared one JVM, and the short runs did not cover a full heartbeat cycle or measure collector CPU. They do not establish physical-cluster capacity; representative cross-host and HA validation is still needed.

Possible follow-up PRs

  • Network utilization collection, including throughput counters and bandwidth configuration.
  • Deliver Master-cached observations to ScaleManager through a registerWorkerStats callback.
  • Scaling policies, persisted scale operations, and Worker decommission/recommission coordination.
  • Platform adapters, such as Kubernetes and OpenKruise.

@SteNicholas

SteNicholas commented Sep 24, 2026 •

Copy link
Copy Markdown
Member

@Kalvin2077, I do not agree with the current implementation. Have you verified the heartbeat load caused by adding worker metrics in a large-scale celeborn cluster? Meanwhile, the details of this FLIP have not yet been fully aligned.

@Kalvin2077
Kalvin2077 marked this pull request as draft September 24, 2026 08:55
@Kalvin2077

Copy link
Copy Markdown
Contributor Author

@Kalvin2077, I do not agree with the current implementation. Have you verified the heartbeat load caused by adding worker metrics in a large-scale celeborn cluster? Meanwhile, the details of this FLIP have not yet been fully aligned.

@SteNicholas

Thanks for the feedback. You're right.

The current implementation introduces a typed WorkerStats schema, while CIP-13 still proposes a simple stats map inside WorkerStatus. I should have aligned the schema on the CIP thread before implementing it.

To make sure I address the right risk: is the concern mainly the payload size of the typed schema, or the broader master-side and aggregate heartbeat cost at scale?

@Kalvin2077
Kalvin2077 force-pushed the cip13/worker-metric branch from 16fc232 to b3079e6 Compare October 9, 2026 12:19
Kalvin2077 and others added 3 commits October 10, 2026 14:51
Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6.1-sol
AI-Contributed/Feature: 114/114
AI-Contributed/UT: 302/302
Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6.1-sol
AI-Contributed/Feature: 228/228
AI-Contributed/UT: 354/354
Co-Authored-By: Codex <noreply@openai.com>
AI-Model: gpt-6.1-sol
AI-Contributed/Feature: 3/3
AI-Contributed/UT: 0/0
@Kalvin2077
Kalvin2077 marked this pull request as ready for review October 10, 2026 08:46
@Kalvin2077

Copy link
Copy Markdown
Contributor Author

@zaynt4606

val disks = worker.storageManager.localDisksSnapshot()
val usableDisks = disks.filter(_.status != DiskStatus.IO_HANG)
val usableBytes = usableDisks.map(_.actualUsableSpace.toDouble).sum
val totalBytes = disks.map(_.totalSpace.toDouble).sum

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could DiskUsedRatio use Celeborn's effective allocatable capacity as the denominator? StorageManager caps actualUsableSpace by configuredUsableSpace, but sets totalSpace to the filesystem's physical total. On an otherwise empty 1 TiB disk configured with capacity=100Gi, this reports about 90% used even though Celeborn has used none of its configured capacity. This would give autoscaling an incorrect disk pressure signal. Please use a consistent capacity basis for the numerator and denominator, preserve the physical free space and reserve limit, and add a test where capacity= is smaller than the physical disk.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants