Merge pull request redpanda-data#16067 from mmaslankaprv/shutdown-tracking

Introduced partition shutdown watchdog timer

Author: mmaslankaprv

src/v/cluster/partition_manager.cc

@@ -19,6 +19,7 @@
 #include "cluster/archival_metadata_stm.h"
 #include "cluster/fwd.h"
 #include "cluster/logger.h"
+#include "cluster/partition.h"
 #include "cluster/partition_recovery_manager.h"
 #include "cluster/types.h"
 #include "config/configuration.h"
@@ -38,6 +39,7 @@
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/io_priority_class.hh>
+#include <seastar/core/lowres_clock.hh>
 #include <seastar/core/reactor.hh>
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/core/smp.hh>
@@ -57,15 +59,17 @@ partition_manager::partition_manager(
   ss::sharded<cloud_storage::cache>& cloud_storage_cache,
   ss::lw_shared_ptr<const archival::configuration> archival_conf,
   ss::sharded<features::feature_table>& feature_table,
-  ss::sharded<archival::upload_housekeeping_service>& upload_hks)
+  ss::sharded<archival::upload_housekeeping_service>& upload_hks,
+  config::binding<std::chrono::milliseconds> partition_shutdown_timeout)
   : _storage(storage.local())
   , _raft_manager(raft)
   , _partition_recovery_mgr(recovery_mgr)
   , _cloud_storage_api(cloud_storage_api)
   , _cloud_storage_cache(cloud_storage_cache)
   , _archival_conf(std::move(archival_conf))
   , _feature_table(feature_table)
