Merge 3894a95 into ba306c3

Author: GrigoriyPA

ydb/core/kqp/workload_service/actors/pool_handlers_acors.cpp

@@ -48,25 +48,37 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
 
         TCommonCounters(NMonitoring::TDynamicCounterPtr counters, const TString& database, const TString& poolId, const NResourcePool::TPoolSettings& poolConfig)
             : CountersRoot(counters)
-            , CountersSubgroup(counters->GetSubgroup("pool", CanonizePath(TStringBuilder() << database << "/" << poolId)))
+            , CountersSubgroup(counters ? counters->GetSubgroup("pool", CanonizePath(TStringBuilder() << database << "/" << poolId)) : nullptr)
         {
             Register();
             UpdateConfigCounters(poolConfig);
         }
 
+        explicit operator bool() const {
+            return !!CountersRoot;
+        }
+
         void CollectRequestLatency(TInstant continueTime) {
-            if (continueTime) {
+            if (continueTime && CountersRoot) {
                 RequestsLatencyMs->Collect((TInstant::Now() - continueTime).MilliSeconds());
             }
         }
 
         void UpdateConfigCounters(const NResourcePool::TPoolSettings& poolConfig) {
+            if (!CountersRoot) {
+                return;
+            }
+
             InFlightLimit->Set(std::max(poolConfig.ConcurrentQueryLimit, 0));
             QueueSizeLimit->Set(std::max(poolConfig.QueueSize, 0));
             LoadCpuThreshold->Set(std::max(poolConfig.DatabaseLoadCpuThreshold, 0.0));
         }
 
         void OnCleanup() {
+            if (!CountersRoot) {
+                return;
+            }
+
             ActivePoolHandlers->Dec();
 
             InFlightLimit->Set(0);
@@ -76,6 +88,10 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
 
     private:
         void Register() {
+            if (!CountersRoot) {
+                return;
+            }
+
             ActivePoolHandlers = CountersRoot->GetCounter("ActivePoolHandlers", false);
             ActivePoolHandlers->Inc();
 
@@ -202,7 +218,9 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
         }
 
         TRequest* request = &LocalSessions.insert({sessionId, TRequest(workerActorId, sessionId)}).first->second;
-        Counters.LocalDelayedRequests->Inc();
+        if (Counters) {
+            Counters.LocalDelayedRequests->Inc();
+        }
 
         UpdatePoolConfig(ev->Get()->PoolConfig);
         UpdateSchemeboardSubscription(ev->Get()->PathId);
@@ -303,25 +321,35 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
             LocalInFlight++;
             request->Started = true;
             request->ContinueTime = TInstant::Now();
-            Counters.LocalInFly->Inc();
-            Counters.ContinueOk->Inc();
-            Counters.DelayedTimeMs->Collect((TInstant::Now() - request->StartTime).MilliSeconds());
+            if (Counters) {
+                Counters.LocalInFly->Inc();
+                Counters.ContinueOk->Inc();
+                Counters.DelayedTimeMs->Collect((TInstant::Now() - request->StartTime).MilliSeconds());
+            }
             LOG_D("Reply continue success to " << request->WorkerActorId << ", session id: " << request->SessionId << ", local in flight: " << LocalInFlight);
         } else {
             if (status == Ydb::StatusIds::OVERLOADED) {
-                Counters.ContinueOverloaded->Inc();
+                if (Counters) {
+                    Counters.ContinueOverloaded->Inc();
+                }
                 LOG_I("Reply overloaded to " << request->WorkerActorId << ", session id: " << request->SessionId << ", issues: " << issues.ToOneLineString());
             } else if (status == Ydb::StatusIds::CANCELLED) {
-                Counters.Cancelled->Inc();
+                if (Counters) {
+                    Counters.Cancelled->Inc();
+                }
                 LOG_I("Reply cancelled to " << request->WorkerActorId << ", session id: " << request->SessionId << ", issues: " << issues.ToOneLineString());
             } else {
-                Counters.ContinueError->Inc();
+                if (Counters) {
+                    Counters.ContinueError->Inc();
+                }
                 LOG_W("Reply continue error " << status << " to " << request->WorkerActorId << ", session id: " << request->SessionId << ", issues: " << issues.ToOneLineString());
             }
             RemoveRequest(request);
         }
 
-        Counters.LocalDelayedRequests->Dec();
+        if (Counters) {
+            Counters.LocalDelayedRequests->Dec();
+        }
     }
 
     void FinalReply(TRequest* request, Ydb::StatusIds::StatusCode status, const TString& message) {
@@ -340,8 +368,10 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
 
         if (request->Started) {
             LocalInFlight--;
-            Counters.LocalInFly->Dec();
-        } else {
+            if (Counters) {
+                Counters.LocalInFly->Dec();
+            }
+        } else if (Counters) {
             Counters.LocalDelayedRequests->Dec();
         }
 
@@ -428,11 +458,15 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
         this->Send(request->WorkerActorId, new TEvCleanupResponse(status, issues));
 
         if (status == Ydb::StatusIds::SUCCESS) {
-            Counters.CleanupOk->Inc();
-            Counters.CollectRequestLatency(request->ContinueTime);
+            if (Counters) {
+                Counters.CleanupOk->Inc();
+                Counters.CollectRequestLatency(request->ContinueTime);
+            }
             LOG_D("Reply cleanup success to " << request->WorkerActorId << ", session id: " << request->SessionId << ", local in flight: " << LocalInFlight);
         } else {
-            Counters.CleanupError->Inc();
+            if (Counters) {
+                Counters.CleanupError->Inc();
+            }
             LOG_W("Reply cleanup error " << status << " to " << request->WorkerActorId << ", session id: " << request->SessionId << ", issues: " << issues.ToOneLineString());
         }
     }
