Merge pull request #29830 from andrwng/ct-l1-sg

ct/l1: explicitly use metastore scheduling group

Author: andrwng

src/v/cloud_topics/app.cc

@@ -83,7 +83,8 @@ ss::future<> app::construct(
       ss::sharded_parameter([this] { return &l1_io.local(); }),
       config::node().l1_staging_path(),
       ss::sharded_parameter([&remote] { return &remote->local(); }),
-      bucket);
+      bucket,
+      scheduling_groups::instance().cloud_topics_metastore_sg());
 
     co_await construct_service(
       l1_metastore_router,

src/v/cloud_topics/level_one/domain/db_domain_manager.cc

@@ -197,12 +197,14 @@ db_domain_manager::db_domain_manager(
   std::filesystem::path staging_dir,
   cloud_io::remote* remote,
   cloud_storage_clients::bucket_name bucket,
-  io* object_io)
+  io* object_io,
+  ss::scheduling_group sg)
   : expected_term_(expected_term)
   , staging_dir_(std::move(staging_dir))
   , remote_(remote)
   , bucket_(std::move(bucket))
   , object_io_(object_io)
+  , sg_(sg)
   , stm_(std::move(stm))
   , gc_interval_(
       config::shard_local_cfg()
@@ -211,7 +213,9 @@ db_domain_manager::db_domain_manager(
 }
 
 void db_domain_manager::start() {
-    ssx::spawn_with_gate(gate_, [this] { return gc_loop(); });
+    ssx::spawn_with_gate(gate_, [this] {
+        return ss::with_scheduling_group(sg_, [this] { return gc_loop(); });
+    });
 }
 
 ss::future<> db_domain_manager::stop_and_wait() {
@@ -1375,7 +1379,7 @@ ss::future<std::expected<void, rpc::errc>> db_domain_manager::maybe_open_db() {
     vlog(
       cd_log.debug, "Opening database with expected term {}", expected_term_);
     auto db_res = co_await replicated_database::open(
-      expected_term_, stm_.get(), staging_dir_, remote_, bucket_, as_);
+      expected_term_, stm_.get(), staging_dir_, remote_, bucket_, as_, sg_);
     if (!db_res.has_value()) {
         co_return std::unexpected(
           log_and_convert(db_res.error(), "Failed to open database: "));
@@ -1559,7 +1563,7 @@ db_domain_manager::restore_domain(rpc::restore_domain_request req) {
       "Re-opening database with expected term {}",
       expected_term_);
     auto db_res = co_await replicated_database::open(
-      expected_term_, stm_.get(), staging_dir_, remote_, bucket_, as_);
+      expected_term_, stm_.get(), staging_dir_, remote_, bucket_, as_, sg_);
     if (!db_res.has_value()) {
         co_return rpc::restore_domain_reply{
           .ec = log_and_convert(db_res.error(), "Failed to reopen database: "),

src/v/cloud_topics/level_one/domain/db_domain_manager.h

@@ -17,6 +17,7 @@
 
 #include <seastar/core/abort_source.hh>
 #include <seastar/core/rwlock.hh>
+#include <seastar/core/scheduling.hh>
 
 namespace cloud_topics::l1 {
 class io;
@@ -33,7 +34,8 @@ class db_domain_manager final : public domain_manager {
       std::filesystem::path staging_dir,
       cloud_io::remote* remote,
       cloud_storage_clients::bucket_name bucket,
-      io* object_io);
+      io* object_io,
+      ss::scheduling_group sg);
 
     void start() override;
     ss::future<> stop_and_wait() override;
@@ -169,6 +171,7 @@ class db_domain_manager final : public domain_manager {
     cloud_io::remote* remote_;
     cloud_storage_clients::bucket_name bucket_;
     io* object_io_;
+    ss::scheduling_group sg_;
 
     ss::shared_ptr<stm> stm_;
 

src/v/cloud_topics/level_one/domain/domain_supervisor.cc

@@ -36,12 +36,14 @@ class domain_supervisor::impl {
       io* io,
       std::filesystem::path staging_dir,
       cloud_io::remote* remote,
-      cloud_storage_clients::bucket_name bucket)
+      cloud_storage_clients::bucket_name bucket,
+      ss::scheduling_group sg)
       : _controller(controller)
       , _object_io(io)
       , _staging_dir(std::move(staging_dir))
       , _remote(remote)
       , _bucket(std::move(bucket))
+      , _sg(sg)
       , _queue([](const std::exception_ptr& ex) {
           vlog(cd_log.error, "Unexpected domain supervisor error: {}", ex);
       }) {}
@@ -326,7 +328,8 @@ class domain_supervisor::impl {
               _staging_dir,
               _remote,
               _bucket,
-              _object_io);
+              _object_io,
+              _sg);
         } else {
             domain_mgr = ss::make_shared<simple_domain_manager>(
               stm_manager->get<simple_stm>(), _object_io);
@@ -340,6 +343,7 @@ class domain_supervisor::impl {
     std::filesystem::path _staging_dir;
     cloud_io::remote* _remote;
     cloud_storage_clients::bucket_name _bucket;
+    ss::scheduling_group _sg;
 
     // Queue to process async work associated with starting and stopping domain
     // managers when handling partition notifications.
@@ -360,10 +364,16 @@ domain_supervisor::domain_supervisor(
   io* io,
   std::filesystem::path staging_dir,
   cloud_io::remote* remote,
-  cloud_storage_clients::bucket_name bucket)
+  cloud_storage_clients::bucket_name bucket,
+  ss::scheduling_group sg)
   : _impl(
       std::make_unique<impl>(
-        controller, io, std::move(staging_dir), remote, std::move(bucket))) {}
+        controller,
+        io,
+        std::move(staging_dir),
+        remote,
+        std::move(bucket),
+        sg)) {}
 
 domain_supervisor::~domain_supervisor() = default;
 

src/v/cloud_topics/level_one/domain/domain_supervisor.h

@@ -13,6 +13,7 @@
 #include "model/fundamental.h"
 
 #include <seastar/core/future.hh>
+#include <seastar/core/scheduling.hh>
 #include <seastar/core/shared_ptr.hh>
 #include <seastar/util/optimized_optional.hh>
 
@@ -42,7 +43,8 @@ class domain_supervisor {
       io*,
       std::filesystem::path staging_dir,
       cloud_io::remote*,
-      cloud_storage_clients::bucket_name bucket);
+      cloud_storage_clients::bucket_name bucket,
+      ss::scheduling_group sg);
     domain_supervisor(const domain_supervisor&) = delete;
     domain_supervisor(domain_supervisor&&) = delete;
     domain_supervisor& operator=(const domain_supervisor&) = delete;

