Add volatile transactions to datashard internal ui (#10859)

Author: snaury

ydb/core/cms/ui/datashard_info.js

@@ -105,6 +105,7 @@ function onDataShardInfoLoaded(data) {
     $('#tablet-info-state').text(info.State + (info.IsActive ? ' (active)' : ' (inactive)'));
     $('#tablet-info-shared-blobs').text(info.HasSharedBlobs);
     $('#tablet-info-change-sender').html('<a href="app?TabletID=' + TabletId + '&page=change-sender">Viewer</a>');
+    $('#tablet-info-volatile-txs').html(`<a href="app?TabletID=${TabletId}&page=volatile-txs">Viewer</a>`);
 
     var activities = data.Activities;
     if (activities) {

ydb/core/cms/ui/datashard_rs.js

@@ -8,12 +8,22 @@ var ReadSetsState = {
     rs: new Map(),
     acks: new Map(),
     delayedAcks: new Map(),
+    expectations: new Map(),
+    pipes: new Map(),
 };
 
 function makeRSKey(info) {
     return `${info.TxId}-${info.Origin}-${info.Source}-${info.Destination}-${info.SeqNo}`;
 }
 
+function makeRSExpectationKey(info) {
+    return `${info.TxId}-${info.Source}`;
+}
+
+function makeRSPipeKey(info) {
+    return `${info.Destination}`;
+}
+
 class RSInfo {
     constructor(info) {
         this.key = makeRSKey(info);
@@ -72,6 +82,114 @@ class RSAckInfo {
     }
 }
 
+class RSExpectationInfo {
+    constructor(info, body) {
+        this.key = makeRSExpectationKey(info);
+
+        var trHtml = this._makeTrHtml(info);
+        $(trHtml).appendTo($('#' + body));
+    }
+
+    update(info) {
+        $('#ds-rs-expectation-row-' + this.key).replaceWith(this._makeTrHtml(info));
+    }
+
+    remove() {
+        $('#ds-rs-expectation-row-' + this.key).remove();
+    }
+
+    _makeTrHtml(info) {
+        return `
+            <tr id="ds-rs-expectation-row-${this.key}">
+                <td>${info.TxId}</td>
+                <td>${info.Step}</td>
+                <td><a href="app?TabletID=${info.Source}">${info.Source}</a></td>
+            </tr>
+        `;
+    }
+}
+
+class RSPipeInfo {
+    constructor(info, body) {
+        this.key = makeRSPipeKey(info);
+
+        var trHtml = this._makeTrHtml(info);
+        $(trHtml).appendTo($('#' + body));
+    }
+
+    update(info) {
+        $('#ds-rs-pipe-row-' + this.key).replaceWith(this._makeTrHtml(info));
+    }
+
+    remove() {
+        $('#ds-rs-pipe-row-' + this.key).remove();
+    }
+
+    _makeTrHtml(info) {
+        return `
+            <tr id="ds-rs-pipe-row-${this.key}">
+                <td><a href="app?TabletID=${info.Destination}">${info.Destination}</a></td>
+                <td>${info.OutReadSets}</td>
+                <td>${info.Subscribed}</td>
+            </tr>
+        `;
+    }
+}
+
+function updateReadSetExpectations(data) {
+    var expectations = new Set();
+    if (data.Expectations) {
+        for (var info of data.Expectations) {
+            var key = makeRSExpectationKey(info);
+            expectations.add(key);
+            if (ReadSetsState.expectations.has(key)) {
+                ReadSetsState.expectations.get(key).update(info);
+            } else {
+                ReadSetsState.expectations.set(key, new RSExpectationInfo(info, 'ds-rs-expectations-body'));
+            }
+        }
+    }
+
+    var toRemove = [];
+    for (var key of ReadSetsState.expectations.keys()) {
+        if (!expectations.has(key)) {
+            toRemove.push(key);
+        }
+    }
+
+    for (var key of toRemove) {
+        ReadSetsState.expectations.get(key).remove();
+        ReadSetsState.expectations.delete(key);
+    }
+}
+
+function updateReadSetPipes(data) {
+    var pipes = new Set();
+    if (data.Pipes) {
+        for (var info of data.Pipes) {
+            var key = makeRSPipeKey(info);
+            pipes.add(key);
+            if (ReadSetsState.pipes.has(key)) {
+                ReadSetsState.pipes.get(key).update(info);
+            } else {
+                ReadSetsState.pipes.set(key, new RSPipeInfo(info, 'ds-rs-pipes-body'));
+            }
+        }
+    }
+
+    var toRemove = [];
+    for (var key of ReadSetsState.pipes.keys()) {
+        if (!pipes.has(key)) {
+            toRemove.push(key);
+        }
+    }
+
+    for (var key of toRemove) {
+        ReadSetsState.pipes.get(key).remove();
+        ReadSetsState.pipes.delete(key);
+    }
+}
+
 function onReadSetsLoaded(data) {
     ReadSetsState.loading = false;
 
@@ -154,9 +272,14 @@ function onReadSetsLoaded(data) {
         ReadSetsState.delayedAcks.delete(key);
     }
 
+    updateReadSetExpectations(data);
+    updateReadSetPipes(data);
+
     $('#ds-out-rs-table').trigger('update', [true]);
     $('#ds-out-rs-ack-table').trigger('update', [true]);
     $('#ds-delayed-ack-table').trigger('update', [true]);
+    $('#ds-rs-expectations-table').trigger('update', [true]);
+    $('#ds-rs-pipes-table').trigger('update', [true]);
 
     scheduleLoadReadSets(ReadSetsState.fetchInterval);
 }
@@ -213,6 +336,16 @@ function initReadSetsTab() {
                 sortList: [[0,0]],
                 widgets : ['zebra', 'filter'],
             });
+            $('#ds-rs-expectations-table').tablesorter({
+                theme: 'blue',
+                sortList: [[0,0]],
+                widgets : ['zebra', 'filter'],
+            });
+            $('#ds-rs-pipes-table').tablesorter({
+                theme: 'blue',
+                sortList: [[0,0]],
+                widgets : ['zebra', 'filter'],
+            });
             scheduleLoadReadSets(0);
         }
     });

