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 3fefc4dc63 [CELEBORN-2389] Track S3 resource consumption in
per-app/per-user metrics
3fefc4dc63 is described below
commit 3fefc4dc63f4d8e3ff55339b8e4ffd0a867e2548
Author: hzheng182 <[email protected]>
AuthorDate: Sat Oct 31 19:04:57 2026 +0800
[CELEBORN-2389] Track S3 resource consumption in per-app/per-user metrics
### What changes were proposed in this pull request?
Adds first-class S3 and OSS accounting to `ResourceConsumption`, mirroring
the existing disk/hdfs pattern:
- **`PbResourceConsumption`** / **`Resource.proto` `ResourceConsumption`**:
new fields `s3BytesWritten = 6`, `s3FileCount = 7`, `ossBytesWritten = 8`,
`ossFileCount = 9` (backward compatible; older peers default to 0).
- **`ResourceConsumption`**: new
`s3BytesWritten`/`s3FileCount`/`ossBytesWritten`/`ossFileCount` fields,
defaulted to 0 so all existing construction sites compile unchanged; included
in `add`/`subtract`/`toString`/`simpleString`.
- **`StorageManager.resourceConsumption()`**: tallies
`fileInfos.filter(_.isS3)` and `fileInfos.filter(_.isOSS)` into the new fields.
- **`Worker`**: emits `s3FileCount`/`s3BytesWritten` gauges when
`hasS3Storage` and `ossFileCount`/`ossBytesWritten` gauges when `hasOssStorage`
(and removes them on app cleanup); top-N app ranking and its bytes-written
threshold now sum `disk + hdfs + s3 + oss` bytes.
- **`QuotaManager`**: registers per-user
`s3FileCount`/`s3BytesWritten`/`ossFileCount`/`ossBytesWritten` gauges on the
Master when `celeborn.master.userResourceConsumption.metrics.enabled` is true.
The S3 gauges are registered once a user has S3 consumption, and likewise for
OSS, so clusters without S3/OSS keep the current per-user metric count and stay
within `celeborn.metrics.capacity`.
- **`ResourceConsumptionSource`**:
`S3_FILE_COUNT`/`S3_BYTES_WRITTEN`/`OSS_FILE_COUNT`/`OSS_BYTES_WRITTEN` metric
names.
- **`PbSerDeUtils`**/**`MetaUtil`**: round-trip the new fields.
- **Docs**: `docs/configuration/metrics.md` (top-N ranking and threshold
include S3 and OSS) and `docs/monitoring.md` (new gauges).
### Why are the changes needed?
Worker per-user and per-application `ResourceConsumption` metrics silently
drop S3- and OSS-backed shuffle files, so `diskBytesWritten`/`hdfsBytesWritten`
(and their file counts) read 0 on an S3-only or OSS-only worker, and the top-N
application ranking never sees those apps.
Root cause, in `StorageManager.resourceConsumption()`:
```scala
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
or OSS file (`isDFS=true`, `isHdfs=false`) is excluded from both buckets.
`isS3()`/`isOSS()` exist and are used throughout the storage I/O path but never
in consumption accounting, and `ResourceConsumption` has no S3/OSS fields.
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/OSS storage.
### Does this PR introduce _any_ user-facing change?
Yes:
- New worker gauges
`s3FileCount`/`s3BytesWritten`/`ossFileCount`/`ossBytesWritten`.
- New master per-user gauges with the same names, registered once the user
has S3 or OSS consumption.
- S3 and OSS bytes are included in the top-N app resource consumption
ranking and its bytes-written threshold.
New proto fields are additive (unset ⇒ 0), so mixed-version worker/master
heartbeats are safe.
### How was this patch tested?
- `QuotaManagerSuite`: new test registers a user with nonzero S3/OSS
consumption and asserts the Master exposes the four per-user gauges with the
expected values, and that a disk-only user gets no S3/OSS gauges.
- `PbSerDeUtilsTest`: S3/OSS round-trip serialization with nonzero values.
- `MasterStateMachineSuiteJ`: fixtures use nonzero S3/OSS values. The Raft
snapshot writes workers with `eliminateUserResourceConsumption = true`, and
user resource consumption isn't replicated through the Raft log either; after a
restore or leader change, the leader rebuilds it from the full per-user
snapshot each worker sends with its heartbeat. That heartbeat path serializes
through `PbSerDeUtils`, covered by `PbSerDeUtilsTest` above.
- `WorkerInfoSuite` and `ConfigurationSuite` updated for the new fields and
docs.
Closes #3766 from shellfish007/upstream-pr/s3-resource-consumption-metrics.
Lead-authored-by: hzheng182 <[email protected]>
Co-authored-by: shellfish007 <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/168
AI-Contributed/UT: 0/172
---
common/src/main/proto/TransportMessages.proto | 4 +
.../org/apache/celeborn/common/CelebornConf.scala | 6 +-
.../metrics/source/ResourceConsumptionSource.scala | 8 ++
.../common/quota/ResourceConsumption.scala | 28 +++++-
.../apache/celeborn/common/util/PbSerDeUtils.scala | 10 +-
.../celeborn/common/meta/WorkerInfoSuite.scala | 2 +-
.../celeborn/common/util/PbSerDeUtilsTest.scala | 10 +-
docs/configuration/metrics.md | 4 +-
docs/monitoring.md | 8 ++
.../deploy/master/clustermeta/MetaUtil.java | 10 +-
master/src/main/proto/Resource.proto | 4 +
.../service/deploy/master/quota/QuotaManager.scala | 28 ++++++
.../clustermeta/ha/MasterStateMachineSuiteJ.java | 51 ++++++++--
.../deploy/master/quota/QuotaManagerSuite.scala | 109 +++++++++++++++++++--
.../celeborn/service/deploy/worker/Worker.scala | 50 +++++++++-
.../deploy/worker/storage/StorageManager.scala | 8 +-
16 files changed, 306 insertions(+), 34 deletions(-)
diff --git a/common/src/main/proto/TransportMessages.proto
b/common/src/main/proto/TransportMessages.proto
index b7b04d6cf2..16aa934c4f 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -745,6 +745,10 @@ message PbResourceConsumption {
int64 hdfsBytesWritten = 3;
int64 hdfsFileCount = 4;
map<string, PbResourceConsumption> subResourceConsumptions = 5;
+ int64 s3BytesWritten = 6;
+ int64 s3FileCount = 7;
+ int64 ossBytesWritten = 8;
+ int64 ossFileCount = 9;
}
// deprecated
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index c1dc8aed87..23bb955c58 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -6250,7 +6250,8 @@ object CelebornConf extends Logging {
buildConf("celeborn.metrics.worker.app.topResourceConsumption.count")
.categories("metrics")
.doc("Size for top items about top resource consumption applications
list of worker. " +
- "The top resource consumption is determined by sum of diskBytesWritten
and hdfsBytesWritten. " +
+ "The top resource consumption is determined by sum of
diskBytesWritten, hdfsBytesWritten, " +
+ "s3BytesWritten and ossBytesWritten. " +
"The top resource consumption count prevents the total number of
metrics from exceeding the metrics capacity. " +
"Note: This will add applicationId as label which is considered as a
high cardinality label, " +
"be careful enabling it on metrics systems that are not optimized for
high cardinality columns.")
@@ -6263,7 +6264,8 @@ object CelebornConf extends Logging {
.categories("metrics")
.doc("Threshold of bytes written for top resource consumption
applications list of worker. " +
"The application which has bytes written less than this threshold will
not be included " +
- "in the top resource consumption list, including diskBytesWritten and
hdfsBytesWritten.")
+ "in the top resource consumption list, including diskBytesWritten,
hdfsBytesWritten, " +
+ "s3BytesWritten and ossBytesWritten.")
.version("0.6.0")
.bytesConf(ByteUnit.BYTE)
.createWithDefault(0)
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
index df33310bba..8e5b2d1acd 100644
---
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
@@ -33,4 +33,12 @@ object ResourceConsumptionSource {
val HDFS_FILE_COUNT = "hdfsFileCount"
val HDFS_BYTES_WRITTEN = "hdfsBytesWritten"
+
+ val S3_FILE_COUNT = "s3FileCount"
+
+ val S3_BYTES_WRITTEN = "s3BytesWritten"
+
+ val OSS_FILE_COUNT = "ossFileCount"
+
+ val OSS_BYTES_WRITTEN = "ossBytesWritten"
}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
b/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
index d454bb894c..9c3700696c 100644
---
a/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
@@ -28,7 +28,11 @@ case class ResourceConsumption(
diskFileCount: Long,
hdfsBytesWritten: Long,
hdfsFileCount: Long,
- var subResourceConsumptions: util.Map[String, ResourceConsumption] = null)
{
+ var subResourceConsumptions: util.Map[String, ResourceConsumption] = null,
+ s3BytesWritten: Long = 0,
+ s3FileCount: Long = 0,
+ ossBytesWritten: Long = 0,
+ ossFileCount: Long = 0) {
def withSubResourceConsumptions(
resourceConsumptions: util.Map[String, ResourceConsumption]):
ResourceConsumption = {
@@ -41,7 +45,11 @@ case class ResourceConsumption(
diskBytesWritten + other.diskBytesWritten,
diskFileCount + other.diskFileCount,
hdfsBytesWritten + other.hdfsBytesWritten,
- hdfsFileCount + other.hdfsFileCount)
+ hdfsFileCount + other.hdfsFileCount,
+ s3BytesWritten = s3BytesWritten + other.s3BytesWritten,
+ s3FileCount = s3FileCount + other.s3FileCount,
+ ossBytesWritten = ossBytesWritten + other.ossBytesWritten,
+ ossFileCount = ossFileCount + other.ossFileCount)
}
def subtract(other: ResourceConsumption): ResourceConsumption = {
@@ -49,7 +57,11 @@ case class ResourceConsumption(
diskBytesWritten - other.diskBytesWritten,
diskFileCount - other.diskFileCount,
hdfsBytesWritten - other.hdfsBytesWritten,
- hdfsFileCount - other.hdfsFileCount)
+ hdfsFileCount - other.hdfsFileCount,
+ s3BytesWritten = s3BytesWritten - other.s3BytesWritten,
+ s3FileCount = s3FileCount - other.s3FileCount,
+ ossBytesWritten = ossBytesWritten - other.ossBytesWritten,
+ ossFileCount = ossFileCount - other.ossFileCount)
}
def addSubResourceConsumptions(otherSubResourceConsumptions: Map[
@@ -89,6 +101,10 @@ case class ResourceConsumption(
s" diskFileCount: $diskFileCount," +
s" hdfsBytesWritten: ${Utils.bytesToString(hdfsBytesWritten)}," +
s" hdfsFileCount: $hdfsFileCount," +
+ s" s3BytesWritten: ${Utils.bytesToString(s3BytesWritten)}," +
+ s" s3FileCount: $s3FileCount," +
+ s" ossBytesWritten: ${Utils.bytesToString(ossBytesWritten)}," +
+ s" ossFileCount: $ossFileCount," +
s" subResourceConsumptions: $subResourceConsumptionString)"
}
@@ -96,6 +112,10 @@ case class ResourceConsumption(
s"ResourceConsumption(diskBytesWritten:
${Utils.bytesToString(diskBytesWritten)}," +
s" diskFileCount: $diskFileCount," +
s" hdfsBytesWritten: ${Utils.bytesToString(hdfsBytesWritten)}," +
- s" hdfsFileCount: $hdfsFileCount)"
+ s" hdfsFileCount: $hdfsFileCount," +
+ s" s3BytesWritten: ${Utils.bytesToString(s3BytesWritten)}," +
+ s" s3FileCount: $s3FileCount," +
+ s" ossBytesWritten: ${Utils.bytesToString(ossBytesWritten)}," +
+ s" ossFileCount: $ossFileCount)"
}
}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index 28184a9bfd..b56b5642d3 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -233,7 +233,11 @@ object PbSerDeUtils {
pbResourceConsumption.getDiskFileCount,
pbResourceConsumption.getHdfsBytesWritten,
pbResourceConsumption.getHdfsFileCount,
-
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionsMap))
+
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionsMap),
+ pbResourceConsumption.getS3BytesWritten,
+ pbResourceConsumption.getS3FileCount,
+ pbResourceConsumption.getOssBytesWritten,
+ pbResourceConsumption.getOssFileCount)
def toPbResourceConsumption(resourceConsumption: ResourceConsumption):
PbResourceConsumption = {
PbResourceConsumption.newBuilder
@@ -241,6 +245,10 @@ object PbSerDeUtils {
.setDiskFileCount(resourceConsumption.diskFileCount)
.setHdfsBytesWritten(resourceConsumption.hdfsBytesWritten)
.setHdfsFileCount(resourceConsumption.hdfsFileCount)
+ .setS3BytesWritten(resourceConsumption.s3BytesWritten)
+ .setS3FileCount(resourceConsumption.s3FileCount)
+ .setOssBytesWritten(resourceConsumption.ossBytesWritten)
+ .setOssFileCount(resourceConsumption.ossFileCount)
.putAllSubResourceConsumptions(toPbSubResourceConsumptions(
resourceConsumption.subResourceConsumptions))
.build
diff --git
a/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
b/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
index df77798eea..da1c900acb 100644
---
a/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
+++
b/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
@@ -363,7 +363,7 @@ class WorkerInfoSuite extends CelebornFunSuite {
| DiskInfo1: DiskInfo(maxSlots: 0, availableSlots: 0, committed
shuffles 0, running applications 0, mountPoint: disk1, usableSpace: 2048.0 MiB,
totalSpace: 2048.0 MiB, avgFlushTime: 1 ns, avgFetchTime: 1 ns, activeSlots:
10, storageType: SSD) status: HEALTHY dirs $placeholder shuffleAllocations:
Map()
| DiskInfo2: DiskInfo(maxSlots: 0, availableSlots: 0, committed
shuffles 0, running applications 0, mountPoint: disk2, usableSpace: 2048.0 MiB,
totalSpace: 2048.0 MiB, avgFlushTime: 2 ns, avgFetchTime: 2 ns, activeSlots:
20, storageType: SSD) status: HEALTHY dirs $placeholder shuffleAllocations:
Map()
|UserResourceConsumption: $placeholder
- | UserIdentifier: `tenant1`.`name1`, ResourceConsumption:
ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1,
hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1, subResourceConsumptions:
(application_1697697127390_2171854 -> ResourceConsumption(diskBytesWritten:
20.0 MiB, diskFileCount: 1, hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1,
subResourceConsumptions: empty)))
+ | UserIdentifier: `tenant1`.`name1`, ResourceConsumption:
ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1,
hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1, s3BytesWritten: 0.0 B,
s3FileCount: 0, ossBytesWritten: 0.0 B, ossFileCount: 0,
subResourceConsumptions: (application_1697697127390_2171854 ->
ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1,
hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1, s3BytesWritten: 0.0 B,
s3FileCount: 0, ossBytesWritten: [...]
|WorkerRef: null
|NetworkLocation: /default-rack
|NextInterruptionNotice: None
diff --git
a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala
b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala
index 7c1365eab5..52d8a111ac 100644
---
a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala
+++
b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala
@@ -152,7 +152,15 @@ class PbSerDeUtilsTest extends CelebornFunSuite {
mountPoints.add("/mnt")
val cache = JavaUtils.newConcurrentHashMap[String, UserIdentifier]()
- val resourceConsumption1 = ResourceConsumption(1000, 2000, 3000, 4000)
+ val resourceConsumption1 = ResourceConsumption(
+ 1000,
+ 2000,
+ 3000,
+ 4000,
+ s3BytesWritten = 5000,
+ s3FileCount = 6000,
+ ossBytesWritten = 7000,
+ ossFileCount = 8000)
val resourceConsumption2 = ResourceConsumption(
2000,
4000,
diff --git a/docs/configuration/metrics.md b/docs/configuration/metrics.md
index 93d5c2cbea..b0f9183f65 100644
--- a/docs/configuration/metrics.md
+++ b/docs/configuration/metrics.md
@@ -37,8 +37,8 @@ license: |
| 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 | |
-| celeborn.metrics.worker.app.topResourceConsumption.bytesWrittenThreshold |
0b | false | Threshold of bytes written for top resource consumption
applications list of worker. The application which has bytes written less than
this threshold will not be included in the top resource consumption list,
including diskBytesWritten and hdfsBytesWritten. | 0.6.0 | |
-| celeborn.metrics.worker.app.topResourceConsumption.count | 0 | false | Size
for top items about top resource consumption applications list of worker. The
top resource consumption is determined by sum of diskBytesWritten and
hdfsBytesWritten. The top resource consumption count prevents the total number
of metrics from exceeding the metrics capacity. Note: This will add
applicationId as label which is considered as a high cardinality label, be
careful enabling it on metrics systems that [...]
+| celeborn.metrics.worker.app.topResourceConsumption.bytesWrittenThreshold |
0b | false | Threshold of bytes written for top resource consumption
applications list of worker. The application which has bytes written less than
this threshold will not be included in the top resource consumption list,
including diskBytesWritten, hdfsBytesWritten, s3BytesWritten and
ossBytesWritten. | 0.6.0 | |
+| celeborn.metrics.worker.app.topResourceConsumption.count | 0 | false | Size
for top items about top resource consumption applications list of worker. The
top resource consumption is determined by sum of diskBytesWritten,
hdfsBytesWritten, s3BytesWritten and ossBytesWritten. The top resource
consumption count prevents the total number of metrics from exceeding the
metrics capacity. Note: This will add applicationId as label which is
considered as a high cardinality label, be careful ena [...]
| celeborn.metrics.worker.appLevel.enabled | true | false | When true, enable
worker application level metrics. Note: applicationId is considered as a high
cardinality label, be careful enabling it on metrics systems that are not
optimized for high cardinality columns. | 0.6.0 | |
| celeborn.metrics.worker.pauseSpentTime.forceAppend.threshold | 10 | false |
Force append worker pause spent time even if worker still in pause serving
state. Help user can find worker pause spent time increase, when worker always
been pause state. | | |
<!--end-include-->
diff --git a/docs/monitoring.md b/docs/monitoring.md
index eb0da59170..ff189ad4d5 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -155,6 +155,10 @@ These metrics are exposed by Celeborn master.
| diskBytesWritten | The amount of disk files consumption by each user. |
| hdfsFileCount | The count of hdfs files consumption by each user. |
| hdfsBytesWritten | The amount of hdfs files consumption by each user. |
+ | s3FileCount | The count of s3 files consumption by each user.
Registered once the user has s3 consumption. |
+ | s3BytesWritten | The amount of s3 files consumption by each user.
Registered once the user has s3 consumption. |
+ | ossFileCount | The count of oss files consumption by each user.
Registered once the user has oss consumption. |
+ | ossBytesWritten | The amount of oss files consumption by each user.
Registered once the user has oss consumption. |
- namespace=ThreadPool
- **notes:**
@@ -404,6 +408,10 @@ These metrics are exposed by Celeborn worker.
| diskBytesWritten | The amount of disk files consumption by each user. |
| hdfsFileCount | The count of hdfs files consumption by each user. |
| hdfsBytesWritten | The amount of hdfs files consumption by each user. |
+ | s3FileCount | The count of s3 files consumption by each user. |
+ | s3BytesWritten | The amount of s3 files consumption by each user. |
+ | ossFileCount | The count of oss files consumption by each user. |
+ | ossBytesWritten | The amount of oss files consumption by each user. |
- namespace=ThreadPool
- **notes:**
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
index fb26a4e6f9..981ebc9401 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
@@ -147,7 +147,11 @@ public class MetaUtil {
pbResourceConsumption.getDiskFileCount(),
pbResourceConsumption.getHdfsBytesWritten(),
pbResourceConsumption.getHdfsFileCount(),
-
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionMap()));
+
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionMap()),
+ pbResourceConsumption.getS3BytesWritten(),
+ pbResourceConsumption.getS3FileCount(),
+ pbResourceConsumption.getOssBytesWritten(),
+ pbResourceConsumption.getOssFileCount());
}
public static ResourceProtos.ResourceConsumption toPbResourceConsumption(
@@ -157,6 +161,10 @@ public class MetaUtil {
.setDiskFileCount(resourceConsumption.diskFileCount())
.setHdfsBytesWritten(resourceConsumption.hdfsBytesWritten())
.setHdfsFileCount(resourceConsumption.hdfsFileCount())
+ .setS3BytesWritten(resourceConsumption.s3BytesWritten())
+ .setS3FileCount(resourceConsumption.s3FileCount())
+ .setOssBytesWritten(resourceConsumption.ossBytesWritten())
+ .setOssFileCount(resourceConsumption.ossFileCount())
.putAllSubResourceConsumption(
toPbSubResourceConsumptions(resourceConsumption.subResourceConsumptions()))
.build();
diff --git a/master/src/main/proto/Resource.proto
b/master/src/main/proto/Resource.proto
index d3d71d9390..5956834e8c 100644
--- a/master/src/main/proto/Resource.proto
+++ b/master/src/main/proto/Resource.proto
@@ -231,6 +231,10 @@ message ResourceConsumption {
int64 hdfsBytesWritten = 3;
int64 hdfsFileCount = 4;
map<string, ResourceConsumption> subResourceConsumption = 5;
+ int64 s3BytesWritten = 6;
+ int64 s3FileCount = 7;
+ int64 ossBytesWritten = 8;
+ int64 ossFileCount = 9;
}
enum Status {
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala
index 9b2f4d8818..464d3ded4d 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala
@@ -393,6 +393,34 @@ class QuotaManager(
.map(_.hdfsBytesWritten)
.getOrElse(0L)
}
+ // Register S3/OSS gauges only once the user has used that storage, so
clusters without
+ // S3/OSS do not double the per-user metric count.
+ val consumption = userResourceConsumptionCache.get(userIdentifier)
+ if (consumption != null && (consumption.s3FileCount > 0 ||
consumption.s3BytesWritten > 0)) {
+ resourceConsumptionSource.addGauge(S3_FILE_COUNT,
userIdentifier.toMap) { () =>
+ Option(userResourceConsumptionCache.get(userIdentifier))
+ .map(_.s3FileCount)
+ .getOrElse(0L)
+ }
+ resourceConsumptionSource.addGauge(S3_BYTES_WRITTEN,
userIdentifier.toMap) { () =>
+ Option(userResourceConsumptionCache.get(userIdentifier))
+ .map(_.s3BytesWritten)
+ .getOrElse(0L)
+ }
+ }
+ if (consumption != null &&
+ (consumption.ossFileCount > 0 || consumption.ossBytesWritten > 0)) {
+ resourceConsumptionSource.addGauge(OSS_FILE_COUNT,
userIdentifier.toMap) { () =>
+ Option(userResourceConsumptionCache.get(userIdentifier))
+ .map(_.ossFileCount)
+ .getOrElse(0L)
+ }
+ resourceConsumptionSource.addGauge(OSS_BYTES_WRITTEN,
userIdentifier.toMap) { () =>
+ Option(userResourceConsumptionCache.get(userIdentifier))
+ .map(_.ossBytesWritten)
+ .getOrElse(0L)
+ }
+ }
}
}
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
index bc9b46670b..1518bd1334 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
@@ -166,7 +166,8 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption1 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption1.put(
- new UserIdentifier("tenant1", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
+ new UserIdentifier("tenant1", "name1"),
+ new ResourceConsumption(1000, 1, 1000, 1, null, 1000, 1, 1000, 1));
userResourceConsumption1.put(
new UserIdentifier("tenant1", "name2"),
new ResourceConsumption(
@@ -174,7 +175,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
2,
2000,
2,
- Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId2", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 2000,
+ 2,
+ 2000,
+ 2));
userResourceConsumption1.put(
new UserIdentifier("tenant1", "name3"),
new ResourceConsumption(
@@ -182,7 +188,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
3,
3000,
3,
- Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId3", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 3000,
+ 3,
+ 3000,
+ 3));
Map<String, DiskInfo> disks2 = new HashMap<>();
disks2.put("disk1", new DiskInfo("disk1", 64 * 1024 * 1024 * 1024L, 100,
100, 0));
@@ -191,7 +202,8 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption2 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption2.put(
- new UserIdentifier("tenant2", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
+ new UserIdentifier("tenant2", "name1"),
+ new ResourceConsumption(1000, 1, 1000, 1, null, 1000, 1, 1000, 1));
userResourceConsumption2.put(
new UserIdentifier("tenant2", "name2"),
new ResourceConsumption(
@@ -199,7 +211,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
2,
2000,
2,
- Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId2", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 2000,
+ 2,
+ 2000,
+ 2));
userResourceConsumption2.put(
new UserIdentifier("tenant2", "name3"),
new ResourceConsumption(
@@ -207,7 +224,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
3,
3000,
3,
- Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId3", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 3000,
+ 3,
+ 3000,
+ 3));
Map<String, DiskInfo> disks3 = new HashMap<>();
disks3.put("disk1", new DiskInfo("disk1", 64 * 1024 * 1024 * 1024L, 100,
100, 0));
@@ -216,7 +238,8 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption3 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption3.put(
- new UserIdentifier("tenant3", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
+ new UserIdentifier("tenant3", "name1"),
+ new ResourceConsumption(1000, 1, 1000, 1, null, 1000, 1, 1000, 1));
userResourceConsumption3.put(
new UserIdentifier("tenant3", "name2"),
new ResourceConsumption(
@@ -224,7 +247,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
2,
2000,
2,
- Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId2", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 2000,
+ 2,
+ 2000,
+ 2));
userResourceConsumption3.put(
new UserIdentifier("tenant3", "name3"),
new ResourceConsumption(
@@ -232,7 +260,12 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
3,
3000,
3,
- Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
+ Collections.singletonMap(
+ "appId3", new ResourceConsumption(2000, 2, 2000, 2, null,
2000, 2, 2000, 2)),
+ 3000,
+ 3,
+ 3000,
+ 3));
WorkerInfo info1 = new WorkerInfo("host1", 1, 2, 3, 10, 13, disks1,
userResourceConsumption1);
WorkerInfo info2 = new WorkerInfo("host2", 4, 5, 6, 11, 15, disks2,
userResourceConsumption2);
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala
index 425a3404b4..fce63b0bac 100644
---
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala
@@ -221,7 +221,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 150.0 GiB, " +
"diskFileCount: 15000, " +
"hdfsBytesWritten: 25.0 GiB, " +
- "hdfsFileCount: 20), " +
+ "hdfsFileCount: 20, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: " +
"Quota[" +
"diskBytesWritten=100.0 GiB, " +
@@ -245,7 +249,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 150.0 GiB, " +
"diskFileCount: 15000, " +
"hdfsBytesWritten: 25.0 GiB, " +
- "hdfsFileCount: 20), " +
+ "hdfsFileCount: 20, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: Quota[" +
"diskBytesWritten=100.0 GiB, " +
"diskFileCount=10000, " +
@@ -258,7 +266,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 50.0 GiB, " +
"diskFileCount: 5000, " +
"hdfsBytesWritten: 5.0 GiB, " +
- "hdfsFileCount: 20), " +
+ "hdfsFileCount: 20, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: " +
"Quota[" +
"diskBytesWritten=50.0 GiB, " +
@@ -309,7 +321,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 40.0 GiB, " +
"diskFileCount: 500, " +
"hdfsBytesWritten: 3.0 GiB, " +
- "hdfsFileCount: 20), " +
+ "hdfsFileCount: 20, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: " +
"Quota[diskBytesWritten=20.0 GiB, " +
"diskFileCount=9223372036854775807, " +
@@ -435,6 +451,61 @@ class QuotaManagerSuite extends CelebornFunSuite
clearUserConsumption()
}
+ test("test user resource consumption metrics include S3 and OSS") {
+ val user = UserIdentifier("tenant_s3", "s3User")
+ val appConsumption = ResourceConsumption(
+ 100,
+ 1,
+ 200,
+ 2,
+ s3BytesWritten = 3000,
+ s3FileCount = 3,
+ ossBytesWritten = 4000,
+ ossFileCount = 4)
+ val userConsumption = ResourceConsumption(
+ 100,
+ 1,
+ 200,
+ 2,
+ Map("s3App" -> appConsumption).asJava,
+ s3BytesWritten = 3000,
+ s3FileCount = 3,
+ ossBytesWritten = 4000,
+ ossFileCount = 4)
+ val diskUser = UserIdentifier("tenant_s3", "diskUser")
+ statusSystem.updateApplicationInfo("s3App", user, extraInfo)
+ statusSystem.updateApplicationInfo("diskApp", diskUser, extraInfo)
+ addUserConsumption(user, userConsumption)
+ addUserConsumption(
+ diskUser,
+ ResourceConsumption(
+ 100,
+ 1,
+ 200,
+ 2,
+ Map("diskApp" -> ResourceConsumption(100, 1, 200, 2)).asJava))
+
+ try {
+ quotaManager.updateResourceConsumption()
+
+ val res = resourceConsumptionSource.getMetrics
+ val labels =
s"""$metricsInstanceLabel,name="s3User",role="Master",tenantId="tenant_s3""""
+ assert(res.contains(s"""metrics_s3FileCount_Value{$labels} 3"""))
+ assert(res.contains(s"""metrics_s3BytesWritten_Value{$labels} 3000"""))
+ assert(res.contains(s"""metrics_ossFileCount_Value{$labels} 4"""))
+ assert(res.contains(s"""metrics_ossBytesWritten_Value{$labels} 4000"""))
+
+ // A user without S3/OSS usage gets no S3/OSS gauges.
+ val diskLabels =
+
s"""$metricsInstanceLabel,name="diskUser",role="Master",tenantId="tenant_s3""""
+ assert(res.contains(s"""metrics_diskFileCount_Value{$diskLabels} 1"""))
+ assert(!res.contains(s"""metrics_s3FileCount_Value{$diskLabels}"""))
+ assert(!res.contains(s"""metrics_ossFileCount_Value{$diskLabels}"""))
+ } finally {
+ clearUserConsumption()
+ }
+ }
+
test("test user level conf") {
val conf1 = new CelebornConf()
conf1.set(CelebornConf.DYNAMIC_CONFIG_STORE_BACKEND, "FS")
@@ -521,7 +592,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 150.0 GiB, " +
"diskFileCount: 15000, " +
"hdfsBytesWritten: 25.0 GiB, " +
- "hdfsFileCount: 20), " +
+ "hdfsFileCount: 20, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: " +
"Quota[" +
"diskBytesWritten=100.0 GiB, " +
@@ -537,7 +612,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 80.0 GiB, " +
"diskFileCount: 0, " +
"hdfsBytesWritten: 0.0 B, " +
- "hdfsFileCount: 0), " +
+ "hdfsFileCount: 0, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: " +
"Quota[" +
"diskBytesWritten=10.0 GiB, " +
@@ -643,7 +722,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 150.0 GiB, " +
"diskFileCount: 0, " +
"hdfsBytesWritten: 0.0 B, " +
- "hdfsFileCount: 0), " +
+ "hdfsFileCount: 0, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: Quota[" +
"diskBytesWritten=100.0 GiB, " +
"diskFileCount=10000, " +
@@ -657,7 +740,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 80.0 GiB, " +
"diskFileCount: 0, " +
"hdfsBytesWritten: 0.0 B, " +
- "hdfsFileCount: 0), " +
+ "hdfsFileCount: 0, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: Quota[" +
"diskBytesWritten=150.0 GiB, " +
"diskFileCount=1500, " +
@@ -671,7 +758,11 @@ class QuotaManagerSuite extends CelebornFunSuite
"diskBytesWritten: 150.0 GiB, " +
"diskFileCount: 0, " +
"hdfsBytesWritten: 0.0 B, " +
- "hdfsFileCount: 0), " +
+ "hdfsFileCount: 0, " +
+ "s3BytesWritten: 0.0 B, " +
+ "s3FileCount: 0, " +
+ "ossBytesWritten: 0.0 B, " +
+ "ossFileCount: 0), " +
"Threshold: Quota[" +
"diskBytesWritten=100.0 GiB, " +
"diskFileCount=10000, " +
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index f1b7d21f86..9a96bef46c 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -95,6 +95,8 @@ private[celeborn] class Worker(
private val secretRegistry = new
WorkerSecretRegistryImpl(conf.workerApplicationRegistryCacheSize)
private val hasHDFSStorage = conf.hasHDFSStorage
+ private val hasS3Storage = conf.hasS3Storage
+ private val hasOssStorage = conf.hasOssStorage
if (conf.logCelebornConfEnabled) {
logInfo(getConf)
@@ -764,7 +766,7 @@ private[celeborn] class Worker(
ResourceConsumption]): Unit = {
// Remove application top resource consumption gauges to refresh top
resource consumption metrics.
removeAppResourceConsumption(topApplicationUserIdentifiers.keySet().asScala)
- // Top resource consumption is determined by
diskBytesWritten+hdfsBytesWritten.
+ // Top resource consumption is determined by total bytes written across
all storage types.
userResourceConsumptions.asScala.filter { case (_, resourceConsumption) =>
CollectionUtils.isNotEmpty(resourceConsumption.subResourceConsumptions)
}.flatMap { case (userIdentifier, resourceConsumption) =>
@@ -773,12 +775,14 @@ private[celeborn] class Worker(
}
}.toSeq
.sortBy { case (_, _, appConsumption) =>
- appConsumption.diskBytesWritten + appConsumption.hdfsBytesWritten
+ appConsumption.diskBytesWritten + appConsumption.hdfsBytesWritten +
+ appConsumption.s3BytesWritten + appConsumption.ossBytesWritten
}
.reverse
.take(topAppResourceConsumptionCount).foreach {
case (appId, userIdentifier, appConsumption) =>
- if (appConsumption.diskBytesWritten +
appConsumption.hdfsBytesWritten >=
+ if (appConsumption.diskBytesWritten +
appConsumption.hdfsBytesWritten +
+ appConsumption.s3BytesWritten + appConsumption.ossBytesWritten >=
topAppResourceConsumptionBytesWrittenThreshold) {
topApplicationUserIdentifiers.put(appId, userIdentifier)
gaugeResourceConsumption(userIdentifier, appId, appConsumption)
@@ -817,6 +821,30 @@ private[celeborn] class Worker(
computeResourceConsumption(userIdentifier,
resourceConsumption).hdfsBytesWritten
}
}
+ if (hasS3Storage) {
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.S3_FILE_COUNT,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
resourceConsumption).s3FileCount
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.S3_BYTES_WRITTEN,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
resourceConsumption).s3BytesWritten
+ }
+ }
+ if (hasOssStorage) {
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.OSS_FILE_COUNT,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
resourceConsumption).ossFileCount
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.OSS_BYTES_WRITTEN,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
resourceConsumption).ossBytesWritten
+ }
+ }
}
private def computeResourceConsumption(
@@ -900,6 +928,22 @@ private[celeborn] class Worker(
ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
resourceConsumptionLabel)
}
+ if (hasS3Storage) {
+ resourceConsumptionSource.removeGauge(
+ ResourceConsumptionSource.S3_FILE_COUNT,
+ resourceConsumptionLabel)
+ resourceConsumptionSource.removeGauge(
+ ResourceConsumptionSource.S3_BYTES_WRITTEN,
+ resourceConsumptionLabel)
+ }
+ if (hasOssStorage) {
+ resourceConsumptionSource.removeGauge(
+ ResourceConsumptionSource.OSS_FILE_COUNT,
+ resourceConsumptionLabel)
+ resourceConsumptionSource.removeGauge(
+ ResourceConsumptionSource.OSS_BYTES_WRITTEN,
+ resourceConsumptionLabel)
+ }
}
private def removeAppActiveConnection(applicationIds: JHashSet[String]):
Unit = {
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index 50f2b5f888..c57ab17999 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -1066,12 +1066,18 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
: ResourceConsumption = {
val diskFileInfos = fileInfos.filter(!_.isDFS)
val hdfsFileInfos = fileInfos.filter(_.isHdfs)
+ val s3FileInfos = fileInfos.filter(_.isS3)
+ val ossFileInfos = fileInfos.filter(_.isOSS)
ResourceConsumption(
diskFileInfos.map(_.getFileLength).sum,
diskFileInfos.size,
hdfsFileInfos.map(_.getFileLength).sum,
hdfsFileInfos.size,
- subResourceConsumptions)
+ subResourceConsumptions,
+ s3FileInfos.map(_.getFileLength).sum,
+ s3FileInfos.size,
+ ossFileInfos.map(_.getFileLength).sum,
+ ossFileInfos.size)
}
def notifyFileInfoCommitted(
