forked from apache/cassandra-dtest
-
Notifications
You must be signed in to change notification settings - Fork 0
/
cql_tracing_test.py
216 lines (176 loc) · 8.27 KB
/
cql_tracing_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
import pytest
import logging
from distutils.version import LooseVersion
from dtest import Tester, create_ks
from tools.jmxutils import make_mbean, JolokiaAgent, remove_perf_disable_shared_mem
since = pytest.mark.since
logger = logging.getLogger(__name__)
class TestCqlTracing(Tester):
"""
Smoke test that the default implementation for tracing works. Also test
that Cassandra falls back to the default tracing implementation when the
user specifies an invalid implementation.
# TODO write a mock Tracing implementation and assert, at least, it can be
# instantiated when specified as a custom tracing implementation.
"""
def prepare(self, create_keyspace=True, nodes=3, rf=3, protocol_version=3, jvm_args=None, random_partitioner=False, **kwargs):
if jvm_args is None:
jvm_args = []
jvm_args.append('-Dcassandra.wait_for_tracing_events_timeout_secs=15')
cluster = self.cluster
if random_partitioner:
cluster.set_partitioner("org.apache.cassandra.dht.RandomPartitioner")
else:
cluster.set_partitioner("org.apache.cassandra.dht.Murmur3Partitioner")
cluster.populate(nodes)
node1 = cluster.nodelist()[0]
remove_perf_disable_shared_mem(node1) # necessary for jmx
cluster.start(wait_for_binary_proto=True, jvm_args=jvm_args)
session = self.patient_cql_connection(node1, protocol_version=protocol_version)
if create_keyspace:
create_ks(session, 'ks', rf)
return session
def trace(self, session):
"""
* CREATE a table
* enable TRACING
* SELECT on a known system table and assert it ran with tracing by checking the output
* INSERT a row into the created system table and assert it ran with tracing
* SELECT from the table and assert it ran with tracing
@param session The Session object to use to create a table.
@jira_ticket CASSANDRA-10392
"""
node1 = self.cluster.nodelist()[0]
# Create
session.execute("""
CREATE TABLE ks.users (
userid uuid PRIMARY KEY,
firstname text,
lastname text,
age int
);
""")
out, err, _ = node1.run_cqlsh('TRACING ON')
assert 'Tracing is enabled' in out
out, err, _ = node1.run_cqlsh('TRACING ON; SELECT * from system.peers')
assert 'Tracing session: ' in out
assert 'Request complete ' in out
# Inserts
out, err, _ = node1.run_cqlsh(
"CONSISTENCY ALL; TRACING ON; "
"INSERT INTO ks.users (userid, firstname, lastname, age) "
"VALUES (550e8400-e29b-41d4-a716-446655440000, 'Frodo', 'Baggins', 32)")
logger.debug(out)
assert 'Tracing session: ' in out
assert node1.address_for_current_version_slashy() in out
assert self.cluster.nodelist()[1].address_for_current_version_slashy() in out
assert self.cluster.nodelist()[2].address_for_current_version_slashy() in out
assert 'Parsing INSERT INTO ks.users ' in out
assert 'Request complete ' in out
# Queries
out, err, _ = node1.run_cqlsh('CONSISTENCY ALL; TRACING ON; '
'SELECT firstname, lastname '
'FROM ks.users WHERE userid = 550e8400-e29b-41d4-a716-446655440000')
logger.debug(out)
assert 'Tracing session: ' in out
assert ' 127.0.0.1 ' in out
assert ' 127.0.0.2 ' in out
assert ' 127.0.0.3 ' in out
assert 'Request complete ' in out
assert " Frodo | Baggins" in out
@since('2.2')
def test_tracing_simple(self):
"""
Test tracing using the default tracing class. See trace().
@jira_ticket CASSANDRA-10392
@jira_ticket CASSANDRA-11598
# Restricted to 2.2+ due to flakiness on 2.1. See CASSANDRA-11598 and CASSANDRA-12407 for details.
"""
session = self.prepare()
self.trace(session)
@since('3.4')
def test_tracing_unknown_impl(self):
"""
Test that Cassandra logs an error, but keeps its default tracing
behavior, when a nonexistent tracing class is specified.
* set a nonexistent custom tracing class
* run trace()
* if running the test on a version with custom tracing classes
implemented, check that an error about the nonexistent class was
logged.
@jira_ticket CASSANDRA-10392
"""
expected_error = 'Cannot use class junk for tracing'
self.fixture_dtest_setup.ignore_log_patterns = [expected_error]
session = self.prepare(jvm_args=['-Dcassandra.custom_tracing_class=junk'])
self.trace(session)
errs = self.cluster.nodelist()[0].grep_log_for_errors()
logger.debug('Errors after attempted trace with unknown tracing class: {errs}'.format(errs=errs))
assert len(errs) == 1
if self.cluster.version() >= LooseVersion('3.10'):
# See CASSANDRA-11706 and PR #1281
assert len(errs[0]) > 0
else:
assert len(errs[0]) == 1
err = errs[0][0]
assert expected_error in err
@since('3.4')
def test_tracing_default_impl(self):
"""
Test that Cassandra logs an error, but keeps its default tracing
behavior, when the default tracing class is specified.
This doesn't work because the constructor for the default
implementation isn't accessible.
* set the default tracing class as a custom tracing class
* run trace()
* if running the test on a version with custom tracing classes
implemented, check that an error about the class was
logged.
@jira_ticket CASSANDRA-10392
"""
expected_error = 'Cannot use class org.apache.cassandra.tracing.TracingImpl'
self.fixture_dtest_setup.ignore_log_patterns = [expected_error]
session = self.prepare(jvm_args=['-Dcassandra.custom_tracing_class=org.apache.cassandra.tracing.TracingImpl'])
self.trace(session)
errs = self.cluster.nodelist()[0].grep_log_for_errors()
logger.debug('Errors after attempted trace with default tracing class: {errs}'.format(errs=errs))
assert len(errs) == 1
if self.cluster.version() >= LooseVersion('3.10'):
# See CASSANDRA-11706 and PR #1281
assert len(errs[0]) > 0
else:
assert len(errs[0]) == 1
err = errs[0][0]
assert expected_error in err
# make sure it logged the error for the correct reason. this isn't
# part of the expected error to avoid having to escape parens and
# periods for regexes.
if self.cluster.version() >= LooseVersion('3.10'):
# See CASSANDRA-11706 and PR #1281
check_for_errs_in = errs[0][1]
else:
check_for_errs_in = err
assert "Default constructor for Tracing class 'org.apache.cassandra.tracing.TracingImpl' is inaccessible." \
in check_for_errs_in
@since('3.0')
def test_tracing_does_not_interfere_with_digest_calculation(self):
"""
Test that enabling tracing doesn't interfere with digest responses when using RandomPartitioner.
The use of a threadlocal MessageDigest for generating both DigestResponse messages and for
calculating tokens meant that the DigestResponse was always incorrect when both RP and tracing
were enabled, leading to unnecessary data reads.
@jira_ticket CASSANDRA-13964
"""
session = self.prepare(random_partitioner=True)
self.trace(session)
node1 = self.cluster.nodelist()[0]
rr_count = make_mbean('metrics', type='ReadRepair', name='RepairedBlocking')
with JolokiaAgent(node1) as jmx:
# the MBean may not have been initialized, in which case Jolokia agent will return
# a HTTP 404 response. If we receive such, we know that no digest mismatch was reported
# If we are able to read the MBean attribute, assert that the count is 0
if jmx.has_mbean(rr_count):
# expect 0 digest mismatches
assert 0 == jmx.read_attribute(rr_count, 'Count')
else:
pass