Disable topic metrics

Author: Gazizonoki

cmake/common.cmake

@@ -200,6 +200,7 @@ function(_ydb_sdk_add_library Tgt)
   )
   target_compile_definitions(${Tgt} ${includeMode}
     YDB_SDK_USE_STD_STRING
+    YDB_TOPIC_DISABLE_COUNTERS
   )
 endfunction()
 

include/ydb-cpp-sdk/client/federated_topic/federated_topic.h

@@ -479,8 +479,10 @@ class IFederatedReadSession {
     //! TSessionClosedEvent arrives.
     virtual bool Close(TDuration timeout = TDuration::Max()) = 0;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     //! Reader counters with different stats (see TReaderConuters).
     virtual NTopic::TReaderCounters::TPtr GetCounters() const = 0;
+#endif
 
     //! Get unique identifier of read session.
     virtual std::string GetSessionId() const = 0;

include/ydb-cpp-sdk/client/topic/read_session.h

@@ -1,10 +1,13 @@
 #pragma once
 
-#include "counters.h"
 #include "executor.h"
 #include "read_events.h"
 #include "retry_policy.h"
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
+#include "counters.h"
+#endif
+
 #include <ydb-cpp-sdk/client/common_client/settings.h>
 #include <ydb-cpp-sdk/client/types/tx/tx.h>
 #include <ydb-cpp-sdk/client/types/fluent_settings_helpers.h>
@@ -183,10 +186,12 @@ struct TReadSessionSettings: public TRequestSettings<TReadSessionSettings> {
     //! If not set, default executor will be used.
     FLUENT_SETTING(IExecutor::TPtr, DecompressionExecutor);
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     //! Counters.
     //! If counters are not provided explicitly,
     //! they will be created inside session (without link with parent counters).
     FLUENT_SETTING(TReaderCounters::TPtr, Counters);
+#endif
 
     FLUENT_SETTING_DEFAULT(TDuration, ConnectTimeout, TDuration::Seconds(30));
 
@@ -244,8 +249,10 @@ class IReadSession {
     //! TSessionClosedEvent arrives.
     virtual bool Close(TDuration timeout = TDuration::Max()) = 0;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     //! Reader counters with different stats (see TReaderConuters).
     virtual TReaderCounters::TPtr GetCounters() const = 0;
+#endif
 
     //! Get unique identifier of read session.
     virtual std::string GetSessionId() const = 0;

include/ydb-cpp-sdk/client/topic/write_session.h

@@ -1,11 +1,14 @@
 #pragma once
 
 #include "codecs.h"
-#include "counters.h"
 #include "executor.h"
 #include "retry_policy.h"
 #include "write_events.h"
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
+#include "counters.h"
+#endif
+
 #include <ydb-cpp-sdk/client/types/tx/tx.h>
 #include <ydb-cpp-sdk/client/types/fluent_settings_helpers.h>
 #include <ydb-cpp-sdk/client/types/request_settings.h>
@@ -91,7 +94,9 @@ struct TWriteSessionSettings : public TRequestSettings<TWriteSessionSettings> {
 
     FLUENT_SETTING_DEFAULT(TDuration, ConnectTimeout, TDuration::Seconds(30));
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     FLUENT_SETTING_OPTIONAL(TWriterCounters::TPtr, Counters);
+#endif
 
     //! Executor for compression tasks.
     //! If not set, default executor will be used.
@@ -222,7 +227,9 @@ class ISimpleBlockingWriteSession : public TThrRefBase {
     //! Returns true if write session is alive and acitve. False if session was closed.
     virtual bool IsAlive() const = 0;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     virtual TWriterCounters::TPtr GetCounters() = 0;
+#endif
 
     //! Close immediately and destroy, don't wait for anything.
     virtual ~ISimpleBlockingWriteSession() = default;
@@ -269,8 +276,10 @@ class IWriteSession {
     //! Return true if all writes were completed and acked, false if timeout was reached and some writes were aborted.
     virtual bool Close(TDuration closeTimeout = TDuration::Max()) = 0;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     //! Writer counters with different stats (see TWriterConuters).
     virtual TWriterCounters::TPtr GetCounters() = 0;
+#endif
 
     //! Close() with timeout = 0 and destroy everything instantly.
     virtual ~IWriteSession() = default;

src/client/federated_topic/impl/federated_read_session.h

@@ -151,9 +151,11 @@ class TFederatedReadSessionImpl : public NTopic::TEnableSelfContext<TFederatedRe
         return SessionId;
     }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     inline NTopic::TReaderCounters::TPtr GetCounters() const {
         return Settings.Counters_; // Always not nullptr.
     }
+#endif
 
 private:
     TStringBuilder GetLogPrefix() const;
@@ -236,9 +238,11 @@ class TFederatedReadSession : public IFederatedReadSession,
         return TryGetImpl()->GetSessionId();
     }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     inline NTopic::TReaderCounters::TPtr GetCounters() const override {
         return TryGetImpl()->GetCounters();
     }
+#endif
 
 private:
     void Start() {

src/client/federated_topic/impl/federated_write_session.h

@@ -196,8 +196,9 @@ class TFederatedWriteSession : public NTopic::IWriteSession,
     bool Close(TDuration timeout) override {
         return TryGetImpl()->Close(timeout);
     }
-
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     inline NTopic::TWriterCounters::TPtr GetCounters() override {Y_ABORT("Unimplemented"); } //ToDo - unimplemented;
+#endif
 
 private:
 

src/client/persqueue_public/impl/aliases.h

@@ -32,8 +32,10 @@ using NTopic::Cancel;
 using NTopic::IsErrorMessage;
 using NTopic::MakeErrorFromProto;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
 // counters_logger
 using TCountersLogger = NTopic::TCountersLogger<true>;
+#endif
 
 // read_session_impl
 using NTopic::HasNullCounters;

src/client/persqueue_public/impl/read_session.cpp

@@ -45,7 +45,9 @@ TReadSession::TReadSession(const TReadSessionSettings& settings,
         Settings.RetryPolicy_ = IRetryPolicy::GetDefaultPolicy();
     }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     MakeCountersIfNeeded();
+#endif
 }
 
 TReadSession::~TReadSession() {
@@ -64,9 +66,11 @@ TReadSession::~TReadSession() {
     for (const auto& ctx : CbContexts) {
         ctx->Cancel();
     }
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     if (DumpCountersContext) {
         DumpCountersContext->Cancel();
     }
+#endif
 }
 
 Ydb::PersQueue::ClusterDiscovery::DiscoverClustersRequest TReadSession::MakeClusterDiscoveryRequest() const {
@@ -174,7 +178,9 @@ void TReadSession::ProceedWithoutClusterDiscovery() {
         clusterSessionInfo.Topics = Settings.Topics_;
         CreateClusterSessionsImpl(deferred);
     }
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     SetupCountersLogger();
+#endif
 }
 
 void TReadSession::CreateClusterSessionsImpl(TDeferredActions& deferred) {
@@ -226,7 +232,9 @@ void TReadSession::OnClusterDiscovery(const TStatus& status, const Ydb::PersQueu
         }
 
         if (!status.IsSuccess()) {
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
             ++*Settings.Counters_->Errors;
+#endif
             if (!ClusterDiscoveryRetryState) {
                 ClusterDiscoveryRetryState = Settings.RetryPolicy_->CreateRetryState();
             }
@@ -249,7 +257,9 @@ void TReadSession::OnClusterDiscovery(const TStatus& status, const Ydb::PersQueu
 
         // Init ClusterSessions.
         if (static_cast<size_t>(result.read_sessions_clusters_size()) != Settings.Topics_.size()) {
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
             ++*Settings.Counters_->Errors;
+#endif
             AbortImpl(EStatus::INTERNAL_ERROR, TStringBuilder() << "Unexpected reply from cluster discovery. Sizes of topics arrays don't match: "
                       << result.read_sessions_clusters_size() << " vs " << Settings.Topics_.size(), deferred);
             return;
@@ -309,14 +319,18 @@ void TReadSession::OnClusterDiscovery(const TStatus& status, const Ydb::PersQueu
         }
 
         if (issues) {
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
             ++*Settings.Counters_->Errors;
+#endif
             AbortImpl(errorStatus, std::move(issues), deferred);
             return;
         }
 
         CreateClusterSessionsImpl(deferred);
     }
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     SetupCountersLogger();
+#endif
 }
 
 void TReadSession::RestartClusterDiscoveryImpl(TDuration delay, TDeferredActions& deferred) {
@@ -345,14 +359,13 @@ void TReadSession::RestartClusterDiscoveryImpl(TDuration delay, TDeferredActions
 
 bool TReadSession::Close(TDuration timeout) {
     LOG_LAZY(Log, TLOG_INFO, GetLogPrefix() << "Closing read session. Close timeout: " << timeout);
-
-
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     // the program may not have reached SetupCountersLogger
     if (CountersLogger) {
         // Log final counters.
         CountersLogger->Stop();
     }
-
+#endif
     std::vector<TSingleClusterReadSessionImpl::TPtr> sessions;
     NThreading::TPromise<bool> promise = NThreading::NewPromise<bool>();
     std::shared_ptr<std::atomic<size_t>> count = std::make_shared<std::atomic<size_t>>(0);
@@ -412,7 +425,9 @@ bool TReadSession::Close(TDuration timeout) {
         issues.AddIssue("Session was gracefully closed");
         EventsQueue->Close(TSessionClosedEvent(EStatus::SUCCESS, std::move(issues)), deferred);
     } else {
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
         ++*Settings.Counters_->Errors;
+#endif
         for (const auto& session : sessions) {
             session->Abort();
         }
@@ -438,10 +453,12 @@ void TReadSession::AbortImpl(TDeferredActions&) {
             ClusterDiscoveryDelayContext->Cancel();
             ClusterDiscoveryDelayContext.reset();
         }
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
         if (DumpCountersContext) {
             DumpCountersContext->Cancel();
             DumpCountersContext.reset();
         }
+#endif
         for (auto& [cluster, sessionInfo] : ClusterSessions) {
             if (sessionInfo.Session) {
                 sessionInfo.Session->Abort();
@@ -526,6 +543,7 @@ void TReadSession::ResumeReadingData() {
     }
 }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
 void TReadSession::MakeCountersIfNeeded() {
     if (!Settings.Counters_ || HasNullCounters(*Settings.Counters_)) {
         TReaderCounters::TPtr counters = MakeIntrusive<TReaderCounters>();
@@ -544,6 +562,7 @@ void TReadSession::SetupCountersLogger() {
     DumpCountersContext = CountersLogger->MakeCallbackContext();
     CountersLogger->Start();
 }
+#endif
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 // NPersQueue::TReadSessionEvent

src/client/persqueue_public/impl/read_session.h

@@ -48,9 +48,11 @@ class TReadSession : public IReadSession,
         return SessionId;
     }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     TReaderCounters::TPtr GetCounters() const override {
         return Settings.Counters_; // Always not nullptr.
     }
+#endif
 
     void AddTopic(const TTopicReadSettings& topicReadSettings) /*override*/ {
         Y_UNUSED(topicReadSettings);
@@ -97,8 +99,10 @@ class TReadSession : public IReadSession,
     void AbortImpl(EStatus statusCode, NYdb::NIssue::TIssues&& issues, TDeferredActions& deferred);
     void AbortImpl(EStatus statusCode, const std::string& message, TDeferredActions& deferred);
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     void MakeCountersIfNeeded();
     void SetupCountersLogger();
+#endif
 
 private:
     TReadSessionSettings Settings;
@@ -121,8 +125,10 @@ class TReadSession : public IReadSession,
 
     std::vector<TCallbackContextPtr> CbContexts;
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
     std::shared_ptr<TCountersLogger> CountersLogger;
     std::shared_ptr<TCallbackContext<TCountersLogger>> DumpCountersContext;
+#endif
 };
 
 }  // namespace NYdb::NPersQueue

src/client/persqueue_public/impl/write_session.cpp

@@ -135,9 +135,11 @@ std::optional<TContinuationToken> TSimpleBlockingWriteSession::WaitForToken(cons
     return std::nullopt;
 }
 
+#ifndef YDB_TOPIC_DISABLE_COUNTERS
 TWriterCounters::TPtr TSimpleBlockingWriteSession::GetCounters() {
     return Writer->GetCounters();
 }
+#endif
 
 
 bool TSimpleBlockingWriteSession::IsAlive() const {