Merge pull request #52321 from ClickHouse/fix-keeper-disk-initialization

Skip unsupported disks in Keeper

src/Coordination/KeeperContext.cpp

@@ -41,9 +41,38 @@ void KeeperContext::initialize(const Poco::Util::AbstractConfiguration & config)
     initializeDisks(config);
 }
 
+namespace
+{
+
+bool diskValidator(const Poco::Util::AbstractConfiguration & config, const std::string & disk_config_prefix)
+{
+    const auto disk_type = config.getString(disk_config_prefix + ".type", "local");
+
+    using namespace std::literals;
+    static constexpr std::array supported_disk_types
+    {
+        "s3"sv,
+        "s3_plain"sv,
+        "local"sv
+    };
+
+    if (std::all_of(
+            supported_disk_types.begin(),
+            supported_disk_types.end(),
+            [&](const auto supported_type) { return disk_type != supported_type; }))
+    {
+        LOG_INFO(&Poco::Logger::get("KeeperContext"), "Disk type '{}' is not supported for Keeper", disk_type);
+        return false;
+    }
+
+    return true;
+}
+
+}
+
 void KeeperContext::initializeDisks(const Poco::Util::AbstractConfiguration & config)
 {
-    disk_selector->initialize(config, "storage_configuration.disks", Context::getGlobalContextInstance());
+    disk_selector->initialize(config, "storage_configuration.disks", Context::getGlobalContextInstance(), diskValidator);
 
     log_storage = getLogsPathFromConfig(config);
 

src/Disks/DiskSelector.cpp

@@ -27,7 +27,7 @@ void DiskSelector::assertInitialized() const
 }
 
 
-void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context)
+void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator)
 {
     Poco::Util::AbstractConfiguration::Keys keys;
     config.keys(config_prefix, keys);
@@ -46,6 +46,9 @@ void DiskSelector::initialize(const Poco::Util::AbstractConfiguration & config,
 
         auto disk_config_prefix = config_prefix + "." + disk_name;
 
+        if (disk_validator && !disk_validator(config, disk_config_prefix))
+            continue;
+
         disks.emplace(disk_name, factory.create(disk_name, config, disk_config_prefix, context, disks));
     }
     if (!has_default_disk)

src/Disks/DiskSelector.h

@@ -23,7 +23,8 @@ public:
     DiskSelector() = default;
     DiskSelector(const DiskSelector & from) = default;
 
-    void initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context);
+    using DiskValidator = std::function<bool(const Poco::Util::AbstractConfiguration & config, const String & disk_config_prefix)>;
+    void initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator = {});
 
     DiskSelectorPtr updateFromConfig(
         const Poco::Util::AbstractConfiguration & config,

tests/integration/test_keeper_disks/configs/enable_keeper.xml

@@ -1,6 +1,10 @@
 <clickhouse>
     <storage_configuration>
         <disks>
+            <disk_hdfs>
+                <type>hdfs</type>
+                <endpoint>hdfs://hdfs1:9000/</endpoint>
+            </disk_hdfs>
             <log_local>
                 <type>local</type>
                 <path>/var/lib/clickhouse/coordination/logs/</path>

tests/integration/test_keeper_disks/test.py

@@ -9,7 +9,11 @@ import os
 CURRENT_TEST_DIR = os.path.dirname(os.path.abspath(__file__))
 cluster = ClickHouseCluster(__file__)
 node = cluster.add_instance(
-    "node", main_configs=["configs/enable_keeper.xml"], stay_alive=True, with_minio=True
+    "node",
+    main_configs=["configs/enable_keeper.xml"],
+    stay_alive=True,
+    with_minio=True,
+    with_hdfs=True,
 )
 
 from kazoo.client import KazooClient, KazooState
@@ -117,6 +121,12 @@ def get_local_snapshots():
     return get_local_files("/var/lib/clickhouse/coordination/snapshots")
 
 
+def test_supported_disk_types(started_cluster):
+    node.stop_clickhouse()
+    node.start_clickhouse()
+    node.contains_in_log("Disk type 'hdfs' is not supported for Keeper")
+
+
 def test_logs_with_disks(started_cluster):
     setup_local_storage(started_cluster)