From 44e920cbb0e5faa5256291545d36e824e888b3af Mon Sep 17 00:00:00 2001
From: Benny Halevy <bhalevy@scylladb.com>
Date: Sun, 6 Nov 2022 18:20:49 +0200
Subject: [PATCH 1/3] api: storage_service: add logging for compaction
 operations et al

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
(cherry picked from 85523c45c0d5a849fa6548f8e8d76304bccae41f)
---
 api/storage_service.cc | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/api/storage_service.cc b/api/storage_service.cc
index af092bba72..092a8093a8 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) {
@@ -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);
     });
 

From fe8d8f97e282b59dd8941903263df3adab7fd070 Mon Sep 17 00:00:00 2001
From: Benny Halevy <bhalevy@scylladb.com>
Date: Sun, 6 Nov 2022 12:37:35 +0200
Subject: [PATCH 2/3] table: add perform_cleanup_compaction

Move the integration with compaction_manager
from the api layer to the tabel class so
it can also make sure the memtable is cleaned up in the next patch.

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
(cherry picked from commit fc278be6c4edc028c7908d3c2d236b479032b3f8)
---
 api/storage_service.cc |  2 +-
 replica/database.hh    | 10 ++++++++++
 replica/table.cc       |  4 ++++
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/api/storage_service.cc b/api/storage_service.cc
index 092a8093a8..0a32b66d67 100644
--- a/api/storage_service.cc
+++ b/api/storage_service.cc
@@ -655,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([]{
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..b482e42c70 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -1121,6 +1121,10 @@ 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 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());

From ea56ecace0d2488dfb77e5ffc751af49c8a5cbd9 Mon Sep 17 00:00:00 2001
From: Benny Halevy <bhalevy@scylladb.com>
Date: Sun, 6 Nov 2022 13:42:42 +0200
Subject: [PATCH 3/3] table: perform_cleanup_compaction: flush memtable

We don't explicitly cleanup the memtable, while
it might hold tokens disowned by the current node.

Flush the memtable before performing cleanup compaction
to make sure all tokens in the memtable are cleaned up.

Note that non-owned ranges are invalidate in the cache
in compaction_group::update_main_sstable_list_on_compaction_completion
using desc.ranges_for_cache_invalidation.

\Fixes #1239

Signed-off-by: Benny Halevy <bhalevy@scylladb.com>
(cherry picked from commit eb3a94e2bc45ffcf3b0c0d58c4d9afae9a1c94f5)
---
 replica/table.cc | 1 +
 1 file changed, 1 insertion(+)

diff --git a/replica/table.cc b/replica/table.cc
index b482e42c70..aae0c8c6df 100644
--- a/replica/table.cc
+++ b/replica/table.cc
@@ -1122,6 +1122,7 @@ future<bool> table::perform_offstrategy_compaction() {
 }
 
 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());
 }