@@ -442,8 +476,10 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
         ev->Record.MutableRequest()->SetSessionId(request->SessionId);
         this->Send(MakeKqpProxyID(this->SelfId().NodeId()), ev.release());
 
-        Counters.Cancelled->Inc();
-        Counters.CollectRequestLatency(request->ContinueTime);
+        if (Counters) {
+            Counters.Cancelled->Inc();
+            Counters.CollectRequestLatency(request->ContinueTime);
+        }
         LOG_I("Cancel request for worker " << request->WorkerActorId << ", session id: " << request->SessionId << ", local in flight: " << LocalInFlight);
     }
 
@@ -547,34 +583,54 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
     using TBase = TPoolHandlerActorBase<TFifoPoolHandlerActor>;
 
     struct TCounters {
+        const NMonitoring::TDynamicCounterPtr CountersSubgroup;
+
         // Fifo pool counters
         NMonitoring::TDynamicCounters::TCounterPtr PendingRequestsCount;
         NMonitoring::TDynamicCounters::TCounterPtr FinishingRequestsCount;
         NMonitoring::TDynamicCounters::TCounterPtr GlobalInFly;
         NMonitoring::TDynamicCounters::TCounterPtr GlobalDelayedRequests;
         NMonitoring::THistogramPtr PoolStateUpdatesBacklogMs;
 
-        TCounters(NMonitoring::TDynamicCounterPtr countersSubgroup) {
-            Register(countersSubgroup);
+        TCounters(NMonitoring::TDynamicCounterPtr countersSubgroup)
+            : CountersSubgroup(countersSubgroup)
+        {
+            Register();
+        }
+
+        explicit operator bool() const {
+            return !!CountersSubgroup;
         }
 
         void UpdateGlobalState(const TPoolStateDescription& description) {
+            if (!CountersSubgroup) {
+                return;
+            }
+
             GlobalInFly->Set(description.RunningRequests);
             GlobalDelayedRequests->Set(description.DelayedRequests);
         }
 
         void OnCleanup() {
+            if (!CountersSubgroup) {
+                return;
+            }
+
             GlobalInFly->Set(0);
             GlobalDelayedRequests->Set(0);
         }
 
     private:
-        void Register(NMonitoring::TDynamicCounterPtr countersSubgroup) {
-            PendingRequestsCount = countersSubgroup->GetCounter("PendingRequestsCount", false);
-            FinishingRequestsCount = countersSubgroup->GetCounter("FinishingRequestsCount", false);
-            GlobalInFly = countersSubgroup->GetCounter("GlobalInFly", false);
-            GlobalDelayedRequests = countersSubgroup->GetCounter("GlobalDelayedRequests", false);
-            PoolStateUpdatesBacklogMs = countersSubgroup->GetHistogram("PoolStateUpdatesBacklogMs", NMonitoring::LinearHistogram(20, 0, 3 * LEASE_DURATION.MillisecondsFloat() / 40)); 
+        void Register() {
+            if (!CountersSubgroup) {
+                return;
+            }
+
+            PendingRequestsCount = CountersSubgroup->GetCounter("PendingRequestsCount", false);
+            FinishingRequestsCount = CountersSubgroup->GetCounter("FinishingRequestsCount", false);
+            GlobalInFly = CountersSubgroup->GetCounter("GlobalInFly", false);
+            GlobalDelayedRequests = CountersSubgroup->GetCounter("GlobalDelayedRequests", false);
+            PoolStateUpdatesBacklogMs = CountersSubgroup->GetHistogram("PoolStateUpdatesBacklogMs", NMonitoring::LinearHistogram(20, 0, 3 * LEASE_DURATION.MillisecondsFloat() / 40)); 
         }
     };
 
@@ -628,7 +684,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
         }
 
         PendingRequests.emplace_back(request->SessionId);
-        FifoCounters.PendingRequestsCount->Inc();
+        if (FifoCounters) {
+            FifoCounters.PendingRequestsCount->Inc();
+        }
 
         if (!PreparingFinished) {
             this->Send(MakeKqpWorkloadServiceId(this->SelfId().NodeId()), new TEvPrivate::TEvPrepareTablesRequest(Database, PoolId));
@@ -710,7 +768,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
         });
 
         PendingRequests.clear();
