Merge 'Purge old ip on change' from Petr Gusev

When a node changes IP address we need to remove its old IP from `system.peers` and gossiper.

We do this in `sync_raft_topology_nodes` when the new IP is saved into `system.peers` to avoid losing the mapping if the node crashes between deleting and saving the new IP. We also handle the possible duplicates in this case by dropping them on the read path when the node is restarted.

The PR also fixes the problem with old IPs getting resurrected when a node changes its IP address.
The following scenario is possible: a node `A` changes its IP from `ip1` to `ip2` with restart, other nodes are not yet aware of `ip2` so they keep gossiping `ip1`. After restart `A` receives `ip1` in a gossip message and calls `handle_major_state_change` since it considers it as a new node. Then `on_join` event is called on the gossiper notification handlers, we receive such event in `raft_ip_address_updater` and reverts the IP of the node A back to ip1.

To fix this we ensure that the new gossiper generation number is used when a node registers its IP address in `raft_address_map` at startup.

The `test_change_ip` is adjusted to ensure that the old IPs are properly removed in all cases, even if the node crashes.

Fixes scylladb#16886
Fixes scylladb#16691
Fixes scylladb#17199

Closes scylladb#17162

* github.com:scylladb/scylladb:
  test_change_ip: improve the test
  raft_ip_address_updater: remove stale IPs from gossiper
  raft_address_map: add my ip with the new generation
  system_keyspace::update_peer_info: check ep and host_id are not empty
  system_keyspace::update_peer_info: make host_id an explicit parameter
  system_keyspace::update_peer_info: remove any_set flag optimisation
  system_keyspace: remove duplicate ips for host_id
  system_keyspace: peers table: use coroutines
  storage_service::raft_ip_address_updater: log gossiper event name
  raft topology: ip change: purge old IP
  on_endpoint_change: coroutinize the lambda around sync_raft_topology_nodes

Author: kbr-scylla

db/system_keyspace.cc

@@ -1464,20 +1464,58 @@ struct local_cache {
     system_keyspace::bootstrap_state _state;
 };
 