-  , _upload_hks(upload_hks) {
+  , _upload_hks(upload_hks)
+  , _partition_shutdown_timeout(std::move(partition_shutdown_timeout)) {
     _leader_notify_handle
       = _raft_manager.local().register_leadership_notification(
         [this](
@@ -80,6 +84,8 @@ partition_manager::partition_manager(
                 }
             }
         });
+    _shutdown_watchdog.set_callback(
+      [this] { check_partitions_shutdown_state(); });
 }
 
 partition_manager::~partition_manager() {
@@ -101,6 +107,11 @@ partition_manager::get_topic_partition_table(
     return rs;
 }
 
+ss::future<> partition_manager::start() {
+    maybe_arm_shutdown_watchdog();
+    co_return;
+}
+
 ss::future<consensus_ptr> partition_manager::manage(
   storage::ntp_config ntp_cfg,
   raft::group_id group,
@@ -290,11 +301,17 @@ ss::future<> partition_manager::stop_partitions() {
 
 ss::future<>
 partition_manager::do_shutdown(ss::lw_shared_ptr<partition> partition) {
+    partition_shutdown_state shutdown_state(partition);
+    _partitions_shutting_down.push_back(shutdown_state);
+
     try {
         auto ntp = partition->ntp();
+        shutdown_state.update(partition_shutdown_stage::stopping_raft);
         co_await _raft_manager.local().shutdown(partition->raft());
         _unmanage_watchers.notify(ntp, model::topic_partition_view(ntp.tp));
+        shutdown_state.update(partition_shutdown_stage::stopping_partition);
         co_await partition->stop();
+        shutdown_state.update(partition_shutdown_stage::stopping_storage);
         co_await _storage.log_mgr().shutdown(partition->ntp());
     } catch (...) {
         vassert(
@@ -314,33 +331,34 @@ partition_manager::remove(const model::ntp& ntp, partition_removal_mode mode) {
     auto partition = get(ntp);
 
     if (!partition) {
-        return ss::make_exception_future<>(std::invalid_argument(fmt::format(
+        throw std::invalid_argument(fmt::format(
           "Can not remove partition. NTP {} is not present in partition "
           "manager",
-          ntp)));
+          ntp));
     }
+    partition_shutdown_state shutdown_state(partition);
+    _partitions_shutting_down.push_back(shutdown_state);
     auto group_id = partition->group();
 
     // remove partition from ntp & raft tables
     _ntp_table.erase(ntp);
     _raft_table.erase(group_id);
-
-    return _raft_manager.local()
-      .remove(partition->raft())
-      .then([this, ntp] {
-          _unmanage_watchers.notify(ntp, model::topic_partition_view(ntp.tp));
-      })
-      .then([partition] { return partition->stop(); })
-      .then([partition] { return partition->remove_persistent_state(); })
-      .then([this, ntp] { return _storage.log_mgr().remove(ntp); })
-      .then([this, partition, mode] {
-          if (mode == partition_removal_mode::global) {
-              return partition->finalize_remote_partition(_as);
-          } else {
-              return ss::now();
-          }
-      })
-      .finally([partition] {}); // in the end remove partition
+    shutdown_state.update(partition_shutdown_stage::stopping_raft);
+    co_await _raft_manager.local().remove(partition->raft());
+    _unmanage_watchers.notify(
+      ntp, model::topic_partition_view(partition->ntp().tp));
+    shutdown_state.update(partition_shutdown_stage::stopping_partition);
+    co_await partition->stop();
+    shutdown_state.update(partition_shutdown_stage::removing_persistent_state);
+    co_await partition->remove_persistent_state();
+    shutdown_state.update(partition_shutdown_stage::removing_storage);
+    co_await _storage.log_mgr().remove(partition->ntp());
+
+    if (mode == partition_removal_mode::global) {
+        shutdown_state.update(
+          partition_shutdown_stage::finalizing_remote_storage);
+        co_await partition->finalize_remote_partition(_as);
+    }
 }
 
 ss::future<> partition_manager::shutdown(const model::ntp& ntp) {
@@ -354,14 +372,46 @@ ss::future<> partition_manager::shutdown(const model::ntp& ntp) {
           "manager",
           ntp)));
     }
-
     // remove partition from ntp & raft tables
     _ntp_table.erase(ntp);
     _raft_table.erase(partition->group());
 
     return do_shutdown(partition);
 }
 
+partition_manager::partition_shutdown_state::partition_shutdown_state(
+  ss::lw_shared_ptr<cluster::partition> p)
+  : partition(std::move(p))
+  , stage(partition_manager::partition_shutdown_stage::shutdown_requested)
+  , last_update_timestamp(ss::lowres_clock::now()) {}
+
+void partition_manager::partition_shutdown_state::update(
+  partition_shutdown_stage s) {
+    stage = s;
+    last_update_timestamp = ss::lowres_clock::now();
+}
+
+void partition_manager::check_partitions_shutdown_state() {
+    const auto now = ss::lowres_clock::now();
+    for (auto& state : _partitions_shutting_down) {
+        if (state.last_update_timestamp < now - _partition_shutdown_timeout()) {
+            clusterlog.error(
+              "partition {} shutdown takes longer than expected, current "
+              "shutdown stage: {} time since last update: {} seconds",
+              state.partition->ntp(),
+              state.stage,
+              (now - state.last_update_timestamp) / 1s);
+        }
+    }
+    maybe_arm_shutdown_watchdog();
+}
+
+void partition_manager::maybe_arm_shutdown_watchdog() {
+    if (!_as.abort_requested()) {
+        _shutdown_watchdog.arm(_partition_shutdown_timeout() / 5);
+    }
+}
+
 uint64_t partition_manager::upload_backlog_size() const {
     uint64_t size = 0;
     for (const auto& [_, partition] : _ntp_table) {
@@ -415,4 +465,26 @@ partition_manager::get_cloud_cache_disk_usage_target() const {
       [](auto acc, auto update) { return acc + update; });
 }
 
+std::ostream& operator<<(
+  std::ostream& o, const partition_manager::partition_shutdown_stage& stage) {
+    switch (stage) {
+    case partition_manager::partition_shutdown_stage::shutdown_requested:
+        return o << "shutdown_requested";
+    case partition_manager::partition_shutdown_stage::stopping_raft:
+        return o << "stopping_raft";
+    case partition_manager::partition_shutdown_stage::removing_raft:
+        return o << "removing_raft";
+    case partition_manager::partition_shutdown_stage::stopping_partition:
+        return o << "stopping_partition";
+    case partition_manager::partition_shutdown_stage::removing_persistent_state:
+        return o << "removing_persistent_state";
+    case partition_manager::partition_shutdown_stage::stopping_storage:
+        return o << "stopping_storage";
+    case partition_manager::partition_shutdown_stage::removing_storage:
+        return o << "removing_storage";
+    case partition_manager::partition_shutdown_stage::finalizing_remote_storage:
+        return o << "finalizing_remote_storage";
+    }
+}
+
 } // namespace cluster

src/v/cluster/partition_manager.h

@@ -17,6 +17,7 @@
 #include "cluster/ntp_callbacks.h"
 #include "cluster/partition.h"
 #include "cluster/types.h"
+#include "config/property.h"
 #include "features/feature_table.h"
 #include "model/fundamental.h"
 #include "model/ktp.h"
@@ -25,10 +26,13 @@
 #include "raft/group_manager.h"
 #include "raft/heartbeat_manager.h"
 #include "storage/api.h"
+#include "utils/intrusive_list_helpers.h"
 #include "utils/named_type.h"
 
 #include <absl/container/flat_hash_map.h>
 
+#include <chrono>
+
 namespace cluster {
 class partition_manager
   : public ss::peering_sharded_service<partition_manager> {
@@ -44,7 +48,8 @@ class partition_manager
       ss::sharded<cloud_storage::cache>&,
       ss::lw_shared_ptr<const archival::configuration>,
       ss::sharded<features::feature_table>&,
-      ss::sharded<archival::upload_housekeeping_service>&);
+      ss::sharded<archival::upload_housekeeping_service>&,
+      config::binding<std::chrono::milliseconds>);
 
     ~partition_manager();
 
@@ -81,7 +86,7 @@ class partition_manager
         return nullptr;
     }
 
-    ss::future<> start() { return ss::now(); }
+    ss::future<> start();
     ss::future<> stop_partitions();
     ss::future<consensus_ptr> manage(
       storage::ntp_config,
@@ -206,6 +211,37 @@ class partition_manager
     }
 
 private:
+    enum class partition_shutdown_stage {
+        shutdown_requested,
+        stopping_raft,
+        removing_raft,
+        stopping_partition,
+        removing_persistent_state,
+        stopping_storage,
+        removing_storage,
+        finalizing_remote_storage
+    };
+
+    struct partition_shutdown_state {
+        explicit partition_shutdown_state(ss::lw_shared_ptr<partition>);
+
+        partition_shutdown_state(partition_shutdown_state&&) = delete;
+        partition_shutdown_state(const partition_shutdown_state&) = delete;
+        partition_shutdown_state& operator=(partition_shutdown_state&&)
+          = delete;
+        partition_shutdown_state& operator=(const partition_shutdown_state&)
+          = delete;
+        ~partition_shutdown_state() = default;
+
+        void update(partition_shutdown_stage);
+        // it is more convenient to keep the pointer to partition than an ntp
+        // copy.
+        ss::lw_shared_ptr<partition> partition;
+        partition_shutdown_stage stage;
+        ss::lowres_clock::time_point last_update_timestamp;
+        intrusive_list_hook hook;
+    };
+
     /// Download log if partition_recovery_manager is initialized.
     ///
     /// It might not be initialized if cloud storage is disable.
@@ -217,6 +253,9 @@ class partition_manager
 
     ss::future<> do_shutdown(ss::lw_shared_ptr<partition>);
 
+    void check_partitions_shutdown_state();
+
+    void maybe_arm_shutdown_watchdog();
     storage::api& _storage;
     /// used to wait for concurrent recoveries
     ss::sharded<raft::group_manager>& _raft_manager;
@@ -235,7 +274,11 @@ class partition_manager
     ss::lw_shared_ptr<const archival::configuration> _archival_conf;
     ss::sharded<features::feature_table>& _feature_table;
     ss::sharded<archival::upload_housekeeping_service>& _upload_hks;
+    intrusive_list<partition_shutdown_state, &partition_shutdown_state::hook>
+      _partitions_shutting_down;
     ss::gate _gate;
+    config::binding<std::chrono::milliseconds> _partition_shutdown_timeout;
+    ss::timer<> _shutdown_watchdog;
 
     // In general, all our background work is in partition objects which
     // have their own abort source.  This abort source is only for work that
@@ -250,5 +293,7 @@ class partition_manager
     state_machine_registry _stm_registry;
 
     friend std::ostream& operator<<(std::ostream&, const partition_manager&);
+    friend std::ostream& operator<<(
+      std::ostream&, const partition_manager::partition_shutdown_stage&);
 };
 } // namespace cluster

