storage_service: add mark_nodes_as_cleanup_needed function

The function creates a mutation that sets cleanup to "needed" for each
normal node that, according to the erm, has data it does not own after
successful or unsuccessful topology operation.

service/storage_service.cc

@@ -2726,6 +2726,30 @@ class topology_coordinator {
         );
     }
 
+    std::vector<canonical_mutation> mark_nodes_as_cleanup_needed(node_to_work_on& node, bool rollback) {
+        auto& topo = _topo_sm._topology;
+        std::vector<canonical_mutation> muts;
+        muts.reserve(topo.normal_nodes.size());
+        std::unordered_set<locator::host_id> dirty_nodes;
+
+        for (auto& [_, erm] : _db.get_non_local_strategy_keyspaces_erms()) {
+            const std::unordered_set<locator::host_id>& nodes = rollback ? erm->get_all_pending_nodes() : erm->get_dirty_endpoints();
+            dirty_nodes.insert(nodes.begin(), nodes.end());
+        }
+
+        for (auto& n : dirty_nodes) {
+            auto id = raft::server_id(n.uuid());
+            // mark all nodes (except self) as cleanup needed
+            if (node.id != id) {
+                topology_mutation_builder builder(node.guard.write_timestamp());
+                builder.with_node(id).set("cleanup_status", cleanup_status::running);
+                muts.emplace_back(builder.build());
+                slogger.trace("raft topology: mark node {} as needed cleanup", id);
+            }
+        }
+        return muts;
+    }
+
     // Returns true if the state machine was transitioned into tablet migration path.
     future<bool> maybe_start_tablet_migration(group0_guard);