Merge pull request ClickHouse#87222 from ClickHouse/backport/25.8/86479

Backport ClickHouse#86479 to 25.8: Handling loopback host ID properly to avoid collision when processing DDL tasks:

Author: tuanpach

src/Interpreters/DDLTask.cpp

@@ -1,24 +1,26 @@
-#include <Interpreters/DDLTask.h>
-#include <base/sort.h>
-#include <Common/DNSResolver.h>
-#include <Common/OpenTelemetryTraceContext.h>
-#include <Common/isLocalAddress.h>
-#include <Common/ZooKeeper/ZooKeeper.h>
+#include <Core/ServerSettings.h>
+#include <Core/ServerUUID.h>
 #include <Core/Settings.h>
 #include <Databases/DatabaseReplicated.h>
-#include <Interpreters/DatabaseCatalog.h>
-#include <Interpreters/Context.h>
-#include <IO/WriteHelpers.h>
-#include <IO/ReadHelpers.h>
 #include <IO/Operators.h>
 #include <IO/ReadBufferFromString.h>
-#include <Poco/Net/NetException.h>
-#include <Common/logger_useful.h>
+#include <IO/ReadHelpers.h>
+#include <IO/WriteHelpers.h>
+#include <Interpreters/Context.h>
+#include <Interpreters/DDLTask.h>
+#include <Interpreters/DDLWorker.h>
+#include <Interpreters/DatabaseCatalog.h>
 #include <Parsers/ASTQueryWithOnCluster.h>
+#include <Parsers/ASTQueryWithTableAndOutput.h>
 #include <Parsers/ParserQuery.h>
 #include <Parsers/parseQuery.h>