-        FifoCounters.PendingRequestsCount->Set(0);
+        if (FifoCounters) {
+            FifoCounters.PendingRequestsCount->Set(0);
+        }
     }
 
     void Handle(TEvPrivate::TEvRefreshPoolStateResponse::TPtr& ev) {
@@ -723,7 +783,7 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
             return;
         }
 
-        if (LastRefreshTime) {
+        if (LastRefreshTime && FifoCounters) {
             FifoCounters.PoolStateUpdatesBacklogMs->Collect((TInstant::Now() - LastRefreshTime).MilliSeconds());
         }
         LastRefreshTime = TInstant::Now();
@@ -744,7 +804,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
             RemoveBackRequests(PendingRequests, std::min(delayedRequests - QueueSizeLimit, PendingRequests.size()), [this](TRequest* request) {
                 ReplyContinue(request, Ydb::StatusIds::OVERLOADED, TStringBuilder() << "Too many pending requests for pool " << PoolId);
             });
-            FifoCounters.PendingRequestsCount->Set(PendingRequests.size());
+            if (FifoCounters) {
+                FifoCounters.PendingRequestsCount->Set(PendingRequests.size());
+            }
         }
 
         if (PendingRequests.empty() && delayedRequestsCount > QueueSizeLimit) {
@@ -774,7 +836,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
         }
 
         GlobalState.DelayedRequests++;
-        FifoCounters.GlobalDelayedRequests->Inc();
+        if (FifoCounters) {
+            FifoCounters.GlobalDelayedRequests->Inc();
+        }
         LOG_D("succefully delayed request, session id: " << ev->Get()->SessionId);
 
         DoStartDelayedRequest(GetLoadCpuThreshold());
@@ -849,12 +913,16 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
                     request->UsedCpuQuota = !!GetLoadCpuThreshold();
                     requestFound = true;
                     GlobalState.RunningRequests++;
-                    FifoCounters.GlobalInFly->Inc();
+                    if (FifoCounters) {
+                        FifoCounters.GlobalInFly->Inc();
+                    }
                     ReplyContinue(request);
                 } else {
                     // Request was dropped due to lease expiration 
                     PendingRequests.emplace_front(request->SessionId);
-                    FifoCounters.PendingRequestsCount->Inc();
+                    if (FifoCounters) {
+                        FifoCounters.PendingRequestsCount->Inc();
+                    }
                 }
             });
             DelayedRequests.pop_front();
@@ -947,7 +1015,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
             RunningOperation = true;
             this->Register(CreateCleanupRequestsActor(this->SelfId(), Database, PoolId, FinishedRequests, Counters.CountersSubgroup));
             FinishedRequests.clear();
-            FifoCounters.FinishingRequestsCount->Set(0);
+            if (FifoCounters) {
+                FifoCounters.FinishingRequestsCount->Set(0);
+            }
         }
     }
 
@@ -971,7 +1041,9 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
 
         RemoveFinishedRequests(PendingRequests);
         RemoveFinishedRequests(DelayedRequests);
