This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new 562ff0cf23 [CELEBORN-2389] Track S3 resource consumption in
per-app/per-user metrics
562ff0cf23 is described below
commit 562ff0cf232bd234fe4a0df5698c1e43407a6eb4
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
(cherry picked from commit 3fefc4dc63f4d8e3ff55339b8e4ffd0a867e2548)
Signed-off-by: zhengtao <[email protected]>
---
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 a813a9e501..bf3e06fd94 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -702,6 +702,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 fb557fee18..ae0cf3afc2 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -6046,7 +6046,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.")
@@ -6059,7 +6060,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 cac7550871..fa2e26d75a 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 3010f7e049..de9b327840 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
@@ -151,7 +151,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 148c49f074..f3b355776e 100644
--- a/docs/configuration/metrics.md
+++ b/docs/configuration/metrics.md
@@ -31,8 +31,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 1aa4b3749a..07c3e46c56 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -154,6 +154,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:**
@@ -389,6 +393,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 d8fceee156..f19a26ea0a 100644
--- a/master/src/main/proto/Resource.proto
+++ b/master/src/main/proto/Resource.proto
@@ -221,6 +221,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 46a9c68e61..721171d3da 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
@@ -137,7 +137,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(
@@ -145,7 +146,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(
@@ -153,7 +159,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));
@@ -162,7 +173,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(
@@ -170,7 +182,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(
@@ -178,7 +195,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));
@@ -187,7 +209,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(
@@ -195,7 +218,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(
@@ -203,7 +231,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 1903e31b7f..3b8930b997 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)
@@ -748,7 +750,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) =>
@@ -757,12 +759,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)
@@ -801,6 +805,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(
@@ -884,6 +912,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 80aee90a77..ff82222394 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
@@ -1064,12 +1064,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(
