forked from apache/cassandra-dtest
-
Notifications
You must be signed in to change notification settings - Fork 3
/
bootstrap_test.py
1144 lines (938 loc) · 49.4 KB
/
bootstrap_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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import os
import random
import re
import shutil
import tempfile
import threading
import time
import logging
import signal
from distutils.version import LooseVersion
from cassandra import ConsistencyLevel
from cassandra.concurrent import execute_concurrent_with_args
from cassandra.query import SimpleStatement
from ccmlib.node import NodeError, TimeoutError, ToolError, Node
import pytest
from dtest import Tester, create_ks, create_cf, data_size, mk_bman_path
from tools.assertions import (assert_almost_equal, assert_bootstrap_state, assert_not_running,
assert_one, assert_stderr_clean)
from tools.data import query_c1c2
from tools.intervention import InterruptBootstrap, KillOnBootstrap, KillOnReadyToBootstrap
from tools.misc import new_node, generate_ssl_stores
since = pytest.mark.since
ported_to_in_jvm = pytest.mark.ported_to_in_jvm
logger = logging.getLogger(__name__)
class BootstrapTester(Tester):
byteman_submit_path_pre_4_0 = mk_bman_path('pre4.0/stream_failure.btm')
byteman_submit_path_4_0 = mk_bman_path('4.0/stream_failure.btm')
@pytest.fixture(autouse=True)
def fixture_add_additional_log_patterns(self, fixture_dtest_setup):
fixture_dtest_setup.allow_log_errors = True
fixture_dtest_setup.ignore_log_patterns = (
# This one occurs when trying to send the migration to a
# node that hasn't started yet, and when it does, it gets
# replayed and everything is fine.
r'Can\'t send migration request: node.*is down',
# ignore streaming error during bootstrap
r'Exception encountered during startup',
r'Streaming error occurred'
)
def _base_bootstrap_test(self, bootstrap=None, bootstrap_from_version=None,
enable_ssl=None):
def default_bootstrap(cluster, token):
node2 = new_node(cluster)
node2.set_configuration_options(values={'initial_token': token})
node2.start(wait_for_binary_proto=True)
return node2
if bootstrap is None:
bootstrap = default_bootstrap
cluster = self.cluster
if enable_ssl:
logger.debug("***using internode ssl***")
generate_ssl_stores(self.fixture_dtest_setup.test_path)
cluster.enable_internode_ssl(self.fixture_dtest_setup.test_path)
tokens = cluster.balanced_tokens(2)
cluster.set_configuration_options(values={'num_tokens': 1})
logger.debug("[node1, node2] tokens: %r" % (tokens,))
keys = 10000
# Create a single node cluster
cluster.populate(1)
node1 = cluster.nodelist()[0]
if bootstrap_from_version:
logger.debug("starting source node on version {}".format(bootstrap_from_version))
node1.set_install_dir(version=bootstrap_from_version)
node1.set_configuration_options(values={'initial_token': tokens[0]})
cluster.start()
session = self.patient_cql_connection(node1)
create_ks(session, 'ks', 1)
create_cf(session, 'cf', columns={'c1': 'text', 'c2': 'text'})
# record the size before inserting any of our own data
empty_size = data_size(node1, 'ks','cf')
logger.debug("node1 empty size for ks.cf: %s" % float(empty_size))
insert_statement = session.prepare("INSERT INTO ks.cf (key, c1, c2) VALUES (?, 'value1', 'value2')")
execute_concurrent_with_args(session, insert_statement, [['k%d' % k] for k in range(keys)])
node1.flush()
node1.compact()
initial_size = data_size(node1,'ks','cf')
logger.debug("node1 size for ks.cf before bootstrapping node2: %s" % float(initial_size))
# Reads inserted data all during the bootstrap process. We shouldn't
# get any error
query_c1c2(session, random.randint(0, keys - 1), ConsistencyLevel.ONE)
session.shutdown()
# Bootstrapping a new node in the current version
node2 = bootstrap(cluster, tokens[1])
node2.compact()
node1.cleanup()
logger.debug("node1 size for ks.cf after cleanup: %s" % float(data_size(node1,'ks','cf')))
node1.compact()
logger.debug("node1 size for ks.cf after compacting: %s" % float(data_size(node1,'ks','cf')))
logger.debug("node2 size for ks.cf after compacting: %s" % float(data_size(node2,'ks','cf')))
size1 = float(data_size(node1,'ks','cf'))
size2 = float(data_size(node2,'ks','cf'))
assert_almost_equal(size1, size2, error=0.3)
assert_almost_equal(float(initial_size - empty_size), 2 * (size1 - float(empty_size)))
assert_bootstrap_state(self, node2, 'COMPLETED')
@pytest.mark.no_vnodes
def test_simple_bootstrap_with_ssl(self):
self._base_bootstrap_test(enable_ssl=True)
@pytest.mark.no_vnodes
def test_simple_bootstrap(self):
self._base_bootstrap_test()
@pytest.mark.no_vnodes
def test_bootstrap_on_write_survey(self):
def bootstrap_on_write_survey_and_join(cluster, token):
node2 = new_node(cluster)
node2.set_configuration_options(values={'initial_token': token})
node2.start(jvm_args=["-Dcassandra.write_survey=true"], wait_for_binary_proto=True)
assert len(node2.grep_log('Startup complete, but write survey mode is active, not becoming an active ring member.'))
# bootstrapping is considered complete if streaming is successful. Post CEP-21 this is distinct from
# fully joining the ring or leaving write survey mode
bootstrap_state = 'COMPLETED' if self.cluster.version() >= LooseVersion('5.1') else 'IN_PROGRESS'
assert_bootstrap_state(self, node2, bootstrap_state)
node2.nodetool("join")
assert len(node2.grep_log('Leaving write survey mode and joining ring at operator request'))
return node2
self._base_bootstrap_test(bootstrap_on_write_survey_and_join)
def _test_bootstrap_with_compatibility_flag_on(self, bootstrap_from_version):
def bootstrap_with_compatibility_flag_on(cluster, token):
node2 = new_node(cluster)
node2.set_configuration_options(values={'initial_token': token})
# cassandra.force_3_0_protocol_version parameter is needed to allow schema
# changes during the bootstrapping for upgrades from 3.0.14+ to anything upwards for 3.0.x or 3.x clusters.
# @jira_ticket CASSANDRA-13004 for detailed context on `cassandra.force_3_0_protocol_version` flag
node2.start(jvm_args=["-Dcassandra.force_3_0_protocol_version=true"],
wait_for_binary_proto=True)
return node2
self._base_bootstrap_test(bootstrap_with_compatibility_flag_on,
bootstrap_from_version=bootstrap_from_version)
@since('3.10')
@pytest.mark.no_vnodes
def test_simple_bootstrap_small_keepalive_period(self):
"""
@jira_ticket CASSANDRA-11841
Test that bootstrap completes if it takes longer than streaming_socket_timeout_in_ms or
2*streaming_keep_alive_period_in_secs to receive a single sstable
"""
cluster = self.cluster
yaml_opts = {'streaming_keep_alive_period_in_secs': 2}
if cluster.version() < '4.0':
yaml_opts['streaming_socket_timeout_in_ms'] = 1000
cluster.set_configuration_options(values=yaml_opts)
# Create a single node cluster
cluster.populate(1)
node1 = cluster.nodelist()[0]
logger.debug("Setting up byteman on {}".format(node1.name))
# set up byteman
node1.byteman_port = '8100'
node1.import_config_files()
cluster.start()
# Create more than one sstable larger than 1MB
node1.stress(['write', 'n=1K', '-rate', 'threads=8', '-schema',
'compaction(strategy=SizeTieredCompactionStrategy, enabled=false)'])
cluster.flush()
logger.debug("Submitting byteman script to {} to".format(node1.name))
# Sleep longer than streaming_socket_timeout_in_ms to make sure the node will not be killed
node1.byteman_submit([mk_bman_path('stream_5s_sleep.btm')])
# Bootstraping a new node with very small streaming_socket_timeout_in_ms
node2 = new_node(cluster)
node2.start(wait_for_binary_proto=True)
# Shouldn't fail due to streaming socket timeout timeout
assert_bootstrap_state(self, node2, 'COMPLETED')
if cluster.version() < '4.0':
for node in cluster.nodelist():
assert node.grep_log('Scheduling keep-alive task with 2s period.', filename='debug.log')
assert node.grep_log('Sending keep-alive', filename='debug.log')
assert node.grep_log('Received keep-alive', filename='debug.log')
def test_simple_bootstrap_nodata(self):
"""
@jira_ticket CASSANDRA-11010
Test that bootstrap completes if streaming from nodes with no data
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
# Create a two-node cluster
cluster.populate(2)
cluster.start()
# Bootstrapping a new node
node3 = new_node(cluster)
node3.start(wait_for_binary_proto=True)
assert_bootstrap_state(self, node3, 'COMPLETED')
def test_schema_removed_nodes(self):
"""
@jira_ticket CASSANDRA-16577
Test that nodes can bootstrap after a schema change performed with a node removed
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(2)
cluster.start()
node1, node2 = cluster.nodelist()
node2.decommission(force=cluster.version() > '4')
session = self.patient_cql_connection(node1)
session.execute("CREATE KEYSPACE k WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};")
node3 = new_node(cluster)
node3.start(wait_for_binary_proto=True)
def test_read_from_bootstrapped_node(self):
"""
Test bootstrapped node sees existing data
@jira_ticket CASSANDRA-6648
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(3)
if cluster.version() <= '5.0':
# Nodes need to be started one by one pre-TCM, not all at once. See CASSANDRA-19097
cluster.nodelist()[0].start(wait_for_binary_proto=True)
cluster.nodelist()[1].start(wait_for_binary_proto=True)
cluster.nodelist()[2].start(wait_for_binary_proto=True)
cluster.start()
node1 = cluster.nodes['node1']
node1.stress(['write', 'n=10K', 'no-warmup', '-rate', 'threads=8', '-schema', 'replication(factor=2)'])
session = self.patient_cql_connection(node1)
stress_table = 'keyspace1.standard1'
query = SimpleStatement("SELECT * FROM %s" % (stress_table))
original_rows = list(session.execute(query))
node4 = new_node(cluster)
node4.start(wait_for_binary_proto=True)
session = self.patient_exclusive_cql_connection(node4)
new_rows = list(session.execute(query))
assert original_rows == new_rows
@since('3.0')
def test_bootstrap_waits_for_streaming_to_finish(self):
"""
Test that bootstrap completes and is marked as such after streaming finishes.
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
logger.debug("Create a cluster")
cluster.populate(1)
node1 = cluster.nodelist()[0]
logger.debug("Start node 1")
node1.start(wait_for_binary_proto=True)
logger.debug("Insert 10k rows")
node1.stress(['write', 'n=10K', 'no-warmup', '-rate', 'threads=8', '-schema', 'replication(factor=2)'])
logger.debug("Bootstrap node 2 with delay")
node2 = new_node(cluster, byteman_port='4200')
node2.update_startup_byteman_script(mk_bman_path('bootstrap_5s_sleep.btm'))
node2.start(wait_for_binary_proto=True)
assert_bootstrap_state(self, node2, 'COMPLETED')
assert node2.grep_log('Bootstrap completed', filename='debug.log')
def test_consistent_range_movement_true_with_replica_down_should_fail(self):
self._bootstrap_test_with_replica_down(True)
def test_consistent_range_movement_false_with_replica_down_should_succeed(self):
self._bootstrap_test_with_replica_down(False)
def test_consistent_range_movement_true_with_rf1_should_fail(self):
self._bootstrap_test_with_replica_down(True, rf=1)
def test_consistent_range_movement_false_with_rf1_should_succeed(self):
self._bootstrap_test_with_replica_down(False, rf=1)
def test_rf_gt_nodes_multidc_should_succeed(self):
"""
Validating a KS with RF > N on multi DC doesn't break bootstrap
@jira_ticket CASSANDRA-16296 CASSANDRA-16411
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate([1, 1])
cluster.start()
node1 = cluster.nodelist()[0]
node2 = cluster.nodelist()[1]
session = self.patient_exclusive_cql_connection(node1)
session.execute("CREATE KEYSPACE k WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'dc1' : '3'}")
# Since CEP-21, this warning is logged twice on the coordinator (node1) as it is a CMS member so it applies the
# schema transformation once as a dry run for validation, then again when committing to the metadata log. With
# TCM, node2 also logs the warning as it applies the transform when it gets replicated to it by the CMS.
if cluster.version() >= '4.0':
warning = 'Your replication factor 3 for keyspace k is higher than the number of nodes 1 for datacenter dc1'
if cluster.version() >= LooseVersion('5.1'): # we now log this on all nodes
assert len(node1.grep_log(warning)) == 2
assert len(node2.grep_log(warning)) == 1
else:
assert len(node1.grep_log(warning)) == 1
assert len(node2.grep_log(warning)) == 0
session.execute("ALTER KEYSPACE k WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'dc1' : '2'}")
session.execute("CREATE TABLE k.testgtrfmultidc (KEY text PRIMARY KEY)")
session.execute("INSERT INTO k.testgtrfmultidc (KEY) VALUES ('test_rf_gt_nodes_multidc_should_succeed')")
if cluster.version() >= '4.0':
warning = 'Your replication factor 2 for keyspace k is higher than the number of nodes 1 for datacenter dc1'
if cluster.version() >= LooseVersion('5.1'):
assert len(node1.grep_log(warning)) == 2 # we now log this on all nodes
assert len(node2.grep_log(warning)) == 1
else:
assert len(node1.grep_log(warning)) == 1
assert len(node2.grep_log(warning)) == 0
marks = map(lambda n: n.mark_log(), cluster.nodelist())
node3 = Node(name='node3',
cluster=cluster,
auto_bootstrap=True,
thrift_interface=('127.0.0.3', 9160),
storage_interface=('127.0.0.3', 7000),
jmx_port='7300',
remote_debug_port='0',
initial_token=None,
binary_interface=('127.0.0.3', 9042))
cluster.add(node3, is_seed=False, data_center="dc1")
node3.start(wait_for_binary_proto=True)
if cluster.version() >= '4.0':
warning = 'is higher than the number of nodes'
for (node, mark) in zip(cluster.nodelist(), marks):
assert len(node.grep_log(warning, from_mark=mark)) == 0
session3 = self.patient_exclusive_cql_connection(node3)
assert_one(session3, "SELECT * FROM k.testgtrfmultidc", ["test_rf_gt_nodes_multidc_should_succeed"])
def _bootstrap_test_with_replica_down(self, consistent_range_movement, rf=2):
"""
Test to check consistent bootstrap will not succeed when there are insufficient replicas
@jira_ticket CASSANDRA-11848
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(2)
node1, node2 = cluster.nodelist()
node3_token = None
# Make token assignment deterministic
if not self.dtest_config.use_vnodes:
cluster.set_configuration_options(values={'num_tokens': 1})
tokens = cluster.balanced_tokens(3)
logger.debug("non-vnode tokens: %r" % (tokens,))
node1.set_configuration_options(values={'initial_token': tokens[0]})
node2.set_configuration_options(values={'initial_token': tokens[2]})
node3_token = tokens[1] # Add node 3 between node1 and node2
cluster.start()
node1.stress(['write', 'n=10K', 'no-warmup', '-rate', 'threads=8', '-schema', 'replication(factor={})'.format(rf)])
# change system_auth keyspace to 2 (default is 1) to avoid
# "Unable to find sufficient sources for streaming" warning
if cluster.cassandra_version() >= '2.2.0':
session = self.patient_cql_connection(node1)
session.execute("""
ALTER KEYSPACE system_auth
WITH replication = {'class':'SimpleStrategy', 'replication_factor':2};
""")
# Stop node2, so node3 will not be able to perform consistent range movement
node2.stop(wait_other_notice=True)
successful_bootstrap_expected = not consistent_range_movement
node3 = new_node(cluster, token=node3_token)
jvmargs = ["-Dcassandra.consistent.rangemovement={}".format(consistent_range_movement)]
if cluster.version() >= LooseVersion('5.1'):
node3.set_configuration_options(values={'progress_barrier_min_consistency_level': 'NODE_LOCAL', 'progress_barrier_default_consistency_level': 'NODE_LOCAL', 'progress_barrier_timeout': '2000ms'})
node3.start(wait_for_binary_proto=successful_bootstrap_expected,
wait_other_notice=successful_bootstrap_expected,
jvm_args=jvmargs)
if successful_bootstrap_expected:
# with rf=1 and cassandra.consistent.rangemovement=false, missing sources are ignored
if not consistent_range_movement and rf == 1:
node3.watch_log_for("Unable to find sufficient sources for streaming range")
assert node3.is_running()
assert_bootstrap_state(self, node3, 'COMPLETED')
else:
if consistent_range_movement:
if cluster.version() < '4.0':
node3.watch_log_for("A node required to move the data consistently is down")
else:
node3.watch_log_for("Necessary replicas for strict consistency were removed by source filters")
else:
node3.watch_log_for("Unable to find sufficient sources for streaming range")
assert_not_running(node3)
@since('2.2')
def test_resumable_bootstrap(self):
"""
Test resuming bootstrap after data streaming failure
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(2)
node1 = cluster.nodes['node1']
# set up byteman
node1.byteman_port = '8100'
node1.import_config_files()
cluster.start()
# kill stream to node3 in the middle of streaming to let it fail
if cluster.version() < '4.0':
node1.byteman_submit([self.byteman_submit_path_pre_4_0])
else:
node1.byteman_submit([self.byteman_submit_path_4_0])
node1.stress(['write', 'n=1K', 'no-warmup', 'cl=TWO', '-schema', 'replication(factor=2)', '-rate', 'threads=50'])
cluster.flush()
# start bootstrapping node3 and wait for streaming
node3 = new_node(cluster)
node3.start(jvm_args=['-Dcassandra.reset_bootstrap_progress=false'], wait_other_notice=False)
# let streaming fail as we expect
node3.watch_log_for('Some data streaming failed')
# bring back node3 and invoke nodetool bootstrap to resume bootstrapping
node3.nodetool(BootstrapTester.nodetool_resume_command(cluster))
node3.wait_for_binary_interface()
assert_bootstrap_state(self, node3, 'COMPLETED')
# cleanup to guarantee each node will only have sstables of its ranges
cluster.cleanup()
logger.debug("Check data is present")
# Let's check stream bootstrap completely transferred data
stdout, stderr, _ = node3.stress(['read', 'n=1k', 'no-warmup', '-schema', 'replication(factor=2)', '-rate', 'threads=8'])
if stdout is not None:
assert "FAILURE" not in stdout
@since('2.2')
def test_bootstrap_with_reset_bootstrap_state(self):
"""Test bootstrap with resetting bootstrap progress"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.set_configuration_options(values={'stream_throughput_outbound_megabits_per_sec': 1})
cluster.populate(2).start()
node1 = cluster.nodes['node1']
node1.stress(['write', 'n=100K', '-schema', 'replication(factor=2)'])
node1.flush()
# kill node1 in the middle of streaming to let it fail
t = InterruptBootstrap(node1)
t.start()
# start bootstrapping node3 and wait for streaming
node3 = new_node(cluster)
try:
node3.start()
except NodeError:
pass # node doesn't start as expected
t.join()
node1.start()
# restart node3 bootstrap with resetting bootstrap progress
node3.stop(signal_event=signal.SIGKILL)
mark = node3.mark_log()
node3.start(jvm_args=["-Dcassandra.reset_bootstrap_progress=true"])
# check if we reset bootstrap state
node3.watch_log_for("Resetting bootstrap progress to start fresh", from_mark=mark)
# wait for node3 ready to query, 180s as the node needs to bootstrap
node3.wait_for_binary_interface(from_mark=mark, timeout=180)
# check if 2nd bootstrap succeeded
assert_bootstrap_state(self, node3, 'COMPLETED')
def test_manual_bootstrap(self):
"""
Test adding a new node and bootstrapping it manually. No auto_bootstrap.
This test also verify that all data are OK after the addition of the new node.
@jira_ticket CASSANDRA-9022
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(2)
if cluster.version() <= '5.0':
# Nodes need to be started one by one pre-TCM, not all at once. See CASSANDRA-19097
cluster.nodelist()[0].start(wait_for_binary_proto=True)
cluster.nodelist()[1].start(wait_for_binary_proto=True)
cluster.start()
(node1, node2) = cluster.nodelist()
node1.stress(['write', 'n=1K', 'no-warmup', '-schema', 'replication(factor=2)',
'-rate', 'threads=1', '-pop', 'dist=UNIFORM(1..1000)'])
session = self.patient_exclusive_cql_connection(node2)
stress_table = 'keyspace1.standard1'
query = SimpleStatement("SELECT * FROM %s" % stress_table)
original_rows = list(session.execute(query))
# Add a new node
node3 = new_node(cluster, bootstrap=False)
node3.start(wait_for_binary_proto=True)
node3.repair()
node1.cleanup()
current_rows = list(session.execute(query))
assert original_rows == current_rows
def test_local_quorum_bootstrap(self):
"""
Test that CL local_quorum works while a node is bootstrapping.
@jira_ticket CASSANDRA-8058
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate([1, 1])
cluster.start()
node1 = cluster.nodes['node1']
yaml_config = """
# Create the keyspace and table
keyspace: keyspace1
keyspace_definition: |
CREATE KEYSPACE keyspace1 WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': 1, 'dc2': 1};
table: users
table_definition:
CREATE TABLE users (
username text,
first_name text,
last_name text,
email text,
PRIMARY KEY(username)
) WITH compaction = {'class':'SizeTieredCompactionStrategy'};
insert:
partitions: fixed(1)
batchtype: UNLOGGED
queries:
read:
cql: select * from users where username = ?
fields: samerow
"""
with tempfile.NamedTemporaryFile(mode='w+') as stress_config:
stress_config.write(yaml_config)
stress_config.flush()
node1.stress(['user', 'profile=' + stress_config.name, 'n=200K', 'no-warmup',
'ops(insert=1)', '-rate', 'threads=10'])
node3 = new_node(cluster, data_center='dc2')
node3.start(jvm_args=["-Dcassandra.write_survey=true"], no_wait=True)
node3_seen = False
for _ in range(30): # give node3 up to 30 seconds to start
ntout = node1.nodetool('status').stdout
if re.search(r'UJ\s+' + node3.ip_addr, ntout):
node3_seen = True
break
time.sleep(1)
assert node3_seen, "expected {} in status:\n{}".format(node3.ip_addr, ntout)
out, err, _ = node1.stress(['user', 'profile=' + stress_config.name, 'ops(insert=1)',
'n=10k', 'no-warmup', 'cl=LOCAL_QUORUM',
'-rate', 'threads=10',
'-errors', 'retries=2'])
ntout = node1.nodetool('status').stdout
assert re.search(r'UJ\s+' + node3.ip_addr, ntout), ntout
logger.debug(out)
assert_stderr_clean(err)
regex = re.compile("Operation.+error inserting key.+Exception")
failure = regex.search(str(out))
assert failure is None, "Error during stress while bootstrapping"
def test_shutdown_wiped_node_cannot_join(self):
self._wiped_node_cannot_join_test(gently=True)
def test_killed_wiped_node_cannot_join(self):
self._wiped_node_cannot_join_test(gently=False)
def _wiped_node_cannot_join_test(self, gently):
"""
@jira_ticket CASSANDRA-9765
Test that if we stop a node and wipe its data then the node cannot join
when it is not a seed. Test both a nice shutdown or a forced shutdown, via
the gently parameter.
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(3)
if cluster.version() <= '5.0':
# Nodes need to be started one by one pre-TCM, not all at once. See CASSANDRA-19097
cluster.nodelist()[0].start(wait_for_binary_proto=True)
cluster.nodelist()[1].start(wait_for_binary_proto=True)
cluster.nodelist()[2].start(wait_for_binary_proto=True)
cluster.start()
stress_table = 'keyspace1.standard1'
# write some data
node1 = cluster.nodelist()[0]
node1.stress(['write', 'n=10K', 'no-warmup', '-rate', 'threads=8'])
session = self.patient_cql_connection(node1)
query = SimpleStatement("SELECT * FROM {}".format(stress_table))
original_rows = list(session.execute(query))
# Add a new node, bootstrap=True ensures that it is not a seed
node4 = new_node(cluster, bootstrap=True)
node4.start(wait_for_binary_proto=True)
session = self.patient_cql_connection(node4)
assert original_rows == list(session.execute(query))
# Stop the new node and wipe its data
node4.stop(gently=gently)
self._cleanup(node4)
# Now start it, it should not be allowed to join.
mark = node4.mark_log()
node4.start(no_wait=True, wait_other_notice=False)
node4.watch_log_for("A node with address {} already exists, cancelling join".format(node4.address_for_current_version_slashy()), from_mark=mark)
def test_decommissioned_wiped_node_can_join(self):
"""
@jira_ticket CASSANDRA-9765
Test that if we decommission a node and then wipe its data, it can join the cluster.
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(3)
if cluster.version() <= '5.0':
# Nodes need to be started one by one pre-TCM, not all at once. See CASSANDRA-19097
cluster.nodelist()[0].start(wait_for_binary_proto=True)
cluster.nodelist()[1].start(wait_for_binary_proto=True)
cluster.nodelist()[2].start(wait_for_binary_proto=True)
cluster.start()
stress_table = 'keyspace1.standard1'
# write some data
node1 = cluster.nodelist()[0]
node1.stress(['write', 'n=10K', 'no-warmup', '-rate', 'threads=8'])
session = self.patient_cql_connection(node1)
query = SimpleStatement("SELECT * FROM {}".format(stress_table))
original_rows = list(session.execute(query))
# Add a new node, bootstrap=True ensures that it is not a seed
node4 = new_node(cluster, bootstrap=True)
node4.start(wait_for_binary_proto=True)
session = self.patient_cql_connection(node4)
assert original_rows == list(session.execute(query))
# Decommission the new node and wipe its data
node4.decommission()
node4.stop()
self._cleanup(node4)
# Now start it, it should be allowed to join
mark = node4.mark_log()
node4.start()
node4.watch_log_for("JOINING:", from_mark=mark)
def test_decommissioned_wiped_node_can_gossip_to_single_seed(self):
"""
@jira_ticket CASSANDRA-8072
@jira_ticket CASSANDRA-8422
Test that if we decommission a node, kill it and wipe its data, it can join a cluster with a single
seed node.
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(1)
cluster.start()
node1 = cluster.nodelist()[0]
# Add a new node, bootstrap=True ensures that it is not a seed
node2 = new_node(cluster, bootstrap=True)
node2.start(wait_for_binary_proto=True)
session = self.patient_cql_connection(node1)
if cluster.version() >= '2.2':
# reduce system_distributed RF to 2 so we don't require forceful decommission
session.execute("ALTER KEYSPACE system_distributed WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':'1'};")
session.execute("ALTER KEYSPACE system_traces WITH REPLICATION = {'class':'SimpleStrategy', 'replication_factor':'1'};")
# Decommision the new node and kill it
logger.debug("Decommissioning & stopping node2")
node2.decommission()
node2.stop(wait_other_notice=False)
# Wipe its data
self._cleanup(node2)
# Now start it, it should be allowed to join
mark = node2.mark_log()
logger.debug("Restarting wiped node2")
node2.start(wait_other_notice=False)
node2.watch_log_for("JOINING:", from_mark=mark)
def test_failed_bootstrap_wiped_node_can_join(self):
"""
@jira_ticket CASSANDRA-9765
Test that if a node fails to bootstrap, it can join the cluster even if the data is wiped.
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(1)
cluster.set_configuration_options(values={'stream_throughput_outbound_megabits_per_sec': 1})
cluster.start()
stress_table = 'keyspace1.standard1'
# write some data, enough for the bootstrap to fail later on
node1 = cluster.nodelist()[0]
node1.stress(['write', 'n=100K', 'no-warmup', '-rate', 'threads=8'])
node1.flush()
session = self.patient_cql_connection(node1)
original_rows = list(session.execute("SELECT * FROM {}".format(stress_table)))
# Add a new node, bootstrap=True ensures that it is not a seed
node2 = new_node(cluster, bootstrap=True)
mark = node1.mark_log()
# kill node2 in the middle of bootstrap
t = KillOnBootstrap(node2)
t.start()
node2.start()
t.join()
assert not node2.is_running()
# wipe any data for node2
self._cleanup(node2)
if cluster.version() >= LooseVersion('5.1'):
node1.watch_log_for("127.0.0.2:7000 is now DOWN", from_mark=mark)
res = node1.nodetool('abortbootstrap --ip 127.0.0.2')
# Now start it again, it should be allowed to join
mark = node2.mark_log()
node2.start()
node2.watch_log_for("JOINING:", from_mark=mark)
@since('3.0')
@ported_to_in_jvm('4.1')
def test_node_cannot_join_as_hibernating_node_without_replace_address(self):
"""
@jira_ticket CASSANDRA-14559
Test that a node cannot bootstrap without replace_address if a hibernating node exists with that address
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(2)
# Setting seed node to first node to make sure replaced node is not in own seed list
cluster.set_configuration_options({
'seed_provider': [{'class_name': 'org.apache.cassandra.locator.SimpleSeedProvider',
'parameters': [{'seeds': '127.0.0.1'}]
}]
})
cluster.start()
node1 = cluster.nodelist()[0]
node2 = cluster.nodelist()[1]
replacement_address = node2.address()
node2.stop()
jvm_option = 'replace_address'
logger.debug("Starting replacement node {} with jvm_option '{}={}'".format(replacement_address, jvm_option,
replacement_address))
replacement_node = Node('replacement', cluster=self.cluster, auto_bootstrap=True,
thrift_interface=None, storage_interface=(replacement_address, 7000),
jmx_port='7400', remote_debug_port='0', initial_token=None,
binary_interface=(replacement_address, 9042))
cluster.add(replacement_node, False)
extra_jvm_args = []
extra_jvm_args.extend(["-Dcassandra.{}={}".format(jvm_option, replacement_address),
"-Dcassandra.ring_delay_ms=10000",
"-Dcassandra.broadcast_interval_ms=10000"])
wait_other_notice = False
wait_for_binary_proto = False
# Killing node earlier in bootstrap to prevent node making it to 'normal' status.
t = KillOnReadyToBootstrap(replacement_node)
t.start()
replacement_node.start(jvm_args=extra_jvm_args,
wait_for_binary_proto=wait_for_binary_proto, wait_other_notice=wait_other_notice)
t.join()
logger.debug("Asserting that original replacement node is not running")
assert not replacement_node.is_running()
# Assert node is actually in hibernate for test to be accurate.
logger.debug("Asserting that node is actually in hibernate status for test accuracy")
assert 'hibernate' in node1.nodetool("gossipinfo").stdout
extra_jvm_args = []
extra_jvm_args.extend(["-Dcassandra.ring_delay_ms=10000",
"-Dcassandra.broadcast_interval_ms=10000"])
logger.debug("Starting blind replacement node {}".format(replacement_address))
blind_replacement_node = Node('blind_replacement', cluster=self.cluster, auto_bootstrap=True,
thrift_interface=None, storage_interface=(replacement_address, 7000),
jmx_port='7400', remote_debug_port='0', initial_token=None,
binary_interface=(replacement_address, 9042))
cluster.add(blind_replacement_node, False)
wait_other_notice = False
wait_for_binary_proto = False
blind_replacement_node.start(wait_for_binary_proto=wait_for_binary_proto, wait_other_notice=wait_other_notice)
# Asserting that the new node has correct log entry
self.assert_log_had_msg(blind_replacement_node, "A node with the same IP in hibernate status was detected", timeout=60)
# Waiting two seconds to give node a chance to stop in case above assertion is True.
# When this happens cassandra may not shut down fast enough and the below assertion fails.
time.sleep(15)
# Asserting that then new node is not running.
# This tests the actual expected state as opposed to just checking for the existance of the above error message.
assert not blind_replacement_node.is_running()
# this test relies on startup of .2 being slow, allowing .3 to start before .2 has finished joining
# with tcm this is much quicker, and this is now covered by better tests
@since('2.1.1', max_version='5.0.x')
def test_simultaneous_bootstrap(self):
"""
Attempt to bootstrap two nodes at once, to assert the second bootstrapped node fails, and does not interfere.
Start a one node cluster and run a stress write workload.
Start up a second node, and wait for the first node to detect it has joined the cluster.
While the second node is bootstrapping, start a third node. This should fail.
@jira_ticket CASSANDRA-7069
@jira_ticket CASSANDRA-9484
"""
bootstrap_error = "Other bootstrapping/leaving/moving nodes detected," \
" cannot bootstrap while cassandra.consistent.rangemovement is true"
cluster = self.cluster
configuration_options = {
'request_timeout_in_ms': 120000,
'read_request_timeout_in_ms': 120000,
'range_request_timeout_in_ms': 120000
}
cluster.set_configuration_options(configuration_options)
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.populate(1)
cluster.start()
node1, = cluster.nodelist()
node1.stress(['write', 'n=500K', 'no-warmup', '-schema', 'replication(factor=1)',
'-rate', 'threads=10'])
node2 = new_node(cluster)
node2.start()
for _ in range(30): # wait until node2 shows up
ntout = node1.nodetool('status').stdout
if re.search(r'UJ\s+' + node2.ip_addr, ntout):
break
time.sleep(0.1)
node3 = new_node(cluster, remote_debug_port='2003')
try:
node3.start(wait_other_notice=False, verbose=False)
except NodeError:
pass # node doesn't start as expected
time.sleep(.5)
node2.watch_log_for("Starting listening for CQL clients")
node3.watch_log_for(bootstrap_error)
session = self.patient_exclusive_cql_connection(node2)
# Repeat the select count(*) query, to help catch
# bugs like 9484, where count(*) fails at higher
# data loads.
for _ in range(5):
logger.info("Querying: SELECT count(*) from keyspace1.standard1")
# Improve reliability for slower/loaded test systems by using larger client timeout
assert_one(session, "SELECT count(*) from keyspace1.standard1", [500000], cl=ConsistencyLevel.ONE, timeout=180)
logger.info("Querying completed")
def test_cleanup(self):
"""
@jira_ticket CASSANDRA-11179
Make sure we remove processed files during cleanup
"""
cluster = self.cluster
cluster.set_environment_variable('CASSANDRA_TOKEN_PREGENERATION_DISABLED', 'True')
cluster.set_configuration_options(values={'concurrent_compactors': 4})
cluster.populate(1)
cluster.start()
node1, = cluster.nodelist()
for x in range(0, 5):
node1.stress(['write', 'n=100k', 'no-warmup', '-schema', 'compaction(strategy=SizeTieredCompactionStrategy,enabled=false)', 'replication(factor=1)', '-rate', 'threads=10'])
node1.flush()
node2 = new_node(cluster)
node2.start(wait_for_binary_proto=True)
event = threading.Event()
failed = threading.Event()
jobs = 1
thread = threading.Thread(target=self._monitor_datadir, args=(node1, event, len(node1.get_sstables("keyspace1", "standard1")), jobs, failed))
thread.setDaemon(True)
thread.start()
node1.nodetool("cleanup -j {} keyspace1 standard1".format(jobs))
event.set()
thread.join()
assert not failed.is_set()
def _monitor_datadir(self, node, event, basecount, jobs, failed):
while True:
sstables = [s for s in node.get_sstables("keyspace1", "standard1") if "tmplink" not in s]
if len(sstables) > basecount + jobs:
logger.error("---")
for sstable in sstables:
logger.error(sstable)
logger.error("Current count is {}, basecount was {}".format(len(sstables), basecount))
failed.set()
return
if event.is_set():
return
time.sleep(.1)
def _cleanup(self, node):
commitlog_dir = os.path.join(node.get_path(), 'commitlogs')
for data_dir in node.data_directories():
logger.debug("Deleting {}".format(data_dir))
shutil.rmtree(data_dir)
shutil.rmtree(commitlog_dir)
metadata_dir = os.path.join(node.get_path(), 'metadata')
if os.path.exists(metadata_dir):
logger.debug("Deleting {}".format(metadata_dir))
shutil.rmtree(metadata_dir)
@since('2.2')
@pytest.mark.ported_to_in_jvm # see org.apache.cassandra.distributed.test.BootstrapBinaryDisabledTest
def test_bootstrap_binary_disabled(self):
"""
Test binary while bootstrapping and streaming fails.
This test was ported to jvm-dtest org.apache.cassandra.distributed.test.BootstrapBinaryDisabledTest,
as of this writing there are a few limitations with jvm-dtest which requries this test to
stay, namely vnode support (ci also tests under different configs). Once jvm-dtest supports
vnodes, this test can go away in favor of that class.