NodeBroker: use deltas when returning recently added nodes (#9122)

Author: snaury

ydb/core/mind/dynamic_nameserver.cpp

@@ -286,6 +286,7 @@ void TDynamicNameserver::UpdateState(const NKikimrNodeBroker::TNodesInfo &rec,
         ctx.Schedule(config->Epoch.End - ctx.Now(),
                      new TEvPrivate::TEvUpdateEpoch(domain, config->Epoch.Id + 1));
     } else {
+        // Note: this update may be optimized to only include new nodes
         for (auto &node : rec.GetNodes()) {
             auto nodeId = node.GetNodeId();
             if (!config->DynamicNodes.contains(nodeId))

ydb/core/mind/node_broker.cpp

@@ -290,14 +290,45 @@ void TNodeBroker::AddDelayedListNodesRequest(ui64 epoch,
 
 void TNodeBroker::ProcessListNodesRequest(TEvNodeBroker::TEvListNodes::TPtr &ev)
 {
-    ui64 version = ev->Get()->Record.GetCachedVersion();
+    auto *msg = ev->Get();
 
     NKikimrNodeBroker::TNodesInfo info;
     Epoch.Serialize(*info.MutableEpoch());
     info.SetDomain(AppData()->DomainsInfo->GetDomain()->DomainUid);
     TAutoPtr<TEvNodeBroker::TEvNodesInfo> resp = new TEvNodeBroker::TEvNodesInfo(info);
-    if (version != Epoch.Version)
+
+    bool optimized = false;
+
+    if (msg->Record.HasCachedVersion()) {
+        if (msg->Record.GetCachedVersion() == Epoch.Version) {
+            // Client has an up-to-date list already
+            optimized = true;
+        } else {
+            // We may be able to only send added nodes in the same epoch when
+            // all deltas are cached up to the current epoch inclusive.
+            ui64 neededFirstVersion = msg->Record.GetCachedVersion() + 1;
+            if (!EpochDeltasVersions.empty() &&
+                EpochDeltasVersions.front() <= neededFirstVersion &&
+                EpochDeltasVersions.back() == Epoch.Version &&
+                neededFirstVersion <= Epoch.Version)
+            {
+                ui64 firstIndex = neededFirstVersion - EpochDeltasVersions.front();
+                if (firstIndex > 0) {
+                    // Note: usually there is a small number of nodes added
+                    // between subsequent requests, so this substr should be
+                    // very cheap.
+                    resp->PreSerializedData = EpochDeltasCache.substr(EpochDeltasEndOffsets[firstIndex - 1]);
+                } else {
+                    resp->PreSerializedData = EpochDeltasCache;
+                }
+                optimized = true;
+            }
+        }
+    }
+
+    if (!optimized) {
         resp->PreSerializedData = EpochCache;
+    }
 
     TabletCounters->Percentile()[COUNTER_LIST_NODES_BYTES].IncrementFor(resp->GetCachedByteSize());
     LOG_TRACE_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
@@ -308,12 +339,16 @@ void TNodeBroker::ProcessListNodesRequest(TEvNodeBroker::TEvListNodes::TPtr &ev)
 
 void TNodeBroker::ProcessDelayedListNodesRequests()
 {
+    THashSet<TActorId> processed;
     while (!DelayedListNodesRequests.empty()) {
         auto it = DelayedListNodesRequests.begin();
         if (it->first > Epoch.Id)
             break;
 
-        ProcessListNodesRequest(it->second);
+        // Avoid processing more than one request from the same sender
+        if (processed.insert(it->second->Sender).second) {
+            ProcessListNodesRequest(it->second);
+        }
         DelayedListNodesRequests.erase(it);
     }
 }
@@ -432,6 +467,11 @@ void TNodeBroker::PrepareEpochCache()
 
     Y_PROTOBUF_SUPPRESS_NODISCARD info.SerializeToString(&EpochCache);
     TabletCounters->Simple()[COUNTER_EPOCH_SIZE_BYTES].Set(EpochCache.Size());
+
+    EpochDeltasCache.clear();
+    EpochDeltasVersions.clear();
+    EpochDeltasEndOffsets.clear();
+    TabletCounters->Simple()[COUNTER_EPOCH_DELTAS_SIZE_BYTES].Set(EpochDeltasCache.size());
 }
 
 void TNodeBroker::AddNodeToEpochCache(const TNodeInfo &node)
@@ -447,6 +487,17 @@ void TNodeBroker::AddNodeToEpochCache(const TNodeInfo &node)
 
     EpochCache += delta;
     TabletCounters->Simple()[COUNTER_EPOCH_SIZE_BYTES].Set(EpochCache.Size());
+
+    if (!EpochDeltasVersions.empty() && EpochDeltasVersions.back() + 1 != Epoch.Version) {
+        EpochDeltasCache.clear();
+        EpochDeltasVersions.clear();
+        EpochDeltasEndOffsets.clear();
+    }
+
+    EpochDeltasCache += delta;
+    EpochDeltasVersions.push_back(Epoch.Version);
+    EpochDeltasEndOffsets.push_back(EpochDeltasCache.size());
+    TabletCounters->Simple()[COUNTER_EPOCH_DELTAS_SIZE_BYTES].Set(EpochDeltasCache.size());
 }
 
 void TNodeBroker::SubscribeForConfigUpdates(const TActorContext &ctx)

ydb/core/mind/node_broker_impl.h

@@ -341,6 +341,10 @@ class TNodeBroker : public TActor<TNodeBroker>
     TSchedulerCookieHolder EpochTimerCookieHolder;
     TString EpochCache;
 
+    TString EpochDeltasCache;
+    TVector<ui64> EpochDeltasVersions;
+    TVector<ui64> EpochDeltasEndOffsets;
+
     TTabletCountersBase* TabletCounters;
     TAutoPtr<TTabletCountersBase> TabletCountersPtr;
 

ydb/core/mind/node_broker_ut.cpp

@@ -858,6 +858,48 @@ Y_UNIT_TEST_SUITE(TNodeBrokerTest) {
         UNIT_ASSERT_VALUES_EQUAL(epoch1.GetId(), epoch.GetId() + 5);
     }
 
+    Y_UNIT_TEST(TestListNodesEpochDeltas)
+    {
+        TTestBasicRuntime runtime(8, false);
+        Setup(runtime, 10);
+        TActorId sender = runtime.AllocateEdgeActor();
+
+        WaitForEpochUpdate(runtime, sender);
+        WaitForEpochUpdate(runtime, sender);
+
+        auto epoch0 = GetEpoch(runtime, sender);
+        CheckRegistration(runtime, sender, "host1", 1001, "host1.yandex.net", "1.2.3.4",
+                          1, 2, 3, 4, TStatus::OK, NODE1, epoch0.GetNextEnd());
+        auto epoch1 = CheckFilteredNodesList(runtime, sender, {NODE1}, {}, 0, epoch0.GetVersion());
+        CheckRegistration(runtime, sender, "host2", 1001, "host2.yandex.net", "1.2.3.5",
+                          1, 2, 3, 5, TStatus::OK, NODE2, epoch1.GetNextEnd());
+        auto epoch2 = CheckFilteredNodesList(runtime, sender, {NODE2}, {}, 0, epoch1.GetVersion());
+        CheckRegistration(runtime, sender, "host3", 1001, "host3.yandex.net", "1.2.3.6",
+                          1, 2, 3, 6, TStatus::OK, NODE3, epoch2.GetNextEnd());
+        auto epoch3 = CheckFilteredNodesList(runtime, sender, {NODE3}, {}, 0, epoch2.GetVersion());
+
+        CheckFilteredNodesList(runtime, sender, {NODE1, NODE2, NODE3}, {}, 0, epoch0.GetVersion());
+        CheckFilteredNodesList(runtime, sender, {NODE2, NODE3}, {}, 0, epoch1.GetVersion());
+        CheckFilteredNodesList(runtime, sender, {}, {}, 0, epoch3.GetVersion());
+
+        RebootTablet(runtime, MakeNodeBrokerID(), sender);
+        CheckFilteredNodesList(runtime, sender, {}, {}, 0, epoch3.GetVersion());
+
+        CheckRegistration(runtime, sender, "host4", 1001, "host4.yandex.net", "1.2.3.7",
+                          1, 2, 3, 7, TStatus::OK, NODE4, epoch3.GetNextEnd());
+        auto epoch4 = CheckFilteredNodesList(runtime, sender, {NODE4}, {}, 0, epoch3.GetVersion());
+
+        // NodeBroker doesn't have enough history in memory and replies with the full node list
+        CheckFilteredNodesList(runtime, sender, {NODE1, NODE2, NODE3, NODE4}, {}, 0, epoch2.GetVersion());
+
+        WaitForEpochUpdate(runtime, sender);
+        auto epoch5 = GetEpoch(runtime, sender);
+        CheckFilteredNodesList(runtime, sender, {}, {}, 0, epoch5.GetVersion());
+
+        // New epoch may remove nodes, so deltas are not returned on epoch change
+        CheckFilteredNodesList(runtime, sender, {NODE1, NODE2, NODE3, NODE4}, {}, 0, epoch3.GetVersion());
+    }
+
     Y_UNIT_TEST(TestRandomActions)
     {
         TTestBasicRuntime runtime(8, false);

ydb/core/protos/counters_node_broker.proto

@@ -8,6 +8,7 @@ option (NKikimr.TabletTypeName) = "NodeBroker"; // Used as prefix for all counte
 
 enum ESimpleCounters {
     COUNTER_EPOCH_SIZE_BYTES = 0           [(CounterOpts) = {Name: "EpochSizeBytes"}];
+    COUNTER_EPOCH_DELTAS_SIZE_BYTES = 1    [(CounterOpts) = {Name: "EpochDeltasSizeBytes"}];
 }
 
 enum ECumulativeCounters {