forked from thepaul/cassandra-dtest
-
Notifications
You must be signed in to change notification settings - Fork 2
/
replace_address_test.py
127 lines (105 loc) · 5.41 KB
/
replace_address_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
from dtest import Tester, debug, DISABLE_VNODES
import unittest
from tools import *
from ccmlib.cluster import Cluster
from ccmlib.node import NodeError
import time
from cql import OperationalError
from cql.cassandra.ttypes import UnavailableException
class NodeUnavailable(Exception):
pass
class TestReplaceAddress(Tester):
def __init__(self, *args, **kwargs):
# Ignore these log patterns:
self.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',
# This is caused by starting a node improperly (replacing active/nonexistent)
r'Exception encountered during startup',
# This is caused by trying to replace a nonexistent node
r'Exception in thread Thread'
]
Tester.__init__(self, *args, **kwargs)
def replace_stopped_node_test(self):
"""Check that the replace address function correctly replaces a node that has failed in a cluster.
Create a cluster, cause a node to fail, and bring up a new node with the replace_address parameter.
Check that tokens are migrated and that data is replicated properly.
"""
debug("Starting cluster with 3 nodes.")
cluster = self.cluster
cluster.populate(3).start()
[node1,node2, node3] = cluster.nodelist()
debug("Inserting Data...")
node1.stress(['write', 'n=10000', '-schema', 'replication(factor=3)'])
cursor = self.patient_cql_connection(node1).cursor()
cursor.execute('select * from "Keyspace1"."Standard1" LIMIT 1', consistency_level='THREE')
initialData = cursor.fetchall()
#stop node, query should not work with consistency 3
debug("Stopping node 3.")
node3.stop(gently=False)
time.sleep(5)
debug("Testing node stoppage (query should fail).")
with self.assertRaises(NodeUnavailable):
try:
cursor.execute('select * from "Keyspace1"."Standard1" LIMIT 1', consistency_level='THREE')
except (UnavailableException, OperationalError):
raise NodeUnavailable("Node could not be queried.")
#replace node 3 with node 4
debug("Starting node 4 to replace node 3")
node4 = Node('node4', cluster, True, ('127.0.0.4', 9160), ('127.0.0.4', 7000), '7400', '0', None, ('127.0.0.4',9042))
cluster.add(node4, False)
node4.start(replace_address='127.0.0.3')
#query should work again
debug("Verifying querying works again.")
cursor.execute('select * from "Keyspace1"."Standard1" LIMIT 1', consistency_level='THREE')
finalData = cursor.fetchall()
self.assertListEqual(initialData, finalData)
debug("Verifying tokens migrated sucessfully")
movedTokensList = node4.grep_log("Token .* changing ownership from /127.0.0.3 to /127.0.0.4")
debug(movedTokensList[0])
if DISABLE_VNODES:
self.assertEqual(len(movedTokensList), 1)
else:
self.assertEqual(len(movedTokensList), 256)
#check that restarting node 3 doesn't work
debug("Try to restart node 3 (should fail)")
node3.start()
checkCollision = node1.grep_log("between /127.0.0.3 and /127.0.0.4; /127.0.0.4 is the new owner")
debug(checkCollision)
self.assertEqual(len(checkCollision), 1)
def replace_active_node_test(self):
debug("Starting cluster with 3 nodes.")
cluster = self.cluster
cluster.populate(3).start()
[node1,node2, node3] = cluster.nodelist()
debug("Inserting Data...")
node1.stress(['write', 'n=10000', '-schema', 'replication(factor=3)'])
cursor = self.patient_cql_connection(node1).cursor()
cursor.execute('select * from "Keyspace1"."Standard1" LIMIT 1', consistency_level='THREE')
initialData = cursor.fetchall()
#replace active node 3 with node 4
debug("Starting node 4 to replace active node 3")
node4 = Node('node4', cluster, True, ('127.0.0.4', 9160), ('127.0.0.4', 7000), '7400', '0', None, ('127.0.0.4',9042))
cluster.add(node4, False)
with self.assertRaises(NodeError):
node4.start(replace_address='127.0.0.3')
checkError = node4.grep_log("java.lang.UnsupportedOperationException: Cannnot replace a live node...")
self.assertEqual(len(checkError), 1)
def replace_nonexistent_node_test(self):
debug("Starting cluster with 3 nodes.")
cluster = self.cluster
cluster.populate(3).start()
[node1,node2, node3] = cluster.nodelist()
debug("Inserting Data...")
node1.stress(['write', 'n=10000', '-schema', 'replication(factor=3)'])
cursor = self.patient_cql_connection(node1).cursor()
cursor.execute('select * from "Keyspace1"."Standard1" LIMIT 1', consistency_level='THREE')
initialData = cursor.fetchall()
debug('Start node 4 and replace an address with no node')
node4 = Node('node4', cluster, True, ('127.0.0.4', 9160), ('127.0.0.4', 7000), '7400', '0', None, ('127.0.0.4',9042))
cluster.add(node4, False)
#try to replace an unassigned ip address
with self.assertRaises(NodeError):
node4.start(replace_address='127.0.0.5')