forked from scylladb/scylla-cluster-tests
-
Notifications
You must be signed in to change notification settings - Fork 0
/
reduce_cluster_test.py
172 lines (142 loc) · 6.52 KB
/
reduce_cluster_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
#!/usr/bin/env python
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2016 ScyllaDB
# TODO: this test seem to totally unused, hence disabling all pylint checks
# pylint: disable=all
import logging
import datetime
import time
from sdcm.tester import ClusterTester
from sdcm.tester import teardown_on_exception
from sdcm.nemesis import Nemesis
from sdcm.nemesis import log_time_elapsed_and_status
class DecommissionNoAddMonkey(Nemesis):
@log_time_elapsed_and_status
def disrupt(self):
self.disrupt_nodetool_decommission(add_node=False)
self.monitoring_set.reconfigure_scylla_monitoring()
class ReduceClusterTest(ClusterTester):
"""
Test scylla cluster reduction (removing nodes to shrink cluster from initial cluster size to target cluster size).
"""
@teardown_on_exception
def setUp(self):
self.credentials = None
self.db_cluster = None
self.loaders = None
self.monitors = None
logging.getLogger('botocore').setLevel(logging.CRITICAL)
logging.getLogger('boto3').setLevel(logging.CRITICAL)
self.stress_thread = None
def wait_for_init(self, cluster_starting_size, cluster_target_size):
# We're starting the cluster with more than 3 nodes due to
# replication factor settings we're using with cassandra-stress
self._cluster_starting_size = cluster_starting_size
self._cluster_target_size = cluster_target_size
loader_info = {'n_nodes': 1, 'device_mappings': None,
'type': None}
db_info = {'n_nodes': self._cluster_starting_size,
'device_mappings': None, 'type': None}
monitor_info = {'n_nodes': 1, 'device_mappings': None,
'type': None}
self.init_resources(loader_info=loader_info, db_info=db_info,
monitor_info=monitor_info)
self.loaders.wait_for_init()
self.db_cluster.wait_for_init()
nodes_monitored = [node.private_ip_address for node in self.db_cluster.nodes]
nodes_monitored += [node.private_ip_address for node in self.loaders.nodes]
self.monitors.wait_for_init(targets=nodes_monitored)
def get_stress_cmd(self):
"""
Get a cassandra stress cmd string suitable for grow cluster purposes.
:param duration: Duration of stress (minutes).
:param threads: Number of threads used by cassandra stress.
:param population_size: Size of the -pop seq1..%s argument.
:return: Cassandra stress string
:rtype: basestring
"""
ip = self.db_cluster.get_node_private_ips()[0]
population_size = 1000000
duration = self.params.get('test_duration')
threads = 1000
return ("cassandra-stress write cl=QUORUM duration=%sm "
"-schema 'replication(factor=3)' -port jmx=6868 "
"-mode cql3 native -rate threads=%s "
"-pop seq=1..%s -node %s" %
(duration, threads, population_size, ip))
def reduce_cluster(self, cluster_starting_size, cluster_target_size=3):
self.wait_for_init(cluster_starting_size, cluster_target_size)
nodes_to_remove = self._cluster_starting_size - self._cluster_target_size
# 60 minutes should be long enough for each node to do decommission
duration = 60 * nodes_to_remove
cs_thread_pool = self.run_stress_thread(stress_cmd=self.get_stress_cmd(),
duration=duration)
# Wait for cluster is filled with data
# Set space_node_threshold in config file for the size
self.db_cluster.wait_total_space_used_per_node()
start = datetime.datetime.now()
self.log.info('Starting to reduce cluster: %s', str(start))
self.db_cluster.add_nemesis(nemesis=DecommissionNoAddMonkey,
tester_obj=self)
# Have c-s run for 2 + 3 minutes before we start to do decommission
time.sleep(2 * 60)
while len(self.db_cluster.nodes) > cluster_target_size:
# Sleep 3 minutes before decommission
time.sleep(3 * 60)
# Run DecommissionNoAddMonkey once to decommission one node a time
self.db_cluster.start_nemesis(interval=10)
self.db_cluster.stop_nemesis(timeout=None)
end = datetime.datetime.now()
self.log.info('Reducing cluster finished: %s', str(end))
self.log.info('Reducing cluster costs: %s', str(end - start))
# Run 2 more minutes before stop c-s
time.sleep(2 * 60)
# Kill c-s when decommission is done
self.kill_stress_thread()
self.verify_stress_thread(cs_thread_pool=cs_thread_pool)
def test_reduce_4_to_3(self):
"""
Reduce cluster from 4 to 3
1) Start a 4 node cluster
2) Start cassandra-stress on the loader node
3) Decommission a node
4) Keep repeating 3) until we get to the target number of 3 nodes
"""
self.reduce_cluster(cluster_starting_size=4, cluster_target_size=3)
def test_reduce_5_to_3(self):
"""
Reduce cluster from 5 to 3
1) Start a 5 node cluster
2) Start cassandra-stress on the loader node
3) Decommission a node
4) Keep repeating 3) until we get to the target number of 3 nodes
"""
self.reduce_cluster(cluster_starting_size=5, cluster_target_size=3)
def test_reduce_10_to_3(self):
"""
Reduce cluster from 10 to 3
1) Start a 10 node cluster
2) Start cassandra-stress on the loader node
3) Decommission a node
4) Keep repeating 3) until we get to the target number of 3 nodes
"""
self.reduce_cluster(cluster_starting_size=10, cluster_target_size=3)
def test_reduce_30_to_3(self):
"""
Reduce cluster from 10 to 3
1) Start a 10 node cluster
2) Start cassandra-stress on the loader node
3) Decommission a node
4) Keep repeating 3) until we get to the target number of 3 nodes
"""
self.reduce_cluster(cluster_starting_size=30, cluster_target_size=3)