Merge pull request #59945 from ClickHouse/keeper-prometheus-filtered-events

Send only Keeper related metrics/events for Prometheus

programs/keeper/CMakeLists.txt

@@ -39,6 +39,7 @@ if (BUILD_STANDALONE_KEEPER)
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/KeeperContext.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/KeeperStateManager.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/KeeperStorage.cpp
+        ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/KeeperConstants.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/KeeperAsynchronousMetrics.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/pathUtils.cpp
         ${CMAKE_CURRENT_SOURCE_DIR}/../../src/Coordination/SessionExpiryQueue.cpp

programs/keeper/Keeper.cpp

@@ -32,9 +32,10 @@
 #include <Coordination/KeeperAsynchronousMetrics.h>
 
 #include <Server/HTTP/HTTPServer.h>
-#include <Server/TCPServer.h>
 #include <Server/HTTPHandlerFactory.h>
 #include <Server/KeeperReadinessHandler.h>
+#include <Server/PrometheusMetricsWriter.h>
+#include <Server/TCPServer.h>
 
 #include "Core/Defines.h"
 #include "config.h"
@@ -488,18 +489,27 @@ try
 
         /// Prometheus (if defined and not setup yet with http_port)
         port_name = "prometheus.port";
-        createServer(listen_host, port_name, listen_try, [&, my_http_context = std::move(http_context)](UInt16 port) mutable
+        createServer(
+            listen_host,
+            port_name,
+            listen_try,
+            [&, my_http_context = std::move(http_context)](UInt16 port) mutable
             {
                 Poco::Net::ServerSocket socket;
                 auto address = socketBindListen(socket, listen_host, port);
                 socket.setReceiveTimeout(my_http_context->getReceiveTimeout());
                 socket.setSendTimeout(my_http_context->getSendTimeout());
+                auto metrics_writer = std::make_shared<KeeperPrometheusMetricsWriter>(config, "prometheus", async_metrics);
                 servers->emplace_back(
                     listen_host,
                     port_name,
                     "Prometheus: http://" + address.toString(),
                     std::make_unique<HTTPServer>(
-                    std::move(my_http_context), createPrometheusMainHandlerFactory(*this, config_getter(), async_metrics, "PrometheusHandler-factory"), server_pool, socket, http_params));
+                        std::move(my_http_context),
+                        createPrometheusMainHandlerFactory(*this, config_getter(), metrics_writer, "PrometheusHandler-factory"),
+                        server_pool,
+                        socket,
+                        http_params));
             });
 
         /// HTTP control endpoints

src/Common/CurrentMetrics.cpp

@@ -2,6 +2,8 @@
 
 
 /// Available metrics. Add something here as you wish.
+/// If the metric is generic (i.e. not server specific)
+/// it should be also added to src/Coordination/KeeperConstant.cpp
 #define APPLY_FOR_BUILTIN_METRICS(M) \
     M(Query, "Number of executing queries") \
     M(Merge, "Number of executing background merges") \

src/Common/ProfileEvents.cpp

@@ -4,6 +4,8 @@
 
 
 /// Available events. Add something here as you wish.
+/// If the event is generic (i.e. not server specific)
+/// it should be also added to src/Coordination/KeeperConstant.cpp
 #define APPLY_FOR_BUILTIN_EVENTS(M) \
     M(Query, "Number of queries to be interpreted and potentially executed. Does not include queries that failed to parse or were rejected due to AST size limits, quota limits or limits on the number of simultaneously running queries. May include internal queries initiated by ClickHouse itself. Does not count subqueries.") \
     M(SelectQuery, "Same as Query, but only for SELECT queries.") \

src/Coordination/KeeperConstants.cpp

