| import os |
| import os.path |
| import threading |
| import time |
| import re |
| import pytest |
| import logging |
| |
| from collections import namedtuple |
| from threading import Thread |
| |
| from cassandra import ConsistencyLevel |
| from cassandra.query import SimpleStatement |
| from ccmlib.node import ToolError |
| |
| from dtest import FlakyRetryPolicy, Tester, create_ks, create_cf |
| from tools.data import insert_c1c2, query_c1c2 |
| |
| since = pytest.mark.since |
| logger = logging.getLogger(__name__) |
| |
| |
| def _repair_options(version, ks='', cf=None, sequential=True): |
| """ |
| Function for assembling appropriate repair CLI options, |
| based on C* version, as defaults have changed. |
| @param ks The keyspace to repair |
| @param cf The table to repair |
| @param sequential If the repair should be a sequential repair [vs parallel] |
| """ |
| opts = [] |
| |
| # since version 2.2, default is parallel, otherwise it's sequential |
| if sequential: |
| if version >= '2.2': |
| opts += ['-seq'] |
| else: |
| if version < '2.2': |
| opts += ['-par'] |
| |
| # test with full repair |
| if version >= '2.2': |
| opts += ['-full'] |
| if ks: |
| opts += [ks] |
| if cf: |
| opts += [cf] |
| return opts |
| |
| |
| class BaseRepairTest(Tester): |
| |
| def check_rows_on_node(self, node_to_check, rows, found=None, missings=None, restart=True): |
| """ |
| Function to verify the rows on a given node, without interference |
| from the other nodes in the cluster |
| @param node_to_check The given node to check. Should be the node, not the index |
| @param rows The number of rows we expect |
| @param found A list of partition keys that we expect to be on the node |
| @param missings A list of partition keys we expect NOT to be on the node |
| @param restart Whether or not we should restart the nodes we shut down to perform the assertions. Should only be False if the call to check_rows_on_node is the last line in the test. |
| """ |
| if found is None: |
| found = [] |
| if missings is None: |
| missings = [] |
| stopped_nodes = [] |
| |
| for node in list(self.cluster.nodes.values()): |
| if node.is_running() and node is not node_to_check: |
| stopped_nodes.append(node) |
| node.stop(wait_other_notice=True) |
| |
| session = self.patient_exclusive_cql_connection(node_to_check, 'ks') |
| result = list(session.execute("SELECT * FROM cf LIMIT {}".format(rows * 2), timeout=10)) |
| assert len(result) == rows |
| |
| for k in found: |
| query_c1c2(session, k, ConsistencyLevel.ONE) |
| |
| for k in missings: |
| query = SimpleStatement("SELECT c1, c2 FROM cf WHERE key='k{}'".format(k), consistency_level=ConsistencyLevel.ONE) |
| res = list(session.execute(query)) |
| assert len([x for x in res if len(x) != 0]) == 0, res |
| |
| if restart: |
| for node in stopped_nodes: |
| node.start(wait_other_notice=True) |
| |
| def _populate_cluster(self, start=True): |
| cluster = self.cluster |
| |
| # Disable hinted handoff and set batch commit log so this doesn't |
| # interfere with the test (this must be after the populate) |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start() |
| node1, node2, node3 = cluster.nodelist() |
| |
| session = self.patient_cql_connection(node1, retry_policy=FlakyRetryPolicy(max_retries=15)) |
| create_ks(session, 'ks', 3) |
| if cluster.version() < '4.0': |
| create_cf(session, 'cf', read_repair=0.0, columns={'c1': 'text', 'c2': 'text'}) |
| else: |
| create_cf(session, 'cf', columns={'c1': 'text', 'c2': 'text'}) |
| |
| # Insert 1000 keys, kill node 3, insert 1 key, restart node 3, insert 1000 more keys |
| logger.debug("Inserting data...") |
| insert_c1c2(session, n=1000, consistency=ConsistencyLevel.ALL) |
| node3.flush() |
| node3.stop(wait_other_notice=True) |
| insert_c1c2(session, keys=(1000, ), consistency=ConsistencyLevel.TWO) |
| node3.start(wait_other_notice=True, wait_for_binary_proto=True) |
| insert_c1c2(session, keys=list(range(1001, 2001)), consistency=ConsistencyLevel.ALL) |
| |
| cluster.flush() |
| |
| def _repair_and_verify(self, sequential=True): |
| cluster = self.cluster |
| node1, node2, node3 = cluster.nodelist() |
| |
| # Verify that node3 has only 2000 keys |
| logger.debug("Checking data on node3...") |
| self.check_rows_on_node(node3, 2000, missings=[1000]) |
| |
| # Verify that node1 has 2001 keys |
| logger.debug("Checking data on node1...") |
| self.check_rows_on_node(node1, 2001, found=[1000]) |
| |
| # Verify that node2 has 2001 keys |
| logger.debug("Checking data on node2...") |
| self.check_rows_on_node(node2, 2001, found=[1000]) |
| |
| time.sleep(10) # see CASSANDRA-4373 |
| # Run repair |
| start = time.time() |
| logger.debug("starting repair...") |
| node1.repair(_repair_options(self.cluster.version(), ks='ks', sequential=sequential)) |
| logger.debug("Repair time: {end}".format(end=time.time() - start)) |
| |
| # Validate that only one range was transfered |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| |
| assert len(out_of_sync_logs) == 2, "Lines matching: " + str([elt[0] for elt in out_of_sync_logs]) |
| |
| valid_out_of_sync_pairs = [{node1.address(), node3.address()}, |
| {node2.address(), node3.address()}] |
| |
| for line, m in out_of_sync_logs: |
| num_out_of_sync_ranges, out_of_sync_nodes = m.group(3), {m.group(1), m.group(2)} |
| assert int(num_out_of_sync_ranges) == 1, \ |
| "Expecting 1 range out of sync for {}, but saw {}".format(out_of_sync_nodes, num_out_of_sync_ranges) |
| assert out_of_sync_nodes, valid_out_of_sync_pairs in str(out_of_sync_nodes) |
| |
| # Check node3 now has the key |
| self.check_rows_on_node(node3, 2001, found=[1000], restart=False) |
| |
| |
| class TestRepair(BaseRepairTest): |
| |
| @since('2.2.1', max_version='4') |
| def test_no_anticompaction_after_dclocal_repair(self): |
| """ |
| * Launch a four node, two DC cluster |
| * Start a -local repair on node1 in dc1 |
| * Assert that the dc1 nodes see repair messages |
| * Assert that the dc2 nodes do not see repair messages |
| * Assert no nodes anticompact |
| # TODO: Verify the anticompaction with sstablemetadata, not just logs |
| @jira_ticket CASSANDRA-10422 |
| """ |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| cluster.populate([2, 2]).start(wait_for_binary_proto=True) |
| node1_1, node2_1, node1_2, node2_2 = cluster.nodelist() |
| node1_1.stress(stress_options=['write', 'n=50K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=4)', '-rate', 'threads=50']) |
| node1_1.nodetool("repair -local keyspace1 standard1") |
| assert node1_1.grep_log("Not a global repair") |
| assert node2_1.grep_log("Not a global repair") |
| |
| # dc2 should not see these messages: |
| assert not node1_2.grep_log("Not a global repair") |
| assert not node2_2.grep_log("Not a global repair") |
| |
| # and no nodes should do anticompaction: |
| for node in cluster.nodelist(): |
| assert not node.grep_log("Starting anticompaction") |
| |
| # @pytest.mark.skipif(CASSANDRA_VERSION_FROM_BUILD == '3.9', reason="Test doesn't run on 3.9") |
| @pytest.mark.skip_version('3.9') |
| def test_nonexistent_table_repair(self): |
| """ |
| * Check that repairing a non-existent table fails |
| @jira_ticket CASSANDRA-12279 |
| """ |
| self.fixture_dtest_setup.ignore_log_patterns = [r'Unknown keyspace/cf pair'] |
| cluster = self.cluster |
| logger.debug('Starting nodes') |
| cluster.populate(2).start(wait_for_binary_proto=True) |
| node1, _ = cluster.nodelist() |
| logger.debug('Creating keyspace and tables') |
| node1.stress(stress_options=['write', 'n=1', 'no-warmup', |
| 'cl=ONE', '-schema', 'replication(factor=2)', |
| '-rate', 'threads=1']) |
| logger.debug('Repairing non-existent table') |
| |
| def repair_non_existent_table(): |
| global nodetool_error |
| try: |
| node1.nodetool('repair keyspace1 standard2') |
| except Exception as e: |
| nodetool_error = e |
| |
| # Launch in a external thread so it does not hang process |
| t = Thread(target=repair_non_existent_table) |
| t.start() |
| |
| t.join(timeout=60) |
| assert not t.is_alive(), 'Repair thread on inexistent table is still running' |
| |
| if self.cluster.version() >= '2.2': |
| node1.watch_log_for("Unknown keyspace/cf pair", timeout=60) |
| # Repair only finishes with error status after CASSANDRA-12508 on 3.0+ |
| if self.cluster.version() >= '3.0': |
| assert 'nodetool_error' in globals() and isinstance(nodetool_error, ToolError), \ |
| 'Repair thread on inexistent table did not throw exception' |
| logger.debug(repr(nodetool_error)) |
| assert 'Unknown keyspace/cf pair' in repr(nodetool_error),\ |
| 'Repair thread on inexistent table did not detect inexistent table.' |
| |
| @since('2.2.1', max_version='4') |
| def test_no_anticompaction_after_hostspecific_repair(self): |
| """ |
| * Launch a four node, two DC cluster |
| * Start a repair on all nodes, by enumerating with -hosts |
| * Assert all nodes see a repair messages |
| * Assert no nodes anticompact |
| # TODO: Verify the anticompaction with sstablemetadata, not just logs |
| @jira_ticket CASSANDRA-10422 |
| """ |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| cluster.populate([2, 2]).start(wait_for_binary_proto=True) |
| node1_1, node2_1, node1_2, node2_2 = cluster.nodelist() |
| node1_1.stress(stress_options=['write', 'n=100K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=4)', '-rate', 'threads=50']) |
| node1_1.nodetool("repair -hosts 127.0.0.1,127.0.0.2,127.0.0.3,127.0.0.4 keyspace1 standard1") |
| for node in cluster.nodelist(): |
| assert node.grep_log("Not a global repair") |
| for node in cluster.nodelist(): |
| assert not node.grep_log("Starting anticompaction") |
| |
| @since('2.2.4', max_version='4') |
| def test_no_anticompaction_after_subrange_repair(self): |
| """ |
| * Launch a three node, two DC cluster |
| * Start a repair on a token range |
| * Assert all nodes see repair messages |
| * Assert no nodes anticompact |
| # TODO: Verify the anticompaction with sstablemetadata, not just logs |
| @jira_ticket CASSANDRA-10422 |
| """ |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| node1.stress(stress_options=['write', 'n=50K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=3)', '-rate', 'threads=50']) |
| node1.nodetool("repair -st 0 -et 1000 keyspace1 standard1") |
| for node in cluster.nodelist(): |
| assert node.grep_log("Not a global repair") |
| for node in cluster.nodelist(): |
| assert not node.grep_log("Starting anticompaction") |
| |
| def _get_repaired_data(self, node, keyspace): |
| """ |
| Based on incremental_repair_test.py:TestIncRepair implementation. |
| """ |
| _sstable_name = re.compile('SSTable: (.+)') |
| _repaired_at = re.compile(r'Repaired at: (\d+)') |
| _sstable_data = namedtuple('_sstabledata', ('name', 'repaired')) |
| |
| out = node.run_sstablemetadata(keyspace=keyspace).stdout |
| |
| def matches(pattern): |
| return [_f for _f in [pattern.match(l) for l in out.split('\n')] if _f] |
| |
| names = [m.group(1) for m in matches(_sstable_name)] |
| repaired_times = [int(m.group(1)) for m in matches(_repaired_at)] |
| |
| assert names |
| assert repaired_times |
| return [_sstable_data(*a) for a in zip(names, repaired_times)] |
| |
| @since('2.2.10', max_version='4') |
| def test_no_anticompaction_of_already_repaired(self): |
| """ |
| * Launch three node cluster and stress with RF2 |
| * Do incremental repair to have all sstables flagged as repaired |
| * Stop node2, stress, start again and run full -pr repair |
| * Verify that none of the already repaired sstables have been anti-compacted again |
| @jira_ticket CASSANDRA-13153 |
| """ |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| # disable JBOD conf since the test expects sstables to be on the same disk |
| cluster.set_datadir_count(1) |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| # we use RF to make sure to cover only a set of sub-ranges when doing -full -pr |
| node1.stress(stress_options=['write', 'n=50K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=2)', '-rate', 'threads=50']) |
| # disable compaction to make sure that we won't create any new sstables with repairedAt 0 |
| node1.nodetool('disableautocompaction keyspace1 standard1') |
| # Do incremental repair of all ranges. All sstables are expected for have repairedAt set afterwards. |
| node1.nodetool("repair keyspace1 standard1") |
| meta = self._get_repaired_data(node1, 'keyspace1') |
| repaired = set([m for m in meta if m.repaired > 0]) |
| assert len(repaired) == len(meta) |
| |
| # stop node2, stress and start full repair to find out how synced ranges affect repairedAt values |
| node2.stop(wait_other_notice=True) |
| node1.stress(stress_options=['write', 'n=40K', 'no-warmup', 'cl=ONE', '-rate', 'threads=50']) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| node1.nodetool("repair -full -pr keyspace1 standard1") |
| |
| meta = self._get_repaired_data(node1, 'keyspace1') |
| repairedAfterFull = set([m for m in meta if m.repaired > 0]) |
| # already repaired sstables must remain untouched |
| assert repaired.intersection(repairedAfterFull) == repaired |
| |
| @since('2.2.1', '4') |
| def test_anticompaction_after_normal_repair(self): |
| """ |
| * Launch a four node, two DC cluster |
| * Start a normal repair |
| * Assert every node anticompacts |
| @jira_ticket CASSANDRA-10422 |
| """ |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| cluster.populate([2, 2]).start(wait_for_binary_proto=True) |
| node1_1, node2_1, node1_2, node2_2 = cluster.nodelist() |
| node1_1.stress(stress_options=['write', 'n=50K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=4)']) |
| node1_1.nodetool("repair keyspace1 standard1") |
| for node in cluster.nodelist(): |
| assert "Starting anticompaction" |
| |
| def test_simple_sequential_repair(self): |
| """ |
| Calls simple repair test with a sequential repair |
| """ |
| self._simple_repair(sequential=True) |
| |
| def test_simple_parallel_repair(self): |
| """ |
| Calls simple repair test with a parallel repair |
| """ |
| self._simple_repair(sequential=False) |
| |
| def test_empty_vs_gcable_sequential_repair(self): |
| """ |
| Calls empty_vs_gcable repair test with a sequential repair |
| """ |
| self._empty_vs_gcable_no_repair(sequential=True) |
| |
| def test_empty_vs_gcable_parallel_repair(self): |
| """ |
| Calls empty_vs_gcable repair test with a parallel repair |
| """ |
| self._empty_vs_gcable_no_repair(sequential=False) |
| |
| def test_range_tombstone_digest_sequential_repair(self): |
| """ |
| Calls range_tombstone_digest with a sequential repair |
| """ |
| self._range_tombstone_digest(sequential=True) |
| |
| def test_range_tombstone_digest_parallel_repair(self): |
| """ |
| Calls range_tombstone_digest with a parallel repair |
| """ |
| self._range_tombstone_digest(sequential=False) |
| |
| @since('2.1') |
| def test_shadowed_cell_digest_sequential_repair(self): |
| """ |
| Calls _cell_shadowed_by_range_tombstone with sequential repair |
| """ |
| self._cell_shadowed_by_range_tombstone(sequential=True) |
| |
| @since('2.1') |
| def test_shadowed_cell_digest_parallel_repair(self): |
| """ |
| Calls _cell_shadowed_by_range_tombstone with parallel repair |
| """ |
| self._cell_shadowed_by_range_tombstone(sequential=False) |
| |
| @since('3.0') |
| def test_shadowed_range_tombstone_digest_sequential_repair(self): |
| """ |
| Calls _range_tombstone_shadowed_by_range_tombstone with sequential repair |
| """ |
| self._range_tombstone_shadowed_by_range_tombstone(sequential=True) |
| |
| @since('3.0') |
| def test_shadowed_range_tombstone_digest_parallel_repair(self): |
| """ |
| Calls _range_tombstone_shadowed_by_range_tombstone with parallel repair |
| """ |
| self._range_tombstone_shadowed_by_range_tombstone(sequential=False) |
| |
| @pytest.mark.no_vnodes |
| def test_simple_repair_order_preserving(self): |
| """ |
| Calls simple repair test with OPP and sequential repair |
| @jira_ticket CASSANDRA-5220 |
| """ |
| self._simple_repair(order_preserving_partitioner=True) |
| |
| def _simple_repair(self, order_preserving_partitioner=False, sequential=True): |
| """ |
| * Configure a three node cluster to not use hinted handoff, and to use batch commitlog |
| * Launch the cluster |
| * Create a keyspace at RF 3 and table |
| * Insert one thousand rows at CL ALL |
| * Flush on node3 and shut it down |
| * Insert one row at CL TWO |
| * Restart node3 |
| * Insert one thousand more rows at CL ALL |
| * Flush all nodes |
| * Check node3 only has 2000 keys |
| * Check node1 and node2 have 2001 keys |
| * Perform the repair type specified by the parent test |
| * Assert the appropriate messages are logged |
| * Assert node3 now has all data |
| |
| @jira_ticket CASSANDRA-4373 |
| """ |
| if order_preserving_partitioner: |
| self.cluster.set_partitioner('org.apache.cassandra.dht.ByteOrderedPartitioner') |
| |
| self._populate_cluster() |
| self._repair_and_verify(sequential) |
| |
| def _empty_vs_gcable_no_repair(self, sequential): |
| """ |
| Repairing empty partition and tombstoned partition older than gc grace |
| should be treated as the same and no repair is necessary. |
| @jira_ticket CASSANDRA-8979. |
| """ |
| cluster = self.cluster |
| cluster.populate(2) |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| cluster.start() |
| node1, node2 = cluster.nodelist() |
| |
| session = self.patient_cql_connection(node1) |
| # create keyspace with RF=2 to be able to be repaired |
| create_ks(session, 'ks', 2) |
| # we create two tables, one has low gc grace seconds so that the data |
| # can be dropped during test (but we don't actually drop them). |
| # the other has default gc. |
| # compaction is disabled not to purge data |
| query = """ |
| CREATE TABLE cf1 ( |
| key text, |
| c1 text, |
| c2 text, |
| PRIMARY KEY (key, c1) |
| ) |
| WITH gc_grace_seconds=1 |
| AND compaction = {'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'}; |
| """ |
| session.execute(query) |
| time.sleep(.5) |
| query = """ |
| CREATE TABLE cf2 ( |
| key text, |
| c1 text, |
| c2 text, |
| PRIMARY KEY (key, c1) |
| ) |
| WITH compaction = {'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'}; |
| """ |
| session.execute(query) |
| time.sleep(.5) |
| |
| # take down node2, so that only node1 has gc-able data |
| node2.stop(wait_other_notice=True) |
| for cf in ['cf1', 'cf2']: |
| # insert some data |
| for i in range(0, 10): |
| for j in range(0, 1000): |
| query = SimpleStatement("INSERT INTO {} (key, c1, c2) VALUES ('k{}', 'v{}', 'value')".format(cf, i, j), consistency_level=ConsistencyLevel.ONE) |
| session.execute(query) |
| node1.flush() |
| # delete those data, half with row tombstone, and the rest with cell range tombstones |
| for i in range(0, 5): |
| query = SimpleStatement("DELETE FROM {} WHERE key='k{}'".format(cf, i), consistency_level=ConsistencyLevel.ONE) |
| session.execute(query) |
| node1.flush() |
| for i in range(5, 10): |
| for j in range(0, 1000): |
| query = SimpleStatement("DELETE FROM {} WHERE key='k{}' AND c1='v{}'".format(cf, i, j), consistency_level=ConsistencyLevel.ONE) |
| session.execute(query) |
| node1.flush() |
| |
| # sleep until gc grace seconds pass so that cf1 can be dropped |
| time.sleep(2) |
| |
| # bring up node2 and repair |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| node2.repair(_repair_options(self.cluster.version(), ks='ks', sequential=sequential)) |
| |
| # check no rows will be returned |
| for cf in ['cf1', 'cf2']: |
| for i in range(0, 10): |
| query = SimpleStatement("SELECT c1, c2 FROM {} WHERE key='k{}'".format(cf, i), consistency_level=ConsistencyLevel.ALL) |
| res = list(session.execute(query)) |
| assert len([x for x in res if len(x) != 0]) == 0, res |
| |
| # check log for no repair happened for gcable data |
| out_of_sync_logs = node2.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync for cf1".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 0, "GC-able data does not need to be repaired with empty data: " + str([elt[0] for elt in out_of_sync_logs]) |
| # check log for actual repair for non gcable data |
| out_of_sync_logs = node2.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync for cf2".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) > 0, "Non GC-able data should be repaired" |
| |
| def _range_tombstone_digest(self, sequential): |
| """ |
| multiple range tombstones for same partition and interval must not create a digest mismatch as long |
| as the most recent tombstone is present. |
| @jira_ticket cassandra-11349. |
| """ |
| |
| def withsession(session, node1): |
| session.execute("delete from table1 where c1 = 'a' and c2 = 'b'") |
| node1.flush() |
| # recreate same tombstone (will be flushed by repair, so we end up with 2x on node1 and 1x on node2) |
| session.execute("delete from table1 where c1 = 'a' and c2 = 'b'") |
| |
| self._repair_digest(sequential, withsession) |
| |
| def _cell_shadowed_by_range_tombstone(self, sequential): |
| """ |
| Cells shadowed by range tombstones must not effect repairs (given tombstones are present on all nodes) |
| @jira_ticket CASSANDRA-11349. |
| """ |
| |
| def withSession(session, node1): |
| session.execute("INSERT INTO table1 (c1, c2, c3, c4) VALUES ('a', 'b', 'c', 1)") |
| node1.flush() |
| session.execute("DELETE FROM table1 WHERE c1 = 'a' AND c2 = 'b'") |
| |
| self._repair_digest(sequential, withSession) |
| |
| def _range_tombstone_shadowed_by_range_tombstone(self, sequential): |
| """ |
| Range tombstones shadowed by other range tombstones must not effect repairs |
| @jira_ticket CASSANDRA-11349. |
| """ |
| |
| def withSession(session, node1): |
| session.execute("DELETE FROM table1 WHERE c1 = 'a' AND c2 = 'b' AND c3 = 'c'") |
| node1.flush() |
| session.execute("DELETE FROM table1 WHERE c1 = 'a' AND c2 = 'b'") |
| node1.flush() |
| session.execute("DELETE FROM table1 WHERE c1 = 'a' AND c2 = 'b' AND c3 = 'd'") |
| node1.flush() |
| session.execute("DELETE FROM table1 WHERE c1 = 'a' AND c2 = 'b' AND c3 = 'a'") |
| |
| self._repair_digest(sequential, withSession) |
| |
| def _repair_digest(self, sequential, populate): |
| cluster = self.cluster |
| cluster.populate(2) |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| cluster.start() |
| node1, node2 = cluster.nodelist() |
| |
| session = self.patient_cql_connection(node1) |
| # create keyspace with RF=2 to be able to be repaired |
| create_ks(session, 'ks', 2) |
| query = """ |
| CREATE TABLE IF NOT EXISTS table1 ( |
| c1 text, |
| c2 text, |
| c3 text, |
| c4 float, |
| PRIMARY KEY (c1, c2, c3) |
| ) |
| WITH compaction = {'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'}; |
| """ |
| session.execute(query) |
| |
| populate(session, node1) |
| |
| node2.repair(_repair_options(self.cluster.version(), ks='ks', sequential=sequential)) |
| |
| # check log for no repair happened for gcable data |
| out_of_sync_logs = node2.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync for table1".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 0, "Digest mismatch for range tombstone: {}".format(str([elt[0] for elt in out_of_sync_logs])) |
| |
| def test_local_dc_repair(self): |
| """ |
| * Set up a multi DC cluster |
| * Perform a -local repair on one DC |
| * Assert only nodes in that DC are repaired |
| """ |
| cluster = self._setup_multi_dc() |
| node1 = cluster.nodes["node1"] |
| node2 = cluster.nodes["node2"] |
| |
| logger.debug("starting repair...") |
| opts = ["-local"] |
| opts += _repair_options(self.cluster.version(), ks="ks") |
| node1.repair(opts) |
| |
| # Verify that only nodes in dc1 are involved in repair |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 1, "Lines matching: {}".format(len(out_of_sync_logs)) |
| |
| line, m = out_of_sync_logs[0] |
| num_out_of_sync_ranges, out_of_sync_nodes = m.group(3), {m.group(1), m.group(2)} |
| |
| assert int(num_out_of_sync_ranges) == 1, "Expecting 1 range out of sync for {}, but saw {}".format(out_of_sync_nodes, num_out_of_sync_ranges) |
| valid_out_of_sync_pairs = {node1.address(), node2.address()} |
| assert out_of_sync_nodes == valid_out_of_sync_pairs, "Unrelated node found in local repair: {}, expected {}".format(out_of_sync_nodes, valid_out_of_sync_pairs) |
| # Check node2 now has the key |
| self.check_rows_on_node(node2, 2001, found=[1000], restart=False) |
| |
| def test_dc_repair(self): |
| """ |
| * Set up a multi DC cluster |
| * Perform a -dc repair on two dc's |
| * Assert only nodes on those dcs were repaired |
| """ |
| cluster = self._setup_multi_dc() |
| node1 = cluster.nodes["node1"] |
| node2 = cluster.nodes["node2"] |
| node3 = cluster.nodes["node3"] |
| |
| logger.debug("starting repair...") |
| opts = ["-dc", "dc1", "-dc", "dc2"] |
| opts += _repair_options(self.cluster.version(), ks="ks") |
| node1.repair(opts) |
| |
| # Verify that only nodes in dc1 and dc2 are involved in repair |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 2, "Lines matching: " + str([elt[0] for elt in out_of_sync_logs]) |
| valid_out_of_sync_pairs = [{node1.address(), node2.address()}, |
| {node2.address(), node3.address()}] |
| |
| for line, m in out_of_sync_logs: |
| num_out_of_sync_ranges, out_of_sync_nodes = m.group(3), {m.group(1), m.group(2)} |
| assert int(num_out_of_sync_ranges) == 1, "Expecting 1 range out of sync for {}, but saw {}".format(out_of_sync_nodes, num_out_of_sync_ranges) |
| assert out_of_sync_nodes, valid_out_of_sync_pairs in str(out_of_sync_nodes) |
| |
| # Check node2 now has the key |
| self.check_rows_on_node(node2, 2001, found=[1000], restart=False) |
| |
| def test_dc_parallel_repair(self): |
| """ |
| * Set up a multi DC cluster |
| * Perform a -dc repair on two dc's, with -dcpar |
| * Assert only nodes on those dcs were repaired |
| """ |
| cluster = self._setup_multi_dc() |
| node1 = cluster.nodes["node1"] |
| node2 = cluster.nodes["node2"] |
| node3 = cluster.nodes["node3"] |
| |
| logger.debug("starting repair...") |
| opts = ["-dc", "dc1", "-dc", "dc2", "-dcpar"] |
| opts += _repair_options(self.cluster.version(), ks="ks", sequential=False) |
| node1.repair(opts) |
| |
| # Verify that only nodes in dc1 and dc2 are involved in repair |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 2, "Lines matching: " + str([elt[0] for elt in out_of_sync_logs]) |
| valid_out_of_sync_pairs = [{node1.address(), node2.address()}, |
| {node2.address(), node3.address()}] |
| |
| for line, m in out_of_sync_logs: |
| num_out_of_sync_ranges, out_of_sync_nodes = m.group(3), {m.group(1), m.group(2)} |
| assert int(num_out_of_sync_ranges) == 1, "Expecting 1 range out of sync for {}, but saw {}".format(out_of_sync_nodes, num_out_of_sync_ranges) |
| assert out_of_sync_nodes, valid_out_of_sync_pairs in str(out_of_sync_nodes) |
| |
| # Check node2 now has the key |
| self.check_rows_on_node(node2, 2001, found=[1000], restart=False) |
| |
| # Check the repair was a dc parallel repair |
| if self.cluster.version() >= '2.2': |
| assert len(node1.grep_log('parallelism: dc_parallel')) == 1, str(node1.grep_log('parallelism')) |
| else: |
| assert len(node1.grep_log('parallelism=PARALLEL')) == 1, str(node1.grep_log('parallelism')) |
| |
| def _setup_multi_dc(self): |
| """ |
| Sets up 3 DCs (2 nodes in 'dc1', and one each in 'dc2' and 'dc3'). |
| After set up, node2 in dc1 lacks some data and needs to be repaired. |
| """ |
| cluster = self.cluster |
| |
| # Disable hinted handoff and set batch commit log so this doesn't |
| # interfer with the test (this must be after the populate) |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| # populate 2 nodes in dc1, and one node each in dc2 and dc3 |
| cluster.populate([2, 1, 1]).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3, node4 = cluster.nodelist() |
| session = self.patient_cql_connection(node1) |
| session.execute("CREATE KEYSPACE ks WITH replication = {'class': 'NetworkTopologyStrategy', 'dc1': 2, 'dc2': 1, 'dc3':1}") |
| session.execute("USE ks") |
| if cluster.version() < '4.0': |
| create_cf(session, 'cf', read_repair=0.0, columns={'c1': 'text', 'c2': 'text'}) |
| else: |
| create_cf(session, 'cf', columns={'c1': 'text', 'c2': 'text'}) |
| |
| # Insert 1000 keys, kill node 2, insert 1 key, restart node 2, insert 1000 more keys |
| logger.debug("Inserting data...") |
| insert_c1c2(session, n=1000, consistency=ConsistencyLevel.ALL) |
| node2.flush() |
| node2.stop(wait_other_notice=True) |
| insert_c1c2(session, keys=(1000, ), consistency=ConsistencyLevel.THREE) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| node1.watch_log_for_alive(node2) |
| insert_c1c2(session, keys=list(range(1001, 2001)), consistency=ConsistencyLevel.ALL) |
| |
| cluster.flush() |
| |
| # Verify that only node2 has only 2000 keys and others have 2001 keys |
| logger.debug("Checking data...") |
| self.check_rows_on_node(node2, 2000, missings=[1000]) |
| for node in [node1, node3, node4]: |
| self.check_rows_on_node(node, 2001, found=[1000]) |
| return cluster |
| |
| @since('2.2') |
| def parallel_table_repair_noleak(self): |
| """ |
| @jira_ticket CASSANDRA-11215 |
| |
| Tests that multiple parallel repairs on the same table isn't |
| causing reference leaks. |
| """ |
| self.fixture_dtest_setup.ignore_log_patterns = [ |
| "Cannot start multiple repair sessions over the same sstables", # The message we are expecting |
| "Validation failed in", # Expecting validation to fail |
| "RMI Runtime", # JMX Repair failures |
| "Session completed with the following error", # The nodetool repair error |
| "ValidationExecutor", # Errors by the validation executor |
| "RepairJobTask" # Errors by the repair job task |
| ] |
| |
| cluster = self.cluster |
| logger.debug("Starting cluster..") |
| cluster.populate([3]).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| node1.stress(stress_options=['write', 'n=10k', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=3)', '-rate', 'threads=50']) |
| |
| # Start multiple repairs in parallel |
| threads = [] |
| for i in range(3): |
| t = threading.Thread(target=node1.nodetool, args=("repair keyspace1 standard1",)) |
| threads.append(t) |
| t.start() |
| |
| # Wait for the repairs to finish |
| for t in threads: |
| t.join() |
| |
| found_message = False |
| # All nodes should reject multiple repairs and have no reference leaks |
| for node in cluster.nodelist(): |
| if len(node.grep_log("Cannot start multiple repair sessions over the same sstables")) > 0: |
| found_message = True |
| break |
| assert found_message |
| |
| @pytest.mark.no_vnodes |
| def test_token_range_repair(self): |
| """ |
| Test repair using the -st and -et options |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on a range that only belongs to node1 |
| * Verify that nodes 1 and 2, and only nodes 1+2, are repaired |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| self._parameterized_range_repair(repair_opts=['-st', str(node3.initial_token), '-et', str(node1.initial_token)]) |
| |
| @pytest.mark.no_vnodes |
| def test_token_range_repair_with_cf(self): |
| """ |
| @jira_ticket CASSANDRA-11866 |
| |
| Test repair using the -st and -et, and -cf options |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on a range that only belongs to node1 on the wrong cf |
| * Verify that the data did not get repaired |
| * Issue a repair on a range that belongs to the right cf |
| * Verify that the data was repaired |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| # Insert data, kill node 2, insert more data, restart node 2, insert another set of data |
| logger.debug("Inserting data...") |
| node1.stress(['write', 'n=1k', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', 'threads=30']) |
| node2.flush() |
| node2.stop(wait_other_notice=True) |
| node1.stress(['write', 'n=1K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=2)', '-rate', 'threads=30', '-pop', 'seq=20..40K']) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| node1.stress(['write', 'n=1K', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', 'threads=30', '-pop', 'seq=40..60K']) |
| cluster.flush() |
| |
| # Repair only the range node 1 owns on the wrong CF, assert everything is still broke |
| opts = ['-st', str(node3.initial_token), '-et', str(node1.initial_token), ] |
| opts += _repair_options(self.cluster.version(), ks='keyspace1', cf='counter1', sequential=False) |
| node1.repair(opts) |
| assert len(node1.grep_log('are consistent for standard1')) == 0, "Nodes 1 and 2 should not be consistent." |
| assert len(node3.grep_log('Repair command')) == 0, "Node 3 should not have been involved in the repair." |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| assert len(out_of_sync_logs) == 0, "We repaired the wrong CF == so things should still be broke" |
| |
| # Repair only the range node 1 owns on the right CF, assert everything is fixed |
| opts = ['-st', str(node3.initial_token), '-et', str(node1.initial_token), ] |
| opts += _repair_options(self.cluster.version(), ks='keyspace1', cf='standard1', sequential=False) |
| node1.repair(opts) |
| assert len(node1.grep_log('are consistent for standard1')) == 0, "Nodes 1 and 2 should not be consistent." |
| assert len(node3.grep_log('Repair command')) == 0, "Node 3 should not have been involved in the repair." |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| _, matches = out_of_sync_logs[0] |
| out_of_sync_nodes = {matches.group(1), matches.group(2)} |
| valid_out_of_sync_pairs = [{node1.address(), node2.address()}] |
| assert out_of_sync_nodes, valid_out_of_sync_pairs in str(out_of_sync_nodes) |
| |
| @pytest.mark.no_vnodes |
| def test_partitioner_range_repair(self): |
| """ |
| Test repair using the -pr option |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on a range that only belongs to node1 |
| * Verify that nodes 1 and 2, and only nodes 1+2, are repaired |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| self._parameterized_range_repair(repair_opts=['-pr']) |
| |
| @since('3.10') |
| @pytest.mark.no_vnodes |
| def test_pull_repair(self): |
| """ |
| Test repair using the --pull option |
| @jira_ticket CASSANDRA-9876 |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a pull repair on a range that only belongs to node1 |
| * Verify that nodes 1 and 2, and only nodes 1+2, are repaired |
| * Verify that node1 only received data |
| * Verify that node2 only sent data |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| node1_address = node1.network_interfaces['binary'][0] |
| node2_address = node2.network_interfaces['binary'][0] |
| |
| self._parameterized_range_repair(repair_opts=['--pull', '--in-hosts', node1_address + ',' + node2_address, '-st', str(node3.initial_token), '-et', str(node1.initial_token)]) |
| |
| # Node 1 should only receive files (as we ran a pull repair on node1) |
| assert len(node1.grep_log("Receiving [1-9][0-9]* files")) > 0 |
| assert len(node1.grep_log("sending [1-9][0-9]* files")) == 0 |
| assert len(node1.grep_log("sending 0 files")) > 0 |
| |
| # Node 2 should only send files (as we ran a pull repair on node1) |
| assert len(node2.grep_log("Receiving [1-9][0-9]* files")) == 0 |
| assert len(node2.grep_log("Receiving 0 files")) > 0 |
| assert len(node2.grep_log("sending [1-9][0-9]* files")) > 0 |
| |
| def _parameterized_range_repair(self, repair_opts): |
| """ |
| @param repair_opts A list of strings which represent cli args to nodetool repair |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on a range that only belongs to node1, using repair_opts |
| * Verify that nodes 1 and 2, and only nodes 1+2, are repaired |
| """ |
| cluster = self.cluster |
| node1, node2, node3 = cluster.nodelist() |
| |
| # Insert data, kill node 2, insert more data, restart node 2, insert another set of data |
| logger.debug("Inserting data...") |
| node1.stress(['write', 'n=20K', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', 'threads=30']) |
| |
| node2.flush() |
| node2.stop(wait_other_notice=True) |
| |
| node1.stress(['write', 'n=20K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=2)', '-rate', 'threads=30', '-pop', 'seq=20..40K']) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| |
| node1.stress(['write', 'n=20K', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', 'threads=30', '-pop', 'seq=40..60K']) |
| |
| cluster.flush() |
| |
| # Repair only the range node 1 owns |
| opts = repair_opts |
| opts += _repair_options(self.cluster.version(), ks='keyspace1', cf='standard1', sequential=False) |
| node1.repair(opts) |
| |
| assert len(node1.grep_log('are consistent for standard1')) == 0, "Nodes 1 and 2 should not be consistent." |
| assert len(node3.grep_log('Repair command')) == 0, "Node 3 should not have been involved in the repair." |
| |
| out_of_sync_logs = node1.grep_log(r"{} and {} have ([0-9]+) range\(s\) out of sync".format(cluster.address_regex(), cluster.address_regex())) |
| _, matches = out_of_sync_logs[0] |
| out_of_sync_nodes = {matches.group(1), matches.group(2)} |
| |
| valid_out_of_sync_pairs = [{node1.address(), node2.address()}] |
| |
| assert out_of_sync_nodes, valid_out_of_sync_pairs in str(out_of_sync_nodes) |
| |
| @since('2.2') |
| def test_trace_repair(self): |
| """ |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on to node1, setting job threads to 2 and with tracing enabled |
| * Check the trace data was written, and that the right job thread count was used |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| logger.debug("Inserting data...") |
| node1.stress(['write', 'n=20K', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', 'threads=30']) |
| |
| node2.flush() |
| node2.stop(wait_other_notice=True) |
| |
| node1.stress(['write', 'n=20K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=2)', '-rate', 'threads=30', '-pop', 'seq=20..40K']) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| |
| cluster.flush() |
| |
| job_thread_count = '2' |
| opts = ['-tr', '-j', job_thread_count] |
| opts += _repair_options(self.cluster.version(), ks='keyspace1', cf='standard1', sequential=False) |
| node1.repair(opts) |
| |
| time.sleep(5) # Give the trace table some time to populate |
| |
| session = self.patient_cql_connection(node1) |
| rows = list(session.execute("SELECT activity FROM system_traces.events")) |
| # This check assumes that the only (or at least first) thing to write to `system_traces.events.activity` is |
| # the repair task triggered in the test. |
| assert 'job threads: {}'.format(job_thread_count) in rows[0][0], \ |
| 'Expected {} job threads in repair options. Instead we saw {}'.format(job_thread_count, rows[0][0]) |
| |
| @since('2.2') |
| def test_thread_count_repair(self): |
| """ |
| * Launch a three node cluster |
| * Insert some data at RF 2 |
| * Shut down node2, insert more data, restore node2 |
| * Issue a repair on to node1, setting job threads |
| * Check the right job thread count was used |
| * Repeat steps 2 through 5 with all job count options |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Starting cluster..") |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| |
| node1, node2, node3 = cluster.nodelist() |
| |
| # Valid job thread counts: 1, 2, 3, and 4 |
| for job_thread_count in range(1, 5): |
| logger.debug("Inserting data...") |
| node1.stress(['write', 'n=2K', 'no-warmup', 'cl=ALL', '-schema', 'replication(factor=2)', '-rate', |
| 'threads=30', '-pop', 'seq={}..{}K'.format(2 * (job_thread_count - 1), 2 * job_thread_count)]) |
| |
| node2.flush() |
| node2.stop(wait_other_notice=True) |
| |
| node1.stress(['write', 'n=2K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=2)', '-rate', |
| 'threads=30', '-pop', 'seq={}..{}K'.format(2 * (job_thread_count), 2 * (job_thread_count + 1))]) |
| node2.start(wait_for_binary_proto=True, wait_other_notice=True) |
| |
| cluster.flush() |
| session = self.patient_cql_connection(node1) |
| session.execute("TRUNCATE system_traces.events") |
| |
| opts = ['-tr', '-j', str(job_thread_count)] |
| opts += _repair_options(self.cluster.version(), ks='keyspace1', cf='standard1', sequential=False) |
| node1.repair(opts) |
| |
| time.sleep(5) # Give the trace table some time to populate |
| |
| rows = list(session.execute("SELECT activity FROM system_traces.events")) |
| # This check assumes that the only (or at least first) thing to write to `system_traces.events.activity` is |
| # the repair task triggered in the test. |
| assert 'job threads: {}'.format(job_thread_count) in rows[0][0], \ |
| 'Expected {} job threads in repair options. Instead we saw {}'.format(job_thread_count, rows[0][0]) |
| |
| @pytest.mark.no_vnodes |
| def test_multiple_concurrent_repairs(self): |
| """ |
| @jira_ticket CASSANDRA-11451 |
| Make sure we can run sub range repairs in parallel - and verify that we actually do repair |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| node2.stop(wait_other_notice=True) |
| node1.stress(['write', 'n=1M', 'no-warmup', '-schema', 'replication(factor=3)', '-rate', 'threads=30']) |
| node2.start(wait_for_binary_proto=True) |
| t1 = threading.Thread(target=node1.nodetool, args=('repair keyspace1 standard1 -full -st {} -et {}'.format(str(node3.initial_token), str(node1.initial_token)),)) |
| t2 = threading.Thread(target=node2.nodetool, args=('repair keyspace1 standard1 -full -st {} -et {}'.format(str(node1.initial_token), str(node2.initial_token)),)) |
| t3 = threading.Thread(target=node3.nodetool, args=('repair keyspace1 standard1 -full -st {} -et {}'.format(str(node2.initial_token), str(node3.initial_token)),)) |
| t1.start() |
| t2.start() |
| t3.start() |
| t1.join() |
| t2.join() |
| t3.join() |
| node1.stop(wait_other_notice=True) |
| node3.stop(wait_other_notice=True) |
| _, _, rc = node2.stress(['read', 'n=1M', 'no-warmup', '-rate', 'threads=30'], whitelist=True) |
| assert rc == 0 |
| |
| @since('4.0') |
| def test_wide_row_repair(self): |
| """ |
| @jira_ticket CASSANDRA-13899 |
| Make sure compressed vs uncompressed blocks are handled correctly when stream decompressing |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.populate(2).start(wait_for_binary_proto=True) |
| node1, node2 = cluster.nodelist() |
| node2.stop(wait_other_notice=True) |
| profile_path = os.path.join(os.getcwd(), 'stress_profiles/repair_wide_rows.yaml') |
| logger.info(("yaml = " + profile_path)) |
| node1.stress(['user', 'profile=' + profile_path, 'n=50', 'ops(insert=1)', 'no-warmup', '-rate', 'threads=8', |
| '-insert', 'visits=FIXED(100K)', 'revisit=FIXED(100K)']) |
| node2.start(wait_for_binary_proto=True) |
| node2.repair() |
| |
| @since('2.1', max_version='4') |
| def test_dead_coordinator(self): |
| """ |
| @jira_ticket CASSANDRA-11824 |
| Make sure parent repair session is cleared out if the repair coordinator dies |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| node1.stress(['write', 'n=100k', '-schema', 'replication(factor=3)', '-rate', 'threads=30']) |
| |
| def run_repair(): |
| try: |
| if cluster.version() >= "2.2": |
| node1.repair() |
| else: |
| node1.nodetool('repair keyspace1 standard1 -inc -par') |
| except ToolError: |
| logger.debug("got expected exception during repair, ignoring") |
| |
| t1 = threading.Thread(target=run_repair) |
| t1.start() |
| if cluster.version() > "2.2": |
| node2.watch_log_for('Validating ValidationRequest', filename='debug.log') |
| else: |
| node1.watch_log_for('requesting merkle trees', filename='system.log') |
| time.sleep(2) |
| |
| logger.debug("stopping node1") |
| node1.stop(gently=False, wait_other_notice=True) |
| t1.join() |
| logger.debug("starting node1 - first repair should have failed") |
| node1.start(wait_for_binary_proto=True, wait_other_notice=True) |
| logger.debug("running second repair") |
| if cluster.version() >= "2.2": |
| node1.repair() |
| else: |
| node1.nodetool('repair keyspace1 standard1 -inc -par') |
| |
| @since('2.2') |
| def test_dead_sync_initiator(self): |
| """ |
| @jira_ticket CASSANDRA-12901 |
| """ |
| self._test_failure_during_repair(phase='sync', initiator=True) |
| |
| @since('2.2') |
| def test_dead_sync_participant(self): |
| """ |
| @jira_ticket CASSANDRA-12901 |
| """ |
| self._test_failure_during_repair(phase='sync', initiator=False,) |
| |
| @since('2.2', max_version='4') |
| def test_failure_during_anticompaction(self): |
| """ |
| @jira_ticket CASSANDRA-12901 |
| """ |
| self._test_failure_during_repair(phase='anticompaction',) |
| |
| @since('2.2') |
| def test_failure_during_validation(self): |
| """ |
| @jira_ticket CASSANDRA-12901 |
| """ |
| self._test_failure_during_repair(phase='validation') |
| |
| def _test_failure_during_repair(self, phase, initiator=False): |
| cluster = self.cluster |
| # We are not interested in specific errors, but |
| # that the repair session finishes on node failure without hanging |
| self.fixture_dtest_setup.ignore_log_patterns = [ |
| "Endpoint .* died", |
| "Streaming error occurred", |
| "StreamReceiveTask", |
| "Stream failed", |
| "Session completed with the following error", |
| "Repair session .* for range .* failed with error", |
| "Sync failed between .* and .*", |
| "failed to send a stream message/file to peer", |
| "failed to send a stream message/data to peer" |
| ] |
| |
| # Disable hinted handoff and set batch commit log so this doesn't |
| # interfere with the test (this must be after the populate) |
| cluster.set_configuration_options(values={'hinted_handoff_enabled': False}) |
| cluster.set_batch_commitlog(enabled=True) |
| logger.debug("Setting up cluster..") |
| cluster.populate(3) |
| node1, node2, node3 = cluster.nodelist() |
| |
| node_to_kill = node2 if (phase == 'sync' and initiator) else node3 |
| logger.debug("Setting up byteman on {}".format(node_to_kill.name)) |
| # set up byteman |
| node_to_kill.byteman_port = '8100' |
| node_to_kill.import_config_files() |
| |
| logger.debug("Starting cluster..") |
| cluster.start(wait_other_notice=True) |
| |
| logger.debug("stopping node3") |
| node3.stop(gently=False, wait_other_notice=True) |
| |
| self.patient_exclusive_cql_connection(node1) |
| logger.debug("inserting data while node3 is down") |
| node1.stress(stress_options=['write', 'n=1k', |
| 'no-warmup', 'cl=ONE', |
| '-schema', 'replication(factor=3)', |
| '-rate', 'threads=10']) |
| |
| logger.debug("bring back node3") |
| node3.start(wait_other_notice=True, wait_for_binary_proto=True) |
| |
| script = 'stream_sleep.btm' if phase == 'sync' else 'repair_{}_sleep.btm'.format(phase) |
| logger.debug("Submitting byteman script to {}".format(node_to_kill.name)) |
| # Sleep on anticompaction/stream so there will be time for node to be killed |
| node_to_kill.byteman_submit(['./byteman/{}'.format(script)]) |
| |
| def node1_repair(): |
| global nodetool_error |
| try: |
| node1.nodetool('repair keyspace1 standard1') |
| except Exception as e: |
| nodetool_error = e |
| |
| logger.debug("repair node1") |
| # Launch in a external thread so it does not hang process |
| t = Thread(target=node1_repair) |
| t.start() |
| |
| logger.debug("Will kill {} in middle of {}".format(node_to_kill.name, phase)) |
| msg_to_wait = 'streaming plan for Repair' |
| if phase == 'anticompaction': |
| msg_to_wait = 'Got anticompaction request' |
| elif phase == 'validation': |
| msg_to_wait = 'Validating' |
| node_to_kill.watch_log_for(msg_to_wait, filename='debug.log') |
| node_to_kill.stop(gently=False, wait_other_notice=True) |
| |
| logger.debug("Killed {}, now waiting repair to finish".format(node_to_kill.name)) |
| t.join(timeout=60) |
| assert not t.is_alive(), 'Repair still running after sync {} was killed'\ |
| .format("initiator" if initiator else "participant") |
| |
| if cluster.version() < '4.0' or phase != 'sync': |
| # the log entry we're watching for in the sync task came from the |
| # anti compaction at the end of the repair, which has been removed in 4.0 |
| node1.watch_log_for('Endpoint .* died', timeout=60) |
| node1.watch_log_for('Repair command .* finished', timeout=60) |
| |
| |
| RepairTableContents = namedtuple('RepairTableContents', |
| ['parent_repair_history', 'repair_history']) |
| |
| |
| @since('2.2') |
| @pytest.mark.resource_intensive |
| class TestRepairDataSystemTable(Tester): |
| """ |
| @jira_ticket CASSANDRA-5839 |
| |
| Tests the `system_distributed.parent_repair_history` and |
| `system_distributed.repair_history` tables by writing thousands of records |
| to a cluster, then ensuring these tables are in valid states before and |
| after running repair. |
| """ |
| @pytest.fixture(scope='function', autouse=True) |
| def fixture_set_cluster_settings(self, fixture_dtest_setup): |
| """ |
| Prepares a cluster for tests of the repair history tables by starting |
| a 5-node cluster, then inserting 5000 values with RF=3. |
| """ |
| fixture_dtest_setup.cluster.populate(5).start(wait_for_binary_proto=True) |
| self.node1 = self.cluster.nodelist()[0] |
| self.session = fixture_dtest_setup.patient_cql_connection(self.node1) |
| |
| self.node1.stress(stress_options=['write', 'n=5K', 'no-warmup', 'cl=ONE', '-schema', 'replication(factor=3)']) |
| |
| fixture_dtest_setup.cluster.flush() |
| |
| def repair_table_contents(self, node, include_system_keyspaces=True): |
| """ |
| @param node the node to connect to and query |
| @param include_system_keyspaces if truthy, return repair information about all keyspaces. If falsey, filter out keyspaces whose name contains 'system' |
| |
| Return a `RepairTableContents` `namedtuple` containing the rows in |
| `node`'s `system_distributed.parent_repair_history` and |
| `system_distributed.repair_history` tables. If `include_system_keyspaces`, |
| include all results. If not `include_system_keyspaces`, filter out |
| repair information about system keyspaces, or at least keyspaces with |
| 'system' in their names. |
| """ |
| session = self.patient_cql_connection(node) |
| |
| def execute_with_all(stmt): |
| return session.execute(SimpleStatement(stmt, consistency_level=ConsistencyLevel.ALL)) |
| |
| parent_repair_history = execute_with_all('SELECT * FROM system_distributed.parent_repair_history;') |
| repair_history = execute_with_all('SELECT * FROM system_distributed.repair_history;') |
| |
| if not include_system_keyspaces: |
| parent_repair_history = [row for row in parent_repair_history |
| if 'system' not in row.keyspace_name] |
| repair_history = [row for row in repair_history if |
| 'system' not in row.keyspace_name] |
| return RepairTableContents(parent_repair_history=parent_repair_history, |
| repair_history=repair_history) |
| |
| @pytest.mark.skip(reason='hangs CI') |
| def test_initial_empty_repair_tables(self): |
| logger.debug('repair tables:') |
| logger.debug(self.repair_table_contents(node=self.node1, include_system_keyspaces=False)) |
| repair_tables_dict = self.repair_table_contents(node=self.node1, include_system_keyspaces=False)._asdict() |
| for table_name, table_contents in list(repair_tables_dict.items()): |
| assert not table_contents, '{} is non-empty'.format(table_name) |
| |
| def test_repair_parent_table(self): |
| """ |
| Test that `system_distributed.parent_repair_history` is properly populated |
| after repair by: |
| |
| - running repair on `node` and |
| - checking that there are a non-zero number of entries in `parent_repair_history`. |
| """ |
| self.node1.repair() |
| parent_repair_history, _ = self.repair_table_contents(node=self.node1, include_system_keyspaces=False) |
| assert len(parent_repair_history) |
| |
| def test_repair_table(self): |
| """ |
| Test that `system_distributed.repair_history` is properly populated |
| after repair by: |
| |
| - running repair on `node` and |
| - checking that there are a non-zero number of entries in `repair_history`. |
| """ |
| self.node1.repair() |
| _, repair_history = self.repair_table_contents(node=self.node1, include_system_keyspaces=False) |
| assert len(repair_history) |
