summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorAlexander Rutkovsky <[email protected]>2025-09-26 17:20:44 +0300
committerGitHub <[email protected]>2025-09-26 17:20:44 +0300
commitdddc9b67eb4e47ac62df3c80c950548cbe8de7b8 (patch)
tree06eec01b3324c73647b2deb181be21e952a6eb27
parent3db6964e6ff0a5738b4b443bf93369ee20b6328a (diff)
Propagate StorageConfig up the distconf tree (#25867)
-rw-r--r--ydb/core/blobstorage/nodewarden/distconf_binding.cpp35
-rw-r--r--ydb/core/blobstorage/nodewarden/distconf_cache.cpp2
-rw-r--r--ydb/core/blobstorage/nodewarden/distconf_fsm.cpp2
-rw-r--r--ydb/core/blobstorage/nodewarden/node_warden_events.h6
-rw-r--r--ydb/core/protos/blobstorage_distributed_config.proto2
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.