From dddc9b67eb4e47ac62df3c80c950548cbe8de7b8 Mon Sep 17 00:00:00 2001 From: Alexander Rutkovsky Date: Fri, 26 Sep 2025 17:20:44 +0300 Subject: Propagate StorageConfig up the distconf tree (#25867) --- .../blobstorage/nodewarden/distconf_binding.cpp | 35 ++++++++++++++++++---- ydb/core/blobstorage/nodewarden/distconf_cache.cpp | 2 +- ydb/core/blobstorage/nodewarden/distconf_fsm.cpp | 2 +- .../blobstorage/nodewarden/node_warden_events.h | 6 +++- .../protos/blobstorage_distributed_config.proto | 2 ++ 5 files changed, 38 insertions(+), 9 deletions(-) diff --git a/ydb/core/blobstorage/nodewarden/distconf_binding.cpp b/ydb/core/blobstorage/nodewarden/distconf_binding.cpp index 5905375ca6f..02d6a87defa 100644 --- a/ydb/core/blobstorage/nodewarden/distconf_binding.cpp +++ b/ydb/core/blobstorage/nodewarden/distconf_binding.cpp @@ -432,22 +432,37 @@ namespace NKikimr::NStorage { FanOutReversePush(); } + std::unique_ptr pendingPush; + auto getPendingPushRecord = [&] { + if (!pendingPush) { + pendingPush.reset(new TEvNodeConfigPush); + } + return &pendingPush->Record; + }; + // process cache updates, if needed if (record.HasCacheUpdate()) { auto *cacheUpdate = record.MutableCacheUpdate(); ApplyCacheUpdates(cacheUpdate, senderNodeId); if (cacheUpdate->RequestedKeysSize()) { - auto ev = std::make_unique(); for (const TString& key : cacheUpdate->GetRequestedKeys()) { if (const auto it = Cache.find(key); it != Cache.end()) { - AddCacheUpdate(ev->Record.MutableCacheUpdate(), it, true); + AddCacheUpdate(getPendingPushRecord()->MutableCacheUpdate(), it, true); } } - SendEvent(*Binding, std::move(ev)); } } + if (StorageConfig && record.HasRequestStorageConfigGeneration() && + record.GetRequestStorageConfigGeneration() <= StorageConfig->GetGeneration()) { + getPendingPushRecord()->MutableStorageConfig()->CopyFrom(*StorageConfig); + } + + if (pendingPush) { + SendEvent(*Binding, std::move(pendingPush)); + } + // if we have a root node id that is not ours, then we drop all inbound pile connections; also execute any // pending commands if (Binding && Binding->RootNodeId) { @@ -537,11 +552,14 @@ namespace NKikimr::NStorage { // check if we have to send our current config to the peer const NKikimrBlobStorage::TStorageConfig *configToPeer = nullptr; - if (record.GetInitial()) { + std::optional requestStorageConfigGeneration; + if (StorageConfig) { for (const auto& item : record.GetBoundNodes()) { if (item.GetNodeId().GetNodeId() == senderNodeId) { - if (StorageConfig && item.GetMeta().GetGeneration() < StorageConfig->GetGeneration()) { + if (item.GetMeta().GetGeneration() < StorageConfig->GetGeneration()) { configToPeer = StorageConfig.get(); + } else if (StorageConfig->GetGeneration() < item.GetMeta().GetGeneration()) { + requestStorageConfigGeneration.emplace(item.GetMeta().GetGeneration()); } break; } @@ -606,7 +624,8 @@ namespace NKikimr::NStorage { GetRootNodeId()); TBoundNode& info = it->second; if (inserted) { - auto response = std::make_unique(GetRootNodeId(), configToPeer); + auto response = std::make_unique(GetRootNodeId(), configToPeer, + requestStorageConfigGeneration); if (record.GetInitial()) { auto *cache = record.MutableCacheUpdate(); @@ -675,6 +694,10 @@ namespace NKikimr::NStorage { ApplyCacheUpdates(record.MutableCacheUpdate(), senderNodeId); } + if (record.HasStorageConfig()) { + ApplyStorageConfig(record.GetStorageConfig(), /*fromBinding=*/ false); + } + if (pushEv && pushEv->IsUseful()) { SendEvent(*Binding, std::move(pushEv)); } diff --git a/ydb/core/blobstorage/nodewarden/distconf_cache.cpp b/ydb/core/blobstorage/nodewarden/distconf_cache.cpp index b8fbe421b99..8dfd8649845 100644 --- a/ydb/core/blobstorage/nodewarden/distconf_cache.cpp +++ b/ydb/core/blobstorage/nodewarden/distconf_cache.cpp @@ -48,7 +48,7 @@ namespace NKikimr::NStorage { } // propagate backwards for (auto& [nodeId, info] : DirectBoundNodes) { - auto ev = std::make_unique(GetRootNodeId(), nullptr); + auto ev = std::make_unique(GetRootNodeId(), nullptr, std::nullopt); info.LastReportedRootNodeId = GetRootNodeId(); ev->Record.MutableCacheUpdate()->CopyFrom(updates); SendEvent(nodeId, info, std::move(ev)); diff --git a/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp b/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp index 44715a1f63e..a8967de2e5a 100644 --- a/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp +++ b/ydb/core/blobstorage/nodewarden/distconf_fsm.cpp @@ -715,7 +715,7 @@ namespace NKikimr::NStorage { const bool needToUpdateConfig = nodeIdsToUpdate.contains(nodeId); if (needToUpdateConfig || info.LastReportedRootNodeId != rootNodeId) { SendEvent(nodeId, info, std::make_unique(rootNodeId, - needToUpdateConfig ? StorageConfig.get() : nullptr)); + needToUpdateConfig ? StorageConfig.get() : nullptr, std::nullopt)); info.LastReportedRootNodeId = GetRootNodeId(); } } diff --git a/ydb/core/blobstorage/nodewarden/node_warden_events.h b/ydb/core/blobstorage/nodewarden/node_warden_events.h index 59ef0ea63c8..007c8b0e4f3 100644 --- a/ydb/core/blobstorage/nodewarden/node_warden_events.h +++ b/ydb/core/blobstorage/nodewarden/node_warden_events.h @@ -22,11 +22,15 @@ namespace NKikimr::NStorage { { TEvNodeConfigReversePush() = default; - TEvNodeConfigReversePush(ui32 rootNodeId, const NKikimrBlobStorage::TStorageConfig *committedConfig) { + TEvNodeConfigReversePush(ui32 rootNodeId, const NKikimrBlobStorage::TStorageConfig *committedConfig, + std::optional requestStorageConfigGeneration) { Record.SetRootNodeId(rootNodeId); if (committedConfig) { Record.MutableCommittedStorageConfig()->CopyFrom(*committedConfig); } + if (requestStorageConfigGeneration) { + Record.SetRequestStorageConfigGeneration(*requestStorageConfigGeneration); + } } static std::unique_ptr MakeRejected(const NKikimrBlobStorage::TStorageConfig *config) { diff --git a/ydb/core/protos/blobstorage_distributed_config.proto b/ydb/core/protos/blobstorage_distributed_config.proto index 4afbe634a69..5748a2c916a 100644 --- a/ydb/core/protos/blobstorage_distributed_config.proto +++ b/ydb/core/protos/blobstorage_distributed_config.proto @@ -78,6 +78,7 @@ message TEvNodeConfigPush { repeated TBoundNode BoundNodes = 2; // a list of bound node updates (including itself) repeated TNodeIdentifier DeletedBoundNodeIds = 3; // a list of detached nodes TCacheUpdate CacheUpdate = 4; + TStorageConfig StorageConfig = 5; } // Used to reverse-propagate configuration and to confirm/reject initial TEvNodePushBinding query. @@ -87,6 +88,7 @@ message TEvNodeConfigReversePush { TStorageConfig CommittedStorageConfig = 3; // last known committed storage configuration bool RecurseConfigUpdate = 4; TCacheUpdate CacheUpdate = 5; + optional uint64 RequestStorageConfigGeneration = 6; } // Remove node from bound list. -- cgit v1.3