@@ -0,0 +1,376 @@
+#include <Common/ProfileEvents.h>
+#include <Common/CurrentMetrics.h>
+
+/// Events which are useful for Keeper.
+/// New events should be added manually.
+#define APPLY_FOR_KEEPER_PROFILE_EVENTS(M) \
+    M(FileOpen) \
+    M(Seek) \
+    M(ReadBufferFromFileDescriptorRead) \
+    M(ReadBufferFromFileDescriptorReadFailed) \
+    M(ReadBufferFromFileDescriptorReadBytes) \
+    M(WriteBufferFromFileDescriptorWrite) \
+    M(WriteBufferFromFileDescriptorWriteFailed) \
+    M(WriteBufferFromFileDescriptorWriteBytes) \
+    M(FileSync) \
+    M(DirectorySync) \
+    M(FileSyncElapsedMicroseconds) \
+    M(DirectorySyncElapsedMicroseconds) \
+    M(ReadCompressedBytes) \
+    M(CompressedReadBufferBlocks) \
+    M(CompressedReadBufferBytes) \
+    M(AIOWrite) \
+    M(AIOWriteBytes) \
+    M(AIORead) \
+    M(AIOReadBytes) \
+    M(IOBufferAllocs) \
+    M(IOBufferAllocBytes) \
+    M(ArenaAllocChunks) \
+    M(ArenaAllocBytes) \
+    M(CreatedReadBufferOrdinary) \
+    M(CreatedReadBufferDirectIO) \
+    M(CreatedReadBufferDirectIOFailed) \
+    M(CreatedReadBufferMMap) \
+    M(CreatedReadBufferMMapFailed) \
+    M(DiskReadElapsedMicroseconds) \
+    M(DiskWriteElapsedMicroseconds) \
+    M(NetworkReceiveElapsedMicroseconds) \
+    M(NetworkSendElapsedMicroseconds) \
+    M(NetworkReceiveBytes) \
+    M(NetworkSendBytes) \
+\
+    M(DiskS3GetRequestThrottlerCount) \
+    M(DiskS3GetRequestThrottlerSleepMicroseconds) \
+    M(DiskS3PutRequestThrottlerCount) \
+    M(DiskS3PutRequestThrottlerSleepMicroseconds) \
+    M(S3GetRequestThrottlerCount) \
+    M(S3GetRequestThrottlerSleepMicroseconds) \
+    M(S3PutRequestThrottlerCount) \
+    M(S3PutRequestThrottlerSleepMicroseconds) \
+    M(RemoteReadThrottlerBytes) \
+    M(RemoteReadThrottlerSleepMicroseconds) \
+    M(RemoteWriteThrottlerBytes) \
+    M(RemoteWriteThrottlerSleepMicroseconds) \
+    M(LocalReadThrottlerBytes) \
+    M(LocalReadThrottlerSleepMicroseconds) \
+    M(LocalWriteThrottlerBytes) \
+    M(LocalWriteThrottlerSleepMicroseconds) \
+    M(ThrottlerSleepMicroseconds) \
+\
+    M(SlowRead) \
+    M(ReadBackoff) \
+\
+    M(ContextLock) \
+    M(ContextLockWaitMicroseconds) \
+\
+    M(RWLockAcquiredReadLocks) \
+    M(RWLockAcquiredWriteLocks) \
+    M(RWLockReadersWaitMilliseconds) \
+    M(RWLockWritersWaitMilliseconds) \
+    M(DNSError) \
+    M(RealTimeMicroseconds) \
+    M(UserTimeMicroseconds) \
+    M(SystemTimeMicroseconds) \
+    M(MemoryOvercommitWaitTimeMicroseconds) \
+    M(MemoryAllocatorPurge) \
+    M(MemoryAllocatorPurgeTimeMicroseconds) \
+    M(SoftPageFaults) \
+    M(HardPageFaults) \
+\
+    M(OSIOWaitMicroseconds) \
+    M(OSCPUWaitMicroseconds) \
+    M(OSCPUVirtualTimeMicroseconds) \
+    M(OSReadBytes) \
+    M(OSWriteBytes) \
+    M(OSReadChars) \
+    M(OSWriteChars) \
+\
+    M(PerfCpuCycles) \
+    M(PerfInstructions) \
+    M(PerfCacheReferences) \
+    M(PerfCacheMisses) \
+    M(PerfBranchInstructions) \
+    M(PerfBranchMisses) \
+    M(PerfBusCycles) \
+    M(PerfStalledCyclesFrontend) \
+    M(PerfStalledCyclesBackend) \
+    M(PerfRefCpuCycles) \
+\
+    M(PerfCpuClock) \
+    M(PerfTaskClock) \
+    M(PerfContextSwitches) \
+    M(PerfCpuMigrations) \
+    M(PerfAlignmentFaults) \
+    M(PerfEmulationFaults) \
+    M(PerfMinEnabledTime) \
+    M(PerfMinEnabledRunningTime) \
+    M(PerfDataTLBReferences) \
+    M(PerfDataTLBMisses) \
+    M(PerfInstructionTLBReferences) \
+    M(PerfInstructionTLBMisses) \
+    M(PerfLocalMemoryReferences) \
+    M(PerfLocalMemoryMisses) \
+\
+    M(CreatedHTTPConnections) \
+    M(CannotWriteToWriteBufferDiscard) \
+\
+    M(S3ReadMicroseconds) \
+    M(S3ReadRequestsCount) \
+    M(S3ReadRequestsErrors) \
+    M(S3ReadRequestsThrottling) \
+    M(S3ReadRequestsRedirects) \
+\
+    M(S3WriteMicroseconds) \
+    M(S3WriteRequestsCount) \
+    M(S3WriteRequestsErrors) \
+    M(S3WriteRequestsThrottling) \
+    M(S3WriteRequestsRedirects) \
+\
+    M(DiskS3ReadMicroseconds) \
+    M(DiskS3ReadRequestsCount) \
+    M(DiskS3ReadRequestsErrors) \
+    M(DiskS3ReadRequestsThrottling) \
+    M(DiskS3ReadRequestsRedirects) \
+\
+    M(DiskS3WriteMicroseconds) \
+    M(DiskS3WriteRequestsCount) \
+    M(DiskS3WriteRequestsErrors) \
+    M(DiskS3WriteRequestsThrottling) \
+    M(DiskS3WriteRequestsRedirects) \
+\
+    M(S3DeleteObjects) \
+    M(S3CopyObject) \
+    M(S3ListObjects) \
+    M(S3HeadObject) \
+    M(S3GetObjectAttributes) \
+    M(S3CreateMultipartUpload) \
+    M(S3UploadPartCopy) \
+    M(S3UploadPart) \
+    M(S3AbortMultipartUpload) \
+    M(S3CompleteMultipartUpload) \
+    M(S3PutObject) \
+    M(S3GetObject) \
+\
+    M(AzureUploadPart) \
+    M(DiskAzureUploadPart) \
+    M(AzureCopyObject) \
+    M(DiskAzureCopyObject) \
+    M(AzureDeleteObjects) \
+    M(AzureListObjects) \
+\
+    M(DiskS3DeleteObjects) \
+    M(DiskS3CopyObject) \
+    M(DiskS3ListObjects) \
+    M(DiskS3HeadObject) \
+    M(DiskS3GetObjectAttributes) \
+    M(DiskS3CreateMultipartUpload) \
+    M(DiskS3UploadPartCopy) \
+    M(DiskS3UploadPart) \
+    M(DiskS3AbortMultipartUpload) \
+    M(DiskS3CompleteMultipartUpload) \
+    M(DiskS3PutObject) \
+    M(DiskS3GetObject) \
+\
+    M(S3Clients) \
+    M(TinyS3Clients) \
+\
+    M(ReadBufferFromS3Microseconds) \
+    M(ReadBufferFromS3InitMicroseconds) \
+    M(ReadBufferFromS3Bytes) \
+    M(ReadBufferFromS3RequestsErrors) \
+    M(ReadBufferFromS3ResetSessions) \
+    M(ReadBufferFromS3PreservedSessions) \
+\
+    M(ReadWriteBufferFromHTTPPreservedSessions) \
+\
+    M(WriteBufferFromS3Microseconds) \
+    M(WriteBufferFromS3Bytes) \
+    M(WriteBufferFromS3RequestsErrors) \
+    M(WriteBufferFromS3WaitInflightLimitMicroseconds) \
+    M(RemoteFSSeeks) \
+    M(RemoteFSPrefetches) \
+    M(RemoteFSCancelledPrefetches) \
+    M(RemoteFSUnusedPrefetches) \
+    M(RemoteFSPrefetchedReads) \
+    M(RemoteFSPrefetchedBytes) \
+    M(RemoteFSUnprefetchedReads) \
+    M(RemoteFSUnprefetchedBytes) \
+    M(RemoteFSLazySeeks) \
+    M(RemoteFSSeeksWithReset) \
+    M(RemoteFSBuffers) \
+\
+    M(ThreadpoolReaderTaskMicroseconds) \
+    M(ThreadpoolReaderPrepareMicroseconds) \
+    M(ThreadpoolReaderReadBytes) \
+    M(ThreadpoolReaderSubmit) \
+    M(ThreadpoolReaderSubmitReadSynchronously) \
+    M(ThreadpoolReaderSubmitReadSynchronouslyBytes) \
+    M(ThreadpoolReaderSubmitReadSynchronouslyMicroseconds) \
+    M(ThreadpoolReaderSubmitLookupInCacheMicroseconds) \
+    M(AsynchronousReaderIgnoredBytes) \
+\
+    M(FileSegmentWaitReadBufferMicroseconds) \
+    M(FileSegmentReadMicroseconds) \
+    M(FileSegmentCacheWriteMicroseconds) \
+    M(FileSegmentPredownloadMicroseconds) \
+    M(FileSegmentUsedBytes) \
+\
+    M(ReadBufferSeekCancelConnection) \
+\
+    M(SleepFunctionCalls) \
+    M(SleepFunctionMicroseconds) \
+    M(SleepFunctionElapsedMicroseconds) \
+\
+    M(ThreadPoolReaderPageCacheHit) \
+    M(ThreadPoolReaderPageCacheHitBytes) \
+    M(ThreadPoolReaderPageCacheHitElapsedMicroseconds) \
+    M(ThreadPoolReaderPageCacheMiss) \
+    M(ThreadPoolReaderPageCacheMissBytes) \
+    M(ThreadPoolReaderPageCacheMissElapsedMicroseconds) \
+\
+    M(AsynchronousReadWaitMicroseconds) \
+    M(SynchronousReadWaitMicroseconds) \
+    M(AsynchronousRemoteReadWaitMicroseconds) \
+    M(SynchronousRemoteReadWaitMicroseconds) \
+\
+    M(ExternalDataSourceLocalCacheReadBytes) \
+\
+    M(MainConfigLoads) \
+\
+    M(KeeperPacketsSent) \
+    M(KeeperPacketsReceived) \
+    M(KeeperRequestTotal) \
+    M(KeeperLatency) \
+    M(KeeperCommits) \
+    M(KeeperCommitsFailed) \
+    M(KeeperSnapshotCreations) \
+    M(KeeperSnapshotCreationsFailed) \
+    M(KeeperSnapshotApplys) \
+    M(KeeperSnapshotApplysFailed) \
+    M(KeeperReadSnapshot) \
+    M(KeeperSaveSnapshot) \
+    M(KeeperCreateRequest) \
+    M(KeeperRemoveRequest) \
+    M(KeeperSetRequest) \
+    M(KeeperReconfigRequest) \
+    M(KeeperCheckRequest) \
+    M(KeeperMultiRequest) \
+    M(KeeperMultiReadRequest) \
+    M(KeeperGetRequest) \
+    M(KeeperListRequest) \
+    M(KeeperExistsRequest) \
+\
+    M(IOUringSQEsSubmitted) \
+    M(IOUringSQEsResubmits) \
+    M(IOUringCQEsCompleted) \
+    M(IOUringCQEsFailed) \
+\
+    M(LogTest) \
+    M(LogTrace) \
+    M(LogDebug) \
+    M(LogInfo) \
+    M(LogWarning) \
+    M(LogError) \
+    M(LogFatal) \
+\
+    M(InterfaceHTTPSendBytes) \
+    M(InterfaceHTTPReceiveBytes) \
+    M(InterfaceNativeSendBytes) \
+    M(InterfaceNativeReceiveBytes) \
+    M(InterfacePrometheusSendBytes) \
+    M(InterfacePrometheusReceiveBytes) \
+    M(InterfaceInterserverSendBytes) \
+    M(InterfaceInterserverReceiveBytes) \
+    M(InterfaceMySQLSendBytes) \
+    M(InterfaceMySQLReceiveBytes) \
+    M(InterfacePostgreSQLSendBytes) \
+    M(InterfacePostgreSQLReceiveBytes)
+
+namespace ProfileEvents
+{
+#define M(NAME) extern const Event NAME;
+    APPLY_FOR_KEEPER_PROFILE_EVENTS(M)
+#undef M
+
+#define M(NAME) NAME,
+extern const std::vector<Event> keeper_profile_events
+{
+    APPLY_FOR_KEEPER_PROFILE_EVENTS(M)
+};
+#undef M
+}
+
+/// Metrics which are useful for Keeper.
+/// New metrics should be added manually.
+#define APPLY_FOR_KEEPER_METRICS(M) \
+    M(BackgroundCommonPoolTask) \
+    M(BackgroundCommonPoolSize) \
+    M(TCPConnection) \
+    M(HTTPConnection) \
+    M(OpenFileForRead) \
+    M(OpenFileForWrite) \
+    M(Read) \
+    M(RemoteRead) \
+    M(Write) \
+    M(NetworkReceive) \
+    M(NetworkSend) \
+    M(MemoryTracking) \
+    M(ContextLockWait) \
+    M(Revision) \
+    M(VersionInteger) \
+    M(RWLockWaitingReaders) \
+    M(RWLockWaitingWriters) \
+    M(RWLockActiveReaders) \
+    M(RWLockActiveWriters) \
+    M(GlobalThread) \
+    M(GlobalThreadActive) \
+    M(GlobalThreadScheduled) \
+    M(LocalThread) \
+    M(LocalThreadActive) \
+    M(LocalThreadScheduled) \
+    M(IOPrefetchThreads) \
+    M(IOPrefetchThreadsActive) \
+    M(IOPrefetchThreadsScheduled) \
+    M(IOWriterThreads) \
+    M(IOWriterThreadsActive) \
+    M(IOWriterThreadsScheduled) \
+    M(IOThreads) \
+    M(IOThreadsActive) \
+    M(IOThreadsScheduled) \
+    M(ThreadPoolRemoteFSReaderThreads) \
+    M(ThreadPoolRemoteFSReaderThreadsActive) \
+    M(ThreadPoolRemoteFSReaderThreadsScheduled) \
+    M(ThreadPoolFSReaderThreads) \
+    M(ThreadPoolFSReaderThreadsActive) \
+    M(ThreadPoolFSReaderThreadsScheduled) \
+    M(DiskObjectStorageAsyncThreads) \
+    M(DiskObjectStorageAsyncThreadsActive) \
+    M(ObjectStorageS3Threads) \
+    M(ObjectStorageS3ThreadsActive) \
+    M(ObjectStorageS3ThreadsScheduled) \
+    M(ObjectStorageAzureThreads) \
+    M(ObjectStorageAzureThreadsActive) \
+    M(ObjectStorageAzureThreadsScheduled) \
+    M(MMappedFiles) \
+    M(MMappedFileBytes) \
+    M(AsynchronousReadWait) \
+    M(S3Requests) \
+    M(KeeperAliveConnections) \
+    M(KeeperOutstandingRequets) \
+    M(ThreadsInOvercommitTracker) \
+    M(IOUringPendingEvents) \
+    M(IOUringInFlightEvents) \
+
+namespace CurrentMetrics
+{
+#define M(NAME) extern const Metric NAME;
+    APPLY_FOR_KEEPER_METRICS(M)
+#undef M
+
+#define M(NAME) NAME,
+extern const std::vector<Metric> keeper_metrics
+{
+    APPLY_FOR_KEEPER_METRICS(M)
+};
+#undef M
+}

