Counter PQ/LatencyTxCalcPredicate

ydb/core/persqueue/pqtablet/partition/partition.cpp

@@ -1663,6 +1663,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 = {};
 
@@ -2195,7 +2197,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/pqtablet/partition/partition.h

@@ -57,13 +57,14 @@ class IAutopartitioningManager;
 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)
     {
         AFL_ENSURE(Tx);
     }
@@ -123,6 +124,7 @@ struct TTransaction {
     NWilson::TSpan CommitSpan;
 
     TInstant WriteInfoResponseTimestamp;
+    TInstant CalcPredicateTimestamp;
 };
 class TPartitionCompaction;
 

ydb/core/persqueue/ut/partition_ut.cpp

@@ -282,11 +282,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);
@@ -1106,17 +1101,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"}];
 }