[+] tracing

Author: Alek5andr-Kotov

ydb/core/persqueue/events/internal.h

@@ -198,6 +198,7 @@ struct TEvPQ {
         EvRunCompaction,
         EvMirrorTopicDescription,
         EvBroadcastPartitionError,
+        EvForceCompaction,
         EvEnd
     };
 
@@ -1278,6 +1279,15 @@ struct TEvPQ {
 
         ui64 BlobsCount = 0;
     };
+
+    struct TEvForceCompaction : TEventLocal<TEvForceCompaction, EvForceCompaction> {
+        explicit TEvForceCompaction(const ui32 partitionId) :
+            PartitionId(partitionId)
+        {
+        }
+
+        ui32 PartitionId = 0;
+    };
 };
 
 } //NKikimr

ydb/core/persqueue/pqtablet/partition/partition.h

@@ -246,6 +246,7 @@ class TPartition : public TBaseActor<TPartition> {
     void Handle(TEvents::TEvPoisonPill::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvPQ::TEvSubDomainStatus::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvPQ::TEvRunCompaction::TPtr& ev);
+    void Handle(TEvPQ::TEvForceCompaction::TPtr& ev);
     void Handle(TEvPQ::TEvExclusiveLockAcquired::TPtr& ev);
     void Handle(TEvPQ::TBroadcastPartitionError::TPtr& ev, const TActorContext& ctx);
     void HandleMonitoring(TEvPQ::TEvMonRequest::TPtr& ev, const TActorContext& ctx);
@@ -596,6 +597,7 @@ class TPartition : public TBaseActor<TPartition> {
             HFuncTraced(TEvPQ::TEvDeletePartition, HandleOnInit);
             IgnoreFunc(TEvPQ::TEvTxBatchComplete);
             hFuncTraced(TEvPQ::TEvRunCompaction, Handle);
+            hFuncTraced(TEvPQ::TEvForceCompaction, Handle);
         default:
             if (!Initializer.Handle(ev)) {
                 ALOG_ERROR(NKikimrServices::PERSQUEUE, "Unexpected " << EventStr("StateInit", ev));
@@ -664,6 +666,7 @@ class TPartition : public TBaseActor<TPartition> {
             HFuncTraced(TEvPQ::TEvDeletePartition, Handle);
             IgnoreFunc(TEvPQ::TEvTxBatchComplete);
             hFuncTraced(TEvPQ::TEvRunCompaction, Handle);
+            hFuncTraced(TEvPQ::TEvForceCompaction, Handle);
         default:
             ALOG_ERROR(NKikimrServices::PERSQUEUE, "Unexpected " << EventStr("StateIdle", ev));
             break;
@@ -1116,7 +1119,7 @@ class TPartition : public TBaseActor<TPartition> {
                         const TEvPQ::TEvBlobResponse* blobResponse,
                         const TActorContext& ctx);
 
-    void TryRunCompaction();
+    void TryRunCompaction(bool force = false);
     void BlobsForCompactionWereRead(const TVector<NPQ::TRequestedBlob>& blobs);
     void BlobsForCompactionWereWrite();
     ui64 NextReadCookie();

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

@@ -161,7 +161,7 @@ void TPartition::DumpKeysForBlobsCompaction() const
     LOG_D("===================================");
 }
 
-void TPartition::TryRunCompaction()
+void TPartition::TryRunCompaction(bool force)
 {
     if (CompactionInProgress) {
         LOG_D("Blobs compaction in progress");
@@ -174,12 +174,11 @@ void TPartition::TryRunCompaction()
     }
 
     //DumpKeysForBlobsCompaction();
-    DumpZones(__FILE__, __LINE__);
 
     const ui64 blobsKeyCountLimit = GetBodyKeysCountLimit();
     const ui64 compactedBlobSizeLowerBound = GetCompactedBlobSizeLowerBound();
 
-    if ((BlobEncoder.DataKeysBody.size() < blobsKeyCountLimit) && (BlobEncoder.GetSize() < GetCumulativeSizeLimit())) {
+    if ((BlobEncoder.DataKeysBody.size() < blobsKeyCountLimit) && (BlobEncoder.GetSize() < GetCumulativeSizeLimit()) && !force) {
         LOG_D("No data for blobs compaction");
         return;
     }
@@ -200,13 +199,19 @@ void TPartition::TryRunCompaction()
             LOG_D("Blob key for rename " << k.Key.ToString());
         }
     }
+
     LOG_D(blobsCount << " keys were taken away. Let's read " << blobsSize << " bytes");
 
     CompactionInProgress = true;
 
     Send(SelfId(), new TEvPQ::TEvRunCompaction(blobsCount));
 }
 
+void TPartition::Handle(TEvPQ::TEvForceCompaction::TPtr&)
+{
+    TryRunCompaction(true);
+}
+
 void TPartition::Handle(TEvPQ::TEvRunCompaction::TPtr& ev)
 {
     const ui64 blobsCount = ev->Get()->BlobsCount;
@@ -261,11 +266,14 @@ bool TPartition::CompactRequestedBlob(const TRequestedBlob& requestedBlob,
     TMaybe<ui64> firstBlobOffset = requestedBlob.Offset;
 
     for (TBlobIterator it(requestedBlob.Key, requestedBlob.Value); it.IsValid(); it.Next()) {
+        LOG_D("Compaction: case 18");
         TBatch batch = it.GetBatch();
         batch.Unpack();
 
         for (const auto& blob : batch.Blobs) {
+            LOG_D("Compaction: case 19");
             if (wasThePreviousBlobBig && blob.PartData && (blob.PartData->PartNo != 0)) {
+                LOG_D("Compaction: case 20");
                 // надо продолжить писать большое сообщение
                 CompactionBlobEncoder.NewHead.PartNo = blob.PartData->PartNo;
                 CompactionBlobEncoder.NewPartitionedBlob(Partition,
@@ -278,6 +286,8 @@ bool TPartition::CompactRequestedBlob(const TRequestedBlob& requestedBlob,
                                                          needToCompactHead,
                                                          MaxBlobSize,
                                                          blob.PartData->PartNo);
+            } else {
+                LOG_D("Compaction: case 21");
             }
             wasThePreviousBlobBig = false;
 
@@ -310,12 +320,17 @@ bool TPartition::CompactRequestedBlob(const TRequestedBlob& requestedBlob,
 
             blobCreationUnixTime = std::max(blobCreationUnixTime, blob.WriteTimestamp);
             if (!ExecRequestForCompaction(msg, parameters, compactionRequest, blobCreationUnixTime)) {
+                LOG_D("Compaction: case 22");
                 return false;
+            } else {
+                LOG_D("Compaction: case 23");
             }
 
             firstBlobOffset = Nothing();
         }
+        LOG_D("Compaction: case 25");
     }
+    LOG_D("Compaction: case 26");
 
     return true;
 }
@@ -329,6 +344,7 @@ void TPartition::RenameCompactedBlob(TDataKey& k,
     const auto& ctx = ActorContext();
 
     if (!CompactionBlobEncoder.PartitionedBlob.IsInited()) {
+        LOG_D("Compaction: case 12");
         CompactionBlobEncoder.NewPartitionedBlob(Partition,
                                                  CompactionBlobEncoder.NewHead.Offset,
                                                  "",                      // SourceId
@@ -338,22 +354,30 @@ void TPartition::RenameCompactedBlob(TDataKey& k,
                                                  parameters.HeadCleared,  // headCleared
                                                  needToCompactHead,       // needCompactHead
                                                  MaxBlobSize);
+    } else {
+        LOG_D("Compaction: case 13");
     }
     auto write = CompactionBlobEncoder.PartitionedBlob.Add(k.Key, size, k.Timestamp, false);
     if (write && !write->Value.empty()) {
+        LOG_D("Compaction: case 14");
         // надо записать содержимое головы перед первым большим блобом
         AddCmdWrite(write, compactionRequest, k.Timestamp, ctx);
         CompactionBlobEncoder.CompactedKeys.emplace_back(write->Key, write->Value.size());
+    } else {
+        LOG_D("Compaction: case 15");
     }
 
     if (const auto& formedBlobs = CompactionBlobEncoder.PartitionedBlob.GetFormedBlobs(); !formedBlobs.empty()) {
+        LOG_D("Compaction: case 16");
         ui32 curWrites = RenameTmpCmdWrites(compactionRequest);
         RenameFormedBlobs(formedBlobs,
                           parameters,
                           curWrites,
                           compactionRequest,
                           CompactionBlobEncoder,
                           ctx);
+    } else {
+        LOG_D("Compaction: case 17");
     }
 
     k.BlobKeyToken->NeedDelete = false;
@@ -376,8 +400,11 @@ void TPartition::BlobsForCompactionWereRead(const TVector<NPQ::TRequestedBlob>&
     if (!CompactionBlobEncoder.Head.GetCount() &&
         !CompactionBlobEncoder.NewHead.GetCount() &&
         CompactionBlobEncoder.IsEmpty()) {
+        LOG_D("Compaction: case 09");
         // если это первое сообщение, то надо поправить StartOffset
         CompactionBlobEncoder.StartOffset = BlobEncoder.StartOffset;
+    } else {
+        LOG_D("Compaction: case 10");
     }
 
     CompactionBlobEncoder.NewHead.Clear();
@@ -399,15 +426,20 @@ void TPartition::BlobsForCompactionWereRead(const TVector<NPQ::TRequestedBlob>&
     DumpZones(__FILE__, __LINE__);
 
     for (size_t i = 0; i < KeysForCompaction.size(); ++i) {
+        LOG_D("Compaction: case 11");
         auto& [k, pos] = KeysForCompaction[i];
         bool needToCompactHead = (parameters.CurOffset < k.Key.GetOffset());
 
         if (pos == Max<size_t>()) {
+            LOG_D("Compaction: case 01");
             // большой блоб надо переименовать
             LOG_D("Rename key " << k.Key.ToString());
 
             if (!WasTheLastBlobBig) {
+                LOG_D("Compaction: case 02");
                 needToCompactHead = true;
+            } else {
+                LOG_D("Compaction: case 03");
             }
             LOG_D("Need to compact head " << needToCompactHead);
 
@@ -427,32 +459,41 @@ void TPartition::BlobsForCompactionWereRead(const TVector<NPQ::TRequestedBlob>&
 
             WasTheLastBlobBig = true;
         } else {
+            LOG_D("Compaction: case 04");
             // маленький блоб надо дописать
             LOG_D("Append blob for key " << k.Key.ToString());
             LOG_D("Need to compact head " << needToCompactHead);
 
             const TRequestedBlob& requestedBlob = blobs[pos];
             if (!CompactRequestedBlob(requestedBlob, parameters, needToCompactHead, compactionRequest.Get(), blobCreationUnixTime, WasTheLastBlobBig)) {
+                LOG_D("Compaction: case 05");
                 LOG_D("Can't append blob for key " << k.Key.ToString());
                 Y_FAIL("Something went wrong");
                 return;
+            } else {
+                LOG_D("Compaction: case 06");
             }
 
             WasTheLastBlobBig = false;
         }
 
         DumpZones(__FILE__, __LINE__);
     }
+    LOG_D("Compaction: case 24");
 
     if (!CompactionBlobEncoder.IsLastBatchPacked()) {
+        LOG_D("Compaction: case 07");
         CompactionBlobEncoder.PackLastBatch();
+    } else {
+        LOG_D("Compaction: case 08");
     }
 
     CompactionBlobEncoder.HeadCleared = parameters.HeadCleared;
 
     EndProcessWritesForCompaction(compactionRequest.Get(), blobCreationUnixTime, ctx);
 
     DumpZones(__FILE__, __LINE__);
+
     // for debugging purposes
     DumpKeyValueRequest(compactionRequest->Record);
 

ydb/core/persqueue/pqtablet/pq_impl.cpp

@@ -5238,6 +5238,23 @@ void TPersQueue::ProcessPendingEvents()
     }
 }
 
+void TPersQueue::Handle(TEvPQ::TEvForceCompaction::TPtr& ev, const TActorContext& ctx)
+{
+    PQ_LOG_D("TPersQueue::Handle(TEvPQ::TEvForceCompaction)");
+
+    const auto& event = *ev->Get();
+    const TPartitionId partitionId(event.PartitionId);
+
+    if (!Partitions.contains(partitionId)) {
+        PQ_LOG_D("Unknown partition id " << event.PartitionId);
+        return;
+    }
+
+    auto p = Partitions.find(partitionId);
+    ctx.Send(p->second.Actor,
+             new TEvPQ::TEvForceCompaction(event.PartitionId));
+}
+
 bool TPersQueue::HandleHook(STFUNC_SIG)
 {
     TRACE_EVENT(NKikimrServices::PERSQUEUE);
@@ -5285,6 +5302,7 @@ bool TPersQueue::HandleHook(STFUNC_SIG)
         HFuncTraced(TEvPQ::TEvReadingPartitionStatusRequest, Handle);
         HFuncTraced(TEvPQ::TEvDeletePartitionDone, Handle);
         HFuncTraced(TEvPQ::TEvTransactionCompleted, Handle);
+        HFuncTraced(TEvPQ::TEvForceCompaction, Handle);
         default:
             return false;
     }

ydb/core/persqueue/pqtablet/pq_impl.h

@@ -603,6 +603,7 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
 
     void ResendSplitMergeRequests(const TActorContext& ctx);
 
+    void Handle(TEvPQ::TEvForceCompaction::TPtr& ev, const TActorContext& ctx);
 
     TIntrusivePtr<NJaegerTracing::TSamplingThrottlingControl> SamplingControl;
     NWilson::TSpan WriteTxsSpan;

ydb/core/persqueue/ut/common/pq_ut_common.cpp

@@ -1250,4 +1250,19 @@ THolder<TEvPersQueue::TEvPeriodicTopicStats> GetReadBalancerPeriodicTopicStats(T
     return runtime.GrabEdgeEvent<TEvPersQueue::TEvPeriodicTopicStats>(TDuration::Seconds(2));
 }
 
+void CmdRunCompaction(TTestActorRuntime& runtime,
+                      ui64 tabletId,
+                      const TActorId& sender,
+                      const ui32 partition)
+{
+    auto event = MakeHolder<TEvPQ::TEvForceCompaction>(partition);
+    runtime.SendToPipe(tabletId, sender, event.Release(), 0, GetPipeConfigWithRetries());
+}
+
+void CmdRunCompaction(const ui32 partition,
+                      TTestContext& tc)
+{
+    CmdRunCompaction(*tc.Runtime, tc.TabletId, tc.Edge, partition);
+}
+
 } // namespace NKikimr::NPQ

ydb/core/persqueue/ut/common/pq_ut_common.h

@@ -644,6 +644,13 @@ struct TCmdWriteOptions {
 };
 void CmdWrite(const TCmdWriteOptions&);
 
+void CmdRunCompaction(TTestActorRuntime& runtime,
+                      ui64 tabletId,
+                      const TActorId& sender,
+                      const ui32 partition);
+void CmdRunCompaction(const ui32 partition,
+                      TTestContext& tc);
+
 THolder<TEvPersQueue::TEvPeriodicTopicStats> GetReadBalancerPeriodicTopicStats(TTestActorRuntime& runtime, ui64 balancerId);
 
 } // namespace NKikimr::NPQ