diff options
| author | Alexander Rutkovsky <[email protected]> | 2025-09-26 17:20:44 +0300 |
|---|---|---|
| committer | GitHub <[email protected]> | 2025-09-26 17:20:44 +0300 |
| commit | dddc9b67eb4e47ac62df3c80c950548cbe8de7b8 (patch) | |
| tree | 06eec01b3324c73647b2deb181be21e952a6eb27 | |
| parent | 3db6964e6ff0a5738b4b443bf93369ee20b6328a (diff) | |
Propagate StorageConfig up the distconf tree (#25867)
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<TEvNodeConfigPush> 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<TEvNodeConfigPush>(); 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<ui64> 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<TEvNodeConfigReversePush>(GetRootNodeId(), configToPeer); + auto response = std::make_unique<TEvNodeConfigReversePush>(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<TEvNodeConfigReversePush>(GetRootNodeId(), nullptr); + auto ev = std::make_unique<TEvNodeConfigReversePush>(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<TEvNodeConfigReversePush>(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<ui64> requestStorageConfigGeneration) { Record.SetRootNodeId(rootNodeId); if (committedConfig) { Record.MutableCommittedStorageConfig()->CopyFrom(*committedConfig); } + if (requestStorageConfigGeneration) { + Record.SetRequestStorageConfigGeneration(*requestStorageConfigGeneration); + } } static std::unique_ptr<TEvNodeConfigReversePush> 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. |
