Merge bfe24b2 into 66d836f

vporyadke · web-flow · commit 51835ea6e368 · 2024-07-24T13:44:08.000Z
diff --git a/ydb/core/base/pool_stats_collector.cpp b/ydb/core/base/pool_stats_collector.cpp
@@ -46,9 +46,9 @@ class TStatsCollectingActor : public NActors::TStatsCollectingActor {
     void OnWakeup(const TActorContext &ctx) override {
         MiniKQLPoolStats.Update();
 
-        TVector<std::tuple<TString, double, ui32>> pools;
+        TVector<std::tuple<TString, double, ui32, ui32>> pools;
         for (const auto& pool : PoolCounters) {
-            pools.emplace_back(pool.Name, pool.Usage, pool.Threads);
+            pools.emplace_back(pool.Name, pool.Usage, pool.Threads, pool.LimitThreads);
         }
 
         ctx.Send(NNodeWhiteboard::MakeNodeWhiteboardServiceId(ctx.SelfID.NodeId()), new NNodeWhiteboard::TEvWhiteboard::TEvSystemStateUpdate(pools));
diff --git a/ydb/core/mind/hive/tx__update_tablet_metrics.cpp b/ydb/core/mind/hive/tx__update_tablet_metrics.cpp
@@ -54,6 +54,7 @@ class TTxUpdateTabletMetrics : public TTransactionBase<THive> {
         }
         TNodeInfo* node = Self->FindNode(nodeId);
         if (node != nullptr) {
+            node->UpdateResourceMaximum(record.GetResourceMaximum());
             node->UpdateResourceTotalUsage(record);
             node->Statistics.SetLastAliveTimestamp(now.MilliSeconds());
             node->ActualizeNodeStatistics(now);
diff --git a/ydb/core/mind/local.cpp b/ydb/core/mind/local.cpp
@@ -109,6 +109,7 @@ class TLocalNodeRegistrar : public TActorBootstrapped<TLocalNodeRegistrar> {
     ui64 UserPoolUsage = 0; // (usage uS x threads) / sec
     ui64 MemUsage = 0;
     ui64 MemLimit = 0;
+    ui64 CpuLimit = 0; // PotentialMaxThreadCount of UserPool
     double NodeUsage = 0;
 
     bool SentDrainNode = false;
@@ -272,28 +273,28 @@ class TLocalNodeRegistrar : public TActorBootstrapped<TLocalNodeRegistrar> {
         HandlePipeDestroyed(ctx);
     }
 
-    void SendStatusOk(const TActorContext &ctx) {
-        LOG_DEBUG_S(ctx, NKikimrServices::LOCAL, "TLocalNodeRegistrar SendStatusOk");
-        TAutoPtr<TEvLocal::TEvStatus> eventStatus = new TEvLocal::TEvStatus(TEvLocal::TEvStatus::StatusOk);
-        auto& record = eventStatus->Record;
-        record.SetStartTime(StartTime.GetValue());
-        record.MutableResourceMaximum()->CopyFrom(ResourceLimit);
-        if (!record.GetResourceMaximum().HasCPU()) {
-            TExecutorPoolStats poolStats;
-            TVector<TExecutorThreadStats> statsCopy;
-            TVector<TExecutorThreadStats> sharedStatsCopy;
-            ctx.ExecutorThread.ActorSystem->GetPoolStats(AppData()->UserPoolId, poolStats, statsCopy, sharedStatsCopy);
-            if (!statsCopy.empty()) {
-                record.MutableResourceMaximum()->SetCPU(poolStats.CurrentThreadCount * 1000000);
+    void FillResourceMaximum(NKikimrTabletBase::TMetrics* record) {
+        record->CopyFrom(ResourceLimit);
+        if (!record->HasCPU()) {
+            if (CpuLimit != 0) {
+                record->SetCPU(CpuLimit);
             }
         }
-        if (!record.GetResourceMaximum().HasMemory()) {
+        if (!record->HasMemory()) {
             if (MemLimit != 0) {
-                record.MutableResourceMaximum()->SetMemory(MemLimit);
+                record->SetMemory(MemLimit);
             } else {
-                record.MutableResourceMaximum()->SetMemory(NSystemInfo::TotalMemorySize());
+                record->SetMemory(NSystemInfo::TotalMemorySize());
             }
         }
+    }
+
+    void SendStatusOk(const TActorContext &ctx) {
+        LOG_DEBUG_S(ctx, NKikimrServices::LOCAL, "TLocalNodeRegistrar SendStatusOk");
+        TAutoPtr<TEvLocal::TEvStatus> eventStatus = new TEvLocal::TEvStatus(TEvLocal::TEvStatus::StatusOk);
+        auto& record = eventStatus->Record;
+        record.SetStartTime(StartTime.GetValue());
+        FillResourceMaximum(record.MutableResourceMaximum());
         NTabletPipe::SendData(ctx, HivePipeClient, eventStatus.Release());
     }
 
@@ -587,6 +588,7 @@ class TLocalNodeRegistrar : public TActorBootstrapped<TLocalNodeRegistrar> {
                 record.MutableTotalResourceUsage()->SetMemory(MemUsage);
             }
             record.SetTotalNodeUsage(NodeUsage);
+            FillResourceMaximum(record.MutableResourceMaximum());
             NTabletPipe::SendData(ctx, HivePipeClient, event.Release());
             SendTabletMetricsTime = ctx.Now();
         } else {
@@ -649,7 +651,8 @@ class TLocalNodeRegistrar : public TActorBootstrapped<TLocalNodeRegistrar> {
             const NKikimrWhiteboard::TSystemStateInfo& info = record.GetSystemStateInfo(0);
             if (static_cast<ui32>(info.PoolStatsSize()) > AppData()->UserPoolId) {
                 const auto& poolStats(info.GetPoolStats(AppData()->UserPoolId));
-                UserPoolUsage = poolStats.usage() * poolStats.threads() * 1000000; // uS
+                CpuLimit = poolStats.limit() * 1'000'000; // microseconds
+                UserPoolUsage = poolStats.usage() * CpuLimit; // microseconds
             }
 
             // Note: we use allocated memory because MemoryUsed(AnonRSS) has lag
diff --git a/ydb/core/node_whiteboard/node_whiteboard.h b/ydb/core/node_whiteboard/node_whiteboard.h
@@ -361,12 +361,13 @@ struct TEvWhiteboard{
             }
         }
 
-        TEvSystemStateUpdate(const TVector<std::tuple<TString, double, ui32>>& poolStats) {
+        TEvSystemStateUpdate(const TVector<std::tuple<TString, double, ui32, ui32>>& poolStats) {
             for (const auto& row : poolStats) {
                 auto& pb = *Record.AddPoolStats();
                 pb.SetName(std::get<0>(row));
                 pb.SetUsage(std::get<1>(row));
                 pb.SetThreads(std::get<2>(row));
+                pb.SetLimit(std::get<3>(row));
             }
         }
 
diff --git a/ydb/core/protos/hive.proto b/ydb/core/protos/hive.proto
@@ -246,6 +246,7 @@ message TEvTabletMetrics {
     repeated TTabletMetrics TabletMetrics = 1;
     optional NKikimrTabletBase.TMetrics TotalResourceUsage = 2;
     optional double TotalNodeUsage = 3;
+    optional NKikimrTabletBase.TMetrics ResourceMaximum = 4;
 }
 
 message TEvReassignTablet {
diff --git a/ydb/core/protos/node_whiteboard.proto b/ydb/core/protos/node_whiteboard.proto
@@ -277,6 +277,7 @@ message TSystemStateInfo {
         optional string Name = 1;
         optional double Usage = 2 [(InsignificantChangePercent) = 30];
         optional uint32 Threads = 3;
+        optional uint32 Limit = 4;
     }
 
     message TEndpoint {
diff --git a/ydb/library/actors/core/actorsystem.cpp b/ydb/library/actors/core/actorsystem.cpp
@@ -338,4 +338,13 @@ namespace NActors {
         CpuManager->Cleanup();
         Scheduler.Destroy();
     }
+
+    void TActorSystem::GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const {
+        CpuManager->GetExecutorPoolState(poolId, state);
+    }
+
+    void TActorSystem::GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const {
+        CpuManager->GetExecutorPoolStates(states);
+    }
+
 }
diff --git a/ydb/library/actors/core/actorsystem.h b/ydb/library/actors/core/actorsystem.h
@@ -306,5 +306,8 @@ namespace NActors {
             return CpuManager->GetBasicExecutorPools();
         }
 
+        void GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const;
+        void GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const;
+
     };
 }
diff --git a/ydb/library/actors/core/cpu_manager.cpp b/ydb/library/actors/core/cpu_manager.cpp
@@ -1,5 +1,6 @@
 #include "cpu_manager.h"
 #include "executor_pool_jail.h"
+#include "mon_stats.h"
 #include "probes.h"
 
 #include "executor_pool_basic.h"
@@ -172,4 +173,17 @@ namespace NActors {
         }
     }
 
+    void TCpuManager::GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const {
+        if (static_cast<ui32>(poolId) < ExecutorPoolCount) {
+            Executors[poolId]->GetExecutorPoolState(state);
+        }
+    }
+
+    void TCpuManager::GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const {
+        states.resize(ExecutorPoolCount);
+        for (i16 poolId = 0; poolId < static_cast<ui16>(ExecutorPoolCount); ++poolId) {
+            GetExecutorPoolState(poolId, states[poolId]);
+        }
+    }
+
 }
diff --git a/ydb/library/actors/core/cpu_manager.h b/ydb/library/actors/core/cpu_manager.h
@@ -5,6 +5,7 @@
 #include "harmonizer.h"
 #include "executor_pool.h"
 #include "executor_pool_shared.h"
+#include "mon_stats.h"
 #include <memory>
 
 namespace NActors {
@@ -47,6 +48,8 @@ namespace NActors {
         }
 
         void GetPoolStats(ui32 poolId, TExecutorPoolStats& poolStats, TVector<TExecutorThreadStats>& statsCopy, TVector<TExecutorThreadStats>& sharedStatsCopy) const;
+        void GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const;
+        void GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const;
 
         THarmonizerStats GetHarmonizerStats() const {
             if (Harmonizer) {
diff --git a/ydb/library/actors/core/executor_pool.h b/ydb/library/actors/core/executor_pool.h
@@ -9,6 +9,7 @@ namespace NActors {
     struct TMailboxHeader;
     struct TWorkerContext;
     struct TExecutorPoolStats;
+    struct TExecutorPoolState;
     struct TExecutorThreadStats;
     class TExecutorPoolJail;
     class ISchedulerCookie;
@@ -108,6 +109,10 @@ namespace NActors {
             Y_UNUSED(statsCopy);
         }
 
+        virtual void GetExecutorPoolState(TExecutorPoolState &poolState) const {
+            Y_UNUSED(poolState);
+        }
+
         virtual TString GetName() const {
             return TString();
         }
diff --git a/ydb/library/actors/core/executor_pool_basic.cpp b/ydb/library/actors/core/executor_pool_basic.cpp
@@ -9,6 +9,7 @@
 #include "mailbox.h"
 #include "thread_context.h"
 #include <atomic>
+#include <memory>
 #include <ydb/library/actors/util/affinity.h>
 #include <ydb/library/actors/util/datetime.h>
 
@@ -425,6 +426,19 @@ namespace NActors {
         }
     }
 
+    void TBasicExecutorPool::GetExecutorPoolState(TExecutorPoolState &poolState) const {
+        if (Harmonizer) {
+            TPoolHarmonizerStats stats = Harmonizer->GetPoolStats(PoolId);
+            poolState.UsedCpu = stats.AvgConsumedCpu;
+            poolState.PossibleMaxLimit = stats.PotentialMaxThreadCount;
+        } else {
+            poolState.PossibleMaxLimit = poolState.MaxLimit;
+        }
+        poolState.CurrentLimit = GetThreadCount();
+        poolState.MaxLimit = GetMaxThreadCount();
+        poolState.MinLimit = GetDefaultThreadCount();
+    }
+
     void TBasicExecutorPool::Prepare(TActorSystem* actorSystem, NSchedulerQueue::TReader** scheduleReaders, ui32* scheduleSz) {
         TAffinityGuard affinityGuard(Affinity());
 
diff --git a/ydb/library/actors/core/executor_pool_basic.h b/ydb/library/actors/core/executor_pool_basic.h
@@ -251,6 +251,7 @@ namespace NActors {
         void Shutdown() override;
 
         void GetCurrentStats(TExecutorPoolStats& poolStats, TVector<TExecutorThreadStats>& statsCopy) const override;
+        void GetExecutorPoolState(TExecutorPoolState &poolState) const override;
         TString GetName() const override {
             return PoolName;
         }
diff --git a/ydb/library/actors/core/executor_pool_io.cpp b/ydb/library/actors/core/executor_pool_io.cpp
@@ -12,14 +12,16 @@ namespace NActors {
         , PoolName(poolName)
     {}
 
-    TIOExecutorPool::TIOExecutorPool(const TIOExecutorPoolConfig& cfg)
+    TIOExecutorPool::TIOExecutorPool(const TIOExecutorPoolConfig& cfg, IHarmonizer *harmonizer)
         : TIOExecutorPool(
             cfg.PoolId,
             cfg.Threads,
             cfg.PoolName,
             new TAffinity(cfg.Affinity)
         )
-    {}
+    {
+        Harmonizer = harmonizer;
+    }
 
     TIOExecutorPool::~TIOExecutorPool() {
         Threads.Destroy();
@@ -148,6 +150,17 @@ namespace NActors {
         }
     }
 
+    void TIOExecutorPool::GetExecutorPoolState(TExecutorPoolState &poolState) const {
+        if (Harmonizer) {
+            TPoolHarmonizerStats stats = Harmonizer->GetPoolStats(PoolId);
+            poolState.UsedCpu = stats.AvgConsumedCpu;
+        }
+        poolState.CurrentLimit = PoolThreads;
+        poolState.MaxLimit = PoolThreads;
+        poolState.MinLimit = PoolThreads;
+        poolState.PossibleMaxLimit = PoolThreads;
+    }
+
     TString TIOExecutorPool::GetName() const {
         return PoolName;
     }
diff --git a/ydb/library/actors/core/executor_pool_io.h b/ydb/library/actors/core/executor_pool_io.h
@@ -3,6 +3,7 @@
 #include "actorsystem.h"
 #include "executor_thread.h"
 #include "executor_thread_ctx.h"
+#include "harmonizer.h"
 #include "scheduler_queue.h"
 #include "executor_pool_base.h"
 #include <ydb/library/actors/actor_type/indexes.h>
@@ -20,12 +21,13 @@ namespace NActors {
 
         THolder<NSchedulerQueue::TQueueType> ScheduleQueue;
         TTicketLock ScheduleLock;
+        IHarmonizer *Harmonizer = nullptr;
 
         const TString PoolName;
         const ui32 ActorSystemIndex = NActors::TActorTypeOperator::GetActorSystemIndex();
     public:
         TIOExecutorPool(ui32 poolId, ui32 threads, const TString& poolName = "", TAffinity* affinity = nullptr);
-        explicit TIOExecutorPool(const TIOExecutorPoolConfig& cfg);
+        explicit TIOExecutorPool(const TIOExecutorPoolConfig& cfg, IHarmonizer *harmonizer = nullptr);
         ~TIOExecutorPool();
 
         ui32 GetReadyActivation(TWorkerContext& wctx, ui64 revolvingCounter) override;
@@ -42,6 +44,7 @@ namespace NActors {
         void Shutdown() override;
 
         void GetCurrentStats(TExecutorPoolStats& poolStats, TVector<TExecutorThreadStats>& statsCopy) const override;
+        void GetExecutorPoolState(TExecutorPoolState &poolState) const override;
         TString GetName() const override;
     };
 }
diff --git a/ydb/library/actors/core/harmonizer.cpp b/ydb/library/actors/core/harmonizer.cpp
diff --git a/ydb/library/actors/core/harmonizer.h b/ydb/library/actors/core/harmonizer.h
diff --git a/ydb/library/actors/core/mon_stats.h b/ydb/library/actors/core/mon_stats.h
diff --git a/ydb/library/actors/helpers/pool_stats_collector.h b/ydb/library/actors/helpers/pool_stats_collector.h

Original file line number	Diff line number	Diff line change
`@@ -54,6 +54,7 @@ class TTxUpdateTabletMetrics : public TTransactionBase<THive> {`
`54`	`54`	`}`
`55`	`55`	`TNodeInfo* node = Self->FindNode(nodeId);`
`56`	`56`	`if (node != nullptr) {`
	`57`	`+ node->UpdateResourceMaximum(record.GetResourceMaximum());`
`57`	`58`	`node->UpdateResourceTotalUsage(record);`
`58`	`59`	`node->Statistics.SetLastAliveTimestamp(now.MilliSeconds());`
`59`	`60`	`node->ActualizeNodeStatistics(now);`
Original file line number	Diff line number	Diff line change
`@@ -361,12 +361,13 @@ struct TEvWhiteboard{`
`361`	`361`	`}`
`362`	`362`	`}`
`363`	`363`
`364`		`- TEvSystemStateUpdate(const TVector<std::tuple<TString, double, ui32>>& poolStats) {`
	`364`	`+ TEvSystemStateUpdate(const TVector<std::tuple<TString, double, ui32, ui32>>& poolStats) {`
`365`	`365`	`for (const auto& row : poolStats) {`
`366`	`366`	`auto& pb = *Record.AddPoolStats();`
`367`	`367`	`pb.SetName(std::get<0>(row));`
`368`	`368`	`pb.SetUsage(std::get<1>(row));`
`369`	`369`	`pb.SetThreads(std::get<2>(row));`
	`370`	`+ pb.SetLimit(std::get<3>(row));`
`370`	`371`	`}`
`371`	`372`	`}`
`372`	`373`
Original file line number	Diff line number	Diff line change
`@@ -246,6 +246,7 @@ message TEvTabletMetrics {`
`246`	`246`	`repeated TTabletMetrics TabletMetrics = 1;`
`247`	`247`	`optional NKikimrTabletBase.TMetrics TotalResourceUsage = 2;`
`248`	`248`	`optional double TotalNodeUsage = 3;`
	`249`	`+ optional NKikimrTabletBase.TMetrics ResourceMaximum = 4;`
`249`	`250`	`}`
`250`	`251`
`251`	`252`	`message TEvReassignTablet {`
Original file line number	Diff line number	Diff line change
`@@ -277,6 +277,7 @@ message TSystemStateInfo {`
`277`	`277`	`optional string Name = 1;`
`278`	`278`	`optional double Usage = 2 [(InsignificantChangePercent) = 30];`
`279`	`279`	`optional uint32 Threads = 3;`
	`280`	`+ optional uint32 Limit = 4;`
`280`	`281`	`}`
`281`	`282`
`282`	`283`	`message TEndpoint {`
Original file line number	Diff line number	Diff line change
`@@ -338,4 +338,13 @@ namespace NActors {`
`338`	`338`	`CpuManager->Cleanup();`
`339`	`339`	`Scheduler.Destroy();`
`340`	`340`	`}`
	`341`	`+`
	`342`	`+ void TActorSystem::GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const {`
	`343`	`+ CpuManager->GetExecutorPoolState(poolId, state);`
	`344`	`+ }`
	`345`	`+`
	`346`	`+ void TActorSystem::GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const {`
	`347`	`+ CpuManager->GetExecutorPoolStates(states);`
	`348`	`+ }`
	`349`	`+`
`341`	`350`	`}`
Original file line number	Diff line number	Diff line change
`@@ -306,5 +306,8 @@ namespace NActors {`
`306`	`306`	`return CpuManager->GetBasicExecutorPools();`
`307`	`307`	`}`
`308`	`308`
	`309`	`+ void GetExecutorPoolState(i16 poolId, TExecutorPoolState &state) const;`
	`310`	`+ void GetExecutorPoolStates(std::vector<TExecutorPoolState> &states) const;`
	`311`	`+`
`309`	`312`	`};`
`310`	`313`	`}`