src/v/cloud_topics/level_one/domain/tests/db_domain_manager_test.cc

@@ -76,7 +76,8 @@ struct domain_manager_node {
           staging_directory.get_path(),
           remote,
           bucket,
-          &object_io);
+          &object_io,
+          ss::default_scheduling_group());
         if (start_gc) {
             mgr->start();
         }

src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc

@@ -61,7 +61,8 @@ replicated_database::open(
   const std::filesystem::path& staging_directory,
   cloud_io::remote* remote,
   const cloud_storage_clients::bucket_name& bucket,
-  ss::abort_source& as) {
+  ss::abort_source& as,
+  ss::scheduling_group sg) {
     auto term_result = co_await s->sync(std::chrono::seconds(30));
     if (!term_result.has_value()) {
         co_return std::unexpected(
@@ -142,6 +143,7 @@ replicated_database::open(
       lsm::database::open(
         lsm::options{
           .database_epoch = epoch(),
+          .compaction_scheduling_group = sg,
           .file_deletion_delay = absl::FromChrono(
             config::shard_local_cfg()
               .cloud_topics_long_term_file_deletion_delay()),
@@ -193,13 +195,15 @@ replicated_database::open(
         }
     }
     auto ret = std::unique_ptr<replicated_database>(
-      new replicated_database(term, domain_uuid, s, std::move(db), as));
+      new replicated_database(term, domain_uuid, s, std::move(db), as, sg));
     ret->start();
     co_return std::move(ret);
 }
 
 void replicated_database::start() {
-    ssx::spawn_with_gate(gate_, [this] { return apply_loop(); });
+    ssx::spawn_with_gate(gate_, [this] {
+        return ss::with_scheduling_group(sg_, [this] { return apply_loop(); });
+    });
 }
 
 ss::future<std::expected<void, replicated_database::error>>

src/v/cloud_topics/level_one/metastore/lsm/replicated_db.h

@@ -20,6 +20,8 @@
 #include "model/fundamental.h"
 #include "utils/detailed_error.h"
 
+#include <seastar/core/scheduling.hh>
+
 #include <expected>
 #include <filesystem>
 
@@ -63,7 +65,8 @@ class replicated_database {
       const std::filesystem::path& staging_directory,
       cloud_io::remote* remote,
       const cloud_storage_clients::bucket_name& bucket,
-      ss::abort_source& as);
+      ss::abort_source& as,
+      ss::scheduling_group sg);
 
     replicated_database(replicated_database&&) = delete;
     ~replicated_database() = default;
@@ -99,12 +102,14 @@ class replicated_database {
       domain_uuid domain_uuid,
       stm* s,
       lsm::database db,
-      ss::abort_source& as)
+      ss::abort_source& as,
+      ss::scheduling_group sg)
       : term_(term)
       , expected_domain_uuid_(domain_uuid)
       , stm_(s)
       , db_(std::move(db))
-      , as_(as) {}
+      , as_(as)
+      , sg_(sg) {}
 
     ss::future<> apply_loop();
 
@@ -133,6 +138,7 @@ class replicated_database {
 
     ss::gate gate_;
     ss::abort_source& as_;
+    ss::scheduling_group sg_;
 };
 
 } // namespace cloud_topics::l1

src/v/cloud_topics/level_one/metastore/lsm/tests/replicated_db_test.cc

@@ -82,7 +82,8 @@ struct replicated_db_node {
           staging_directory.get_path(),
           remote,
           bucket,
-          as);
+          as,
+          ss::default_scheduling_group());
         if (!ret.has_value()) {
             co_return std::unexpected(ret.error());
         }