Merge pull request #10449 from azat/metrics-for-new-bg-schedule-pools

Add tasks/memory metrics for distributed/buffer schedule pools

src/Common/CurrentMetrics.cpp

@@ -12,6 +12,8 @@
     M(BackgroundPoolTask, "Number of active tasks in BackgroundProcessingPool (merges, mutations, fetches, or replication queue bookkeeping)") \
     M(BackgroundMovePoolTask, "Number of active tasks in BackgroundProcessingPool for moves") \
     M(BackgroundSchedulePoolTask, "Number of active tasks in BackgroundSchedulePool. This pool is used for periodic ReplicatedMergeTree tasks, like cleaning old data parts, altering data parts, replica re-initialization, etc.") \
+    M(BackgroundBufferFlushSchedulePoolTask, "Number of active tasks in BackgroundBufferFlushSchedulePool. This pool is used for periodic Buffer flushes") \
+    M(BackgroundDistributedSchedulePoolTask, "Number of active tasks in BackgroundDistributedSchedulePool. This pool is used for distributed sends that is done in background.") \
     M(CacheDictionaryUpdateQueueBatches, "Number of 'batches' (a set of keys) in update queue in CacheDictionaries.") \
     M(CacheDictionaryUpdateQueueKeys, "Exact number of keys in update queue in CacheDictionaries.") \
     M(DiskSpaceReservedForMerge, "Disk space reserved for currently running background merges. It is slightly more than the total size of currently merging parts.") \
@@ -34,6 +36,8 @@
     M(MemoryTrackingInBackgroundProcessingPool, "Total amount of memory (bytes) allocated in background processing pool (that is dedicated for backround merges, mutations and fetches). Note that this value may include a drift when the memory was allocated in a context of background processing pool and freed in other context or vice-versa. This happens naturally due to caches for tables indexes and doesn't indicate memory leaks.") \
     M(MemoryTrackingInBackgroundMoveProcessingPool, "Total amount of memory (bytes) allocated in background processing pool (that is dedicated for backround moves). Note that this value may include a drift when the memory was allocated in a context of background processing pool and freed in other context or vice-versa. This happens naturally due to caches for tables indexes and doesn't indicate memory leaks.") \
     M(MemoryTrackingInBackgroundSchedulePool, "Total amount of memory (bytes) allocated in background schedule pool (that is dedicated for bookkeeping tasks of Replicated tables).") \
+    M(MemoryTrackingInBackgroundBufferFlushSchedulePool, "Total amount of memory (bytes) allocated in background buffer flushes pool (that is dedicated for background buffer flushes).") \
+    M(MemoryTrackingInBackgroundDistributedSchedulePool, "Total amount of memory (bytes) allocated in background distributed schedule pool (that is dedicated for distributed sends).") \
     M(MemoryTrackingForMerges, "Total amount of memory (bytes) allocated for background merges. Included in MemoryTrackingInBackgroundProcessingPool. Note that this value may include a drift when the memory was allocated in a context of background processing pool and freed in other context or vice-versa. This happens naturally due to caches for tables indexes and doesn't indicate memory leaks.") \
     M(LeaderElection, "Number of Replicas participating in leader election. Equals to total number of replicas in usual cases.") \
     M(EphemeralNode, "Number of ephemeral nodes hold in ZooKeeper.") \

src/Core/BackgroundSchedulePool.cpp

@@ -1,6 +1,5 @@
 #include "BackgroundSchedulePool.h"
 #include <Common/MemoryTracker.h>
-#include <Common/CurrentMetrics.h>
 #include <Common/Exception.h>
 #include <Common/setThreadName.h>
 #include <Common/Stopwatch.h>
@@ -10,12 +9,6 @@
 #include <ext/scope_guard.h>
 
 
