[CELEBORN-2389] Track S3 resource consumption in per-app/per-user metrics

[shellfish007]

What changes were proposed in this pull request?

Adds first-class S3 accounting to ResourceConsumption, mirroring the existing disk/hdfs pattern:

• PbResourceConsumption — new fields s3BytesWritten = 6, s3FileCount = 7 (backward compatible; older peers default to 0).
• ResourceConsumption — new s3BytesWritten/s3FileCount fields, defaulted to 0 so all existing construction sites compile unchanged; included in add/subtract/toString/simpleString.
• StorageManager.resourceConsumption() — tallies fileInfos.filter(_.isS3) into the S3 fields.
• Worker — emits s3FileCount/s3BytesWritten gauges (and removes them on app cleanup) when hasS3Storage; top-N app ranking now sums disk + hdfs + s3 bytes.
• ResourceConsumptionSource — S3_FILE_COUNT/S3_BYTES_WRITTEN metric names.
• PbSerDeUtils/MetaUtil — round-trip the new fields.
• Tests updated (PbSerDeUtilsTest now covers S3 round-trip; WorkerInfoSuite toString; MasterStateMachineSuiteJ).

OSS has the identical gap but is intentionally out of scope here.

Why are the changes needed?

Worker per-user and per-application ResourceConsumption metrics silently drop S3-backed shuffle files, so diskBytesWritten/hdfsBytesWritten (and their file counts) read 0 on an S3-only worker, and the top-N application ranking never sees S3 apps.

Root cause — StorageManager.resourceConsumption():

val diskFileInfos = fileInfos.filter(!_.isDFS)   // local disk
val hdfsFileInfos = fileInfos.filter(_.isHdfs)   // HDFS only

DiskFileInfo.isDFS() = HDFS || S3 || OSS and isHdfs() = HDFS, so an S3 file (isDFS=true, isHdfs=false) is excluded from both buckets. isS3() exists and is used throughout the storage I/O path but never in consumption accounting, and ResourceConsumption has no S3 field.

This affects both celeborn.metrics.worker.app.topResourceConsumption.count (per-app) and celeborn.master.userResourceConsumption.metrics.enabled (per-user/tenant) — both read 0 on S3 storage.

Does this PR introduce any user-facing change?

Yes — adds new worker gauges s3FileCount/s3BytesWritten and includes S3 bytes in top-N app resource consumption ranking. New proto fields are additive (unset ⇒ 0), so mixed-version worker/master heartbeats are safe.

How was this patch tested?

PbSerDeUtilsTest covers S3 round-trip serialization; WorkerInfoSuite and MasterStateMachineSuiteJ updated for the new fields.

[Copilot]

Pull request overview

This PR extends Celeborn’s per-user/per-application ResourceConsumption accounting and related metrics to properly include S3-backed shuffle files, aligning S3 behavior with existing disk/HDFS tracking and ensuring S3-only workers are visible in resource metrics and top-N rankings.

Changes:

• Adds S3 fields (s3BytesWritten, s3FileCount) to ResourceConsumption and both protobuf representations, including serde support in PbSerDeUtils/MetaUtil.
• Updates worker-side accounting (StorageManager.resourceConsumption) to bucket and sum S3 file infos, and updates Worker to emit S3 gauges and include S3 bytes in top-N ranking.
• Updates tests and string expectations to reflect the expanded ResourceConsumption model.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala	Adds S3-aware top-N ranking and S3 gauges lifecycle management.
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala	Adds S3 bucketing in resourceConsumption() so S3-backed files are accounted.
master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java	Updates Java test construction to include new S3 fields.
master/src/main/proto/Resource.proto	Adds new S3 fields to master proto ResourceConsumption.
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java	Updates master-side serde to round-trip S3 fields.
common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala	Extends serde tests to cover S3 fields.
common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala	Updates expected toString output to include S3 fields.
common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala	Updates common serde to include S3 fields.
common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala	Adds S3 fields and incorporates them into arithmetic / toString.
common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala	Adds metric name constants for S3 gauges.
common/src/main/proto/TransportMessages.proto	Adds new S3 fields to transport proto PbResourceConsumption.

Comments suppressed due to low confidence (1)

common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala:103

• simpleString still omits the new S3 consumption fields, so any logs/UI paths that use it will continue to hide S3 usage (and it no longer matches the expanded ResourceConsumption state). Consider including s3BytesWritten/s3FileCount here for consistency with toString.

      s" s3FileCount: $s3FileCount," +
      s" subResourceConsumptions: $subResourceConsumptionString)"
  }

  def simpleString: String = {

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[SteNicholas]

@shellfish007, could you also add oss resource consumption in per-app/per-user metric?