Skip to content

Commit 65176c7

Browse files
GrigoriyPAGrigoriyPA
committed
KqpRun improved multi query mode and storage settings (ydb-platform#13177)
1 parent 95333f6 commit 65176c7

File tree

20 files changed

+519
-177
lines changed

20 files changed

+519
-177
lines changed

‎ydb/core/driver_lib/run/config_helpers.cpp‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
#include "config_helpers.h"
22

3+
#include <ydb/core/base/localdb.h>
4+
#include <ydb/core/protos/bootstrap.pb.h>
5+
#include <ydb/core/protos/resource_broker.pb.h>
6+
37
#include <ydb/library/actors/util/affinity.h>
48

59

@@ -114,4 +118,31 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
114118

115119
} // namespace NActorSystemConfigHelpers
116120

121+
namespace NKikimrConfigHelpers {
122+
123+
NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config) {
124+
NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
125+
auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
126+
if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
127+
resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
128+
}
129+
for (const auto& queue : resourceBrokerConfig.GetQueues()) {
130+
if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
131+
if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
132+
resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
133+
}
134+
}
135+
}
136+
};
137+
if (config.HasBootstrapConfig() && config.GetBootstrapConfig().HasResourceBroker()) {
138+
mergeResourceBrokerConfigs(config.GetBootstrapConfig().GetResourceBroker());
139+
}
140+
if (config.HasResourceBrokerConfig()) {
141+
mergeResourceBrokerConfigs(config.GetResourceBrokerConfig());
142+
}
143+
return resourceBrokerSelfConfig;
144+
}
145+
146+
} // namespace NKikimrConfigHelpers
147+
117148
} // namespace NKikimr

‎ydb/core/driver_lib/run/config_helpers.h‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#pragma once
22

3+
#include <ydb/core/memory_controller/memory_controller.h>
34
#include <ydb/core/protos/config.pb.h>
45

56
#include <ydb/library/actors/core/config.h>
@@ -15,4 +16,10 @@ NActors::TSchedulerConfig CreateSchedulerConfig(const NKikimrConfig::TActorSyste
1516

1617
} // namespace NActorSystemConfigHelpers
1718

19+
namespace NKikimrConfigHelpers {
20+
21+
NMemory::TResourceBrokerConfig CreateMemoryControllerResourceBrokerConfig(const NKikimrConfig::TAppConfig& config);
22+
23+
} // namespace NKikimrConfigHelpers
24+
1825
} // namespace NKikimr

‎ydb/core/driver_lib/run/kikimr_services_initializers.cpp‎

