Statistics service metrics (#8322)

Author: andrewstalin

ydb/core/protos/counters_statistics_aggregator.proto

@@ -25,3 +25,22 @@ enum ETxTypes {
     TXTYPE_ANALYZE_TABLE_DELIVERY_PROBLEM = 15 [(TxTypeOpts) = {Name: "TTxAnalyzeTableDeliveryProblem"}];
     TXTYPE_ANALYZE_DEADLINE = 16 [(TxTypeOpts) = {Name: "TTxAnalyzeDeadline"}];
 }
+
+enum ESimpleCounters {
+    COUNTER_FORCE_TRAVERSALS_INFLIGHT_SIZE = 0      [(CounterOpts) = {Name: "ForceTraversalsInflightSize"}];
+    COUNTER_AGGREGATION_TIME = 1                    [(CounterOpts) = {Name: "AggregationTime"}];
+    COUNTER_FORCE_TRAVERSAL_INFLIGHT_MAX_TIME = 2   [(CounterOpts) = {Name: "ForceTraversalsInflightMaxTime"}];
+}
+
+enum ECumulativeCounters {
+    COUNTER_CUMULATIVE_IGNORE = 0;
+}
+
+enum EPercentileCounters {
+    option (GlobalCounterOpts) = {
+        Ranges { Value: 0 Name: "0 ms" }
+        Ranges { Value: 1 Name: "1 ms" }
+    };
+
+    COUNTER_PERCENTILE_IGNORE = 0;
+}

ydb/core/statistics/aggregator/aggregator_impl.cpp

@@ -5,6 +5,8 @@
 
 #include <ydb/core/engine/minikql/flat_local_tx_factory.h>
 #include <ydb/core/protos/feature_flags.pb.h>
+#include <ydb/core/protos/counters_statistics_aggregator.pb.h>
+#include <ydb/core/tablet/tablet_counters_protobuf.h>
 #include <ydb/core/tx/datashard/datashard.h>
 #include <ydb/core/tx/scheme_cache/scheme_cache.h>
 
@@ -21,6 +23,14 @@ TStatisticsAggregator::TStatisticsAggregator(const NActors::TActorId& tablet, TT
 
     auto seed = std::random_device{}();
     RandomGenerator.seed(seed);
+
+    TabletCountersPtr.Reset(new TProtobufTabletCounters<
+        ESimpleCounters_descriptor,
+        ECumulativeCounters_descriptor,
+        EPercentileCounters_descriptor,
+        ETxTypes_descriptor
+    >());
+    TabletCounters = TabletCountersPtr.Get();
 }
 
 void TStatisticsAggregator::OnDetach(const TActorContext& ctx) {
@@ -34,6 +44,7 @@ void TStatisticsAggregator::OnTabletDead(TEvTablet::TEvTabletDead::TPtr&, const
 void TStatisticsAggregator::OnActivateExecutor(const TActorContext& ctx) {
     SA_LOG_I("[" << TabletID() << "] OnActivateExecutor");
 
+    Executor()->RegisterExternalTabletCounters(TabletCountersPtr);
     Execute(CreateTxInitSchema(), ctx);
 }
 
@@ -483,8 +494,15 @@ void TStatisticsAggregator::Handle(TEvPrivate::TEvRequestDistribution::TPtr&) {
 }
 
 void TStatisticsAggregator::Handle(TEvStatistics::TEvAggregateKeepAlive::TPtr& ev) {
+    const auto round = ev->Get()->Record.GetRound();
+    if (round == GlobalTraversalRound && AggregationRequestBeginTime) {
+        TInstant now = AppData(TlsActivationContext->AsActorContext())->TimeProvider->Now();
+        TDuration time = now - AggregationRequestBeginTime;
+        TabletCounters->Simple()[COUNTER_AGGREGATION_TIME].Set(time.MicroSeconds());
+    }
+
     auto ack = std::make_unique<TEvStatistics::TEvAggregateKeepAliveAck>();
-    ack->Record.SetRound(ev->Get()->Record.GetRound());
+    ack->Record.SetRound(round);
     Send(ev->Sender, ack.release());
     Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++KeepAliveSeqNo));
 }
@@ -779,6 +797,7 @@ void TStatisticsAggregator::DeleteForceTraversalOperation(const TString& operati
     }
 
     ForceTraversals.remove_if([operationId](const TForceTraversalOperation& elem) { return elem.OperationId == operationId;});
+    TabletCounters->Simple()[COUNTER_FORCE_TRAVERSALS_INFLIGHT_SIZE].Set(ForceTraversals.size());
 }
 
 TStatisticsAggregator::TForceTraversalTable* TStatisticsAggregator::ForceTraversalTable(const TString& operationId, const TPathId& pathId) {

ydb/core/statistics/aggregator/aggregator_impl.h

@@ -14,6 +14,7 @@
 #include <ydb/core/cms/console/configs_dispatcher.h>
 #include <ydb/core/cms/console/console.h>
 
+#include <ydb/core/tablet/tablet_counters.h>
 #include <ydb/core/tablet_flat/tablet_flat_executed.h>
 #include <ydb/core/tx/datashard/datashard.h>
 #include <ydb/core/tx/scheme_cache/scheme_cache.h>
@@ -228,6 +229,11 @@ class TStatisticsAggregator : public TActor<TStatisticsAggregator>, public NTabl
 
     std::mt19937_64 RandomGenerator;
 
+    TTabletCountersBase* TabletCounters;
+    TAutoPtr<TTabletCountersBase> TabletCountersPtr;
+
+    TInstant AggregationRequestBeginTime;
+
     bool EnableStatistics = false;
     bool EnableColumnStatistics = false;
 

ydb/core/statistics/aggregator/tx_aggr_stat_response.cpp

@@ -28,6 +28,9 @@ struct TStatisticsAggregator::TTxAggregateStatisticsResponse : public TTxBase {
 
         ++Self->KeepAliveSeqNo; // cancel timeout events
 
+        Self->TabletCounters->Simple()[COUNTER_AGGREGATION_TIME].Set(0);
+        Self->AggregationRequestBeginTime = TInstant::Zero();
+
         NIceDb::TNiceDb db(txc.DB);
 
         for (auto& column : Record.GetColumns()) {
@@ -122,6 +125,7 @@ struct TStatisticsAggregator::TTxAggregateStatisticsResponse : public TTxBase {
         case EAction::SendAggregate:
             ctx.Send(MakeStatServiceID(Self->SelfId().NodeId()), Request.release());
             ctx.Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++Self->KeepAliveSeqNo));
+            Self->AggregationRequestBeginTime = AppData(ctx)->TimeProvider->Now();
             break;
 
         default:

ydb/core/statistics/aggregator/tx_analyze.cpp

@@ -82,6 +82,7 @@ struct TStatisticsAggregator::TTxAnalyze : public TTxBase {
         }
 
         Self->ForceTraversals.emplace_back(operation);
+        Self->TabletCounters->Simple()[COUNTER_FORCE_TRAVERSALS_INFLIGHT_SIZE].Set(Self->ForceTraversals.size());
 
         db.Table<Schema::ForceTraversalOperations>().Key(operationId).Update(
             NIceDb::TUpdate<Schema::ForceTraversalOperations::OperationId>(operationId),

ydb/core/statistics/aggregator/tx_init.cpp

@@ -210,6 +210,8 @@ struct TStatisticsAggregator::TTxInit : public TTxBase {
                 }
             }
 
+            Self->TabletCounters->Simple()[COUNTER_FORCE_TRAVERSALS_INFLIGHT_SIZE].Set(Self->ForceTraversals.size());
+
             SA_LOG_D("[" << Self->TabletID() << "] Loaded ForceTraversalOperations: "
                 << "table count# " << Self->ForceTraversals.size());
         }

ydb/core/statistics/aggregator/tx_response_tablet_distribution.cpp

@@ -101,6 +101,7 @@ struct TStatisticsAggregator::TTxResponseTabletDistribution : public TTxBase {
         case EAction::SendAggregate:
             ctx.Send(MakeStatServiceID(Self->SelfId().NodeId()), AggregateStatisticsRequest.release());
             ctx.Schedule(KeepAliveTimeout, new TEvPrivate::TEvAckTimeout(++Self->KeepAliveSeqNo));
+            Self->AggregationRequestBeginTime = AppData(ctx)->TimeProvider->Now();
             break;
 
         default:

ydb/core/statistics/aggregator/tx_schedule_traversal.cpp

@@ -11,12 +11,18 @@ struct TStatisticsAggregator::TTxScheduleTrasersal : public TTxBase {
 
     TTxType GetTxType() const override { return TXTYPE_SCHEDULE_TRAVERSAL; }
 
-    bool Execute(TTransactionContext& txc, const TActorContext&) override {
+    bool Execute(TTransactionContext& txc, const TActorContext& ctx) override {
 
         if (!Self->EnableColumnStatistics) {
             return true;
         }
 
+        TDuration time = TDuration ::Zero();
+        if (!Self->ForceTraversals.empty()) {
+            time = ctx.Now() - Self->ForceTraversals.front().CreatedAt;
+        }
+        Self->TabletCounters->Simple()[COUNTER_FORCE_TRAVERSAL_INFLIGHT_MAX_TIME].Set(time.MicroSeconds());
+
         if (Self->TraversalPathId) {
             SA_LOG_T("[" << Self->TabletID() << "] TTxScheduleTrasersal::Execute. Traverse is in progress. PathId " << Self->TraversalPathId);
             return true;