diff --git a/api/storage_service.cc b/api/storage_service.cc
index af092bba72..0a32b66d67 100644
--- a/api/storage_service.cc
+++ b/api/storage_service.cc
@@ -610,6 +610,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         if (column_families.empty()) {
             column_families = map_keys(ctx.db.local().find_keyspace(keyspace).metadata().get()->cf_meta_data());
         }
+        apilog.debug("force_keyspace_compaction: keyspace={} tables={}", keyspace, column_families);
         return ctx.db.invoke_on_all([keyspace, column_families] (replica::database& db) -> future<> {
             auto table_ids = boost::copy_range<std::vector<utils::UUID>>(column_families | boost::adaptors::transformed([&] (auto& cf_name) {
                 return db.find_uuid(keyspace, cf_name);
@@ -634,6 +635,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         if (column_families.empty()) {
             column_families = map_keys(ctx.db.local().find_keyspace(keyspace).metadata().get()->cf_meta_data());
         }
+        apilog.info("force_keyspace_cleanup: keyspace={} tables={}", keyspace, column_families);
         return ss.local().is_cleanup_allowed(keyspace).then([&ctx, keyspace,
                 column_families = std::move(column_families)] (bool is_cleanup_allowed) mutable {
             if (!is_cleanup_allowed) {
@@ -653,7 +655,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
                 // as a table can be dropped during loop below, let's find it before issuing the cleanup request.
                 for (auto& id : table_ids) {
                     replica::table& t = db.find_column_family(id);
-                    co_await cm.perform_cleanup(owned_ranges_ptr, t.as_table_state());
+                    co_await t.perform_cleanup_compaction(owned_ranges_ptr);
                 }
                 co_return;
             }).then([]{
@@ -663,6 +665,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     });
 
     ss::perform_keyspace_offstrategy_compaction.set(r, wrap_ks_cf(ctx, [] (http_context& ctx, std::unique_ptr<request> req, sstring keyspace, std::vector<sstring> tables) -> future<json::json_return_type> {
+        apilog.info("perform_keyspace_offstrategy_compaction: keyspace={} tables={}", keyspace, tables);
         co_return co_await ctx.db.map_reduce0([&keyspace, &tables] (replica::database& db) -> future<bool> {
             bool needed = false;
             for (const auto& table : tables) {
@@ -676,6 +679,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::upgrade_sstables.set(r, wrap_ks_cf(ctx, [] (http_context& ctx, std::unique_ptr<request> req, sstring keyspace, std::vector<sstring> column_families) {
         bool exclude_current_version = req_param<bool>(*req, "exclude_current_version", false);
 
+        apilog.info("upgrade_sstables: keyspace={} tables={} exclude_current_version={}", keyspace, column_families, exclude_current_version);
         return ctx.db.invoke_on_all([=] (replica::database& db) {
             auto owned_ranges_ptr = compaction::make_owned_ranges_ptr(db.get_keyspace_local_ranges(keyspace));
             return do_for_each(column_families, [=, &db](sstring cfname) {
@@ -691,6 +695,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::force_keyspace_flush.set(r, [&ctx](std::unique_ptr<request> req) -> future<json::json_return_type> {
         auto keyspace = validate_keyspace(ctx, req->param);
         auto column_families = parse_tables(keyspace, ctx, req->query_parameters, "cf");
+        apilog.info("perform_keyspace_flush: keyspace={} tables={}", keyspace, column_families);
         auto &db = ctx.db.local();
         if (column_families.empty()) {
             co_await db.flush_on_all(keyspace);
@@ -702,6 +707,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
 
 
     ss::decommission.set(r, [&ss](std::unique_ptr<request> req) {
+        apilog.info("decommission");
         return ss.local().decommission().then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
@@ -717,6 +723,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::remove_node.set(r, [&ss](std::unique_ptr<request> req) {
         auto host_id = req->get_query_param("host_id");
         std::vector<sstring> ignore_nodes_strs= split(req->get_query_param("ignore_nodes"), ",");
+        apilog.info("remove_node: host_id={} ignore_nodes={}", host_id, ignore_nodes_strs);
         auto ignore_nodes = std::list<gms::inet_address>();
         for (std::string n : ignore_nodes_strs) {
             try {
@@ -789,6 +796,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     });
 
     ss::drain.set(r, [&ss](std::unique_ptr<request> req) {
+        apilog.info("drain");
         return ss.local().drain().then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
@@ -822,12 +830,14 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     });
 
     ss::stop_gossiping.set(r, [&ss](std::unique_ptr<request> req) {
+        apilog.info("stop_gossiping");
         return ss.local().stop_gossiping().then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
     });
 
     ss::start_gossiping.set(r, [&ss](std::unique_ptr<request> req) {
+        apilog.info("start_gossiping");
         return ss.local().start_gossiping().then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
@@ -930,6 +940,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
 
     ss::rebuild.set(r, [&ss](std::unique_ptr<request> req) {
         auto source_dc = req->get_query_param("source_dc");
+        apilog.info("rebuild: source_dc={}", source_dc);
         return ss.local().rebuild(std::move(source_dc)).then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
@@ -966,6 +977,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         // FIXME: We should truncate schema tables if more than one node in the cluster.
         auto& sp = service::get_storage_proxy();
         auto& fs = sp.local().features();
+        apilog.info("reset_local_schema");
         return db::schema_tables::recalculate_schema_version(sys_ks, sp, fs).then([] {
             return make_ready_future<json::json_return_type>(json_void());
         });
@@ -973,6 +985,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
 
     ss::set_trace_probability.set(r, [](std::unique_ptr<request> req) {
         auto probability = req->get_query_param("probability");
+        apilog.info("set_trace_probability: probability={}", probability);
         return futurize_invoke([probability] {
             double real_prob = std::stod(probability.c_str());
             return tracing::tracing::tracing_instance().invoke_on_all([real_prob] (auto& local_tracing) {
@@ -1010,6 +1023,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         auto ttl = req->get_query_param("ttl");
         auto threshold = req->get_query_param("threshold");
         auto fast = req->get_query_param("fast");
+        apilog.info("set_slow_query: enable={} ttl={} threshold={} fast={}", enable, ttl, threshold, fast);
         try {
             return tracing::tracing::tracing_instance().invoke_on_all([enable, ttl, threshold, fast] (auto& local_tracing) {
                 if (threshold != "") {
@@ -1036,6 +1050,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         auto keyspace = validate_keyspace(ctx, req->param);
         auto tables = parse_tables(keyspace, ctx, req->query_parameters, "cf");
 
+        apilog.info("enable_auto_compaction: keyspace={} tables={}", keyspace, tables);
         return set_tables_autocompaction(ctx, keyspace, tables, true);
     });
 
@@ -1043,6 +1058,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
         auto keyspace = validate_keyspace(ctx, req->param);
         auto tables = parse_tables(keyspace, ctx, req->query_parameters, "cf");
 
+        apilog.info("disable_auto_compaction: keyspace={} tables={}", keyspace, tables);
         return set_tables_autocompaction(ctx, keyspace, tables, false);
     });
 
diff --git a/replica/database.hh b/replica/database.hh
index 83f6f9b42e..e8675a63ce 100644
--- a/replica/database.hh
+++ b/replica/database.hh
@@ -909,6 +909,8 @@ public:
     // a future<bool> that is resolved when offstrategy_compaction completes.
     // The future value is true iff offstrategy compaction was required.
     future<bool> perform_offstrategy_compaction();
+    future<> perform_cleanup_compaction(owned_ranges_ptr sorted_owned_ranges);
+
     void set_compaction_strategy(sstables::compaction_strategy_type strategy);
     const sstables::compaction_strategy& get_compaction_strategy() const {
         return _compaction_strategy;
@@ -918,6 +920,14 @@ public:
         return _compaction_strategy;
     }
 
+    const compaction_manager& get_compaction_manager() const noexcept {
+        return _compaction_manager;
+    }
+
+    compaction_manager& get_compaction_manager() noexcept {
+        return _compaction_manager;
+    }
+
     table_stats& get_stats() const {
         return _stats;
     }
diff --git a/replica/table.cc b/replica/table.cc
index 8023a2d00f..aae0c8c6df 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -1121,6 +1121,11 @@ future<bool> table::perform_offstrategy_compaction() {
     return _compaction_manager.perform_offstrategy(as_table_state());
 }
 
+future<> table::perform_cleanup_compaction(compaction::owned_ranges_ptr sorted_owned_ranges) {
+    co_await flush();
+    co_await get_compaction_manager().perform_cleanup(std::move(sorted_owned_ranges), as_table_state());
+}
+
 void table::set_compaction_strategy(sstables::compaction_strategy_type strategy) {
     tlogger.debug("Setting compaction strategy of {}.{} to {}", _schema->ks_name(), _schema->cf_name(), sstables::compaction_strategy::name(strategy));
     auto new_cs = make_compaction_strategy(strategy, _schema->compaction_strategy_options());