PR from branch users/eivanov89/-simplify-ds-load-actors

fix VERIFY crashes in info collector and polish it

change log to trace

update upsert.sh

3 files changed, 48 insertions, 24 deletions

diff --git a/ydb/core/tx/datashard/testload/bulk_mkql_upsert.cpp b/ydb/core/tx/datashard/testload/bulk_mkql_upsert.cpp
index e0f23650ff8..5440b5c3c01 100644
--- a/ydb/core/tx/datashard/testload/bulk_mkql_upsert.cpp
+++ b/ydb/core/tx/datashard/testload/bulk_mkql_upsert.cpp
@@ -235,7 +235,7 @@ private:
     }
 
     void Handle(TEvDataShard::TEvUploadRowsResponse::TPtr ev, const TActorContext& ctx) {
-        LOG_DEBUG_S(ctx, NKikimrServices::DS_LOAD_TEST, "Tag# " << Tag
+        LOG_TRACE_S(ctx, NKikimrServices::DS_LOAD_TEST, "Tag# " << Tag
             << " TUpsertActor received from " << ev->Sender << ": " << ev->Get()->Record);
         --Inflight;
 
@@ -250,7 +250,7 @@ private:
     }
 
     void Handle(TEvTablet::TEvLocalMKQLResponse::TPtr ev, const TActorContext& ctx) {
-        LOG_DEBUG_S(ctx, NKikimrServices::DS_LOAD_TEST, "Tag# " << Tag
+        LOG_TRACE_S(ctx, NKikimrServices::DS_LOAD_TEST, "Tag# " << Tag
             << " TUpsertActor received from " << ev->Sender << ": " << ev->Get()->Record);
         --Inflight;
 
diff --git a/ydb/core/tx/datashard/testload/info_collector.cpp b/ydb/core/tx/datashard/testload/info_collector.cpp
index da2a002442b..3e5e03fea17 100644
--- a/ydb/core/tx/datashard/testload/info_collector.cpp
+++ b/ydb/core/tx/datashard/testload/info_collector.cpp
@@ -5,7 +5,7 @@ namespace NKikimr::NDataShardLoad {
 
 namespace {
 
-constexpr ui64 WakeupSeconds = 1;
+constexpr ui64 WakeupMs = 500;
 
 class TInfoCollector : public TActorBootstrapped<TInfoCollector> {
     const TActorId Parent;
@@ -30,7 +30,7 @@ public:
             ctx.Send(actorId, new TEvDataShardLoad::TEvTestLoadInfoRequest());
         }
 
-        ctx.Schedule(TDuration::Seconds(WakeupSeconds), new TEvents::TEvWakeup());
+        ctx.Schedule(TDuration::MilliSeconds(WakeupMs), new TEvents::TEvWakeup());
     }
 
     void Handle(TEvDataShardLoad::TEvTestLoadInfoResponse::TPtr& ev, const TActorContext& ctx) {
@@ -54,7 +54,6 @@ public:
         ctx.Send(Parent, response.release());
     }
 
-
     void HandleWakeup(const TActorContext& ctx) {
         LOG_DEBUG_S(ctx, NKikimrServices::DS_LOAD_TEST, "TInfoCollector# " << Parent
             << " actor timeout: waiting# " << ResponsesPending << " out of actorsCount# " << Actors.size());
diff --git a/ydb/core/tx/datashard/testload/test_load_actor.cpp b/ydb/core/tx/datashard/testload/test_load_actor.cpp
index 60146c1248e..62032488b29 100644
--- a/ydb/core/tx/datashard/testload/test_load_actor.cpp
+++ b/ydb/core/tx/datashard/testload/test_load_actor.cpp
@@ -63,6 +63,7 @@ private:
     TString Session;
 
     TActorId HttpInfoCollector;
+    THashSet<TActorId> HttpInfoWaiters;
 
     // info about finished actors
     TVector<TFinishedTestInfo> FinishedTests;
@@ -467,44 +468,56 @@ public:
     }
 
     void Handle(TEvDataShardLoad::TEvTestLoadInfoRequest::TPtr& ev, const TActorContext& ctx) {
+        if (LoadActors.empty()) {
+            TStringStream ss;
+            ss << "TLoad# " << Tag << " started on " << StartTs
+               << " with no subactors, finished# " << FinishedTests.size()
+               << ", in state# " << State;
+
+            auto response = std::make_unique<TEvDataShardLoad::TEvTestLoadInfoResponse>();
+            auto* info = response->Record.AddInfos();
+            info->SetTag(Tag);
+            info->SetData(ss.Str());
+            ctx.Send(ev->Sender, response.release());
+            return;
+        }
+
+        HttpInfoWaiters.insert(ev->Sender);
         if (HttpInfoCollector)
             return;
 
         TVector<TActorId> actors;
         actors.reserve(LoadActors.size());
-
-        // send messages to subactors
         for (const auto& actorId : LoadActors) {
             actors.push_back(actorId);
         }
-
-        // note that only parent can send us this request
-        Y_VERIFY(ev->Sender == Parent);
         HttpInfoCollector = ctx.Register(CreateInfoCollector(SelfId(), std::move(actors)));
     }
 
     void Handle(TEvDataShardLoad::TEvTestLoadInfoResponse::TPtr& ev, const TActorContext& ctx) {
-        HttpInfoCollector = {};
-
-        auto response = std::make_unique<TEvDataShardLoad::TEvTestLoadInfoResponse>();
-        auto& record = response->Record;
-
         TStringStream ss;
         ss << "TLoad# " << Tag << " started on " << StartTs
            << " with subactors active# " << LoadActors.size()
-           << ", finished# " << FinishedTests.size();
+           << ", finished# " << FinishedTests.size()
+           << ", subactors infos: ";
 
+        for (auto& info: ev->Get()->Record.GetInfos()) {
+            ss << "{ tag: " << info.GetTag() << ", data: " << info.GetData() << " }";
+        }
+
+        NKikimrDataShardLoad::TEvTestLoadInfoResponse record;
         auto* info = record.AddInfos();
         info->SetTag(Tag);
         info->SetData(ss.Str());
 
-        // bad that we copy, but to have it sorted by tag we had to prepend with our info
-        auto& collectedRecord = ev->Get()->Record;
-        for (auto& info: collectedRecord.GetInfos()) {
-            *record.AddInfos() = std::move(info);
+        for (const auto& actorId: HttpInfoWaiters) {
+            auto response = std::make_unique<TEvDataShardLoad::TEvTestLoadInfoResponse>();
+            response->Record = record;
+            ctx.Send(actorId, response.release());
         }
 
-        ctx.Send(Parent, response.release());
+        HttpInfoWaiters.clear();
+        HttpInfoCollector = {};
     }
 
     void HandlePoison(const TActorContext& ctx) {
@@ -686,14 +699,17 @@ public:
             return;
         }
 
+        if (LoadActors.empty()) {
+            auto response = std::make_unique<TEvDataShardLoad::TEvTestLoadInfoResponse>();
+            ctx.Send(ctx.SelfID, response.release());
+            return;
+        }
+
         TVector<TActorId> actors;
         actors.reserve(LoadActors.size());
-
-        // send messages to subactors
         for (const auto& kv : LoadActors) {
             actors.push_back(kv.second.ActorId);
         }
-
         HttpInfoCollector = ctx.Register(CreateInfoCollector(SelfId(), std::move(actors)));
     }
 
@@ -702,6 +718,15 @@ public:
 
         TStringStream str;
         HTML(str) {
+            DIV_CLASS("panel panel-info") {
+                DIV_CLASS("panel-heading") {
+                    str << "State";
+                }
+                DIV_CLASS("panel-body") {
+                    str << "Running# " << LoadActors.size() << ", Finished# " << FinishedTests.size();
+                }
+            }
+
             for (const auto& info: record.GetInfos()) {
                 DIV_CLASS("panel panel-info") {
                     DIV_CLASS("panel-heading") {