Add slot names generation to node broker (#2808)

Author: pixcc

ydb/core/mind/node_broker.cpp

@@ -64,6 +64,8 @@ void TNodeBroker::OnActivateExecutor(const TActorContext &ctx)
     MinDynamicId = Max(MaxStaticId + 1, (ui64)Min(appData->DynamicNameserviceConfig->MinDynamicNodeId, TActorId::MaxNodeId));
     MaxDynamicId = Max(MinDynamicId, (ui64)Min(appData->DynamicNameserviceConfig->MaxDynamicNodeId, TActorId::MaxNodeId));
 
+    EnableSlotNameGeneration = appData->FeatureFlags.GetEnableSlotNameGeneration();
+
     ClearState();
 
     ProcessTx(CreateTxInitScheme(), ctx);
@@ -125,11 +127,38 @@ bool TNodeBroker::OnRenderAppHtmlPage(NMon::TEvRemoteHttpInfo::TPtr ev,
                     << "   Location: " << node.Location.ToString() << Endl
                     << "   Lease: " << node.Lease << Endl
                     << "   Expire: " << node.ExpirationString() << Endl
-                    << "   AuthorizedByCertificate: " << (node.AuthorizedByCertificate ? "true" : "false") << Endl;
+                    << "   AuthorizedByCertificate: " << (node.AuthorizedByCertificate ? "true" : "false") << Endl
+                    << "   ServicedSubDomain: " << node.ServicedSubDomain << Endl
+                    << "   SlotIndex: " << node.SlotIndex << Endl;
             }
             str << Endl;
 
             str << "Free Node IDs count: " << FreeIds.Count() << Endl;
+
+            str << Endl;
+            str << "Slot Indexes Pools usage: " << Endl;
+            size_t totalSize = 0;
+            size_t totalCapacity = 0;
+            for (const auto &[subdomainKey, slotIndexesPool] : SlotIndexesPools) {
+                const size_t size = slotIndexesPool.Size();
+                totalSize += size;
+                const size_t capacity = slotIndexesPool.Capacity();
+                totalCapacity += capacity;
+                const double usagePercent = floor(size * 100.0 / capacity);
+                str << "   " << subdomainKey
+                    << " = " << usagePercent << "% (" << size << " of " << capacity << ")"
+                    << Endl;
+            }
+            str << Endl;
+
+            if (totalCapacity > 0) {
+                const double totalUsagePercent = floor(totalSize * 100.0 / totalCapacity);
+                str << "   Total"
+                    << " = " << totalUsagePercent << "% (" << totalSize << " of " << totalCapacity << ")"
+                    << Endl;
+            } else {
+                str << "   No Slot Indexes Pools" << Endl;
+            }
         }
     }
 
@@ -159,11 +188,15 @@ void TNodeBroker::ClearState()
     Hosts.clear();
 
     RecomputeFreeIds();
+    RecomputeSlotIndexesPools();
 }
 
 void TNodeBroker::AddNode(const TNodeInfo &info)
 {
     FreeIds.Reset(info.NodeId);
+    if (info.SlotIndex.has_value()) {
+        SlotIndexesPools[info.ServicedSubDomain].Acquire(info.SlotIndex.value());
+    }    
 
     if (info.Expire > Epoch.Start) {
         LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
@@ -215,6 +248,24 @@ void TNodeBroker::RecomputeFreeIds()
     }
 }
 
+void TNodeBroker::RecomputeSlotIndexesPools()
+{
+    for (auto &[_, slotIndexesPool] : SlotIndexesPools) {
+        slotIndexesPool.ReleaseAll();
+    }
+
+    for (const auto &[_, node] : Nodes) {
+        if (node.SlotIndex.has_value()) {
+            SlotIndexesPools[node.ServicedSubDomain].Acquire(node.SlotIndex.value());
+        }
+    }
+    for (const auto &[_, node] : ExpiredNodes) {
+        if (node.SlotIndex.has_value()) {
+            SlotIndexesPools[node.ServicedSubDomain].Acquire(node.SlotIndex.value());
+        }
+    }
+}
+
 bool TNodeBroker::IsBannedId(ui32 id) const
 {
     for (auto &pr : BannedIds)
@@ -288,6 +339,10 @@ void TNodeBroker::FillNodeInfo(const TNodeInfo &node,
     info.SetAddress(node.Address);
     info.SetExpire(node.Expire.GetValue());
     node.Location.Serialize(info.MutableLocation(), false);
+    if (EnableSlotNameGeneration && node.SlotIndex.has_value()) {
+        const TString slotName = TStringBuilder() << "slot-" << node.SlotIndex;
+        info.SetSlotName(slotName);
+    }
 }
 
 void TNodeBroker::ComputeNextEpochDiff(TStateDiff &diff)
@@ -328,9 +383,13 @@ void TNodeBroker::ApplyStateDiff(const TStateDiff &diff)
         LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
                     "Remove node " << it->second.IdString());
 