Lines changed: 1 addition & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -2049,28 +2049,8 @@ void TMemoryControllerInitializer::InitializeServices(
20492049
NActors::TActorSystemSetup* setup,
20502050
const NKikimr::TAppData* appData)
20512051
{
2052-
NMemory::TResourceBrokerConfig resourceBrokerSelfConfig; // for backward compatibility
2053-
auto mergeResourceBrokerConfigs = [&](const NKikimrResourceBroker::TResourceBrokerConfig& resourceBrokerConfig) {
2054-
if (resourceBrokerConfig.HasResourceLimit() && resourceBrokerConfig.GetResourceLimit().HasMemory()) {
2055-
resourceBrokerSelfConfig.LimitBytes = resourceBrokerConfig.GetResourceLimit().GetMemory();
2056-
}
2057-
for (const auto& queue : resourceBrokerConfig.GetQueues()) {
2058-
if (queue.GetName() == NLocalDb::KqpResourceManagerQueue) {
2059-
if (queue.HasLimit() && queue.GetLimit().HasMemory()) {
2060-
resourceBrokerSelfConfig.QueryExecutionLimitBytes = queue.GetLimit().GetMemory();
2061-
}
2062-
}
2063-
}
2064-
};
2065-
if (Config.HasBootstrapConfig() && Config.GetBootstrapConfig().HasResourceBroker()) {
2066-
mergeResourceBrokerConfigs(Config.GetBootstrapConfig().GetResourceBroker());
2067-
}
2068-
if (Config.HasResourceBrokerConfig()) {
2069-
mergeResourceBrokerConfigs(Config.GetResourceBrokerConfig());
2070-
}
2071-
20722052
auto* actor = NMemory::CreateMemoryController(TDuration::Seconds(1), ProcessMemoryInfoProvider,
2073-
Config.GetMemoryControllerConfig(), resourceBrokerSelfConfig,
2053+
Config.GetMemoryControllerConfig(), NKikimrConfigHelpers::CreateMemoryControllerResourceBrokerConfig(Config),
20742054
appData->Counters);
20752055
setup->LocalServices.emplace_back(
20762056
NMemory::MakeMemoryControllerId(0),

‎ydb/core/testlib/basics/helpers.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@ namespace NFake {
1818
ui64 SectorSize = 0;
1919
ui64 ChunkSize = 0;
2020
ui64 DiskSize = 0;
21+
bool FormatDisk = true;
22+
TString DiskPath;
2123
};
2224

2325
struct INode {

‎ydb/core/testlib/basics/storage.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,15 +54,15 @@ namespace NKikimr {
5454

5555
static ui64 keySalt = 0;
5656
ui64 salt = ++keySalt;
57-
TString baseDir = Runtime.GetTempDir();
57+
TString baseDir = conf.DiskPath ? conf.DiskPath : Runtime.GetTempDir();
5858

5959
if (Conf.UseDisk) {
6060
MakeDirIfNotExist(baseDir.c_str());
6161
}
6262

6363
PDiskPath = TStringBuilder() << baseDir << "pdisk_1.dat";
6464

65-
if (!Mock) {
65+
if (!Mock && conf.FormatDisk) {
6666
FormatPDisk(PDiskPath,
6767
Conf.DiskSize, Conf.SectorSize, Conf.ChunkSize, PDiskGuid,
6868
0x123 + salt, 0x456 + salt, 0x789 + salt, mainKey,

‎ydb/core/testlib/test_client.cpp‎

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@
9393
#include <ydb/core/mind/tenant_slot_broker.h>
9494
#include <ydb/core/mind/tenant_node_enumeration.h>
9595
#include <ydb/core/mind/node_broker.h>
96+
#include <ydb/core/mon_alloc/monitor.h>
9697
#include <ydb/core/kesus/tablet/events.h>
9798
#include <ydb/core/sys_view/service/sysview_service.h>
9899
#include <yql/essentials/minikql/mkql_function_registry.h>
@@ -610,18 +611,21 @@ namespace Tests {
610611

611612
NKikimrBlobStorage::TDefineBox boxConfig;
612613
boxConfig.SetBoxId(Settings->BOX_ID);
614+
boxConfig.SetItemConfigGeneration(Settings->StorageGeneration);
613615

614616
ui32 nodeId = Runtime->GetNodeId(0);
615617
Y_ABORT_UNLESS(nodesInfo->Nodes[0].NodeId == nodeId);
616618
auto& nodeInfo = nodesInfo->Nodes[0];
617619

618620
NKikimrBlobStorage::TDefineHostConfig hostConfig;
619621
hostConfig.SetHostConfigId(nodeId);
622+
hostConfig.SetItemConfigGeneration(Settings->StorageGeneration);
620623
TString path;
621624
if (Settings->UseSectorMap) {
622625
path ="SectorMap:test-client[:2000]";
623626
} else {
624-
path = TStringBuilder() << Runtime->GetTempDir() << "pdisk_1.dat";
627+
TString diskPath = Settings->CustomDiskParams.DiskPath;
628+
path = TStringBuilder() << (diskPath ? diskPath : Runtime->GetTempDir()) << "pdisk_1.dat";
625629
}
626630
hostConfig.AddDrive()->SetPath(path);
627631
if (Settings->Verbose) {
@@ -637,7 +641,9 @@ namespace Tests {
637641

638642
for (const auto& [poolKind, storagePool] : Settings->StoragePoolTypes) {
639643
if (storagePool.GetNumGroups() > 0) {
640-
bsConfigureRequest->Record.MutableRequest()->AddCommand()->MutableDefineStoragePool()->CopyFrom(storagePool);
644+
auto* command = bsConfigureRequest->Record.MutableRequest()->AddCommand()->MutableDefineStoragePool();
645+
command->CopyFrom(storagePool);
646+
command->SetItemConfigGeneration(Settings->StorageGeneration);
641647
}
642648
}
643649

@@ -1071,6 +1077,28 @@ namespace Tests {
10711077
}
10721078
}
10731079

1080+
{
1081+
if (Settings->NeedStatsCollectors) {
1082+
TString filePathPrefix;
1083+
if (Settings->AppConfig->HasMonitoringConfig()) {
1084+
filePathPrefix = Settings->AppConfig->GetMonitoringConfig().GetMemAllocDumpPathPrefix();
1085+
}
1086+
1087+
const TIntrusivePtr<NMemory::IProcessMemoryInfoProvider> processMemoryInfoProvider(MakeIntrusive<NMemory::TProcessMemoryInfoProvider>());
1088+
1089+
IActor* monitorActor = CreateMemProfMonitor(TDuration::Seconds(1), processMemoryInfoProvider,
1090+
Runtime->GetAppData(nodeIdx).Counters, filePathPrefix);
1091+
const TActorId monitorActorId = Runtime->Register(monitorActor, nodeIdx, Runtime->GetAppData(nodeIdx).BatchPoolId);
1092+
Runtime->RegisterService(MakeMemProfMonitorID(Runtime->GetNodeId(nodeIdx)), monitorActorId, nodeIdx);
1093+
1094+
IActor* controllerActor = NMemory::CreateMemoryController(TDuration::Seconds(1), processMemoryInfoProvider,
1095+
Settings->AppConfig->GetMemoryControllerConfig(), NKikimrConfigHelpers::CreateMemoryControllerResourceBrokerConfig(*Settings->AppConfig),
1096+
Runtime->GetAppData(nodeIdx).Counters);
1097+
const TActorId controllerActorId = Runtime->Register(controllerActor, nodeIdx, Runtime->GetAppData(nodeIdx).BatchPoolId);
1098+
Runtime->RegisterService(NMemory::MakeMemoryControllerId(0), controllerActorId, nodeIdx);
1099+
}
1100+
}
1101+
10741102
{
10751103
IActor* kesusService = NKesus::CreateKesusProxyService();
10761104
TActorId kesusServiceId = Runtime->Register(kesusService, nodeIdx, userPoolId);

‎ydb/core/testlib/test_client.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,7 @@ namespace Tests {
122122
TString DomainName = TestDomainName;
123123
ui32 NodeCount = 1;
124124
ui32 DynamicNodeCount = 0;
125+
ui64 StorageGeneration = 0;
125126
NFake::TStorage CustomDiskParams;
126127
TControls Controls;
127128
TAppPrepare::TFnReg FrFactory = &DefaultFrFactory;
@@ -177,6 +178,7 @@ namespace Tests {
177178
TServerSettings& SetDomainName(const TString& value);
178179
TServerSettings& SetNodeCount(ui32 value) { NodeCount = value; return *this; }
179180
TServerSettings& SetDynamicNodeCount(ui32 value) { DynamicNodeCount = value; return *this; }
181+
TServerSettings& SetStorageGeneration(ui64 value) { StorageGeneration = value; return *this; }
180182
TServerSettings& SetCustomDiskParams(const NFake::TStorage& value) { CustomDiskParams = value; return *this; }
181183
TServerSettings& SetControls(const TControls& value) { Controls = value; return *this; }
182184
TServerSettings& SetFrFactory(const TAppPrepare::TFnReg& value) { FrFactory = value; return *this; }

‎ydb/tests/tools/kqprun/.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
1+
storage
12
sync_dir
23
example
34
udfs
5+
46
*.log
57
*.json
68
*.sql

‎ydb/tests/tools/kqprun/flame_graph.sh‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
#!/usr/bin/env bash
22

3-
# For svg graph download https://github.com/brendangregg/FlameGraph
4-
# and run `FlameGraph/stackcollapse-perf.pl profdata.txt | FlameGraph/flamegraph.pl > profdata.svg`
3+
set -eux
54

6-
pid=$(pgrep -u $USER kqprun)
5+
kqprun_pid=$(pgrep -u $USER kqprun)
76

8-
echo "Target process id: ${pid}"
9-
10-
sudo perf record -F 50 --call-graph dwarf -g --proc-map-timeout=10000 --pid $pid -v -o profdata -- sleep 30
7+
sudo perf record -F 50 --call-graph dwarf -g --proc-map-timeout=10000 --pid $kqprun_pid -v -o profdata -- sleep ${1:-'30'}
118
sudo perf script -i profdata > profdata.txt
9+
10+
flame_graph_tool="../../../../contrib/tools/flame-graph/"
11+
12+
${flame_graph_tool}/stackcollapse-perf.pl profdata.txt | ${flame_graph_tool}/flamegraph.pl > profdata.svg

0 commit comments

Comments
 (0)