ydb/core/protos/tx_datashard.proto

@@ -1205,10 +1205,24 @@ message TEvGetRSInfoResponse {
         optional uint64 SeqNo = 6;
     }
 
+    message TExpectation {
+        optional uint64 Source = 1;
+        optional uint64 TxId = 2;
+        optional uint64 Step = 3;
+    }
+
+    message TPipe {
+        optional uint64 Destination = 1;
+        optional uint64 OutReadSets = 2;
+        optional bool Subscribed = 3;
+    }
+
     optional TStatus Status = 1;
     repeated TOutRSInfo OutReadSets = 2;
     repeated TRSAckInfo OutRSAcks = 3;
     repeated TRSAckInfo DelayedRSAcks = 4;
+    repeated TExpectation Expectations = 5;
+    repeated TPipe Pipes = 6;
 }
 
 message TEvGetDataHistogramRequest {

ydb/core/tx/datashard/datashard.cpp

@@ -2223,12 +2223,12 @@ bool TDataShard::OnRenderAppHtmlPage(NMon::TEvRemoteHttpInfo::TPtr ev, const TAc
 
     if (const auto& action = cgi.Get("action")) {
         if (action == "cleanup-borrowed-parts") {
-            Execute(CreateTxMonitoringCleanupBorrowedParts(this, ev), ctx);
+            HandleMonCleanupBorrowedParts(ev);
             return true;
         }
 
         if (action == "reset-schema-version") {
-            Execute(CreateTxMonitoringResetSchemaVersion(this, ev), ctx);
+            HandleMonResetSchemaVersion(ev);
             return true;
         }
 
@@ -2254,13 +2254,16 @@ bool TDataShard::OnRenderAppHtmlPage(NMon::TEvRemoteHttpInfo::TPtr ev, const TAc
                 ctx.Send(ev->Sender, new NMon::TEvRemoteHttpInfoRes("Change sender is not running"));
                 return true;
             }
+        } else if (page == "volatile-txs") {
+            HandleMonVolatileTxs(ev);
+            return true;
         } else {
             ctx.Send(ev->Sender, new NMon::TEvRemoteBinaryInfoRes(NMonitoring::HTTPNOTFOUND));
             return true;
         }
     }
 
-    Execute(CreateTxMonitoring(this, ev), ctx);
+    HandleMonIndexPage(ev);
     return true;
 }
 