src/Server/HTTPHandlerFactory.cpp

@@ -1,3 +1,4 @@
+#include <memory>
 #include <Server/HTTPHandlerFactory.h>
 
 #include <Server/HTTP/HTTPRequestHandler.h>
@@ -7,6 +8,7 @@
 #include <Poco/Util/AbstractConfiguration.h>
 
 #include "HTTPHandler.h"
+#include "Server/PrometheusMetricsWriter.h"
 #include "StaticRequestHandler.h"
 #include "ReplicasStatusHandler.h"
 #include "InterserverIOHTTPHandler.h"
@@ -113,7 +115,10 @@ HTTPRequestHandlerFactoryPtr createHandlerFactory(IServer & server, const Poco::
     else if (name == "InterserverIOHTTPHandler-factory" || name == "InterserverIOHTTPSHandler-factory")
         return createInterserverHTTPHandlerFactory(server, name);
     else if (name == "PrometheusHandler-factory")
-        return createPrometheusMainHandlerFactory(server, config, async_metrics, name);
+    {
+        auto metrics_writer = std::make_shared<PrometheusMetricsWriter>(config, "prometheus", async_metrics);
+        return createPrometheusMainHandlerFactory(server, config, metrics_writer, name);
+    }
 
     throw Exception(ErrorCodes::LOGICAL_ERROR, "LOGICAL ERROR: Unknown HTTP handler factory name.");
 }