-        ExpiredNodes.erase(it);
-        if (!IsBannedId(id) && id >= MinDynamicId && id <= MaxDynamicId)
+        if (!IsBannedId(id) && id >= MinDynamicId && id <= MaxDynamicId) {
             FreeIds.Set(id);
+        }
+        if (it->second.SlotIndex.has_value()) {
+            SlotIndexesPools[it->second.ServicedSubDomain].Release(it->second.SlotIndex.value());
+        }
+        ExpiredNodes.erase(it);
     }
 
     LOG_DEBUG_S(TActorContext::AsActorContext(), NKikimrServices::NODE_BROKER,
@@ -437,7 +496,9 @@ void TNodeBroker::DbAddNode(const TNodeInfo &node,
                 << " dc=" << node.Location.GetDataCenterId()
                 << " location=" << node.Location.ToString()
                 << " lease=" << node.Lease
-                << " expire=" << node.ExpirationString());
+                << " expire=" << node.ExpirationString()
+                << " servicedsubdomain=" << node.ServicedSubDomain
+                << " slotindex= " << node.SlotIndex);
 
     NIceDb::TNiceDb db(txc.DB);
     using T = Schema::Nodes;
@@ -448,7 +509,16 @@ void TNodeBroker::DbAddNode(const TNodeInfo &node,
         .Update<T::Address>(node.Address)
         .Update<T::Lease>(node.Lease)
         .Update<T::Expire>(node.Expire.GetValue())
-        .Update<T::Location>(node.Location.GetSerializedLocation());
+        .Update<T::Location>(node.Location.GetSerializedLocation())
+        .Update<T::ServicedSubDomain>(node.ServicedSubDomain);
+
+    if (node.SlotIndex.has_value()) {
+        db.Table<T>().Key(node.NodeId)
+            .Update<T::SlotIndex>(node.SlotIndex.value());
+    } else {
+        db.Table<T>().Key(node.NodeId)
+            .UpdateToNull<T::SlotIndex>();
+    }
 }
 
 void TNodeBroker::DbApplyStateDiff(const TStateDiff &diff,
@@ -590,7 +660,10 @@ bool TNodeBroker::DbLoadState(TTransactionContext &txc,
 
             info.Lease = nodesRowset.GetValue<T::Lease>();
             info.Expire = expire;
-
+            info.ServicedSubDomain = TSubDomainKey(nodesRowset.GetValueOrDefault<T::ServicedSubDomain>());
+            if (nodesRowset.HaveValue<T::SlotIndex>()) {
+                info.SlotIndex = nodesRowset.GetValue<T::SlotIndex>();
+            } 
             AddNode(info);
 
             LOG_DEBUG_S(ctx, NKikimrServices::NODE_BROKER,
@@ -776,6 +849,7 @@ void TNodeBroker::Handle(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
         TEvNodeBroker::TEvRegistrationRequest::TPtr Ev;
         TNodeBroker *Self;
         NActors::TScopeId ScopeId;
+        TSubDomainKey ServicedSubDomain;
 
     public:
         static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -821,8 +895,9 @@ void TNodeBroker::Handle(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
                 } else {
                     ScopeId = {response.DomainInfo->DomainKey.OwnerId, response.DomainInfo->DomainKey.LocalPathId};
                 }
+                ServicedSubDomain = TSubDomainKey(response.DomainInfo->DomainKey.OwnerId, response.DomainInfo->DomainKey.LocalPathId);
             } else {
-                LOG_WARN_S(ctx, NKikimrServices::NODE_BROKER, "Cannot resolve scope id"
+                LOG_WARN_S(ctx, NKikimrServices::NODE_BROKER, "Cannot resolve tenant"
                     << ": request# " << Ev->Get()->Record.ShortDebugString()
                     << ", response# " << response.ToString(*AppData()->TypeRegistry));
             }
@@ -835,11 +910,12 @@ void TNodeBroker::Handle(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
         }
 
         void Finish(const TActorContext& ctx) {
-            LOG_TRACE_S(ctx, NKikimrServices::NODE_BROKER, "Finished resolving scope id"
+            LOG_TRACE_S(ctx, NKikimrServices::NODE_BROKER, "Finished resolving tenant"
                 << ": request# " << Ev->Get()->Record.ShortDebugString()
-                << ": scope id# " << ScopeIdToString(ScopeId));
+                << ": scope id# " << ScopeIdToString(ScopeId)
+                << ": serviced subdomain# " << ServicedSubDomain);
 
-            Self->ProcessTx(Self->CreateTxRegisterNode(Ev, ScopeId), ctx);
+            Self->ProcessTx(Self->CreateTxRegisterNode(Ev, ScopeId, ServicedSubDomain), ctx);
             Die(ctx);
         }
 

ydb/core/mind/node_broker__register_node.cpp

@@ -10,10 +10,12 @@ using namespace NKikimrNodeBroker;
 
 class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
 public:
-    TTxRegisterNode(TNodeBroker *self, TEvNodeBroker::TEvRegistrationRequest::TPtr &ev, const NActors::TScopeId& scopeId)
+    TTxRegisterNode(TNodeBroker *self, TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
+                    const NActors::TScopeId& scopeId, const TSubDomainKey& servicedSubDomain)
         : TBase(self)
         , Event(ev)
         , ScopeId(scopeId)
+        , ServicedSubDomain(servicedSubDomain)
         , NodeId(0)
         , ExtendLease(false)
         , FixNodeId(false)
@@ -60,6 +62,12 @@ class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
                          ctx);
         }
 
+        if (Self->EnableSlotNameGeneration && rec.HasPath() && ServicedSubDomain == InvalidSubDomainKey) {
+            return Error(TStatus::ERROR,
+                         TStringBuilder() << "Cannot resolve subdomain key for path " << rec.GetPath(),
+                         ctx);
+        }
+
         // Already registered?
         auto it = Self->Hosts.find(std::make_tuple(host, addr, port));
         if (it != Self->Hosts.end()) {
@@ -91,7 +99,21 @@ class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
                 ExtendLease = true;
             }
             node.AuthorizedByCertificate = rec.GetAuthorizedByCertificate();
-
+            
+            if (Self->EnableSlotNameGeneration) {
+                if (ServicedSubDomain != node.ServicedSubDomain) {
+                    if (node.SlotIndex.has_value()) {
+                        Self->SlotIndexesPools[node.ServicedSubDomain].Release(node.SlotIndex.value());
+                    }
+                    node.ServicedSubDomain = ServicedSubDomain;
+                    node.SlotIndex = Self->SlotIndexesPools[node.ServicedSubDomain].AcquireLowestFreeIndex();
+                    Self->DbAddNode(node, txc);
+                } else if (!node.SlotIndex.has_value()) {    
+                    node.SlotIndex = Self->SlotIndexesPools[node.ServicedSubDomain].AcquireLowestFreeIndex();
+                    Self->DbAddNode(node, txc);
+                }
+            }
+            
             Response->Record.MutableStatus()->SetCode(TStatus::OK);
             Self->FillNodeInfo(node, *Response->Record.MutableNode());
 
@@ -109,6 +131,11 @@ class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
         Node->Lease = 1;
         Node->Expire = expire;
 
+        if (Self->EnableSlotNameGeneration) {
+            Node->ServicedSubDomain = ServicedSubDomain;
+            Node->SlotIndex = Self->SlotIndexesPools[Node->ServicedSubDomain].AcquireLowestFreeIndex();
+        }
+
         Response->Record.MutableStatus()->SetCode(TStatus::OK);
 
         Self->DbAddNode(*Node, txc);
@@ -151,16 +178,19 @@ class TNodeBroker::TTxRegisterNode : public TTransactionBase<TNodeBroker> {
 private:
     TEvNodeBroker::TEvRegistrationRequest::TPtr Event;
     const NActors::TScopeId ScopeId;
+    const TSubDomainKey ServicedSubDomain;
     TAutoPtr<TEvNodeBroker::TEvRegistrationResponse> Response;
     THolder<TNodeInfo> Node;
     ui32 NodeId;
     bool ExtendLease;
     bool FixNodeId;
 };
 
-ITransaction *TNodeBroker::CreateTxRegisterNode(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev, const NActors::TScopeId& scopeId)
+ITransaction *TNodeBroker::CreateTxRegisterNode(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
+                                                const NActors::TScopeId& scopeId,
+                                                const TSubDomainKey& servicedSubDomain)
 {
-    return new TTxRegisterNode(this, ev, scopeId);
+    return new TTxRegisterNode(this, ev, scopeId, servicedSubDomain);
 }
 
 } // NNodeBroker

ydb/core/mind/node_broker__scheme.h

@@ -2,6 +2,7 @@
 
 #include "defs.h"
 
+#include <ydb/core/base/subdomain.h>
 #include <ydb/core/scheme/scheme_types_defs.h>
 #include <ydb/core/tablet_flat/flat_cxx_database.h>
 
@@ -22,9 +23,22 @@ struct Schema : NIceDb::Schema {
         struct Lease : Column<10, NScheme::NTypeIds::Uint32> {};
         struct Expire : Column<11, NScheme::NTypeIds::Uint64> {};
         struct Location : Column<12, NScheme::NTypeIds::String> {};
+        struct ServicedSubDomain : Column<13, NScheme::NTypeIds::String> { using Type = NKikimrSubDomains::TDomainKey; };
+        struct SlotIndex : Column<14, NScheme::NTypeIds::Uint32> {};
 
         using TKey = TableKey<ID>;
-        using TColumns = TableColumns<ID, Host, Port, ResolveHost, Address, Lease, Expire, Location>;
+        using TColumns = TableColumns<
+            ID,
+            Host,
+            Port,
+            ResolveHost,
+            Address,
+            Lease,
+            Expire,
+            Location,
+            ServicedSubDomain,
+            SlotIndex
+        >;
     };
 
     struct Config : Table<2> {

ydb/core/mind/node_broker_impl.h

@@ -1,8 +1,10 @@
 #pragma once
 
 #include "node_broker.h"
+#include "slot_indexes_pool.h"
 
 #include <ydb/core/base/tablet_pipe.h>
+#include <ydb/core/base/subdomain.h>
 #include <ydb/core/cms/console/console.h>
 #include <ydb/core/cms/console/configs_dispatcher.h>
 #include <ydb/core/cms/console/tx_processor.h>
@@ -114,6 +116,8 @@ class TNodeBroker : public TActor<TNodeBroker>
         ui32 Lease;
         TInstant Expire;
         bool AuthorizedByCertificate = false;
+        std::optional<ui32> SlotIndex;
+        TSubDomainKey ServicedSubDomain;
     };
 
     // State changes to apply while moving to the next epoch.
@@ -134,7 +138,9 @@ class TNodeBroker : public TActor<TNodeBroker>
     ITransaction *CreateTxExtendLease(TEvNodeBroker::TEvExtendLeaseRequest::TPtr &ev);
     ITransaction *CreateTxInitScheme();
     ITransaction *CreateTxLoadState();
-    ITransaction *CreateTxRegisterNode(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev, const NActors::TScopeId& scopeId);
+    ITransaction *CreateTxRegisterNode(TEvNodeBroker::TEvRegistrationRequest::TPtr &ev,
+                                       const NActors::TScopeId& scopeId,
+                                       const TSubDomainKey& servicedSubDomain);
     ITransaction *CreateTxUpdateConfig(TEvConsole::TEvConfigNotificationRequest::TPtr &ev);
     ITransaction *CreateTxUpdateConfig(TEvNodeBroker::TEvSetConfigRequest::TPtr &ev);
     ITransaction *CreateTxUpdateConfigSubscription(TEvConsole::TEvReplaceConfigSubscriptionsResponse::TPtr &ev);
@@ -208,6 +214,7 @@ class TNodeBroker : public TActor<TNodeBroker>
     void ExtendLease(TNodeInfo &node);
     void FixNodeId(TNodeInfo &node);
     void RecomputeFreeIds();
+    void RecomputeSlotIndexesPools();
     bool IsBannedId(ui32 id) const;
 
     void AddDelayedListNodesRequest(ui64 epoch,
@@ -292,6 +299,9 @@ class TNodeBroker : public TActor<TNodeBroker>
     THashMap<std::tuple<TString, TString, ui16>, ui32> Hosts;
     // Bitmap with free Node IDs (with no lower 5 bits).
     TDynBitMap FreeIds;
+    // Maps tenant to its slot indexes pool.
+    std::unordered_map<TSubDomainKey, TSlotIndexesPool, THash<TSubDomainKey>> SlotIndexesPools;
+    bool EnableSlotNameGeneration = false;
     // Epoch info.
     TEpochInfo Epoch;
     // Current config.