|
| 1 | +# This program is distributed in the hope that it will be useful, |
| 2 | +# but WITHOUT ANY WARRANTY; without even the implied warranty of |
| 3 | +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. |
| 4 | +# |
| 5 | +# See LICENSE for more details. |
| 6 | +# |
| 7 | +# Copyright (c) 2016 ScyllaDB |
| 8 | + |
| 9 | +import time |
| 10 | + |
| 11 | +from longevity_test import LongevityTest |
| 12 | +from sdcm.utils.adaptive_timeouts import adaptive_timeout, Operations |
| 13 | +from sdcm.utils.cluster_tools import group_nodes_by_dc_idx |
| 14 | +from sdcm.sct_events.system import InfoEvent |
| 15 | +from sdcm.sct_events import Severity |
| 16 | +from sdcm.cluster import MAX_TIME_WAIT_FOR_NEW_NODE_UP, BaseScyllaCluster |
| 17 | + |
| 18 | + |
| 19 | +class ScaleClusterTest(LongevityTest): |
| 20 | + @staticmethod |
| 21 | + def is_target_reached(current: list[int], target: list[int]) -> bool: |
| 22 | + """ Check that cluster size reached target size in each dc""" |
| 23 | + return all([x >= y for x, y in zip(current, target)]) |
| 24 | + |
| 25 | + @staticmethod |
| 26 | + def init_nodes(db_cluster: BaseScyllaCluster): |
| 27 | + """ |
| 28 | + method is required to be rewritten to support setup large clusters. |
| 29 | + """ |
| 30 | + db_cluster.set_seeds(first_only=True) |
| 31 | + db_cluster.wait_for_init(node_list=db_cluster.nodes, timeout=MAX_TIME_WAIT_FOR_NEW_NODE_UP) |
| 32 | + db_cluster.set_seeds() |
| 33 | + db_cluster.update_seed_provider() |
| 34 | + |
| 35 | + @property |
| 36 | + def cluster_target_size(self) -> list[int]: |
| 37 | + cluster_target_size = self.params.get('cluster_target_size') |
| 38 | + if not cluster_target_size: |
| 39 | + return [] |
| 40 | + return list(map(int, cluster_target_size.split())) if isinstance(cluster_target_size, str) else [cluster_target_size] |
| 41 | + |
| 42 | + def grow_to_cluster_target_size(self, cluster_target_size: list[int]): |
| 43 | + """ Bootstrap node in each dc in each rack while cluster size less than target size""" |
| 44 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 45 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 46 | + if self.is_target_reached(current_cluster_size, cluster_target_size): |
| 47 | + self.log.debug("Cluster has required size, no need to grow") |
| 48 | + return |
| 49 | + InfoEvent( |
| 50 | + message=f"Starting to grow cluster from {current_cluster_size} to {cluster_target_size}").publish() |
| 51 | + |
| 52 | + add_node_cnt = self.params.get('add_node_cnt') |
| 53 | + try: |
| 54 | + while not self.is_target_reached(current_cluster_size, cluster_target_size): |
| 55 | + for dcx, target in enumerate(cluster_target_size): |
| 56 | + if current_cluster_size[dcx] >= target: |
| 57 | + continue |
| 58 | + add_nodes_num = add_node_cnt if ( |
| 59 | + target - current_cluster_size[dcx]) >= add_node_cnt else target - current_cluster_size[dcx] |
| 60 | + |
| 61 | + for rack in range(self.db_cluster.racks_count): |
| 62 | + added_nodes = [] |
| 63 | + InfoEvent( |
| 64 | + message=f"Adding next number of nodes {add_nodes_num} to dc_idx {dcx} and rack {rack}").publish() |
| 65 | + added_nodes.extend(self.db_cluster.add_nodes( |
| 66 | + count=add_nodes_num, enable_auto_bootstrap=True, dc_idx=dcx, rack=rack)) |
| 67 | + self.monitors.reconfigure_scylla_monitoring() |
| 68 | + up_timeout = MAX_TIME_WAIT_FOR_NEW_NODE_UP |
| 69 | + with adaptive_timeout(Operations.NEW_NODE, node=self.db_cluster.data_nodes[0], timeout=up_timeout): |
| 70 | + self.db_cluster.wait_for_init( |
| 71 | + node_list=added_nodes, timeout=up_timeout, check_node_health=False) |
| 72 | + self.db_cluster.wait_for_nodes_up_and_normal(nodes=added_nodes) |
| 73 | + InfoEvent(f"New nodes up and normal {[node.name for node in added_nodes]}").publish() |
| 74 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 75 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 76 | + finally: |
| 77 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 78 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 79 | + InfoEvent(message=f"Grow cluster finished, cluster size is {current_cluster_size}").publish() |
| 80 | + |
| 81 | + def shrink_to_cluster_target_size(self, cluster_target_size: list[int]): |
| 82 | + """Decommission node in each dc in each rack while cluster size more than target size""" |
| 83 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 84 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 85 | + if self.is_target_reached(cluster_target_size, current_cluster_size): |
| 86 | + self.log.debug("Cluster has required size, no need to shrink") |
| 87 | + return |
| 88 | + InfoEvent( |
| 89 | + message=f"Starting to shrink cluster from {current_cluster_size} to {cluster_target_size}").publish() |
| 90 | + try: |
| 91 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 92 | + while not self.is_target_reached(cluster_target_size, current_cluster_size): |
| 93 | + for dcx, _ in enumerate(current_cluster_size): |
| 94 | + nodes_by_racks = self.db_cluster.get_nodes_per_datacenter_and_rack_idx(nodes_by_dcx[dcx]) |
| 95 | + for nodes in nodes_by_racks.values(): |
| 96 | + decommissioning_node = nodes[-1] |
| 97 | + decommissioning_node.running_nemesis = "Decommissioning node" |
| 98 | + self.db_cluster.decommission(node=decommissioning_node, timeout=7200) |
| 99 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 100 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 101 | + finally: |
| 102 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 103 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 104 | + InfoEvent( |
| 105 | + message=f"Reached cluster size {current_cluster_size}").publish() |
| 106 | + |
| 107 | + def create_schema(self): |
| 108 | + number_of_table = self.params.get( |
| 109 | + 'user_profile_table_count') or 0 |
| 110 | + cs_user_profiles = self.params.get('cs_user_profiles') |
| 111 | + keyspace_num = self.params.get('keyspace_num') |
| 112 | + if not number_of_table and not cs_user_profiles: |
| 113 | + self.log.debug("User schema will not be created") |
| 114 | + return |
| 115 | + if not cs_user_profiles: |
| 116 | + region_dc_names = self.db_cluster.get_datacenter_name_per_region(self.db_cluster.nodes) |
| 117 | + replication_factor = self.db_cluster.racks_count |
| 118 | + InfoEvent("Create keyspace and 100 empty tables").publish() |
| 119 | + for i in range(1, keyspace_num + 1): |
| 120 | + self.create_keyspace(keyspace_name=f"testing_keyspace_{i}", replication_factor={ |
| 121 | + dc_name: replication_factor for dc_name in region_dc_names.values()}) |
| 122 | + for j in range(1, number_of_table + 1): |
| 123 | + self.create_table(name=f"table_{j}", keyspace_name=f"testing_keyspace_{i}") |
| 124 | + InfoEvent(f"{keyspace_num} Keyspaces and {number_of_table} tables were created").publish() |
| 125 | + else: |
| 126 | + self._pre_create_templated_user_schema() |
| 127 | + |
| 128 | + def test_grow_shrink_cluster(self): |
| 129 | + """ |
| 130 | + Test allow to test cluster reaching target size by growing and shrinking. |
| 131 | + 1. Create schema if needed |
| 132 | + 2. Grow cluster to target size |
| 133 | + 3. if bootstrap failed during grow, try to shrink cluster to initial size |
| 134 | + 4. If shrink failed during step 3, just log error and finish test |
| 135 | +
|
| 136 | + """ |
| 137 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 138 | + init_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 139 | + InfoEvent(message=f"Cluster size is {init_cluster_size}").publish() |
| 140 | + self.create_schema() |
| 141 | + try: |
| 142 | + InfoEvent("Start grow cluster").publish() |
| 143 | + self.grow_to_cluster_target_size() |
| 144 | + except Exception as ex: # noqa: BLE001 |
| 145 | + self.log.error(f"Failed to grow cluster: {ex}") |
| 146 | + InfoEvent(f"Grow cluster failed with error: {ex}", severity=Severity.ERROR).publish() |
| 147 | + |
| 148 | + try: |
| 149 | + InfoEvent("Start shrink cluster").publish() |
| 150 | + self.shrink_to_cluster_target_size(init_cluster_size) |
| 151 | + except Exception as ex: # noqa: BLE001 |
| 152 | + self.log.error(f"Failed to shrink cluster: {ex}") |
| 153 | + InfoEvent(f"Shrink cluster failed with error: {ex}", severity=Severity.ERROR).publish() |
| 154 | + nodes_by_dcx = group_nodes_by_dc_idx(self.db_cluster.data_nodes) |
| 155 | + current_cluster_size = [len(nodes_by_dcx[dcx]) for dcx in sorted(nodes_by_dcx)] |
| 156 | + InfoEvent(message=f"Cluster size is {current_cluster_size}").publish() |
| 157 | + |
| 158 | + assert current_cluster_size == init_cluster_size, f"Cluster size {current_cluster_size} is not equal to initial {init_cluster_size}" |
| 159 | + |
| 160 | + def test_no_workloads_idle_custom_time(self): |
| 161 | + """ |
| 162 | + The aim of test is nemesis execution without any workload |
| 163 | + with configured user schema during idle_duration time. |
| 164 | + """ |
| 165 | + self.create_schema() |
| 166 | + self.grow_to_cluster_target_size(self.cluster_target_size) |
| 167 | + self.db_cluster.add_nemesis(nemesis=self.get_nemesis_class(), tester_obj=self) |
| 168 | + self.db_cluster.start_nemesis() |
| 169 | + duration = self.params.get('idle_duration') |
| 170 | + InfoEvent(f"Wait {duration} minutes while cluster resizing").publish() |
| 171 | + time.sleep(duration * 60) |
| 172 | + |
| 173 | + self.shrink_to_cluster_target_size(self.params.total_db_nodes) |
| 174 | + InfoEvent("Test done").publish() |
0 commit comments