Merge pull request #28996 from dotnwat/manual-backport-28935-v25.3.x-95

[v25.3.x] Manual backport of #28935 and #28947

Author: dotnwat

src/v/cloud_topics/level_zero/gc/BUILD

@@ -36,6 +36,7 @@ redpanda_cc_library(
         "//src/v/cloud_io:io_result",
         "//src/v/cloud_storage_clients",
         "//src/v/cloud_topics:types",
+        "//src/v/config",
         "//src/v/container:chunked_hash_map",
         "//src/v/model",
         "@seastar",

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

@@ -324,7 +324,7 @@ level_zero_gc::level_zero_gc(
   level_zero_gc_config config,
   std::unique_ptr<object_storage> storage,
   std::unique_ptr<epoch_source> epoch_source)
-  : config_(config)
+  : config_(std::move(config))
   , storage_(std::move(storage))
   , epoch_source_(std::move(epoch_source))
   , should_run_(false) // begin in a stopped state
@@ -337,10 +337,18 @@ level_zero_gc::level_zero_gc(
   cloud_storage_clients::bucket_name bucket,
   seastar::sharded<cluster::health_monitor_frontend>* health_monitor,
   seastar::sharded<cluster::controller_stm>* controller_stm,
-  seastar::sharded<cluster::topic_table>* topic_table,
-  level_zero_gc_config config)
+  seastar::sharded<cluster::topic_table>* topic_table)
   : level_zero_gc(
-      config,
+      level_zero_gc_config{
+        .deletion_grace_period
+        = config::shard_local_cfg()
+            .cloud_topics_short_term_gc_minimum_object_age.bind(),
+        .throttle_progress
+        = config::shard_local_cfg().cloud_topics_short_term_gc_interval.bind(),
+        .throttle_no_progress
+        = config::shard_local_cfg()
+            .cloud_topics_short_term_gc_backoff_interval.bind(),
+      },
       std::make_unique<object_storage_remote_impl>(remote, std::move(bucket)),
       std::make_unique<epoch_source_impl>(
         health_monitor, controller_stm, topic_table)) {}
@@ -402,16 +410,16 @@ seastar::future<> level_zero_gc::worker() {
             auto res = co_await try_to_collect();
             if (res.has_value()) {
                 if (res.value() > 0) {
-                    backoff = config_.throttle_progress;
+                    backoff = config_.throttle_progress();
                 } else {
-                    backoff = config_.throttle_no_progress;
+                    backoff = config_.throttle_no_progress();
                 }
             } else {
                 switch (res.error()) {
                 case collection_error::service_error:
                 case collection_error::invalid_object_name:
                 case collection_error::no_collectible_epoch:
-                    backoff = config_.throttle_no_progress;
+                    backoff = config_.throttle_no_progress();
                 }
             }
 
@@ -420,7 +428,7 @@ seastar::future<> level_zero_gc::worker() {
               cd_log.info,
               "Level zero GC restarting after error: {}",
               std::current_exception());
-            backoff = config_.throttle_no_progress;
+            backoff = config_.throttle_no_progress();
         }
     }
 
@@ -455,7 +463,7 @@ level_zero_gc::try_to_collect() {
     }
 
     const auto max_gc_birthday = std::chrono::system_clock::now()
