[+] Counter PQ/LatencyTxCalcPredicate #28149

Author: Alek5andr-Kotov

ydb/core/persqueue/partition.cpp

@@ -1532,6 +1532,8 @@ void TPartition::Handle(TEvPQ::TEvGetWriteInfoError::TPtr& ev, const TActorConte
 void TPartition::ReplyToProposeOrPredicate(TSimpleSharedPtr<TTransaction>& tx, bool isPredicate) {
 
     if (isPredicate) {
+        TabletCounters.Percentile()[COUNTER_LATENCY_PQ_TXCALCPREDICATE].IncrementFor((Now() - tx->CalcPredicateTimestamp).MilliSeconds());
+
         tx->CalcPredicateSpan.End();
         tx->CalcPredicateSpan = {};
 
@@ -2044,7 +2046,7 @@ void TPartition::Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext&
 
 void TPartition::PushBackDistrTx(TSimpleSharedPtr<TEvPQ::TEvTxCalcPredicate> event)
 {
-    UserActionAndTransactionEvents.emplace_back(MakeSimpleShared<TTransaction>(std::move(event)));
+    UserActionAndTransactionEvents.emplace_back(MakeSimpleShared<TTransaction>(std::move(event), Now()));
     RequestWriteInfoIfRequired();
 }
 

ydb/core/persqueue/partition.h

@@ -63,12 +63,14 @@ class TPartitionCompaction;
 
 struct TTransaction {
 
-    explicit TTransaction(TSimpleSharedPtr<TEvPQ::TEvTxCalcPredicate> tx,
-                          TMaybe<bool> predicate = Nothing())
+    TTransaction(TSimpleSharedPtr<TEvPQ::TEvTxCalcPredicate> tx,
+                 TInstant calcPredicateTimestamp,
+                 TMaybe<bool> predicate = Nothing())
         : Tx(tx)
         , Predicate(predicate)
         , SupportivePartitionActor(tx->SupportivePartitionActor)
         , CalcPredicateSpan(std::move(tx->Span))
+        , CalcPredicateTimestamp(calcPredicateTimestamp)
     {
         Y_ABORT_UNLESS(Tx);
     }
@@ -128,6 +130,7 @@ struct TTransaction {
     NWilson::TSpan CommitSpan;
 
     TInstant WriteInfoResponseTimestamp;
+    TInstant CalcPredicateTimestamp;
 };
 class TPartitionCompaction;
 

ydb/core/persqueue/ut/partition_ut.cpp

@@ -281,11 +281,6 @@ class TPartitionFixture : public NUnitTest::TBaseFixture {
     void SendChangePartitionConfig(const TConfigParams& config = {});
     void WaitPartitionConfigChanged(const TChangePartitionConfigMatcher& matcher = {});
 
-    TTransaction MakeTransaction(ui64 step, ui64 txId,
-                                 TString consumer,
-                                 ui64 begin, ui64 end,
-                                 TMaybe<bool> predicate = Nothing());
-
     void SendSubDomainStatus(bool subDomainOutOfSpace = false);
     void SendReserveBytes(const ui64 cookie, const ui32 size, const TString& ownerCookie, const ui64 messageNo, bool lastRequest = false);
     void SendChangeOwner(const ui64 cookie, const TString& owner, const TActorId& pipeClient, const bool force = true);
@@ -1098,17 +1093,6 @@ void TPartitionFixture::WaitPartitionConfigChanged(const TChangePartitionConfigM
     }
 }
 
-TTransaction TPartitionFixture::MakeTransaction(ui64 step, ui64 txId,
-                                                TString consumer,
-                                                ui64 begin, ui64 end,
-                                                TMaybe<bool> predicate)
-{
-    auto event = MakeSimpleShared<TEvPQ::TEvTxCalcPredicate>(step, txId);
-    event->AddOperation(std::move(consumer), begin, end);
-
-    return TTransaction(event, predicate);
-}
-
 template<class TIterable>
 void CompareVectors(const TVector<ui64>& expected, const TIterable& actual) {
     auto i = 0u;

ydb/core/protos/counters_pq.proto

@@ -141,6 +141,8 @@ enum EPercentileCounters {
     COUNTER_LATENCY_PQ_SPLIT_MESSAGE_GROUP = 19                 [(CounterOpts) = {Name: "LatencySplitMessageGroup"}];
     COUNTER_LATENCY_PQ_PUBLISH_READ = 20                        [(CounterOpts) = {Name: "LatencyPublishRead"}];
     COUNTER_LATENCY_PQ_FORGET_READ = 21                         [(CounterOpts) = {Name: "LatencyForgetRead"}];
+
+    COUNTER_LATENCY_PQ_TXCALCPREDICATE = 22                     [(CounterOpts) = {Name: "LatencyTxCalcPredicate"}];
 }