-        FifoCounters.PendingRequestsCount->Set(PendingRequests.size());
+        if (FifoCounters) {
+            FifoCounters.PendingRequestsCount->Set(PendingRequests.size());
+        }
     }
 
     void RemoveFinishedRequests(std::deque<TString>& requests) {
@@ -1012,13 +1084,17 @@ class TFifoPoolHandlerActor : public TPoolHandlerActorBase<TFifoPoolHandlerActor
     TString PopPendingRequest() {
         TString sessionId = PendingRequests.front();
         PendingRequests.pop_front();
-        FifoCounters.PendingRequestsCount->Dec();
+        if (FifoCounters) {
+            FifoCounters.PendingRequestsCount->Dec();
+        }
         return sessionId;
     }
 
     void AddFinishedRequest(const TString& sessionId) {
         FinishedRequests.emplace_back(sessionId);
-        FifoCounters.FinishingRequestsCount->Inc();
+        if (FifoCounters) {
+            FifoCounters.FinishingRequestsCount->Inc();
+        }
     }
 
 private:

ydb/core/kqp/workload_service/kqp_workload_service.cpp

@@ -74,6 +74,7 @@ class TKqpWorkloadService : public TActorBootstrapped<TKqpWorkloadService> {
 
         EnabledResourcePools = AppData()->FeatureFlags.GetEnableResourcePools();
         EnabledResourcePoolsOnServerless = AppData()->FeatureFlags.GetEnableResourcePoolsOnServerless();
+        EnableResourcePoolsCounters = AppData()->FeatureFlags.GetEnableResourcePoolsCounters();
         if (EnabledResourcePools) {
             InitializeWorkloadService();
         }
@@ -101,6 +102,7 @@ class TKqpWorkloadService : public TActorBootstrapped<TKqpWorkloadService> {
 
         EnabledResourcePools = event.GetConfig().GetFeatureFlags().GetEnableResourcePools();
         EnabledResourcePoolsOnServerless = event.GetConfig().GetFeatureFlags().GetEnableResourcePoolsOnServerless();
+        EnableResourcePoolsCounters = event.GetConfig().GetFeatureFlags().GetEnableResourcePoolsCounters();
         if (EnabledResourcePools) {
             LOG_I("Resource pools was enanbled");
             InitializeWorkloadService();
@@ -526,7 +528,7 @@ class TKqpWorkloadService : public TActorBootstrapped<TKqpWorkloadService> {
 
         LOG_I("Creating new handler for pool " << poolKey);
 
-        const auto poolHandler = Register(CreatePoolHandlerActor(database, poolId, poolConfig, Counters.Counters));
+        const auto poolHandler = Register(CreatePoolHandlerActor(database, poolId, poolConfig, EnableResourcePoolsCounters ? Counters.Counters : nullptr));
         const auto poolState = &PoolIdToState.insert({poolKey, TPoolState{.PoolHandler = poolHandler, .ActorContext = ActorContext()}}).first->second;
 
         Counters.ActivePools->Inc();
@@ -560,6 +562,7 @@ class TKqpWorkloadService : public TActorBootstrapped<TKqpWorkloadService> {
 
     bool EnabledResourcePools = false;
     bool EnabledResourcePoolsOnServerless = false;
+    bool EnableResourcePoolsCounters = false;
     bool ServiceInitialized = false;
     bool IdleChecksStarted = false;
     ETablesCreationStatus TablesCreationStatus = ETablesCreationStatus::Cleanup;

ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.cpp

@@ -230,6 +230,7 @@ class TWorkloadServiceYdbSetup : public IYdbSetup {
     TAppConfig GetAppConfig() const {
         TAppConfig appConfig;
         appConfig.MutableFeatureFlags()->SetEnableResourcePools(Settings_.EnableResourcePools_);
+        appConfig.MutableFeatureFlags()->SetEnableResourcePoolsCounters(true);
 
         return appConfig;
     }

ydb/core/protos/feature_flags.proto

@@ -151,4 +151,5 @@ message TFeatureFlags {
     optional bool EnableResourcePoolsOnServerless = 132 [default = false];
     optional bool EnableVectorIndex = 133 [default = false];
     optional bool EnableChangefeedsOnIndexTables = 134 [default = false];
+    optional bool EnableResourcePoolsCounters = 135 [default = false];
 }