24-3: Replication counters (#11081)

Author: CyberROFL

ydb/core/protos/counters_replication.proto

@@ -7,10 +7,17 @@ option (TabletTypeName) = "ReplicationController"; // Used as prefix for all cou
 
 enum ESimpleCounters {
     COUNTER_SIMPLE_IGNORE = 0;
+    COUNTER_SESSIONS = 1 [(CounterOpts) = {Name: "Sessions"}];
+    COUNTER_WORKERS = 2 [(CounterOpts) = {Name: "Workers"}];
+    COUNTER_BOOT_QUEUE = 3 [(CounterOpts) = {Name: "BootQueue"}];
+    COUNTER_STOP_QUEUE = 4 [(CounterOpts) = {Name: "StopQueue"}];
+    COUNTER_DATA_LAG = 5 [(CounterOpts) = {Name: "DataLag"}];
 }
 
 enum ECumulativeCounters {
     COUNTER_CUMULATIVE_IGNORE = 0;
+    COUNTER_CREATE_SESSION = 1 [(CounterOpts) = {Name: "CreateSession"}];
+    COUNTER_DELETE_SESSION = 2 [(CounterOpts) = {Name: "DeleteSession"}];
 }
 
 enum EPercentileCounters {

ydb/core/tx/replication/controller/controller.cpp

@@ -4,6 +4,7 @@
 #include <ydb/core/base/appdata.h>
 #include <ydb/core/discovery/discovery.h>
 #include <ydb/core/engine/minikql/flat_local_tx_factory.h>
+#include <ydb/core/tablet/tablet_counters_protobuf.h>
 
 namespace NKikimr::NReplication {
 
@@ -13,6 +14,13 @@ TController::TController(const TActorId& tablet, TTabletStorageInfo* info)
     : TActor(&TThis::StateInit)
     , TTabletExecutedFlat(info, tablet, new NMiniKQL::TMiniKQLFactory)
     , LogPrefix(this)
+    , TabletCountersPtr(new TProtobufTabletCounters<
+             ESimpleCounters_descriptor,
+             ECumulativeCounters_descriptor,
+             EPercentileCounters_descriptor,
+             ETxTypes_descriptor
+        >())
+    , TabletCounters(TabletCountersPtr.Get())
 {
 }
 
@@ -30,6 +38,7 @@ void TController::OnTabletDead(TEvTablet::TEvTabletDead::TPtr&, const TActorCont
 
 void TController::OnActivateExecutor(const TActorContext& ctx) {
     CLOG_T(ctx, "OnActivateExecutor");
+    Executor()->RegisterExternalTabletCounters(TabletCountersPtr.Release());
     RunTxInitSchema(ctx);
 }
 
@@ -292,9 +301,11 @@ void TController::Handle(TEvDiscovery::TEvError::TPtr& ev, const TActorContext&
 void TController::CreateSession(ui32 nodeId, const TActorContext& ctx) {
     CLOG_D(ctx, "Create session"
         << ": nodeId# " << nodeId);
+    TabletCounters->Cumulative()[COUNTER_CREATE_SESSION] += 1;
 
     Y_ABORT_UNLESS(!Sessions.contains(nodeId));
     Sessions.emplace(nodeId, TSessionInfo());
+    TabletCounters->Simple()[COUNTER_SESSIONS] = Sessions.size();
 
     auto ev = MakeHolder<TEvService::TEvHandshake>(TabletID(), Executor()->Generation());
     ui32 flags = 0;
@@ -308,6 +319,7 @@ void TController::CreateSession(ui32 nodeId, const TActorContext& ctx) {
 void TController::DeleteSession(ui32 nodeId, const TActorContext& ctx) {
     CLOG_D(ctx, "Delete session"
         << ": nodeId# " << nodeId);
+    TabletCounters->Cumulative()[COUNTER_DELETE_SESSION] += 1;
 
     Y_ABORT_UNLESS(Sessions.contains(nodeId));
     auto& session = Sessions[nodeId];
@@ -327,6 +339,8 @@ void TController::DeleteSession(ui32 nodeId, const TActorContext& ctx) {
     }
 
     Sessions.erase(nodeId);
+    TabletCounters->Simple()[COUNTER_SESSIONS] = Sessions.size();
+
     CloseSession(nodeId, ctx);
     ScheduleProcessQueues();
 }
@@ -431,6 +445,9 @@ void TController::UpdateLag(const TWorkerId& id, TDuration lag) {
     }
 
     target->UpdateLag(id.WorkerId(), lag);
+    if (const auto lag = replication->GetLag()) {
+        TabletCounters->Simple()[COUNTER_DATA_LAG] = lag->MilliSeconds();
+    }
 }
 
 void TController::Handle(TEvService::TEvRunWorker::TPtr& ev, const TActorContext& ctx) {
@@ -486,6 +503,7 @@ TWorkerInfo* TController::GetOrCreateWorker(const TWorkerId& id, NKikimrReplicat
     auto it = Workers.find(id);
     if (it == Workers.end()) {
         it = Workers.emplace(id, cmd).first;
+        TabletCounters->Simple()[COUNTER_WORKERS] = Workers.size();
     }
 
     auto replication = Find(id.ReplicationId());
@@ -499,6 +517,9 @@ TWorkerInfo* TController::GetOrCreateWorker(const TWorkerId& id, NKikimrReplicat
 }
 
 void TController::ScheduleProcessQueues() {
+    TabletCounters->Simple()[COUNTER_BOOT_QUEUE] = BootQueue.size();
+    TabletCounters->Simple()[COUNTER_STOP_QUEUE] = StopQueue.size();
+
     if (ProcessQueuesScheduled || (!BootQueue && !StopQueue)) {
         return;
     }
@@ -652,6 +673,7 @@ void TController::RemoveWorker(const TWorkerId& id, const TActorContext& ctx) {
 
     RemoveQueue.erase(id);
     Workers.erase(id);
+    TabletCounters->Simple()[COUNTER_WORKERS] = Workers.size();
 
     auto replication = Find(id.ReplicationId());
     if (!replication) {

ydb/core/tx/replication/controller/controller_impl.h

@@ -12,6 +12,7 @@
 #include <ydb/core/base/blobstorage.h>
 #include <ydb/core/base/defs.h>
 #include <ydb/core/protos/counters_replication.pb.h>
+#include <ydb/core/tablet/tablet_counters.h>
 #include <ydb/core/tablet_flat/tablet_flat_executed.h>
 #include <ydb/core/tx/replication/service/service.h>
 #include <ydb/library/actors/core/interconnect.h>
@@ -167,6 +168,8 @@ class TController
 
 private:
     const TTabletLogPrefix LogPrefix;
+    THolder<TTabletCountersBase> TabletCountersPtr;
+    TTabletCountersBase* TabletCounters;
 
     TSysParams SysParams;
     THashMap<ui64, TReplication::TPtr> Replications;

ydb/core/tx/replication/controller/ya.make

@@ -5,6 +5,7 @@ PEERDIR(
     ydb/core/discovery
     ydb/core/engine/minikql
     ydb/core/protos
+    ydb/core/tablet
     ydb/core/tablet_flat
     ydb/core/tx/replication/common
     ydb/core/tx/replication/ydb_proxy