Persist config upon being changed by distconf (#14809)

Author: alexvru

ydb/core/blobstorage/nodewarden/distconf.cpp

@@ -74,25 +74,15 @@ namespace NKikimr::NStorage {
             MainConfigFetchYamlHash = 0;
 
             if (config.HasConfigComposite()) {
-                try {
-                    // parse the composite stream
-                    TStringInput ss(config.GetConfigComposite());
-                    TZstdDecompress zstd(&ss);
-                    MainConfigYaml = TString::Uninitialized(LoadSize(&zstd));
-                    zstd.LoadOrFail(MainConfigYaml.Detach(), MainConfigYaml.size());
-                    MainConfigFetchYaml = TString::Uninitialized(LoadSize(&zstd));
-                    zstd.LoadOrFail(MainConfigFetchYaml.Detach(), MainConfigFetchYaml.size());
-
-                    // extract _current_ config version
-                    auto metadata = NYamlConfig::GetMainMetadata(MainConfigYaml);
-                    Y_DEBUG_ABORT_UNLESS(metadata.Version.has_value());
-                    MainConfigYamlVersion = metadata.Version.value_or(0);
-
-                    // and _fetched_ config hash
-                    MainConfigFetchYamlHash = NYaml::GetConfigHash(MainConfigFetchYaml);
-                } catch (const std::exception& ex) {
-                    Y_ABORT("ConfigComposite format incorrect: %s", ex.what());
+                // parse the composite stream
+                auto error = DecomposeConfig(config.GetConfigComposite(), &MainConfigYaml,
+                    &MainConfigYamlVersion.emplace(), &MainConfigFetchYaml);
+                if (error) {
+                    Y_ABORT("ConfigComposite format incorrect: %s", error->data());
                 }
+
+                // and _fetched_ config hash
+                MainConfigFetchYamlHash = NYaml::GetConfigHash(MainConfigFetchYaml);
             }
 
             // now extract the additional storage section
@@ -380,6 +370,34 @@ namespace NKikimr::NStorage {
         TActivationContext::Send(ev.Release());
     }
 
+
+    std::optional<TString> DecomposeConfig(const TString& configComposite, TString *mainConfigYaml,
+            ui64 *mainConfigVersion, TString *mainConfigFetchYaml) {
+        try {
+            TStringInput ss(configComposite);
+            TZstdDecompress zstd(&ss);
+
+            TString yaml = TString::Uninitialized(LoadSize(&zstd));
+            zstd.LoadOrFail(yaml.Detach(), yaml.size());
+            if (mainConfigVersion) {
+                auto metadata = NYamlConfig::GetMainMetadata(yaml);
+                Y_DEBUG_ABORT_UNLESS(metadata.Version.has_value());
+                *mainConfigVersion = metadata.Version.value_or(0);
+            }
+            if (mainConfigYaml) {
+                *mainConfigYaml = std::move(yaml);
+            }
+
+            if (mainConfigFetchYaml) {
+                *mainConfigFetchYaml = TString::Uninitialized(LoadSize(&zstd));
+                zstd.LoadOrFail(mainConfigFetchYaml->Detach(), mainConfigFetchYaml->size());
+            }
+        } catch (const std::exception& ex) {
+            return ex.what();
+        }
+        return std::nullopt;
+    }
+
 } // NKikimr::NStorage
 
 template<>

ydb/core/blobstorage/nodewarden/distconf.h

@@ -738,6 +738,9 @@ namespace NKikimr::NStorage {
 
     std::optional<TString> ValidateConfig(const NKikimrBlobStorage::TStorageConfig& config);
 
+    std::optional<TString> DecomposeConfig(const TString& configComposite, TString *mainConfigYaml,
+        ui64 *mainConfigVersion, TString *mainConfigFetchYaml);
+
 } // NKikimr::NStorage
 
 template<>

ydb/core/blobstorage/nodewarden/node_warden_impl.cpp

@@ -640,6 +640,10 @@ void TNodeWarden::PersistConfig(const TString& configYaml, ui64 version, std::op
         return;
     }
 
+    if (YamlConfig && version <= YamlConfig->GetConfigVersion()) {
+        return; // some kind of a race
+    }
+
     struct TSaveContext {
         TString ConfigStorePath;
         TString ConfigYaml;

ydb/core/blobstorage/nodewarden/node_warden_resource.cpp

@@ -1,11 +1,13 @@
 #include "node_warden.h"
 #include "node_warden_impl.h"
+#include "distconf.h"
 #include <ydb/core/base/statestorage_impl.h>
 #include <ydb/core/blobstorage/crypto/default.h>
 #include <ydb/core/blobstorage/incrhuge/incrhuge_keeper.h>
 #include <ydb/core/blobstorage/nodewarden/node_warden_events.h>
 #include <ydb/library/pdisk_io/file_params.h>
 #include <ydb/library/pdisk_io/wcache.h>
+#include <library/cpp/streams/zstd/zstd.h>
 #include <util/string/split.h>
 
 using namespace NKikimr;
@@ -112,6 +114,30 @@ void TNodeWarden::Handle(TEvNodeWardenStorageConfig::TPtr ev) {
         Send(subscriber, new TEvNodeWardenStorageConfig(StorageConfig, nullptr, SelfManagementEnabled));
     }
 
+    if (StorageConfig.HasConfigComposite()) {
+        TString mainConfigYaml;
+        ui64 mainConfigYamlVersion;
+        auto error = DecomposeConfig(StorageConfig.GetConfigComposite(), &mainConfigYaml, &mainConfigYamlVersion, nullptr);
+        if (error) {
+            STLOG_DEBUG_FAIL(BS_NODE, NW49, "failed to decompose yaml configuration", (Error, error));
+        } else if (mainConfigYaml) {
+            std::optional<TString> storageConfigYaml;
+            if (StorageConfig.HasCompressedStorageYaml()) {
+                try {
+                    TStringInput s(StorageConfig.GetCompressedStorageYaml());
+                    storageConfigYaml.emplace(TZstdDecompress(&s).ReadAll());
+                } catch (const std::exception& ex) {
+                    Y_ABORT("CompressedStorageYaml format incorrect: %s", ex.what());
+                }
+            }
+
+            // TODO(alexvru): make this blocker for confirmation?
+            PersistConfig(std::move(mainConfigYaml), mainConfigYamlVersion, std::move(storageConfigYaml));
+        }
+    } else {
+        Y_DEBUG_ABORT_UNLESS(!StorageConfig.HasCompressedStorageYaml());
+    }
+
     TActivationContext::Send(new IEventHandle(TEvBlobStorage::EvNodeWardenStorageConfigConfirm, 0, ev->Sender, SelfId(),
         nullptr, ev->Cookie));
 }