Merge pull request ClickHouse#87777 from ClickHouse/backport/25.8/87584

Backport ClickHouse#87584 to 25.8: Avoid leaking of tracked memory for async logging and text_log

Author: azat

docs/en/operations/system-tables/trace_log.md

@@ -37,6 +37,8 @@ Columns:
   - `MemorySample` represents collecting random allocations and deallocations.
   - `MemoryPeak` represents collecting updates of peak memory usage.
   - `ProfileEvent` represents collecting of increments of profile events.
+  - `JemallocSample` represents collecting of jemalloc samples.
+  - `MemoryAllocatedWithoutCheck` represents collection of significant allocations (>16MiB) that is done with ignoring any memory limits (for ClickHouse developers only).
 - `thread_id` ([UInt64](../../sql-reference/data-types/int-uint.md)) — Thread identifier.
 - `query_id` ([String](../../sql-reference/data-types/string.md)) — Query identifier that can be used to get details about a query that was running from the [query_log](/operations/system-tables/query_log) system table.
 - `trace` ([Array(UInt64)](../../sql-reference/data-types/array.md)) — Stack trace at the moment of sampling. Each element is a virtual memory address inside ClickHouse server process.

src/Common/MemoryTracker.cpp

@@ -252,8 +252,9 @@ AllocationTrace MemoryTracker::allocImpl(Int64 size, bool throw_if_memory_exceed
         if (level == VariableContext::Global)
         {
             /// For global memory tracker always update memory usage.
-            amount.fetch_add(size, std::memory_order_relaxed);
+            Int64 will_be = amount.fetch_add(size, std::memory_order_relaxed);
             rss.fetch_add(size, std::memory_order_relaxed);
+            updatePeak(will_be, /*log_memory_usage=*/ false);
 
             auto metric_loaded = metric.load(std::memory_order_relaxed);
             if (metric_loaded != CurrentMetrics::end())

src/Common/MemoryTrackerDebugBlockerInThread.h

@@ -2,6 +2,7 @@
 
 #include <base/defines.h> /// DEBUG_OR_SANITIZER_BUILD
 
+/// Use it if you need to suppress MemoryAllocatedWithoutCheck for known big allocations.
 #ifdef DEBUG_OR_SANITIZER_BUILD
 struct MemoryTrackerDebugBlockerInThread
 {

src/Common/SystemLogBase.cpp

@@ -30,6 +30,7 @@
 #include <Interpreters/DeadLetterQueue.h>
 #include <IO/S3/BlobStorageLogWriter.h>
 
+#include <Common/MemoryTrackerDebugBlockerInThread.h>
 #include <Common/MemoryTrackerBlockerInThread.h>
 #include <Common/SystemLogBase.h>
 #include <Common/ThreadPool.h>
@@ -65,7 +66,7 @@ SystemLogQueue<LogElement>::SystemLogQueue(const SystemLogQueueSettings & settin
 static thread_local bool recursive_push_call = false;
 
 template <typename LogElement>
-void SystemLogQueue<LogElement>::push(LogElement&& element)
+void SystemLogQueue<LogElement>::push(LogElement && element)
 {
     /// It is possible that the method will be called recursively.
     /// Better to drop these events to avoid complications.
@@ -74,10 +75,11 @@ void SystemLogQueue<LogElement>::push(LogElement&& element)
     recursive_push_call = true;
     SCOPE_EXIT({ recursive_push_call = false; });
 
-    /// Memory can be allocated while resizing on queue.push_back.
-    /// The size of allocation can be in order of a few megabytes.
-    /// But this should not be accounted for query memory usage.
-    /// Otherwise the tests like 01017_uniqCombined_memory_usage.sql will be flaky.
+
+    /// Queue resize can allocate memory
+    /// - MemoryTrackerDebugBlockerInThread here due to the allocation can hit the limit for MemoryAllocatedWithoutCheck, let's suppress it.
+    /// - MemoryTrackerBlockerInThread here because this allocation should not be take into account in the query scope (since it will be freed outside of it)
+    [[maybe_unused]] MemoryTrackerDebugBlockerInThread blocker;
     MemoryTrackerBlockerInThread temporarily_disable_memory_tracker;
 
     /// Should not log messages under mutex.
@@ -308,6 +310,8 @@ void SystemLogBase<LogElement>::stopFlushThread()
 template <typename LogElement>
 void SystemLogBase<LogElement>::add(LogElement element)
 {
+    /// This allocation should not be take into account in the query scope (since it will be freed outside of it)
+    MemoryTrackerBlockerInThread temporarily_disable_memory_tracker;
     queue->push(std::move(element));
 }
 

src/Common/logger_useful.h

@@ -9,6 +9,7 @@
 #include <Common/AtomicLogger.h>
 #include <Common/CurrentThreadHelpers.h>
 #include <Common/Logger.h>
+#include <Common/MemoryTrackerBlockerInThread.h>
 #include <Common/LoggingFormatStringHelpers.h>
 #include <Common/ProfileEvents.h>
 #include <Common/Stopwatch.h>
@@ -73,6 +74,7 @@ constexpr bool constexprContains(std::string_view haystack, std::string_view nee
 {                                                                                                                   \
     static_assert(!constexprContains(#__VA_ARGS__, "formatWithSecretsOneLine"), "Think twice!");                    \
     static_assert(!constexprContains(#__VA_ARGS__, "formatWithSecretsMultiLine"), "Think twice!");                  \
+                                                                                                                    \
     auto _logger = ::impl::getLoggerHelper(logger);                                                                 \
     const bool _is_clients_log = DB::currentThreadHasGroup() && DB::currentThreadLogsLevel() >= (priority);         \
     if (!_is_clients_log && !_logger->is((PRIORITY)))                                                               \
@@ -87,6 +89,11 @@ constexpr bool constexprContains(std::string_view haystack, std::string_view nee
     constexpr size_t _nargs = CH_VA_ARGS_NARGS(__VA_ARGS__);                                                        \
     using LogTypeInfo = FormatStringTypeInfo<std::decay_t<decltype(LOG_IMPL_FIRST_ARG(__VA_ARGS__))>>;              \
                                                                                                                     \
+    /* Note, we need to block memory tracking to avoid taking into account this memory in the query context */      \
+    /* Since this memory will be freed either in OwnAsyncSplitChannel::runChannel() (in case of async logging) */   \
+    /* Or in a system thread for flushing system.text_log (in case of sync logging) */                              \
+    MemoryTrackerBlockerInThread block_memory_tracker(VariableContext::Global);                                     \
+                                                                                                                    \
     std::string_view _format_string;                                                                                \
     std::string _formatted_message;                                                                                 \
     std::vector<std::string> _format_string_args;                                                                   \

src/Loggers/OwnSplitChannel.cpp

@@ -138,7 +138,6 @@ void logToSystemTextLogQueue(
 void OwnSplitChannel::logSplit(
     const ExtendedLogMessage & msg_ext, const std::shared_ptr<InternalTextLogsQueue> & logs_queue, const std::string & msg_thread_name)
 {
-    LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global);
     const Poco::Message & msg = *msg_ext.base;
 
     try
@@ -393,7 +392,6 @@ void OwnAsyncSplitChannel::log(const Poco::Message & msg)
 
 void OwnAsyncSplitChannel::log(Poco::Message && msg)
 {
-    LockMemoryExceptionInThread lock_memory_tracker(VariableContext::Global);
     try
     {
         /// Based on logger_useful.h this won't be called if the message is not needed

src/Loggers/OwnSplitChannel.h

@@ -33,6 +33,8 @@ using TextLogQueue = SystemLogQueue<TextLogElement>;
 using AsyncLogQueueSize = std::pair<std::string, size_t>;
 using AsyncLogQueueSizes = std::vector<AsyncLogQueueSize>;
 
+class ExtendedLogMessage;
+
 class OwnSplitChannelBase : public Poco::Channel
 {
 public:

tests/integration/test_keeper_memory_soft_limit/test.py

@@ -1,7 +1,4 @@
 #!/usr/bin/env python3
-import random
-import string
-
 import pytest
 from kazoo.client import KazooClient
 from helpers.cluster import ClickHouseCluster
@@ -18,10 +15,6 @@
 )
 
 
-def random_string(length):
-    return "".join(random.choices(string.ascii_lowercase + string.digits, k=length))
-
-
 def get_connection_zk(nodename, timeout=30.0):
     # NOTE: here we need KazooClient without implicit retries! (KazooClientWithImplicitRetries)
     _fake_zk_instance = KazooClient(
@@ -64,7 +57,8 @@ def test_soft_limit_create(started_cluster):
 
         txn.create("/test_soft_limit/node_1000001" + str(i), b"abcde")
         txn.commit()
-        assert "0\n"  == node.query("select sum(ProfileEvent_ZooKeeperHardwareExceptions) from system.metric_log")
+        node.query("system flush logs metric_log")
+        assert int(node.query("select sum(ProfileEvent_ZooKeeperHardwareExceptions) from system.metric_log").strip()) > 0
         return
 
     raise Exception("all records are inserted but no error occurs")