src/v/config/configuration.cc

@@ -261,6 +261,14 @@ configuration::configuration()
         .min = 0,     // It is not mandatory to reserve any capacity
         .max = 131072 // Same max as topic_partitions_per_shard
       })
+  , partition_manager_shutdown_watchdog_timeout(
+      *this,
+      "partition_manager_shutdown_watchdog_timeout",
+      "A threshold value to detect partitions which shutdown might have been "
+      "stuck. After this threshold a watchdog in partition manager will log "
+      "information about partition shutdown not making progress",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      30s)
   , admin_api_require_auth(
       *this,
       "admin_api_require_auth",

src/v/config/configuration.h

@@ -80,6 +80,8 @@ struct configuration final : public config_store {
     bounded_property<std::optional<int32_t>> topic_fds_per_partition;
     bounded_property<uint32_t> topic_partitions_per_shard;
     bounded_property<uint32_t> topic_partitions_reserve_shard0;
+    property<std::chrono::milliseconds>
+      partition_manager_shutdown_watchdog_timeout;
 
     // Admin API
     property<bool> admin_api_require_auth;

src/v/redpanda/application.cc

@@ -1441,7 +1441,11 @@ void application::wire_up_redpanda_services(
             }
         }),
       std::ref(feature_table),
-      std::ref(_archival_upload_housekeeping))
+      std::ref(_archival_upload_housekeeping),
+      ss::sharded_parameter([] {
+          return config::shard_local_cfg()
+            .partition_manager_shutdown_watchdog_timeout.bind();
+      }))
       .get();
     vlog(_log.info, "Partition manager started");
     construct_service(