You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2024/01/26 10:17:47 UTC
(incubator-celeborn) branch branch-0.4 updated: [CELEBORN-1236][METRICS] Celeborn add metrics about thread pool
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 <an...@gmail.com>
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 <an...@gmail.com>
Signed-off-by: Angerszhuuuu <an...@gmail.com>
(cherry picked from commit 3ffed66c409340e866012c823093dd6ee294238a)
Signed-off-by: Angerszhuuuu <an...@gmail.com>
---
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))
}
}