@@ -208,7 +213,7 @@ void addDefaultHandlersFactory(
     /// Otherwise it will be created separately, see createHandlerFactory(...).
     if (config.has("prometheus") && config.getInt("prometheus.port", 0) == 0)
     {
-        PrometheusMetricsWriter writer(config, "prometheus", async_metrics);
+        auto writer = std::make_shared<PrometheusMetricsWriter>(config, "prometheus", async_metrics);
         auto creator = [&server, writer] () -> std::unique_ptr<PrometheusRequestHandler>
         {
             return std::make_unique<PrometheusRequestHandler>(server, writer);

src/Server/HTTPHandlerFactory.h

@@ -6,6 +6,7 @@
 #include <Server/HTTPHandlerRequestFilter.h>
 #include <Server/HTTPRequestHandlerFactoryMain.h>
 #include <Common/StringUtils/StringUtils.h>
+#include <Server/PrometheusMetricsWriter.h>
 
 #include <Poco/Util/AbstractConfiguration.h>
 
@@ -130,10 +131,10 @@ createPrometheusHandlerFactory(IServer & server,
     AsynchronousMetrics & async_metrics,
     const std::string & config_prefix);
 
-HTTPRequestHandlerFactoryPtr
-createPrometheusMainHandlerFactory(IServer & server,
+HTTPRequestHandlerFactoryPtr createPrometheusMainHandlerFactory(
+    IServer & server,
     const Poco::Util::AbstractConfiguration & config,
-    AsynchronousMetrics & async_metrics,
+    PrometheusMetricsWriterPtr metrics_writer,
     const std::string & name);
 
 /// @param server - used in handlers to check IServer::isCancelled()

src/Server/PrometheusMetricsWriter.cpp

@@ -4,6 +4,8 @@
 #include <Common/re2.h>
 #include <algorithm>
 
+#include "config.h"
+
 namespace
 {
 
@@ -38,8 +40,83 @@ void convertHelpToSingleLine(std::string & help)
     std::replace(help.begin(), help.end(), '\n', ' ');
 }
 
+constexpr auto profile_events_prefix = "ClickHouseProfileEvents_";
+constexpr auto current_metrics_prefix = "ClickHouseMetrics_";
+constexpr auto asynchronous_metrics_prefix = "ClickHouseAsyncMetrics_";
+constexpr auto error_metrics_prefix = "ClickHouseErrorMetric_";
+
+void writeEvent(DB::WriteBuffer & wb, ProfileEvents::Event event)
+{
+    const auto counter = ProfileEvents::global_counters[event].load(std::memory_order_relaxed);
+
+    std::string metric_name{ProfileEvents::getName(static_cast<ProfileEvents::Event>(event))};
+    std::string metric_doc{ProfileEvents::getDocumentation(static_cast<ProfileEvents::Event>(event))};
+
+    convertHelpToSingleLine(metric_doc);
+
+    if (!replaceInvalidChars(metric_name))
+        return;
+
+    std::string key{profile_events_prefix + metric_name};
+
+    writeOutLine(wb, "# HELP", key, metric_doc);
+    writeOutLine(wb, "# TYPE", key, "counter");
+    writeOutLine(wb, key, counter);
 }
 
+void writeMetric(DB::WriteBuffer & wb, size_t metric)
+{
+    const auto value = CurrentMetrics::values[metric].load(std::memory_order_relaxed);
+
+    std::string metric_name{CurrentMetrics::getName(static_cast<CurrentMetrics::Metric>(metric))};
+    std::string metric_doc{CurrentMetrics::getDocumentation(static_cast<CurrentMetrics::Metric>(metric))};
+
+    convertHelpToSingleLine(metric_doc);
+
+    if (!replaceInvalidChars(metric_name))
+        return;
+
+    std::string key{current_metrics_prefix + metric_name};
+
+    writeOutLine(wb, "# HELP", key, metric_doc);
+    writeOutLine(wb, "# TYPE", key, "gauge");
+    writeOutLine(wb, key, value);
+}
+
+void writeAsyncMetrics(DB::WriteBuffer & wb, const DB::AsynchronousMetricValues & values)
+{
+    for (const auto & name_value : values)
+    {
+        std::string key{asynchronous_metrics_prefix + name_value.first};
+
+        if (!replaceInvalidChars(key))
+            continue;
+
+        auto value = name_value.second;
+
+        std::string metric_doc{value.documentation};
+        convertHelpToSingleLine(metric_doc);
+
+        writeOutLine(wb, "# HELP", key, metric_doc);
+        writeOutLine(wb, "# TYPE", key, "gauge");
+        writeOutLine(wb, key, value.value);
+    }
+}
+
+}
+
+#if USE_NURAFT
+namespace ProfileEvents
+{
+    extern const std::vector<Event> keeper_profile_events;
+}
+
+namespace CurrentMetrics
+{
+    extern const std::vector<Metric> keeper_metrics;
+}
+#endif
+
 
 namespace DB
 {
@@ -60,65 +137,17 @@ void PrometheusMetricsWriter::write(WriteBuffer & wb) const
     if (send_events)
     {
         for (ProfileEvents::Event i = ProfileEvents::Event(0), end = ProfileEvents::end(); i < end; ++i)
-        {
-            const auto counter = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
-
-            std::string metric_name{ProfileEvents::getName(static_cast<ProfileEvents::Event>(i))};
-            std::string metric_doc{ProfileEvents::getDocumentation(static_cast<ProfileEvents::Event>(i))};
-
-            convertHelpToSingleLine(metric_doc);
-
-            if (!replaceInvalidChars(metric_name))
-                continue;
-            std::string key{profile_events_prefix + metric_name};
-
-            writeOutLine(wb, "# HELP", key, metric_doc);
-            writeOutLine(wb, "# TYPE", key, "counter");
-            writeOutLine(wb, key, counter);
-        }
+            writeEvent(wb, i);
     }
 
     if (send_metrics)
     {
         for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
-        {
-            const auto value = CurrentMetrics::values[i].load(std::memory_order_relaxed);
-
-            std::string metric_name{CurrentMetrics::getName(static_cast<CurrentMetrics::Metric>(i))};
-            std::string metric_doc{CurrentMetrics::getDocumentation(static_cast<CurrentMetrics::Metric>(i))};
-
-            convertHelpToSingleLine(metric_doc);
-
-            if (!replaceInvalidChars(metric_name))
-                continue;
-            std::string key{current_metrics_prefix + metric_name};
-
-            writeOutLine(wb, "# HELP", key, metric_doc);
-            writeOutLine(wb, "# TYPE", key, "gauge");
-            writeOutLine(wb, key, value);
-        }
+            writeMetric(wb, i);
     }
 
     if (send_asynchronous_metrics)
-    {
-        auto async_metrics_values = async_metrics.getValues();
-        for (const auto & name_value : async_metrics_values)
-        {
-            std::string key{asynchronous_metrics_prefix + name_value.first};
-
-            if (!replaceInvalidChars(key))
-                continue;
-
-            auto value = name_value.second;
-
-            std::string metric_doc{value.documentation};
-            convertHelpToSingleLine(metric_doc);
-
-            writeOutLine(wb, "# HELP", key, metric_doc);
-            writeOutLine(wb, "# TYPE", key, "gauge");
-            writeOutLine(wb, key, value.value);
-        }
-    }
+        writeAsyncMetrics(wb, async_metrics.getValues());
 
     if (send_errors)
     {
@@ -152,4 +181,24 @@ void PrometheusMetricsWriter::write(WriteBuffer & wb) const
 
 }
 
+void KeeperPrometheusMetricsWriter::write([[maybe_unused]] WriteBuffer & wb) const
+{
+#if USE_NURAFT
+    if (send_events)
+    {
+        for (auto event : ProfileEvents::keeper_profile_events)
+            writeEvent(wb, event);
+    }
+
+    if (send_metrics)
+    {
+        for (auto metric : CurrentMetrics::keeper_metrics)
+            writeMetric(wb, metric);
+    }
+
+    if (send_asynchronous_metrics)
+        writeAsyncMetrics(wb, async_metrics.getValues());
+#endif
+}
+
 }

src/Server/PrometheusMetricsWriter.h

@@ -3,6 +3,7 @@
 #include <string>
 
 #include <Common/AsynchronousMetrics.h>
+#include <Common/ProfileEvents.h>
 #include <IO/WriteBuffer.h>
 
 #include <Poco/Util/AbstractConfiguration.h>
@@ -19,20 +20,25 @@ public:
         const Poco::Util::AbstractConfiguration & config, const std::string & config_name,
         const AsynchronousMetrics & async_metrics_);
 
-    void write(WriteBuffer & wb) const;
+    virtual void write(WriteBuffer & wb) const;
 
-private:
+    virtual ~PrometheusMetricsWriter() = default;
+
+protected:
     const AsynchronousMetrics & async_metrics;
-
     const bool send_events;
     const bool send_metrics;
     const bool send_asynchronous_metrics;
     const bool send_errors;
-
-    static inline constexpr auto profile_events_prefix = "ClickHouseProfileEvents_";
-    static inline constexpr auto current_metrics_prefix = "ClickHouseMetrics_";
-    static inline constexpr auto asynchronous_metrics_prefix = "ClickHouseAsyncMetrics_";
-    static inline constexpr auto error_metrics_prefix = "ClickHouseErrorMetric_";
 };
 
+class KeeperPrometheusMetricsWriter : public PrometheusMetricsWriter
+{
+    using PrometheusMetricsWriter::PrometheusMetricsWriter;
+
+    void write(WriteBuffer & wb) const override;
+};
+
+using PrometheusMetricsWriterPtr = std::shared_ptr<PrometheusMetricsWriter>;
+
 }

