Merge pull request #17690 from zverevgeny/merge_25.1.1.4

Merge 25.1.1.4 into 25-1-analytics

Author: maximyurchuk

ydb/apps/ydb/CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed return code of command `ydb workload * run --check-canonical` for the case when benchmark query results differ from canonical ones.
 * Added support for dual configuration mode in the `ydb admin cluster config fetch` command, allowing it to handle separate cluster and storage config sections.
 * Add options for client certificates in SSL/TLS connections.
 * Add `ydb admin node config init` command to initialize directory with node config files.

ydb/core/base/blobstorage.h

@@ -757,6 +757,9 @@ struct TEvBlobStorage {
         EvHugeQueryForbiddenChunks,
         EvHugeForbiddenChunks,
         EvContinueShred,
+        EvQuerySyncToken,
+        EvSyncToken,
+        EvReleaseSyncToken,
 
         EvYardInitResult = EvPut + 9 * 512,                     /// 268 636 672
         EvLogResult,
@@ -870,6 +873,8 @@ struct TEvBlobStorage {
         EvControllerShredResponse                   = 0x1003162a,
         EvControllerFetchConfigRequest              = 0x1003162b,
         EvControllerFetchConfigResponse             = 0x1003162c,
+        EvControllerDistconfRequest                 = 0x1003162d,
+        EvControllerDistconfResponse                = 0x1003162e,
 
         // BSC interface result section
         EvControllerNodeServiceSetUpdate            = 0x10031802,
@@ -2523,6 +2528,8 @@ struct TEvBlobStorage {
     struct TEvControllerShredResponse;
     struct TEvControllerFetchConfigRequest;
     struct TEvControllerFetchConfigResponse;
+    struct TEvControllerDistconfRequest;
+    struct TEvControllerDistconfResponse;
 
     struct TEvMonStreamQuery;
     struct TEvMonStreamActorDeathNote;

ydb/core/base/services/blobstorage_service_id.h

@@ -76,6 +76,11 @@ inline TActorId MakeBlobStorageReplBrokerID() {
     return TActorId(0, TStringBuf(x, 12));
 }
 
+inline TActorId MakeBlobStorageSyncBrokerID() {
+    char x[12] = {'b', 's', 's', 'y', 'n', 'c', 'b', 'r', 'o', 'k', 'e', 'r'};
+    return TActorId(0, TStringBuf(x, 12));
+}
+
 inline TActorId MakeBlobStorageNodeWardenID(ui32 node) {
     char x[12] = {'b','s','n','o','d','e','c','n','t','r','l','r'};
     return TActorId(node, TStringBuf(x, 12));

ydb/core/blobstorage/base/blobstorage_console_events.h

@@ -10,9 +10,10 @@ namespace NKikimr {
             NKikimrBlobStorage::TEvControllerProposeConfigRequest, EvControllerProposeConfigRequest> {
         TEvControllerProposeConfigRequest() = default;
 
-        TEvControllerProposeConfigRequest(ui64 configHash, ui64 configVersion) {
+        TEvControllerProposeConfigRequest(ui64 configHash, ui64 configVersion, bool distconf) {
             Record.SetConfigHash(configHash);
             Record.SetConfigVersion(configVersion);
+            Record.SetDistconf(distconf);
         }
 
         TString ToString() const override {
@@ -32,14 +33,8 @@ namespace NKikimr {
             NKikimrBlobStorage::TEvControllerConsoleCommitRequest, EvControllerConsoleCommitRequest> {
         TEvControllerConsoleCommitRequest() = default;
 
-        TEvControllerConsoleCommitRequest(
-            const TString& yamlConfig,
-            bool allowUnknownFields = false,
-            bool bypassMetadataChecks = false) {
-
+        TEvControllerConsoleCommitRequest(const TString& yamlConfig) {
             Record.SetYAML(yamlConfig);
-            Record.SetAllowUnknownFields(allowUnknownFields);
-            Record.SetBypassMetadataChecks(bypassMetadataChecks);
         }
 
         TString ToString() const override {
@@ -82,14 +77,9 @@ namespace NKikimr {
             NKikimrBlobStorage::TEvControllerReplaceConfigRequest, EvControllerReplaceConfigRequest> {
         TEvControllerReplaceConfigRequest() = default;
 
-        TEvControllerReplaceConfigRequest(
-            std::optional<TString> clusterYaml,
-            std::optional<TString> storageYaml,
-            std::optional<bool> switchDedicatedStorageSection,
-            bool dedicatedConfigMode,
-            bool allowUnknownFields,
-            bool bypassMetadataChecks) {
-
+        TEvControllerReplaceConfigRequest(std::optional<TString> clusterYaml, std::optional<TString> storageYaml,
+                std::optional<bool> switchDedicatedStorageSection, bool dedicatedConfigMode, bool allowUnknownFields,
+                bool bypassMetadataChecks, bool enableConfigV2, bool disableConfigV2) {
             if (clusterYaml) {
                 Record.SetClusterYaml(*clusterYaml);
             }
@@ -102,6 +92,11 @@ namespace NKikimr {
             Record.SetDedicatedConfigMode(dedicatedConfigMode);
             Record.SetAllowUnknownFields(allowUnknownFields);
             Record.SetBypassMetadataChecks(bypassMetadataChecks);
+            if (enableConfigV2) {
+                Record.SetSwitchEnableConfigV2(true);
+            } else if (disableConfigV2) {
+                Record.SetSwitchEnableConfigV2(false);
+            }
         }
 
         TString ToString() const override {
@@ -130,4 +125,10 @@ namespace NKikimr {
     struct TEvBlobStorage::TEvControllerFetchConfigResponse : TEventPB<TEvControllerFetchConfigResponse,
         NKikimrBlobStorage::TEvControllerFetchConfigResponse, EvControllerFetchConfigResponse> {};
 
+    struct TEvBlobStorage::TEvControllerDistconfRequest : TEventPB<TEvControllerDistconfRequest,
+        NKikimrBlobStorage::TEvControllerDistconfRequest, EvControllerDistconfRequest> {};
+
+    struct TEvBlobStorage::TEvControllerDistconfResponse : TEventPB<TEvControllerDistconfResponse,
+        NKikimrBlobStorage::TEvControllerDistconfResponse, EvControllerDistconfResponse> {};
+
 }

ydb/core/blobstorage/dsproxy/dsproxy_get_block.cpp

@@ -11,7 +11,7 @@ namespace NKikimr {
 
 class TBlobStorageGroupGetBlockRequest : public TBlobStorageGroupRequestActor {
     const ui64 TabletId;
-    ui64 Generation;
+    ui64 Generation = 0;
     const TInstant Deadline;
     ui64 Requests = 0;
     ui64 Responses = 0;

ydb/core/blobstorage/nodewarden/distconf.h

@@ -227,6 +227,7 @@ namespace NKikimr::NStorage {
         // pending event queue
         std::deque<TAutoPtr<IEventHandle>> PendingEvents;
         std::vector<ui32> NodeIds;
+        THashSet<ui32> NodeIdsSet;
         TNodeIdentifier SelfNode;
 
         // scatter tasks
@@ -351,8 +352,13 @@ namespace NKikimr::NStorage {
         bool HasQuorum() const;
         void ProcessCollectConfigs(TEvGather::TCollectConfigs *res);
 
-        using TProcessCollectConfigsResult = std::variant<std::monostate, TString, NKikimrBlobStorage::TStorageConfig>;
-        TProcessCollectConfigsResult ProcessCollectConfigs(TEvGather::TCollectConfigs *res, const TString *selfAssemblyUUID);
+        struct TProcessCollectConfigsResult {
+            std::variant<std::monostate, TString, NKikimrBlobStorage::TStorageConfig> Outcome;
+            bool IsDistconfDisabledQuorum = false;
+        };
+        TProcessCollectConfigsResult ProcessCollectConfigs(TEvGather::TCollectConfigs *res,
+            std::optional<TStringBuf> selfAssemblyUUID);
+
         std::optional<TString> ProcessProposeStorageConfig(TEvGather::TProposeStorageConfig *res);
 
         struct TExConfigError : yexception {};

ydb/core/blobstorage/nodewarden/distconf_binding.cpp

@@ -58,6 +58,7 @@ namespace NKikimr::NStorage {
         // issue updates
         NodeIds = std::move(nodeIds);
         BindQueue.Update(NodeIds);
+        NodeIdsSet = {NodeIds.begin(), NodeIds.end()};
     }
 
     void TDistributedConfigKeeper::IssueNextBindRequest() {
@@ -323,7 +324,8 @@ namespace NKikimr::NStorage {
         const auto [it, inserted] = AllBoundNodes.try_emplace(std::move(nodeId));
         TIndirectBoundNode& node = it->second;
 
-        if (inserted) { // disable this node from target binding set, this is the first mention of this node
+        if (inserted && NodeIdsSet.contains(it->first.NodeId())) {
+            // disable this node from target binding set, this is the first mention of this node
             BindQueue.Disable(it->first.NodeId());
         }
 
@@ -363,7 +365,9 @@ namespace NKikimr::NStorage {
 
         if (node.Refs.empty()) {
             AllBoundNodes.erase(it);
-            BindQueue.Enable(nodeId.NodeId());
+            if (NodeIdsSet.contains(nodeId.NodeId())) {
+                BindQueue.Enable(nodeId.NodeId());
+            }
             if (msg) {
                 nodeId.Serialize(msg->Record.AddDeletedBoundNodeIds());
             }
@@ -382,6 +386,12 @@ namespace NKikimr::NStorage {
         STLOG(PRI_DEBUG, BS_NODE, NWDC02, "TEvNodeConfigPush", (NodeId, senderNodeId), (Cookie, ev->Cookie),
             (SessionId, ev->InterconnectSession), (Binding, Binding), (Record, record));
 
+        if (!NodeIdsSet.contains(senderNodeId)) {
+            // node has been already deleted from the config, but new subscription is coming through -- ignoring it
+            SendEvent(*ev, TEvNodeConfigReversePush::MakeRejected());
+            return;
+        }
+
         // check if we can't accept this message (or else it would make a cycle)
         if (record.GetInitial() && senderNodeId == GetRootNodeId()) {
             STLOG(PRI_DEBUG, BS_NODE, NWDC28, "TEvNodeConfigPush rejected", (NodeId, senderNodeId),

ydb/core/blobstorage/nodewarden/distconf_console.cpp

@@ -35,8 +35,14 @@ namespace NKikimr::NStorage {
             return; // still waiting for previous one
         }
 
+        ProposeRequestInFlight = true;
+
         if (!StorageConfig || !StorageConfig->HasConfigComposite()) {
-            return; // no config yet
+            // send empty proposition just to connect to console
+            auto ev = std::make_unique<TEvBlobStorage::TEvControllerProposeConfigRequest>();
+            ev->Record.SetDistconf(true);
+            NTabletPipe::SendData(SelfId(), ConsolePipeId, ev.release(), ++ProposeRequestCookie);
+            return;
         }
 
         Y_ABORT_UNLESS(MainConfigYamlVersion);
@@ -51,11 +57,14 @@ namespace NKikimr::NStorage {
             MainConfigFetchYamlHash, *MainConfigYamlVersion));
         ProposedConfigHashVersion.emplace(MainConfigFetchYamlHash, *MainConfigYamlVersion);
         NTabletPipe::SendData(SelfId(), ConsolePipeId, new TEvBlobStorage::TEvControllerProposeConfigRequest(
-            MainConfigFetchYamlHash, *MainConfigYamlVersion), ++ProposeRequestCookie);
-        ProposeRequestInFlight = true;
+            MainConfigFetchYamlHash, *MainConfigYamlVersion, true), ++ProposeRequestCookie);
     }
 
     void TDistributedConfigKeeper::Handle(TEvBlobStorage::TEvControllerValidateConfigResponse::TPtr ev) {
+        STLOG(PRI_DEBUG, BS_NODE, NWDC10, "received TEvControllerValidateConfigResponse",
+            (Sender, ev->Sender), (Cookie, ev->Cookie), (Record, ev->Get()->Record),
+            (ConsoleConfigValidationQ.size, ConsoleConfigValidationQ.size()));
+
         auto& q = ConsoleConfigValidationQ;
         auto pred = [&](const auto& item) {
             const auto& [actorId, yaml, cookie] = item;
@@ -114,7 +123,7 @@ namespace NKikimr::NStorage {
                 break;
 
             case NKikimrBlobStorage::TEvControllerProposeConfigResponse::ReverseCommit:
-                Y_DEBUG_ABORT();
+                // just do nothing, we didn't have the config in distconf, possibly it is being enabled
                 break;
         }
     }
@@ -204,7 +213,7 @@ namespace NKikimr::NStorage {
         if (!fetched) { // fill in 'to-be-fetched' version of config with version incremented by one
             try {
                 auto metadata = NYamlConfig::GetMainMetadata(yaml);
-                metadata.Cluster = metadata.Cluster.value_or("unknown"); // TODO: fix this
+                metadata.Cluster = metadata.Cluster.value_or(AppData()->ClusterName);
                 metadata.Version = metadata.Version.value_or(0) + 1;
                 temp = NYamlConfig::ReplaceMetadata(yaml, metadata);
             } catch (const std::exception& ex) {

ydb/core/blobstorage/nodewarden/distconf_fsm.cpp

@@ -110,16 +110,17 @@ namespace NKikimr::NStorage {
     }
 
     void TDistributedConfigKeeper::ProcessCollectConfigs(TEvGather::TCollectConfigs *res) {
-        TOverloaded handler{
-            [&](std::monostate&&) {
+        auto r = ProcessCollectConfigs(res, std::nullopt);
+        std::visit(TOverloaded{
+            [&](std::monostate&) {
                 STLOG(PRI_DEBUG, BS_NODE, NWDC61, "ProcessCollectConfigs: monostate");
                 RootState = ERootState::RELAX;
             },
-            [&](TString&& error) {
+            [&](TString& error) {
                 STLOG(PRI_DEBUG, BS_NODE, NWDC63, "ProcessCollectConfigs: error", (Error, error));
                 SwitchToError(error);
             },
-            [&](NKikimrBlobStorage::TStorageConfig&& proposedConfig) {
+            [&](NKikimrBlobStorage::TStorageConfig& proposedConfig) {
                 STLOG(PRI_DEBUG, BS_NODE, NWDC64, "ProcessCollectConfigs: proposed new config",
                     (ProposedConfig, proposedConfig));
                 TEvScatter task;
@@ -130,12 +131,11 @@ namespace NKikimr::NStorage {
                 propose->MutableConfig()->Swap(&proposedConfig);
                 IssueScatterTask(TActorId(), std::move(task));
             }
-        };
-        std::visit(handler, ProcessCollectConfigs(res, nullptr));
+        }, r.Outcome);
     }
 
     TDistributedConfigKeeper::TProcessCollectConfigsResult TDistributedConfigKeeper::ProcessCollectConfigs(
-            TEvGather::TCollectConfigs *res, const TString *selfAssemblyUUID) {
+            TEvGather::TCollectConfigs *res, std::optional<TStringBuf> selfAssemblyUUID) {
         auto generateSuccessful = [&](auto&& callback) {
             for (const auto& item : res->GetNodes()) {
                 for (const auto& node : item.GetNodeIds()) {
@@ -171,10 +171,24 @@ namespace NKikimr::NStorage {
 
         if (nodeQuorum && !configQuorum) {
             // check if there is quorum of no-distconf config along the cluster
+            auto generateNodesWithoutDistconf = [&](auto&& callback) {
+                for (const auto& item : res->GetNodes()) {
+                    if (item.GetBaseConfig().GetSelfManagementConfig().GetEnabled()) {
+                        continue;
+                    }
+                    for (const auto& node : item.GetNodeIds()) {
+                        callback(node);
+                    }
+                }
+            };
+            if (HasNodeQuorum(*StorageConfig, generateNodesWithoutDistconf)) {
+                // yes, distconf is disabled on the majority of the nodes, so we can't do anything about it
+                return {.IsDistconfDisabledQuorum = true};
+            }
         }
 
         if (!nodeQuorum || !configQuorum) {
-            return "no quorum for CollectConfigs";
+            return {"no quorum for CollectConfigs"};
         }
 
         // TODO: validate self-assembly UUID
@@ -223,7 +237,7 @@ namespace NKikimr::NStorage {
                 (BaseConfigs.size, baseConfigs.size()));
             Y_DEBUG_ABORT("Multiple nonintersecting node sets have quorum of BaseConfig");
             Halt();
-            return "Multiple nonintersecting node sets have quorum of BaseConfig";
+            return {"Multiple nonintersecting node sets have quorum of BaseConfig"};
         }
         NKikimrBlobStorage::TStorageConfig *baseConfig = nullptr;
         for (auto& [meta, info] : baseConfigs) {
@@ -289,13 +303,13 @@ namespace NKikimr::NStorage {
                     (Generation, generation), (Configs, configs));
                 Y_DEBUG_ABORT("Multiple nonintersecting node sets have quorum of persistent config");
                 Halt();
-                return "Multiple nonintersecting node sets have quorum of persistent config";
+                return {"Multiple nonintersecting node sets have quorum of persistent config"};
             }
             Y_ABORT_UNLESS(configs.size() == 1);
             persistedConfig = configs.front();
         }
         if (maxSeenGeneration && (!persistedConfig || persistedConfig->GetGeneration() < maxSeenGeneration)) {
-            return "couldn't obtain quorum for configuration that was seen in effect";
+            return {"couldn't obtain quorum for configuration that was seen in effect"};
         }
 
         // let's try to find possibly proposed config, but without a quorum, and try to reconstruct it
@@ -311,7 +325,7 @@ namespace NKikimr::NStorage {
                             (PersistentConfig, *persistedConfig), (ProposedConfig, config));
                         Y_DEBUG_ABORT("persistently proposed config has too big generation");
                         Halt();
-                        return "persistently proposed config has too big generation";
+                        return {"persistently proposed config has too big generation"};
                     }
                 }
                 if (proposedConfig && (proposedConfig->GetGeneration() != config.GetGeneration() ||
@@ -361,11 +375,11 @@ namespace NKikimr::NStorage {
                     if (!CurrentSelfAssemblyUUID) {
                         CurrentSelfAssemblyUUID.emplace(CreateGuidAsString());
                     }
-                    selfAssemblyUUID = &CurrentSelfAssemblyUUID.value();
+                    selfAssemblyUUID.emplace(CurrentSelfAssemblyUUID.value());
                 }
                 propositionBase.emplace(*baseConfig);
-                if (auto error = GenerateFirstConfig(baseConfig, *selfAssemblyUUID)) {
-                    return *error;
+                if (auto error = GenerateFirstConfig(baseConfig, TString(*selfAssemblyUUID))) {
+                    return {*error};
                 }
                 configToPropose = baseConfig;
             }
@@ -392,12 +406,12 @@ namespace NKikimr::NStorage {
 
             if (error) {
                 Y_DEBUG_ABORT("incorrect config proposition");
-                return "incorrect config proposition";
+                return {"incorrect config proposition"};
             }
 
             if (propositionBase) {
                 if (auto error = ValidateConfig(*propositionBase)) {
-                    return TStringBuilder() << "failed to propose configuration, base config contains errors: " << *error;
+                    return {TStringBuilder() << "failed to propose configuration, base config contains errors: " << *error};
                 }
                 if (auto error = ValidateConfigUpdate(*propositionBase, *configToPropose)) {
                     Y_FAIL_S("incorrect config proposed: " << *error);
@@ -408,7 +422,7 @@ namespace NKikimr::NStorage {
                 }
             }
 
-            return std::move(*configToPropose);
+            return {std::move(*configToPropose)};
         }
 
         return {};