Metrics for buffer actor (#11304)

ydb-platform · Nov 6, 2024 · 5c69ec9 · 5c69ec9
1 parent 2bf1a6c
commit 5c69ec9
Show file tree

Hide file tree

Showing 6 changed files with 148 additions and 11 deletions.
diff --git a/ydb/core/kqp/common/buffer/buffer.h b/ydb/core/kqp/common/buffer/buffer.h
@@ -9,6 +9,8 @@ namespace NKqp {
 struct TKqpBufferWriterSettings {
     TActorId SessionActorId;
     IKqpTransactionManagerPtr TxManager;
+    NWilson::TTraceId TraceId;
+    TIntrusivePtr<TKqpCounters> Counters;
 };
 
 NActors::IActor* CreateKqpBufferWriterActor(TKqpBufferWriterSettings&& settings);

diff --git a/ydb/core/kqp/counters/kqp_counters.cpp b/ydb/core/kqp/counters/kqp_counters.cpp
@@ -817,14 +817,29 @@ TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, co
     /* sink writes */
     WriteActorsShardResolve = KqpGroup->GetCounter("SinkWrites/WriteActorShardResolve", true);
     WriteActorsCount = KqpGroup->GetCounter("SinkWrites/WriteActorsCount", false);
+    BufferActorsCount = KqpGroup->GetCounter("SinkWrites/BufferActorsCount", false);
+    ForwardActorsCount = KqpGroup->GetCounter("SinkWrites/ForwardActorsCount", false);
+
     WriteActorImmediateWrites = KqpGroup->GetCounter("SinkWrites/WriteActorImmediateWrites", true);
     WriteActorImmediateWritesRetries = KqpGroup->GetCounter("SinkWrites/WriteActorImmediateWritesRetries", true);
+    WriteActorPrepareWrites = KqpGroup->GetCounter("SinkWrites/WriteActorPrepareWrites", true);
+
+    BufferActorFlushes = KqpGroup->GetCounter("SinkWrites/BufferActorFlushes", true);
+    BufferActorImmediateCommits = KqpGroup->GetCounter("SinkWrites/BufferActorImmediateCommits", true);
+    BufferActorDistributedCommits = KqpGroup->GetCounter("SinkWrites/BufferActorDistributedCommits", true);
+    BufferActorRollbacks = KqpGroup->GetCounter("SinkWrites/BufferActorRollbacks", true);
+
     WriteActorWritesSizeHistogram =
         KqpGroup->GetHistogram("SinkWrites/WriteActorWritesSize", NMonitoring::ExponentialHistogram(28, 2, 1));
     WriteActorWritesOperationsHistogram =
         KqpGroup->GetHistogram("SinkWrites/WriteActorWritesOperations", NMonitoring::ExponentialHistogram(20, 2, 1));
     WriteActorWritesLatencyHistogram =
         KqpGroup->GetHistogram("SinkWrites/WriteActorWritesLatencyMs", NMonitoring::ExponentialHistogram(20, 2, 1));
+
+    ForwardActorWritesSizeHistogram =
+        KqpGroup->GetHistogram("SinkWrites/ForwardActorWritesSize", NMonitoring::ExponentialHistogram(28, 2, 1));
+    ForwardActorWritesLatencyHistogram =
+        KqpGroup->GetHistogram("SinkWrites/ForwardActorWritesLatencyMs", NMonitoring::ExponentialHistogram(20, 2, 1));
 
     /* sequencers */
 

diff --git a/ydb/core/kqp/counters/kqp_counters.h b/ydb/core/kqp/counters/kqp_counters.h
@@ -412,12 +412,25 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
     // Sink write counters
     ::NMonitoring::TDynamicCounters::TCounterPtr WriteActorsShardResolve;
     ::NMonitoring::TDynamicCounters::TCounterPtr WriteActorsCount;
+    ::NMonitoring::TDynamicCounters::TCounterPtr BufferActorsCount;
+    ::NMonitoring::TDynamicCounters::TCounterPtr ForwardActorsCount;
+
     ::NMonitoring::TDynamicCounters::TCounterPtr WriteActorImmediateWrites;
     ::NMonitoring::TDynamicCounters::TCounterPtr WriteActorImmediateWritesRetries;
+    ::NMonitoring::TDynamicCounters::TCounterPtr WriteActorPrepareWrites;
+
+    ::NMonitoring::TDynamicCounters::TCounterPtr BufferActorFlushes;
+    ::NMonitoring::TDynamicCounters::TCounterPtr BufferActorImmediateCommits;
+    ::NMonitoring::TDynamicCounters::TCounterPtr BufferActorDistributedCommits;
+    ::NMonitoring::TDynamicCounters::TCounterPtr BufferActorRollbacks;
+
     NMonitoring::THistogramPtr WriteActorWritesSizeHistogram;
     NMonitoring::THistogramPtr WriteActorWritesOperationsHistogram;
     NMonitoring::THistogramPtr WriteActorWritesLatencyHistogram;
 
+    NMonitoring::THistogramPtr ForwardActorWritesSizeHistogram;
+    NMonitoring::THistogramPtr ForwardActorWritesLatencyHistogram;
+
     // Scheduler signals
     ::NMonitoring::TDynamicCounters::TCounterPtr SchedulerThrottled;
     ::NMonitoring::TDynamicCounters::TCounterPtr SchedulerCapacity;