From 6f9be0747e8efa7c7653abaed1788936d0b693fc Mon Sep 17 00:00:00 2001 From: ZeDRoman Date: Mon, 6 Feb 2023 16:40:02 +0300 Subject: [PATCH] controller: clean up topic orphan files When redpanda is restarted while delete operation is not finish Partition files might be left on disk. We need to cleanup orphan partition files after bootstrap. --- src/v/cluster/controller_backend.cc | 118 ++++++++++++++++++++++++++++ src/v/cluster/controller_backend.h | 14 ++++ src/v/storage/fs_utils.h | 20 +++++ src/v/storage/log_manager.cc | 95 ++++++++++++++++++++++ src/v/storage/log_manager.h | 12 +++ 5 files changed, 259 insertions(+) diff --git a/src/v/cluster/controller_backend.cc b/src/v/cluster/controller_backend.cc index 1d7ed6dee460a..f22ccb9daa0b4 100644 --- a/src/v/cluster/controller_backend.cc +++ b/src/v/cluster/controller_backend.cc @@ -43,6 +43,7 @@ #include #include +#include #include #include @@ -288,10 +289,77 @@ void controller_backend::setup_metrics() { "Number of partitions with ongoing/requested operations")), }); } +/** + * Create snapshot of topic table that contains all ntp that + * must be presented on local disk storage and their revision + * This snapshot will help to define orphan topic files + * If topic is not presented in this snapshot or its revision + * its revision is less than in this snapshot then this topic + * directory is orphan + */ +static absl::flat_hash_map +create_topic_table_snapshot( + ss::sharded& topics, model::node_id current_node) { + absl::flat_hash_map snapshot; + + for (const auto& nt : topics.local().all_topics()) { + auto ntp_view = model::topic_namespace_view(nt); + auto ntp_meta = topics.local().get_topic_metadata(ntp_view); + if (!ntp_meta) { + continue; + } + for (const auto& p : ntp_meta->get_assignments()) { + auto ntp = model::ntp(nt.ns, nt.tp, p.id); + if (!ntp_meta->is_topic_replicable()) { + snapshot.emplace(ntp, 0); + continue; + } + auto revision_id = ntp_meta->get_revision(); + if (cluster::contains_node(p.replicas, current_node)) { + snapshot.emplace(ntp, revision_id); + continue; + } + auto target_replica_set = topics.local().get_target_replica_set( + ntp); + if ( + target_replica_set + && cluster::contains_node(*target_replica_set, current_node)) { + snapshot.emplace(ntp, revision_id); + continue; + } + auto previous_replica_set = topics.local().get_previous_replica_set( + ntp); + if ( + previous_replica_set + && cluster::contains_node(*previous_replica_set, current_node)) { + snapshot.emplace(ntp, revision_id); + continue; + } + } + } + return snapshot; +} ss::future<> controller_backend::start() { setup_metrics(); return bootstrap_controller_backend().then([this] { + if (ss::this_shard_id() == cluster::controller_stm_shard) { + auto bootstrap_revision = _topics.local().last_applied_revision(); + auto snapshot = create_topic_table_snapshot(_topics, _self); + ssx::spawn_with_gate( + _gate, + [this, bootstrap_revision, snapshot = std::move(snapshot)] { + return clear_orphan_topic_files( + bootstrap_revision, std::move(snapshot)) + .handle_exception( + [](const std::exception_ptr& err) { + vlog( + clusterlog.error, + "Exception while cleaning oprhan files {}", + err); + }); + }); + } start_topics_reconciliation_loop(); _housekeeping_timer.set_callback([this] { housekeeping(); }); _housekeeping_timer.arm(_housekeeping_timer_interval); @@ -570,6 +638,56 @@ ss::future<> controller_backend::fetch_deltas() { }); } +/** + * Topic files is qualified as orphan if we don't have it in topic table + * or it's revision is less than revision it topic table + * And it's revision is less than topic table snapshot revision + */ +bool topic_files_are_orphan( + const model::ntp& ntp, + storage::partition_path::metadata ntp_directory_data, + const absl::flat_hash_map& + topic_table_snapshot, + model::revision_id last_applied_revision) { + vlog(clusterlog.debug, "Checking topic files for ntp {} are orphan", ntp); + + if (ntp_directory_data.revision_id > last_applied_revision) { + return false; + } + if ( + topic_table_snapshot.contains(ntp) + && ntp_directory_data.revision_id + >= topic_table_snapshot.find(ntp)->second) { + return false; + } + return true; +} + +ss::future<> controller_backend::clear_orphan_topic_files( + model::revision_id bootstrap_revision, + absl::flat_hash_map topic_table_snapshot) { + vlog( + clusterlog.info, + "Cleaning up orphan topic files. bootstrap_revision: {}", + bootstrap_revision); + // Init with default namespace to clean if there is no topics + absl::flat_hash_set namespaces = {{model::kafka_namespace}}; + for (const auto& t : _topics.local().all_topics()) { + namespaces.emplace(t.ns); + } + + return _storage.local().log_mgr().remove_orphan_files( + _data_directory, + std::move(namespaces), + [&, + bootstrap_revision, + topic_table_snapshot = std::move(topic_table_snapshot)]( + model::ntp ntp, storage::partition_path::metadata p) { + return topic_files_are_orphan( + ntp, p, topic_table_snapshot, bootstrap_revision); + }); +} + void controller_backend::start_topics_reconciliation_loop() { ssx::spawn_with_gate(_gate, [this] { return ss::do_until( diff --git a/src/v/cluster/controller_backend.h b/src/v/cluster/controller_backend.h index 37580299e021a..5268077d0cb16 100644 --- a/src/v/cluster/controller_backend.h +++ b/src/v/cluster/controller_backend.h @@ -270,6 +270,20 @@ class controller_backend // Topics ss::future<> bootstrap_controller_backend(); + /** + * Function that will clean orphan topic files on redpanda startup + * Orphan topic files is files that left on disk after some node + * manipullations and redpanda doesn't know about these files, so it is + * impossible to remove them with default approach. Currently we may leave + * orphan topic files when we restart redpanda while partition deletion + * operation was evaluating but hasn't finished yet. + * We assume that we can leave orphan files only on redpanda restart + * so we run clean process on startup, after bootstrap when redpanda + * already knows about all topics that it should containt on disk + **/ + ss::future<> clear_orphan_topic_files( + model::revision_id bootstrap_revision, + absl::flat_hash_map topic_table_snapshot); void start_topics_reconciliation_loop(); ss::future<> fetch_deltas(); diff --git a/src/v/storage/fs_utils.h b/src/v/storage/fs_utils.h index 2e5354dacaed8..50a4e40e3d147 100644 --- a/src/v/storage/fs_utils.h +++ b/src/v/storage/fs_utils.h @@ -67,6 +67,26 @@ class segment_full_path; class partition_path { public: + struct metadata { + model::partition_id partition_id; + model::revision_id revision_id; + }; + + /// Parse ntp directory name + static std::optional + parse_partition_directory(const ss::sstring& name) { + const std::regex re(R"(^(\d+)_(\d+)$)"); + std::cmatch match; + if (!std::regex_match(name.c_str(), match, re)) { + return std::nullopt; + } + return metadata{ + .partition_id = model::partition_id( + boost::lexical_cast(match[1].str())), + .revision_id = model::revision_id( + boost::lexical_cast(match[2].str()))}; + } + partition_path(const ntp_config& ntpc); operator std::filesystem::path() const { return make_path(); } diff --git a/src/v/storage/log_manager.cc b/src/v/storage/log_manager.cc index 83bfcde73f0d5..ac12aa60bfd65 100644 --- a/src/v/storage/log_manager.cc +++ b/src/v/storage/log_manager.cc @@ -9,9 +9,12 @@ #include "storage/log_manager.h" +#include "cluster/cluster_utils.h" +#include "cluster/topic_table.h" #include "config/configuration.h" #include "likely.h" #include "model/fundamental.h" +#include "model/metadata.h" #include "model/timestamp.h" #include "resource_mgmt/io_priority.h" #include "ssx/async-clear.h" @@ -46,6 +49,7 @@ #include #include #include +#include #include #include @@ -437,6 +441,97 @@ ss::future<> log_manager::remove(model::ntp ntp) { co_await dispatch_topic_dir_deletion(topic_dir); } +ss::future<> remove_orphan_partition_files( + ss::sstring topic_directory_path, + model::topic_namespace nt, + ss::noncopyable_function& + orphan_filter) { + return directory_walker::walk( + topic_directory_path, + [topic_directory_path, nt, &orphan_filter]( + ss::directory_entry entry) -> ss::future<> { + auto ntp_directory_data = partition_path::parse_partition_directory( + entry.name); + if (!ntp_directory_data) { + return ss::now(); + } + + auto ntp = model::ntp(nt.ns, nt.tp, ntp_directory_data->partition_id); + if (orphan_filter(ntp, *ntp_directory_data)) { + auto ntp_directory = std::filesystem::path(topic_directory_path) + / std::filesystem::path(entry.name); + vlog(stlog.info, "Cleaning up ntp directory {} ", ntp_directory); + return ss::recursive_remove_directory(ntp_directory) + .handle_exception_type([ntp_directory]( + std::filesystem:: + filesystem_error const& err) { + vlog( + stlog.error, + "Exception while cleaning oprhan files for {} Error: {}", + ntp_directory, + err); + }); + } + return ss::now(); + }); +} + +ss::future<> log_manager::remove_orphan_files( + ss::sstring data_directory_path, + absl::flat_hash_set namespaces, + ss::noncopyable_function + orphan_filter) { + auto data_directory_exist = co_await ss::file_exists(data_directory_path); + if (!data_directory_exist) { + co_return; + } + + for (const auto& ns : namespaces) { + auto namespace_directory = std::filesystem::path(data_directory_path) + / std::filesystem::path(ss::sstring(ns)); + auto namespace_directory_exist = co_await ss::file_exists( + namespace_directory.string()); + if (!namespace_directory_exist) { + continue; + } + co_await directory_walker::walk( + namespace_directory.string(), + [this, &namespace_directory, &ns, &orphan_filter]( + ss::directory_entry entry) -> ss::future<> { + if (entry.type != ss::directory_entry_type::directory) { + return ss::now(); + } + auto topic_directory = namespace_directory + / std::filesystem::path(entry.name); + return remove_orphan_partition_files( + topic_directory.string(), + model::topic_namespace(ns, model::topic(entry.name)), + orphan_filter) + .then([this, topic_directory]() { + vlog( + stlog.info, + "Trying to clean up topic directory {} ", + topic_directory); + return dispatch_topic_dir_deletion( + topic_directory.string()); + }) + .handle_exception_type( + [](std::filesystem::filesystem_error const& err) { + vlog( + stlog.error, + "Exception while cleaning oprhan files {}", + err); + }); + }) + .handle_exception_type( + [](std::filesystem::filesystem_error const& err) { + vlog( + stlog.error, "Exception while cleaning oprhan files {}", err); + }); + } + co_return; +} + ss::future<> log_manager::dispatch_topic_dir_deletion(ss::sstring dir) { return ss::smp::submit_to( 0, diff --git a/src/v/storage/log_manager.h b/src/v/storage/log_manager.h index 13f767152c564..1990e2f9780c4 100644 --- a/src/v/storage/log_manager.h +++ b/src/v/storage/log_manager.h @@ -11,9 +11,11 @@ #pragma once +#include "cluster/topic_table.h" #include "config/property.h" #include "features/feature_table.h" #include "model/fundamental.h" +#include "model/metadata.h" #include "random/simple_time_jitter.h" #include "seastarx.h" #include "storage/batch_cache.h" @@ -174,6 +176,16 @@ class log_manager { * rebalancing partitions across the cluster, etc... */ ss::future<> remove(model::ntp); + /** + * This function walsk through entire directory structure in an async fiber + * to remove all orphan files in that directory. It checks if file is orphan + * with special orphan_filter + */ + ss::future<> remove_orphan_files( + ss::sstring data_directory_path, + absl::flat_hash_set namespaces, + ss::noncopyable_function + orphan_filter); ss::future<> stop();