KqpRun added CPU monitoring and yaml configs (#6675)

Author: GrigoriyPA

ydb/core/testlib/actors/test_runtime.cpp

@@ -3,6 +3,7 @@
 #include <ydb/core/base/appdata.h>
 #include <ydb/core/base/blobstorage.h>
 #include <ydb/core/base/counters.h>
+#include <ydb/core/base/pool_stats_collector.h>
 #include <ydb/core/mon/sync_http_mon.h>
 #include <ydb/core/mon/async_http_mon.h>
 #include <ydb/core/mon_alloc/profiler.h>
@@ -43,6 +44,10 @@ namespace NActors {
         InitNodes();
     }
 
+    void TTestActorRuntime::SetupStatsCollectors() {
+        NeedStatsCollectors = true;
+    }
+
     TTestActorRuntime::TTestActorRuntime(THeSingleSystemEnv d)
         : TPortManager(false)
         , TTestActorRuntimeBase{d}
@@ -212,7 +217,14 @@ namespace NActors {
         return MonPorts[nodeIndex];
     }
 
-    void TTestActorRuntime::InitActorSystemSetup(TActorSystemSetup& /*setup*/) {
+    void TTestActorRuntime::InitActorSystemSetup(TActorSystemSetup& setup, TNodeDataBase* node) {
+        if (NeedMonitoring && NeedStatsCollectors) {
+            NActors::IActor* statsCollector = NKikimr::CreateStatsCollector(1, setup, node->DynamicCounters);
+            setup.LocalServices.push_back({
+                TActorId(),
+                NActors::TActorSetupCmd(statsCollector, NActors::TMailboxType::HTSwap, node->GetAppData<NKikimr::TAppData>()->SystemPoolId)
+            });
+        }
     }
 
     NKikimr::TAppData& TTestActorRuntime::GetAppData(ui32 nodeIndex) {

ydb/core/testlib/actors/test_runtime.h

@@ -62,6 +62,7 @@ namespace NActors {
 
         void AddAppDataInit(std::function<void(ui32, NKikimr::TAppData&)> callback);
         virtual void Initialize(TEgg);
+        void SetupStatsCollectors();
 
         ui16 GetMonPort(ui32 nodeIndex = 0) const;
 
@@ -111,7 +112,7 @@ namespace NActors {
     private:
         void Initialize() override;
         TIntrusivePtr<::NMonitoring::TDynamicCounters> GetCountersForComponent(TIntrusivePtr<::NMonitoring::TDynamicCounters> counters, const char* component) override;
-        void InitActorSystemSetup(TActorSystemSetup& setup) override;
+        void InitActorSystemSetup(TActorSystemSetup& setup, TNodeDataBase* node) override;
 
         TNodeData* GetNodeById(size_t idx) override {
             return static_cast<TNodeData*>(TTestActorRuntimeBase::GetNodeById(idx));
@@ -129,5 +130,6 @@ namespace NActors {
         TVector<ui16> MonPorts;
         TActorId SleepEdgeActor;
         TVector<std::function<void(ui32, NKikimr::TAppData&)>> AppDataInit_;
+        bool NeedStatsCollectors = false;
     };
 } // namespace NActors

ydb/core/testlib/test_client.cpp

@@ -225,6 +225,9 @@ namespace Tests {
 
         NKikimr::SetupChannelProfiles(app);
 
+        if (Settings->NeedStatsCollectors) {
+            Runtime->SetupStatsCollectors();
+        }
         Runtime->SetupMonitoring(Settings->MonitoringPortOffset, Settings->MonitoringTypeAsync);
         Runtime->SetLogBackend(Settings->LogBackend);
 

ydb/core/testlib/test_client.h

@@ -109,6 +109,7 @@ namespace Tests {
         int GrpcMaxMessageSize = 0;  // 0 - default (4_MB), -1 - no limit
         ui16 MonitoringPortOffset = 0;
         bool MonitoringTypeAsync = false;
+        bool NeedStatsCollectors = false;
         NKikimrProto::TAuthConfig AuthConfig;
         NKikimrPQ::TPQConfig PQConfig;
         NKikimrPQ::TPQClusterDiscoveryConfig PQClusterDiscoveryConfig;
@@ -164,6 +165,7 @@ namespace Tests {
         TServerSettings& SetGrpcPort(ui16 value) { GrpcPort = value; return *this; }
         TServerSettings& SetGrpcMaxMessageSize(int value) { GrpcMaxMessageSize = value; return *this; }
         TServerSettings& SetMonitoringPortOffset(ui16 value, bool monitoringTypeAsync = false) { MonitoringPortOffset = value; MonitoringTypeAsync = monitoringTypeAsync; return *this; }
+        TServerSettings& SetNeedStatsCollectors(bool value) { NeedStatsCollectors = value; return *this; }
         TServerSettings& SetSupportsRedirect(bool value) { SupportsRedirect = value; return *this; }
         TServerSettings& SetTracePath(const TString& value) { TracePath = value; return *this; }
         TServerSettings& SetDomain(ui32 value) { Domain = value; return *this; }

ydb/library/actors/testlib/test_runtime.cpp

@@ -1688,14 +1688,20 @@ namespace NActors {
         THolder<TActorSystemSetup> setup(new TActorSystemSetup);
         setup->NodeId = FirstNodeId + nodeIndex;
 
+        IHarmonizer* harmonizer = nullptr;
+        if (node) {
+            node->Harmonizer.reset(MakeHarmonizer(GetCycleCountFast()));
+            harmonizer = node->Harmonizer.get();
+        }
+
         if (UseRealThreads) {
             setup->ExecutorsCount = 5;
             setup->Executors.Reset(new TAutoPtr<IExecutorPool>[5]);
-            setup->Executors[0].Reset(new TBasicExecutorPool(0, 2, 20));
-            setup->Executors[1].Reset(new TBasicExecutorPool(1, 2, 20));
-            setup->Executors[2].Reset(new TIOExecutorPool(2, 1));
-            setup->Executors[3].Reset(new TBasicExecutorPool(3, 2, 20));
-            setup->Executors[4].Reset(new TBasicExecutorPool(4, 1, 20));
+            setup->Executors[0].Reset(new TBasicExecutorPool(0, 2, 20, "System", harmonizer));
+            setup->Executors[1].Reset(new TBasicExecutorPool(1, 2, 20, "User", harmonizer));
+            setup->Executors[2].Reset(new TIOExecutorPool(2, 1, "IO"));
+            setup->Executors[3].Reset(new TBasicExecutorPool(3, 2, 20, "Batch", harmonizer));
+            setup->Executors[4].Reset(new TBasicExecutorPool(4, 1, 20, "IC", harmonizer));
             setup->Scheduler.Reset(new TBasicSchedulerThread(TSchedulerConfig(512, 100)));
         } else {
             setup->ExecutorsCount = 1;
@@ -1704,7 +1710,7 @@ namespace NActors {
             setup->Executors[0].Reset(new TExecutorPoolStub(this, nodeIndex, node, 0));
         }
 
-        InitActorSystemSetup(*setup);
+        InitActorSystemSetup(*setup, node);
 
         return setup;
     }
@@ -1714,7 +1720,9 @@ namespace NActors {
 
         node->ExecutorPools.resize(setup->ExecutorsCount);
         for (ui32 i = 0; i < setup->ExecutorsCount; ++i) {
-            node->ExecutorPools[i] = setup->Executors[i].Get();
+            IExecutorPool* executor = setup->Executors[i].Get();
+            node->ExecutorPools[i] = executor;
+            node->Harmonizer->AddPool(executor);
         }
 
         const auto& interconnectCounters = GetCountersForComponent(node->DynamicCounters, "interconnect");

ydb/library/actors/testlib/test_runtime.h

@@ -618,8 +618,8 @@ namespace NActors {
 
         THolder<TActorSystemSetup> MakeActorSystemSetup(ui32 nodeIndex, TNodeDataBase* node);
         THolder<TActorSystem> MakeActorSystem(ui32 nodeIndex, TNodeDataBase* node);
-        virtual void InitActorSystemSetup(TActorSystemSetup& setup) {
-            Y_UNUSED(setup);
+        virtual void InitActorSystemSetup(TActorSystemSetup& setup, TNodeDataBase* node) {
+            Y_UNUSED(setup, node);
         }
 
    private:
@@ -703,6 +703,7 @@ namespace NActors {
             THolder<IExecutorPool> SchedulerPool;
             TVector<IExecutorPool*> ExecutorPools;
             THolder<TExecutorThread> ExecutorThread;
+            std::unique_ptr<IHarmonizer> Harmonizer;
         };
 
         struct INodeFactory {

ydb/tests/tools/kqprun/kqprun.cpp

@@ -13,6 +13,7 @@
 
 #include <ydb/core/base/backtrace.h>
 
+#include <ydb/library/yaml_config/yaml_config.h>
 #include <ydb/library/yql/minikql/invoke_builtins/mkql_builtins.h>
 #include <ydb/library/yql/providers/yt/gateway/file/yql_yt_file.h>
 #include <ydb/library/yql/providers/yt/gateway/file/yql_yt_file_comp_nodes.h>
@@ -305,7 +306,10 @@ class TMain : public TMainClassArgs {
             .DefaultValue("./configuration/app_config.conf")
             .Handler1([this](const NLastGetopt::TOptsParser* option) {
                 TString file(option->CurValOrDef());
-                if (!google::protobuf::TextFormat::ParseFromString(LoadFile(file), &RunnerOptions.YdbSettings.AppConfig)) {
+                if (file.EndsWith(".yaml")) {
+                    auto document = NKikimr::NFyaml::TDocument::Parse(LoadFile(file));
+                    RunnerOptions.YdbSettings.AppConfig = NKikimr::NYamlConfig::YamlToProto(document.Root());
+                } else if (!google::protobuf::TextFormat::ParseFromString(LoadFile(file), &RunnerOptions.YdbSettings.AppConfig)) {
                     ythrow yexception() << "Bad format of app configuration";
                 }
             });

ydb/tests/tools/kqprun/src/actors.h

@@ -1,3 +1,5 @@
+#pragma once
+
 #include <ydb/core/kqp/common/events/events.h>
 #include <ydb/core/kqp/executer_actor/kqp_executer.h>
 

ydb/tests/tools/kqprun/src/ydb_setup.cpp

@@ -141,6 +141,7 @@ class TYdbSetup::TImpl {
         if (Settings_.MonitoringEnabled) {
             serverSettings.InitKikimrRunConfig();
             serverSettings.SetMonitoringPortOffset(Settings_.MonitoringPortOffset);
+            serverSettings.SetNeedStatsCollectors(true);
         }
 
         return serverSettings;