Merge pull request #29794 from redpanda-data/ct/core-15649/gc-auto-pause

Author: oleiman

proto/redpanda/core/admin/internal/cloud_topics/v1/level_zero.proto

@@ -174,6 +174,7 @@ enum Status {
     L0_GC_STATUS_STOPPING = 3;
     L0_GC_STATUS_STOPPED = 4;
     L0_GC_STATUS_RESETTING = 5;
+    L0_GC_STATUS_SAFETY_BLOCKED = 6;
 }
 
 // Request the L0 size estimate for a single partition.

src/v/cloud_topics/level_zero/gc/BUILD

@@ -30,6 +30,7 @@ redpanda_cc_library(
         "//src/v/cloud_topics:logger",
         "//src/v/cloud_topics:object_utils",
         "//src/v/cluster",
+        "//src/v/config",
         "//src/v/ssx:semaphore",
         "//src/v/ssx:work_queue",
     ],

src/v/cloud_topics/level_zero/gc/level_zero_gc.cc

@@ -19,6 +19,7 @@
 #include "cluster/health_monitor_frontend.h"
 #include "cluster/members_table.h"
 #include "cluster/topic_table.h"
+#include "config/configuration.h"
 #include "ssx/semaphore.h"
 #include "ssx/work_queue.h"
 
@@ -32,7 +33,8 @@
 
 namespace {
 constexpr ss::lowres_clock::duration control_timeout = 5s;
-}
+constexpr ss::lowres_clock::duration health_report_query_timeout = 10s;
+} // namespace
 
 namespace cloud_topics {
 
@@ -484,7 +486,6 @@ class epoch_source_impl : public level_zero_gc::epoch_source {
          * Get a recent health report. Partitions use the health reporting
          * mechanism to self-report their max GC eligible epoch.
          */
-        constexpr auto health_report_query_timeout = 10s;
 
         auto health_report
           = co_await health_monitor_->local().get_cluster_health(
@@ -615,13 +616,82 @@ class node_info_impl : public level_zero_gc::node_info {
     seastar::sharded<cluster::members_table>* members_table_;
 };
 
+class cluster_safety_monitor : public level_zero_gc::safety_monitor {
+public:
+    explicit cluster_safety_monitor(
+      seastar::sharded<cluster::health_monitor_frontend>* health_monitor,
+      config::binding<std::chrono::milliseconds> check_interval)
+      : health_monitor_(health_monitor)
+      , check_interval_(std::move(check_interval))
+      , cached_result_{.ok = false, .reason = "awaiting first health check"}
+      , poll_loop_(do_poll_loop()) {}
+
+    result can_proceed() const override { return cached_result_; }
+
+    void start() override { started_ = true; }
+
+    seastar::future<> stop() override {
+        started_ = false;
+        as_.request_abort();
+        co_await std::exchange(poll_loop_, seastar::make_ready_future<>());
+    }
+
+private:
+    seastar::future<> do_poll_loop() noexcept {
+        while (!as_.abort_requested()) {
+            if (started_) {
+                auto poll_fut = co_await ss::coroutine::as_future(
+                  poll_health());
+                if (poll_fut.failed()) {
+                    auto ex = poll_fut.get_exception();
+                    cached_result_ = result{
+                      .ok = false,
+                      .reason = fmt::format("health check failed: {}", ex)};
+                }
+            }
+
+            auto sleep_fut = co_await seastar::coroutine::as_future(
+              seastar::sleep_abortable(check_interval_(), as_));
+            if (sleep_fut.failed()) {
+                sleep_fut.ignore_ready_future();
+                break;
+            }
+        }
+    }
+
+    seastar::future<> poll_health() {
+        auto overview
+          = co_await health_monitor_->local().get_cluster_health_overview(
+            model::timeout_clock::now() + health_report_query_timeout);
+
+        if (overview.is_healthy()) {
+            cached_result_ = result{.ok = true, .reason = std::nullopt};
+        } else {
+            cached_result_ = result{
+              .ok = false,
+              .reason = overview.unhealthy_reasons.empty()
+                          ? "cluster unhealthy"
+                          : overview.unhealthy_reasons.front()};
+        }
+    }
+
+    seastar::sharded<cluster::health_monitor_frontend>* health_monitor_;
+    config::binding<std::chrono::milliseconds> check_interval_;
+    result cached_result_;
+    bool started_{false};
+    seastar::abort_source as_;
+    seastar::future<> poll_loop_;
+};
+
 level_zero_gc::level_zero_gc(
   level_zero_gc_config config,
   std::unique_ptr<object_storage> storage,
   std::unique_ptr<epoch_source> epoch_source,
-  std::unique_ptr<node_info> node_info)
+  std::unique_ptr<node_info> node_info,
+  std::unique_ptr<safety_monitor> safety_monitor)
   : config_(std::move(config))
   , epoch_source_(std::move(epoch_source))
+  , safety_monitor_(std::move(safety_monitor))
   , should_run_(false) // begin in a stopped state
   , should_shutdown_(false)
   , worker_(worker())
@@ -654,7 +724,11 @@ level_zero_gc::level_zero_gc(
       std::make_unique<object_storage_remote_impl>(remote, std::move(bucket)),
       std::make_unique<epoch_source_impl>(
         health_monitor, controller_stm, topic_table),
-      std::make_unique<node_info_impl>(self, members_table)) {}
+      std::make_unique<node_info_impl>(self, members_table),
+      std::make_unique<cluster_safety_monitor>(
+        health_monitor,
+        config::shard_local_cfg()
+          .cloud_topics_gc_health_check_interval.bind())) {}
 
 level_zero_gc::~level_zero_gc() = default;
 
@@ -665,6 +739,7 @@ seastar::future<> level_zero_gc::start() {
     }
     vlog(cd_log.info, "Starting cloud topics L0 GC worker");
     delete_worker_->start();
+    safety_monitor_->start();
     should_run_ = true;
     worker_cv_.signal();
 }