-                                 - config_.deletion_grace_period;
+                                 - config_.deletion_grace_period();
 
     vlog(
       cd_log.debug,

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

@@ -13,6 +13,7 @@
 #include "cloud_storage_clients/client.h"
 #include "cloud_topics/level_zero/gc/level_zero_gc_probe.h"
 #include "cloud_topics/types.h"
+#include "config/property.h"
 #include "container/chunked_hash_map.h"
 
 #include <seastar/core/condition-variable.hh>
@@ -136,16 +137,9 @@ namespace cloud_topics {
  * epochs starting at 0.
  */
 struct level_zero_gc_config {
-    /*
-     * TODO(noah): the grace period controls the minimum age of objects before
-     * they are eligible for removal, and likely deserves to be a proper
-     * configuration tunable. The throttling parameters are for preventing the
-     * polling worker from spinning. The throttling policy is very crude, and
-     * will need to be revisited, but should keep things in check for now.
-     */
-    std::chrono::milliseconds deletion_grace_period{10s};
-    std::chrono::milliseconds throttle_progress{2s};
-    std::chrono::milliseconds throttle_no_progress{10s};
+    config::binding<std::chrono::milliseconds> deletion_grace_period;
+    config::binding<std::chrono::milliseconds> throttle_progress;
+    config::binding<std::chrono::milliseconds> throttle_no_progress;
 };
 
 class level_zero_gc {
@@ -249,8 +243,7 @@ class level_zero_gc {
       cloud_storage_clients::bucket_name,
       seastar::sharded<cluster::health_monitor_frontend>*,
       seastar::sharded<cluster::controller_stm>*,
-      seastar::sharded<cluster::topic_table>*,
-      level_zero_gc_config = {});
+      seastar::sharded<cluster::topic_table>*);
 
     /*
      * Request that GC be started or paused. These can be called multiple times

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

@@ -110,9 +110,12 @@ class LevelZeroGCTest : public testing::Test {
     LevelZeroGCTest()
       : gc(
           cloud_topics::level_zero_gc_config{
-            .deletion_grace_period = 12h,
-            .throttle_progress = 10ms,
-            .throttle_no_progress = 10ms,
+            .deletion_grace_period
+            = config::mock_binding<std::chrono::milliseconds>(12h),
+            .throttle_progress
+            = config::mock_binding<std::chrono::milliseconds>(10ms),
+            .throttle_no_progress
+            = config::mock_binding<std::chrono::milliseconds>(10ms),
           },
           std::make_unique<object_storage_test_impl>(&listed, &deleted),
           std::make_unique<epoch_source_test_impl>(&max_epoch)) {}

src/v/config/configuration.cc

@@ -4576,6 +4576,27 @@ configuration::configuration()
       "The local cache duration of a cluster wide epoch.",
       {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
       1min)
+  , cloud_topics_short_term_gc_minimum_object_age(
+      *this,
+      "cloud_topics_short_term_gc_minimum_object_age",
+      "The minimum age of an L0 object before it becomes eligible for garbage "
+      "collection.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      12h)
+  , cloud_topics_short_term_gc_interval(
+      *this,
+      "cloud_topics_short_term_gc_interval",
+      "The interval between invocations of the L0 garbage collection work loop "
+      "when progress is being made.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      10s)
+  , cloud_topics_short_term_gc_backoff_interval(
+      *this,
+      "cloud_topics_short_term_gc_backoff_interval",
+      "The interval between invocations of the L0 garbage collection work loop "
+      "when no progress is being made or errors are occurring.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      1min)
   , development_feature_property_testing_only(
       *this,
       "development_feature_property_testing_only",

src/v/config/configuration.h

@@ -823,6 +823,12 @@ struct configuration final : public config_store {
     property<std::chrono::milliseconds>
       cloud_topics_epoch_service_local_epoch_cache_duration;
 
+    property<std::chrono::milliseconds>
+      cloud_topics_short_term_gc_minimum_object_age;
+    property<std::chrono::milliseconds> cloud_topics_short_term_gc_interval;
+    property<std::chrono::milliseconds>
+      cloud_topics_short_term_gc_backoff_interval;
+
     development_feature_property<int> development_feature_property_testing_only;
 
 private:

tests/rptest/tests/cloud_topics/l0_gc_test.py

@@ -38,6 +38,9 @@ def __init__(self, test_context: TestContext):
             "cloud_topics_reconciliation_interval": 2000,
             "cloud_topics_epoch_service_epoch_increment_interval": 5000,
             "cloud_topics_epoch_service_local_epoch_cache_duration": 5000,
+            "cloud_topics_short_term_gc_minimum_object_age": 10000,
+            "cloud_topics_short_term_gc_interval": 2000,
+            "cloud_topics_short_term_gc_backoff_interval": 10000,
         }
         super(CloudTopicsL0GCTest, self).__init__(
             test_context=test_context,