diff --git a/src/Disks/DiskSelector.cpp b/src/Disks/DiskSelector.cpp index 441451ad5ab37..130286306ac0e 100644 --- a/src/Disks/DiskSelector.cpp +++ b/src/Disks/DiskSelector.cpp @@ -143,7 +143,7 @@ catch (...) } DiskSelectorPtr DiskSelector::updateFromConfig( - const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context) const + const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, const NameSet & disks_to_skip) const { assertInitialized(); @@ -166,6 +166,14 @@ DiskSelectorPtr DiskSelector::updateFromConfig( auto disk_config_prefix = config_prefix + "." + disk_name; if (!result->getDisksMap().contains(disk_name)) { + if (disks_to_skip.contains(disk_name)) + { + LOG_WARNING( + getLogger("DiskSelector"), + "Disk {} is not created from the configuration: a custom disk with this name is being created", + backQuote(disk_name)); + continue; + } auto created_disk = factory.create( disk_name, config, disk_config_prefix, context, result->getDisksMap(), /*attach*/ false, /*custom_disk*/ false, skip_types); if (created_disk) diff --git a/src/Disks/DiskSelector.h b/src/Disks/DiskSelector.h index acf491daa1c3e..f85820db1bfe9 100644 --- a/src/Disks/DiskSelector.h +++ b/src/Disks/DiskSelector.h @@ -29,10 +29,13 @@ class DiskSelector using DiskValidator = std::function; void initialize(const Poco::Util::AbstractConfiguration & config, const String & config_prefix, ContextPtr context, DiskValidator disk_validator = {}); + /// Configured disks named in `disks_to_skip` are not created by this reload: their names are being + /// created as custom disks, and creating them here could open the same storage twice. DiskSelectorPtr updateFromConfig( const Poco::Util::AbstractConfiguration & config, const String & config_prefix, - ContextPtr context) const; + ContextPtr context, + const NameSet & disks_to_skip = {}) const; /// Get disk by name DiskPtr get(const String & name) const; diff --git a/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp new file mode 100644 index 0000000000000..34e6e9390987a --- /dev/null +++ b/src/Disks/tests/gtest_cas_startup_stale_mounts.cpp @@ -0,0 +1,525 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/// Server start re-creates the `disk = disk(...)` custom disks of every table it loads. A CAS disk +/// whose previous process died without a farewell waits out the stale mount lease inside that +/// creation. These tests pin how such waits compose with the rest of startup. + +namespace DB::ErrorCodes +{ +extern const int BAD_ARGUMENTS; +} + +namespace +{ + +namespace fs = std::filesystem; +using namespace DB::Cas; + +constexpr uint64_t lease_ttl_ms = 3000; +constexpr uint64_t renew_period_ms = 600; +/// `mountObservationThresholdMs`: TTL + TTL/20 + renew_period/2. +constexpr uint64_t observation_ms = lease_ttl_ms + lease_ttl_ms / 20 + renew_period_ms / 2; + +uint64_t elapsedMs(std::chrono::steady_clock::time_point since) +{ + return static_cast( + std::chrono::duration_cast(std::chrono::steady_clock::now() - since).count()); +} + +uint64_t wallNowMs() +{ + return static_cast(std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()).count()); +} + +struct StaleCasDisk +{ + String name; + String server_root_id; + fs::path pool_dir; + fs::path scratch_dir; + + std::map settings() const + { + return { + {"type", "object_storage"}, + {"object_storage_type", "local"}, + {"metadata_type", "cas"}, + {"path", pool_dir.string() + "/"}, + {"cas_server_root_id", server_root_id}, + {"cas_scratch_path", scratch_dir.string() + "/"}, + {"cas_mount_lease_ttl_ms", std::to_string(lease_ttl_ms)}, + {"cas_mount_renew_period_ms", std::to_string(renew_period_ms)}, + {"cas_attempt_timeout_ms", "200"}, + {"cas_lease_safety_margin_ms", "200"}, + {"cas_gc_enabled", "0"}, + }; + } + + /// The table-metadata form: what server start parses from `SETTINGS disk = disk(...)`. + DB::ASTPtr diskFunctionAst() const + { + String text = "disk(name = '" + name + "'"; + for (const auto & [key, value] : settings()) + text += ", " + key + " = '" + value + "'"; + text += ")"; + DB::ParserFunction parser; + return DB::parseQuery(parser, text, "", 0, 0, 0); + } + + Poco::AutoPtr xmlConfig() const + { + std::ostringstream xml; + xml << ""; + for (const auto & [key, value] : settings()) + xml << "<" << key << ">" << value << ""; + xml << ""; + std::istringstream in(xml.str()); + return Poco::AutoPtr(new Poco::Util::XMLConfiguration(in)); + } +}; + +/// Leaves `disk` as a process killed after it minted a writer epoch and claimed the mount: a live- +/// looking lease with no farewell, held by this server's own uuid under an older epoch. +void seedStaleMount(const StaleCasDisk & disk) +{ + { + auto config = disk.xmlConfig(); + auto seeded = DB::DiskFactory::instance().create( + disk.name + "_seed", *config, "disk", getContext().context, {}, /*attach=*/true, /*custom_disk=*/true); + seeded->shutdown(); + } + + auto object_storage = std::make_shared( + DB::LocalObjectStorageSettings(disk.name + "_seed_backend", disk.pool_dir.string() + "/", /*read_only_=*/false)); + auto backend = std::make_shared(object_storage, ObjectStorageBackend::Mode::EmulatedSingleProcess); + DB::Cas::tests::OperationForTest op(backend); + const Layout layout{"ca"}; + + const auto mount = (*op).read(layout.mountKey(disk.server_root_id), Retry::standard()); + ASSERT_TRUE(mount.has_value()); + const MountLease farewell = decodeMountLease(mount->bytes); + ASSERT_EQ(farewell.min_active_build_sequence, std::numeric_limits::max()) + << "the seeding open must end in a clean farewell"; + + const auto epoch_object = (*op).read(layout.epochKey(disk.server_root_id), Retry::standard()); + ASSERT_TRUE(epoch_object.has_value()); + const uint64_t crashed_epoch = decodeServerEpoch(epoch_object->bytes).next_writer_epoch; + ASSERT_TRUE(std::holds_alternative((*op).replace( + layout.epochKey(disk.server_root_id), encodeServerEpoch(ServerEpoch{.next_writer_epoch = crashed_epoch + 1}), + epoch_object->etag, Retry::standard()))); + + const MountClaimResult claim = claimMount( + *op, layout, disk.server_root_id, farewell.server_uuid, crashed_epoch, wallNowMs(), lease_ttl_ms); + ASSERT_EQ(claim.kind, MountClaimResult::Claimed); +} + +/// Signals when a CAS mount starts waiting out a stale lease. +class StaleMountWaitChannel : public Poco::Channel +{ +public: + void log(const Poco::Message & msg) override + { + if (msg.getText().find("stale-looking mount lease") == std::string::npos) + return; + std::lock_guard lock(mutex); + ++waits_started; + cv.notify_all(); + } + + bool waitForWaitStarted(size_t count) + { + std::unique_lock lock(mutex); + return cv.wait_for(lock, std::chrono::seconds(30), [&] { return waits_started >= count; }); + } + +private: + std::mutex mutex; + std::condition_variable cv; + size_t waits_started = 0; +}; + +class CASStartupStaleMounts : public ::testing::Test +{ +public: + static void SetUpTestSuite() + { + DB::ServerUUID::setRandomForUnitTests(); + DB::clearDiskRegistry(); + DB::registerDisks(/*global_skip_access_check=*/true); + } + + static void TearDownTestSuite() + { + DB::clearDiskRegistry(); + } + +protected: + void SetUp() override + { + root = fs::temp_directory_path() + / ("cas_startup_stale_mounts_" + std::to_string(::getpid()) + "_" + + ::testing::UnitTest::GetInstance()->current_test_info()->name()); + fs::remove_all(root); + fs::create_directories(root); + } + + void TearDown() override + { + for (const auto & disk : disks) + if (auto created = getContext().context->getDisksMap(); created.contains(disk.name)) + created.at(disk.name)->shutdown(); + /// Restored only after the disks stop, so no CAS worker logs while the channel is swapped. + if (stale_mount_waits) + { + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + pool_logger.setChannel(previous_pool_channel); + pool_logger.setLevel(previous_pool_level); + } + fs::remove_all(root); + } + + /// Disks stay in the global context after a test, so every name is unique within the process. + static String uniqueDiskName(const String & prefix) + { + static std::atomic counter = 0; + return prefix + "_" + ::testing::UnitTest::GetInstance()->current_test_info()->name() + "_" + std::to_string(++counter); + } + + StaleCasDisk & addStaleDisk(const String & suffix) + { + StaleCasDisk disk{ + .name = uniqueDiskName("cas_stale_" + suffix), + .server_root_id = "stale_" + suffix, + .pool_dir = root / ("pool_" + suffix), + .scratch_dir = root / ("scratch_" + suffix), + }; + fs::create_directories(disk.scratch_dir); + seedStaleMount(disk); + return disks.emplace_back(std::move(disk)); + } + + StaleMountWaitChannel & watchStaleMountWaits() + { + Poco::Logger & pool_logger = Poco::Logger::get("CasPool"); + previous_pool_channel = Poco::AutoPtr(pool_logger.getChannel(), /*shared=*/true); + previous_pool_level = pool_logger.getLevel(); + stale_mount_waits = new StaleMountWaitChannel; + pool_logger.setChannel(stale_mount_waits); + pool_logger.setLevel(Poco::Message::PRIO_INFORMATION); + return *stale_mount_waits; + } + + fs::path root; + std::vector disks; + Poco::AutoPtr stale_mount_waits; + Poco::AutoPtr previous_pool_channel; + int previous_pool_level = 0; +}; + +} + +/// Each stale mount must be waited out once, but the waits of different disks are independent. +/// Waiting them one after another makes server start N × the observation window. +TEST_F(CASStartupStaleMounts, StaleMountsOfDifferentDisksAreWaitedOutConcurrently) +{ + constexpr size_t disk_count = 3; + for (size_t i = 0; i < disk_count; ++i) + addStaleDisk(std::to_string(i)); + ASSERT_FALSE(HasFatalFailure()); + + std::vector per_disk_ms(disk_count, 0); + std::vector errors(disk_count); + const auto start = std::chrono::steady_clock::now(); + { + std::vector loaders; + for (size_t i = 0; i < disk_count; ++i) + loaders.emplace_back([&, i] + { + try + { + DB::DiskFromAST::createCustomDisk(disks[i].diskFunctionAst(), getContext().context, /*attach=*/true); + } + catch (...) + { + errors[i] = std::current_exception(); + } + per_disk_ms[i] = elapsedMs(start); + }); + for (auto & loader : loaders) + loader.join(); + } + const uint64_t total_ms = elapsedMs(start); + + for (size_t i = 0; i < disk_count; ++i) + { + ASSERT_FALSE(errors[i]) << DB::getExceptionMessage(errors[i], false); + EXPECT_GE(per_disk_ms[i], observation_ms) << "disk " << i << " had no stale mount to wait out"; + } + EXPECT_LT(total_ms, 2 * observation_ms) + << disk_count << " stale mounts took " << total_ms << " ms to open; one observation window is " + << observation_ms << " ms, so the waits ran one after another"; +} + +/// A disk lookup by an unrelated table load must not queue behind a stale-mount wait: the wait is +/// up to a full lease TTL, and every table load and system log flush resolves disks. +TEST_F(CASStartupStaleMounts, DiskLookupDoesNotWaitForAStaleMountObservation) +{ + const StaleCasDisk & stale = addStaleDisk("0"); + ASSERT_FALSE(HasFatalFailure()); + StaleMountWaitChannel & waits = watchStaleMountWaits(); + + std::exception_ptr error; + std::thread loader([&] + { + try + { + DB::DiskFromAST::createCustomDisk(stale.diskFunctionAst(), getContext().context, /*attach=*/true); + } + catch (...) + { + error = std::current_exception(); + } + }); + + const bool wait_started = waits.waitForWaitStarted(1); + uint64_t lookup_ms = 0; + if (wait_started) + { + const auto lookup_start = std::chrono::steady_clock::now(); + (void)getContext().context->getDisksMap(); + lookup_ms = elapsedMs(lookup_start); + } + loader.join(); + + ASSERT_FALSE(error) << DB::getExceptionMessage(error, false); + ASSERT_TRUE(wait_started) << "the stale mount was never observed"; + EXPECT_LT(lookup_ms, observation_ms / 3) + << "a disk lookup took " << lookup_ms << " ms while a stale mount was being observed for up to " + << observation_ms << " ms"; +} + +/// Two table loads that name the same custom disk must share one creation: a second open of the +/// same CAS root while the first is still waiting out the stale mount would be a live twin. +TEST_F(CASStartupStaleMounts, ConcurrentCreationsOfOneDiskShareOneOpen) +{ + const StaleCasDisk & stale = addStaleDisk("0"); + ASSERT_FALSE(HasFatalFailure()); + StaleMountWaitChannel & waits = watchStaleMountWaits(); + + const auto config = stale.xmlConfig(); + std::atomic creator_calls = 0; + const auto creator = [&](const DB::DisksMap & disks_map) + { + ++creator_calls; + return DB::DiskFactory::instance().create( + stale.name, *config, "disk", getContext().context, disks_map, /*attach=*/true, /*custom_disk=*/true); + }; + + DB::DiskPtr first_disk; + DB::DiskPtr second_disk; + std::exception_ptr first_error; + std::exception_ptr second_error; + std::thread first([&] + { + try + { + first_disk = getContext().context->getOrCreateDisk(stale.name, creator); + } + catch (...) + { + first_error = std::current_exception(); + } + }); + /// Without this wait the second request could find the disk already created. + const bool first_is_waiting = waits.waitForWaitStarted(1); + std::thread second([&] + { + try + { + second_disk = getContext().context->getOrCreateDisk(stale.name, creator); + } + catch (...) + { + second_error = std::current_exception(); + } + }); + first.join(); + second.join(); + + ASSERT_TRUE(first_is_waiting) << "the stale mount was never observed"; + ASSERT_FALSE(first_error) << DB::getExceptionMessage(first_error, false); + ASSERT_FALSE(second_error) << DB::getExceptionMessage(second_error, false); + EXPECT_EQ(creator_calls.load(), 1u); + EXPECT_EQ(first_disk, second_disk); + EXPECT_EQ(first_disk, getContext().context->getDisk(stale.name)); +} + +/// A failed creation is not remembered: the next request for the name creates the disk again. +TEST_F(CASStartupStaleMounts, FailedDiskCreationIsRetriedByTheNextRequest) +{ + const String name = uniqueDiskName("failed_creation"); + std::vector calls; + EXPECT_THROW( + getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + calls.push_back("first"); + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "creation failed"); + }), + DB::Exception); + EXPECT_FALSE(getContext().context->getDisksMap().contains(name)); + EXPECT_THROW( + getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + calls.push_back("second"); + throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "creation failed again"); + }), + DB::Exception); + EXPECT_EQ(calls, (std::vector{"first", "second"})); +} + +/// A request that arrives while another creation of the same name fails gets that failure without +/// running its own creator. Each caller gets its own copy of the exception, of the original type. +TEST_F(CASStartupStaleMounts, ConcurrentRequestsShareOneFailure) +{ + const String name = uniqueDiskName("shared_failure"); + std::mutex mutex; + std::condition_variable cv; + bool first_started = false; + bool second_returned = false; + std::atomic creator_calls = 0; + + const auto failing_creator = [&](const DB::DisksMap &) -> DB::DiskPtr + { + ++creator_calls; + std::unique_lock lock(mutex); + first_started = true; + cv.notify_all(); + /// Gives the second request time to join this creation; it returns early only if it did not join. + /// A late scheduler can only cause a false failure, never a false pass. + cv.wait_for(lock, std::chrono::seconds(5), [&] { return second_returned; }); + DB::ErrnoException::throwWithErrno(DB::ErrorCodes::BAD_ARGUMENTS, ENOSPC, "creation failed"); + }; + + struct Outcome + { + bool got_errno_exception = false; + int code = 0; + int saved_errno = 0; + String message; + }; + const auto request = [&](const String & caller, Outcome & outcome) + { + try + { + getContext().context->getOrCreateDisk(name, failing_creator); + } + catch (DB::ErrnoException & e) + { + outcome.got_errno_exception = true; + outcome.code = e.code(); + outcome.saved_errno = e.getErrno(); + e.addMessage("seen by " + caller); + outcome.message = e.message(); + } + catch (...) + { + outcome.message = DB::getCurrentExceptionMessage(false); + } + }; + + Outcome first_outcome; + Outcome second_outcome; + std::thread first([&] { request("first", first_outcome); }); + bool first_is_creating = false; + { + std::unique_lock lock(mutex); + first_is_creating = cv.wait_for(lock, std::chrono::seconds(30), [&] { return first_started; }); + } + std::thread second([&] + { + request("second", second_outcome); + std::lock_guard lock(mutex); + second_returned = true; + cv.notify_all(); + }); + first.join(); + second.join(); + + ASSERT_TRUE(first_is_creating) << "the first creator never started"; + EXPECT_EQ(creator_calls.load(), 1u); + for (const Outcome * outcome : {&first_outcome, &second_outcome}) + { + EXPECT_TRUE(outcome->got_errno_exception) << outcome->message; + EXPECT_EQ(outcome->code, DB::ErrorCodes::BAD_ARGUMENTS); + EXPECT_EQ(outcome->saved_errno, ENOSPC); + } + EXPECT_TRUE(first_outcome.message.contains("seen by first")) << first_outcome.message; + EXPECT_FALSE(first_outcome.message.contains("seen by second")) << first_outcome.message; + EXPECT_TRUE(second_outcome.message.contains("seen by second")) << second_outcome.message; + EXPECT_FALSE(second_outcome.message.contains("seen by first")) << second_outcome.message; + EXPECT_FALSE(getContext().context->getDisksMap().contains(name)); +} + +/// A configuration reload during a custom disk creation must not open a configured disk of the +/// same name: both could describe the same storage. +TEST_F(CASStartupStaleMounts, ConfigReloadSkipsADiskBeingCreated) +{ + const String name = uniqueDiskName("reload_race"); + const fs::path config_disk_path = root / "config_disk"; + const fs::path custom_disk_path = root / "custom_disk"; + fs::create_directories(config_disk_path); + fs::create_directories(custom_disk_path); + + std::istringstream xml( + "<" + name + ">local" + config_disk_path.string() + + "/"); + Poco::AutoPtr reloaded_config(new Poco::Util::XMLConfiguration(xml)); + + DB::DiskPtr created; + const auto disk = getContext().context->getOrCreateDisk(name, [&](const DB::DisksMap &) -> DB::DiskPtr + { + getContext().context->updateStorageConfiguration(*reloaded_config); + created = std::make_shared(name, custom_disk_path.string() + "/"); + return created; + }); + + EXPECT_EQ(disk, created); + EXPECT_EQ(getContext().context->getDisk(name), created); +} diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp index 1109c5ae62036..2c411a8aa7bb6 100644 --- a/src/Interpreters/Context.cpp +++ b/src/Interpreters/Context.cpp @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -664,6 +665,8 @@ struct ContextSharedPart : boost::noncopyable mutable std::unique_ptr compression_codec_selector TSA_GUARDED_BY(mutex); /// Storage disk chooser for MergeTree engines mutable std::shared_ptr merge_tree_disk_selector TSA_GUARDED_BY(storage_policies_mutex); + /// Custom disks being created by `getOrCreateDisk`, by name + mutable std::unordered_map> disks_being_created TSA_GUARDED_BY(storage_policies_mutex); /// Storage policy chooser for MergeTree engines mutable std::shared_ptr merge_tree_storage_policy_selector TSA_GUARDED_BY(storage_policies_mutex); @@ -6775,20 +6778,90 @@ DiskPtr Context::getDisk(const String & name) const return disk_selector->get(name); } +namespace +{ + +/// All callers of one creation get the same stored exception. Each rethrows its own copy, so a +/// caller that modifies its exception (e.g. `addMessage`) does not race with the others. +DiskPtr getDiskCreationResult(const std::shared_future & creation) +{ + try + { + return creation.get(); + } + catch (...) + { + std::rethrow_exception(copyMutableException(std::current_exception())); + } +} + +} + DiskPtr Context::getOrCreateDisk(const String & name, DiskCreator creator) const { - std::lock_guard lock(shared->storage_policies_mutex); + std::promise promise; + std::shared_future creation; + bool is_creator = false; + DisksMap disks_map; + { + std::lock_guard lock(shared->storage_policies_mutex); - auto disk_selector = getDiskSelector(lock); + auto disk_selector = getDiskSelector(lock); + if (auto disk = disk_selector->tryGet(name)) + return disk; - auto disk = disk_selector->tryGet(name); - if (!disk) + if (auto it = shared->disks_being_created.find(name); it != shared->disks_being_created.end()) + { + creation = it->second; + } + else + { + disks_map = disk_selector->getDisksMap(); + creation = promise.get_future().share(); + shared->disks_being_created.emplace(name, creation); + is_creator = true; + } + } + + if (!is_creator) + return getDiskCreationResult(creation); + + /// A remote disk's startup can block on the network for a long time, so the creator runs without + /// `storage_policies_mutex`. Concurrent requests for the same name wait for this creation rather + /// than open the same storage a second time. + DiskPtr disk; + std::exception_ptr error; + try + { + disk = creator(disks_map); + } + catch (...) { - disk = creator(getDisksMap(lock)); - const_cast(disk_selector.get())->addToDiskMap(name, disk); + error = std::current_exception(); + } + + { + std::lock_guard lock(shared->storage_policies_mutex); + shared->disks_being_created.erase(name); + if (!error) + { + try + { + const_cast(getDiskSelector(lock).get())->addToDiskMap(name, disk); + } + catch (...) + { + error = std::current_exception(); + } + } } - return disk; + if (error) + promise.set_exception(error); + else + promise.set_value(disk); + + return getDiskCreationResult(creation); } StoragePolicyPtr Context::getStoragePolicy(const String & name) const @@ -6891,8 +6964,13 @@ void Context::updateStorageConfiguration(const Poco::Util::AbstractConfiguration std::lock_guard lock(shared->storage_policies_mutex); Strings disks_to_reinit; if (shared->merge_tree_disk_selector) - shared->merge_tree_disk_selector - = shared->merge_tree_disk_selector->updateFromConfig(config, "storage_configuration.disks", shared_from_this()); + { + NameSet disks_being_created; + for (const auto & [name, _] : shared->disks_being_created) + disks_being_created.insert(name); + shared->merge_tree_disk_selector = shared->merge_tree_disk_selector->updateFromConfig( + config, "storage_configuration.disks", shared_from_this(), disks_being_created); + } if (shared->merge_tree_storage_policy_selector) { diff --git a/src/Interpreters/Context.h b/src/Interpreters/Context.h index 5a9943af651d8..9427c4b39198e 100644 --- a/src/Interpreters/Context.h +++ b/src/Interpreters/Context.h @@ -1685,6 +1685,10 @@ class Context: public ContextData, public std::enable_shared_from_this /// Provides storage disks DiskPtr getDisk(const String & name) const; using DiskCreator = std::function; + /// Returns the disk `name`, creating it with `creator` if it does not exist yet. The creator runs + /// without the storage lock and gets a snapshot of the disks. Concurrent calls for the same name + /// share one creation: they all get its disk or a copy of its exception. A failed creation is not + /// remembered. The creator must not request a disk of its own name. DiskPtr getOrCreateDisk(const String & name, DiskCreator creator) const; StoragePoliciesMap getPoliciesMap() const;