@@ -687,6 +762,7 @@ seastar::future<> level_zero_gc::stop() {
     worker_cv_.signal();
     co_await delete_worker_->stop();
     co_await std::exchange(worker_, seastar::make_ready_future<>());
+    co_await safety_monitor_->stop();
     vlog(cd_log.info, "Stopped cloud_topics L0 GC worker");
 }
 
@@ -732,6 +808,8 @@ std::string_view to_string_view(level_zero_gc::state s) {
         return "level_zero_gc::state::stopping";
     case stopped:
         return "level_zero_gc::state::stopped";
+    case safety_blocked:
+        return "level_zero_gc::state::safety_blocked";
     }
     vunreachable("Unrecognized GC state: {}", s);
 }
@@ -746,7 +824,11 @@ auto level_zero_gc::get_state() const -> state {
         if (resetting_) {
             return state::resetting;
         }
-        return should_run_ ? state::running : state::paused;
+        if (!should_run_) {
+            return state::paused;
+        }
+        return safety_monitor_->can_proceed().ok ? state::running
+                                                 : state::safety_blocked;
     }();
     vlog(cd_log.debug, "cloud_topics L0 GC worker state: {}", st);
     return st;
@@ -779,6 +861,19 @@ seastar::future<> level_zero_gc::worker() {
             // ensure that the abort source is unreferenced at this time.
             asrc_ = {};
 
+            if (auto safety = safety_monitor_->can_proceed(); !safety.ok) {
+                vlog(
+                  cd_log.debug,
+                  "L0 GC blocked by safety monitor: {}",
+                  safety.reason.value_or("unknown"));
+                probe_.safety_blocked();
+                (co_await seastar::coroutine::as_future(
+                   seastar::sleep_abortable(
+                     config_.throttle_no_progress(), asrc_)))
+                  .ignore_ready_future();
+                continue;
+            }
+
             if (backoff.count() > 0) {
                 auto t0 = ss::lowres_clock::now();
                 (co_await seastar::coroutine::as_future(

src/v/cloud_topics/level_zero/gc/level_zero_gc.h

@@ -143,6 +143,74 @@ struct level_zero_gc_config {
     config::binding<std::chrono::milliseconds> throttle_no_progress;
 };
 
+/*
+ * State Machine
+ * =============
+ *
+ *  Construction
+ *       |
+ *       v
+ *  +--------+
+ *  | paused |<-----------------------------+
+ *  +---+----+                              |
+ *      | start()                           |
+ *      v                                   |
+ *  +----------------+                      |
+ *  | worker loop top|<-----------+         |
+ *  +-------+--------+            |         |
+ *          |                     |         |
+ *          | check safety        |         |
+ *          |                     |         |
+ *     ok +-+-+ !ok               |         |
+ *    +---+   +----+              |         |
+ *    v            v              |         |
+ *  +---------+ +---------------+ |         |
+ *  | running | |safety_blocked | |         |
+ *  |         | |               | |         |
+ *  | backoff | | increment     | |         |
+ *  |  then   | |  probe,       | |         |
+ *  | collect | |  sleep        | |         |
+ *  +----+----+ +-------+-------+ |         |
+ *       |              |         |         |
+ *       +------+-------+         |         |
+ *              |                 |         |
+ *              +-----------------+         |
+ *                  loop back               |
+ *                                          |
+ *  pause() from running or safety_blocked  |
+ *  ----------------------------------------+
+ *
+ *  reset() from any non-stopped state:
+ *
+ *  +----------+
+ *  |resetting | drains delete worker, then
+ *  +----+-----+ restores prior run/pause state
+ *       |
+ *       v
+ *  was_running?
+ *    yes -> running
+ *    no  -> paused
+ *
+ *  stop() from any state:
+ *
+ *  +---------+  worker done  +---------+
+ *  |stopping |-------------->| stopped |
+ *  +---------+               +---------+
+ *                             (terminal)
+ *
+ *  The running/safety_blocked distinction is not an
+ *  explicit transition. Both are phases of the worker
+ *  loop -- each iteration checks can_proceed() and
+ *  takes the corresponding branch. The state returned
+ *  by get_state() reflects whichever branch would be
+ *  taken at query time.
+ *
+ *  get_state() priority:
+ *    should_shutdown_ > resetting_ > !should_run_ >
+ *    safety_monitor_.can_proceed()
+ *
+ *  start()/pause() block while resetting_ is true.
+ */
 class level_zero_gc {
 public:
     /*
@@ -253,6 +321,33 @@ class level_zero_gc {
         virtual size_t total_shards() const = 0;
     };
 
+    /// Interface for gating GC on system-level safety conditions.
+    ///
+    /// Production implementations poll external signals (e.g. cluster health)
+    /// in a background fiber and cache the result so that can_proceed() is
+    /// synchronous and cheap. This is orthogonal to admin pause/start — even
+    /// if an operator calls start(), GC will not proceed while the safety
+    /// monitor reports not-ok.
+    class safety_monitor {
+    public:
+        safety_monitor() = default;
+        safety_monitor(const safety_monitor&) = delete;
+        safety_monitor(safety_monitor&&) = delete;
+        safety_monitor& operator=(const safety_monitor&) = delete;
+        safety_monitor& operator=(safety_monitor&&) = delete;
+        virtual ~safety_monitor() = default;
+
+        struct result {
+            bool ok;
+            std::optional<ss::sstring> reason;
+        };
+
+        virtual result can_proceed() const = 0;
+
+        virtual void start() {}
+        virtual seastar::future<> stop() { return seastar::now(); }
+    };
+
 public:
     /*
      * Construct with the given storage and epoch providers. This interface is
@@ -262,7 +357,8 @@ class level_zero_gc {
       level_zero_gc_config,
       std::unique_ptr<object_storage>,
       std::unique_ptr<epoch_source>,
-      std::unique_ptr<node_info>);
+      std::unique_ptr<node_info>,
+      std::unique_ptr<safety_monitor>);
 
     /*
      * Construct with default implementations of storage and epoch providers.
@@ -308,13 +404,16 @@ class level_zero_gc {
      *   - resetting: reset() is draining in-flight work
      *   - stopping: stop() requested but there may be work still in flight
      *   - stopped: Permanently stopped.
+     *   - safety_blocked: GC is started but the safety monitor is preventing
+     *     collection (e.g. cluster unhealthy)
      */
     enum class state : uint8_t {
         paused,
         running,
         resetting,
         stopping,
         stopped,
+        safety_blocked,
     };
 
     /**
@@ -325,6 +424,7 @@ class level_zero_gc {
 private:
     level_zero_gc_config config_;
     std::unique_ptr<epoch_source> epoch_source_;
+    std::unique_ptr<safety_monitor> safety_monitor_;
 
     bool should_run_;
     bool should_shutdown_;

src/v/cloud_topics/level_zero/gc/level_zero_gc_probe.cc

@@ -95,6 +95,13 @@ void level_zero_gc_probe::setup_internal_metrics(bool disable) {
             "Cumulative time in seconds spent in backoff between L0 "
             "garbage collection rounds."),
           labels),
+        sm::make_counter(
+          "safety_blocked_rounds_total",
+          [this] { return safety_blocked_rounds_; },
+          sm::description(
+            "Number of L0 GC rounds skipped because the safety monitor "
+            "reported an unsafe condition."),
+          labels),
         sm::make_gauge(
           "min_partition_gc_epoch",
           [this] {

src/v/cloud_topics/level_zero/gc/level_zero_gc_probe.h

@@ -33,6 +33,7 @@ class level_zero_gc_probe {
     void list_error() { list_errors_++; }
     void delete_error() { delete_errors_++; }
     void add_backpressure(double seconds) { backpressure_seconds_ += seconds; }
+    void safety_blocked() { safety_blocked_rounds_++; }
     void set_max_gc_eligible_epoch(cluster_epoch epoch) {
         max_gc_eligible_epoch_ = epoch;
     }
@@ -59,6 +60,7 @@ class level_zero_gc_probe {
     uint64_t list_errors_{0};
     uint64_t delete_errors_{0};
     double backpressure_seconds_{0};
+    uint64_t safety_blocked_rounds_{0};
 
     std::optional<cluster_epoch> min_partition_gc_epoch_;
     std::optional<cluster_epoch> max_gc_eligible_epoch_;

src/v/cloud_topics/level_zero/gc/tests/level_zero_gc_mt_test.cc

@@ -171,6 +171,14 @@ class mt_node_info : public level_zero_gc::node_info {
     size_t total_shards() const override { return ss::smp::count; }
 };
 
+class safety_monitor_test_impl
+  : public cloud_topics::level_zero_gc::safety_monitor {
+public:
+    result can_proceed() const override {
+        return {.ok = true, .reason = std::nullopt};
+    }
+};
+
 /*
  * Test fixture for multithreaded GC tests.
  */
@@ -200,7 +208,9 @@ struct level_zero_gc_mt_test : public seastar_test {
               return std::make_unique<mt_epoch_source>(g_bucket_state.get());
           }),
           ss::sharded_parameter(
-            [] { return std::make_unique<mt_node_info>(); }));
+            [] { return std::make_unique<mt_node_info>(); }),
+          ss::sharded_parameter(
+            [] { return std::make_unique<safety_monitor_test_impl>(); }));
     }
 
     ss::future<> TearDownAsync() override {