src/Server/PrometheusRequestHandler.cpp

@@ -7,6 +7,7 @@
 #include <Common/CurrentMetrics.h>
 #include <Common/Exception.h>
 #include <Common/ProfileEvents.h>
+#include "Server/PrometheusMetricsWriter.h"
 
 #include <Poco/Util/LayeredConfiguration.h>
 
@@ -34,7 +35,7 @@ void PrometheusRequestHandler::handleRequest(HTTPServerRequest & request, HTTPSe
         WriteBufferFromHTTPServerResponse wb(response, request.getMethod() == Poco::Net::HTTPRequest::HTTP_HEAD, keep_alive_timeout, write_event);
         try
         {
-            metrics_writer.write(wb);
+            metrics_writer->write(wb);
             wb.finalize();
         }
         catch (...)
@@ -54,7 +55,7 @@ HTTPRequestHandlerFactoryPtr createPrometheusHandlerFactory(
     AsynchronousMetrics & async_metrics,
     const std::string & config_prefix)
 {
-    PrometheusMetricsWriter writer(config, config_prefix + ".handler", async_metrics);
+    auto writer = std::make_shared<PrometheusMetricsWriter>(config, config_prefix + ".handler", async_metrics);
     auto creator = [&server, writer]() -> std::unique_ptr<PrometheusRequestHandler>
     {
         return std::make_unique<PrometheusRequestHandler>(server, writer);
@@ -66,13 +67,12 @@ HTTPRequestHandlerFactoryPtr createPrometheusHandlerFactory(
 }
 
 HTTPRequestHandlerFactoryPtr createPrometheusMainHandlerFactory(
-    IServer & server, const Poco::Util::AbstractConfiguration & config, AsynchronousMetrics & async_metrics, const std::string & name)
+    IServer & server, const Poco::Util::AbstractConfiguration & config, PrometheusMetricsWriterPtr metrics_writer, const std::string & name)
 {
     auto factory = std::make_shared<HTTPRequestHandlerFactoryMain>(name);
-    PrometheusMetricsWriter writer(config, "prometheus", async_metrics);
-    auto creator = [&server, writer]() -> std::unique_ptr<PrometheusRequestHandler>
+    auto creator = [&server, metrics_writer]
     {
-        return std::make_unique<PrometheusRequestHandler>(server, writer);
+        return std::make_unique<PrometheusRequestHandler>(server, metrics_writer);
     };
 
     auto handler = std::make_shared<HandlingRuleHTTPHandlerFactory<PrometheusRequestHandler>>(std::move(creator));

src/Server/PrometheusRequestHandler.h

@@ -13,12 +13,12 @@ class PrometheusRequestHandler : public HTTPRequestHandler
 {
 private:
     IServer & server;
-    const PrometheusMetricsWriter & metrics_writer;
+    PrometheusMetricsWriterPtr metrics_writer;
 
 public:
-    PrometheusRequestHandler(IServer & server_, const PrometheusMetricsWriter & metrics_writer_)
+    PrometheusRequestHandler(IServer & server_, PrometheusMetricsWriterPtr metrics_writer_)
         : server(server_)
-        , metrics_writer(metrics_writer_)
+        , metrics_writer(std::move(metrics_writer_))
     {
     }
 

utils/check-style/check-style

@@ -76,6 +76,7 @@ EXTERN_TYPES_EXCLUDES=(
     ProfileEvents::getProfileEvents
     ProfileEvents::ThreadIdToCountersSnapshot
     ProfileEvents::LOCAL_NAME
+    ProfileEvents::keeper_profile_events
     ProfileEvents::CountersIncrement
 
     CurrentMetrics::add
@@ -87,6 +88,7 @@ EXTERN_TYPES_EXCLUDES=(
     CurrentMetrics::Metric
     CurrentMetrics::values
     CurrentMetrics::Value
+    CurrentMetrics::keeper_metrics
 
     ErrorCodes::ErrorCode
     ErrorCodes::getName
@@ -106,7 +108,7 @@ for extern_type in ${!EXTERN_TYPES[@]}; do
     find $ROOT_PATH/{src,base,programs,utils} -name '*.h' -or -name '*.cpp' | {
         # NOTE: the check is pretty dumb and distinguish only by the type_of_extern,
         # and this matches with zkutil::CreateMode
-        grep -v 'src/Common/ZooKeeper/Types.h'
+        grep -v -e 'src/Common/ZooKeeper/Types.h' -e 'src/Coordination/KeeperConstants.cpp'
     } | {
         grep -vP $EXCLUDE_DIRS | xargs grep -l -P "extern const $type_of_extern $allowed_chars"
     } | while read file; do