-namespace CurrentMetrics
-{
-    extern const Metric BackgroundSchedulePoolTask;
-    extern const Metric MemoryTrackingInBackgroundSchedulePool;
-}
-
 namespace DB
 {
 
@@ -95,7 +88,7 @@ bool BackgroundSchedulePoolTaskInfo::activateAndSchedule()
 void BackgroundSchedulePoolTaskInfo::execute()
 {
     Stopwatch watch;
-    CurrentMetrics::Increment metric_increment{CurrentMetrics::BackgroundSchedulePoolTask};
+    CurrentMetrics::Increment metric_increment{pool.tasks_metric};
 
     std::lock_guard lock_exec(exec_mutex);
 
@@ -155,10 +148,13 @@ Coordination::WatchCallback BackgroundSchedulePoolTaskInfo::getWatchCallback()
 }
 
 
-BackgroundSchedulePool::BackgroundSchedulePool(size_t size_)
+BackgroundSchedulePool::BackgroundSchedulePool(size_t size_, CurrentMetrics::Metric tasks_metric_, CurrentMetrics::Metric memory_metric_, const char *thread_name_)
     : size(size_)
+    , tasks_metric(tasks_metric_)
+    , memory_metric(memory_metric_)
+    , thread_name(thread_name_)
 {
-    LOG_INFO(&Logger::get("BackgroundSchedulePool"), "Create BackgroundSchedulePool with " << size << " threads");
+    LOG_INFO(&Logger::get("BackgroundSchedulePool/" + thread_name), "Create BackgroundSchedulePool with " << size << " threads");
 
     threads.resize(size);
     for (auto & thread : threads)
@@ -181,7 +177,7 @@ BackgroundSchedulePool::~BackgroundSchedulePool()
         queue.wakeUpAll();
         delayed_thread.join();
 
-        LOG_TRACE(&Logger::get("BackgroundSchedulePool"), "Waiting for threads to finish.");
+        LOG_TRACE(&Logger::get("BackgroundSchedulePool/" + thread_name), "Waiting for threads to finish.");
         for (auto & thread : threads)
             thread.join();
     }
@@ -247,12 +243,12 @@ void BackgroundSchedulePool::attachToThreadGroup()
 
 void BackgroundSchedulePool::threadFunction()
 {
-    setThreadName("BackgrSchedPool");
+    setThreadName(thread_name.c_str());
 
     attachToThreadGroup();
     SCOPE_EXIT({ CurrentThread::detachQueryIfNotDetached(); });
     if (auto * memory_tracker = CurrentThread::getMemoryTracker())
-        memory_tracker->setMetric(CurrentMetrics::MemoryTrackingInBackgroundSchedulePool);
+        memory_tracker->setMetric(memory_metric);
 
     while (!shutdown)
     {
@@ -267,7 +263,7 @@ void BackgroundSchedulePool::threadFunction()
 
 void BackgroundSchedulePool::delayExecutionThreadFunction()
 {
-    setThreadName("BckSchPoolDelay");
+    setThreadName((thread_name + "/D").c_str());
 
     attachToThreadGroup();
     SCOPE_EXIT({ CurrentThread::detachQueryIfNotDetached(); });

src/Core/BackgroundSchedulePool.h

@@ -12,6 +12,7 @@
 #include <functional>
 #include <boost/noncopyable.hpp>
 #include <Common/ZooKeeper/Types.h>
+#include <Common/CurrentMetrics.h>
 #include <Common/CurrentThread.h>
 #include <Common/ThreadPool.h>
 
@@ -49,7 +50,8 @@ public:
 
     size_t getNumberOfThreads() const { return size; }
 
-    BackgroundSchedulePool(size_t size_);
+    /// thread_name_ cannot be longer then 13 bytes (2 bytes is reserved for "/D" suffix for delayExecutionThreadFunction())
+    BackgroundSchedulePool(size_t size_, CurrentMetrics::Metric tasks_metric_, CurrentMetrics::Metric memory_metric_, const char *thread_name_);
     ~BackgroundSchedulePool();
 
 private:
@@ -82,6 +84,10 @@ private:
     /// Thread group used for profiling purposes
     ThreadGroupStatusPtr thread_group;
 
+    CurrentMetrics::Metric tasks_metric;
+    CurrentMetrics::Metric memory_metric;
+    std::string thread_name;
+
     void attachToThreadGroup();
 };
 

src/Interpreters/Context.cpp

@@ -68,9 +68,17 @@ namespace ProfileEvents
 namespace CurrentMetrics
 {
     extern const Metric ContextLockWait;
-    extern const Metric MemoryTrackingForMerges;
     extern const Metric BackgroundMovePoolTask;
     extern const Metric MemoryTrackingInBackgroundMoveProcessingPool;
+
+    extern const Metric BackgroundSchedulePoolTask;
+    extern const Metric MemoryTrackingInBackgroundSchedulePool;
+
+    extern const Metric BackgroundBufferFlushSchedulePoolTask;
+    extern const Metric MemoryTrackingInBackgroundBufferFlushSchedulePool;
+
+    extern const Metric BackgroundDistributedSchedulePoolTask;
+    extern const Metric MemoryTrackingInBackgroundDistributedSchedulePool;
 }
 
 
@@ -1352,7 +1360,11 @@ BackgroundSchedulePool & Context::getBufferFlushSchedulePool()
 {
     auto lock = getLock();
     if (!shared->buffer_flush_schedule_pool)
-        shared->buffer_flush_schedule_pool.emplace(settings.background_buffer_flush_schedule_pool_size);
+        shared->buffer_flush_schedule_pool.emplace(
+            settings.background_buffer_flush_schedule_pool_size,
+            CurrentMetrics::BackgroundBufferFlushSchedulePoolTask,
+            CurrentMetrics::MemoryTrackingInBackgroundBufferFlushSchedulePool,
+            "BgBufSchPool");
     return *shared->buffer_flush_schedule_pool;
 }
 
@@ -1360,7 +1372,11 @@ BackgroundSchedulePool & Context::getSchedulePool()
 {
     auto lock = getLock();
     if (!shared->schedule_pool)
-        shared->schedule_pool.emplace(settings.background_schedule_pool_size);
+        shared->schedule_pool.emplace(
+            settings.background_schedule_pool_size,
+            CurrentMetrics::BackgroundSchedulePoolTask,
+            CurrentMetrics::MemoryTrackingInBackgroundSchedulePool,
+            "BgSchPool");
     return *shared->schedule_pool;
 }
 
@@ -1368,7 +1384,11 @@ BackgroundSchedulePool & Context::getDistributedSchedulePool()
 {
     auto lock = getLock();
     if (!shared->distributed_schedule_pool)
-        shared->distributed_schedule_pool.emplace(settings.background_distributed_schedule_pool_size);
+        shared->distributed_schedule_pool.emplace(
+            settings.background_distributed_schedule_pool_size,
+            CurrentMetrics::BackgroundDistributedSchedulePoolTask,
+            CurrentMetrics::MemoryTrackingInBackgroundDistributedSchedulePool,
+            "BgDistSchPool");
     return *shared->distributed_schedule_pool;
 }