-#include <Parsers/ASTQueryWithTableAndOutput.h>
-
+#include <base/sort.h>
+#include <Poco/Net/NetException.h>
+#include <Common/DNSResolver.h>
+#include <Common/OpenTelemetryTraceContext.h>
+#include <Common/ZooKeeper/ZooKeeper.h>
+#include <Common/isLocalAddress.h>
+#include <Common/logger_useful.h>
 
 namespace DB
 {
@@ -31,7 +33,7 @@ namespace Setting
     extern const SettingsUInt64 max_parser_depth;
     extern const SettingsUInt64 max_parser_backtracks;
     extern const SettingsUInt64 max_query_size;
-}
+    }
 
 namespace ErrorCodes
 {
@@ -55,7 +57,27 @@ bool HostID::isLocalAddress(UInt16 clickhouse_port) const
 {
     try
     {
-        return DB::isLocalAddress(DNSResolver::instance().resolveAddress(host_name, port), clickhouse_port);
+        auto address = DNSResolver::instance().resolveAddress(host_name, port);
+        return DB::isLocalAddress(address, clickhouse_port);
+    }
+    catch (const DB::NetException &)
+    {
+        /// Avoid "Host not found" exceptions
+        return false;
+    }
+    catch (const Poco::Net::NetException &)
+    {
+        /// Avoid "Host not found" exceptions
+        return false;
+    }
+}
+
+bool HostID::isLoopbackHost() const
+{
+    try
+    {
+        auto address = DNSResolver::instance().resolveAddress(host_name, port);
+        return address.host().isLoopback();
     }
     catch (const DB::NetException &)
     {
@@ -270,10 +292,7 @@ bool DDLTask::findCurrentHostID(ContextPtr global_context, LoggerPtr log, const
 
     if (config_host_name)
     {
-        bool is_local_port = (maybe_secure_port && HostID(*config_host_name, *maybe_secure_port).isLocalAddress(*maybe_secure_port)) ||
-                             HostID(*config_host_name, port).isLocalAddress(port);
-
-        if (!is_local_port)
+        if (!IsSelfHostname(*config_host_name, maybe_secure_port, port))
             throw Exception(
                 ErrorCodes::DNS_ERROR,
                 "{} is not a local address. Check parameter 'host_name' in the configuration",
@@ -298,12 +317,28 @@ bool DDLTask::findCurrentHostID(ContextPtr global_context, LoggerPtr log, const
 
         try
         {
-            /// The port is considered local if it matches TCP or TCP secure port that the server is listening.
-            bool is_local_port
-                = (maybe_secure_port && host.isLocalAddress(*maybe_secure_port)) || host.isLocalAddress(port);
-
-            if (!is_local_port)
+            if (!IsSelfHostID(host, maybe_secure_port, port))
                 continue;
+
+            if (host.isLoopbackHost())
+            {
+                String current_host_id_str = host.toString();
+                String active_id = toString(ServerUUID::get());
+                String active_path = fs::path(global_context->getDDLWorker().getReplicasDir()) / current_host_id_str / "active";
+                String content;
+                Coordination::Stat stat;
+                if (!zookeeper->tryGet(active_path, content, &stat))
+                {
+                    LOG_TRACE(log, "HostID {} is a loopback host which has not been claimed by any replica", current_host_id_str);
+                    continue;
+                }
+
+                if (content != active_id)
+                {
+                    LOG_TRACE(log, "HostID {} is a loopback host which is claimed by another replica {}", current_host_id_str, content);
+                    continue;
+                }
+            }
         }
         catch (const Exception & e)
         {
@@ -506,6 +541,20 @@ String DDLTask::getShardID() const
     return res;
 }
 
+bool DDLTask::IsSelfHostID(const HostID & checking_host_id, std::optional<UInt16> maybe_self_secure_port, UInt16 self_port)
+{
+    // If the checking_host_id has a loopback address, it is not considered as the self host_id.
+    // Because all replicas will try to claim it as their own hosts.
+    return (maybe_self_secure_port && checking_host_id.isLocalAddress(*maybe_self_secure_port))
+        || checking_host_id.isLocalAddress(self_port);
+}
+
+bool DDLTask::IsSelfHostname(const String & checking_host_name, std::optional<UInt16> maybe_self_secure_port, UInt16 self_port)
+{
+    return (maybe_self_secure_port && HostID(checking_host_name, *maybe_self_secure_port).isLocalAddress(*maybe_self_secure_port))
+        || HostID(checking_host_name, self_port).isLocalAddress(self_port);
+}
+
 DatabaseReplicatedTask::DatabaseReplicatedTask(const String & name, const String & path, DatabaseReplicated * database_)
     : DDLTaskBase(name, path)
     , database(database_)

src/Interpreters/DDLTask.h

@@ -61,6 +61,7 @@ struct HostID
     }
 
     bool isLocalAddress(UInt16 clickhouse_port) const;
+    bool isLoopbackHost() const;
 
     static String applyToString(const HostID & host_id)
     {
@@ -157,6 +158,9 @@ struct DDLTask : public DDLTaskBase
 
     String getShardID() const override;
 
+    static bool IsSelfHostID(const HostID & checking_host_id, std::optional<UInt16> maybe_self_secure_port, UInt16 self_port);
+    static bool IsSelfHostname(const String & checking_host_name, std::optional<UInt16> maybe_self_secure_port, UInt16 self_port);
+
 private:
     bool tryFindHostInCluster();
     bool tryFindHostInClusterViaResolving(ContextPtr context);

src/Interpreters/DDLWorker.cpp

@@ -1,14 +1,16 @@
 
+#include <Core/ServerSettings.h>
 #include <Core/ServerUUID.h>
 #include <Core/Settings.h>
+#include <IO/NullWriteBuffer.h>
 #include <IO/ReadBufferFromString.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteHelpers.h>
 #include <Interpreters/Cluster.h>
 #include <Interpreters/Context.h>
-#include <Interpreters/DatabaseCatalog.h>
 #include <Interpreters/DDLTask.h>
 #include <Interpreters/DDLWorker.h>
+#include <Interpreters/DatabaseCatalog.h>
 #include <Interpreters/ZooKeeperLog.h>
 #include <Interpreters/executeQuery.h>
 #include <Parsers/ASTAlterQuery.h>
@@ -19,7 +21,6 @@
 #include <Parsers/ASTQueryWithOnCluster.h>
 #include <Parsers/ASTQueryWithTableAndOutput.h>
 #include <Parsers/ParserQuery.h>
-#include <IO/NullWriteBuffer.h>
 #include <Storages/IStorage.h>
 #include <Storages/StorageReplicatedMergeTree.h>
 #include <Poco/Timestamp.h>
@@ -1349,10 +1350,7 @@ void DDLWorker::markReplicasActive(bool reinitialized)
         try
         {
             HostID host = HostID::fromString(host_id);
-            /// The port is considered local if it matches TCP or TCP secure port that the server is listening.
-            bool is_local_host = (maybe_secure_port && host.isLocalAddress(*maybe_secure_port)) || host.isLocalAddress(port);
-
-            if (is_local_host)
+            if (DDLTask::IsSelfHostID(host, maybe_secure_port, port))
                 local_host_ids.emplace(host_id);
         }
         catch (const Exception & e)
@@ -1366,17 +1364,36 @@ void DDLWorker::markReplicasActive(bool reinitialized)
     {
         auto it = active_node_holders.find(host_id);
         if (it != active_node_holders.end())
-        {
             continue;
-        }
 
         String active_path = fs::path(replicas_dir) / host_id / "active";
         String active_id = toString(ServerUUID::get());
 
         LOG_TRACE(log, "Trying to mark a replica active: active_path={}, active_id={}", active_path, active_id);
+        if (HostID::fromString(host_id).isLoopbackHost())
+        {
+            String content;
+            Coordination::Stat stat;
+            if (zookeeper->tryGet(active_path, content, &stat))
+            {
+                // For a loopback host, many replicas might try to claim it as their own host.
+                // If the host is claimed by a replica, we skip it.
+                // Loopback host is supposed to be used in test environment.
+                if (content != active_id)
+                {
+                    LOG_TRACE(log, "HostID {} is a loopback host which is claimed by another replica {}", host_id, content);
+                    continue;
+                }
 
-        zookeeper->deleteEphemeralNodeIfContentMatches(active_path, active_id);
-
+                auto code = zookeeper->tryRemove(active_path, stat.version);
+                if (code != Coordination::Error::ZOK && code != Coordination::Error::ZNONODE)
+                    throw Coordination::Exception::fromPath(code, active_path);
+            }
+        }
+        else
+        {
+            zookeeper->deleteEphemeralNodeIfContentMatches(active_path, active_id);
+        }
         zookeeper->create(active_path, active_id, zkutil::CreateMode::Ephemeral);
         auto active_node_holder_zookeeper = zookeeper;
         auto active_node_holder = zkutil::EphemeralNodeHolder::existing(active_path, *active_node_holder_zookeeper);

tests/integration/test_ddl_worker_with_loopback_hosts/__init__.py

@@ -0,0 +1,33 @@
+<clickhouse>
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <internal_replication>true</internal_replication>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_cluster>
+        <test_loopback_cluster1>
+            <shard>
+                <replica>
+                    <host>127.0.0.1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_loopback_cluster1>
+        <test_loopback_cluster2>
+            <shard>
+                <replica>
+                    <host>localhost</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_loopback_cluster2>
+    </remote_servers>
+</clickhouse>

tests/integration/test_ddl_worker_with_loopback_hosts/configs/remote_servers.xml

@@ -0,0 +1,98 @@
+import time
+
+import pytest
+
+from helpers.client import QueryRuntimeException
+from helpers.cluster import ClickHouseCluster
+from helpers.network import PartitionManager
+
+cluster = ClickHouseCluster(__file__)
+node1 = cluster.add_instance(
+    "node1",
+    main_configs=[
+        "configs/remote_servers.xml",
+    ],
+    with_zookeeper=True,
+)
+node2 = cluster.add_instance(
+    "node2",
+    main_configs=[
+        "configs/remote_servers.xml",
+    ],
+    with_zookeeper=True,
+)
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+def count_table(node, table_name):
+    return int(
+        node.query(
+            f"SELECT count() FROM system.tables WHERE name='{table_name}'"
+        ).strip()
+    )
+
+
+def test_ddl_worker_with_loopback_hosts(
+    started_cluster,
+):
+    node1.query("DROP TABLE IF EXISTS t1 SYNC")
+    node2.query("DROP TABLE IF EXISTS t1 SYNC")
+    node1.query("DROP TABLE IF EXISTS t2 SYNC")
+    node2.query("DROP TABLE IF EXISTS t2 SYNC")
+    node1.query("DROP TABLE IF EXISTS t3 SYNC")
+    node2.query("DROP TABLE IF EXISTS t3 SYNC")
+    node1.query("DROP TABLE IF EXISTS t4 SYNC")
+    node2.query("DROP TABLE IF EXISTS t4 SYNC")
+
+    node1.query(
+        "CREATE TABLE t1 ON CLUSTER 'test_cluster' (x INT) ENGINE=MergeTree() ORDER BY x",
+        settings={
+            "distributed_ddl_task_timeout": 10,
+        },
+    )
+
+    node2.query(
+        "CREATE TABLE t2 ON CLUSTER 'test_cluster' (x INT) ENGINE=MergeTree() ORDER BY x",
+        settings={
+            "distributed_ddl_task_timeout": 10,
+        },
+    )
+
+    assert count_table(node1, "t2") == 1
+    assert count_table(node2, "t1") == 1
+
+    node1.query(
+        "CREATE TABLE t3 ON CLUSTER 'test_loopback_cluster1' (x INT) ENGINE=MergeTree() ORDER BY x",
+        settings={
+            "distributed_ddl_task_timeout": 10,
+        },
+    )
+    # test_loopback_cluster1 has a loopback host, only 1 replica processed the query
+    assert count_table(node1, "t3") == 1 or count_table(node2, "t3") == 1
+
+    node2.query(
+        "CREATE TABLE t4 ON CLUSTER 'test_loopback_cluster2' (x INT) ENGINE=MergeTree() ORDER BY x",
+        settings={
+            "distributed_ddl_task_timeout": 10,
+        },
+    )
+    # test_loopback_cluster2 has a loopback host, only 1 replica processed the query
+    assert count_table(node1, "t4") == 1 or count_table(node2, "t4") == 1
+
+    node1.query("DROP TABLE IF EXISTS t1 SYNC")
+    node2.query("DROP TABLE IF EXISTS t1 SYNC")
+    node1.query("DROP TABLE IF EXISTS t2 SYNC")
+    node2.query("DROP TABLE IF EXISTS t2 SYNC")
+    node1.query("DROP TABLE IF EXISTS t3 SYNC")
+    node2.query("DROP TABLE IF EXISTS t3 SYNC")
+    node1.query("DROP TABLE IF EXISTS t4 SYNC")
+    node2.query("DROP TABLE IF EXISTS t4 SYNC")