+future<> system_keyspace::peers_table_read_fixup() {
+    assert(this_shard_id() == 0);
+    if (_peers_table_read_fixup_done) {
+        co_return;
+    }
+    _peers_table_read_fixup_done = true;
+
+    const auto cql = format("SELECT peer, host_id, WRITETIME(host_id) as ts from system.{}", PEERS);
+    std::unordered_map<utils::UUID, std::pair<net::inet_address, int64_t>> map{};
+    const auto cql_result = co_await execute_cql(cql);
+    for (const auto& row : *cql_result) {
+        const auto peer = row.get_as<net::inet_address>("peer");
+        if (!row.has("host_id")) {
+            slogger.error("Peer {} has no host_id in system.{}, the record is broken, removing it",
+                peer, system_keyspace::PEERS);
+            co_await remove_endpoint(gms::inet_address{peer});
+            continue;
+        }
+        const auto host_id = row.get_as<utils::UUID>("host_id");
+        const auto ts = row.get_as<int64_t>("ts");
+        const auto it = map.find(host_id);
+        if (it == map.end()) {
+            map.insert({host_id, {peer, ts}});
+            continue;
+        }
+        if (it->second.second >= ts) {
+            slogger.error("Peer {} with host_id {} has newer IP {} in system.{}, the record is stale, removing it",
+                peer, host_id, it->second.first, system_keyspace::PEERS);
+            co_await remove_endpoint(gms::inet_address{peer});
+        } else {
+            slogger.error("Peer {} with host_id {} has newer IP {} in system.{}, the record is stale, removing it",
+                it->second.first, host_id, peer, system_keyspace::PEERS);
+            co_await remove_endpoint(gms::inet_address{it->second.first});
+            it->second = {peer, ts};
+        }
+    }
+}
+
 future<std::unordered_map<gms::inet_address, locator::endpoint_dc_rack>> system_keyspace::load_dc_rack_info() {
-    auto msg = co_await execute_cql(format("SELECT peer, data_center, rack from system.{}", PEERS));
+    co_await peers_table_read_fixup();
+
+    const auto msg = co_await execute_cql(format("SELECT peer, data_center, rack from system.{}", PEERS));
 
     std::unordered_map<gms::inet_address, locator::endpoint_dc_rack> ret;
     for (const auto& row : *msg) {
-        net::inet_address peer = row.template get_as<net::inet_address>("peer");
         if (!row.has("data_center") || !row.has("rack")) {
             continue;
         }
-        gms::inet_address gms_addr(std::move(peer));
-        sstring dc = row.template get_as<sstring>("data_center");
-        sstring rack = row.template get_as<sstring>("rack");
-
-        ret.emplace(gms_addr, locator::endpoint_dc_rack{ dc, rack });
+        ret.emplace(row.get_as<net::inet_address>("peer"), locator::endpoint_dc_rack {
+            row.get_as<sstring>("data_center"),
+            row.get_as<sstring>("rack")
+        });
     }
 
     co_return ret;
@@ -1735,114 +1773,98 @@ static std::vector<cdc::generation_id_v2> decode_cdc_generations_ids(const set_t
     return gen_ids_list;
 }
 
-static locator::host_id get_host_id(const gms::inet_address& peer, const cql3::untyped_result_set::row& row) {
-    locator::host_id host_id;
-    if (row.has("host_id")) {
-        host_id = locator::host_id(row.get_as<utils::UUID>("host_id"));
-    }
-    if (!host_id) {
-        slogger.warn("Peer {} has no host_id in system.{}", peer, system_keyspace::PEERS);
-    }
-    return host_id;
-}
-
 future<std::unordered_map<gms::inet_address, std::unordered_set<dht::token>>> system_keyspace::load_tokens() {
-    sstring req = format("SELECT peer, host_id, tokens FROM system.{}", PEERS);
-    return execute_cql(req).then([] (::shared_ptr<cql3::untyped_result_set> cql_result) {
-        std::unordered_map<gms::inet_address, std::unordered_set<dht::token>> ret;
-        for (auto& row : *cql_result) {
-            auto peer = gms::inet_address(row.get_as<net::inet_address>("peer"));
-            if (!get_host_id(peer, row)) {
-                continue;
-            }
-            if (row.has("tokens")) {
-                ret.emplace(peer, decode_tokens(deserialize_set_column(*peers(), row, "tokens")));
-            }
+    co_await peers_table_read_fixup();
+
+    const sstring req = format("SELECT peer, tokens FROM system.{}", PEERS);
+    std::unordered_map<gms::inet_address, std::unordered_set<dht::token>> ret;
+    const auto cql_result = co_await execute_cql(req);
+    for (const auto& row : *cql_result) {
+        if (row.has("tokens")) {
+            ret.emplace(gms::inet_address(row.get_as<net::inet_address>("peer")),
+                decode_tokens(deserialize_set_column(*peers(), row, "tokens")));
         }
-        return ret;
-    });
+    }
+    co_return ret;
 }
 
 future<std::unordered_map<gms::inet_address, locator::host_id>> system_keyspace::load_host_ids() {
-    sstring req = format("SELECT peer, host_id FROM system.{}", PEERS);
-    return execute_cql(req).then([] (::shared_ptr<cql3::untyped_result_set> cql_result) {
-        std::unordered_map<gms::inet_address, locator::host_id> ret;
-        for (auto& row : *cql_result) {
-            auto peer = gms::inet_address(row.get_as<net::inet_address>("peer"));
-            if (auto host_id = get_host_id(peer, row)) {
-                ret.emplace(peer, host_id);
-            }
-        }
-        return ret;
-    });
+    co_await peers_table_read_fixup();
+
+    const sstring req = format("SELECT peer, host_id FROM system.{}", PEERS);
+    std::unordered_map<gms::inet_address, locator::host_id> ret;
+    const auto cql_result = co_await execute_cql(req);
+    for (const auto& row : *cql_result) {
+        ret.emplace(gms::inet_address(row.get_as<net::inet_address>("peer")),
+            locator::host_id(row.get_as<utils::UUID>("host_id")));
+    }
+    co_return ret;
 }
 
 future<std::vector<gms::inet_address>> system_keyspace::load_peers() {
-    auto res = co_await execute_cql(format("SELECT peer, host_id, tokens FROM system.{}", PEERS));
+    co_await peers_table_read_fixup();
+
+    const auto res = co_await execute_cql(format("SELECT peer, tokens FROM system.{}", PEERS));
     assert(res);
 
     std::vector<gms::inet_address> ret;
-    for (auto& row: *res) {
-        auto peer = gms::inet_address(row.get_as<net::inet_address>("peer"));
-        if (!get_host_id(peer, row)) {
-            continue;
-        }
+    for (const auto& row: *res) {
         if (!row.has("tokens")) {
             // Ignore rows that don't have tokens. Such rows may
             // be introduced by code that persists parts of peer
             // information (such as RAFT_ID) which may potentially
             // race with deleting a peer (during node removal).
             continue;
         }
-        ret.emplace_back(peer);
+        ret.emplace_back(gms::inet_address(row.get_as<net::inet_address>("peer")));
     }
     co_return ret;
 }
 
 future<std::unordered_map<gms::inet_address, sstring>> system_keyspace::load_peer_features() {
-    sstring req = format("SELECT peer, supported_features FROM system.{}", PEERS);
-    return execute_cql(req).then([] (::shared_ptr<cql3::untyped_result_set> cql_result) {
-        std::unordered_map<gms::inet_address, sstring> ret;
-        for (auto& row : *cql_result) {
-            if (row.has("supported_features")) {
-                ret.emplace(row.get_as<net::inet_address>("peer"),
-                        row.get_as<sstring>("supported_features"));
-            }
+    co_await peers_table_read_fixup();
+
+    const sstring req = format("SELECT peer, supported_features FROM system.{}", PEERS);
+    std::unordered_map<gms::inet_address, sstring> ret;
+    const auto cql_result = co_await execute_cql(req);
+    for (const auto& row : *cql_result) {
+        if (row.has("supported_features")) {
+            ret.emplace(row.get_as<net::inet_address>("peer"),
+                    row.get_as<sstring>("supported_features"));
         }
-        return ret;
-    });
+    }
+    co_return ret;
 }
 
 future<std::unordered_map<gms::inet_address, gms::inet_address>> system_keyspace::get_preferred_ips() {
-    sstring req = format("SELECT peer, preferred_ip FROM system.{}", PEERS);
-    return execute_cql(req).then([] (::shared_ptr<cql3::untyped_result_set> cql_res_set) {
-        std::unordered_map<gms::inet_address, gms::inet_address> res;
-
-        for (auto& r : *cql_res_set) {
-            if (r.has("preferred_ip")) {
-                res.emplace(gms::inet_address(r.get_as<net::inet_address>("peer")),
-                            gms::inet_address(r.get_as<net::inet_address>("preferred_ip")));
-            }
+    co_await peers_table_read_fixup();
+
+    const sstring req = format("SELECT peer, preferred_ip FROM system.{}", PEERS);
+    std::unordered_map<gms::inet_address, gms::inet_address> res;
+
+    const auto cql_result = co_await execute_cql(req);
+    for (const auto& r : *cql_result) {
+        if (r.has("preferred_ip")) {
+            res.emplace(gms::inet_address(r.get_as<net::inet_address>("peer")),
+                        gms::inet_address(r.get_as<net::inet_address>("preferred_ip")));
         }
+    }
 
-        return res;
-    });
+    co_return res;
 }
 
 namespace {
 template <typename T>
-static data_value_or_unset make_data_value_or_unset(const std::optional<T>& opt, bool& any_set) {
+static data_value_or_unset make_data_value_or_unset(const std::optional<T>& opt) {
     if (opt) {
-        any_set = true;
         return data_value(*opt);
     } else {
         return unset_value{};
     }
 };
 
-static data_value_or_unset make_data_value_or_unset(const std::optional<std::unordered_set<dht::token>>& opt, bool& any_set) {
+static data_value_or_unset make_data_value_or_unset(const std::optional<std::unordered_set<dht::token>>& opt) {
     if (opt) {
-        any_set = true;
         auto set_type = set_type_impl::get_instance(utf8_type, true);
         return make_set_value(set_type, prepare_tokens(*opt));
     } else {
@@ -1851,29 +1873,30 @@ static data_value_or_unset make_data_value_or_unset(const std::optional<std::uno
 };
 }
 
-future<> system_keyspace::update_peer_info(gms::inet_address ep, const peer_info& info) {
+future<> system_keyspace::update_peer_info(gms::inet_address ep, locator::host_id hid, const peer_info& info) {
+    if (ep == gms::inet_address{}) {
+        on_internal_error(slogger, format("update_peer_info called with empty inet_address, host_id {}", hid));
+    }
+    if (!hid) {
+        on_internal_error(slogger, format("update_peer_info called with empty host_id, ep {}", ep));
+    }
     if (_db.get_token_metadata().get_topology().is_me(ep)) {
         on_internal_error(slogger, format("update_peer_info called for this node: {}", ep));
     }
 
-    bool any_set = false;
     data_value_list values = {
         data_value_or_unset(data_value(ep.addr())),
-        make_data_value_or_unset(info.data_center, any_set),
-        make_data_value_or_unset(info.host_id, any_set),
-        make_data_value_or_unset(info.preferred_ip, any_set),
-        make_data_value_or_unset(info.rack, any_set),
-        make_data_value_or_unset(info.release_version, any_set),
-        make_data_value_or_unset(info.rpc_address, any_set),
-        make_data_value_or_unset(info.schema_version, any_set),
-        make_data_value_or_unset(info.tokens, any_set),
-        make_data_value_or_unset(info.supported_features, any_set),
+        make_data_value_or_unset(info.data_center),
+        data_value_or_unset(hid.id),
+        make_data_value_or_unset(info.preferred_ip),
+        make_data_value_or_unset(info.rack),
+        make_data_value_or_unset(info.release_version),
+        make_data_value_or_unset(info.rpc_address),
+        make_data_value_or_unset(info.schema_version),
+        make_data_value_or_unset(info.tokens),
+        make_data_value_or_unset(info.supported_features),
     };
 
-    if (!any_set) {
-        co_return;
-    }
-
     auto query = fmt::format("INSERT INTO system.{} "
             "(peer,data_center,host_id,preferred_ip,rack,release_version,rpc_address,schema_version,tokens,supported_features) VALUES"
             "(?,?,?,?,?,?,?,?,?,?)", PEERS);
@@ -1928,7 +1951,7 @@ future<> system_keyspace::update_schema_version(table_schema_version version) {
  * Remove stored tokens being used by another node
  */
 future<> system_keyspace::remove_endpoint(gms::inet_address ep) {
-    sstring req = format("DELETE FROM system.{} WHERE peer = ?", PEERS);
+    const sstring req = format("DELETE FROM system.{} WHERE peer = ?", PEERS);
     slogger.debug("DELETE FROM system.{} WHERE peer = {}", PEERS, ep);
     co_await execute_cql(req, ep.addr()).discard_result();
 }

db/system_keyspace.hh

@@ -114,6 +114,7 @@ class system_keyspace : public seastar::peering_sharded_service<system_keyspace>
     replica::database& _db;
     std::unique_ptr<local_cache> _cache;
     virtual_tables_registry _virtual_tables_registry;
+    bool _peers_table_read_fixup_done = false;
 
     static schema_ptr raft_snapshot_config();
     static schema_ptr local();
@@ -128,6 +129,15 @@ class system_keyspace : public seastar::peering_sharded_service<system_keyspace>
     static schema_ptr large_cells();
     static schema_ptr scylla_local();
     future<> force_blocking_flush(sstring cfname);
+    // This function is called when the system.peers table is read,
+    // and it fixes some types of inconsistencies that can occur
+    // due to node crashes:
+    //  * missing host_id. This is possible in the old versions of the code. Such records
+    //  are removed and the warning is written to the log.
+    //  * duplicate IPs for a given host_id. This is possible when some node changes its IP
+    //  and this node crashes after adding a new IP but before removing the old one. The
+    //  record with older timestamp is removed, the warning is written to the log.
+    future<> peers_table_read_fixup();
 public:
     static schema_ptr size_estimates();
 public:
@@ -265,7 +275,6 @@ public:
 public:
     struct peer_info {
         std::optional<sstring> data_center;
-        std::optional<utils::UUID> host_id;
         std::optional<net::inet_address> preferred_ip;
         std::optional<sstring> rack;
         std::optional<sstring> release_version;
@@ -275,7 +284,7 @@ public:
         std::optional<sstring> supported_features;
     };
 
-    future<> update_peer_info(gms::inet_address ep, const peer_info& info);
+    future<> update_peer_info(gms::inet_address ep, locator::host_id hid, const peer_info& info);
 
     future<> remove_endpoint(gms::inet_address ep);
 

main.cc

@@ -1792,8 +1792,10 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             // Set up group0 service earlier since it is needed by group0 setup just below
             ss.local().set_group0(group0_service, raft_topology_change_enabled);
 
+            const auto generation_number = gms::generation_type(sys_ks.local().increment_and_get_generation().get());
+
             // Load address_map from system.peers and subscribe to gossiper events to keep it updated.
-            ss.local().init_address_map(raft_address_map.local()).get();
+            ss.local().init_address_map(raft_address_map.local(), generation_number).get();
             auto cancel_address_map_subscription = defer_verbose_shutdown("storage service uninit address map", [&ss] {
                 ss.local().uninit_address_map().get();
             });
@@ -1814,7 +1816,7 @@ To start the scylla server proper, simply invoke as: scylla server (or just scyl
             }).get();
 
             with_scheduling_group(maintenance_scheduling_group, [&] {
-                return ss.local().join_cluster(sys_dist_ks, proxy, gossiper, service::start_hint_manager::yes);
+                return ss.local().join_cluster(sys_dist_ks, proxy, gossiper, service::start_hint_manager::yes, generation_number);
             }).get();
 
             sl_controller.invoke_on_all([&lifecycle_notifier] (qos::service_level_controller& controller) {

service/raft/raft_group0.cc

@@ -387,12 +387,6 @@ future<> raft_group0::start_server_for_group0(raft::group_id group0_id, service:
     // The address map may miss our own id in case we connect
     // to an existing Raft Group 0 leader.
     auto my_id = load_my_id();
-    _raft_gr.address_map().add_or_update_entry(my_id, _gossiper.get_broadcast_address());
-    // At this time the group registry is already up and running,
-    // so the address map is getting all the notifications from
-    // the gossiper. By reading the application state *after* subscribing to new gossip events,
-    // we ensure we haven't missed any IP update in the map.
-    load_initial_raft_address_map();
     group0_log.info("Server {} is starting group 0 with id {}", my_id, group0_id);
     auto srv_for_group0 = create_server_for_group0(group0_id, my_id, ss, qp, mm, topology_change_enabled);
     auto& persistence = srv_for_group0.persistence;
@@ -749,25 +743,6 @@ future<> raft_group0::setup_group0(
     co_await _client.set_group0_upgrade_state(group0_upgrade_state::use_post_raft_procedures);
 }
 
-void raft_group0::load_initial_raft_address_map() {
-    _gossiper.for_each_endpoint_state([this] (const gms::inet_address& ip_addr, const gms::endpoint_state& state) {
-        auto* value = state.get_application_state_ptr(gms::application_state::HOST_ID);
-        if (value == nullptr) {
-            return;
-        }
-        auto server_id = utils::UUID(value->value());
-        if (server_id == utils::UUID{}) {
-            upgrade_log.error("empty Host ID for host {} ", ip_addr);
-            return;
-        }
-        // The failure detector needs the IPs on all shards. We
-        // can safely overwrite existing entries since are loading
-        // them directly from gossiper app state - which is most
-        // recent.
-        _raft_gr.address_map().add_or_update_entry(raft::server_id{server_id}, ip_addr);
-    });
-}
-
 future<> raft_group0::finish_setup_after_join(service::storage_service& ss, cql3::query_processor& qp, service::migration_manager& mm, bool topology_change_enabled) {
     if (joined_group0()) {
         group0_log.info("finish_setup_after_join: group 0 ID present, loading server info.");

service/raft/raft_group0.hh

@@ -361,10 +361,6 @@ private:
     // Retries on raft::commit_status_unknown.
     future<> make_raft_config_nonvoter(const std::unordered_set<raft::server_id>&);
 
-    // Load the initial Raft <-> IP address map as seen by
-    // the gossiper.
-    void load_initial_raft_address_map();
-
     // Returns true if raft is enabled
     future<bool> use_raft();
 };