Merge pull request #26790 from WillemKauf/tx_pr_6_recovery_rpc

`raft`: add `remake_learner_state` to `consensus` protocol [SC&R #5]

Author: WillemKauf

src/v/cluster/cluster_utils.cc

@@ -145,6 +145,7 @@ cluster::errc map_update_interruption_error_code(std::error_code ec) {
         case raft::errc::group_not_exists:
         case raft::errc::replicate_first_stage_exception:
         case raft::errc::invalid_input_records:
+        case raft::errc::not_learner:
             return errc::replication_error;
         }
         __builtin_unreachable();

src/v/cluster/controller.cc

@@ -655,6 +655,8 @@ ss::future<> controller::start(
       std::ref(_partition_manager),
       std::ref(_as));
 
+    co_await set_raft_manager_remake_cb();
+
     co_await _members_backend.invoke_on(
       members_manager::shard, &members_backend::start);
     co_await _config_manager.invoke_on(
@@ -919,6 +921,7 @@ ss::future<> controller::stop() {
     co_await _data_migration_frontend.stop();
     co_await _topic_mount_handler.stop();
     co_await _config_manager.stop();
+    co_await clear_raft_manager_remake_cb();
     co_await _api.stop();
     co_await _shard_balancer.stop();
     co_await _backend.stop();
@@ -1262,4 +1265,25 @@ controller::validate_configuration_invariants() {
     co_return invariants;
 }
 
+ss::future<std::error_code> controller::trigger_remake_cb(raft::group_id g) {
+    auto ec = co_await _api.local().remake_partition(g);
+    if (ec) {
+        vlog(clusterlog.warn, "Unable to remake group {}, {}", g, ec);
+    }
+    co_return ec;
+}
+
+ss::future<> controller::set_raft_manager_remake_cb() {
+    co_await _raft_manager.invoke_on_all([this](raft::group_manager& gm) {
+        gm.set_remake_cb(
+          [this](raft::group_id g) -> ss::future<std::error_code> {
+              return trigger_remake_cb(g);
+          });
+    });
+}
+
+ss::future<> controller::clear_raft_manager_remake_cb() {
+    co_await _raft_manager.invoke_on_all(&raft::group_manager::clear_remake_cb);
+}
+
 } // namespace cluster

src/v/cluster/controller.h

@@ -272,6 +272,15 @@ class controller {
 private:
     friend controller_probe;
 
+    using remake_cb_t
+      = ss::noncopyable_function<ss::future<std::error_code>(model::ntp)>;
+
+    ss::future<std::error_code> trigger_remake_cb(raft::group_id g);
+
+    ss::future<> set_raft_manager_remake_cb();
+
+    ss::future<> clear_raft_manager_remake_cb();
+
     /**
      * Create a \c bootstrap_cluster_cmd, replicate-and-wait it to the current
      * quorum, retry infinitely if replicate-and-wait fails.

src/v/cluster/controller_api.cc

@@ -579,17 +579,31 @@ controller_api::get_global_reconciliation_state(
     co_return state;
 }
 
-ss::future<std::error_code>
-controller_api::remake_partition(const model::ntp& ntp) {
-    auto shard_for_opt = shard_for(ntp);
+ss::future<std::error_code> controller_api::remake_partition(raft::group_id g) {
+    auto shard_for_opt = shard_for(g);
     if (!shard_for_opt.has_value()) {
         co_return errc::partition_not_exists;
     }
 
     auto shard = shard_for_opt.value();
+    auto ntp_opt = co_await _partition_manager.invoke_on(
+      shard, [g](cluster::partition_manager& pm) -> std::optional<model::ntp> {
+          auto p = pm.partition_for(g);
+          if (!p) {
+              return std::nullopt;
+          }
+          return p->ntp();
+      });
+
+    if (!ntp_opt.has_value()) {
+        co_return errc::partition_not_exists;
+    }
+
+    auto ntp = std::move(ntp_opt).value();
+
     co_return co_await _backend.invoke_on(
       shard, [&ntp](cluster::controller_backend& b) {
-          return b.remake_partition(ntp);
+          return b.remake_partition(std::move(ntp));
       });
 }
 

src/v/cluster/controller_api.h

@@ -88,7 +88,8 @@ class controller_api {
     std::optional<ss::shard_id> shard_for(const raft::group_id& group) const;
     std::optional<ss::shard_id> shard_for(const model::ntp& ntp) const;
 
-    ss::future<std::error_code> remake_partition(const model::ntp& ntp);
+    // Remakes the partition for the provided raft group.
+    ss::future<std::error_code> remake_partition(raft::group_id g);
 
 private:
     ss::future<std::optional<backend_operation>>

src/v/cluster/tests/remake_partition_tests.cc

@@ -113,7 +113,9 @@ FIXTURE_TEST(remake_partition_test, remake_partition_fixture) {
     add_topic(model::topic_namespace_view{test_ntp}).get();
     wait_for_leader(test_ntp).get();
 
-    auto ec = controller->get_api().local().remake_partition(test_ntp).get();
+    auto group
+      = controller->get_partition_manager().local().get(test_ntp)->group();
+    auto ec = controller->get_api().local().remake_partition(group).get();
     BOOST_REQUIRE_EQUAL(ec, cluster::errc::success);
 
     // Wait till partition is recreated
@@ -229,7 +231,8 @@ FIXTURE_TEST(remake_partition_with_produce_test, remake_partition_fixture) {
         BOOST_REQUIRE_EQUAL(records.size(), total_num_records);
     }
 
-    auto ec = controller->get_api().local().remake_partition(test_ntp).get();
+    auto group = p->group();
+    auto ec = controller->get_api().local().remake_partition(group).get();
     BOOST_REQUIRE_EQUAL(ec, cluster::errc::success);
 
     // Wait till partition is recreated

src/v/kafka/server/group.cc

@@ -2146,6 +2146,7 @@ kafka::error_code map_store_offset_error_code(std::error_code ec) {
         case raft::errc::group_not_exists:
         case raft::errc::replicate_first_stage_exception:
         case raft::errc::transfer_to_current_leader:
+        case raft::errc::not_learner:
             return error_code::unknown_server_error;
         }
     }

src/v/raft/buffered_protocol.cc

@@ -185,6 +185,20 @@ buffered_protocol::transfer_leadership(
       &consensus_client_protocol::transfer_leadership);
 }
 
+ss::future<result<remake_learner_state_reply>>
+buffered_protocol::remake_learner_state(
+  model::node_id target_node,
+  remake_learner_state_request req,
+  rpc::client_opts opts) {
+    return apply_with_gate(
+      _gate,
+      _base_protocol,
+      target_node,
+      std::move(req),
+      std::move(opts),
+      &consensus_client_protocol::remake_learner_state);
+}
+
 ss::future<bool> buffered_protocol::ensure_disconnect(model::node_id node_id) {
     return _base_protocol.ensure_disconnect(node_id);
 }

src/v/raft/buffered_protocol.h

@@ -152,6 +152,9 @@ class buffered_protocol : public consensus_client_protocol::impl {
 
     ss::future<> reset_backoff(model::node_id n) final;
 
+    ss::future<result<remake_learner_state_reply>> remake_learner_state(
+      model::node_id, remake_learner_state_request, rpc::client_opts) final;
+
     ss::future<> stop();
 
 private:

src/v/raft/consensus.cc

@@ -106,7 +106,8 @@ consensus::consensus(
   config::binding<std::chrono::milliseconds> disk_timeout,
   config::binding<bool> enable_longest_log_detection,
   consensus_client_protocol client,
-  consensus::leader_cb_t cb,
+  remake_cb_t remake_cb,
+  consensus::leader_cb_t leader_cb,
   storage::api& storage,
   std::optional<std::reference_wrapper<coordinated_recovery_throttle>>
     recovery_throttle,
@@ -123,7 +124,8 @@ consensus::consensus(
   , _disk_timeout(std::move(disk_timeout))
   , _enable_longest_log_detection(std::move(enable_longest_log_detection))
   , _client_protocol(client)
-  , _leader_notification(std::move(cb))
+  , _remake_notification(std::move(remake_cb))
+  , _leader_notification(std::move(leader_cb))
   , _fstats(_self)
   , _batcher(this, config::shard_local_cfg().raft_replicate_batch_window_size())
   , _event_manager(this)
@@ -1709,6 +1711,18 @@ ss::future<> consensus::write_last_applied(model::offset o) {
       storage::kvstore::key_space::consensus, std::move(key), std::move(val));
 }
 
+ss::future<> consensus::truncate_state(model::offset truncate_at) {
+    co_await _log->truncate(storage::truncate_config(truncate_at));
+    _probe->log_truncated();
+    // update flushed offset
+    _flushed_offset = std::min(
+      model::prev_offset(truncate_at), _flushed_offset);
+
+    co_await _configuration_manager.truncate(truncate_at);
+    _probe->configuration_update();
+    update_follower_stats(_configuration_manager.get_latest());
+}
+
 model::offset consensus::read_last_applied() const {
     const auto key = last_applied_key();
     auto value = _storage.kvs().get(
@@ -2235,7 +2249,6 @@ consensus::do_append_entries(append_entries_request&& r) {
           last_visible_index(),
           _last_leader_visible_offset,
           truncate_at);
-        _probe->log_truncated();
 
         _majority_replicated_index = std::min(
           model::prev_offset(truncate_at), _majority_replicated_index);
@@ -2248,17 +2261,7 @@ consensus::do_append_entries(append_entries_request&& r) {
           model::prev_offset(truncate_at), _flushed_offset);
 
         try {
-            co_await _log->truncate(storage::truncate_config(truncate_at));
-            // update flushed offset once again after truncation as flush is
-            // executed concurrently to append entries and it may race with
-            // the truncation
-            _flushed_offset = std::min(
-              model::prev_offset(truncate_at), _flushed_offset);
-
-            co_await _configuration_manager.truncate(truncate_at);
-            _probe->configuration_update();
-            update_follower_stats(_configuration_manager.get_latest());
-
+            co_await truncate_state(truncate_at);
             auto lstats = _log->offsets();
             if (unlikely(lstats.dirty_offset != adjusted_prev_log_index)) {
                 vlog(
@@ -4287,4 +4290,85 @@ size_t consensus::bytes_to_deliver_to_learners() const {
     return total;
 }
 
+ss::future<remake_learner_state_reply>
+consensus::remake_learner_state(vnode target) {
+    _probe->recovery_reset();
+    remake_learner_state_request req{
+      .node_id = _self,
+      .target_node_id = target,
+      .group = _group,
+      .term = _term};
+    vlog(_ctxlog.info, "Issuing remake group request {}", req);
+    static constexpr auto timeout = 10s;
+    result<remake_learner_state_reply> reply
+      = co_await _client_protocol.remake_learner_state(
+        target.id(), req, rpc::client_opts(timeout));
+    if (!reply) {
+        vlog(
+          _ctxlog.warn,
+          "Unable to issue remake group request {}, {}",
+          req,
+          reply.error());
+        co_return remake_learner_state_reply{};
+    }
+
+    co_return reply.value();
+}
+
+ss::future<remake_learner_state_reply>
+consensus::do_remake_learner_state(remake_learner_state_request req) {
+    remake_learner_state_reply reply{};
+    using is_success = remake_learner_state_reply::is_success;
+    try {
+        auto units = co_await _op_lock.get_units();
+
+        // Perform validation of request under _op_lock
+        auto maybe_err = [&]() -> std::optional<raft::errc> {
+            if (req.term != _term) {
+                return raft::errc::not_leader;
+            }
+            if (req.source_node() != _leader_id) {
+                return raft::errc::leadership_transfer_in_progress;
+            }
+            if (req.target_node() != _self) {
+                return raft::errc::invalid_target_node;
+            }
+            if (!is_learner()) {
+                return raft::errc::not_learner;
+            }
+            if (req.group != _group) {
+                return raft::errc::group_not_exists;
+            }
+
+            return std::nullopt;
+        }();
+
+        if (maybe_err.has_value()) {
+            reply.success = is_success::no;
+            vlog(
+              _ctxlog.warn,
+              "Unable to process remake group request {}, raft::errc {}",
+              req,
+              maybe_err.value());
+        } else {
+            auto cluster_err = co_await _remake_notification(req.group);
+            reply.success = cluster_err ? is_success::no : is_success::yes;
+            vlog(
+              _ctxlog.warn,
+              "Unable to process remake group request {}, cluster::errc {}",
+              req,
+              cluster_err);
+        }
+    } catch (...) {
+        vlog(
+          _ctxlog.warn,
+          "Unable to process remake group request {}, caught exception: {}",
+          req,
+          std::current_exception());
+        reply.success = is_success::no;
+    }
+
+    co_return reply;
+}
+
 } // namespace raft