api: Add REST endpoint for migration finalization

The endpoint is the following:

    POST /storage_service/vnode_tablet_migrations/keyspaces/{keyspace}/finalization

When called, it issues a `finalize_migration` topology request and waits
for its completion.

Signed-off-by: Nikos Dragazis <nikolaos.dragazis@scylladb.com>

api/api-doc/storage_service.json

@@ -3206,6 +3206,26 @@
              ]
          }]
       },
+      {
+         "path":"/storage_service/vnode_tablet_migrations/keyspaces/{keyspace}/finalization",
+         "operations":[{
+             "method":"POST",
+             "summary":"Finalize vnodes-to-tablets migration for all tables in a keyspace",
+             "type":"void",
+             "nickname":"finalize_vnode_tablet_migration",
+             "produces":["application/json"],
+             "parameters":[
+                 {
+                     "name":"keyspace",
+                     "description":"Keyspace name",
+                     "required":true,
+                     "allowMultiple":false,
+                     "type":"string",
+                     "paramType":"path"
+                 }
+             ]
+         }]
+      },
       {
          "path":"/storage_service/quiesce_topology",
          "operations":[

api/storage_service.cc

@@ -1753,6 +1753,20 @@ rest_set_vnode_tablet_migration_node_storage_mode(http_context& ctx, sharded<ser
     co_return json_void();
 }
 
+static
+future<json::json_return_type>
+rest_finalize_vnode_tablet_migration(http_context& ctx, sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
+    if (!ss.local().get_feature_service().vnodes_to_tablets_migrations) {
+        apilog.warn("finalize_vnode_tablet_migration: called before the cluster feature was enabled");
+        throw std::runtime_error("vnodes-to-tablets migration requires all nodes to support the VNODES_TO_TABLETS_MIGRATIONS cluster feature");
+    }
+    auto keyspace = validate_keyspace(ctx, req);
+    validate_keyspace(ctx, keyspace);
+
+    co_await ss.local().finalize_tablets_migration(keyspace);
+    co_return json_void();
+}
+
 static
 future<json::json_return_type>
 rest_quiesce_topology(sharded<service::storage_service>& ss, std::unique_ptr<http::request> req) {
@@ -1905,6 +1919,7 @@ void set_storage_service(http_context& ctx, routes& r, sharded<service::storage_
     ss::tablet_balancing_enable.set(r, rest_bind(rest_tablet_balancing_enable, ss));
     ss::create_vnode_tablet_migration.set(r, rest_bind(rest_create_vnode_tablet_migration, ctx, ss));
     ss::set_vnode_tablet_migration_node_storage_mode.set(r, rest_bind(rest_set_vnode_tablet_migration_node_storage_mode, ctx, ss));
+    ss::finalize_vnode_tablet_migration.set(r, rest_bind(rest_finalize_vnode_tablet_migration, ctx, ss));
     ss::quiesce_topology.set(r, rest_bind(rest_quiesce_topology, ss));
     sp::get_schema_versions.set(r, rest_bind(rest_get_schema_versions, ss));
     ss::drop_quarantined_sstables.set(r, rest_bind(rest_drop_quarantined_sstables, ctx, ss));
@@ -1986,6 +2001,7 @@ void unset_storage_service(http_context& ctx, routes& r) {
     ss::tablet_balancing_enable.unset(r);
     ss::create_vnode_tablet_migration.unset(r);
     ss::set_vnode_tablet_migration_node_storage_mode.unset(r);
+    ss::finalize_vnode_tablet_migration.unset(r);
     ss::quiesce_topology.unset(r);
     sp::get_schema_versions.unset(r);
     ss::drop_quarantined_sstables.unset(r);

service/storage_service.cc

@@ -4199,6 +4199,78 @@ future<> storage_service::set_node_intended_storage_mode(intended_storage_mode m
     slogger.info("Successfully set intended storage mode for node {} to {}", raft_server.id(), mode);
 }
 
+future<> storage_service::finalize_tablets_migration(const sstring& ks_name) {
+    if (this_shard_id() != 0) {
+        co_return co_await container().invoke_on(0, [&ks_name] (auto& ss) {
+            return ss.finalize_tablets_migration(ks_name);
+        });
+    }
+
+    slogger.info("Finalizing vnodes-to-tablets migration for keyspace '{}'", ks_name);
+
+    utils::UUID request_id;
+
+    while (true) {
+        auto guard = co_await _group0->client().start_operation(_group0_as, raft_timeout{});
+
+        auto& db = _db.local();
+        auto& ks = db.find_keyspace(ks_name);
+
+        if (ks.uses_tablets()) {
+            throw std::runtime_error(fmt::format("Keyspace '{}' already uses tablets", ks_name));
+        }
+
+        const auto& tm = get_token_metadata();
+        const auto& tablet_metadata = tm.tablets();
+
+        auto tables = ks.metadata()->tables();
+        if (tables.empty()) {
+            throw std::runtime_error(fmt::format("Keyspace '{}' has no tables", ks_name));
+        }
+
+        for (const auto& schema : tables) {
+            if (!tablet_metadata.has_tablet_map(schema->id())) {
+                throw std::runtime_error(fmt::format("Table {}.{} does not have a tablet map; "
+                    "all tables in keyspace '{}' must be prepared for migration before finalizing",
+                    ks_name, schema->cf_name(), ks_name));
+            }
+        }
+
+        slogger.info("All {} table(s) in keyspace '{}' have tablet maps, submitting finalization request",
+                     tables.size(), ks_name);
+
+        request_id = guard.new_group0_state_id();
+
+        topology_mutation_builder builder(guard.write_timestamp());
+        builder.queue_global_topology_request_id(request_id);
+
+        topology_request_tracking_mutation_builder rtbuilder(request_id, _feature_service.topology_requests_type_column);
+        rtbuilder.set("done", false)
+                 .set("start_time", db_clock::now())
+                 .set("request_type", global_topology_request::finalize_migration)
+                 .set_finalize_migration_data(ks_name);
+
+        topology_change change{{builder.build(), rtbuilder.build()}};
+        group0_command g0_cmd = _group0->client().prepare_command(std::move(change), guard,
+            fmt::format("finalize vnodes-to-tablets migration for keyspace '{}'", ks_name));
+
+        try {
+            co_await _group0->client().add_entry(std::move(g0_cmd), std::move(guard), _group0_as, raft_timeout{});
+        } catch (group0_concurrent_modification&) {
+            slogger.info("finalize_tablets_migration: concurrent modification, retrying");
+            continue;
+        }
+        break;
+    }
+
+    auto error = co_await wait_for_topology_request_completion(request_id);
+    if (!error.empty()) {
+        throw std::runtime_error(fmt::format("Migration finalization failed for keyspace '{}': {}", ks_name, error));
+    }
+
+    slogger.info("Successfully finalized vnodes-to-tablets migration for keyspace '{}'", ks_name);
+}
+
 future<> storage_service::process_tablet_split_candidate(table_id table) noexcept {
     tasks::task_info tablet_split_task_info;
 

service/storage_service.hh

@@ -289,6 +289,7 @@ public:
     // persists them to group0.
     future<> prepare_for_tablets_migration(const sstring& ks_name);
     future<> set_node_intended_storage_mode(intended_storage_mode mode);
+    future<> finalize_tablets_migration(const sstring& ks_name);
 
     void start_tablet_split_monitor();
 private:

test/pylib/rest_client.py

@@ -336,6 +336,10 @@ class ScyllaRESTAPIClient:
         """Set the node's intended storage mode to vnodes"""
         await self.client.put_json(f"/storage_service/vnode_tablet_migrations/node/storage_mode?intended_mode=vnodes", host=node_ip)
 
+    async def finalize_vnode_tablet_migration(self, node_ip: str, ks: str) -> None:
+        """Finalize vnodes-to-tablets migration for all tables in a keyspace"""
+        await self.client.post(f"/storage_service/vnode_tablet_migrations/keyspaces/{ks}/finalization", host=node_ip)
+
     async def keyspace_upgrade_sstables(self, node_ip: str, ks: str) -> None:
         await self.client.get(f"/storage_service/keyspace_upgrade_sstables/{ks}", host=node_ip)