This is an automated email from the ASF dual-hosted git repository.
angerszhuuuu pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.4 by this push:
new 7fcd2e43e [CELEBORN-1236][METRICS] Celeborn add metrics about thread
pool
7fcd2e43e is described below
commit 7fcd2e43e5befa0314d03b56aefcd5f8ec3538d1
Author: Angerszhuuuu <[email protected]>
AuthorDate: Fri Jan 26 18:14:05 2024 +0800
[CELEBORN-1236][METRICS] Celeborn add metrics about thread pool
Add metrics about worker's thread pool, help admin to observe the thread
pool's work status.
ThreadPool list as below:
1. celeborn-dispatcher
2. celeborn-netty-rpc-connection-executor
3. worker-disk-{mount_point}-cleaner
4. worker-device-checker
5. flusher-{mount_point}
6. worker-file-sorter-executor
7. worker-data-replicator
8. worker-files-committer
9. worker-expired-shuffle-cleaner
```
metrics_active_thread_count_Value{role="Worker",threadPool="celeborn-dispatcher"}
64 1706237338484
metrics_pending_task_count_Value{role="Worker",threadPool="celeborn-dispatcher"}
0 1706237338484
metrics_pool_size_Value{role="Worker",threadPool="celeborn-dispatcher"} 64
1706237338484
metrics_core_pool_size_Value{role="Worker",threadPool="celeborn-dispatcher"} 64
1706237338484
metrics_maximum_pool_size_Value{role="Worker",threadPool="celeborn-dispatcher"}
64 1706237338484
metrics_largest_pool_size_Value{role="Worker",threadPool="celeborn-dispatcher"}
64 1706237338484
metrics_active_thread_count_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
0 1706237338484
metrics_pending_task_count_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
0 1706237338484
metrics_pool_size_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
0 1706237338484
metrics_core_pool_size_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
64 1706237338484
metrics_maximum_pool_size_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
64 1706237338484
metrics_largest_pool_size_Value{role="Worker",threadPool="celeborn-netty-rpc-connection-executor"}
1 1706237338484
metrics_active_thread_count_Value{role="Worker",threadPool="worker-disk-/-cleaner"}
0 1706237338484
metrics_pending_task_count_Value{role="Worker",threadPool="worker-disk-/-cleaner"}
0 1706237338484
metrics_pool_size_Value{role="Worker",threadPool="worker-disk-/-cleaner"} 0
1706237338484
metrics_core_pool_size_Value{role="Worker",threadPool="worker-disk-/-cleaner"}
4 1706237338484
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-disk-/-cleaner"}
4 1706237338484
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-disk-/-cleaner"}
0 1706237338485
metrics_active_thread_count_Value{role="Worker",threadPool="worker-device-checker"}
0 1706237338485
metrics_pending_task_count_Value{role="Worker",threadPool="worker-device-checker"}
0 1706237338485
metrics_pool_size_Value{role="Worker",threadPool="worker-device-checker"} 2
1706237338485
metrics_core_pool_size_Value{role="Worker",threadPool="worker-device-checker"}
5 1706237338485
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-device-checker"}
5 1706237338485
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-device-checker"}
2 1706237338485
metrics_thread_count_Value{role="Worker",threadPool="LocalFlusher1441328175-/"}
2 1706237338485
metrics_thread_is_terminated_count_Value{role="Worker",threadPool="LocalFlusher1441328175-/"}
0 1706237338485
metrics_thread_is_shutdown_count_Value{role="Worker",threadPool="LocalFlusher1441328175-/"}
0 1706237338485
metrics_active_thread_count_Value{role="Worker",threadPool="worker-file-sorter-executor"}
0 1706237338485
metrics_pending_task_count_Value{role="Worker",threadPool="worker-file-sorter-executor"}
0 1706237338485
metrics_pool_size_Value{role="Worker",threadPool="worker-file-sorter-executor"}
0 1706237338485
metrics_core_pool_size_Value{role="Worker",threadPool="worker-file-sorter-executor"}
24 1706237338485
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-file-sorter-executor"}
24 1706237338485
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-file-sorter-executor"}
0 1706237338485
metrics_active_thread_count_Value{role="Worker",threadPool="worker-data-replicator"}
0 1706237338485
metrics_pending_task_count_Value{role="Worker",threadPool="worker-data-replicator"}
0 1706237338485
metrics_pool_size_Value{role="Worker",threadPool="worker-data-replicator"}
0 1706237338485
metrics_core_pool_size_Value{role="Worker",threadPool="worker-data-replicator"}
64 1706237338485
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-data-replicator"}
64 1706237338485
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-data-replicator"}
0 1706237338485
metrics_active_thread_count_Value{role="Worker",threadPool="worker-files-committer"}
0 1706237338485
metrics_pending_task_count_Value{role="Worker",threadPool="worker-files-committer"}
0 1706237338485
metrics_pool_size_Value{role="Worker",threadPool="worker-files-committer"}
0 1706237338485
metrics_core_pool_size_Value{role="Worker",threadPool="worker-files-committer"}
32 1706237338485
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-files-committer"}
32 1706237338485
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-files-committer"}
0 1706237338485
metrics_active_thread_count_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
0 1706237338485
metrics_pending_task_count_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
0 1706237338485
metrics_pool_size_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
2 1706237338485
metrics_core_pool_size_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
64 1706237338485
metrics_maximum_pool_size_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
64 1706237338485
metrics_largest_pool_size_Value{role="Worker",threadPool="worker-expired-shuffle-cleaner"}
2 1706237338485
```
Help observe server status
No
MT
Closes #2239 from AngersZhuuuu/CLEBORN-1236.
Authored-by: Angerszhuuuu <[email protected]>
Signed-off-by: Angerszhuuuu <[email protected]>
(cherry picked from commit 3ffed66c409340e866012c823093dd6ee294238a)
Signed-off-by: Angerszhuuuu <[email protected]>
---
METRICS.md | 238 +++++++++++----------
.../common/metrics/source/ThreadPoolSource.scala | 179 ++++++++++++++++
.../celeborn/common/rpc/netty/NettyRpcEnv.scala | 2 +-
.../apache/celeborn/common/util/ThreadUtils.scala | 11 +-
docs/monitoring.md | 26 +++
.../celeborn/service/deploy/master/Master.scala | 24 ++-
.../worker/storage/PartitionFilesSorter.java | 2 +-
.../celeborn/service/deploy/worker/Worker.scala | 27 +--
.../deploy/worker/storage/DeviceMonitor.scala | 2 +-
.../service/deploy/worker/storage/Flusher.scala | 5 +-
.../deploy/worker/storage/StorageManager.scala | 7 +-
11 files changed, 376 insertions(+), 147 deletions(-)
diff --git a/METRICS.md b/METRICS.md
index 674fb9638..7fc24083e 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -74,119 +74,131 @@ Here is an example of Grafana dashboard importing.
## Details
-| MetricName | Scope |
Description
|
-|:--------------------------------------:|:-----------------:|:-----------------------------------------------------------------------------------------------------------------:|
-| RegisteredShuffleCount | master and worker |
The value means count of registered shuffle.
|
-| RunningApplicationCount | master and worker |
The value means count of running applications.
|
-| ActiveShuffleSize | master and worker | The value
means the active shuffle size for workers or a worker including master replica
and slave replica. |
-| ActiveShuffleFileCount | master and worker | The value means
the active shuffle file count for workers or a worker including master replica
and slave replica. |
-| diskFileCount | master and worker |
The count of disk files consumption by each user.
|
-| diskBytesWritten | master and worker |
The amount of disk files consumption by each user.
|
-| hdfsFileCount | master and worker |
The count of hdfs files consumption by each user.
|
-| hdfsBytesWritten | master and worker |
The amount of hdfs files consumption by each user.
|
-| WorkerCount | master |
The count of active workers.
|
-| LostWorkers | master |
The count of workers in lost list.
|
-| ExcludedWorkerCount | master |
The count of workers in excluded list.
|
-| IsActiveMaster | master |
Whether the current master is active.
|
-| PartitionSize | master | The
estimated partition size of last 20 flush window whose length is 15 seconds by
defaults. |
-| OfferSlotsTime | master |
The time of offer slots.
|
-| OpenStreamTime | worker |
OpenStream means read a shuffle file and send client about chunks size and
stream index. |
-| FetchChunkTime | worker |
FetchChunk means read a chunk from a shuffle file and send to client.
|
-| ChunkStreamCount | worker |
The stream count for reduce partition reading streams in current worker.
|
-| OpenStreamSuccessCount | worker |
The count of opening stream succeed in current worker.
|
-| OpenStreamFailCount | worker |
The count of opening stream failed in current worker.
|
-| FetchChunkSuccessCount | worker |
The count of fetching chunk succeed in current worker.
|
-| FetchChunkFailCount | worker |
The count of fetching chunk failed in current worker.
|
-| PrimaryPushDataTime | worker |
PrimaryPushData means handle PushData of primary partition location.
|
-| ReplicaPushDataTime | worker |
ReplicaPushData means handle PushData of replica partition location.
|
-| WriteDataHardSplitCount | worker | The
count of writing PushData or PushMergedData to HARD_SPLIT partition in current
worker. |
-| WriteDataSuccessCount | worker |
The count of writing PushData or PushMergedData succeed in current worker.
|
-| WriteDataFailCount | worker |
The count of writing PushData or PushMergedData failed in current worker.
|
-| ReplicateDataFailCount | worker |
The count of replicating PushData or PushMergedData failed in current worker.
|
-| ReplicateDataWriteFailCount | worker | The
count of replicating PushData or PushMergedData failed caused by write failure
in peer worker. |
-| ReplicateDataCreateConnectionFailCount | worker | The count of
replicating PushData or PushMergedData failed caused by creating connection
failed in peer worker. |
-| ReplicateDataConnectionExceptionCount | worker | The count
of replicating PushData or PushMergedData failed caused by connection exception
in peer worker. |
-| ReplicateDataTimeoutCount | worker | The
count of replicating PushData or PushMergedData failed caused by push timeout
in peer worker. |
-| PushDataHandshakeFailCount | worker |
The count of PushDataHandshake failed in current worker.
|
-| RegionStartFailCount | worker |
The count of RegionStart failed in current worker.
|
-| RegionFinishFailCount | worker |
The count of RegionFinish failed in current worker.
|
-| PrimaryPushDataHandshakeTime | worker |
PrimaryPushDataHandshake means handle PushData of primary partition location.
|
-| ReplicaPushDataHandshakeTime | worker |
ReplicaPushDataHandshake means handle PushData of replica partition location.
|
-| PrimaryRegionStartTime | worker |
PrimaryRegionStart means handle RegionStart of primary partition location.
|
-| ReplicaRegionStartTime | worker |
ReplicaRegionStart means handle RegionStart of replica partition location.
|
-| PrimaryRegionFinishTime | worker |
PrimaryRegionFinish means handle RegionFinish of primary partition location.
|
-| ReplicaRegionFinishTime | worker |
ReplicaRegionFinish means handle RegionFinish of replica partition location.
|
-| PausePushDataTime | worker |
PausePushData means stop receiving data from client.
|
-| PausePushDataAndReplicateTime | worker |
PausePushDataAndReplicate means stop receiving data from client and other
workers. |
-| PausePushData | worker |
The count of stopping receiving data from client in current worker.
|
-| PausePushDataAndReplicate | worker |
The count of stopping receiving data from client and other workers in current
worker. |
-| TakeBufferTime | worker |
TakeBuffer means get a disk buffer from disk flusher.
|
-| FlushDataTime | worker |
FlushData means flush a disk buffer to disk.
|
-| CommitFilesTime | worker |
CommitFiles means flush and close a shuffle partition file.
|
-| SlotsAllocated | worker |
Slots allocated in last hour
|
-| ReserveSlotsTime | worker |
ReserveSlots means acquire a disk buffer and record partition location.
|
-| ActiveConnectionCount | worker |
The value means count of active network connection.
|
-| NettyMemory | worker |
The value measures all kinds of transport memory used by netty.
|
-| SortTime | worker |
SortTime measures the time used by sorting a shuffle file.
|
-| SortMemory | worker |
SortMemory means total reserved memory for sorting shuffle files .
|
-| SortingFiles | worker |
This value means the count of sorting shuffle files.
|
-| SortedFiles | worker |
This value means the count of sorted shuffle files.
|
-| SortedFileSize | worker |
This value means the count of sorted shuffle files 's total size.
|
-| DiskBuffer | worker | Disk buffers
are part of netty used memory, means data need to write to disk but haven't
been written to disk. |
-| BufferStreamReadBuffer | worker |
This value means memory used by credit stream read buffer.
|
-| ReadBufferDispatcherRequestsLength | worker |
This value means the queue size of read buffer allocation requests.
|
-| ReadBufferAllocatedCount | worker |
This value means count of allocated read buffer.
|
-| CreditStreamCount | worker |
This value means count of stream for map partition reading streams.
|
-| ActiveMapPartitionCount | worker |
This value means count of active map partition reading streams.
|
-| DeviceOSFreeBytes | worker |
This value means actual usable space of OS for device monitor.
|
-| DeviceOSTotalBytes | worker |
This value means total usable space of OS for device monitor.
|
-| DeviceCelebornFreeBytes | worker |
This value means actual usable space of Celeborn for device monitor.
|
-| DeviceCelebornTotalBytes | worker |
This value means configured usable space of Celeborn for device monitor.
|
-| PotentialConsumeSpeed | worker |
This value means speed of potential consumption for congestion control.
|
-| UserProduceSpeed | worker |
This value means speed of user production for congestion control.
|
-| WorkerConsumeSpeed | worker |
This value means speed of worker consumption for congestion control.
|
-| jvm_gc_count | JVM |
The GC count of each garbage collector.
|
-| jvm_gc_time | JVM |
The GC cost time of each garbage collector.
|
-| jvm_memory_heap_init | JVM |
The amount of heap init memory.
|
-| jvm_memory_heap_max | JVM |
The amount of heap max memory.
|
-| jvm_memory_heap_used | JVM |
The amount of heap used memory.
|
-| jvm_memory_heap_committed | JVM |
The amount of heap committed memory.
|
-| jvm_memory_heap_usage | JVM |
The percentage of heap memory usage.
|
-| jvm_memory_non_heap_init | JVM |
The amount of non-heap init memory.
|
-| jvm_memory_non_heap_max | JVM |
The amount of non-heap max memory.
|
-| jvm_memory_non_heap_used | JVM |
The amount of non-heap uesd memory.
|
-| jvm_memory_non_heap_committed | JVM |
The amount of non-heap committed memory.
|
-| jvm_memory_non_heap_usage | JVM |
The percentage of non-heap memory usage.
|
-| jvm_memory_pools_init | JVM |
The amount of each memory pool's init memory.
|
-| jvm_memory_pools_max | JVM |
The amount of each memory pool's max memory.
|
-| jvm_memory_pools_used | JVM |
The amount of each memory pool's used memory.
|
-| jvm_memory_pools_committed | JVM |
The amount of each memory pool's committed memory.
|
-| jvm_memory_pools_used_after_gc | JVM |
The amount of each memory pool's used memory after GC.
|
-| jvm_memory_pools_usage | JVM |
The percentage of each memory pool's memory usage.
|
-| jvm_memory_total_init | JVM |
The amount of total init memory.
|
-| jvm_memory_total_max | JVM |
The amount of total max memory.
|
-| jvm_memory_total_used | JVM |
The amount of total used memory.
|
-| jvm_memory_total_committed | JVM |
The amount of each memory pool's committed memory.
|
-| jvm_direct_capacity | JVM |
An estimate of the total capacity of the buffers in this pool
|
-| jvm_direct_count | JVM |
An estimate of the number of buffers in the pool
|
-| jvm_direct_used | JVM |
An estimate of the memory that JVM is using for this buffer pool
|
-| jvm_mapped_capacity | JVM |
An estimate of the total capacity of the buffers in this pool
|
-| jvm_mapped_count | JVM |
An estimate of the number of buffers in the pool
|
-| jvm_mapped_used | JVM |
An estimate of the memory that JVM is using for this buffer pool
|
-| jvm_thread_count | JVM |
The current number of threads.
|
-| jvm_thread_daemon_count | JVM |
The current number of daemon threads.
|
-| jvm_thread_blocked_count | JVM |
The current number of threads having blocked state.
|
-| jvm_thread_deadlock_count | JVM |
The current number of threads having deadlock state.
|
-| jvm_thread_new_count | JVM |
The current number of threads having new state.
|
-| jvm_thread_runnable_count | JVM |
The current number of threads having runnable state.
|
-| jvm_thread_terminated_count | JVM |
The current number of threads having terminated state.
|
-| jvm_thread_timed_waiting_count | JVM |
The current number of threads having timed_waiting state.
|
-| jvm_thread_waiting_count | JVM |
The current number of threads having waiting state.
|
-| jvm_classloader_loaded | JVM |
The total number of classes loaded since the start of the JVM.
|
-| jvm_classloader_unloaded | JVM |
The total number of classes unloaded since the start of the JVM.
|
-| JVMCPUTime | system |
The JVM costs cpu time.
|
-| AvailableProcessors | system |
The amount of system available processors.
|
-| LastMinuteSystemLoad | system |
The last minute load of system.
|
+| MetricName | Scope |
Description
|
+|:--------------------------------------:|:-----------------:|:----------------------------------------------------------------------------------------------------------------------:|
+| RegisteredShuffleCount | master and worker |
The value means count of registered shuffle.
|
+| RunningApplicationCount | master and worker |
The value means count of running applications.
|
+| ActiveShuffleSize | master and worker | The value
means the active shuffle size for workers or a worker including master replica
and slave replica. |
+| ActiveShuffleFileCount | master and worker | The value
means the active shuffle file count for workers or a worker including master
replica and slave replica. |
+| diskFileCount | master and worker |
The count of disk files consumption by each user.
|
+| diskBytesWritten | master and worker |
The amount of disk files consumption by each user.
|
+| hdfsFileCount | master and worker |
The count of hdfs files consumption by each user.
|
+| hdfsBytesWritten | master and worker |
The amount of hdfs files consumption by each user.
|
+| WorkerCount | master |
The count of active workers.
|
+| LostWorkers | master |
The count of workers in lost list.
|
+| ExcludedWorkerCount | master |
The count of workers in excluded list.
|
+| IsActiveMaster | master |
Whether the current master is active.
|
+| PartitionSize | master |
The estimated partition size of last 20 flush window whose length is 15 seconds
by defaults. |
+| OfferSlotsTime | master |
The time of offer slots.
|
+| OpenStreamTime | worker |
OpenStream means read a shuffle file and send client about chunks size and
stream index. |
+| FetchChunkTime | worker |
FetchChunk means read a chunk from a shuffle file and send to client.
|
+| ChunkStreamCount | worker |
The stream count for reduce partition reading streams in current worker.
|
+| OpenStreamSuccessCount | worker |
The count of opening stream succeed in current worker.
|
+| OpenStreamFailCount | worker |
The count of opening stream failed in current worker.
|
+| FetchChunkSuccessCount | worker |
The count of fetching chunk succeed in current worker.
|
+| FetchChunkFailCount | worker |
The count of fetching chunk failed in current worker.
|
+| PrimaryPushDataTime | worker |
PrimaryPushData means handle PushData of primary partition location.
|
+| ReplicaPushDataTime | worker |
ReplicaPushData means handle PushData of replica partition location.
|
+| WriteDataHardSplitCount | worker |
The count of writing PushData or PushMergedData to HARD_SPLIT partition in
current worker. |
+| WriteDataSuccessCount | worker |
The count of writing PushData or PushMergedData succeed in current
worker. |
+| WriteDataFailCount | worker |
The count of writing PushData or PushMergedData failed in current worker.
|
+| ReplicateDataFailCount | worker |
The count of replicating PushData or PushMergedData failed in current
worker. |
+| ReplicateDataWriteFailCount | worker | The
count of replicating PushData or PushMergedData failed caused by write failure
in peer worker. |
+| ReplicateDataCreateConnectionFailCount | worker | The count of
replicating PushData or PushMergedData failed caused by creating connection
failed in peer worker. |
+| ReplicateDataConnectionExceptionCount | worker | The count
of replicating PushData or PushMergedData failed caused by connection exception
in peer worker. |
+| ReplicateDataTimeoutCount | worker | The
count of replicating PushData or PushMergedData failed caused by push timeout
in peer worker. |
+| PushDataHandshakeFailCount | worker |
The count of PushDataHandshake failed in current worker.
|
+| RegionStartFailCount | worker |
The count of RegionStart failed in current worker.
|
+| RegionFinishFailCount | worker |
The count of RegionFinish failed in current worker.
|
+| PrimaryPushDataHandshakeTime | worker |
PrimaryPushDataHandshake means handle PushData of primary partition
location. |
+| ReplicaPushDataHandshakeTime | worker |
ReplicaPushDataHandshake means handle PushData of replica partition
location. |
+| PrimaryRegionStartTime | worker |
PrimaryRegionStart means handle RegionStart of primary partition
location. |
+| ReplicaRegionStartTime | worker |
ReplicaRegionStart means handle RegionStart of replica partition
location. |
+| PrimaryRegionFinishTime | worker |
PrimaryRegionFinish means handle RegionFinish of primary partition
location. |
+| ReplicaRegionFinishTime | worker |
ReplicaRegionFinish means handle RegionFinish of replica partition
location. |
+| PausePushDataTime | worker |
PausePushData means stop receiving data from client.
|
+| PausePushDataAndReplicateTime | worker |
PausePushDataAndReplicate means stop receiving data from client and other
workers. |
+| PausePushData | worker |
The count of stopping receiving data from client in current worker.
|
+| PausePushDataAndReplicate | worker |
The count of stopping receiving data from client and other workers in current
worker. |
+| TakeBufferTime | worker |
TakeBuffer means get a disk buffer from disk flusher.
|
+| FlushDataTime | worker |
FlushData means flush a disk buffer to disk.
|
+| CommitFilesTime | worker |
CommitFiles means flush and close a shuffle partition file.
|
+| SlotsAllocated | worker |
Slots allocated in last hour
|
+| ReserveSlotsTime | worker |
ReserveSlots means acquire a disk buffer and record partition location.
|
+| ActiveConnectionCount | worker |
The value means count of active network connection.
|
+| NettyMemory | worker |
The value measures all kinds of transport memory used by netty.
|
+| SortTime | worker |
SortTime measures the time used by sorting a shuffle file.
|
+| SortMemory | worker |
SortMemory means total reserved memory for sorting shuffle files .
|
+| SortingFiles | worker |
This value means the count of sorting shuffle files.
|
+| SortedFiles | worker |
This value means the count of sorted shuffle files.
|
+| SortedFileSize | worker |
This value means the count of sorted shuffle files 's total size.
|
+| DiskBuffer | worker | Disk
buffers are part of netty used memory, means data need to write to disk but
haven't been written to disk. |
+| BufferStreamReadBuffer | worker |
This value means memory used by credit stream read buffer.
|
+| ReadBufferDispatcherRequestsLength | worker |
This value means the queue size of read buffer allocation requests.
|
+| ReadBufferAllocatedCount | worker |
This value means count of allocated read buffer.
|
+| CreditStreamCount | worker |
This value means count of stream for map partition reading streams.
|
+| ActiveMapPartitionCount | worker |
This value means count of active map partition reading streams.
|
+| DeviceOSFreeBytes | worker |
This value means actual usable space of OS for device monitor.
|
+| DeviceOSTotalBytes | worker |
This value means total usable space of OS for device monitor.
|
+| DeviceCelebornFreeBytes | worker |
This value means actual usable space of Celeborn for device monitor.
|
+| DeviceCelebornTotalBytes | worker |
This value means configured usable space of Celeborn for device monitor.
|
+| PotentialConsumeSpeed | worker |
This value means speed of potential consumption for congestion control.
|
+| UserProduceSpeed | worker |
This value means speed of user production for congestion control.
|
+| WorkerConsumeSpeed | worker |
This value means speed of worker consumption for congestion control.
|
+| jvm_gc_count | JVM |
The GC count of each garbage collector.
|
+| jvm_gc_time | JVM |
The GC cost time of each garbage collector.
|
+| jvm_memory_heap_init | JVM |
The amount of heap init memory.
|
+| jvm_memory_heap_max | JVM |
The amount of heap max memory.
|
+| jvm_memory_heap_used | JVM |
The amount of heap used memory.
|
+| jvm_memory_heap_committed | JVM |
The amount of heap committed memory.
|
+| jvm_memory_heap_usage | JVM |
The percentage of heap memory usage.
|
+| jvm_memory_non_heap_init | JVM |
The amount of non-heap init memory.
|
+| jvm_memory_non_heap_max | JVM |
The amount of non-heap max memory.
|
+| jvm_memory_non_heap_used | JVM |
The amount of non-heap uesd memory.
|
+| jvm_memory_non_heap_committed | JVM |
The amount of non-heap committed memory.
|
+| jvm_memory_non_heap_usage | JVM |
The percentage of non-heap memory usage.
|
+| jvm_memory_pools_init | JVM |
The amount of each memory pool's init memory.
|
+| jvm_memory_pools_max | JVM |
The amount of each memory pool's max memory.
|
+| jvm_memory_pools_used | JVM |
The amount of each memory pool's used memory.
|
+| jvm_memory_pools_committed | JVM |
The amount of each memory pool's committed memory.
|
+| jvm_memory_pools_used_after_gc | JVM |
The amount of each memory pool's used memory after GC.
|
+| jvm_memory_pools_usage | JVM |
The percentage of each memory pool's memory usage.
|
+| jvm_memory_total_init | JVM |
The amount of total init memory.
|
+| jvm_memory_total_max | JVM |
The amount of total max memory.
|
+| jvm_memory_total_used | JVM |
The amount of total used memory.
|
+| jvm_memory_total_committed | JVM |
The amount of each memory pool's committed memory.
|
+| jvm_direct_capacity | JVM |
An estimate of the total capacity of the buffers in this pool
|
+| jvm_direct_count | JVM |
An estimate of the number of buffers in the pool
|
+| jvm_direct_used | JVM |
An estimate of the memory that JVM is using for this buffer pool
|
+| jvm_mapped_capacity | JVM |
An estimate of the total capacity of the buffers in this pool
|
+| jvm_mapped_count | JVM |
An estimate of the number of buffers in the pool
|
+| jvm_mapped_used | JVM |
An estimate of the memory that JVM is using for this buffer pool
|
+| jvm_thread_count | JVM |
The current number of threads.
|
+| jvm_thread_daemon_count | JVM |
The current number of daemon threads.
|
+| jvm_thread_blocked_count | JVM |
The current number of threads having blocked state.
|
+| jvm_thread_deadlock_count | JVM |
The current number of threads having deadlock state.
|
+| jvm_thread_new_count | JVM |
The current number of threads having new state.
|
+| jvm_thread_runnable_count | JVM |
The current number of threads having runnable state.
|
+| jvm_thread_terminated_count | JVM |
The current number of threads having terminated state.
|
+| jvm_thread_timed_waiting_count | JVM |
The current number of threads having timed_waiting state.
|
+| jvm_thread_waiting_count | JVM |
The current number of threads having waiting state.
|
+| jvm_classloader_loaded | JVM |
The total number of classes loaded since the start of the JVM.
|
+| jvm_classloader_unloaded | JVM |
The total number of classes unloaded since the start of the JVM.
|
+| JVMCPUTime | system |
The JVM costs cpu time.
|
+| AvailableProcessors | system |
The amount of system available processors.
|
+| LastMinuteSystemLoad | system |
The last minute load of system.
|
+| active_thread_count | JVM |
The approximate number of threads that are actively executing tasks.
|
+| pending_task_count | JVM |
The pending task not executed in block queue.
|
+| pool_size | JVM |
The current number of threads in the pool.
|
+| core_pool_size | JVM |
The core number of threads.
|
+| maximum_pool_size | JVM |
The maximum allowed number of threads.
|
+| largest_pool_size | JVM |
The largest number of threads that have ever simultaneously been in the
pool. |
+| is_terminating | JVM | If this
executor is in the process of terminating after shutdown() or shutdownNow() but
has not completely terminated. |
+| is_terminated | JVM | If this
executor is in the process of terminated after shutdown() or shutdownNow() and
has completely terminated. |
+| is_shutdown | JVM |
If this executor is shutdown.
|
+| thread_count | JVM |
The thread count of current thread group.
|
+| thread_is_terminated_count | JVM |
The terminated thread count of current thread group.
|
+| thread_is_shutdown_count | JVM |
The shutdown thread count of current thread group.
|
## Implementation
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/ThreadPoolSource.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ThreadPoolSource.scala
new file mode 100644
index 000000000..944611dee
--- /dev/null
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ThreadPoolSource.scala
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.common.metrics.source
+
+import java.util.concurrent.{ExecutorService, ThreadPoolExecutor}
+
+import com.codahale.metrics.Gauge
+
+import org.apache.celeborn.common.CelebornConf
+
+class ThreadPoolSource(
+ conf: CelebornConf,
+ role: String)
+ extends AbstractSource(conf, role) {
+ override val sourceName = "ThreadPool"
+ def registerSource(
+ threadPoolName: String,
+ threadPoolExecutor: ThreadPoolExecutor): Unit = {
+ val label = Map("threadPool" -> threadPoolName)
+ addGauge(
+ "active_thread_count",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getActiveCount
+ }
+ })
+ addGauge(
+ "pending_task_count",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getQueue.size()
+ }
+ })
+ addGauge(
+ "pool_size",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getPoolSize
+ }
+ })
+ addGauge(
+ "core_pool_size",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getCorePoolSize
+ }
+ })
+ addGauge(
+ "maximum_pool_size",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getMaximumPoolSize
+ }
+ })
+ addGauge(
+ "largest_pool_size",
+ label,
+ new Gauge[Long] {
+ override def getValue: Long = {
+ threadPoolExecutor.getLargestPoolSize
+ }
+ })
+ addGauge(
+ "is_terminating",
+ label,
+ new Gauge[Boolean] {
+ override def getValue: Boolean = {
+ threadPoolExecutor.isTerminating
+ }
+ })
+ addGauge(
+ "is_terminated",
+ label,
+ new Gauge[Boolean] {
+ override def getValue: Boolean = {
+ threadPoolExecutor.isTerminated
+ }
+ })
+ addGauge(
+ "is_shutdown",
+ label,
+ new Gauge[Boolean] {
+ override def getValue: Boolean = {
+ threadPoolExecutor.isShutdown
+ }
+ })
+ }
+
+ def unregisterSource(threadPoolName: String): Unit = {
+ val label = Map("threadPool" -> threadPoolName)
+ removeGauge("active_thread_count", label)
+ removeGauge("pending_task_count", label)
+ removeGauge("pool_size", label)
+ removeGauge("core_pool_size", label)
+ removeGauge("maximum_pool_size", label)
+ removeGauge("largest_pool_size", label)
+ removeGauge("is_terminating", label)
+ removeGauge("is_terminated", label)
+ removeGauge("is_shutdown", label)
+ removeGauge("thread_count", label)
+ removeGauge("thread_is_terminated_count", label)
+ removeGauge("thread_is_shutdown_count", label)
+ }
+
+ def registerSource(threadPoolName: String, threads: Array[ExecutorService]):
Unit = {
+ val label = Map("threadPool" -> threadPoolName)
+ addGauge(
+ "thread_count",
+ label,
+ new Gauge[Int] {
+ override def getValue: Int = {
+ threads.size
+ }
+ })
+ addGauge(
+ "thread_is_terminated_count",
+ label,
+ new Gauge[Int] {
+ override def getValue: Int = {
+ threads.count(_.isTerminated)
+ }
+ })
+ addGauge(
+ "thread_is_shutdown_count",
+ label,
+ new Gauge[Int] {
+ override def getValue: Int = {
+ threads.count(_.isShutdown)
+ }
+ })
+ }
+}
+
+object ThreadPoolSource {
+ var threadPoolSource: Option[ThreadPoolSource] = None
+
+ def apply(conf: CelebornConf, role: String): ThreadPoolSource = {
+ if (threadPoolSource.isEmpty) {
+ ThreadPoolSource.getClass.synchronized {
+ if (threadPoolSource.isEmpty) {
+ threadPoolSource = Option(new ThreadPoolSource(conf, role))
+ }
+ }
+ }
+ threadPoolSource.get
+ }
+
+ def registerSource(threadPoolName: String, threadPoolExecutor:
ThreadPoolExecutor): Unit = {
+ threadPoolSource.foreach(_.registerSource(threadPoolName,
threadPoolExecutor))
+ }
+
+ def registerSource(threadPoolName: String, threads: Array[ExecutorService]):
Unit = {
+ threadPoolSource.foreach(_.registerSource(threadPoolName, threads))
+ }
+
+ def unregisterSource(threadPoolName: String): Unit = {
+ threadPoolSource.foreach(_.unregisterSource(threadPoolName))
+ }
+}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
index 293e113cb..7c13f2a3e 100644
---
a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/NettyRpcEnv.scala
@@ -69,7 +69,7 @@ class NettyRpcEnv(
// Because TransportClientFactory.createClient is blocking, we need to run
it in this thread pool
// to implement non-blocking send/ask.
private[celeborn] val clientConnectionExecutor =
ThreadUtils.newDaemonCachedThreadPool(
- "netty-rpc-connection",
+ "celeborn-netty-rpc-connection-executor",
celebornConf.rpcConnectThreads)
@volatile private var server: TransportServer = _
diff --git
a/common/src/main/scala/org/apache/celeborn/common/util/ThreadUtils.scala
b/common/src/main/scala/org/apache/celeborn/common/util/ThreadUtils.scala
index 233bacd0b..fe6605fb1 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/ThreadUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/ThreadUtils.scala
@@ -30,6 +30,7 @@ import com.google.common.util.concurrent.{MoreExecutors,
ThreadFactoryBuilder}
import org.apache.celeborn.common.exception.CelebornException
import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.metrics.source.ThreadPoolSource
object ThreadUtils {
@@ -145,7 +146,10 @@ object ThreadUtils {
* unique, sequentially assigned integer.
*/
def newDaemonCachedThreadPool(prefix: String): ThreadPoolExecutor = {
-
Executors.newCachedThreadPool(namedThreadFactory(prefix)).asInstanceOf[ThreadPoolExecutor]
+ val threadPool =
+
Executors.newCachedThreadPool(namedThreadFactory(prefix)).asInstanceOf[ThreadPoolExecutor]
+ ThreadPoolSource.registerSource(prefix, threadPool)
+ threadPool
}
/**
@@ -164,6 +168,7 @@ object ThreadUtils {
new LinkedBlockingQueue[Runnable],
namedThreadFactory(prefix))
threadPool.allowCoreThreadTimeOut(true)
+ ThreadPoolSource.registerSource(prefix, threadPool)
threadPool
}
@@ -172,8 +177,10 @@ object ThreadUtils {
* unique, sequentially assigned integer.
*/
def newDaemonFixedThreadPool(nThreads: Int, prefix: String):
ThreadPoolExecutor = {
- Executors.newFixedThreadPool(nThreads, namedThreadFactory(prefix))
+ val threadPool = Executors.newFixedThreadPool(nThreads,
namedThreadFactory(prefix))
.asInstanceOf[ThreadPoolExecutor]
+ ThreadPoolSource.registerSource(prefix, threadPool)
+ threadPool
}
/**
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 4e9cf5e6b..89dffbef0 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -127,6 +127,19 @@ These metrics are exposed by Celeborn master.
- hdfsFileCount
- hdfsBytesWritten
+ - namespace=ThreadPool
+ - **notes:**
+ - This metrics data is generated for each thread pool and they are
identified using a metric tag by thread pool name.
+ - active_thread_count
+ - pending_task_count
+ - pool_size
+ - core_pool_size
+ - maximum_pool_size
+ - largest_pool_size
+ - is_terminating
+ - is_terminated
+ - is_shutdown
+
#### Worker
These metrics are exposed by Celeborn worker.
@@ -288,6 +301,19 @@ These metrics are exposed by Celeborn worker.
- hdfsFileCount
- hdfsBytesWritten
+ - namespace=ThreadPool
+ - **notes:**
+ - This metrics data is generated for each thread pool and they are
identified using a metric tag by thread pool name.
+ - active_thread_count
+ - pending_task_count
+ - pool_size
+ - core_pool_size
+ - maximum_pool_size
+ - largest_pool_size
+ - is_terminating
+ - is_terminated
+ - is_shutdown
+
**Note:**
The Netty DirectArenaMetrics named like `push/fetch/replicate_server_numXX`
are not exposed by default, nor in Grafana dashboard.
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index fed3c8a7a..4a0df0a1b 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -36,12 +36,13 @@ import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DiskInfo, WorkerInfo}
import org.apache.celeborn.common.metrics.MetricsSystem
-import org.apache.celeborn.common.metrics.source.{JVMCPUSource, JVMSource,
ResourceConsumptionSource, SystemMiscSource}
+import org.apache.celeborn.common.metrics.source.{JVMCPUSource, JVMSource,
ResourceConsumptionSource, SystemMiscSource, ThreadPoolSource}
import org.apache.celeborn.common.protocol._
import org.apache.celeborn.common.protocol.message.{ControlMessages,
StatusCode}
import org.apache.celeborn.common.protocol.message.ControlMessages._
import org.apache.celeborn.common.quota.{QuotaManager, ResourceConsumption}
import org.apache.celeborn.common.rpc._
+import org.apache.celeborn.common.rpc.netty.NettyRpcEnv
import org.apache.celeborn.common.util.{CelebornHadoopUtils, JavaUtils,
PbSerDeUtils, ThreadUtils, Utils}
import org.apache.celeborn.server.common.{HttpService, Service}
import
org.apache.celeborn.service.deploy.master.clustermeta.SingleMasterMetaManager
@@ -58,6 +59,17 @@ private[celeborn] class Master(
override val metricsSystem: MetricsSystem =
MetricsSystem.createMetricsSystem(serviceName, conf)
+ // init and register master metrics
+ private val resourceConsumptionSource =
+ new ResourceConsumptionSource(conf, MetricsSystem.ROLE_MASTER)
+ private val threadPoolSource = ThreadPoolSource(conf,
MetricsSystem.ROLE_MASTER)
+ private val masterSource = new MasterSource(conf)
+ metricsSystem.registerSource(resourceConsumptionSource)
+ metricsSystem.registerSource(masterSource)
+ metricsSystem.registerSource(threadPoolSource)
+ metricsSystem.registerSource(new JVMSource(conf, MetricsSystem.ROLE_MASTER))
+ metricsSystem.registerSource(new JVMCPUSource(conf,
MetricsSystem.ROLE_MASTER))
+ metricsSystem.registerSource(new SystemMiscSource(conf,
MetricsSystem.ROLE_MASTER))
override val rpcEnv: RpcEnv = RpcEnv.create(
RpcNameConstants.MASTER_SYS,
@@ -151,10 +163,6 @@ private[celeborn] class Master(
TimeUnit.MILLISECONDS)
private val slotsAssignPolicy = conf.masterSlotAssignPolicy
- // init and register master metrics
- private val resourceConsumptionSource =
- new ResourceConsumptionSource(conf, MetricsSystem.ROLE_MASTER)
- private val masterSource = new MasterSource(conf)
private var hadoopFs: FileSystem = _
masterSource.addGauge(MasterSource.REGISTERED_SHUFFLE_COUNT) { () =>
statusSystem.registeredShuffle.size
@@ -190,12 +198,6 @@ private[celeborn] class Master(
}
masterSource.addGauge(MasterSource.IS_ACTIVE_MASTER) { () => isMasterActive }
- metricsSystem.registerSource(resourceConsumptionSource)
- metricsSystem.registerSource(masterSource)
- metricsSystem.registerSource(new JVMSource(conf, MetricsSystem.ROLE_MASTER))
- metricsSystem.registerSource(new JVMCPUSource(conf,
MetricsSystem.ROLE_MASTER))
- metricsSystem.registerSource(new SystemMiscSource(conf,
MetricsSystem.ROLE_MASTER))
-
rpcEnv.setupEndpoint(RpcNameConstants.MASTER_EP, this)
// start threads to check timeout for workers and applications
diff --git
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
index dd5a49733..d2a6388a8 100644
---
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
+++
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java
@@ -127,7 +127,7 @@ public class PartitionFilesSorter extends
ShuffleRecoverHelper {
fileSorterExecutors =
ThreadUtils.newDaemonCachedThreadPool(
- "worker-file-sorter-execute", conf.partitionSorterThreads(), 120);
+ "worker-file-sorter-executor", conf.partitionSorterThreads(), 120);
indexCache =
CacheBuilder.newBuilder()
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 b237f21ee..cca324445 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
@@ -37,7 +37,7 @@ import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DiskInfo, WorkerInfo,
WorkerPartitionLocationInfo}
import org.apache.celeborn.common.metrics.MetricsSystem
-import org.apache.celeborn.common.metrics.source.{JVMCPUSource, JVMSource,
ResourceConsumptionSource, SystemMiscSource}
+import org.apache.celeborn.common.metrics.source.{JVMCPUSource, JVMSource,
ResourceConsumptionSource, SystemMiscSource, ThreadPoolSource}
import org.apache.celeborn.common.network.TransportContext
import org.apache.celeborn.common.protocol.{PartitionType,
PbRegisterWorkerResponse, PbWorkerLostResponse, RpcNameConstants,
TransportModuleConstants}
import org.apache.celeborn.common.protocol.message.ControlMessages._
@@ -65,6 +65,16 @@ private[celeborn] class Worker(
override val metricsSystem: MetricsSystem =
MetricsSystem.createMetricsSystem(serviceName, conf)
+ val workerSource = new WorkerSource(conf)
+ private val resourceConsumptionSource =
+ new ResourceConsumptionSource(conf, MetricsSystem.ROLE_WORKER)
+ private val threadPoolSource = ThreadPoolSource(conf,
MetricsSystem.ROLE_WORKER)
+ metricsSystem.registerSource(workerSource)
+ metricsSystem.registerSource(threadPoolSource)
+ metricsSystem.registerSource(resourceConsumptionSource)
+ metricsSystem.registerSource(new JVMSource(conf, MetricsSystem.ROLE_WORKER))
+ metricsSystem.registerSource(new JVMCPUSource(conf,
MetricsSystem.ROLE_WORKER))
+ metricsSystem.registerSource(new SystemMiscSource(conf,
MetricsSystem.ROLE_WORKER))
val rpcEnv: RpcEnv = RpcEnv.create(
RpcNameConstants.WORKER_SYS,
@@ -106,15 +116,6 @@ private[celeborn] class Worker(
}
}
- private val resourceConsumptionSource =
- new ResourceConsumptionSource(conf, MetricsSystem.ROLE_WORKER)
- val workerSource = new WorkerSource(conf)
- metricsSystem.registerSource(resourceConsumptionSource)
- metricsSystem.registerSource(workerSource)
- metricsSystem.registerSource(new JVMSource(conf, MetricsSystem.ROLE_WORKER))
- metricsSystem.registerSource(new JVMCPUSource(conf,
MetricsSystem.ROLE_WORKER))
- metricsSystem.registerSource(new SystemMiscSource(conf,
MetricsSystem.ROLE_WORKER))
-
val storageManager = new StorageManager(conf, workerSource)
val memoryManager: MemoryManager = MemoryManager.initialize(conf)
@@ -249,12 +250,12 @@ private[celeborn] class Worker(
private var checkFastFailTask: ScheduledFuture[_] = _
val replicateThreadPool: ThreadPoolExecutor =
- ThreadUtils.newDaemonCachedThreadPool("worker-replicate-data",
conf.workerReplicateThreads)
+ ThreadUtils.newDaemonCachedThreadPool("worker-data-replicator",
conf.workerReplicateThreads)
val commitThreadPool: ThreadPoolExecutor =
- ThreadUtils.newDaemonCachedThreadPool("worker-commit-files",
conf.workerCommitThreads)
+ ThreadUtils.newDaemonCachedThreadPool("worker-files-committer",
conf.workerCommitThreads)
val cleanThreadPool: ThreadPoolExecutor =
ThreadUtils.newDaemonCachedThreadPool(
- "worker-clean-expired-shuffle-keys",
+ "worker-expired-shuffle-cleaner",
conf.workerCleanThreads)
val asyncReplyPool: ScheduledExecutorService =
ThreadUtils.newDaemonSingleThreadScheduledExecutor("worker-rpc-async-replier")
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 ad961c0b0..b480b5175 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
@@ -208,7 +208,7 @@ object DeviceMonitor extends Logging {
workerSource: AbstractSource): DeviceMonitor = {
try {
if (conf.workerDiskMonitorEnabled) {
- deviceCheckThreadPool =
ThreadUtils.newDaemonCachedThreadPool("device-check-thread", 5)
+ deviceCheckThreadPool =
ThreadUtils.newDaemonCachedThreadPool("worker-device-checker", 5)
val monitor =
new LocalDeviceMonitor(conf, deviceObserver, deviceInfos, diskInfos,
workerSource)
monitor.init()
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
index caa758ae8..c97b6defc 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/Flusher.scala
@@ -29,7 +29,7 @@ import io.netty.buffer.{CompositeByteBuf,
PooledByteBufAllocator, Unpooled}
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DiskStatus, TimeWindow}
-import org.apache.celeborn.common.metrics.source.AbstractSource
+import org.apache.celeborn.common.metrics.source.{AbstractSource,
ThreadPoolSource}
import org.apache.celeborn.common.protocol.StorageInfo
import org.apache.celeborn.common.util.ThreadUtils
import org.apache.celeborn.service.deploy.worker.WorkerSource
@@ -90,6 +90,7 @@ abstract private[worker] class Flusher(
}
})
}
+ ThreadPoolSource.registerSource(s"$this", workers)
}
def getWorkerIndex: Int = synchronized {
@@ -172,10 +173,10 @@ final private[worker] class HdfsFlusher(
allocator,
maxComponents,
null) with Logging {
- override def toString: String = s"HdfsFlusher@$flusherId"
override def processIOException(e: IOException, deviceErrorType:
DiskStatus): Unit = {
logError(s"$this write failed, reason $deviceErrorType ,exception: $e")
}
+ override def toString: String = s"HdfsFlusher@$flusherId"
}
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 b12fc1874..79b03c709 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
@@ -38,7 +38,7 @@ import org.apache.celeborn.common.exception.CelebornException
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
import org.apache.celeborn.common.meta.{DeviceInfo, DiskInfo, DiskStatus,
FileInfo, TimeWindow}
-import org.apache.celeborn.common.metrics.source.AbstractSource
+import org.apache.celeborn.common.metrics.source.{AbstractSource,
ThreadPoolSource}
import org.apache.celeborn.common.network.util.{NettyUtils, TransportConf}
import org.apache.celeborn.common.protocol.{PartitionLocation,
PartitionSplitMode, PartitionType, StorageInfo}
import org.apache.celeborn.common.quota.ResourceConsumption
@@ -94,7 +94,7 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
cleaners.put(
diskInfo.mountPoint,
ThreadUtils.newDaemonCachedThreadPool(
- s"disk-cleaner-${diskInfo.mountPoint}",
+ s"worker-disk-${diskInfo.mountPoint}-cleaner",
conf.workerDiskCleanThreads))
}
cleaners
@@ -165,6 +165,7 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
if (diskStatus == DiskStatus.CRITICAL_ERROR) {
logInfo(s"Disk $mountPoint faces critical error, will remove its disk
operator.")
val operator = diskOperators.remove(mountPoint)
+ ThreadPoolSource.unregisterSource(s"worker-disk-$mountPoint-cleaner")
if (operator != null) {
operator.shutdown()
}
@@ -176,7 +177,7 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
diskOperators.put(
mountPoint,
ThreadUtils.newDaemonCachedThreadPool(
- s"disk-cleaner-$mountPoint",
+ s"worker-disk-$mountPoint-cleaner",
conf.workerDiskCleanThreads))
}
}
