fix(get_rack_names_per_datacenter_and_rack_idx): support topology change

when there a topology change during this function being called

it can fail like the following:
```
File ".../sdcm/stress_thread.py", line 298, in _run_cs_stress
 stress_cmd = self.create_stress_cmd(cmd_runner, keyspace_idx, loader)
File ".../sdcm/stress_thread.py", line 187, in create_stress_cmd
 stress_cmd = self.adjust_cmd_connection_options(stress_cmd, loader, cmd_runner)
File ".../sdcm/stress_thread.py", line 159, in adjust_cmd_connection_options
 stress_cmd = self.adjust_cmd_node_option(stress_cmd, loader, cmd_runner)
File ".../sdcm/stress_thread.py", line 142, in adjust_cmd_node_option
 rack_names = self.loader_set.get_rack_names_per_datacenter_and_rack_idx(db_nodes=self.node_list)
File ".../sdcm/cluster.py", line 3224, in get_rack_names_per_datacenter_and_rack_idx
 rack_names_mapping[(region, rack)] = status[nodes[0]]['rack']
```

since one of the nodes it might use as a key, isn't in `nodetool status` anymore.

so we using only the intersection of the nodes from the argument
and from the response of `nodetool status`

Author: fruch

sdcm/cluster.py

@@ -12,6 +12,8 @@
 # Copyright (c) 2016 ScyllaDB
 
 # pylint: disable=too-many-lines
+from __future__ import annotations
+
 import contextlib
 import queue
 import logging
@@ -2602,7 +2604,7 @@ def check_node_health(self, retries: int = CHECK_NODE_HEALTH_RETRIES) -> None:
                          CHECK_NODE_HEALTH_RETRY_DELAY, self.name)
             time.sleep(CHECK_NODE_HEALTH_RETRY_DELAY)
 
-    def get_nodes_status(self):
+    def get_nodes_status(self) -> dict[BaseNode, dict]:
         nodes_status = {}
         try:
             statuses = self.parent_cluster.get_nodetool_status(verification_node=self)
@@ -2660,7 +2662,7 @@ def get_peers_info(self):
         return peers_details
 
     @retrying(n=5, sleep_time=10, raise_on_exceeded=False)
-    def get_gossip_info(self):
+    def get_gossip_info(self) -> dict[BaseNode, dict]:
         gossip_info = self.run_nodetool('gossipinfo', verbose=False, warning_event_on_exception=(Exception,),
                                         publish_event=False)
         LOGGER.debug("get_gossip_info: %s", gossip_info)
@@ -3180,7 +3182,7 @@ def tags(self) -> Dict[str, str]:
     def dead_nodes_ip_address_list(self):
         return [node.ip_address for node in self.dead_nodes_list]
 
-    def get_ip_to_node_map(self):
+    def get_ip_to_node_map(self) -> dict[str, BaseNode]:
         """returns {ip: node} map for all nodes in cluster to get node by ip"""
         return {ip: node for node in self.nodes for ip in node.get_all_ip_addresses()}
 
@@ -3219,8 +3221,12 @@ def get_rack_names_per_datacenter_and_rack_idx(self, db_nodes: list[BaseNode] |
         db_nodes = db_nodes if db_nodes else self.nodes
         status = db_nodes[0].get_nodes_status()
 
+        # intersection of asked nodes and nodes returned by nodetool status
+        # since topology might change during this command execution
+        actual_db_nodes = set(status.keys()).intersection(db_nodes)
+
         rack_names_mapping = {}
-        for (region, rack), nodes in self.nodes_by_racks_idx_and_regions(nodes=db_nodes).items():
+        for (region, rack), nodes in self.nodes_by_racks_idx_and_regions(nodes=actual_db_nodes).items():
             rack_names_mapping[(region, rack)] = status[nodes[0]]['rack']
 
         return rack_names_mapping