@@ -4181,60 +4184,6 @@ void TDataShard::Handle(TEvSchemeShard::TEvDescribeSchemeResult::TPtr ev, const
     Execute(new TTxStoreTablePath(this, pathId, rec.GetPath()), ctx);
 }
 
-void TDataShard::Handle(TEvDataShard::TEvGetInfoRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    Execute(CreateTxGetInfo(this, ev), ctx);
-}
-
-void TDataShard::Handle(TEvDataShard::TEvListOperationsRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    Execute(CreateTxListOperations(this, ev), ctx);
-}
-
-void TDataShard::Handle(TEvDataShard::TEvGetOperationRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    Execute(CreateTxGetOperation(this, ev), ctx);
-}
-
-void TDataShard::Handle(TEvDataShard::TEvGetDataHistogramRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    auto *response = new TEvDataShard::TEvGetDataHistogramResponse;
-    response->Record.MutableStatus()->SetCode(Ydb::StatusIds::SUCCESS);
-    const auto& rec = ev->Get()->Record;
-
-    if (rec.GetCollectKeySampleMs() > 0) {
-        EnableKeyAccessSampling(ctx,
-            AppData(ctx)->TimeProvider->Now() + TDuration::MilliSeconds(rec.GetCollectKeySampleMs()));
-    }
-
-    if (rec.GetActualData()) {
-        if (CurrentKeySampler == DisabledKeySampler) {
-            // datashard stores expired stats
-            ctx.Send(ev->Sender, response);
-            return;
-        }
-    }
-
-    for (const auto &pr : TableInfos) {
-        const auto &tinfo = *pr.second;
-        const NTable::TStats &stats = tinfo.Stats.DataStats;
-
-        auto &hist = *response->Record.AddTableHistograms();
-        hist.SetTableName(pr.second->Name);
-        for (ui32 ki : tinfo.KeyColumnIds)
-            hist.AddKeyNames(tinfo.Columns.FindPtr(ki)->Name);
-        SerializeHistogram(tinfo, stats.DataSizeHistogram, *hist.MutableSizeHistogram());
-        SerializeHistogram(tinfo, stats.RowCountHistogram, *hist.MutableCountHistogram());
-        SerializeKeySample(tinfo, tinfo.Stats.AccessStats, *hist.MutableKeyAccessSample());
-    }
-
-    ctx.Send(ev->Sender, response);
-}
-
 void TDataShard::Handle(TEvDataShard::TEvGetReadTableSinkStateRequest::TPtr &ev,
                                const TActorContext &ctx)
 {
@@ -4329,60 +4278,6 @@ void TDataShard::Handle(TEvDataShard::TEvGetReadTableStreamStateRequest::TPtr &e
     ctx.Send(ev->Forward(tx->GetStreamSink()));
 }
 
-void TDataShard::Handle(TEvDataShard::TEvGetRSInfoRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    auto *response = new TEvDataShard::TEvGetRSInfoResponse;
-    response->Record.MutableStatus()->SetCode(Ydb::StatusIds::SUCCESS);
-
-    for (auto &pr : OutReadSets.CurrentReadSets) {
-        auto &rs = *response->Record.AddOutReadSets();
-        rs.SetTxId(pr.second.TxId);
-        rs.SetOrigin(pr.second.Origin);
-        rs.SetSource(pr.second.From);
-        rs.SetDestination(pr.second.To);
-        rs.SetSeqNo(pr.first);
-    }
-
-    for (auto &p : OutReadSets.ReadSetAcks) {
-        auto &rec = p->Record;
-        auto &ack = *response->Record.AddOutRSAcks();
-        ack.SetTxId(rec.GetTxId());
-        ack.SetStep(rec.GetStep());
-        ack.SetOrigin(rec.GetTabletConsumer());
-        ack.SetSource(rec.GetTabletSource());
-        ack.SetDestination(rec.GetTabletDest());
-        ack.SetSeqNo(rec.GetSeqno());
-    }
-
-    for (auto &pr : Pipeline.GetDelayedAcks()) {
-        for (auto &ack : pr.second) {
-            auto *ev = ack->CastAsLocal<TEvTxProcessing::TEvReadSetAck>();
-            if (ev) {
-                auto &rec = ev->Record;
-                auto &ack = *response->Record.AddDelayedRSAcks();
-                ack.SetTxId(rec.GetTxId());
-                ack.SetStep(rec.GetStep());
-                ack.SetOrigin(rec.GetTabletConsumer());
-                ack.SetSource(rec.GetTabletSource());
-                ack.SetDestination(rec.GetTabletDest());
-                ack.SetSeqNo(rec.GetSeqno());
-            }
-        }
-    }
-
-    ctx.Send(ev->Sender, response);
-}
-
-void TDataShard::Handle(TEvDataShard::TEvGetSlowOpProfilesRequest::TPtr &ev,
-                               const TActorContext &ctx)
-{
-    auto *response = new TEvDataShard::TEvGetSlowOpProfilesResponse;
-    response->Record.MutableStatus()->SetCode(Ydb::StatusIds::SUCCESS);
-    Pipeline.FillStoredExecutionProfiles(response->Record);
-    ctx.Send(ev->Sender, response);
-}
-
 void TDataShard::Handle(TEvDataShard::TEvRefreshVolatileSnapshotRequest::TPtr& ev, const TActorContext& ctx) {
     Execute(new TTxRefreshVolatileSnapshot(this, std::move(ev)), ctx);
 }

ydb/core/tx/datashard/datashard__cleanup_borrowed.cpp

@@ -255,11 +255,8 @@ class TDataShard::TTxMonitoringCleanupBorrowedParts
     bool DryRun = true;
 };
 
-ITransaction* TDataShard::CreateTxMonitoringCleanupBorrowedParts(
-        TDataShard* self,
-        NMon::TEvRemoteHttpInfo::TPtr ev)
-{
-    return new TTxMonitoringCleanupBorrowedParts(self, ev);
+void TDataShard::HandleMonCleanupBorrowedParts(NMon::TEvRemoteHttpInfo::TPtr& ev) {
+    Execute(new TTxMonitoringCleanupBorrowedParts(this, ev));
 }
 
 }

ydb/core/tx/datashard/datashard__mon_reset_schema_version.cpp

@@ -54,11 +54,8 @@ class TDataShard::TTxMonitoringResetSchemaVersion
     size_t Updates = 0;
 };
 
-ITransaction* TDataShard::CreateTxMonitoringResetSchemaVersion(
-        TDataShard* self,
-        NMon::TEvRemoteHttpInfo::TPtr ev)
-{
-    return new TTxMonitoringResetSchemaVersion(self, ev);
+void TDataShard::HandleMonResetSchemaVersion(NMon::TEvRemoteHttpInfo::TPtr& ev) {
+    Execute(new TTxMonitoringResetSchemaVersion(this, ev));
 }
 
 }