| import time |
| import pytest |
| import re |
| import logging |
| |
| from datetime import datetime |
| from collections import Counter, namedtuple |
| from re import findall, compile |
| from uuid import UUID, uuid1 |
| |
| from cassandra import ConsistencyLevel |
| from cassandra.query import SimpleStatement |
| from cassandra.metadata import Murmur3Token |
| from ccmlib.common import is_win |
| from ccmlib.node import Node, ToolError |
| |
| from dtest import Tester, create_ks, create_cf |
| from tools.assertions import assert_almost_equal, assert_one |
| from tools.data import create_c1c2_table, insert_c1c2 |
| from tools.misc import new_node, ImmutableMapping |
| from tools.jmxutils import make_mbean, JolokiaAgent |
| |
| since = pytest.mark.since |
| logger = logging.getLogger(__name__) |
| |
| |
| class ConsistentState(object): |
| PREPARING = 0 |
| PREPARED = 1 |
| REPAIRING = 2 |
| FINALIZE_PROMISED = 3 |
| FINALIZED = 4 |
| FAILED = 5 |
| |
| |
| def assert_parent_repair_session_count(nodes, expected): |
| for node in nodes: |
| with JolokiaAgent(node) as jmx: |
| result = jmx.execute_method("org.apache.cassandra.db:type=RepairService", |
| "parentRepairSessionsCount") |
| assert expected == result, "The number of cached ParentRepairSessions should be {} but was {}. " \ |
| "This may mean that PRS objects are leaking on node {}. Check " \ |
| "ActiveRepairService for PRS clean up code.".format(expected, result, node.name) |
| |
| |
| class TestIncRepair(Tester): |
| |
| @pytest.fixture(autouse=True) |
| def fixture_add_additional_log_patterns(self, fixture_dtest_setup): |
| fixture_dtest_setup.ignore_log_patterns = ( |
| r'Can\'t send migration request: node.*is down' |
| ) |
| |
| @classmethod |
| def _get_repaired_data(cls, node, keyspace): |
| _sstable_name = compile('SSTable: (.+)') |
| _repaired_at = compile('Repaired at: (\d+)') |
| _pending_repair = compile('Pending repair: (\-\-|null|[a-f0-9\-]+)') |
| _sstable_data = namedtuple('_sstabledata', ('name', 'repaired', 'pending_id')) |
| |
| out = node.run_sstablemetadata(keyspace=keyspace).stdout |
| |
| def matches(pattern): |
| return filter(None, [pattern.match(l) for l in out.split('\n')]) |
| names = [m.group(1) for m in matches(_sstable_name)] |
| repaired_times = [int(m.group(1)) for m in matches(_repaired_at)] |
| |
| def uuid_or_none(s): |
| return None if s == 'null' or s == '--' else UUID(s) |
| pending_repairs = [uuid_or_none(m.group(1)) for m in matches(_pending_repair)] |
| assert names |
| assert repaired_times |
| assert pending_repairs |
| assert len(names) == len(repaired_times) == len(pending_repairs) |
| return [_sstable_data(*a) for a in zip(names, repaired_times, pending_repairs)] |
| |
| def assertNoRepairedSSTables(self, node, keyspace): |
| """ Checks that no sstables are marked repaired, and none are marked pending repair """ |
| data = self._get_repaired_data(node, keyspace) |
| assert all([t.repaired == 0 for t in data]), '{}'.format(data) |
| assert all([t.pending_id is None for t in data]) |
| |
| def assertAllPendingRepairSSTables(self, node, keyspace, pending_id=None): |
| """ Checks that no sstables are marked repaired, and all are marked pending repair """ |
| data = self._get_repaired_data(node, keyspace) |
| assert all([t.repaired == 0 for t in data]), '{}'.format(data) |
| if pending_id: |
| assert all([t.pending_id == pending_id for t in data]) |
| else: |
| assert all([t.pending_id is not None for t in data]) |
| |
| def assertAllRepairedSSTables(self, node, keyspace): |
| """ Checks that all sstables are marked repaired, and none are marked pending repair """ |
| data = self._get_repaired_data(node, keyspace) |
| assert all([t.repaired > 0 for t in data]), '{}'.format(data) |
| assert all([t.pending_id is None for t in data]), '{}'.format(data) |
| |
| def assertRepairedAndUnrepaired(self, node, keyspace): |
| """ Checks that a node has both repaired and unrepaired sstables for a given keyspace """ |
| data = self._get_repaired_data(node, keyspace) |
| assert any([t.repaired > 0 for t in data]), '{}'.format(data) |
| assert any([t.repaired == 0 for t in data]), '{}'.format(data) |
| assert all([t.pending_id is None for t in data]), '{}'.format(data) |
| |
| @since('4.0') |
| def test_consistent_repair(self): |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500}) |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| # make data inconsistent between nodes |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i)) |
| node3.flush() |
| time.sleep(1) |
| node3.stop(gently=False) |
| stmt.consistency_level = ConsistencyLevel.QUORUM |
| |
| session = self.exclusive_cql_connection(node1) |
| for i in range(10): |
| session.execute(stmt, (i + 10, i + 10)) |
| node1.flush() |
| time.sleep(1) |
| node1.stop(gently=False) |
| node3.start(wait_for_binary_proto=True) |
| session = self.exclusive_cql_connection(node2) |
| for i in range(10): |
| session.execute(stmt, (i + 20, i + 20)) |
| node1.start(wait_for_binary_proto=True) |
| |
| # flush and check that no sstables are marked repaired |
| for node in self.cluster.nodelist(): |
| node.flush() |
| self.assertNoRepairedSSTables(node, 'ks') |
| session = self.patient_exclusive_cql_connection(node) |
| results = list(session.execute("SELECT * FROM system.repairs")) |
| assert len(results) == 0, str(results) |
| |
| # disable compaction so we can verify sstables are marked pending repair |
| for node in self.cluster.nodelist(): |
| node.nodetool('disableautocompaction ks tbl') |
| |
| node1.repair(options=['ks']) |
| |
| # check that all participating nodes have the repair recorded in their system |
| # table, that all nodes are listed as participants, and that all sstables are |
| # (still) marked pending repair |
| expected_participants = {n.address() for n in self.cluster.nodelist()} |
| expected_participants_wp = {n.address_and_port() for n in self.cluster.nodelist()} |
| recorded_pending_ids = set() |
| for node in self.cluster.nodelist(): |
| session = self.patient_exclusive_cql_connection(node) |
| results = list(session.execute("SELECT * FROM system.repairs")) |
| assert len(results) == 1 |
| result = results[0] |
| assert set(result.participants) == expected_participants |
| if hasattr(result, "participants_wp"): |
| assert set(result.participants_wp) == expected_participants_wp |
| assert result.state, ConsistentState.FINALIZED == "4=FINALIZED" |
| pending_id = result.parent_id |
| self.assertAllPendingRepairSSTables(node, 'ks', pending_id) |
| recorded_pending_ids.add(pending_id) |
| |
| assert len(recorded_pending_ids) == 1 |
| |
| # sstables are compacted out of pending repair by a compaction |
| # task, we disabled compaction earlier in the test, so here we |
| # force the compaction and check that all sstables are promoted |
| for node in self.cluster.nodelist(): |
| node.nodetool('compact ks tbl') |
| self.assertAllRepairedSSTables(node, 'ks') |
| |
| def test_sstable_marking(self): |
| """ |
| * Launch a three node cluster |
| * Stop node3 |
| * Write 10K rows with stress |
| * Start node3 |
| * Issue an incremental repair, and wait for it to finish |
| * Run sstablemetadata on every node, assert that all sstables are marked as repaired |
| """ |
| # hinted handoff can create SSTable that we don't need after node3 restarted |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false'}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| node3.stop(gently=True) |
| |
| node1.stress(['write', 'n=10K', 'no-warmup', '-schema', 'replication(factor=3)']) |
| node1.flush() |
| node2.flush() |
| |
| node3.start() |
| if node3.get_cassandra_version() < '2.2': |
| log_file = 'system.log' |
| else: |
| log_file = 'debug.log' |
| node3.watch_log_for("Initializing keyspace1.standard1", filename=log_file) |
| # wait for things to settle before starting repair |
| time.sleep(1) |
| if self.cluster.version() >= "2.2": |
| node3.repair() |
| else: |
| node3.nodetool("repair -par -inc") |
| |
| if self.cluster.version() >= '4.0': |
| # sstables are compacted out of pending repair by a compaction |
| for node in self.cluster.nodelist(): |
| node.nodetool('compact keyspace1 standard1') |
| |
| for out in (node.run_sstablemetadata(keyspace='keyspace1').stdout for node in self.cluster.nodelist()): |
| assert 'Repaired at: 0' not in out |
| |
| def test_multiple_repair(self): |
| """ |
| * Launch a three node cluster |
| * Create a keyspace with RF 3 and a table |
| * Insert 49 rows |
| * Stop node3 |
| * Insert 50 more rows |
| * Restart node3 |
| * Issue an incremental repair on node3 |
| * Stop node2 |
| * Insert a final50 rows |
| * Restart node2 |
| * Issue an incremental repair on node2 |
| * Replace node3 with a new node |
| * Verify data integrity |
| # TODO: Several more verifications of data need to be interspersed throughout the test. The final assertion is insufficient. |
| @jira_ticket CASSANDRA-10644 |
| """ |
| cluster = self.cluster |
| cluster.populate(3).start() |
| node1, node2, node3 = cluster.nodelist() |
| |
| session = self.patient_cql_connection(node1) |
| 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'}) |
| |
| logger.debug("insert data") |
| |
| insert_c1c2(session, keys=list(range(1, 50)), consistency=ConsistencyLevel.ALL) |
| node1.flush() |
| |
| logger.debug("bringing down node 3") |
| node3.flush() |
| node3.stop(gently=False) |
| |
| logger.debug("inserting additional data into node 1 and 2") |
| insert_c1c2(session, keys=list(range(50, 100)), consistency=ConsistencyLevel.TWO) |
| node1.flush() |
| node2.flush() |
| |
| logger.debug("restarting and repairing node 3") |
| node3.start(wait_for_binary_proto=True) |
| |
| if cluster.version() >= "2.2": |
| node3.repair() |
| else: |
| node3.nodetool("repair -par -inc") |
| |
| # wait stream handlers to be closed on windows |
| # after session is finished (See CASSANDRA-10644) |
| if is_win: |
| time.sleep(2) |
| |
| logger.debug("stopping node 2") |
| node2.stop(gently=False) |
| |
| logger.debug("inserting data in nodes 1 and 3") |
| insert_c1c2(session, keys=list(range(100, 150)), consistency=ConsistencyLevel.TWO) |
| node1.flush() |
| node3.flush() |
| |
| logger.debug("start and repair node 2") |
| node2.start(wait_for_binary_proto=True) |
| |
| if cluster.version() >= "2.2": |
| node2.repair() |
| else: |
| node2.nodetool("repair -par -inc") |
| |
| logger.debug("replace node and check data integrity") |
| node3.stop(gently=False) |
| node5 = Node('node5', cluster, True, ('127.0.0.5', 9160), ('127.0.0.5', 7000), '7500', '0', None, ('127.0.0.5', 9042)) |
| cluster.add(node5, False, data_center="dc1") |
| node5.start(replace_address='127.0.0.3') |
| |
| assert_one(session, "SELECT COUNT(*) FROM ks.cf LIMIT 200", [149]) |
| |
| def test_sstable_repairedset(self): |
| """ |
| * Launch a two node cluster |
| * Insert data with stress |
| * Stop node2 |
| * Run sstablerepairedset against node2 |
| * Start node2 |
| * Run sstablemetadata on both nodes, pipe to a file |
| * Verify the output of sstablemetadata shows no repairs have occurred |
| * Stop node1 |
| * Insert more data with stress |
| * Start node1 |
| * Issue an incremental repair |
| * Run sstablemetadata on both nodes again, pipe to a new file |
| * Verify repairs occurred and repairedAt was updated |
| """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false'}) |
| self.init_default_config() |
| self.cluster.populate(2).start() |
| node1, node2 = self.cluster.nodelist() |
| node1.stress(['write', 'n=10K', 'no-warmup', '-schema', 'replication(factor=2)', 'compaction(strategy=SizeTieredCompactionStrategy,enabled=false)', '-rate', 'threads=50']) |
| |
| node1.flush() |
| node2.flush() |
| |
| node2.stop(gently=False) |
| |
| node2.run_sstablerepairedset(keyspace='keyspace1') |
| node2.start(wait_for_binary_proto=True) |
| |
| initialOut1 = node1.run_sstablemetadata(keyspace='keyspace1').stdout |
| initialOut2 = node2.run_sstablemetadata(keyspace='keyspace1').stdout |
| |
| matches = findall('(?<=Repaired at:).*', '\n'.join([initialOut1, initialOut2])) |
| logger.debug("Repair timestamps are: {}".format(matches)) |
| |
| uniquematches = set(matches) |
| matchcount = Counter(matches) |
| |
| assert len(uniquematches) >= 2, uniquematches |
| |
| assert len(max(matchcount)) >= 1, matchcount |
| |
| assert re.search('Repaired at: 0', '\n'.join([initialOut1, initialOut2])) |
| |
| node1.stop() |
| node2.stress(['write', 'n=15K', 'no-warmup', '-schema', 'replication(factor=2)']) |
| node2.flush() |
| node1.start(wait_for_binary_proto=True) |
| |
| if self.cluster.version() >= "2.2": |
| node1.repair() |
| else: |
| node1.nodetool("repair -par -inc") |
| |
| if self.cluster.version() >= '4.0': |
| # sstables are compacted out of pending repair by a compaction |
| for node in self.cluster.nodelist(): |
| node.nodetool('compact keyspace1 standard1') |
| |
| finalOut1 = node1.run_sstablemetadata(keyspace='keyspace1').stdout |
| if not isinstance(finalOut1, str): |
| finalOut1 = finalOut1 |
| finalOut2 = node2.run_sstablemetadata(keyspace='keyspace1').stdout |
| if not isinstance(finalOut2, str): |
| finalOut2 = finalOut2 |
| |
| matches = findall('(?<=Repaired at:).*', '\n'.join([finalOut1, finalOut2])) |
| |
| logger.debug(matches) |
| |
| uniquematches = set(matches) |
| matchcount = Counter(matches) |
| |
| assert len(uniquematches) >= 2 |
| |
| assert len(max(matchcount)) >= 2 |
| |
| assert not re.search('Repaired at: 0', '\n'.join([finalOut1, finalOut2])) |
| |
| def test_compaction(self): |
| """ |
| Test we can major compact after an incremental repair |
| * Launch a three node cluster |
| * Create a keyspace with RF 3 and a table |
| * Stop node3 |
| * Insert 100 rows |
| * Restart node3 |
| * Issue an incremental repair |
| * Insert 50 more rows |
| * Perform a major compaction on node3 |
| * Verify all data is present |
| # TODO: I have no idea what this is testing. The assertions do not verify anything meaningful. |
| # TODO: Fix all the string formatting |
| """ |
| cluster = self.cluster |
| cluster.populate(3).start() |
| node1, node2, node3 = cluster.nodelist() |
| |
| session = self.patient_cql_connection(node1) |
| create_ks(session, 'ks', 3) |
| session.execute("create table tab(key int PRIMARY KEY, val int);") |
| |
| node3.stop() |
| |
| for x in range(0, 100): |
| session.execute("insert into tab(key,val) values(" + str(x) + ",0)") |
| node1.flush() |
| |
| node3.start(wait_for_binary_proto=True) |
| |
| if cluster.version() >= "2.2": |
| node3.repair() |
| else: |
| node3.nodetool("repair -par -inc") |
| for x in range(0, 150): |
| session.execute("insert into tab(key,val) values(" + str(x) + ",1)") |
| |
| cluster.flush() |
| |
| node3.nodetool('compact') |
| |
| for x in range(0, 150): |
| assert_one(session, "select val from tab where key =" + str(x), [1]) |
| |
| @since("2.2") |
| def test_multiple_full_repairs_lcs(self): |
| """ |
| @jira_ticket CASSANDRA-11172 - repeated full repairs should not cause infinite loop in getNextBackgroundTask |
| """ |
| cluster = self.cluster |
| cluster.populate(2).start() |
| node1, node2 = cluster.nodelist() |
| for x in range(0, 10): |
| node1.stress(['write', 'n=100k', 'no-warmup', '-rate', 'threads=10', '-schema', 'compaction(strategy=LeveledCompactionStrategy,sstable_size_in_mb=10)', 'replication(factor=2)']) |
| cluster.flush() |
| cluster.wait_for_compactions() |
| node1.nodetool("repair -full keyspace1 standard1") |
| |
| @pytest.mark.env("long") |
| @pytest.mark.skip(reason='hangs CI') |
| def test_multiple_subsequent_repair(self): |
| """ |
| @jira_ticket CASSANDRA-8366 |
| |
| There is an issue with subsequent inc repairs increasing load size. |
| So we perform several repairs and check that the expected amount of data exists. |
| * Launch a three node cluster |
| * Write 5M rows with stress |
| * Wait for minor compactions to finish |
| * Issue an incremental repair on each node, sequentially |
| * Issue major compactions on each node |
| * Sleep for a while so load size can be propagated between nodes |
| * Verify the correct amount of data is on each node |
| """ |
| cluster = self.cluster |
| cluster.populate(3).start() |
| node1, node2, node3 = cluster.nodelist() |
| |
| logger.debug("Inserting data with stress") |
| node1.stress(['write', 'n=5M', 'no-warmup', '-rate', 'threads=10', '-schema', 'replication(factor=3)']) |
| |
| logger.debug("Flushing nodes") |
| cluster.flush() |
| |
| logger.debug("Waiting compactions to finish") |
| cluster.wait_for_compactions() |
| |
| if self.cluster.version() >= '2.2': |
| logger.debug("Repairing node1") |
| node1.nodetool("repair") |
| logger.debug("Repairing node2") |
| node2.nodetool("repair") |
| logger.debug("Repairing node3") |
| node3.nodetool("repair") |
| else: |
| logger.debug("Repairing node1") |
| node1.nodetool("repair -par -inc") |
| logger.debug("Repairing node2") |
| node2.nodetool("repair -par -inc") |
| logger.debug("Repairing node3") |
| node3.nodetool("repair -par -inc") |
| |
| # Using "print" instead of logger.debug() here is on purpose. The compactions |
| # take a long time and don't print anything by default, which can result |
| # in the test being timed out after 20 minutes. These print statements |
| # prevent it from being timed out. |
| print("compacting node1") |
| node1.compact() |
| print("compacting node2") |
| node2.compact() |
| print("compacting node3") |
| node3.compact() |
| |
| # wait some time to be sure the load size is propagated between nodes |
| logger.debug("Waiting for load size info to be propagated between nodes") |
| time.sleep(45) |
| |
| load_size_in_kb = float(sum([n.data_size() for n in [node1, node2, node3]])) |
| load_size = load_size_in_kb / 1024 / 1024 |
| logger.debug("Total Load size: {}GB".format(load_size)) |
| |
| # There is still some overhead, but it's lot better. We tolerate 25%. |
| expected_load_size = 4.5 # In GB |
| assert_almost_equal(load_size, expected_load_size, error=0.25) |
| |
| @pytest.mark.resource_intensive |
| def test_sstable_marking_not_intersecting_all_ranges(self): |
| """ |
| @jira_ticket CASSANDRA-10299 |
| * Launch a four node cluster |
| * Insert data with stress |
| * Issue an incremental repair on each node sequentially |
| * Assert no extra, unrepaired sstables are generated |
| """ |
| cluster = self.cluster |
| cluster.populate(4).start() |
| node1, node2, node3, node4 = cluster.nodelist() |
| |
| logger.debug("Inserting data with stress") |
| node1.stress(['write', 'n=3', 'no-warmup', '-rate', 'threads=1', '-schema', 'replication(factor=3)']) |
| |
| logger.debug("Flushing nodes") |
| cluster.flush() |
| |
| repair_options = '' if self.cluster.version() >= '2.2' else '-inc -par' |
| |
| logger.debug("Repairing node 1") |
| node1.nodetool("repair {}".format(repair_options)) |
| logger.debug("Repairing node 2") |
| node2.nodetool("repair {}".format(repair_options)) |
| logger.debug("Repairing node 3") |
| node3.nodetool("repair {}".format(repair_options)) |
| logger.debug("Repairing node 4") |
| node4.nodetool("repair {}".format(repair_options)) |
| |
| if cluster.version() >= '4.0': |
| # sstables are compacted out of pending repair by a compaction |
| for node in cluster.nodelist(): |
| node.nodetool('compact keyspace1 standard1') |
| |
| for out in (node.run_sstablemetadata(keyspace='keyspace1').stdout for node in cluster.nodelist() if len(node.get_sstables('keyspace1', 'standard1')) > 0): |
| assert 'Repaired at: 0' not in out |
| |
| @pytest.mark.no_vnodes |
| @since('4.0') |
| def test_move(self): |
| """ Test repaired data remains in sync after a move """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(4, tokens=[0, 2**32, 2**48, -(2**32)]).start() |
| node1, node2, node3, node4 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| |
| # insert some data |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| for i in range(1000): |
| session.execute(stmt, (i, i)) |
| |
| node1.repair(options=['ks']) |
| |
| for i in range(1000): |
| v = i + 1000 |
| session.execute(stmt, (v, v)) |
| |
| # everything should be in sync |
| for node in self.cluster.nodelist(): |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| node2.nodetool('move {}'.format(2**16)) |
| |
| # everything should still be in sync |
| for node in self.cluster.nodelist(): |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| @pytest.mark.no_vnodes |
| @since('4.0') |
| def test_decommission(self): |
| """ Test repaired data remains in sync after a decommission """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(4).start() |
| node1, node2, node3, node4 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| |
| # insert some data |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| for i in range(1000): |
| session.execute(stmt, (i, i)) |
| |
| node1.repair(options=['ks']) |
| |
| for i in range(1000): |
| v = i + 1000 |
| session.execute(stmt, (v, v)) |
| |
| # everything should be in sync |
| for node in self.cluster.nodelist(): |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| node2.nodetool('decommission') |
| |
| # everything should still be in sync |
| for node in [node1, node3, node4]: |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| @pytest.mark.no_vnodes |
| @since('4.0') |
| def test_bootstrap(self): |
| """ Test repaired data remains in sync after a bootstrap """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| |
| # insert some data |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| for i in range(1000): |
| session.execute(stmt, (i, i)) |
| |
| node1.repair(options=['ks']) |
| |
| for i in range(1000): |
| v = i + 1000 |
| session.execute(stmt, (v, v)) |
| |
| # everything should be in sync |
| for node in [node1, node2, node3]: |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| node4 = new_node(self.cluster) |
| node4.start(wait_for_binary_proto=True) |
| |
| assert len(self.cluster.nodelist()) == 4 |
| # everything should still be in sync |
| for node in self.cluster.nodelist(): |
| result = node.repair(options=['ks', '--validate']) |
| assert "Repaired data is in sync" in result.stdout |
| |
| @since('4.0') |
| def test_force(self): |
| """ |
| forcing an incremental repair should incrementally repair any nodes |
| that are up, but should not promote the sstables to repaired |
| """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i)) |
| |
| node2.stop(wait_other_notice=True) |
| |
| # repair should fail because node2 is down |
| with pytest.raises(ToolError): |
| node1.repair(options=['ks']) |
| |
| # run with force flag |
| node1.repair(options=['ks', '--force']) |
| |
| # ... and verify nothing was promoted to repaired |
| self.assertNoRepairedSSTables(node1, 'ks') |
| self.assertNoRepairedSSTables(node2, 'ks') |
| |
| @since('4.0') |
| def test_force_with_none_down(self): |
| """ |
| if we force an incremental repair, but all the involved nodes are up, |
| we should run normally and promote sstables afterwards |
| """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i)) |
| |
| # run with force flag |
| node1.repair(options=['ks', '--force']) |
| |
| # ... and verify everything was still promoted |
| self.assertAllRepairedSSTables(node1, 'ks') |
| self.assertAllRepairedSSTables(node2, 'ks') |
| self.assertAllRepairedSSTables(node3, 'ks') |
| |
| @since('4.0') |
| def test_hosts(self): |
| """ |
| running an incremental repair with hosts specified should incrementally repair |
| the given nodes, but should not promote the sstables to repaired |
| """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i)) |
| |
| # run with force flag |
| node1.repair(options=['ks', '-hosts', ','.join([node1.address(), node2.address()])]) |
| |
| # ... and verify nothing was promoted to repaired |
| self.assertNoRepairedSSTables(node1, 'ks') |
| self.assertNoRepairedSSTables(node2, 'ks') |
| |
| @since('4.0') |
| def test_subrange(self): |
| """ |
| running an incremental repair with hosts specified should incrementally repair |
| the given nodes, but should not promote the sstables to repaired |
| """ |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500, |
| 'partitioner': 'org.apache.cassandra.dht.Murmur3Partitioner'}) |
| self.init_default_config() |
| self.cluster.populate(3).start() |
| node1, node2, node3 = self.cluster.nodelist() |
| |
| session = self.patient_exclusive_cql_connection(node3) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 3}") |
| session.execute("CREATE TABLE ks.tbl (k INT PRIMARY KEY, v INT)") |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k,v) VALUES (%s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| self.assertNoRepairedSSTables(node, 'ks') |
| |
| # only repair the partition k=0 |
| token = Murmur3Token.from_key(bytes([0, 0, 0, 0])) |
| # import ipdb; ipdb.set_trace() |
| # run with force flag |
| node1.repair(options=['ks', '-st', str(token.value - 1), '-et', str(token.value)]) |
| |
| # verify we have a mix of repaired and unrepaired sstables |
| self.assertRepairedAndUnrepaired(node1, 'ks') |
| self.assertRepairedAndUnrepaired(node2, 'ks') |
| self.assertRepairedAndUnrepaired(node3, 'ks') |
| |
| @since('4.0') |
| def test_repaired_tracking_with_partition_deletes(self): |
| """ |
| check that when an tracking repaired data status following a digest mismatch, |
| repaired data mismatches are marked as unconfirmed as we may skip sstables |
| after the partition delete are encountered. |
| @jira_ticket CASSANDRA-14145 |
| """ |
| session, node1, node2 = self.setup_for_repaired_data_tracking() |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k, c, v) VALUES (%s, %s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| self.assertNoRepairedSSTables(node, 'ks') |
| |
| node1.repair(options=['ks']) |
| node2.stop(wait_other_notice=True) |
| |
| session.execute("delete from ks.tbl where k = 5") |
| |
| node1.flush() |
| node2.start() |
| |
| # expect unconfirmed inconsistencies as the partition deletes cause some sstables to be skipped |
| with JolokiaAgent(node1) as jmx: |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5", |
| expect_unconfirmed_inconsistencies=True) |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5 AND c = 5", |
| expect_unconfirmed_inconsistencies=True) |
| # no digest reads for range queries so blocking read repair metric isn't incremented |
| # *all* sstables are read for partition ranges too, and as the repaired set is still in sync there should |
| # be no inconsistencies |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl", expect_read_repair=False) |
| |
| @since('4.0') |
| def test_repaired_tracking_with_varying_sstable_sets(self): |
| """ |
| verify that repaired data digests are computed over the merged data for each replica |
| and that the particular number of sstables on each doesn't affect the comparisons |
| both replicas start with the same repaired set, comprising 2 sstables. node1's is |
| then compacted and additional unrepaired data added (which overwrites some in the |
| repaired set). We expect the repaired digests to still match as the tracking will |
| force all sstables containing the partitions to be read |
| there are two variants of this, for single partition slice & names reads and range reads |
| @jira_ticket CASSANDRA-14145 |
| """ |
| session, node1, node2 = self.setup_for_repaired_data_tracking() |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k, c, v) VALUES (%s, %s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| |
| for i in range(10,20): |
| session.execute(stmt, (i, i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| self.assertNoRepairedSSTables(node, 'ks') |
| |
| node1.repair(options=['ks']) |
| node2.stop(wait_other_notice=True) |
| |
| session.execute("insert into ks.tbl (k, c, v) values (5, 5, 55)") |
| session.execute("insert into ks.tbl (k, c, v) values (15, 15, 155)") |
| node1.flush() |
| node1.compact() |
| node1.compact() |
| node2.start() |
| |
| # we don't expect any inconsistencies as all repaired data is read on both replicas |
| with JolokiaAgent(node1) as jmx: |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5") |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5 AND c = 5") |
| # no digest reads for range queries so read repair metric isn't incremented |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl", expect_read_repair=False) |
| |
| @since('4.0') |
| def test_repaired_tracking_with_mismatching_replicas(self): |
| """ |
| verify that when replicas have different repaired sets, this can be detected via the digests |
| computed at read time. All nodes have start with the same data, but only 1 replica's sstables |
| are marked repaired. Then a divergence is introduced by overwriting on 1 replica only, which |
| is required to trigger a digest mismatch & full data read (for single partition reads). |
| As the repaired sets are different between the replicas, but no other shortcutting occurs |
| (no partition tombstones or sstable skipping) and no sstables are involved in pending repair |
| session, we expect confirmed inconsistencies to be reported. |
| there are two variants of this, for single partition slice & names reads and range reads |
| @jira_ticket CASSANDRA-14145 |
| """ |
| session, node1, node2 = self.setup_for_repaired_data_tracking() |
| stmt = SimpleStatement("INSERT INTO ks.tbl (k, c, v) VALUES (%s, %s, %s)") |
| stmt.consistency_level = ConsistencyLevel.ALL |
| for i in range(10): |
| session.execute(stmt, (i, i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| |
| for i in range(10,20): |
| session.execute(stmt, (i, i, i)) |
| |
| for node in self.cluster.nodelist(): |
| node.flush() |
| self.assertNoRepairedSSTables(node, 'ks') |
| |
| # stop node 2 and mark its sstables repaired |
| node2.stop(wait_other_notice=True) |
| node2.run_sstablerepairedset(keyspace='ks') |
| # before restarting node2 overwrite some data on node1 to trigger digest mismatches |
| session.execute("insert into ks.tbl (k, c, v) values (5, 5, 55)") |
| node2.start(wait_for_binary_proto=True) |
| |
| out1 = node1.run_sstablemetadata(keyspace='ks').stdout |
| out2 = node2.run_sstablemetadata(keyspace='ks').stdout |
| |
| # verify the repaired at times for the sstables on node1/node2 |
| assert all(t == 0 for t in [int(x) for x in [y.split(' ')[0] for y in findall('(?<=Repaired at: ).*', out1)]]) |
| assert all(t > 0 for t in [int(x) for x in [y.split(' ')[0] for y in findall('(?<=Repaired at: ).*', out2)]]) |
| |
| # we expect inconsistencies due to sstables being marked repaired on one replica only |
| # these are marked confirmed because no sessions are pending & all sstables are |
| # skipped due to partition deletes |
| with JolokiaAgent(node1) as jmx: |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5", |
| expect_confirmed_inconsistencies=True) |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl WHERE k = 5 AND c = 5", |
| expect_confirmed_inconsistencies=True) |
| # no digest reads for range queries so read repair metric isn't incremented |
| self.query_and_check_repaired_mismatches(jmx, session, "SELECT * FROM ks.tbl", |
| expect_confirmed_inconsistencies=True, |
| expect_read_repair=False) |
| |
| @since('4.0') |
| def test_parent_repair_session_cleanup(self): |
| """ |
| Calls incremental repair on 3 node cluster and verifies if all ParentRepairSession objects are cleaned |
| @jira_ticket CASSANDRA-16446 |
| """ |
| self.cluster.populate(3).start() |
| session = self.patient_cql_connection(self.cluster.nodelist()[0]) |
| create_ks(session, 'ks', 2) |
| create_c1c2_table(self, session) |
| |
| for node in self.cluster.nodelist(): |
| node.repair(options=['ks']) |
| |
| assert_parent_repair_session_count(self.cluster.nodelist(), 0) |
| |
| def setup_for_repaired_data_tracking(self): |
| self.fixture_dtest_setup.setup_overrides.cluster_options = ImmutableMapping({'hinted_handoff_enabled': 'false', |
| 'num_tokens': 1, |
| 'commitlog_sync_period_in_ms': 500}) |
| self.init_default_config() |
| self.cluster.populate(2) |
| node1, node2 = self.cluster.nodelist() |
| self.cluster.start() |
| |
| session = self.patient_exclusive_cql_connection(node1) |
| session.execute("CREATE KEYSPACE ks WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 2}") |
| session.execute("CREATE TABLE ks.tbl (k INT, c INT, v INT, PRIMARY KEY (k,c)) with read_repair='NONE'") |
| return session, node1, node2 |
| |
| def query_and_check_repaired_mismatches(self, jmx, session, query, |
| expect_read_repair=True, |
| expect_unconfirmed_inconsistencies=False, |
| expect_confirmed_inconsistencies=False): |
| |
| rr_count = make_mbean('metrics', type='ReadRepair', name='ReconcileRead') |
| unconfirmed_count = make_mbean('metrics', type='Table,keyspace=ks', name='RepairedDataInconsistenciesUnconfirmed,scope=tbl') |
| confirmed_count = make_mbean('metrics', type='Table,keyspace=ks', name='RepairedDataInconsistenciesConfirmed,scope=tbl') |
| |
| rr_before = self.get_attribute_count(jmx, rr_count) |
| uc_before = self.get_attribute_count(jmx, unconfirmed_count) |
| cc_before = self.get_attribute_count(jmx, confirmed_count) |
| |
| stmt = SimpleStatement(query) |
| stmt.consistency_level = ConsistencyLevel.ALL |
| session.execute(stmt) |
| |
| rr_after = self.get_attribute_count(jmx, rr_count) |
| uc_after = self.get_attribute_count(jmx, unconfirmed_count) |
| cc_after = self.get_attribute_count(jmx, confirmed_count) |
| |
| logger.debug("Read Repair Count: {before}, {after}".format(before=rr_before, after=rr_after)) |
| logger.debug("Unconfirmed Inconsistency Count: {before}, {after}".format(before=uc_before, after=uc_after)) |
| logger.debug("Confirmed Inconsistency Count: {before}, {after}".format(before=cc_before, after=cc_after)) |
| |
| if expect_read_repair: |
| assert rr_after > rr_before |
| else: |
| assert rr_after == rr_before |
| |
| if expect_unconfirmed_inconsistencies: |
| assert uc_after > uc_before |
| else: |
| assert uc_after == uc_before |
| |
| if expect_confirmed_inconsistencies: |
| assert cc_after > cc_before |
| else: |
| assert cc_after == cc_before |
| |
| def get_attribute_count(self, jmx, bean): |
| # 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 the count can only be 0 |
| if jmx.has_mbean(bean): |
| return jmx.read_attribute(bean, 'Count') |
| else: |
| return 0 |
