aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAleksandr Dmitriev <monster@ydb.tech>2024-01-30 15:54:31 +0300
committerGitHub <noreply@github.com>2024-01-30 15:54:31 +0300
commit3311d10b22651f987a52656368ee1d1050ccdacf (patch)
tree4eb550eae11babd78e9b6688cc128f7d27901f4d
parentae12905f2832ace12b9e9e480bd362d11f785b4d (diff)
downloadydb-3311d10b22651f987a52656368ee1d1050ccdacf.tar.gz
create statistics aggregator tablet during migration (#1393)
-rw-r--r--ydb/core/protos/statistics.proto4
-rw-r--r--ydb/core/statistics/aggregator/aggregator_impl.cpp53
-rw-r--r--ydb/core/statistics/aggregator/aggregator_impl.h15
-rw-r--r--ydb/core/statistics/aggregator/tx_init.cpp7
-rw-r--r--ydb/core/statistics/events.h7
-rw-r--r--ydb/core/statistics/stat_service.cpp76
-rw-r--r--ydb/core/tx/schemeshard/schemeshard_impl.cpp95
-rw-r--r--ydb/core/tx/schemeshard/schemeshard_impl.h4
-rw-r--r--ydb/core/tx/schemeshard/schemeshard_svp_migration.cpp44
-rw-r--r--ydb/core/tx/schemeshard/schemeshard_svp_migration.h8
-rw-r--r--ydb/library/services/services.proto2
11 files changed, 221 insertions, 94 deletions
diff --git a/ydb/core/protos/statistics.proto b/ydb/core/protos/statistics.proto
index b759cf34f3..88efbd1b70 100644
--- a/ydb/core/protos/statistics.proto
+++ b/ydb/core/protos/statistics.proto
@@ -56,3 +56,7 @@ message TEvPropagateStatistics {
}
repeated TStatsEntry Entries = 2;
}
+
+// SA -> nodes
+message TEvStatisticsIsDisabled {
+}
diff --git a/ydb/core/statistics/aggregator/aggregator_impl.cpp b/ydb/core/statistics/aggregator/aggregator_impl.cpp
index 0388fb1c7c..07f98b283b 100644
--- a/ydb/core/statistics/aggregator/aggregator_impl.cpp
+++ b/ydb/core/statistics/aggregator/aggregator_impl.cpp
@@ -35,6 +35,27 @@ void TStatisticsAggregator::DefaultSignalTabletActive(const TActorContext& ctx)
Y_UNUSED(ctx);
}
+void TStatisticsAggregator::SubscribeForConfigChanges(const TActorContext& ctx) {
+ ui32 configKind = (ui32)NKikimrConsole::TConfigItem::FeatureFlagsItem;
+ ctx.Send(NConsole::MakeConfigsDispatcherID(ctx.SelfID.NodeId()),
+ new NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionRequest({configKind}));
+}
+
+void TStatisticsAggregator::HandleConfig(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse::TPtr&) {
+ SA_LOG_I("[" << TabletID() << "] Subscribed for config changes");
+}
+
+void TStatisticsAggregator::HandleConfig(NConsole::TEvConsole::TEvConfigNotificationRequest::TPtr& ev) {
+ const auto& record = ev->Get()->Record;
+ const auto& config = record.GetConfig();
+ if (config.HasFeatureFlags()) {
+ const auto& featureFlags = config.GetFeatureFlags();
+ EnableStatistics = featureFlags.GetEnableStatistics();
+ }
+ auto response = std::make_unique<NConsole::TEvConsole::TEvConfigNotificationResponse>(record);
+ Send(ev->Sender, response.release(), 0, ev->Cookie);
+}
+
void TStatisticsAggregator::Handle(TEvTabletPipe::TEvServerConnected::TPtr &ev) {
auto pipeServerId = ev->Get()->ServerId;
@@ -97,6 +118,12 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvConnectNode::TPtr& ev) {
RequestedSchemeShards.insert(ssEntry.GetSchemeShardId());
}
+ if (!EnableStatistics) {
+ auto disabled = std::make_unique<TEvStatistics::TEvStatisticsIsDisabled>();
+ Send(NStat::MakeStatServiceID(nodeId), disabled.release());
+ return;
+ }
+
if (!IsPropagateInFlight) {
Schedule(PropagateInterval, new TEvPrivate::TEvPropagate());
IsPropagateInFlight = true;
@@ -124,6 +151,12 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvRequestStats::TPtr& ev) {
<< ", node id = " << nodeId
<< ", schemeshard count = " << record.NeedSchemeShardsSize());
+ if (!EnableStatistics) {
+ auto disabled = std::make_unique<TEvStatistics::TEvStatisticsIsDisabled>();
+ Send(NStat::MakeStatServiceID(nodeId), disabled.release());
+ return;
+ }
+
std::vector<TSSId> ssIds;
ssIds.reserve(record.NeedSchemeShardsSize());
for (const auto& ssId : record.GetNeedSchemeShards()) {
@@ -151,6 +184,10 @@ void TStatisticsAggregator::Handle(TEvStatistics::TEvConnectSchemeShard::TPtr& e
void TStatisticsAggregator::Handle(TEvPrivate::TEvFastPropagateCheck::TPtr&) {
SA_LOG_D("[" << TabletID() << "] EvFastPropagateCheck");
+ if (!EnableStatistics) {
+ return;
+ }
+
PropagateFastStatistics();
FastCheckInFlight = false;
@@ -162,7 +199,9 @@ void TStatisticsAggregator::Handle(TEvPrivate::TEvFastPropagateCheck::TPtr&) {
void TStatisticsAggregator::Handle(TEvPrivate::TEvPropagate::TPtr&) {
SA_LOG_D("[" << TabletID() << "] EvPropagate");
- PropagateStatistics();
+ if (EnableStatistics) {
+ PropagateStatistics();
+ }
Schedule(PropagateInterval, new TEvPrivate::TEvPropagate());
}
@@ -176,10 +215,10 @@ void TStatisticsAggregator::ProcessRequests(TNodeId nodeId, const std::vector<TS
for (const auto& ssId : ssIds) {
FastSchemeShards.insert(ssId);
}
- if (!FastCheckInFlight) {
- Schedule(TDuration::MilliSeconds(100), new TEvPrivate::TEvFastPropagateCheck());
- FastCheckInFlight = true;
- }
+ }
+ if (!FastCheckInFlight) {
+ Schedule(TDuration::MilliSeconds(100), new TEvPrivate::TEvFastPropagateCheck());
+ FastCheckInFlight = true;
}
}
@@ -247,6 +286,10 @@ void TStatisticsAggregator::PropagateFastStatistics() {
void TStatisticsAggregator::PropagateStatisticsImpl(
const std::vector<TNodeId>& nodeIds, const std::vector<TSSId>& ssIds)
{
+ if (nodeIds.empty() || ssIds.empty()) {
+ return;
+ }
+
TNodeId leadingNodeId = nodeIds[0];
for (size_t index = 0; index < ssIds.size(); ) {
diff --git a/ydb/core/statistics/aggregator/aggregator_impl.h b/ydb/core/statistics/aggregator/aggregator_impl.h
index 5597fffb9b..d2ce8c500b 100644
--- a/ydb/core/statistics/aggregator/aggregator_impl.h
+++ b/ydb/core/statistics/aggregator/aggregator_impl.h
@@ -9,6 +9,9 @@
#include <ydb/core/statistics/common.h>
#include <ydb/core/statistics/events.h>
+#include <ydb/core/cms/console/configs_dispatcher.h>
+#include <ydb/core/cms/console/console.h>
+
#include <ydb/core/tablet_flat/tablet_flat_executed.h>
#include <random>
@@ -53,10 +56,14 @@ private:
void OnActivateExecutor(const TActorContext& ctx) override;
void DefaultSignalTabletActive(const TActorContext& ctx) override;
bool OnRenderAppHtmlPage(NMon::TEvRemoteHttpInfo::TPtr ev, const TActorContext &ctx) override;
+ void SubscribeForConfigChanges(const TActorContext& ctx);
NTabletFlatExecutor::ITransaction* CreateTxInitSchema();
NTabletFlatExecutor::ITransaction* CreateTxInit();
+ void HandleConfig(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse::TPtr& ev);
+ void HandleConfig(NConsole::TEvConsole::TEvConfigNotificationRequest::TPtr& ev);
+
void Handle(TEvStatistics::TEvConfigureAggregator::TPtr& ev);
void Handle(TEvStatistics::TEvSchemeShardStats::TPtr& ev);
void Handle(TEvPrivate::TEvPropagate::TPtr& ev);
@@ -76,11 +83,13 @@ private:
void PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value);
STFUNC(StateInit) {
- StateInitImpl(ev,SelfId());
+ StateInitImpl(ev, SelfId());
}
STFUNC(StateWork) {
switch(ev->GetTypeRewrite()) {
+ hFunc(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse, HandleConfig)
+ hFunc(NConsole::TEvConsole::TEvConfigNotificationRequest, HandleConfig)
hFunc(TEvStatistics::TEvConfigureAggregator, Handle);
hFunc(TEvStatistics::TEvSchemeShardStats, Handle);
hFunc(TEvPrivate::TEvPropagate, Handle);
@@ -103,10 +112,12 @@ private:
std::mt19937_64 RandomGenerator;
+ bool EnableStatistics = false;
+
static constexpr size_t StatsOptimizeFirstNodesCount = 3; // optimize first nodes - fast propagation
static constexpr size_t StatsSizeLimitBytes = 2 << 20; // limit for stats size in one message
- TDuration PropagateInterval = TDuration::Minutes(3);
+ TDuration PropagateInterval;
bool IsPropagateInFlight = false; // is slow propagation started
std::unordered_map<TSSId, TString> BaseStats; // schemeshard id -> serialized stats for all paths
diff --git a/ydb/core/statistics/aggregator/tx_init.cpp b/ydb/core/statistics/aggregator/tx_init.cpp
index 8cfb57b8cc..88f6428ded 100644
--- a/ydb/core/statistics/aggregator/tx_init.cpp
+++ b/ydb/core/statistics/aggregator/tx_init.cpp
@@ -1,5 +1,8 @@
#include "aggregator_impl.h"
+#include <ydb/core/base/appdata_fwd.h>
+#include <ydb/core/base/feature_flags.h>
+
namespace NKikimr::NStat {
struct TStatisticsAggregator::TTxInit : public TTxBase {
@@ -82,6 +85,10 @@ struct TStatisticsAggregator::TTxInit : public TTxBase {
SA_LOG_D("[" << Self->TabletID() << "] TTxInit::Complete");
Self->SignalTabletActive(ctx);
+
+ Self->EnableStatistics = AppData(ctx)->FeatureFlags.GetEnableStatistics();
+ Self->SubscribeForConfigChanges(ctx);
+
Self->Become(&TThis::StateWork);
}
};
diff --git a/ydb/core/statistics/events.h b/ydb/core/statistics/events.h
index 055e8c0caf..a24600338a 100644
--- a/ydb/core/statistics/events.h
+++ b/ydb/core/statistics/events.h
@@ -54,6 +54,7 @@ struct TEvStatistics {
EvConnectNode,
EvRequestStats,
EvPropagateStatistics,
+ EvStatisticsIsDisabled,
EvEnd
};
@@ -108,6 +109,12 @@ struct TEvStatistics {
NKikimrStat::TEvPropagateStatistics,
EvPropagateStatistics>
{};
+
+ struct TEvStatisticsIsDisabled : public TEventPB<
+ TEvStatisticsIsDisabled,
+ NKikimrStat::TEvStatisticsIsDisabled,
+ EvStatisticsIsDisabled>
+ {};
};
} // NStat
diff --git a/ydb/core/statistics/stat_service.cpp b/ydb/core/statistics/stat_service.cpp
index 9a4c802ca8..ef2864e920 100644
--- a/ydb/core/statistics/stat_service.cpp
+++ b/ydb/core/statistics/stat_service.cpp
@@ -43,6 +43,7 @@ public:
hFunc(TEvStatistics::TEvPropagateStatistics, Handle);
hFunc(TEvTabletPipe::TEvClientConnected, Handle);
hFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
+ hFunc(TEvStatistics::TEvStatisticsIsDisabled, Handle);
cFunc(TEvents::TEvPoison::EventType, PassAway);
default:
LOG_CRIT_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
@@ -51,20 +52,21 @@ public:
}
private:
- bool IsSAUnavailable() {
- return ResolveSAStage == RSA_FINISHED && StatisticsAggregatorId == 0;
- }
-
void HandleConfig(NConsole::TEvConfigsDispatcher::TEvSetConfigSubscriptionResponse::TPtr&) {
LOG_INFO_S(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
- "Subscribed for config changes");
+ "Subscribed for config changes on node " << SelfId().NodeId());
}
void HandleConfig(NConsole::TEvConsole::TEvConfigNotificationRequest::TPtr& ev) {
const auto& record = ev->Get()->Record;
- const auto& featureFlags = record.GetConfig().GetFeatureFlags();
- EnableStatistics = featureFlags.GetEnableStatistics();
-
+ const auto& config = record.GetConfig();
+ if (config.HasFeatureFlags()) {
+ const auto& featureFlags = config.GetFeatureFlags();
+ EnableStatistics = featureFlags.GetEnableStatistics();
+ if (!EnableStatistics) {
+ ReplyAllFailed();
+ }
+ }
auto response = std::make_unique<NConsole::TEvConsole::TEvConfigNotificationResponse>(record);
Send(ev->Sender, response.release(), 0, ev->Cookie);
}
@@ -77,7 +79,7 @@ private:
request.EvCookie = ev->Cookie;
request.StatRequests.swap(ev->Get()->StatRequests);
- if (!EnableStatistics || IsSAUnavailable()) {
+ if (!EnableStatistics) {
ReplyFailed(requestId, true);
return;
}
@@ -106,12 +108,12 @@ private:
auto& entry = navigate->ResultSet.back();
if (entry.Status != TNavigate::EStatus::Ok) {
StatisticsAggregatorId = 0;
- } else {
+ } else if (entry.DomainInfo->Params.HasStatisticsAggregator()) {
StatisticsAggregatorId = entry.DomainInfo->Params.GetStatisticsAggregator();
}
- ResolveSAStage = RSA_FINISHED;
+ ResolveSAStage = StatisticsAggregatorId ? RSA_FINISHED : RSA_INITIAL;
- if (StatisticsAggregatorId != 0) {
+ if (StatisticsAggregatorId) {
ConnectToSA();
SyncNode();
} else {
@@ -127,7 +129,7 @@ private:
}
auto& request = itRequest->second;
- if (!EnableStatistics || IsSAUnavailable()) {
+ if (!EnableStatistics) {
ReplyFailed(requestId, true);
return;
}
@@ -135,7 +137,7 @@ private:
std::unordered_set<ui64> ssIds;
bool isServerless = false;
ui64 aggregatorId = 0;
- TPathId resourcesDomainKey;
+ TPathId domainKey, resourcesDomainKey;
for (const auto& entry : navigate->ResultSet) {
if (entry.Status != TNavigate::EStatus::Ok) {
continue;
@@ -144,6 +146,7 @@ private:
ssIds.insert(domainInfo->ExtractSchemeShard());
aggregatorId = domainInfo->Params.GetStatisticsAggregator();
isServerless = domainInfo->IsServerless();
+ domainKey = domainInfo->DomainKey;
resourcesDomainKey = domainInfo->ResourcesDomainKey;
}
if (ssIds.size() != 1) {
@@ -157,22 +160,31 @@ private:
return;
}
+ auto navigateDomainKey = [this] (TPathId domainKey) {
+ using TNavigate = NSchemeCache::TSchemeCacheNavigate;
+ auto navigate = std::make_unique<TNavigate>();
+ auto& entry = navigate->ResultSet.emplace_back();
+ entry.TableId = TTableId(domainKey.OwnerId, domainKey.LocalPathId);
+ entry.Operation = TNavigate::EOp::OpPath;
+ entry.RequestType = TNavigate::TEntry::ERequestType::ByTableId;
+ entry.RedirectRequired = false;
+ navigate->Cookie = ResolveSACookie;
+ Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
+ ResolveSAStage = RSA_IN_FLIGHT;
+ };
+
switch (ResolveSAStage) {
- case RSA_NOT_RUN:
+ case RSA_INITIAL:
if (!isServerless) {
- StatisticsAggregatorId = aggregatorId;
- ResolveSAStage = RSA_FINISHED;
+ if (aggregatorId) {
+ StatisticsAggregatorId = aggregatorId;
+ ResolveSAStage = RSA_FINISHED;
+ } else {
+ navigateDomainKey(domainKey);
+ return;
+ }
} else {
- using TNavigate = NSchemeCache::TSchemeCacheNavigate;
- auto navigate = std::make_unique<TNavigate>();
- auto& entry = navigate->ResultSet.emplace_back();
- entry.TableId = TTableId(resourcesDomainKey.OwnerId, resourcesDomainKey.LocalPathId);
- entry.Operation = TNavigate::EOp::OpPath;
- entry.RequestType = TNavigate::TEntry::ERequestType::ByTableId;
- entry.RedirectRequired = false;
- navigate->Cookie = ResolveSACookie;
- Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
- ResolveSAStage = RSA_IN_FLIGHT;
+ navigateDomainKey(resourcesDomainKey);
return;
}
break;
@@ -182,7 +194,7 @@ private:
break;
}
- if (IsSAUnavailable()) {
+ if (!StatisticsAggregatorId) {
ReplyFailed(requestId, true);
return;
}
@@ -303,6 +315,10 @@ private:
SyncNode();
}
+ void Handle(TEvStatistics::TEvStatisticsIsDisabled::TPtr&) {
+ ReplyAllFailed();
+ }
+
void ConnectToSA() {
if (SAPipeClientId || !StatisticsAggregatorId) {
return;
@@ -465,11 +481,11 @@ private:
static const ui64 ResolveSACookie = std::numeric_limits<ui64>::max();
enum EResolveSAStage {
- RSA_NOT_RUN,
+ RSA_INITIAL,
RSA_IN_FLIGHT,
RSA_FINISHED
};
- EResolveSAStage ResolveSAStage = RSA_NOT_RUN;
+ EResolveSAStage ResolveSAStage = RSA_INITIAL;
};
THolder<IActor> CreateStatService() {
diff --git a/ydb/core/tx/schemeshard/schemeshard_impl.cpp b/ydb/core/tx/schemeshard/schemeshard_impl.cpp
index fa2662f423..9060584c9d 100644
--- a/ydb/core/tx/schemeshard/schemeshard_impl.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard_impl.cpp
@@ -83,33 +83,7 @@ void TSchemeShard::ActivateAfterInitialization(const TActorContext& ctx, TActiva
}
if (IsDomainSchemeShard) {
- std::queue<TSVPMigrationInfo> migrations;
- for (auto& [pathId, subdomain] : SubDomains) {
- if (subdomain->GetTenantSchemeShardID() == InvalidTabletId) { // no tenant schemeshard
- continue;
- }
- if (subdomain->GetTenantSysViewProcessorID() != InvalidTabletId) { // tenant has SVP
- continue;
- }
-
- auto path = TPath::Init(pathId, this);
- if (path->IsRoot()) { // do not migrate main domain
- continue;
- }
-
- auto workingDir = path.Parent().PathString();
- auto dbName = path.LeafName();
- TSVPMigrationInfo migration{workingDir, dbName};
- migrations.push(std::move(migration));
-
- LOG_INFO_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - creating SVP"
- << ", working dir: " << workingDir
- << ", db name: " << dbName
- << ", at schemeshard: " << TabletID());
- }
-
- SVPMigrator = Register(CreateSVPMigrator(TabletID(), SelfId(), std::move(migrations)).Release());
+ InitializeTabletMigrations();
}
ResumeExports(opts.ExportIds, ctx);
@@ -135,6 +109,57 @@ void TSchemeShard::ActivateAfterInitialization(const TActorContext& ctx, TActiva
Become(&TThis::StateWork);
}
+void TSchemeShard::InitializeTabletMigrations() {
+ std::queue<TMigrationInfo> migrations;
+
+ for (auto& [pathId, subdomain] : SubDomains) {
+ auto path = TPath::Init(pathId, this);
+ if (path->IsRoot()) { // do not migrate main domain
+ continue;
+ }
+ if (subdomain->GetTenantSchemeShardID() == InvalidTabletId) { // no tenant schemeshard
+ continue;
+ }
+
+ bool createSVP = false;
+ bool createSA = false;
+
+ if (subdomain->GetTenantSysViewProcessorID() == InvalidTabletId) {
+ createSVP = true;
+ }
+
+ if (EnableStatistics &&
+ !IsServerlessDomain(subdomain) &&
+ subdomain->GetTenantStatisticsAggregatorID() == InvalidTabletId)
+ {
+ createSA = true;
+ }
+
+ if (!createSVP && !createSA) {
+ continue;
+ }
+
+ auto workingDir = path.Parent().PathString();
+ auto dbName = path.LeafName();
+ TMigrationInfo migration{workingDir, dbName, createSVP, createSA};
+ migrations.push(std::move(migration));
+
+ LOG_INFO_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
+ "TabletMigrator - creating tablets"
+ << ", working dir: " << workingDir
+ << ", db name: " << dbName
+ << ", create SVP: " << createSVP
+ << ", create SA: " << createSA
+ << ", at schemeshard: " << TabletID());
+ }
+
+ if (migrations.empty()) {
+ return;
+ }
+
+ TabletMigrator = Register(CreateTabletMigrator(TabletID(), SelfId(), std::move(migrations)).Release());
+}
+
ui64 TSchemeShard::Generation() const {
return Executor()->Generation();
}
@@ -4204,8 +4229,8 @@ void TSchemeShard::Die(const TActorContext &ctx) {
ctx.Send(TxAllocatorClient, new TEvents::TEvPoisonPill());
ctx.Send(SysPartitionStatsCollector, new TEvents::TEvPoisonPill());
- if (SVPMigrator) {
- ctx.Send(SVPMigrator, new TEvents::TEvPoisonPill());
+ if (TabletMigrator) {
+ ctx.Send(TabletMigrator, new TEvents::TEvPoisonPill());
}
if (CdcStreamScanFinalizer) {
@@ -6990,9 +7015,6 @@ void TSchemeShard::Handle(TEvPrivate::TEvSendBaseStatsToSA::TPtr&, const TActorC
}
void TSchemeShard::InitializeStatistics(const TActorContext& ctx) {
- if (!EnableStatistics) {
- return;
- }
ResolveSA();
ctx.Schedule(TDuration::Seconds(30), new TEvPrivate::TEvSendBaseStatsToSA());
}
@@ -7037,10 +7059,17 @@ void TSchemeShard::ConnectToSA() {
}
void TSchemeShard::SendBaseStatsToSA() {
- if (!EnableStatistics || !SAPipeClientId) {
+ if (!EnableStatistics) {
return;
}
+ if (!SAPipeClientId) {
+ ResolveSA();
+ if (!StatisticsAggregatorId) {
+ return;
+ }
+ }
+
int count = 0;
NKikimrStat::TSchemeShardStats record;
diff --git a/ydb/core/tx/schemeshard/schemeshard_impl.h b/ydb/core/tx/schemeshard/schemeshard_impl.h
index 05c3f4cb7a..295f45ce5e 100644
--- a/ydb/core/tx/schemeshard/schemeshard_impl.h
+++ b/ydb/core/tx/schemeshard/schemeshard_impl.h
@@ -283,7 +283,7 @@ public:
TActorId SysPartitionStatsCollector;
- TActorId SVPMigrator;
+ TActorId TabletMigrator;
TActorId CdcStreamScanFinalizer;
TDuration StatsMaxExecuteTime;
@@ -385,6 +385,8 @@ public:
bool IsSchemeShardConfigured() const;
+ void InitializeTabletMigrations();
+
ui64 Generation() const;
void SubscribeConsoleConfigs(const TActorContext& ctx);
diff --git a/ydb/core/tx/schemeshard/schemeshard_svp_migration.cpp b/ydb/core/tx/schemeshard/schemeshard_svp_migration.cpp
index f1575904f7..eb51ae879d 100644
--- a/ydb/core/tx/schemeshard/schemeshard_svp_migration.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard_svp_migration.cpp
@@ -5,13 +5,13 @@
namespace NKikimr::NSchemeShard {
-class TSVPMigrator : public TActorBootstrapped<TSVPMigrator> {
+class TTabletMigrator : public TActorBootstrapped<TTabletMigrator> {
public:
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
- return NKikimrServices::TActivity::SCHEMESHARD_SVP_MIGRATOR;
+ return NKikimrServices::TActivity::SCHEMESHARD_TABLET_MIGRATOR;
}
- TSVPMigrator(ui64 ssTabletId, TActorId ssActorId, std::queue<TSVPMigrationInfo>&& migrations)
+ TTabletMigrator(ui64 ssTabletId, TActorId ssActorId, std::queue<TMigrationInfo>&& migrations)
: SSTabletId(ssTabletId)
, SSActorId(ssActorId)
, Queue(std::move(migrations))
@@ -19,7 +19,7 @@ public:
void Bootstrap() {
Schedule(TDuration::Seconds(15), new TEvents::TEvWakeup);
- Become(&TSVPMigrator::StateWork);
+ Become(&TTabletMigrator::StateWork);
}
STFUNC(StateWork) {
@@ -32,14 +32,14 @@ public:
cFunc(TEvents::TEvPoison::EventType, PassAway);
default:
LOG_CRIT(*TlsActivationContext, NKikimrServices::FLAT_TX_SCHEMESHARD,
- "TSVPMigrator StateWork unexpected event 0x%08" PRIx32, ev->GetTypeRewrite());
+ "TTabletMigrator StateWork unexpected event 0x%08" PRIx32, ev->GetTypeRewrite());
}
}
private:
void RequestTxId() {
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - send TEvAllocateTxId"
+ "TabletMigrator - send TEvAllocateTxId"
<< ", working dir " << Current.WorkingDir
<< ", db name: " << Current.DbName
<< ", at schemeshard: " << SSTabletId);
@@ -59,10 +59,16 @@ private:
auto& modifySubDomain = *modifyScheme.MutableSubDomain();
modifySubDomain.SetName(Current.DbName);
- modifySubDomain.SetExternalSysViewProcessor(true);
+
+ if (Current.CreateSVP) {
+ modifySubDomain.SetExternalSysViewProcessor(true);
+ }
+ if (Current.CreateSA) {
+ modifySubDomain.SetExternalStatisticsAggregator(true);
+ }
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - send TEvModifySchemeTransaction"
+ "TabletMigrator - send TEvModifySchemeTransaction"
<< ", working dir " << Current.WorkingDir
<< ", db name: " << Current.DbName
<< ", at schemeshard: " << SSTabletId);
@@ -75,7 +81,7 @@ private:
request->Record.SetTxId(txId);
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - send TEvNotifyTxCompletion"
+ "TabletMigrator - send TEvNotifyTxCompletion"
<< ", txId " << txId
<< ", at schemeshard: " << SSTabletId);
@@ -95,7 +101,7 @@ private:
void Handle(TEvents::TEvWakeup::TPtr&) {
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - start processing migrations"
+ "TabletMigrator - start processing migrations"
<< ", queue size: " << Queue.size()
<< ", at schemeshard: " << SSTabletId);
@@ -106,7 +112,7 @@ private:
auto txId = ev->Get()->TxId;
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - handle TEvAllocateTxIdResult"
+ "TabletMigrator - handle TEvAllocateTxIdResult"
<< ", txId: " << txId
<< ", at schemeshard: " << SSTabletId);
@@ -119,7 +125,7 @@ private:
auto txId = record.GetTxId();
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - handle TEvModifySchemeTransactionResult"
+ "TabletMigrator - handle TEvModifySchemeTransactionResult"
<< ", status: " << status
<< ", txId: " << txId
<< ", at schemeshard: " << SSTabletId);
@@ -133,7 +139,7 @@ private:
break;
default:
LOG_ERROR_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - migration failed"
+ "TabletMigrator - migration failed"
<< ", status: " << status
<< ", reason: " << record.GetReason()
<< ", txId: " << txId
@@ -146,7 +152,7 @@ private:
void Handle(TEvSchemeShard::TEvNotifyTxCompletionResult::TPtr& ev) {
LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::FLAT_TX_SCHEMESHARD,
- "SVPMigrator - handle TEvNotifyTxCompletionResult"
+ "TabletMigrator - handle TEvNotifyTxCompletionResult"
<< ", txId: " << ev->Get()->Record.GetTxId()
<< ", at schemeshard: " << SSTabletId);
@@ -156,14 +162,14 @@ private:
private:
const ui64 SSTabletId;
const TActorId SSActorId;
- std::queue<TSVPMigrationInfo> Queue;
- TSVPMigrationInfo Current;
+ std::queue<TMigrationInfo> Queue;
+ TMigrationInfo Current;
};
-THolder<IActor> CreateSVPMigrator(ui64 ssTabletId, TActorId ssActorId,
- std::queue<TSVPMigrationInfo>&& migrations)
+THolder<IActor> CreateTabletMigrator(ui64 ssTabletId, TActorId ssActorId,
+ std::queue<TMigrationInfo>&& migrations)
{
- return MakeHolder<TSVPMigrator>(ssTabletId, ssActorId, std::move(migrations));
+ return MakeHolder<TTabletMigrator>(ssTabletId, ssActorId, std::move(migrations));
}
} // namespace NKikimr::NSchemeShard
diff --git a/ydb/core/tx/schemeshard/schemeshard_svp_migration.h b/ydb/core/tx/schemeshard/schemeshard_svp_migration.h
index 475c06811e..b9a75c180e 100644
--- a/ydb/core/tx/schemeshard/schemeshard_svp_migration.h
+++ b/ydb/core/tx/schemeshard/schemeshard_svp_migration.h
@@ -5,12 +5,14 @@
namespace NKikimr::NSchemeShard {
-struct TSVPMigrationInfo {
+struct TMigrationInfo {
TString WorkingDir;
TString DbName;
+ bool CreateSVP = false;
+ bool CreateSA = false;
};
-THolder<NActors::IActor> CreateSVPMigrator(ui64 ssTabletId, NActors::TActorId ssActorId,
- std::queue<TSVPMigrationInfo>&& migrations);
+THolder<NActors::IActor> CreateTabletMigrator(ui64 ssTabletId, NActors::TActorId ssActorId,
+ std::queue<TMigrationInfo>&& migrations);
} // namespace NKikimr::NSchemeShard
diff --git a/ydb/library/services/services.proto b/ydb/library/services/services.proto
index 49d3bcf55c..d9049aa6c8 100644
--- a/ydb/library/services/services.proto
+++ b/ydb/library/services/services.proto
@@ -987,7 +987,7 @@ message TActivity {
HTTP_MON_SERVICE_MON_REQUEST = 594;
HTTP_MON_SERVICE_NODE_PROXY = 595;
AUDIT_WRITER_ACTOR = 596;
- SCHEMESHARD_SVP_MIGRATOR = 597;
+ SCHEMESHARD_TABLET_MIGRATOR = 597;
SS_FETCHING_ACTOR = 598;
METADATA_SCHEME_DESCRIPTION_ACTOR = 599;
SCHEMESHARD_BACKGROUND_COMPACTION = 600;