diff --git a/docs/en/operations/cluster-discovery.md b/docs/en/operations/cluster-discovery.md index 011eccd2da5c..b1c80f3a0340 100644 --- a/docs/en/operations/cluster-discovery.md +++ b/docs/en/operations/cluster-discovery.md @@ -62,6 +62,8 @@ Traditionally, in ClickHouse, each shard and replica in the cluster needed to be With Cluster Discovery, rather than defining each node explicitly, you simply specify a path in ZooKeeper. All nodes that register under this path in ZooKeeper will be automatically discovered and added to the cluster. +Discovery settings under `remote_servers` (including `user`, `password`, `secret`, `path`, `multicluster_root_path`, and adding or removing discovery clusters) are applied on configuration reload (for example with `SYSTEM RELOAD CONFIG`). A server restart is not required for these changes. + ```xml diff --git a/src/Common/FailPoint.cpp b/src/Common/FailPoint.cpp index 1ec87f9d3aaf..10219d9e0152 100644 --- a/src/Common/FailPoint.cpp +++ b/src/Common/FailPoint.cpp @@ -41,6 +41,8 @@ static struct InitFiu REGULAR(use_delayed_remote_source) \ ONCE(remote_query_executor_cancel_before_send) \ REGULAR(cluster_discovery_faults) \ + REGULAR(cluster_discovery_unregister_fail) \ + ONCE(cluster_discovery_retry_signal_fail) \ REGULAR(stripe_log_sink_write_fallpoint) \ REGULAR(hybrid_watermarks_read_fail) \ ONCE(smt_commit_merge_mutate_zk_fail_after_op) \ diff --git a/src/Interpreters/Cluster.cpp b/src/Interpreters/Cluster.cpp index 2967d2e36a23..12c1dea5487b 100644 --- a/src/Interpreters/Cluster.cpp +++ b/src/Interpreters/Cluster.cpp @@ -386,36 +386,40 @@ void Clusters::updateClusters(const Poco::Util::AbstractConfiguration & new_conf std::lock_guard lock(mutex); - /// If old config is set, remove deleted clusters from impl, otherwise just clear it. + /// If old config is set, remove deleted clusters; otherwise rebuild ownership from scratch + /// while preserving non-automatic entries (e.g. clusters added via setCluster). if (old_config) { for (const auto & key : deleted_keys) { - if (!automatic_clusters.contains(key)) - impl.erase(key); + automatic_clusters.erase(key); + impl.erase(key); } } else { - if (!automatic_clusters.empty()) - std::erase_if(impl, [this](const auto & e) { return automatic_clusters.contains(e.first); }); - else - impl.clear(); + for (const auto & name : automatic_clusters) + impl.erase(name); + automatic_clusters.clear(); } - for (const auto & key : new_config_keys) { if (new_config.has(config_prefix + "." + key + ".discovery")) { - /// Handled in ClusterDiscovery + /// Handled in ClusterDiscovery — must not leave a prior static Cluster in impl, + /// or Context::getCluster / getClusters would prefer the stale static entry. automatic_clusters.insert(key); + impl.erase(key); continue; } if (key.contains('.')) throw Exception(ErrorCodes::SYNTAX_ERROR, "Cluster names with dots are not supported: '{}'", key); + /// Leaving discovery (or never was discovery): drop automatic ownership for this name. + automatic_clusters.erase(key); + /// If old config is set and cluster config wasn't changed, don't update this cluster. if (!old_config || !isSameConfiguration(new_config, *old_config, config_prefix + "." + key)) impl[key] = std::make_shared(new_config, settings, config_prefix, key); diff --git a/src/Interpreters/ClusterDiscovery.cpp b/src/Interpreters/ClusterDiscovery.cpp index 9df92e7a4a52..14c02fb07ec1 100644 --- a/src/Interpreters/ClusterDiscovery.cpp +++ b/src/Interpreters/ClusterDiscovery.cpp @@ -2,7 +2,9 @@ #include #include #include +#include #include +#include #include #include @@ -54,6 +56,8 @@ namespace ErrorCodes namespace FailPoints { extern const char cluster_discovery_faults[]; + extern const char cluster_discovery_unregister_fail[]; + extern const char cluster_discovery_retry_signal_fail[]; } namespace @@ -78,7 +82,7 @@ ClusterDiscovery::ClusterInfo::ClusterInfo(const String & name_, size_t shard_id, bool observer_mode, bool invisible, - size_t zk_root_index_ + const String & multicluster_full_path_ ) : name(name_) , zk_name(zk_name_) @@ -90,7 +94,7 @@ ClusterDiscovery::ClusterInfo::ClusterInfo(const String & name_, , username(username_) , password(password_) , cluster_secret(cluster_secret_) - , zk_root_index(zk_root_index_) + , multicluster_full_path(multicluster_full_path_) { } @@ -103,12 +107,7 @@ template class ClusterDiscovery::Flags { public: - template - Flags(It begin, It end) - { - for (auto it = begin; it != end; ++it) - flags.emplace(*it, false); - } + Flags() = default; void set(const T & key, bool value = true) { @@ -120,6 +119,21 @@ class ClusterDiscovery::Flags cv.notify_one(); } + /// Set an existing key only. Late Keeper watch callbacks must use this so a removed + /// cluster name cannot be reinserted after Flags::remove. + void setIfPresent(const T & key, bool value = true) + { + std::unique_lock lk(mu); + if (stop_flag) + return; + auto it = flags.find(key); + if (it == flags.end()) + return; + it->second = value; + any_need_update |= value; + cv.notify_one(); + } + /// Just notify the condition variable. void set() { @@ -157,6 +171,12 @@ class ClusterDiscovery::Flags cv.notify_one(); } + bool isStopped() const + { + std::unique_lock lk(mu); + return stop_flag; + } + void wakeup() { std::unique_lock lk(mu); @@ -166,7 +186,7 @@ class ClusterDiscovery::Flags private: std::condition_variable cv; - std::mutex mu; + mutable std::mutex mu; /// flag indicates that update is required std::unordered_map flags; @@ -174,17 +194,12 @@ class ClusterDiscovery::Flags bool stop_flag = false; }; -ClusterDiscovery::ClusterDiscovery( +ClusterDiscovery::ParsedDiscoveryConfig ClusterDiscovery::parseDiscoveryConfig( const Poco::Util::AbstractConfiguration & config, - ContextPtr context_, - MultiVersion::Version macros_, + ContextPtr context, const String & config_prefix) - : context(Context::createCopy(context_)) - , current_node_name(toString(ServerUUID::get())) - , log(getLogger("ClusterDiscovery")) - , macros(macros_) { - LOG_DEBUG(log, "Cluster discovery is enabled"); + ParsedDiscoveryConfig result; Poco::Util::AbstractConfiguration::Keys config_keys; config.keys(config_prefix, config_keys); @@ -220,16 +235,14 @@ ClusterDiscovery::ClusterDiscovery( String zk_root = zkutil::extractZooKeeperPath(zk_multicluster_name_and_root, true); String zk_name = zkutil::extractZooKeeperName(zk_multicluster_name_and_root); - MulticlusterDiscovery mcd( - /* zk_name */ zk_name, - /* zk_path */ zk_root, - /* is_secure_connection */ config.getBool(cluster_config_prefix + ".secure", false), - /* username */ config.getString(cluster_config_prefix + ".user", context->getUserName()), - /* password */ password, - /* cluster_secret */ cluster_secret - ); - - multicluster_discovery_paths.push_back(std::move(mcd)); + result.multicluster_roots.push_back(ParsedMulticlusterDiscovery{ + .zk_name = zk_name, + .zk_path = zk_root, + .is_secure_connection = config.getBool(cluster_config_prefix + ".secure", false), + .username = config.getString(cluster_config_prefix + ".user", context->getUserName()), + .password = password, + .cluster_secret = cluster_secret, + }); continue; } @@ -239,51 +252,433 @@ ClusterDiscovery::ClusterDiscovery( String zk_root = zkutil::extractZooKeeperPath(zk_name_and_root, true); String zk_name = zkutil::extractZooKeeperName(zk_name_and_root); - clusters_info.emplace( - key, - ClusterInfo( - /* name_= */ key, - /* zk_name_= */ zk_name, - /* zk_root_= */ zk_root, - /* host_name= */ config.getString(cluster_config_prefix + ".my_hostname", getFQDNOrHostName()), - /* username= */ config.getString(cluster_config_prefix + ".user", context->getUserName()), - /* password= */ password, - /* cluster_secret= */ cluster_secret, - /* port= */ context->getTCPPort(), - /* secure= */ config.getBool(cluster_config_prefix + ".secure", false), - /* shard_id= */ config.getUInt(cluster_config_prefix + ".shard", 0), - /* observer_mode= */ is_observer, - /* invisible= */ ConfigHelper::getBool(config, cluster_config_prefix + ".invisible") - ) - ); + result.static_clusters.push_back(ParsedStaticDiscovery{ + .name = key, + .zk_name = zk_name, + .zk_root = zk_root, + .host_name = config.getString(cluster_config_prefix + ".my_hostname", getFQDNOrHostName()), + .username = config.getString(cluster_config_prefix + ".user", context->getUserName()), + .password = password, + .cluster_secret = cluster_secret, + .secure = config.getBool(cluster_config_prefix + ".secure", false), + .shard_id = config.getUInt(cluster_config_prefix + ".shard", 0), + .observer = is_observer, + .invisible = ConfigHelper::getBool(config, cluster_config_prefix + ".invisible"), + }); } + return result; +} + +void ClusterDiscovery::validateConfig( + const Poco::Util::AbstractConfiguration & config, + ContextPtr context, + const String & config_prefix) +{ + /// Discard result; throws on invalid discovery subtrees. + parseDiscoveryConfig(config, context, config_prefix); +} + +ClusterDiscovery::ClusterDiscovery( + const Poco::Util::AbstractConfiguration & config, + ContextPtr context_, + MultiVersion::Version macros_, + const String & config_prefix) + : context(Context::createCopy(context_)) + , current_node_name(toString(ServerUUID::get())) + , clusters_to_update(std::make_shared()) + , log(getLogger("ClusterDiscovery")) + , macros(macros_) +{ + LOG_DEBUG(log, "Cluster discovery is enabled"); + + auto parsed = parseDiscoveryConfig(config, context, config_prefix); + + for (auto & static_cluster : parsed.static_clusters) + addStaticCluster(std::move(static_cluster)); + + for (auto & root : parsed.multicluster_roots) + addMulticlusterRoot(std::move(root)); + std::vector clusters_info_names; clusters_info_names.reserve(clusters_info.size()); for (const auto & e : clusters_info) clusters_info_names.emplace_back(e.first); LOG_TRACE(log, "Clusters in discovery mode: {}", fmt::join(clusters_info_names, ", ")); - clusters_to_update = std::make_shared(clusters_info_names.begin(), clusters_info_names.end()); +} - /// Init get_nodes_callbacks after init clusters_to_update. - for (const auto & e : clusters_info) - get_nodes_callbacks[e.first] = std::make_shared( - [cluster_name = e.first, my_clusters_to_update = clusters_to_update](auto) - { - my_clusters_to_update->set(cluster_name); - }); +void ClusterDiscovery::addStaticCluster(ParsedStaticDiscovery && parsed) +{ + const String name = parsed.name; - for (auto & path : multicluster_discovery_paths) + if (auto existing = clusters_info.find(name); existing != clusters_info.end()) { - path.watch_callback = std::make_shared( - [my_need_update = path.need_update, my_flag = clusters_to_update](auto) - { - my_need_update->store(true); - my_flag->set(); - } - ); + if (existing->second.isDynamic()) + { + /// Static config wins over multicluster discovery (same as findDynamicClusters). + removeDynamicCluster(name); + } + else + { + LOG_DEBUG(log, "Static discovery cluster '{}' already exists, skip add", name); + return; + } + } + + clusters_info.emplace( + name, + ClusterInfo( + /* name_= */ parsed.name, + /* zk_name_= */ parsed.zk_name, + /* zk_root_= */ parsed.zk_root, + /* host_name= */ parsed.host_name, + /* username= */ parsed.username, + /* password= */ parsed.password, + /* cluster_secret= */ parsed.cluster_secret, + /* port= */ context->getTCPPort(), + /* secure= */ parsed.secure, + /* shard_id= */ parsed.shard_id, + /* observer_mode= */ parsed.observer, + /* invisible= */ parsed.invisible)); + + get_nodes_callbacks[name] = std::make_shared( + [cluster_name = name, my_clusters_to_update = clusters_to_update](auto) + { + my_clusters_to_update->setIfPresent(cluster_name); + }); + + clusters_to_update->set(name); +} + +void ClusterDiscovery::removeStaticCluster(const String & name) +{ + auto it = clusters_info.find(name); + if (it == clusters_info.end() || it->second.isDynamic()) + return; + + /// Drop local tracking even if Keeper remove fails: config already removed the cluster. + /// Keep enough identity to retry ephemeral cleanup so peers stop seeing this node. + /// Aliases that share zk_name/zk_root share one ephemeral; do not remove it while another + /// participant alias still needs the registration. + if (pathHasActiveParticipant(it->second.zk_name, it->second.zk_root, &name)) + { + LOG_DEBUG( + log, + "Skip unregister for removed cluster '{}': another participant still owns path {}:{}", + name, + it->second.zk_name, + it->second.zk_root); + } + else if (!unregisterFromZk(it->second.zk_name, it->second.zk_root, name)) + { + pending_zk_unregisters.push_back(PendingZkUnregister{ + .zk_name = it->second.zk_name, + .zk_root = it->second.zk_root, + .cluster_name = name, + }); + LOG_WARNING( + log, + "Failed to unregister current node from cluster '{}' on config remove; will retry", + name); + } + + clusters_to_update->remove(name); + get_nodes_callbacks.erase(name); + clusters_info.erase(it); + + { + std::lock_guard lock(mutex); + cluster_impls.erase(name); + } + + /// A static entry may have been shadowing a same-named cluster under a multicluster root. + /// Roots are otherwise skipped while need_update is false and no children event fires. + if (!multicluster_discovery_paths.empty()) + { + markMulticlusterRootsNeedUpdate(); + clusters_to_update->set(); + } + + LOG_DEBUG(log, "Static discovery cluster '{}' removed due to config change", name); +} + +void ClusterDiscovery::removeDynamicCluster(const String & name) +{ + auto it = clusters_info.find(name); + if (it == clusters_info.end() || !it->second.isDynamic()) + return; + + removeCluster(name, /* is_dynamic */ true); + clusters_info.erase(name); + LOG_DEBUG(log, "Dynamic discovery cluster '{}' removed to make way for static config", name); +} + +bool ClusterDiscovery::updateStaticClusterFields(ClusterInfo & info, const ParsedStaticDiscovery & parsed) +{ + bool identity_changed = info.zk_name != parsed.zk_name || info.zk_root != parsed.zk_root; + if (identity_changed) + return false; + + bool credentials_changed + = info.username != parsed.username + || info.password != parsed.password + || info.cluster_secret != parsed.cluster_secret + || info.is_secure_connection != parsed.secure; + + String expected_address = parsed.host_name + ":" + toString(context->getTCPPort()); + bool registration_changed + = info.current_node_is_observer != parsed.observer + || info.current_node.shard_id != parsed.shard_id + || info.current_node.address != expected_address + || info.current_node.secure != parsed.secure; + + bool invisible_changed = info.current_cluster_is_invisible != parsed.invisible; + + if (!credentials_changed && !registration_changed && !invisible_changed) + return true; + + info.username = parsed.username; + info.password = parsed.password; + info.cluster_secret = parsed.cluster_secret; + info.is_secure_connection = parsed.secure; + info.current_node_is_observer = parsed.observer; + info.current_cluster_is_invisible = parsed.invisible; + info.current_node = NodeInfo(expected_address, parsed.secure, parsed.shard_id); + + if (registration_changed || invisible_changed) + { + /// Force upsertCluster to re-read ZK payloads; membership UUIDs alone do not change + /// when only address/shard/secure are updated. + if (registration_changed) + info.nodes_info.clear(); + clusters_to_update->set(info.name); + } + else + rebuildClusterObject(info); + + return true; +} + +void ClusterDiscovery::addMulticlusterRoot(ParsedMulticlusterDiscovery && parsed) +{ + const String full_path = parsed.getFullPath(); + if (multicluster_discovery_paths.contains(full_path)) + return; + + MulticlusterDiscovery mcd( + /* zk_name */ parsed.zk_name, + /* zk_path */ parsed.zk_path, + /* is_secure_connection */ parsed.is_secure_connection, + /* username */ parsed.username, + /* password */ parsed.password, + /* cluster_secret */ parsed.cluster_secret); + + mcd.watch_callback = std::make_shared( + [my_need_update = mcd.need_update, my_flag = clusters_to_update](auto) + { + my_need_update->store(true); + my_flag->set(); + }); + + multicluster_discovery_paths.emplace(full_path, std::move(mcd)); + clusters_to_update->set(); + + LOG_DEBUG(log, "Added multicluster discovery root '{}'", full_path); +} + +void ClusterDiscovery::removeMulticlusterRoot(const String & full_path) +{ + if (!multicluster_discovery_paths.erase(full_path)) + return; + + std::vector dynamic_clusters; + for (const auto & [name, info] : clusters_info) + { + if (info.multicluster_full_path == full_path) + dynamic_clusters.push_back(name); + } + + for (const auto & name : dynamic_clusters) + removeDynamicCluster(name); + + clusters_to_update->set(); + + LOG_DEBUG(log, "Removed multicluster discovery root '{}'", full_path); +} + +void ClusterDiscovery::markMulticlusterRootsNeedUpdate() +{ + for (auto & [_, path] : multicluster_discovery_paths) + path.need_update->store(true); +} + +bool ClusterDiscovery::updateMulticlusterRootFields(MulticlusterDiscovery & path, const ParsedMulticlusterDiscovery & parsed) +{ + bool credentials_changed + = path.username != parsed.username + || path.password != parsed.password + || path.cluster_secret != parsed.cluster_secret + || path.is_secure_connection != parsed.is_secure_connection; + + if (!credentials_changed) + return true; + + path.username = parsed.username; + path.password = parsed.password; + path.cluster_secret = parsed.cluster_secret; + path.is_secure_connection = parsed.is_secure_connection; + + const String full_path = path.getFullPath(); + for (auto & [_, info] : clusters_info) + { + if (info.multicluster_full_path != full_path) + continue; + info.username = parsed.username; + info.password = parsed.password; + info.cluster_secret = parsed.cluster_secret; + info.is_secure_connection = parsed.is_secure_connection; + info.current_node.secure = parsed.is_secure_connection; + rebuildClusterObject(info); + } + + return true; +} + +void ClusterDiscovery::rebuildClusterObject(const ClusterInfo & info) +{ + if (info.current_cluster_is_invisible || info.nodes_info.empty()) + { + std::lock_guard lock(mutex); + cluster_impls.erase(info.name); + return; + } + + auto cluster = makeCluster(info); + std::lock_guard lock(mutex); + cluster_impls[info.name] = std::move(cluster); +} + +void ClusterDiscovery::applyParsedConfig(ParsedDiscoveryConfig && parsed) +{ + std::unordered_map desired_static; + for (auto & c : parsed.static_clusters) + desired_static.emplace(c.name, std::move(c)); + + std::unordered_map desired_multi; + for (auto & r : parsed.multicluster_roots) + desired_multi.emplace(r.getFullPath(), std::move(r)); + + std::vector static_to_remove; + std::vector static_identity_replace; + std::vector static_to_add; + std::vector multi_to_remove; + std::vector multi_to_add; + + for (auto & [name, info] : clusters_info) + { + if (info.isDynamic()) + continue; + + auto it = desired_static.find(name); + if (it == desired_static.end()) + { + static_to_remove.push_back(name); + continue; + } + + if (!updateStaticClusterFields(info, it->second)) + static_identity_replace.push_back(name); + else + desired_static.erase(it); + } + + for (auto & [full_path, path] : multicluster_discovery_paths) + { + auto it = desired_multi.find(full_path); + if (it == desired_multi.end()) + { + multi_to_remove.push_back(full_path); + continue; + } + + updateMulticlusterRootFields(path, it->second); + desired_multi.erase(it); + } + + for (const auto & name : static_identity_replace) + { + auto it = desired_static.find(name); + if (it != desired_static.end()) + { + static_to_add.push_back(std::move(it->second)); + desired_static.erase(it); + } + static_to_remove.push_back(name); + } + + for (auto & [_, c] : desired_static) + static_to_add.push_back(std::move(c)); + for (auto & [_, r] : desired_multi) + multi_to_add.push_back(std::move(r)); + + for (const auto & name : static_to_remove) + removeStaticCluster(name); + for (auto & c : static_to_add) + addStaticCluster(std::move(c)); + + for (const auto & full_path : multi_to_remove) + removeMulticlusterRoot(full_path); + for (auto & r : multi_to_add) + addMulticlusterRoot(std::move(r)); +} + +void ClusterDiscovery::updateFromConfig( + const Poco::Util::AbstractConfiguration & config, + const String & config_prefix) +{ + LOG_DEBUG(log, "Scheduling cluster discovery config update"); + auto parsed = parseDiscoveryConfig(config, context, config_prefix); + { + std::lock_guard lock(pending_config_mutex); + pending_config_update = std::move(parsed); + } + ensureWorkerStarted(); + clusters_to_update->set(); +} + +bool ClusterDiscovery::consumePendingConfigUpdate() +{ + std::optional pending; + { + std::lock_guard lock(pending_config_mutex); + pending.swap(pending_config_update); + } + if (!pending) + return false; + + LOG_DEBUG(log, "Applying pending cluster discovery config update"); + applyParsedConfig(std::move(*pending)); + return true; +} + +void ClusterDiscovery::ensureWorkerStarted() +{ + std::lock_guard lock(start_mutex); + if (main_thread.joinable()) + return; + + if (clusters_info.empty() && multicluster_discovery_paths.empty()) + { + std::lock_guard pending_lock(pending_config_mutex); + if (!pending_config_update) + return; + /// Pending update may add the first discovery path; apply it before start(). } + + /// If worker is not running yet, apply pending config inline so startImpl() sees new clusters. + consumePendingConfigUpdate(); + startImpl(); } /// List node in zookeper for cluster @@ -292,7 +687,7 @@ Strings ClusterDiscovery::getNodeNames(zkutil::ZooKeeperPtr & zk, const String & cluster_name, int * version, bool set_callback, - size_t zk_root_index) + const String & multicluster_full_path) { Coordination::Stat stat; Strings nodes; @@ -302,14 +697,23 @@ Strings ClusterDiscovery::getNodeNames(zkutil::ZooKeeperPtr & zk, auto callback = get_nodes_callbacks.find(cluster_name); if (callback == get_nodes_callbacks.end()) { + std::shared_ptr need_update; + if (!multicluster_full_path.empty()) + { + auto path_it = multicluster_discovery_paths.find(multicluster_full_path); + if (path_it != multicluster_discovery_paths.end()) + need_update = path_it->second.need_update; + } + auto watch_dynamic_callback = std::make_shared([ cluster_name, my_clusters_to_update = clusters_to_update, - my_discovery_paths_need_update = multicluster_discovery_paths[zk_root_index - 1].need_update + my_discovery_paths_need_update = need_update ](auto) { - my_discovery_paths_need_update->store(true); - my_clusters_to_update->set(cluster_name); + if (my_discovery_paths_need_update) + my_discovery_paths_need_update->store(true); + my_clusters_to_update->setIfPresent(cluster_name); }); auto res = get_nodes_callbacks.insert(std::make_pair(cluster_name, watch_dynamic_callback)); callback = res.first; @@ -346,10 +750,8 @@ ClusterDiscovery::NodesInfo ClusterDiscovery::getNodes(zkutil::ZooKeeperPtr & zk return result; } -/// Checks if cluster nodes set is changed. -/// Returns true if update required. -/// It performs only shallow check (set of nodes' uuids). -/// So, if node's hostname are changed, then cluster won't be updated. +/// Checks if cluster membership (set of node UUIDs) changed. +/// Used for logging; payload refresh is decided separately in upsertCluster. bool ClusterDiscovery::needUpdate(const Strings & node_uuids, const NodesInfo & nodes) { bool has_difference = node_uuids.size() != nodes.size() || @@ -437,15 +839,18 @@ bool ClusterDiscovery::upsertCluster(ClusterInfo & cluster_info) LOG_DEBUG(log, "Updating cluster '{}'", cluster_info.name); auto zk = context->getDefaultOrAuxiliaryZooKeeper(cluster_info.zk_name); + registerInZk(zk, cluster_info); int start_version = 0; - Strings node_uuids = getNodeNames(zk, cluster_info.zk_root, cluster_info.name, &start_version, false, cluster_info.zk_root_index); + Strings node_uuids = getNodeNames( + zk, cluster_info.zk_root, cluster_info.name, &start_version, false, cluster_info.multicluster_full_path); auto & nodes_info = cluster_info.nodes_info; auto on_exit = [this, start_version, &zk, &cluster_info, &nodes_info]() { /// in case of successful update we still need to check if configuration of cluster still valid and also set watch callback int current_version = 0; - getNodeNames(zk, cluster_info.zk_root, cluster_info.name, ¤t_version, true, cluster_info.zk_root_index); + getNodeNames( + zk, cluster_info.zk_root, cluster_info.name, ¤t_version, true, cluster_info.multicluster_full_path); if (current_version != start_version) { @@ -469,15 +874,17 @@ bool ClusterDiscovery::upsertCluster(ClusterInfo & cluster_info) if (cluster_info.current_cluster_is_invisible) { LOG_DEBUG(log, "Cluster '{}' is invisible.", cluster_info.name); + std::lock_guard lock(mutex); + cluster_impls.erase(cluster_info.name); return true; } if (!needUpdate(node_uuids, nodes_info)) - { - LOG_DEBUG(log, "No update required for cluster '{}'", cluster_info.name); - return on_exit(); - } + LOG_DEBUG(log, "Membership unchanged for cluster '{}', refreshing node payloads", cluster_info.name); + /// Always re-read ephemeral payloads so hostname/shard/secure updates propagate even when + /// the UUID set is unchanged (createOrUpdate does not fire children watches by itself; + /// registerInZk recreates the node when data changes to notify peers). nodes_info = getNodes(zk, cluster_info.zk_root, node_uuids); if (bool ok = on_exit(); !ok) @@ -487,14 +894,15 @@ bool ClusterDiscovery::upsertCluster(ClusterInfo & cluster_info) if (nodes_info.empty()) { - removeCluster(cluster_info.name, /* is_dynamic_cluster */cluster_info.zk_root_index != 0); + String name = cluster_info.name; + if (cluster_info.isDynamic()) + removeDynamicCluster(name); + else + removeCluster(name, /* is_dynamic */ false); return true; } - auto cluster = makeCluster(cluster_info); - std::lock_guard lock(mutex); - cluster_impls[cluster_info.name] = cluster; - + rebuildClusterObject(cluster_info); return true; } @@ -523,6 +931,33 @@ void ClusterDiscovery::registerInZk(zkutil::ZooKeeperPtr & zk, ClusterInfo & inf if (info.current_node_is_observer) { + /// Drop leftover ephemeral registration when transitioning from participant to observer + /// (or if a stale node remained). ZNONODE means already absent. + /// Shared-path aliases: another participant may still own this ephemeral. + if (pathHasActiveParticipant(info.zk_name, info.zk_root)) + { + LOG_DEBUG( + log, + "Current node {} is observer of cluster {} (shared path {}:{} still owned by another participant)", + current_node_name, + info.name, + info.zk_name, + info.zk_root); + return; + } + fiu_do_on(FailPoints::cluster_discovery_retry_signal_fail, + { + throw Exception( + ErrorCodes::KEEPER_EXCEPTION, + "Failpoint cluster_discovery_retry_signal_fail is triggered in observer tryRemove"); + }); + auto code = zk->tryRemove(node_path); + if (code != Coordination::Error::ZOK && code != Coordination::Error::ZNONODE) + throw Exception( + ErrorCodes::KEEPER_EXCEPTION, + "Cannot remove discovery registration for observer node {}: {}", + node_path, + Coordination::errorMessage(code)); LOG_DEBUG(log, "Current node {} is observer of cluster {}", current_node_name, info.name); return; } @@ -535,10 +970,96 @@ void ClusterDiscovery::registerInZk(zkutil::ZooKeeperPtr & zk, ClusterInfo & inf LOG_DEBUG(log, "Registering current node {} in cluster {}", current_node_name, info.name); - zk->createOrUpdate(node_path, info.current_node.serialize(), zkutil::CreateMode::Ephemeral); + const String payload = info.current_node.serialize(); + String existing; + if (zk->tryGet(node_path, existing)) + { + if (existing == payload) + { + LOG_DEBUG(log, "Current node {} already registered in cluster {} with up-to-date data", current_node_name, info.name); + return; + } + /// Recreate ephemeral so children watches fire; setData alone does not notify peers. + zk->tryRemove(node_path); + } + + zk->create(node_path, payload, zkutil::CreateMode::Ephemeral); LOG_DEBUG(log, "Current node {} registered in cluster {}", current_node_name, info.name); } +bool ClusterDiscovery::unregisterFromZk(const String & zk_name, const String & zk_root, const String & cluster_name) +{ + try + { + fiu_do_on(FailPoints::cluster_discovery_unregister_fail, + { + throw Exception( + ErrorCodes::KEEPER_EXCEPTION, + "Failpoint cluster_discovery_unregister_fail is triggered for cluster '{}'", + cluster_name); + }); + + auto zk = context->getDefaultOrAuxiliaryZooKeeper(zk_name); + String node_path = getShardsListPath(zk_root) / current_node_name; + auto code = zk->tryRemove(node_path); + if (code != Coordination::Error::ZOK && code != Coordination::Error::ZNONODE) + { + LOG_WARNING( + log, + "Cannot unregister current node {} from cluster '{}': {}", + current_node_name, + cluster_name, + Coordination::errorMessage(code)); + return false; + } + + LOG_DEBUG(log, "Current node {} unregistered from cluster {}", current_node_name, cluster_name); + return true; + } + catch (...) + { + tryLogCurrentException(log, "Error while unregistering node from cluster '" + cluster_name + "'"); + return false; + } +} + +bool ClusterDiscovery::pathHasActiveParticipant( + const String & zk_name, + const String & zk_root, + const String * exclude_cluster_name) const +{ + for (const auto & [name, info] : clusters_info) + { + if (exclude_cluster_name && name == *exclude_cluster_name) + continue; + if (!info.current_node_is_observer && info.zk_name == zk_name && info.zk_root == zk_root) + return true; + } + return false; +} + +bool ClusterDiscovery::retryPendingUnregisters() +{ + if (pending_zk_unregisters.empty()) + return true; + + std::vector still_pending; + still_pending.reserve(pending_zk_unregisters.size()); + + for (const auto & pending : pending_zk_unregisters) + { + /// Re-add put a participant back on this path; drop stale cleanup instead of deleting the live ephemeral. + if (pathHasActiveParticipant(pending.zk_name, pending.zk_root)) + continue; + + if (!unregisterFromZk(pending.zk_name, pending.zk_root, pending.cluster_name)) + still_pending.push_back(pending); + } + + pending_zk_unregisters = std::move(still_pending); + return pending_zk_unregisters.empty(); +} + void ClusterDiscovery::unregisterFromZk(zkutil::ZooKeeperPtr & zk, ClusterInfo & info) { if (info.current_node_is_observer) @@ -566,7 +1087,7 @@ void ClusterDiscovery::initialUpdate() throw Exception(ErrorCodes::KEEPER_EXCEPTION, "Failpoint cluster_discovery_faults is triggered"); }); - for (const auto & path : multicluster_discovery_paths) + for (const auto & [_, path] : multicluster_discovery_paths) { auto zk = context->getDefaultOrAuxiliaryZooKeeper(path.zk_name); @@ -576,17 +1097,27 @@ void ClusterDiscovery::initialUpdate() findDynamicClusters(clusters_info); - for (auto & [_, info] : clusters_info) + std::vector cluster_names; + cluster_names.reserve(clusters_info.size()); + for (const auto & [name, _] : clusters_info) + cluster_names.push_back(name); + + for (const auto & name : cluster_names) { + auto it = clusters_info.find(name); + if (it == clusters_info.end()) + continue; + + auto & info = it->second; auto zk = context->getDefaultOrAuxiliaryZooKeeper(info.zk_name); registerInZk(zk, info); if (!upsertCluster(info)) { - LOG_WARNING(log, "Error on initial cluster '{}' update, will retry in background", info.name); - clusters_to_update->set(info.name); + LOG_WARNING(log, "Error on initial cluster '{}' update, will retry in background", name); + clusters_to_update->set(name); } - else if (info.zk_root_index) - clusters_to_update->set(info.name, false); + else if (auto after = clusters_info.find(name); after != clusters_info.end() && after->second.isDynamic()) + clusters_to_update->set(name, false); } LOG_DEBUG(log, "Initialized"); @@ -607,18 +1138,17 @@ void ClusterDiscovery::unregisterAll() void ClusterDiscovery::findDynamicClusters( std::unordered_map & info, - std::unordered_set * unchanged_roots) + std::unordered_set * unchanged_roots) { using namespace std::chrono_literals; constexpr auto force_update_interval = 2min; - size_t zk_root_index = 0; - - for (const auto & path : multicluster_discovery_paths) + for (auto & [full_path, path] : multicluster_discovery_paths) { - ++zk_root_index; - + /// Cleared before Keeper I/O so a throw must restore the bit; otherwise the worker + /// may sleep forever with no children watch reinstalled. + bool cleared_need_update = false; if (unchanged_roots) { if (!path.need_update->exchange(false)) @@ -627,63 +1157,92 @@ void ClusterDiscovery::findDynamicClusters( bool force_update = path.watch.elapsedSeconds() > std::chrono::seconds(force_update_interval).count(); if (!force_update) { - unchanged_roots->insert(zk_root_index); + unchanged_roots->insert(full_path); continue; } } + else + cleared_need_update = true; } - auto zk = context->getDefaultOrAuxiliaryZooKeeper(path.zk_name); - - auto clusters = zk->getChildrenWatch( - path.zk_path, - nullptr, - Coordination::WatchCallbackPtrOrEventPtr{path.watch_callback, ProfileEvents::ZooKeeperWatchTriggeredClusterDiscovery}); - - for (const auto & cluster : clusters) + try { - auto p = clusters_info.find(cluster); - if (p != clusters_info.end() && !p->second.zk_root_index) + fiu_do_on(FailPoints::cluster_discovery_retry_signal_fail, { - /// Not a warning - node can register itsefs in one cluster and discover other clusters - LOG_TRACE(log, "Found dynamic duplicate of cluster '{}' in config and Keeper, skipped", cluster); - continue; - } + throw Exception( + ErrorCodes::KEEPER_EXCEPTION, + "Failpoint cluster_discovery_retry_signal_fail is triggered in findDynamicClusters"); + }); + + auto zk = context->getDefaultOrAuxiliaryZooKeeper(path.zk_name); + zk->createAncestors(path.zk_path); + zk->createIfNotExists(path.zk_path, ""); - if (info.contains(cluster)) + auto clusters = zk->getChildrenWatch( + path.zk_path, + nullptr, + Coordination::WatchCallbackPtrOrEventPtr{path.watch_callback, ProfileEvents::ZooKeeperWatchTriggeredClusterDiscovery}); + + for (const auto & cluster : clusters) { - /// Possible with several root paths, it's a configuration error - LOG_WARNING(log, "Found dynamic duplicate of cluster '{}' in Keeper, skipped record by path {}:{}", - cluster, path.zk_name, path.zk_path); - continue; + auto p = clusters_info.find(cluster); + if (p != clusters_info.end() && !p->second.isDynamic()) + { + /// Not a warning - node can register itsefs in one cluster and discover other clusters + LOG_TRACE(log, "Found dynamic duplicate of cluster '{}' in config and Keeper, skipped", cluster); + continue; + } + + if (info.contains(cluster)) + { + /// Possible with several root paths, it's a configuration error + LOG_WARNING(log, "Found dynamic duplicate of cluster '{}' in Keeper, skipped record by path {}:{}", + cluster, path.zk_name, path.zk_path); + continue; + } + + info.emplace( + cluster, + ClusterInfo( + /* name_= */ cluster, + /* zk_name_= */ path.zk_name, + /* zk_root_= */ path.zk_path + "/" + cluster, + /* host_name= */ "", + /* username= */ path.username, + /* password= */ path.password, + /* cluster_secret= */ path.cluster_secret, + /* port= */ context->getTCPPort(), + /* secure= */ path.is_secure_connection, + /* shard_id= */ 0, + /* observer_mode= */ true, + /* invisible= */ false, + /* multicluster_full_path_= */ full_path + ) + ); } - info.emplace( - cluster, - ClusterInfo( - /* name_= */ cluster, - /* zk_name_= */ path.zk_name, - /* zk_root_= */ path.zk_path + "/" + cluster, - /* host_name= */ "", - /* username= */ path.username, - /* password= */ path.password, - /* cluster_secret= */ path.cluster_secret, - /* port= */ context->getTCPPort(), - /* secure= */ path.is_secure_connection, - /* shard_id= */ 0, - /* observer_mode= */ true, - /* invisible= */ false, - /* zk_root_index= */ zk_root_index - ) - ); + path.watch.restart(); + } + catch (...) + { + if (cleared_need_update) + path.need_update->store(true); + throw; } - - path.watch.restart(); } } void ClusterDiscovery::start() { + std::lock_guard lock(start_mutex); + startImpl(); +} + +void ClusterDiscovery::startImpl() +{ + if (main_thread.joinable()) + return; + if (clusters_info.empty() && multicluster_discovery_paths.empty()) { LOG_DEBUG(log, "No defined clusters for discovery"); @@ -693,6 +1252,8 @@ void ClusterDiscovery::start() try { auto component_guard = Coordination::setCurrentComponent("ClusterDiscovery::start"); + /// Apply any queued reload before the first init attempt (same rationale as runMainThread). + consumePendingConfigUpdate(); initialUpdate(); } catch (...) @@ -722,8 +1283,25 @@ void ClusterDiscovery::start() * should not stop discovery forever */ tryLogCurrentException(log, "Caught exception in cluster discovery runMainThread"); + if (clusters_to_update->isStopped()) + break; + /// Flags::wait may have already cleared the only wake bit before the throw. + /// Re-arm so backoff retry does not sleep until an unrelated event. + clusters_to_update->wakeup(); + } + if (finish || clusters_to_update->isStopped()) + break; + + /// Interruptible backoff so shutdown does not wait for a long sleep after errors. + for (auto remaining = backoff_timeout; remaining.count() > 0 && !clusters_to_update->isStopped();) + { + constexpr auto slice = std::chrono::milliseconds(50); + auto step = remaining < slice ? remaining : slice; + std::this_thread::sleep_for(step); + remaining -= step; } - std::this_thread::sleep_for(backoff_timeout); + if (clusters_to_update->isStopped()) + break; backoff_timeout = std::min(backoff_timeout * 2, std::chrono::milliseconds(3min)); } }); @@ -741,6 +1319,12 @@ bool ClusterDiscovery::runMainThread(std::function up_to_date_callback) constexpr auto force_update_interval = 2min; + /// Pending reloads must be applied before retrying init. Otherwise a worker that keeps + /// failing initialUpdate (bad Keeper path, etc.) never reaches the loop body consumer and + /// updateFromConfig stays stuck while ensureWorkerStarted no-ops on the running thread. + consumePendingConfigUpdate(); + retryPendingUnregisters(); + if (!is_initialized) initialUpdate(); @@ -752,110 +1336,172 @@ bool ClusterDiscovery::runMainThread(std::function up_to_date_callback) if (finished) break; - std::unordered_map new_dynamic_clusters_info; - std::unordered_set unchanged_roots; - findDynamicClusters(new_dynamic_clusters_info, &unchanged_roots); + /// Snapshot of work acknowledged by wait(). Must be restored if this iteration throws + /// before watches / register commands are reinstalled; otherwise the worker blocks forever. + RegisterChangeFlag consumed_register_flag = RegisterChangeFlag::RCF_NONE; - std::unordered_set clusters_to_insert; - std::unordered_set clusters_to_remove; - - /// Remove clusters that are not found in new_dynamic_clusters_info - for (const auto & [cluster_name, info] : clusters_info) + try { - if (!info.zk_root_index) - continue; - if (!new_dynamic_clusters_info.erase(cluster_name) - && !unchanged_roots.contains(info.zk_root_index)) - clusters_to_remove.insert(cluster_name); - } - /// new_dynamic_clusters_info now contains only new clusters - for (const auto & [cluster_name, _] : new_dynamic_clusters_info) - clusters_to_insert.insert(cluster_name); - - for (const auto & cluster_name : clusters_to_remove) - removeCluster(cluster_name, /* is_dynamic_cluster */true); + fiu_do_on(FailPoints::cluster_discovery_retry_signal_fail, + { + throw Exception( + ErrorCodes::KEEPER_EXCEPTION, + "Failpoint cluster_discovery_retry_signal_fail is triggered after Flags::wait"); + }); - clusters_info.merge(new_dynamic_clusters_info); + consumePendingConfigUpdate(); - for (const auto & [cluster_name, need_update] : clusters) - { - auto cluster_info_it = clusters_info.find(cluster_name); - if (cluster_info_it == clusters_info.end()) + if (!retryPendingUnregisters()) { - LOG_ERROR(log, "Unknown cluster '{}'", cluster_name); - continue; + /// Keep waking the loop with a short interruptible backoff so ephemeral cleanup + /// retries without failing / rolling back the already-applied config update. + using namespace std::chrono_literals; + for (auto remaining = std::chrono::milliseconds(1000); + remaining.count() > 0 && !clusters_to_update->isStopped();) + { + constexpr auto slice = std::chrono::milliseconds(50); + auto step = remaining < slice ? remaining : slice; + std::this_thread::sleep_for(step); + remaining -= step; + } + if (!clusters_to_update->isStopped()) + clusters_to_update->set(); } - auto & cluster_info = cluster_info_it->second; - if (!need_update) + std::unordered_map new_dynamic_clusters_info; + std::unordered_set unchanged_roots; + findDynamicClusters(new_dynamic_clusters_info, &unchanged_roots); + + std::unordered_set clusters_to_insert; + std::unordered_set clusters_to_remove; + + /// Remove clusters that are not found in new_dynamic_clusters_info + for (const auto & [cluster_name, info] : clusters_info) { - /// force updating periodically - bool force_update = cluster_info.watch.elapsedSeconds() > std::chrono::seconds(force_update_interval).count(); - if (!force_update) + if (!info.isDynamic()) continue; + if (!new_dynamic_clusters_info.erase(cluster_name) + && !unchanged_roots.contains(info.multicluster_full_path)) + clusters_to_remove.insert(cluster_name); } + /// new_dynamic_clusters_info now contains only new clusters + for (const auto & [cluster_name, _] : new_dynamic_clusters_info) + clusters_to_insert.insert(cluster_name); - if (upsertCluster(cluster_info)) + for (const auto & cluster_name : clusters_to_remove) + removeDynamicCluster(cluster_name); + + clusters_info.merge(new_dynamic_clusters_info); + + for (const auto & [cluster_name, need_update] : clusters) { - cluster_info.watch.restart(); - LOG_DEBUG(log, "Cluster '{}' updated successfully", cluster_name); + auto cluster_info_it = clusters_info.find(cluster_name); + if (cluster_info_it == clusters_info.end()) + { + LOG_ERROR(log, "Unknown cluster '{}'", cluster_name); + /// Drop keys resurrected by late Keeper callbacks after removal. + clusters_to_update->remove(cluster_name); + continue; + } + + auto & cluster_info = cluster_info_it->second; + if (!need_update) + { + /// force updating periodically + bool force_update = cluster_info.watch.elapsedSeconds() > std::chrono::seconds(force_update_interval).count(); + if (!force_update) + continue; + } + + String name = cluster_name; + if (upsertCluster(cluster_info)) + { + cluster_info_it = clusters_info.find(name); + if (cluster_info_it != clusters_info.end()) + cluster_info_it->second.watch.restart(); + LOG_DEBUG(log, "Cluster '{}' updated successfully", name); + } + else + { + all_up_to_date = false; + /// no need to trigger convar, will retry after timeout in `wait` + clusters_to_update->set(name); + LOG_WARNING(log, "Cluster '{}' wasn't updated, will retry", name); + } } - else + + for (const auto & cluster_name : clusters_to_insert) { - all_up_to_date = false; - /// no need to trigger convar, will retry after timeout in `wait` - clusters_to_update->set(cluster_name); - LOG_WARNING(log, "Cluster '{}' wasn't updated, will retry", cluster_name); + auto cluster_info_it = clusters_info.find(cluster_name); + if (cluster_info_it == clusters_info.end()) + { + LOG_ERROR(log, "Unknown dynamic cluster '{}'", cluster_name); + clusters_to_update->remove(cluster_name); + continue; + } + auto & cluster_info = cluster_info_it->second; + String name = cluster_name; + if (upsertCluster(cluster_info)) + { + cluster_info_it = clusters_info.find(name); + if (cluster_info_it != clusters_info.end()) + cluster_info_it->second.watch.restart(); + LOG_DEBUG(log, "Dynamic cluster '{}' inserted successfully", name); + } + else + { + all_up_to_date = false; + /// no need to trigger convar, will retry after timeout in `wait` + clusters_to_update->set(name); + LOG_WARNING(log, "Dynamic cluster '{}' wasn't inserted, will retry", name); + } } - } - for (const auto & cluster_name : clusters_to_insert) - { - auto cluster_info_it = clusters_info.find(cluster_name); - if (cluster_info_it == clusters_info.end()) + if (all_up_to_date) { - LOG_ERROR(log, "Unknown dynamic cluster '{}'", cluster_name); - continue; + up_to_date_callback(); } - auto & cluster_info = cluster_info_it->second; - if (upsertCluster(cluster_info)) + + consumed_register_flag = register_change_flag.exchange(RegisterChangeFlag::RCF_NONE); + + if (consumed_register_flag == RegisterChangeFlag::RCF_REGISTER_ALL) { - cluster_info.watch.restart(); - LOG_DEBUG(log, "Dynamic cluster '{}' inserted successfully", cluster_name); + LOG_DEBUG(log, "Register in all dynamic clusters"); + for (auto & [_, info] : clusters_info) + { + auto zk = context->getDefaultOrAuxiliaryZooKeeper(info.zk_name); + registerInZk(zk, info); + } } - else + else if (consumed_register_flag == RegisterChangeFlag::RCF_UNREGISTER_ALL) { - all_up_to_date = false; - /// no need to trigger convar, will retry after timeout in `wait` - clusters_to_update->set(cluster_name); - LOG_WARNING(log, "Dynamic cluster '{}' wasn't inserted, will retry", cluster_name); + LOG_DEBUG(log, "Unregister in all dynamic clusters"); + for (auto & [_, info] : clusters_info) + { + auto zk = context->getDefaultOrAuxiliaryZooKeeper(info.zk_name); + unregisterFromZk(zk, info); + } } - } - if (all_up_to_date) - { - up_to_date_callback(); + consumed_register_flag = RegisterChangeFlag::RCF_NONE; } - - RegisterChangeFlag flag = register_change_flag.exchange(RegisterChangeFlag::RCF_NONE); - - if (flag == RegisterChangeFlag::RCF_REGISTER_ALL) + catch (...) { - LOG_DEBUG(log, "Register in all dynamic clusters"); - for (auto & [_, info] : clusters_info) + for (const auto & [cluster_name, need_update] : clusters) { - auto zk = context->getDefaultOrAuxiliaryZooKeeper(info.zk_name); - registerInZk(zk, info); + if (need_update) + clusters_to_update->set(cluster_name); } - } - else if (flag == RegisterChangeFlag::RCF_UNREGISTER_ALL) - { - LOG_DEBUG(log, "Unregister in all dynamic clusters"); - for (auto & [_, info] : clusters_info) + + if (consumed_register_flag != RegisterChangeFlag::RCF_NONE) { - auto zk = context->getDefaultOrAuxiliaryZooKeeper(info.zk_name); - unregisterFromZk(zk, info); + /// Do not overwrite a newer registerAll/unregisterAll posted while we failed. + RegisterChangeFlag expected = RegisterChangeFlag::RCF_NONE; + register_change_flag.compare_exchange_strong(expected, consumed_register_flag); } + + clusters_to_update->wakeup(); + throw; } } LOG_DEBUG(log, "Worker thread stopped"); @@ -881,12 +1527,65 @@ std::unordered_map ClusterDiscovery::getClusters() const void ClusterDiscovery::shutdown() { LOG_DEBUG(log, "Shutting down"); - clusters_to_update->stop(); + if (clusters_to_update) + clusters_to_update->stop(); + /// Wait for any in-flight startImpl() before joining so we do not race ThreadFromGlobalPool assign. + std::lock_guard lock(start_mutex); if (main_thread.joinable()) main_thread.join(); } +void ClusterDiscovery::disableAndShutdown() +{ + LOG_DEBUG(log, "Disabling cluster discovery"); + + /// Stop the worker before touching clusters_info so upsert cannot re-register. + shutdown(); + + /// Config-reloader thread has no ZooKeeper component; unregister requires one. + auto component_guard = Coordination::setCurrentComponent("ClusterDiscovery::disableAndShutdown"); + + for (const auto & [name, info] : clusters_info) + { + if (info.current_node_is_observer) + continue; + if (!unregisterFromZk(info.zk_name, info.zk_root, name)) + { + LOG_WARNING( + log, + "Failed to unregister current node from cluster '{}' while disabling discovery", + name); + } + } + + for (const auto & pending : pending_zk_unregisters) + { + if (!unregisterFromZk(pending.zk_name, pending.zk_root, pending.cluster_name)) + { + LOG_WARNING( + log, + "Failed to complete pending unregister for cluster '{}' while disabling discovery", + pending.cluster_name); + } + } + pending_zk_unregisters.clear(); + + clusters_info.clear(); + multicluster_discovery_paths.clear(); + get_nodes_callbacks.clear(); + register_change_flag.store(RegisterChangeFlag::RCF_NONE); + + { + std::lock_guard lock(pending_config_mutex); + pending_config_update.reset(); + } + { + std::lock_guard lock(mutex); + cluster_impls.clear(); + } +} + ClusterDiscovery::~ClusterDiscovery() { try diff --git a/src/Interpreters/ClusterDiscovery.h b/src/Interpreters/ClusterDiscovery.h index 3965cd0d8e37..06bfb815d57c 100644 --- a/src/Interpreters/ClusterDiscovery.h +++ b/src/Interpreters/ClusterDiscovery.h @@ -9,7 +9,10 @@ #include +#include #include +#include +#include namespace DB { @@ -33,9 +36,26 @@ class ClusterDiscovery void start(); + /// Apply changes from reloaded remote_servers config (credentials, add/remove discovery paths). + /// Safe to call from the config-reloader thread; the update is applied on the discovery worker. + void updateFromConfig( + const Poco::Util::AbstractConfiguration & config, + const String & config_prefix = "remote_servers"); + + /// Throws if discovery subtrees under `config_prefix` are invalid. No side effects. + /// Call before committing Clusters / clusters_config so a bad reload cannot partially apply. + static void validateConfig( + const Poco::Util::AbstractConfiguration & config, + ContextPtr context, + const String & config_prefix = "remote_servers"); + ClusterPtr getCluster(const String & cluster_name) const; std::unordered_map getClusters() const; + /// Stop the worker, remove participant ephemerals, and drop published clusters. + /// Used when allow_experimental_cluster_discovery is turned off on reload. + void disableAndShutdown(); + ~ClusterDiscovery(); void registerAll(); @@ -92,9 +112,11 @@ class ClusterDiscovery String password; String cluster_secret; - /// For dynamic clusters, index+1 in multicluster_discovery_paths where cluster was found - /// 0 for static clusters - size_t zk_root_index; + /// For dynamic clusters: MulticlusterDiscovery::getFullPath() where cluster was found. + /// Empty for static clusters defined with . + String multicluster_full_path; + + bool isDynamic() const { return !multicluster_full_path.empty(); } ClusterInfo(const String & name_, const String & zk_name_, @@ -108,21 +130,128 @@ class ClusterDiscovery size_t shard_id, bool observer_mode, bool invisible, - size_t zk_root_index_ = 0 + const String & multicluster_full_path_ = {} ); }; + struct ParsedStaticDiscovery + { + String name; + String zk_name; + String zk_root; + String host_name; + String username; + String password; + String cluster_secret; + bool secure = false; + size_t shard_id = 0; + bool observer = false; + bool invisible = false; + }; + + struct ParsedMulticlusterDiscovery + { + String zk_name; + String zk_path; + bool is_secure_connection = false; + String username; + String password; + String cluster_secret; + + String getFullPath() const { return zk_name + ":" + zk_path; } + }; + + struct ParsedDiscoveryConfig + { + std::vector static_clusters; + std::vector multicluster_roots; + }; + + struct MulticlusterDiscovery + { + const String zk_name; + const String zk_path; + bool is_secure_connection; + String username; + String password; + String cluster_secret; + + mutable Stopwatch watch; + mutable std::shared_ptr need_update; + Coordination::WatchCallbackPtr watch_callback; + + MulticlusterDiscovery(const String & zk_name_, + const String & zk_path_, + bool is_secure_connection_, + const String & username_, + const String & password_, + const String & cluster_secret_) + : zk_name(zk_name_) + , zk_path(zk_path_) + , is_secure_connection(is_secure_connection_) + , username(username_) + , password(password_) + , cluster_secret(cluster_secret_) + , need_update(std::make_shared(true)) + {} + + String getFullPath() const { return zk_name + ":" + zk_path; } + }; + + static ParsedDiscoveryConfig parseDiscoveryConfig( + const Poco::Util::AbstractConfiguration & config, + ContextPtr context, + const String & config_prefix); + + void applyParsedConfig(ParsedDiscoveryConfig && parsed); + void addStaticCluster(ParsedStaticDiscovery && parsed); + void removeStaticCluster(const String & name); + /// Remove a multicluster-discovered cluster so a static config entry can take its name. + void removeDynamicCluster(const String & name); + bool updateStaticClusterFields(ClusterInfo & info, const ParsedStaticDiscovery & parsed); + void addMulticlusterRoot(ParsedMulticlusterDiscovery && parsed); + void removeMulticlusterRoot(const String & full_path); + bool updateMulticlusterRootFields(MulticlusterDiscovery & path, const ParsedMulticlusterDiscovery & parsed); + /// Force findDynamicClusters to rescan roots (e.g. after a static name stops shadowing). + void markMulticlusterRootsNeedUpdate(); + + void rebuildClusterObject(const ClusterInfo & info); + void ensureWorkerStarted(); + bool consumePendingConfigUpdate(); + + /// Assumes start_mutex is held. Starts the worker at most once. + void startImpl(); + void initialUpdate(); void registerInZk(zkutil::ZooKeeperPtr & zk, ClusterInfo & info); void unregisterFromZk(zkutil::ZooKeeperPtr & zk, ClusterInfo & info); + struct PendingZkUnregister + { + String zk_name; + String zk_root; + String cluster_name; + }; + + /// Returns false if Keeper remove failed; caller should queue a retry. + bool unregisterFromZk(const String & zk_name, const String & zk_root, const String & cluster_name); + /// True if a non-observer entry still owns this Keeper registration path. + /// When removing `exclude_cluster_name`, pass it so the cluster being dropped is ignored. + bool pathHasActiveParticipant( + const String & zk_name, + const String & zk_root, + const String * exclude_cluster_name = nullptr) const; + /// Retries failed unregisters. Returns true when the queue is empty. + /// Drops pending entries whose path already has an active participant again. + bool retryPendingUnregisters(); + Strings getNodeNames(zkutil::ZooKeeperPtr & zk, const String & zk_root, const String & cluster_name, int * version, bool set_callback, - size_t zk_root_index); + const String & multicluster_full_path); NodesInfo getNodes(zkutil::ZooKeeperPtr & zk, const String & zk_root, const Strings & node_uuids); @@ -135,9 +264,12 @@ class ClusterDiscovery bool runMainThread(std::function up_to_date_callback); void shutdown(); - void findDynamicClusters(std::unordered_map & info, std::unordered_set * unchanged_roots = nullptr); + void findDynamicClusters( + std::unordered_map & info, + std::unordered_set * unchanged_roots = nullptr); /// cluster name -> cluster info (zk root, set of nodes) + /// Mutated only from constructor (before start) and the discovery worker thread. std::unordered_map clusters_info; ContextMutablePtr context; @@ -160,42 +292,28 @@ class ClusterDiscovery std::unordered_map cluster_impls; bool is_initialized = false; + + /// Serializes start() / ensureWorkerStarted() so concurrent config reload and + /// startClusterDiscovery cannot double-assign main_thread (ThreadFromGlobalPool aborts). + /// Lock order: start_mutex before pending_config_mutex. + mutable std::mutex start_mutex; ThreadFromGlobalPool main_thread; LoggerPtr log; - struct MulticlusterDiscovery - { - const String zk_name; - const String zk_path; - bool is_secure_connection; - String username; - String password; - String cluster_secret; - - mutable Stopwatch watch; - mutable std::shared_ptr need_update; - Coordination::WatchCallbackPtr watch_callback; + /// Keyed by MulticlusterDiscovery::getFullPath() + std::unordered_map multicluster_discovery_paths; - MulticlusterDiscovery(const String & zk_name_, - const String & zk_path_, - bool is_secure_connection_, - const String & username_, - const String & password_, - const String & cluster_secret_) - : zk_name(zk_name_) - , zk_path(zk_path_) - , is_secure_connection(is_secure_connection_) - , username(username_) - , password(password_) - , cluster_secret(cluster_secret_) - , need_update(std::make_shared(true)) - {} - - String getFullPath() const { return zk_name + ":" + zk_path; } - }; + /// Config reload posts parsed config here; worker applies it. + /// Never take this lock while a caller without start_mutex may later take start_mutex + /// while holding this one: lock order is start_mutex -> pending_config_mutex. + mutable std::mutex pending_config_mutex; + std::optional pending_config_update; - std::vector multicluster_discovery_paths; + /// Ephemeral registrations that failed to remove during config apply. + /// Local cluster state is already dropped; retry from the worker thread. + /// Accessed only from the discovery worker (same as clusters_info). + std::vector pending_zk_unregisters; MultiVersion::Version macros; diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 0e6c02b1dbbb..aa2b65303b50 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -6094,12 +6094,12 @@ void Context::startClusterDiscovery() /// On repeating calls updates existing clusters and adds new clusters, doesn't delete old clusters void Context::setClustersConfig(const ConfigurationPtr & config, bool enable_discovery, const String & config_name) { + ClusterDiscovery * discovery_to_update = nullptr; + ClusterDiscovery * discovery_just_created_ptr = nullptr; + std::unique_ptr discovery_to_disable; + bool clusters_changed = false; { std::lock_guard lock(shared->clusters_mutex); - if (ConfigHelper::getBool(*config, "allow_experimental_cluster_discovery") && enable_discovery && !shared->cluster_discovery) - { - shared->cluster_discovery = std::make_unique(*config, getGlobalContext(), getMacros()); - } /// Do not update clusters if this part of config wasn't changed. /// Note: clusters_config must be checked for null separately from clusters, because @@ -6107,19 +6107,76 @@ void Context::setClustersConfig(const ConfigurationPtr & config, bool enable_dis /// shared->clusters using the fallback getConfigRef() without setting shared->clusters_config. /// If setClustersConfig() then runs before the config reloader stores its ConfigurationPtr, /// dereferencing shared->clusters_config would throw Poco::NullPointerException. - if (shared->clusters && shared->clusters_config && isSameConfiguration(*config, *shared->clusters_config, config_name)) - return; + /// + /// Still start a discovery object created after server start when only the allow-flag + /// flipped (remote_servers subtree unchanged) — otherwise the worker never runs. + /// The reverse transition (allow 1 -> 0) must tear discovery down even when remote_servers + /// is unchanged; otherwise the worker stays registered until restart. + const bool remote_servers_unchanged + = shared->clusters && shared->clusters_config + && isSameConfiguration(*config, *shared->clusters_config, config_name); + + const bool discovery_enabled + = ConfigHelper::getBool(*config, "allow_experimental_cluster_discovery") && enable_discovery; + + /// Validate discovery before creating the object or committing Clusters so a bad reload + /// cannot leave clusters_config advanced while discovery stays on the previous view. + /// Also validate when allow is turned off: an existing ClusterDiscovery is still updated. + if (!remote_servers_unchanged && (discovery_enabled || shared->cluster_discovery)) + ClusterDiscovery::validateConfig(*config, getGlobalContext(), config_name); + + bool discovery_just_created = false; + if (discovery_enabled) + { + if (!shared->cluster_discovery) + { + shared->cluster_discovery = std::make_unique(*config, getGlobalContext(), getMacros()); + discovery_just_created = true; + } + } + else if (shared->cluster_discovery) + { + discovery_to_disable = std::move(shared->cluster_discovery); + } - auto old_clusters_config = shared->clusters_config; - shared->clusters_config = config; + if (!remote_servers_unchanged) + { + auto old_clusters_config = shared->clusters_config; + shared->clusters_config = config; - if (!shared->clusters) - shared->clusters = std::make_shared(*shared->clusters_config, *settings, getMacros(), config_name); - else - shared->clusters->updateClusters(*shared->clusters_config, *settings, config_name, old_clusters_config); + if (!shared->clusters) + shared->clusters = std::make_shared(*shared->clusters_config, *settings, getMacros(), config_name); + else + shared->clusters->updateClusters(*shared->clusters_config, *settings, config_name, old_clusters_config); + + if (shared->cluster_discovery && !discovery_just_created) + discovery_to_update = shared->cluster_discovery.get(); + + ++shared->clusters_version; + clusters_changed = true; + } - ++shared->clusters_version; + /// Constructor already applied config. Start outside this lock if the server is ready; + /// otherwise programs/server/Server.cpp calls startClusterDiscovery() after listen. + if (discovery_just_created) + discovery_just_created_ptr = shared->cluster_discovery.get(); } + + /// Tear down outside clusters_mutex: joins the worker and may touch ZooKeeper. + if (discovery_to_disable) + discovery_to_disable->disableAndShutdown(); + + /// Apply discovery updates outside clusters_mutex: may start the worker and touch ZooKeeper. + if (discovery_to_update) + discovery_to_update->updateFromConfig(*config, config_name); + + /// Re-check server readiness without clusters_mutex (isServerCompletelyStarted takes shared->mutex). + if (discovery_just_created_ptr && getApplicationType() == ApplicationType::SERVER && isServerCompletelyStarted()) + discovery_just_created_ptr->start(); + + /// Avoid DDL host-id refresh / log noise when remote_servers (and discovery) did not change. + /// Still notify when discovery was just created or disabled (e.g. allow-flag-only reload). + if (clusters_changed || discovery_to_update || discovery_just_created_ptr || discovery_to_disable) { SharedLockGuard lock(shared->mutex); if (shared->ddl_worker) diff --git a/src/Interpreters/tests/gtest_cluster_discovery_flags.cpp b/src/Interpreters/tests/gtest_cluster_discovery_flags.cpp new file mode 100644 index 000000000000..34343ca2ccfc --- /dev/null +++ b/src/Interpreters/tests/gtest_cluster_discovery_flags.cpp @@ -0,0 +1,77 @@ +#include + +#include +#include +#include +#include + +namespace +{ + +/// Mirrors ClusterDiscovery::Flags set / setIfPresent / remove contract used by watch callbacks. +template +class UpdateFlags +{ +public: + void set(const T & key, bool value = true) + { + std::unique_lock lk(mu); + flags[key] = value; + any_need_update |= value; + } + + void setIfPresent(const T & key, bool value = true) + { + std::unique_lock lk(mu); + auto it = flags.find(key); + if (it == flags.end()) + return; + it->second = value; + any_need_update |= value; + } + + void remove(const T & key) + { + std::unique_lock lk(mu); + flags.erase(key); + } + + bool contains(const T & key) const + { + std::unique_lock lk(mu); + return flags.contains(key); + } + +private: + mutable std::mutex mu; + std::unordered_map flags; + bool any_need_update = true; +}; + +} + +TEST(ClusterDiscoveryFlags, SetIfPresentDoesNotResurrectRemovedKey) +{ + UpdateFlags flags; + flags.set("gone"); + ASSERT_TRUE(flags.contains("gone")); + + flags.remove("gone"); + ASSERT_FALSE(flags.contains("gone")); + + /// Late Keeper callback after removal. + flags.setIfPresent("gone"); + EXPECT_FALSE(flags.contains("gone")); + + /// Intentional re-registration may insert again. + flags.set("gone"); + EXPECT_TRUE(flags.contains("gone")); +} + +TEST(ClusterDiscoveryFlags, SetIfPresentUpdatesExistingKey) +{ + UpdateFlags flags; + flags.set("alive", false); + flags.setIfPresent("alive", true); + EXPECT_TRUE(flags.contains("alive")); +} diff --git a/src/Interpreters/tests/gtest_cluster_discovery_start.cpp b/src/Interpreters/tests/gtest_cluster_discovery_start.cpp new file mode 100644 index 000000000000..eee373d5ad1c --- /dev/null +++ b/src/Interpreters/tests/gtest_cluster_discovery_start.cpp @@ -0,0 +1,80 @@ +#include +#include +#include +#include + +#include +#include + +#include + +#include +#include +#include +#include + +using namespace DB; + +namespace +{ + +Poco::AutoPtr makeDiscoveryConfig() +{ + /// Observer mode avoids ephemeral registration; initialUpdate may still fail without ZooKeeper, + /// which is fine — startImpl() still assigns the worker thread after catching the exception. + std::istringstream config_stream{R"( + + + + + /clickhouse/discovery/test_cluster_concurrent_start + + + + + + )"}; + return new Poco::Util::XMLConfiguration(config_stream); +} + +} + +/// Regression: concurrent start() / updateFromConfig (ensureWorkerStarted) must not +/// double-assign ThreadFromGlobalPool (which aborts if already initialized). +TEST(ClusterDiscovery, ConcurrentStartDoesNotAbort) +{ + ServerUUID::setRandomForUnitTests(); + + auto context = Context::createCopy(getContext().context); + auto config = makeDiscoveryConfig(); + auto discovery = std::make_unique(*config, context, context->getMacros()); + + constexpr size_t num_threads = 8; + constexpr size_t iterations = 40; + std::atomic started_calls{0}; + + std::vector threads; + threads.reserve(num_threads); + for (size_t i = 0; i < num_threads; ++i) + { + threads.emplace_back([&] + { + for (size_t j = 0; j < iterations; ++j) + { + if ((j % 2) == 0) + discovery->start(); + else + discovery->updateFromConfig(*config); + started_calls.fetch_add(1, std::memory_order_relaxed); + } + }); + } + + for (auto & t : threads) + t.join(); + + EXPECT_EQ(started_calls.load(), num_threads * iterations); + + /// Destructor joins the worker; surviving to here means no abort on double-start. + discovery.reset(); +} diff --git a/tests/integration/test_cluster_discovery/config/config_discovery_disabled.xml b/tests/integration/test_cluster_discovery/config/config_discovery_disabled.xml new file mode 100644 index 000000000000..194579f950bc --- /dev/null +++ b/tests/integration/test_cluster_discovery/config/config_discovery_disabled.xml @@ -0,0 +1,5 @@ + + 0 + + + diff --git a/tests/integration/test_cluster_discovery/config/config_discovery_disabled_with_path.xml b/tests/integration/test_cluster_discovery/config/config_discovery_disabled_with_path.xml new file mode 100644 index 000000000000..0442acdc7d40 --- /dev/null +++ b/tests/integration/test_cluster_discovery/config/config_discovery_disabled_with_path.xml @@ -0,0 +1,11 @@ + + + 0 + + + + /clickhouse/discovery/test_enable_allow_only + + + + diff --git a/tests/integration/test_cluster_discovery/config/config_reload_discovery.xml b/tests/integration/test_cluster_discovery/config/config_reload_discovery.xml new file mode 100644 index 000000000000..acb57ca89084 --- /dev/null +++ b/tests/integration/test_cluster_discovery/config/config_reload_discovery.xml @@ -0,0 +1,12 @@ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + user1 + password123 + + + + diff --git a/tests/integration/test_cluster_discovery/test_config_reload.py b/tests/integration/test_cluster_discovery/test_config_reload.py new file mode 100644 index 000000000000..3df61176f3f3 --- /dev/null +++ b/tests/integration/test_cluster_discovery/test_config_reload.py @@ -0,0 +1,1060 @@ +import time + +import pytest + +from helpers.cluster import ClickHouseCluster + +from .common import check_on_cluster + +cluster = ClickHouseCluster(__file__) + +nodes = { + "node0": cluster.add_instance( + "node0", + main_configs=["config/config_reload_discovery.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), + "node1": cluster.add_instance( + "node1", + main_configs=["config/config_reload_discovery.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), +} + +CONFIG_PATH = "/etc/clickhouse-server/config.d/config_reload_discovery.xml" + +CONFIG_WITH_PWD = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + user1 + password123 + + + + +""" + +CONFIG_WITH_WRONG_PWD = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + user1 + wrongpass1234 + + + + +""" + +CONFIG_PASSWORD_AND_SECRET = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + user1 + password123 + cluster_secret_value + + + + + + 127.0.0.1 + 9000 + + + + + +""" + +CONFIG_PASSWORD_AND_SECRET_ALLOW_OFF = """ + + 0 + + + + /clickhouse/discovery/test_reload_cluster + user1 + password123 + cluster_secret_value + + + + + + 127.0.0.1 + 9000 + + + + + +""" + +CONFIG_NO_DISCOVERY = """ + + 1 + + + +""" + +CONFIG_WITH_CLUSTER_B = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster_b + + + + +""" + +CONFIG_MULTICLUSTER_ROOT = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + + + + + + /clickhouse/discovery + + + + +""" + +CONFIG_NO_MULTICLUSTER_ROOT = """ + + 1 + + + + /clickhouse/discovery/test_reload_cluster + + + + +""" + +CONFIG_PARTICIPANT = """ + + 1 + + + + /clickhouse/discovery/test_observer_transition + + + + +""" + +CONFIG_OBSERVER = """ + + 1 + + + + /clickhouse/discovery/test_observer_transition + + + + + +""" + +CONFIG_INVISIBLE = """ + + 1 + + + + /clickhouse/discovery/test_invisible_transition + + + + + +""" + +CONFIG_VISIBLE = """ + + 1 + + + + /clickhouse/discovery/test_invisible_transition + + + + +""" + + +@pytest.fixture(scope="module") +def start_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def wait_cluster_query(node, cluster_name, password="passwordAbc", should_succeed=True, retries=10): + query = ( + f"SELECT sum(number) FROM clusterAllReplicas('{cluster_name}', numbers(3)) " + f"GROUP BY hostname()" + ) + last_error = "" + for retry in range(retries): + if should_succeed: + try: + result = node.query(query, password=password) + if result.count("\n") >= 2: + return result + except Exception as e: + last_error = str(e) + else: + try: + error = node.query_and_get_error(query, password=password) + if "Authentication failed" in error or error: + return error + except Exception as e: + last_error = str(e) + time.sleep(1 + retry) + raise AssertionError( + f"wait_cluster_query failed (should_succeed={should_succeed}): {last_error}" + ) + + +def reload_config_on_all(config_body): + for node in nodes.values(): + node.replace_config(CONFIG_PATH, config_body) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + +def reload_config_on_node(node, config_body): + node.replace_config(CONFIG_PATH, config_body) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + +def test_reload_static_discovery_ownership_transitions(start_cluster): + """Static ↔ discovery must not leave a stale Clusters::impl entry that shadows discovery.""" + cluster_name = "test_ownership_cluster" + config_static = f""" + + 1 + + <{cluster_name}> + + + 127.0.0.1 + 9000 + + + + + +""" + config_discovery = f""" + + 1 + + <{cluster_name}> + + /clickhouse/discovery/{cluster_name} + + + + +""" + + def host_names(): + return [ + node.query( + f"SELECT groupArray(host_name) FROM system.clusters WHERE cluster = '{cluster_name}'", + password="passwordAbc", + ).strip() + for node in nodes.values() + ] + + def wait_hosts_contain(needle, msg, retries=10): + for _ in range(retries): + hosts = host_names() + if all(needle in h for h in hosts): + return + time.sleep(1) + raise AssertionError(f"{msg}: {hosts}") + + def wait_cluster_absent(msg, retries=10): + for _ in range(retries): + counts = [ + int( + node.query( + f"SELECT count() FROM system.clusters WHERE cluster = '{cluster_name}'", + password="passwordAbc", + ) + ) + for node in nodes.values() + ] + if all(c == 0 for c in counts): + return + time.sleep(1) + raise AssertionError(f"{msg}: {counts}") + + # Static only — placeholder host must be visible. + reload_config_on_all(config_static) + wait_hosts_contain("127.0.0.1", "Static ownership not applied") + + # Static → discovery: discovery must win (not keep 127.0.0.1 from impl). + reload_config_on_all(config_discovery) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name=cluster_name, + what="count()", + msg="Discovery ownership not applied after static→discovery", + query_params={"password": "passwordAbc"}, + retries=6, + ) + for hosts in host_names(): + if "127.0.0.1" in hosts: + raise AssertionError( + f"Stale static Cluster still shadows discovery after reload: {hosts}" + ) + + # Discovery → static. + reload_config_on_all(config_static) + wait_hosts_contain("127.0.0.1", "Static ownership not restored after discovery→static") + + # Static → removed. + reload_config_on_all(CONFIG_NO_DISCOVERY) + wait_cluster_absent("Cluster still present after removal") + + # static → discovery → removed (skip return to static). + reload_config_on_all(config_static) + wait_hosts_contain("127.0.0.1", "Static ownership not applied before second discovery cycle") + reload_config_on_all(config_discovery) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name=cluster_name, + what="count()", + msg="Discovery ownership not applied on second cycle", + query_params={"password": "passwordAbc"}, + retries=6, + ) + reload_config_on_all(CONFIG_NO_DISCOVERY) + wait_cluster_absent("Cluster still present after discovery→removed") + + +def test_reload_discovery_credentials(start_cluster): + reload_config_on_all(CONFIG_WITH_PWD) + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Wrong nodes count after credential config apply", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + reload_config_on_all(CONFIG_WITH_WRONG_PWD) + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=False) + + reload_config_on_all(CONFIG_WITH_PWD) + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + +def test_reload_invalid_discovery_does_not_partially_apply(start_cluster): + """Invalid discovery must fail the reload before Clusters / discovery diverge.""" + reload_config_on_all(CONFIG_WITH_PWD) + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_PASSWORD_AND_SECRET) + error = node.query_and_get_error("SYSTEM RELOAD CONFIG", password="passwordAbc") + assert "password" in error and "secret" in error, error + + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + for node in nodes.values(): + count = int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_partial_apply_marker'", + password="passwordAbc", + ) + ) + assert count == 0, "Static cluster from rejected config was partially applied" + + reload_config_on_all(CONFIG_WITH_PWD) + + +def test_reload_invalid_discovery_allow_off_does_not_partially_apply(start_cluster): + """Existing discovery must still validate when allow is turned off on reload.""" + reload_config_on_all(CONFIG_WITH_PWD) + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_PASSWORD_AND_SECRET_ALLOW_OFF) + error = node.query_and_get_error("SYSTEM RELOAD CONFIG", password="passwordAbc") + assert "password" in error and "secret" in error, error + + wait_cluster_query(nodes["node0"], "test_reload_cluster", should_succeed=True) + + for node in nodes.values(): + count = int( + node.query( + "SELECT count() FROM system.clusters " + "WHERE cluster = 'test_partial_apply_marker_allow_off'", + password="passwordAbc", + ) + ) + assert count == 0, "Static cluster from rejected allow=0 config was partially applied" + + reload_config_on_all(CONFIG_WITH_PWD) + + +def test_reload_add_remove_discovery_cluster(start_cluster): + reload_config_on_all(CONFIG_NO_DISCOVERY) + time.sleep(2) + + for node in nodes.values(): + count = int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster_b'", + password="passwordAbc", + ) + ) + assert count == 0 + + reload_config_on_all(CONFIG_WITH_CLUSTER_B) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster_b", + what="count()", + msg="Cluster was not added after config reload", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + reload_config_on_all(CONFIG_NO_DISCOVERY) + for retry in range(10): + counts = [ + int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster_b'", + password="passwordAbc", + ) + ) + for node in nodes.values() + ] + if all(c == 0 for c in counts): + break + time.sleep(1) + else: + raise AssertionError(f"Cluster was not removed after config reload: {counts}") + + +def test_reload_remove_retries_failed_unregister(start_cluster): + """Failed Keeper unregister is retried; re-add of the same path must not be undone by that retry.""" + reload_config_on_all(CONFIG_WITH_PWD) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Cluster not ready before unregister-retry test", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + node0 = nodes["node0"] + node1 = nodes["node1"] + + def enable_unregister_failpoint(): + node0.query( + "SYSTEM ENABLE FAILPOINT cluster_discovery_unregister_fail", + password="passwordAbc", + ) + + def disable_unregister_failpoint(): + node0.query( + "SYSTEM DISABLE FAILPOINT cluster_discovery_unregister_fail", + password="passwordAbc", + ) + + def wait_local_cluster_gone(): + for _ in range(10): + count = int( + node0.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster'", + password="passwordAbc", + ) + ) + if count == 0: + return + time.sleep(1) + raise AssertionError("node0 still exposes removed discovery cluster after reload") + + # --- remove while unregister fails, then retry cleanup after failpoint is cleared --- + enable_unregister_failpoint() + try: + reload_config_on_node(node0, CONFIG_NO_DISCOVERY) + wait_local_cluster_gone() + + for _ in range(10): + hosts = int( + node1.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster'", + password="passwordAbc", + ) + ) + if hosts == len(nodes): + break + time.sleep(1) + else: + raise AssertionError( + "Expected node0 ephemeral to remain visible on node1 while unregister failpoint is on" + ) + finally: + disable_unregister_failpoint() + + for _ in range(20): + hosts = int( + node1.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster'", + password="passwordAbc", + ) + ) + if hosts == 1: + break + time.sleep(1) + else: + raise AssertionError( + f"node0 ephemeral was not cleaned up after unregister retry; hosts on node1={hosts}" + ) + + reload_config_on_all(CONFIG_WITH_PWD) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Cluster not restored before remove/re-add unregister test", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + # --- remove, re-add same path while unregister still failing; pending retry must not drop the node --- + enable_unregister_failpoint() + try: + reload_config_on_node(node0, CONFIG_NO_DISCOVERY) + wait_local_cluster_gone() + + reload_config_on_node(node0, CONFIG_WITH_PWD) + check_on_cluster( + [node0, node1], + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Cluster was not restored on node0 after re-add", + query_params={"password": "passwordAbc"}, + retries=6, + ) + finally: + disable_unregister_failpoint() + + for _ in range(15): + hosts = int( + node1.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster'", + password="passwordAbc", + ) + ) + if hosts != len(nodes): + raise AssertionError( + f"Pending unregister deleted re-registered ephemeral; hosts on node1={hosts}" + ) + time.sleep(1) + + reload_config_on_all(CONFIG_WITH_PWD) + + +def test_reload_add_remove_multicluster_root(start_cluster): + reload_config_on_all(CONFIG_MULTICLUSTER_ROOT) + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Static discovery cluster missing", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + # Observer root should discover the static cluster under /clickhouse/discovery + for retry in range(15): + counts = [ + int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_reload_cluster'", + password="passwordAbc", + ) + ) + for node in nodes.values() + ] + if all(c == len(nodes) for c in counts): + break + time.sleep(1) + + reload_config_on_all(CONFIG_NO_MULTICLUSTER_ROOT) + # Static cluster must remain after multicluster root removal + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Static cluster disappeared after multicluster root removal", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + reload_config_on_all(CONFIG_MULTICLUSTER_ROOT) + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_reload_cluster", + what="count()", + msg="Static cluster missing after restoring multicluster root", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + +def test_reload_participant_to_observer_unregisters(start_cluster): + """Participant -> observer reload must remove this node's ephemeral ZK registration.""" + reload_config_on_all(CONFIG_PARTICIPANT) + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_observer_transition", + what="count()", + msg="Both participants should be visible before observer transition", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + reload_config_on_node(nodes["node0"], CONFIG_OBSERVER) + + # node0 must disappear from node1's view without waiting for ZK session expiry. + for retry in range(15): + hosts = ( + nodes["node1"] + .query( + "SELECT host_name FROM system.clusters " + "WHERE cluster = 'test_observer_transition' ORDER BY host_name", + password="passwordAbc", + ) + .strip() + .split("\n") + ) + hosts = [h for h in hosts if h] + if hosts == ["node1"]: + break + time.sleep(1) + else: + raise AssertionError( + f"node0 still advertised after observer reload; hosts on node1: {hosts}" + ) + + # Observer still sees the remaining participant. + check_on_cluster( + [nodes["node0"]], + 1, + cluster_name="test_observer_transition", + what="count()", + msg="Observer should still see the remaining participant", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + +def test_reload_invisible_to_visible_populates_cluster(start_cluster): + """Invisible -> visible reload must upsert and publish nodes promptly.""" + reload_config_on_all(CONFIG_INVISIBLE) + + for retry in range(10): + counts = [ + int( + node.query( + "SELECT count() FROM system.clusters " + "WHERE cluster = 'test_invisible_transition'", + password="passwordAbc", + ) + ) + for node in nodes.values() + ] + if all(c == 0 for c in counts): + break + time.sleep(1) + else: + raise AssertionError( + f"Invisible cluster should not appear in system.clusters: {counts}" + ) + + reload_config_on_all(CONFIG_VISIBLE) + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_invisible_transition", + what="count()", + msg="Cluster did not appear after becoming visible", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + reload_config_on_all(CONFIG_INVISIBLE) + + for retry in range(15): + counts = [ + int( + node.query( + "SELECT count() FROM system.clusters " + "WHERE cluster = 'test_invisible_transition'", + password="passwordAbc", + ) + ) + for node in nodes.values() + ] + if all(c == 0 for c in counts): + break + time.sleep(1) + else: + raise AssertionError( + f"Cluster still visible after invisible reload: {counts}" + ) + + +def test_reload_static_replaces_dynamic_same_name(start_cluster): + """Static for a name already discovered via multicluster must replace it cleanly.""" + config_participant = """ + + 1 + + + + /clickhouse/discovery/test_collision_cluster + + + + +""" + config_multicluster_observer = """ + + 1 + + + + + /clickhouse/discovery + + + + +""" + config_static_observer = """ + + 1 + + + + /clickhouse/discovery/test_collision_cluster + + + + + +""" + config_static_and_multicluster = """ + + 1 + + + + /clickhouse/discovery/test_collision_cluster + + + + + + + /clickhouse/discovery + + + + +""" + + reload_config_on_node(nodes["node1"], config_participant) + reload_config_on_node(nodes["node0"], config_multicluster_observer) + + check_on_cluster( + [nodes["node0"]], + 1, + cluster_name="test_collision_cluster", + what="count()", + msg="Observer should discover dynamic test_collision_cluster", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + # Replace dynamic discovery with static config of the same name. + reload_config_on_node(nodes["node0"], config_static_observer) + + check_on_cluster( + [nodes["node0"]], + 1, + cluster_name="test_collision_cluster", + what="count()", + msg="Static observer should still see the participant after replacing dynamic", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + # Static + multicluster: static shadows the dynamic name. Removing only the static entry + # must rescan roots so the dynamic cluster reappears without waiting for force refresh. + reload_config_on_node(nodes["node0"], config_static_and_multicluster) + check_on_cluster( + [nodes["node0"]], + 1, + cluster_name="test_collision_cluster", + what="count()", + msg="Static+multicluster observer should see the participant", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + reload_config_on_node(nodes["node0"], config_multicluster_observer) + check_on_cluster( + [nodes["node0"]], + 1, + cluster_name="test_collision_cluster", + what="count()", + msg="Dynamic cluster must reappear after static shadow is removed", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + +def _registration_config(hostname, shard): + return f""" + + 1 + + + + /clickhouse/discovery/test_registration_reload + {hostname} + {shard} + + + + +""" + + +def _registration_rows(node): + return node.query( + "SELECT host_name, shard_num FROM system.clusters " + "WHERE cluster = 'test_registration_reload' ORDER BY host_name, shard_num " + "FORMAT TSV", + password="passwordAbc", + ).strip() + + +def test_reload_my_hostname_and_shard_updates_local_and_peer(start_cluster): + """Registration field reload must refresh payloads locally and on peers without membership churn.""" + reload_config_on_node(nodes["node0"], _registration_config("reg-host-node0", 1)) + reload_config_on_node(nodes["node1"], _registration_config("reg-host-node1", 1)) + + expected_initial = "reg-host-node0\t1\nreg-host-node1\t1" + for retry in range(15): + rows = {_registration_rows(node) for node in nodes.values()} + if rows == {expected_initial}: + break + time.sleep(1) + else: + raise AssertionError(f"Initial registration view not ready: {rows}") + + # Change hostname and shard on node1 only; UUID set stays the same. + reload_config_on_node(nodes["node1"], _registration_config("reg-host-node1-renamed", 2)) + + expected_updated = "reg-host-node0\t1\nreg-host-node1-renamed\t2" + for retry in range(15): + rows = {_registration_rows(node) for node in nodes.values()} + if rows == {expected_updated}: + break + time.sleep(1) + else: + raise AssertionError( + f"Hostname/shard reload did not propagate to local and peer system.clusters: {rows}" + ) + + +def test_keeper_exception_after_wait_restores_retry_signal(start_cluster): + """A one-shot Keeper throw after Flags::wait must not leave peer updates stuck forever.""" + reload_config_on_node(nodes["node0"], _registration_config("reg-host-node0", 1)) + reload_config_on_node(nodes["node1"], _registration_config("reg-host-node1", 1)) + + expected_initial = "reg-host-node0\t1\nreg-host-node1\t1" + for retry in range(15): + rows = {_registration_rows(node) for node in nodes.values()} + if rows == {expected_initial}: + break + time.sleep(1) + else: + raise AssertionError(f"Initial registration view not ready: {rows}") + + node0 = nodes["node0"] + node0.query( + "SYSTEM ENABLE FAILPOINT cluster_discovery_retry_signal_fail", + password="passwordAbc", + ) + + # Peer-only registration change: node0 is woken by the Keeper children watch, not by a + # local config reload. Without restoring flags/wake after the failpoint throw, node0 would + # wait until an unrelated event (or never) to see the new payload. + reload_config_on_node(nodes["node1"], _registration_config("reg-host-node1-renamed", 2)) + + expected_updated = "reg-host-node0\t1\nreg-host-node1-renamed\t2" + for retry in range(20): + rows = _registration_rows(node0) + if rows == expected_updated: + break + time.sleep(1) + else: + raise AssertionError( + f"node0 did not recover peer registration update after one-shot Keeper failpoint; " + f"got {rows!r}" + ) + + # Peer that did not hit the failpoint must also converge. + for retry in range(15): + rows = {_registration_rows(node) for node in nodes.values()} + if rows == {expected_updated}: + break + time.sleep(1) + else: + raise AssertionError(f"Cluster views did not converge after retry-signal recovery: {rows}") + + +def _shared_path_aliases_config(include_alias_a, alias_a_observer=False): + alias_a = "" + if include_alias_a: + observer = "\n " if alias_a_observer else "" + alias_a = f""" + + + /clickhouse/discovery/test_shared_path_aliases{observer} + + """ + return f""" + + 1 + {alias_a} + + + /clickhouse/discovery/test_shared_path_aliases + + + + +""" + + +def _alias_b_host_count(node): + return int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'alias_b'", + password="passwordAbc", + ) + ) + + +def test_reload_remove_shared_path_alias_keeps_peer_membership(start_cluster): + """Removing one participant alias must not delete the shared ephemeral while another remains.""" + # node0: two aliases on one Keeper path; node1: only the retained alias. + reload_config_on_node(nodes["node0"], _shared_path_aliases_config(include_alias_a=True)) + reload_config_on_node(nodes["node1"], _shared_path_aliases_config(include_alias_a=False)) + + for retry in range(15): + if _alias_b_host_count(nodes["node1"]) == len(nodes): + break + time.sleep(1) + else: + raise AssertionError("alias_b not ready with both nodes before shared-path alias remove") + + reload_config_on_node(nodes["node0"], _shared_path_aliases_config(include_alias_a=False)) + + # Membership must never transiently drop: the shared ephemeral must stay. + for _ in range(20): + hosts = _alias_b_host_count(nodes["node1"]) + if hosts != len(nodes): + raise AssertionError( + f"Peer lost membership after removing shared-path alias_a; alias_b hosts={hosts}" + ) + time.sleep(0.2) + + # Convert-to-observer on a restored alias_a must also keep the shared registration. + reload_config_on_node(nodes["node0"], _shared_path_aliases_config(include_alias_a=True)) + for retry in range(15): + if _alias_b_host_count(nodes["node1"]) == len(nodes): + break + time.sleep(1) + else: + raise AssertionError("alias_b not ready before shared-path alias observer convert") + + reload_config_on_node( + nodes["node0"], + _shared_path_aliases_config(include_alias_a=True, alias_a_observer=True), + ) + + for _ in range(20): + hosts = _alias_b_host_count(nodes["node1"]) + if hosts != len(nodes): + raise AssertionError( + f"Peer lost membership after converting shared-path alias_a to observer; " + f"alias_b hosts={hosts}" + ) + time.sleep(0.2) diff --git a/tests/integration/test_cluster_discovery/test_enable_after_startup.py b/tests/integration/test_cluster_discovery/test_enable_after_startup.py new file mode 100644 index 000000000000..56ecd27adfc1 --- /dev/null +++ b/tests/integration/test_cluster_discovery/test_enable_after_startup.py @@ -0,0 +1,126 @@ +import time + +import pytest + +from helpers.cluster import ClickHouseCluster + +from .common import check_on_cluster + +cluster = ClickHouseCluster(__file__) + +nodes = { + "node0": cluster.add_instance( + "node0", + main_configs=["config/config_discovery_disabled.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), + "node1": cluster.add_instance( + "node1", + main_configs=["config/config_discovery_disabled.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), +} + +CONFIG_PATH = "/etc/clickhouse-server/config.d/config_discovery_disabled.xml" + +CONFIG_ENABLED = """ + + 1 + + + + /clickhouse/discovery/test_enable_after_startup + + + + +""" + +CONFIG_BAD_AUX_KEEPER = """ + + 1 + + + + missing_aux_keeper:/clickhouse/discovery/test_pending_during_init + + + + +""" + +CONFIG_GOOD_AFTER_BAD_INIT = """ + + 1 + + + + /clickhouse/discovery/test_pending_during_init + + + + +""" + + +@pytest.fixture(scope="module") +def start_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def test_reload_applies_while_initial_update_is_failing(start_cluster): + """Pending config must be consumed before retrying initialUpdate, or a fix reload is stuck.""" + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_BAD_AUX_KEEPER) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + # Let the worker retry failed init on the obsolete auxiliary keeper. + time.sleep(2) + + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_GOOD_AFTER_BAD_INIT) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_pending_during_init", + what="count()", + msg="Corrective reload was not applied while discovery init was failing", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + +def test_enable_discovery_after_startup_starts_worker(start_cluster): + """Creating ClusterDiscovery on reload must start the worker without a process restart.""" + for node in nodes.values(): + count = int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_enable_after_startup'", + password="passwordAbc", + ) + ) + assert count == 0 + + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_ENABLED) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_enable_after_startup", + what="count()", + msg="Discovery cluster missing after enabling feature post-startup", + query_params={"password": "passwordAbc"}, + retries=6, + ) diff --git a/tests/integration/test_cluster_discovery/test_enable_allow_only.py b/tests/integration/test_cluster_discovery/test_enable_allow_only.py new file mode 100644 index 000000000000..e3235bbbb3e7 --- /dev/null +++ b/tests/integration/test_cluster_discovery/test_enable_allow_only.py @@ -0,0 +1,152 @@ +import time + +import pytest + +from helpers.cluster import ClickHouseCluster + +from .common import check_on_cluster + +cluster = ClickHouseCluster(__file__) + +nodes = { + "node0": cluster.add_instance( + "node0", + main_configs=["config/config_discovery_disabled_with_path.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), + "node1": cluster.add_instance( + "node1", + main_configs=["config/config_discovery_disabled_with_path.xml"], + user_configs=["config/users.d/users_with_pwd.xml"], + stay_alive=True, + with_zookeeper=True, + ), +} + +CONFIG_PATH = "/etc/clickhouse-server/config.d/config_discovery_disabled_with_path.xml" + +CONFIG_ALLOW_DISABLED = """ + + 0 + + + + /clickhouse/discovery/test_enable_allow_only + + + + +""" + +CONFIG_ALLOW_ENABLED = """ + + 1 + + + + /clickhouse/discovery/test_enable_allow_only + + + + +""" + + +@pytest.fixture(scope="module") +def start_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def _cluster_host_count(node): + return int( + node.query( + "SELECT count() FROM system.clusters WHERE cluster = 'test_enable_allow_only'", + password="passwordAbc", + ) + ) + + +def test_enable_allow_flag_only_starts_worker(start_cluster): + """Flipping only allow_experimental_cluster_discovery must start discovery (remote_servers unchanged).""" + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_ALLOW_DISABLED) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + for _ in range(15): + if all(_cluster_host_count(node) == 0 for node in nodes.values()): + break + time.sleep(1) + else: + raise AssertionError("Discovery cluster still published after allow=0 baseline") + + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_ALLOW_ENABLED) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_enable_allow_only", + what="count()", + msg="Discovery cluster missing after allow-flag-only reload", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + +def test_disable_allow_flag_only_stops_discovery(start_cluster): + """allow 1 → 0 must unregister and unpublish; 0 → 1 must restore without changing remote_servers.""" + for node in nodes.values(): + node.replace_config(CONFIG_PATH, CONFIG_ALLOW_ENABLED) + node.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_enable_allow_only", + what="count()", + msg="Discovery cluster not ready before allow-disable test", + query_params={"password": "passwordAbc"}, + retries=6, + ) + + node0 = nodes["node0"] + node1 = nodes["node1"] + + node0.replace_config(CONFIG_PATH, CONFIG_ALLOW_DISABLED) + node0.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + for _ in range(15): + if _cluster_host_count(node0) == 0: + break + time.sleep(1) + else: + raise AssertionError("node0 still publishes discovery cluster after allow=0 reload") + + for _ in range(15): + if _cluster_host_count(node1) == 1: + break + time.sleep(1) + else: + raise AssertionError( + f"node1 still sees node0 in Keeper after allow=0 on node0; hosts={_cluster_host_count(node1)}" + ) + + node0.replace_config(CONFIG_PATH, CONFIG_ALLOW_ENABLED) + node0.query("SYSTEM RELOAD CONFIG", password="passwordAbc") + + check_on_cluster( + list(nodes.values()), + len(nodes), + cluster_name="test_enable_allow_only", + what="count()", + msg="Discovery cluster missing after allow 0 → 1 reload", + query_params={"password": "passwordAbc"}, + retries=6, + )