use statistics aggregator for basic statistics KIKIMR-18323 (#623)

ydb/core/protos/counters_statistics_aggregator.proto

@@ -7,7 +7,8 @@ option java_package = "ru.yandex.kikimr.proto";
 option (TabletTypeName) = "StatisticsAggregator";
 
 enum ETxTypes {
-    TXTYPE_INIT_SCHEMA = 0      [(TxTypeOpts) = {Name: "TxInitSchema"}];
-    TXTYPE_INIT = 1             [(TxTypeOpts) = {Name: "TxInit"}];
-    TXTYPE_CONFIGURE = 2        [(TxTypeOpts) = {Name: "TxConfigure"}];
+    TXTYPE_INIT_SCHEMA = 0       [(TxTypeOpts) = {Name: "TxInitSchema"}];
+    TXTYPE_INIT = 1              [(TxTypeOpts) = {Name: "TxInit"}];
+    TXTYPE_CONFIGURE = 2         [(TxTypeOpts) = {Name: "TxConfigure"}];
+    TXTYPE_SCHEMESHARD_STATS = 3 [(TxTypeOpts) = {Name: "TxSchemeShardStats"}];
 }

ydb/core/protos/statistics.proto

@@ -4,21 +4,55 @@ package NKikimrStat;
 
 option java_package = "ru.yandex.kikimr.proto";
 
-message TEvBroadcastStatistics {
-    message TEntry {
-        optional NKikimrProto.TPathID PathId = 1;
-        optional uint64 RowCount = 2;
-        optional uint64 BytesSize = 3;
+message TEvConfigureAggregator {
+    optional string Database = 1;
+}
+
+message TPathEntry {
+    optional NKikimrProto.TPathID PathId = 1;
+    optional uint64 RowCount = 2;
+    optional uint64 BytesSize = 3;
+}
+
+message TSchemeShardStats {
+    repeated TPathEntry Entries = 1;
+}
+
+// SS -> SA
+message TEvConnectSchemeShard {
+    optional fixed64 SchemeShardId = 1;
+}
+
+// SS -> SA
+message TEvSchemeShardStats {
+    optional fixed64 SchemeShardId = 1;
+    optional bytes Stats = 2; // serialized TSchemeShardStats
+}
+
+// nodes -> SA
+message TEvConnectNode {
+    optional uint32 NodeId = 1;
+    repeated fixed64 NeedSchemeShards = 2;
+    message THaveEntry {
+        optional fixed64 SchemeShardId = 1;
+        optional uint64 Timestamp = 2;
     }
-    repeated uint32 NodeIds = 1;
-    repeated TEntry Entries = 2;
+    repeated THaveEntry HaveSchemeShards = 3;
 }
 
-message TEvRegisterNode {
+// nodes -> SA
+message TEvRequestStats {
     optional uint32 NodeId = 1;
-    optional bool HasStatistics = 2;
+    repeated fixed64 NeedSchemeShards = 2;
 }
 
-message TEvConfigureAggregator {
-    optional string Database = 1;
+// SA -> nodes
+message TEvPropagateStatistics {
+    repeated uint32 NodeIds = 1; // hierarchical propagation
+    message TStatsEntry {
+        optional fixed64 SchemeShardId = 1;
+        optional bytes Stats = 2; // serialized TSchemeShardStats
+        optional uint64 Timestamp = 3;
+    }
+    repeated TStatsEntry Entries = 2;
 }

ydb/core/statistics/aggregator/aggregator.cpp

@@ -5,7 +5,11 @@
 namespace NKikimr::NStat {
 
 IActor* CreateStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info) {
-    return new TStatisticsAggregator(tablet, info);
+    return new TStatisticsAggregator(tablet, info, false);
+}
+
+IActor* CreateStatisticsAggregatorForTests(const NActors::TActorId& tablet, TTabletStorageInfo* info) {
+    return new TStatisticsAggregator(tablet, info, true);
 }
 
 } // NKikimr::NStat

ydb/core/statistics/aggregator/aggregator.h

@@ -7,4 +7,6 @@ namespace NKikimr::NStat {
 
 IActor* CreateStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info);
 
+IActor* CreateStatisticsAggregatorForTests(const NActors::TActorId& tablet, TTabletStorageInfo* info);
+
 } // NKikimr::NStat

ydb/core/statistics/aggregator/aggregator_impl.cpp

@@ -1,15 +1,21 @@
 #include "aggregator_impl.h"
 
 #include <ydb/core/engine/minikql/flat_local_tx_factory.h>
+#include <ydb/core/statistics/stat_service.h>
 
 #include <library/cpp/monlib/service/pages/templates.h>
 
 namespace NKikimr::NStat {
 
-TStatisticsAggregator::TStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info)
+TStatisticsAggregator::TStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info, bool forTests)
     : TActor(&TThis::StateInit)
     , TTabletExecutedFlat(info, tablet, new NMiniKQL::TMiniKQLFactory)
-{}
+{
+    PropagateInterval = forTests ? TDuration::Seconds(5) : TDuration::Minutes(3);
+
+    auto seed = std::random_device{}();
+    RandomGenerator.seed(seed);
+}
 
 void TStatisticsAggregator::OnDetach(const TActorContext& ctx) {
     Die(ctx);
@@ -29,8 +35,241 @@ void TStatisticsAggregator::DefaultSignalTabletActive(const TActorContext& ctx)
     Y_UNUSED(ctx);
 }
 
-void TStatisticsAggregator::Handle(TEvPrivate::TEvProcess::TPtr&) {
-    SA_LOG_D("[" << TabletID() << "] Handle TEvPrivate::TEvProcess");
+void TStatisticsAggregator::Handle(TEvTabletPipe::TEvServerConnected::TPtr &ev) {
+    auto pipeServerId = ev->Get()->ServerId;
+
+    SA_LOG_D("[" << TabletID() << "] EvServerConnected"
+        << ", pipe server id = " << pipeServerId);
+}
+
+void TStatisticsAggregator::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr &ev) {
+    auto pipeServerId = ev->Get()->ServerId;
+
+    SA_LOG_D("[" << TabletID() << "] EvServerDisconnected"
+        << ", pipe server id = " << pipeServerId);
+
+    auto itNodeServer = NodePipes.find(pipeServerId);
+    if (itNodeServer != NodePipes.end()) {
+        auto nodeId = itNodeServer->second;
+        auto itNode = Nodes.find(nodeId);
+        if (itNode != Nodes.end()) {
+            --itNode->second;
+            if (itNode->second == 0) {
+                Nodes.erase(itNode);
+            }
+        }
+        NodePipes.erase(itNodeServer);
+        return;
+    }
+
+    auto itShardServer = SchemeShardPipes.find(pipeServerId);
+    if (itShardServer != SchemeShardPipes.end()) {
+        auto ssId = itShardServer->second;
+        auto itShard = SchemeShards.find(ssId);
+        if (itShard != SchemeShards.end()) {
+            --itShard->second;
+            if (itShard->second == 0) {
+                SchemeShards.erase(itShard);
+            }
+        }
+        SchemeShardPipes.erase(itShardServer);
+        return;
+    }
+}
+
+void TStatisticsAggregator::Handle(TEvStatistics::TEvConnectNode::TPtr& ev) {
+    const auto& record = ev->Get()->Record;
+    const TNodeId nodeId = record.GetNodeId();
+    auto pipeServerId = ev->Recipient;
+
+    SA_LOG_D("[" << TabletID() << "] EvConnectNode"
+        << ", pipe server id = " << pipeServerId
+        << ", node id = " << nodeId
+        << ", have schemeshards count = " << record.HaveSchemeShardsSize()
+        << ", need schemeshards count = " << record.NeedSchemeShardsSize());
+
+    if (NodePipes.find(pipeServerId) == NodePipes.end()) {
+        NodePipes[pipeServerId] = nodeId;
+        ++Nodes[nodeId];
+    }
+
+    for (const auto& ssEntry : record.GetHaveSchemeShards()) {
+        RequestedSchemeShards.insert(ssEntry.GetSchemeShardId());
+    }
+
+    if (!IsPropagateInFlight) {
+        Schedule(PropagateInterval, new TEvPrivate::TEvPropagate());
+        IsPropagateInFlight = true;
+    }
+
+    std::vector<TSSId> ssIds;
+    ssIds.reserve(record.NeedSchemeShardsSize());
+    for (const auto& ssId : record.GetNeedSchemeShards()) {
+        ssIds.push_back(ssId);
+        RequestedSchemeShards.insert(ssId);
+    }
+
+    ProcessRequests(nodeId, ssIds);
+}
+
+void TStatisticsAggregator::Handle(TEvStatistics::TEvRequestStats::TPtr& ev) {
+    const auto& record = ev->Get()->Record;
+    const auto nodeId = record.GetNodeId();
+
+    SA_LOG_D("[" << TabletID() << "] EvRequestStats"
+        << ", node id = " << nodeId
+        << ", schemeshard count = " << record.NeedSchemeShardsSize());
+
+    std::vector<TSSId> ssIds;
+    ssIds.reserve(record.NeedSchemeShardsSize());
+    for (const auto& ssId : record.GetNeedSchemeShards()) {
+        ssIds.push_back(ssId);
+    }
+
+    ProcessRequests(nodeId, ssIds);
+}
+
+void TStatisticsAggregator::Handle(TEvStatistics::TEvConnectSchemeShard::TPtr& ev) {
+    const auto& record = ev->Get()->Record;
+    const TSSId schemeShardId = record.GetSchemeShardId();
+    auto pipeServerId = ev->Recipient;
+
+    if (SchemeShardPipes.find(pipeServerId) == SchemeShardPipes.end()) {
+        SchemeShardPipes[pipeServerId] = schemeShardId;
+        ++SchemeShards[schemeShardId];
+    }
+
+    SA_LOG_D("[" << TabletID() << "] EvConnectSchemeShard"
+        << ", pipe server id = " << pipeServerId
+        << ", schemeshard id = " << schemeShardId);
+}
+
+void TStatisticsAggregator::Handle(TEvPrivate::TEvFastPropagateCheck::TPtr&) {
+    SA_LOG_D("[" << TabletID() << "] EvFastPropagateCheck");
+
+    PropagateFastStatistics();
+
+    FastCheckInFlight = false;
+    FastCounter = StatsOptimizeFirstNodesCount;
+    FastNodes.clear();
+    FastSchemeShards.clear();
+}
+
+void TStatisticsAggregator::Handle(TEvPrivate::TEvPropagate::TPtr&) {
+    SA_LOG_D("[" << TabletID() << "] EvPropagate");
+
+    PropagateStatistics();
+
+    Schedule(PropagateInterval, new TEvPrivate::TEvPropagate());
+}
+
+void TStatisticsAggregator::ProcessRequests(TNodeId nodeId, const std::vector<TSSId>& ssIds) {
+    if (FastCounter > 0) {
+        --FastCounter;
+        SendStatisticsToNode(nodeId, ssIds);
+    } else {
+        FastNodes.insert(nodeId);
+        for (const auto& ssId : ssIds) {
+            FastSchemeShards.insert(ssId);
+        }
+        if (!FastCheckInFlight) {
+            Schedule(TDuration::MilliSeconds(100), new TEvPrivate::TEvFastPropagateCheck());
+            FastCheckInFlight = true;
+        }
+    }
+}
+
+void TStatisticsAggregator::SendStatisticsToNode(TNodeId nodeId, const std::vector<TSSId>& ssIds) {
+    SA_LOG_D("[" << TabletID() << "] SendStatisticsToNode()"
+        << ", node id = " << nodeId
+        << ", schemeshard count = " << ssIds.size());
+
+    std::vector<TNodeId> nodeIds;
+    nodeIds.push_back(nodeId);
+
+    PropagateStatisticsImpl(nodeIds, ssIds);
+}
+
+void TStatisticsAggregator::PropagateStatistics() {
+    SA_LOG_D("[" << TabletID() << "] PropagateStatistics()"
+        << ", node count = " << Nodes.size()
+        << ", schemeshard count = " << RequestedSchemeShards.size());
+
+    if (Nodes.empty() || RequestedSchemeShards.empty()) {
+        return;
+    }
+
+    std::vector<TNodeId> nodeIds;
+    nodeIds.reserve(Nodes.size());
+    for (const auto& [nodeId, _] : Nodes) {
+        nodeIds.push_back(nodeId);
+    }
+    std::shuffle(std::begin(nodeIds), std::end(nodeIds), RandomGenerator);
+
+    std::vector<TSSId> ssIds;
+    ssIds.reserve(RequestedSchemeShards.size());
+    for (const auto& ssId : RequestedSchemeShards) {
+        ssIds.push_back(ssId);
+    }
+
+    PropagateStatisticsImpl(nodeIds, ssIds);
+}
+
+void TStatisticsAggregator::PropagateFastStatistics() {
+    SA_LOG_D("[" << TabletID() << "] PropagateFastStatistics()"
+        << ", node count = " << FastNodes.size()
+        << ", schemeshard count = " << FastSchemeShards.size());
+
+    if (FastNodes.empty() || FastSchemeShards.empty()) {
+        return;
+    }
+
+    std::vector<TNodeId> nodeIds;
+    nodeIds.reserve(FastNodes.size());
+    for (const auto& nodeId : FastNodes) {
+        nodeIds.push_back(nodeId);
+    }
+    std::shuffle(std::begin(nodeIds), std::end(nodeIds), RandomGenerator);
+
+    std::vector<TSSId> ssIds;
+    ssIds.reserve(FastSchemeShards.size());
+    for (const auto& ssId : FastSchemeShards) {
+        ssIds.push_back(ssId);
+    }
+
+    PropagateStatisticsImpl(nodeIds, ssIds);
+}
+
+void TStatisticsAggregator::PropagateStatisticsImpl(
+    const std::vector<TNodeId>& nodeIds, const std::vector<TSSId>& ssIds)
+{
+    TNodeId leadingNodeId = nodeIds[0];
+
+    for (size_t index = 0; index < ssIds.size(); ) {
+        auto propagate = std::make_unique<TEvStatistics::TEvPropagateStatistics>();
+        auto* record = propagate->MutableRecord();
+        record->MutableNodeIds()->Reserve(nodeIds.size() - 1);
+        for (size_t i = 1; i < nodeIds.size(); ++i) {
+            record->AddNodeIds(nodeIds[i]);
+        }
+        for (size_t size = 0; index < ssIds.size(); ++index) {
+            auto ssId = ssIds[index];
+            auto* entry = record->AddEntries();
+            entry->SetSchemeShardId(ssId);
+            auto itStats = BaseStats.find(ssId);
+            if (itStats != BaseStats.end()) {
+                entry->SetStats(itStats->second);
+                size += itStats->second.size();
+            } else {
+                entry->SetStats(TString()); // stats are not sent from SA yet
+            }
+            if (size >= StatsSizeLimitBytes) {
+                ++index;
+                break;
+            }
+        }
+        Send(NStat::MakeStatServiceID(leadingNodeId), propagate.release());
+    }
 }
 
 void TStatisticsAggregator::PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value) {

ydb/core/statistics/aggregator/aggregator_impl.h

@@ -11,6 +11,8 @@
 
 #include <ydb/core/tablet_flat/tablet_flat_executed.h>
 
+#include <random>
+
 namespace NKikimr::NStat {
 
 class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabletFlatExecutor::TTabletExecutedFlat {
@@ -19,24 +21,30 @@ public:
         return NKikimrServices::TActivity::STATISTICS_AGGREGATOR;
     }
 
-    TStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info);
+    TStatisticsAggregator(const NActors::TActorId& tablet, TTabletStorageInfo* info, bool forTests);
 
 private:
+    using TSSId = ui64;
+    using TNodeId = ui32;
+
     using Schema = TAggregatorSchema;
     using TTxBase = NTabletFlatExecutor::TTransactionBase<TStatisticsAggregator>;
 
     struct TTxInitSchema;
     struct TTxInit;
     struct TTxConfigure;
+    struct TTxSchemeShardStats;
 
     struct TEvPrivate {
         enum EEv {
-            EvProcess = EventSpaceBegin(TEvents::ES_PRIVATE),
+            EvPropagate = EventSpaceBegin(TEvents::ES_PRIVATE),
+            EvFastPropagateCheck,
 
             EvEnd
         };
 
-        struct TEvProcess : public TEventLocal<TEvProcess, EvProcess> {};
+        struct TEvPropagate : public TEventLocal<TEvPropagate, EvPropagate> {};
+        struct TEvFastPropagateCheck : public TEventLocal<TEvFastPropagateCheck, EvFastPropagateCheck> {};
     };
 
 private:
@@ -50,26 +58,38 @@ private:
     NTabletFlatExecutor::ITransaction* CreateTxInit();
 
     void Handle(TEvStatistics::TEvConfigureAggregator::TPtr& ev);
-    void Handle(TEvPrivate::TEvProcess::TPtr& ev);
+    void Handle(TEvStatistics::TEvSchemeShardStats::TPtr& ev);
+    void Handle(TEvPrivate::TEvPropagate::TPtr& ev);
+    void Handle(TEvStatistics::TEvConnectNode::TPtr& ev);
+    void Handle(TEvStatistics::TEvRequestStats::TPtr& ev);
+    void Handle(TEvStatistics::TEvConnectSchemeShard::TPtr& ev);
+    void Handle(TEvTabletPipe::TEvServerConnected::TPtr& ev);
+    void Handle(TEvTabletPipe::TEvServerDisconnected::TPtr& ev);
+    void Handle(TEvPrivate::TEvFastPropagateCheck::TPtr& ev);
+
+    void ProcessRequests(TNodeId nodeId, const std::vector<TSSId>& ssIds);
+    void SendStatisticsToNode(TNodeId nodeId, const std::vector<TSSId>& ssIds);
+    void PropagateStatistics();
+    void PropagateFastStatistics();
+    void PropagateStatisticsImpl(const std::vector<TNodeId>& nodeIds, const std::vector<TSSId>& ssIds);
 
     void PersistSysParam(NIceDb::TNiceDb& db, ui64 id, const TString& value);
 
     STFUNC(StateInit) {
-        switch(ev->GetTypeRewrite()) {
-            hFunc(TEvStatistics::TEvConfigureAggregator, Handle);
-            IgnoreFunc(TEvPrivate::TEvProcess);
-            default:
-                if (!HandleDefaultEvents(ev, SelfId())) {
-                    LOG_CRIT(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
-                        "TStatisticsAggregator StateInit unexpected event 0x%08" PRIx32, ev->GetTypeRewrite());
-                }
-        }
+        StateInitImpl(ev,SelfId());
     }
 
     STFUNC(StateWork) {
         switch(ev->GetTypeRewrite()) {
             hFunc(TEvStatistics::TEvConfigureAggregator, Handle);
-            hFunc(TEvPrivate::TEvProcess, Handle);
+            hFunc(TEvStatistics::TEvSchemeShardStats, Handle);
+            hFunc(TEvPrivate::TEvPropagate, Handle);
+            hFunc(TEvStatistics::TEvConnectNode, Handle);
+            hFunc(TEvStatistics::TEvRequestStats, Handle);
+            hFunc(TEvStatistics::TEvConnectSchemeShard, Handle);
+            hFunc(TEvTabletPipe::TEvServerConnected, Handle);
+            hFunc(TEvTabletPipe::TEvServerDisconnected, Handle);
+            hFunc(TEvPrivate::TEvFastPropagateCheck, Handle);
             default:
                 if (!HandleDefaultEvents(ev, SelfId())) {
                     LOG_CRIT(TlsActivationContext->AsActorContext(), NKikimrServices::STATISTICS,
@@ -80,6 +100,29 @@ private:
 
 private:
     TString Database;
+
+    std::mt19937_64 RandomGenerator;
+
+    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);
+    bool IsPropagateInFlight = false; // is slow propagation started
+
+    std::unordered_map<TSSId, TString> BaseStats; // schemeshard id -> serialized stats for all paths
+
+    std::unordered_map<TSSId, size_t> SchemeShards; // all connected schemeshards
+    std::unordered_map<TActorId, TSSId> SchemeShardPipes; // schemeshard pipe servers
+
+    std::unordered_map<TNodeId, size_t> Nodes; // all connected nodes
+    std::unordered_map<TActorId, TNodeId> NodePipes; // node pipe servers
+
+    std::unordered_set<TSSId> RequestedSchemeShards; // all schemeshards that were requested from all nodes
+
+    size_t FastCounter = StatsOptimizeFirstNodesCount;
+    bool FastCheckInFlight = false;
+    std::unordered_set<TNodeId> FastNodes; // nodes for fast propagation
+    std::unordered_set<TSSId> FastSchemeShards; // schemeshards for fast propagation
 };
 
 } // NKikimr::NStat

ydb/core/statistics/aggregator/schema.h

@@ -13,8 +13,17 @@ struct TAggregatorSchema : NIceDb::Schema {
         using TColumns = TableColumns<Id, Value>;
     };
 
+    struct BaseStats : Table<2> {
+        struct SchemeShardId : Column<1, NScheme::NTypeIds::Uint64> {};
+        struct Stats : Column<2, NScheme::NTypeIds::String> {};
+
+        using TKey = TableKey<SchemeShardId>;
+        using TColumns = TableColumns<SchemeShardId, Stats>;
+    };
+
     using TTables = SchemaTables<
-        SysParams
+        SysParams,
+        BaseStats
     >;
 
     using TSettings = SchemaSettings<

ydb/core/statistics/aggregator/tx_init.cpp

@@ -16,8 +16,11 @@ struct TStatisticsAggregator::TTxInit : public TTxBase {
 
         { // precharge
             auto sysParamsRowset = db.Table<Schema::SysParams>().Range().Select();
+            auto baseStatsRowset = db.Table<Schema::BaseStats>().Range().Select();
 
-            if (!sysParamsRowset.IsReady()) {
+            if (!sysParamsRowset.IsReady() ||
+                !baseStatsRowset.IsReady())
+            {
                 return false;
             }
         }
@@ -48,6 +51,30 @@ struct TStatisticsAggregator::TTxInit : public TTxBase {
             }
         }
 
+        // BaseStats
+        {
+            Self->BaseStats.clear();
+
+            auto rowset = db.Table<Schema::BaseStats>().Range().Select();
+            if (!rowset.IsReady()) {
+                return false;
+            }
+
+            while (!rowset.EndOfSet()) {
+                ui64 schemeShardId = rowset.GetValue<Schema::BaseStats::SchemeShardId>();
+                TString stats = rowset.GetValue<Schema::BaseStats::Stats>();
+
+                Self->BaseStats[schemeShardId] = stats;
+
+                if (!rowset.Next()) {
+                    return false;
+                }
+            }
+
+            SA_LOG_D("[" << Self->TabletID() << "] Loading base stats: "
+                << "schemeshard count# " << Self->BaseStats.size());
+        }
+
         return true;
     }
 

ydb/core/statistics/aggregator/tx_schemeshard_stats.cpp

@@ -0,0 +1,43 @@
+#include "aggregator_impl.h"
+
+namespace NKikimr::NStat {
+
+struct TStatisticsAggregator::TTxSchemeShardStats : public TTxBase {
+    NKikimrStat::TEvSchemeShardStats Record;
+
+    TTxSchemeShardStats(TSelf* self, NKikimrStat::TEvSchemeShardStats&& record)
+        : TTxBase(self)
+        , Record(std::move(record))
+    {}
+
+    TTxType GetTxType() const override { return TXTYPE_SCHEMESHARD_STATS; }
+
+    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+        ui64 schemeShardId = Record.GetSchemeShardId();
+        const auto& stats = Record.GetStats();
+
+        SA_LOG_D("[" << Self->TabletID() << "] TTxSchemeShardStats::Execute: "
+            << "schemeshard id# " << schemeShardId
+            << ", stats size# " << stats.size());
+
+        NIceDb::TNiceDb db(txc.DB);
+        db.Table<Schema::BaseStats>().Key(schemeShardId).Update(
+            NIceDb::TUpdate<Schema::BaseStats::Stats>(stats));
+
+        Self->BaseStats[schemeShardId] = stats;
+
+        return true;
+    }
+
+    void Complete(const TActorContext&) override {
+        SA_LOG_D("[" << Self->TabletID() << "] TTxSchemeShardStats::Complete");
+    }
+};
+
+void TStatisticsAggregator::Handle(TEvStatistics::TEvSchemeShardStats::TPtr& ev) {
+    auto& record = ev->Get()->Record;
+    Execute(new TTxSchemeShardStats(this, std::move(record)),
+        TActivationContext::AsActorContext());
+}
+
+} // NKikimr::NStat

ydb/core/statistics/aggregator/ya.make

@@ -1,5 +1,10 @@
 LIBRARY()
 
+OWNER(
+    monster
+    g:kikimr
+)
+
 SRCS(
     aggregator.h
     aggregator.cpp
@@ -10,6 +15,7 @@ SRCS(
     tx_configure.cpp
     tx_init.cpp
     tx_init_schema.cpp
+    tx_schemeshard_stats.cpp
 )
 
 PEERDIR(