| import json |
| import os |
| import random |
| import re |
| import subprocess |
| |
| from ccmlib import common |
| from ccmlib.node import ToolError |
| |
| from dtest import Tester, debug, create_ks |
| from tools.decorators import since |
| |
| |
| class TestOfflineTools(Tester): |
| |
| # In 2.0, we will get this error log message due to jamm not being |
| # in the classpath |
| ignore_log_patterns = ["Unable to initialize MemoryMeter"] |
| |
| def sstablelevelreset_test(self): |
| """ |
| Insert data and call sstablelevelreset on a series of |
| tables. Confirm level is reset to 0 using its output. |
| Test a variety of possible errors and ensure response is resonable. |
| @since 2.1.5 |
| @jira_ticket CASSANDRA-7614 |
| """ |
| cluster = self.cluster |
| cluster.populate(1).start(wait_for_binary_proto=True) |
| node1 = cluster.nodelist()[0] |
| |
| # test by trying to run on nonexistent keyspace |
| cluster.stop(gently=False) |
| try: |
| node1.run_sstablelevelreset("keyspace1", "standard1") |
| except ToolError as e: |
| self.assertIn("ColumnFamily not found: keyspace1/standard1", e.message) |
| # this should return exit code 1 |
| self.assertEqual(e.exit_status, 1, "Expected sstablelevelreset to have a return code of 1, but instead return code was {}".format(e.exit_status)) |
| |
| # now test by generating keyspace but not flushing sstables |
| cluster.start(wait_for_binary_proto=True) |
| node1.stress(['write', 'n=100', 'no-warmup', '-schema', 'replication(factor=1)', |
| '-rate', 'threads=8']) |
| cluster.stop(gently=False) |
| |
| output, error, rc = node1.run_sstablelevelreset("keyspace1", "standard1") |
| self._check_stderr_error(error) |
| self.assertIn("Found no sstables, did you give the correct keyspace", output) |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| # test by writing small amount of data and flushing (all sstables should be level 0) |
| cluster.start(wait_for_binary_proto=True) |
| session = self.patient_cql_connection(node1) |
| session.execute("ALTER TABLE keyspace1.standard1 with compaction={'class': 'LeveledCompactionStrategy', 'sstable_size_in_mb':1};") |
| node1.stress(['write', 'n=1K', 'no-warmup', '-schema', 'replication(factor=1)', |
| '-rate', 'threads=8']) |
| node1.flush() |
| cluster.stop(gently=False) |
| |
| output, error, rc = node1.run_sstablelevelreset("keyspace1", "standard1") |
| self._check_stderr_error(error) |
| self.assertIn("since it is already on level 0", output) |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| # test by loading large amount data so we have multiple levels and checking all levels are 0 at end |
| cluster.start(wait_for_binary_proto=True) |
| node1.stress(['write', 'n=50K', 'no-warmup', '-schema', 'replication(factor=1)', |
| '-rate', 'threads=8']) |
| cluster.flush() |
| self.wait_for_compactions(node1) |
| cluster.stop() |
| |
| initial_levels = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"])) |
| _, error, rc = node1.run_sstablelevelreset("keyspace1", "standard1") |
| final_levels = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"])) |
| self._check_stderr_error(error) |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| debug(initial_levels) |
| debug(final_levels) |
| |
| # let's make sure there was at least L1 beforing resetting levels |
| self.assertTrue(max(initial_levels) > 0) |
| |
| # let's check all sstables are on L0 after sstablelevelreset |
| self.assertTrue(max(final_levels) == 0) |
| |
| def get_levels(self, data): |
| (out, err, rc) = data |
| return map(int, re.findall("SSTable Level: ([0-9])", out)) |
| |
| def wait_for_compactions(self, node): |
| pattern = re.compile("pending tasks: 0") |
| while True: |
| output, err, _ = node.nodetool("compactionstats") |
| if pattern.search(output): |
| break |
| |
| def sstableofflinerelevel_test(self): |
| """ |
| Generate sstables of varying levels. |
| Reset sstables to L0 with sstablelevelreset |
| Run sstableofflinerelevel and ensure tables are promoted correctly |
| Also test a variety of bad inputs including nonexistent keyspace and sstables |
| @since 2.1.5 |
| @jira_ticket CASSANRDA-8031 |
| """ |
| cluster = self.cluster |
| cluster.set_configuration_options(values={'compaction_throughput_mb_per_sec': 0}) |
| cluster.populate(1).start(wait_for_binary_proto=True) |
| node1 = cluster.nodelist()[0] |
| |
| # NOTE - As of now this does not return when it encounters Exception and causes test to hang, temporarily commented out |
| # test by trying to run on nonexistent keyspace |
| # cluster.stop(gently=False) |
| # output, error, rc = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True) |
| # self.assertTrue("java.lang.IllegalArgumentException: Unknown keyspace/columnFamily keyspace1.standard1" in error) |
| # # this should return exit code 1 |
| # self.assertEqual(rc, 1, msg=str(rc)) |
| # cluster.start() |
| |
| # now test by generating keyspace but not flushing sstables |
| |
| node1.stress(['write', 'n=1', 'no-warmup', |
| '-schema', 'replication(factor=1)', |
| '-col', 'n=FIXED(10)', 'SIZE=FIXED(1024)', |
| '-rate', 'threads=8']) |
| |
| cluster.stop(gently=False) |
| try: |
| output, error, _ = node1.run_sstableofflinerelevel("keyspace1", "standard1") |
| except ToolError as e: |
| self.assertIn("No sstables to relevel for keyspace1.standard1", e.stdout) |
| self.assertEqual(e.exit_status, 1, msg=str(e.exit_status)) |
| |
| # test by flushing (sstable should be level 0) |
| cluster.start(wait_for_binary_proto=True) |
| session = self.patient_cql_connection(node1) |
| debug("Altering compaction strategy to LCS") |
| session.execute("ALTER TABLE keyspace1.standard1 with compaction={'class': 'LeveledCompactionStrategy', 'sstable_size_in_mb':1};") |
| |
| node1.stress(['write', 'n=1K', 'no-warmup', |
| '-schema', 'replication(factor=1)', |
| '-col', 'n=FIXED(10)', 'SIZE=FIXED(1024)', |
| '-rate', 'threads=8']) |
| |
| node1.flush() |
| cluster.stop() |
| |
| output, _, rc = node1.run_sstableofflinerelevel("keyspace1", "standard1") |
| self.assertIn("L0=1", output) |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| cluster.start(wait_for_binary_proto=True) |
| # test by loading large amount data so we have multiple sstables |
| # must write enough to create more than just L1 sstables |
| keys = 8 * cluster.data_dir_count |
| node1.stress(['write', 'n={0}K'.format(keys), 'no-warmup', |
| '-schema', 'replication(factor=1)', |
| '-col', 'n=FIXED(10)', 'SIZE=FIXED(1200)', |
| '-rate', 'threads=8']) |
| |
| node1.flush() |
| debug("Waiting for compactions to finish") |
| self.wait_for_compactions(node1) |
| debug("Stopping node") |
| cluster.stop() |
| debug("Done stopping node") |
| |
| # Let's reset all sstables to L0 |
| debug("Getting initial levels") |
| initial_levels = list(self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"]))) |
| self.assertNotEqual([], initial_levels) |
| debug('initial_levels:') |
| debug(initial_levels) |
| debug("Running sstablelevelreset") |
| node1.run_sstablelevelreset("keyspace1", "standard1") |
| debug("Getting final levels") |
| final_levels = list(self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"]))) |
| self.assertNotEqual([], final_levels) |
| debug('final levels:') |
| debug(final_levels) |
| |
| # let's make sure there was at least 3 levels (L0, L1 and L2) |
| self.assertGreater(max(initial_levels), 1) |
| # let's check all sstables are on L0 after sstablelevelreset |
| self.assertEqual(max(final_levels), 0) |
| |
| # time to relevel sstables |
| debug("Getting initial levels") |
| initial_levels = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"])) |
| debug("Running sstableofflinerelevel") |
| output, error, _ = node1.run_sstableofflinerelevel("keyspace1", "standard1") |
| debug("Getting final levels") |
| final_levels = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"])) |
| |
| debug(output) |
| debug(error) |
| |
| debug(initial_levels) |
| debug(final_levels) |
| |
| # let's check sstables were promoted after releveling |
| self.assertGreater(max(final_levels), 1) |
| |
| @since('2.2') |
| def sstableverify_test(self): |
| """ |
| Generate sstables and test offline verification works correctly |
| Test on bad input: nonexistent keyspace and sstables |
| Test on potential situations: deleted sstables, corrupted sstables |
| """ |
| |
| cluster = self.cluster |
| cluster.populate(3).start(wait_for_binary_proto=True) |
| node1, node2, node3 = cluster.nodelist() |
| |
| # test on nonexistent keyspace |
| try: |
| (out, err, rc) = node1.run_sstableverify("keyspace1", "standard1") |
| except ToolError as e: |
| self.assertIn("Unknown keyspace/table keyspace1.standard1", e.message) |
| self.assertEqual(e.exit_status, 1, msg=str(e.exit_status)) |
| |
| # test on nonexistent sstables: |
| node1.stress(['write', 'n=100', 'no-warmup', '-schema', 'replication(factor=3)', |
| '-rate', 'threads=8']) |
| (out, err, rc) = node1.run_sstableverify("keyspace1", "standard1") |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| # Generate multiple sstables and test works properly in the simple case |
| node1.stress(['write', 'n=100K', 'no-warmup', '-schema', 'replication(factor=3)', |
| '-rate', 'threads=8']) |
| node1.flush() |
| node1.stress(['write', 'n=100K', 'no-warmup', '-schema', 'replication(factor=3)', |
| '-rate', 'threads=8']) |
| node1.flush() |
| cluster.stop() |
| |
| (out, error, rc) = node1.run_sstableverify("keyspace1", "standard1") |
| |
| self.assertEqual(rc, 0, msg=str(rc)) |
| |
| # STDOUT of the sstableverify command consists of multiple lines which may contain |
| # Java-normalized paths. To later compare these with Python-normalized paths, we |
| # map over each line of out and replace Java-normalized paths with Python equivalents. |
| outlines = map(lambda line: re.sub("(?<=path=').*(?=')", |
| lambda match: os.path.normcase(match.group(0)), |
| line), |
| out.splitlines()) |
| |
| # check output is correct for each sstable |
| sstables = self._get_final_sstables(node1, "keyspace1", "standard1") |
| |
| for sstable in sstables: |
| verified = False |
| hashcomputed = False |
| for line in outlines: |
| if sstable in line: |
| if "Verifying BigTableReader" in line: |
| verified = True |
| elif "Checking computed hash of BigTableReader" in line: |
| hashcomputed = True |
| else: |
| debug(line) |
| |
| debug(verified) |
| debug(hashcomputed) |
| debug(sstable) |
| self.assertTrue(verified and hashcomputed) |
| |
| # now try intentionally corrupting an sstable to see if hash computed is different and error recognized |
| sstable1 = sstables[1] |
| with open(sstable1, 'r') as f: |
| sstabledata = bytearray(f.read()) |
| with open(sstable1, 'w') as out: |
| position = random.randrange(0, len(sstabledata)) |
| sstabledata[position] = (sstabledata[position] + 1) % 256 |
| out.write(sstabledata) |
| |
| # use verbose to get some coverage on it |
| try: |
| (out, error, rc) = node1.run_sstableverify("keyspace1", "standard1", options=['-v']) |
| except ToolError as e: |
| # Process sstableverify output to normalize paths in string to Python casing as above |
| error = re.sub("(?<=Corrupted: ).*", lambda match: os.path.normcase(match.group(0)), e.message) |
| |
| self.assertIn("Corrupted: " + sstable1, error) |
| self.assertEqual(e.exit_status, 1, msg=str(e.exit_status)) |
| |
| def sstableexpiredblockers_test(self): |
| cluster = self.cluster |
| cluster.populate(1).start(wait_for_binary_proto=True) |
| [node1] = cluster.nodelist() |
| session = self.patient_cql_connection(node1) |
| create_ks(session, 'ks', 1) |
| session.execute("create table ks.cf (key int PRIMARY KEY, val int) with gc_grace_seconds=0") |
| # create a blocker: |
| session.execute("insert into ks.cf (key, val) values (1,1)") |
| node1.flush() |
| session.execute("delete from ks.cf where key = 2") |
| node1.flush() |
| session.execute("delete from ks.cf where key = 3") |
| node1.flush() |
| out, error, _ = node1.run_sstableexpiredblockers(keyspace="ks", column_family="cf") |
| self.assertIn("blocks 2 expired sstables from getting dropped", out) |
| |
| # 4.0 removes back compatibility with pre-3.0 versions, so testing upgradesstables for |
| # paths from those versions to 4.0 is invalid (and can only fail). There isn't currently |
| # any difference between the 3.0 and 4.0 sstable format though, but when the version is |
| # bumped for 4.0, remove the max_version & add a case for testing a 3.0 -> 4.0 upgrade |
| @since('2.2', max_version='3.X') |
| def sstableupgrade_test(self): |
| """ |
| Test that sstableupgrade functions properly offline on a same-version Cassandra sstable, a |
| stdout message of "Found 0 sstables that need upgrading." should be returned. |
| """ |
| # Set up original node version to test for upgrade |
| cluster = self.cluster |
| testversion = cluster.version() |
| original_install_dir = cluster.get_install_dir() |
| debug('Original install dir: {}'.format(original_install_dir)) |
| |
| # Set up last major version to upgrade from, assuming 2.1 branch is the oldest tested version |
| if testversion < '2.2': |
| # Upgrading from 2.0->2.1 fails due to the jamm 0.2.5->0.3.0 jar update. |
| # ** This will happen again next time jamm version is upgraded. |
| # CCM doesn't handle this upgrade correctly and results in an error when flushing 2.1: |
| # Error opening zip file or JAR manifest missing : /home/mshuler/git/cassandra/lib/jamm-0.2.5.jar |
| # The 2.1 installed jamm version is 0.3.0, but bin/cassandra.in.sh used by nodetool still has 0.2.5 |
| # (when this is fixed in CCM issue #463, install version='github:apache/cassandra-2.0' as below) |
| self.skipTest('Skipping 2.1 test due to jamm.jar version upgrade problem in CCM node configuration.') |
| elif testversion < '3.0': |
| debug('Test version: {} - installing github:apache/cassandra-2.1'.format(testversion)) |
| cluster.set_install_dir(version='github:apache/cassandra-2.1') |
| # As of 3.5, sstable format 'ma' from 3.0 is still the latest - install 2.2 to upgrade from |
| elif testversion < '4.0': |
| debug('Test version: {} - installing github:apache/cassandra-2.2'.format(testversion)) |
| cluster.set_install_dir(version='github:apache/cassandra-2.2') |
| # From 4.0, one can only upgrade from 3.0 |
| else: |
| debug('Test version: {} - installing github:apache/cassandra-3.0'.format(testversion)) |
| cluster.set_install_dir(version='github:apache/cassandra-3.0') |
| |
| # Start up last major version, write out an sstable to upgrade, and stop node |
| cluster.populate(1).start(wait_for_binary_proto=True) |
| [node1] = cluster.nodelist() |
| # Check that node1 is actually what we expect |
| debug('Downgraded install dir: {}'.format(node1.get_install_dir())) |
| session = self.patient_cql_connection(node1) |
| create_ks(session, 'ks', 1) |
| session.execute('create table ks.cf (key int PRIMARY KEY, val int) with gc_grace_seconds=0') |
| session.execute('insert into ks.cf (key, val) values (1,1)') |
| node1.flush() |
| cluster.stop() |
| debug('Beginning ks.cf sstable: {}'.format(node1.get_sstables(keyspace='ks', column_family='cf'))) |
| |
| # Upgrade Cassandra to original testversion and run sstableupgrade |
| cluster.set_install_dir(original_install_dir) |
| # Check that node1 is actually upgraded |
| debug('Upgraded to original install dir: {}'.format(node1.get_install_dir())) |
| # Perform a node start/stop so system tables get internally updated, otherwise we may get "Unknown keyspace/table ks.cf" |
| cluster.start(wait_for_binary_proto=True) |
| node1.flush() |
| cluster.stop() |
| |
| # A bit hacky, but we can only upgrade to 4.0 from 3.0, but both use the |
| # same sstable major format currently, so there is no upgrading to do. |
| # So on 4.0, we only test that sstable upgrade detect there is no |
| # upgrade. We'll removed that test if 4.0 introduce a major sstable |
| # change before it's release. |
| if testversion < '4.0': |
| (out, error, rc) = node1.run_sstableupgrade(keyspace='ks', column_family='cf') |
| debug(out) |
| debug(error) |
| debug('Upgraded ks.cf sstable: {}'.format(node1.get_sstables(keyspace='ks', column_family='cf'))) |
| self.assertIn('Found 1 sstables that need upgrading.', out) |
| |
| # Check that sstableupgrade finds no upgrade needed on current version. |
| (out, error, rc) = node1.run_sstableupgrade(keyspace='ks', column_family='cf') |
| debug(out) |
| debug(error) |
| self.assertIn('Found 0 sstables that need upgrading.', out) |
| |
| @since('3.0') |
| def sstabledump_test(self): |
| """ |
| Test that sstabledump functions properly offline to output the contents of a table. |
| """ |
| cluster = self.cluster |
| # disable JBOD conf since the test expects exactly one SSTable to be written. |
| cluster.set_datadir_count(1) |
| cluster.populate(1).start(wait_for_binary_proto=True) |
| [node1] = cluster.nodelist() |
| session = self.patient_cql_connection(node1) |
| create_ks(session, 'ks', 1) |
| session.execute('create table ks.cf (key int PRIMARY KEY, val int) with gc_grace_seconds=0') |
| session.execute('insert into ks.cf (key, val) values (1,1)') |
| |
| # delete a partition and then insert a row to test CASSANDRA-13177 |
| session.execute('DELETE FROM ks.cf WHERE key = 2') |
| session.execute('INSERT INTO ks.cf (key, val) VALUES (2, 2)') |
| |
| node1.flush() |
| cluster.stop() |
| [(out, error, rc)] = node1.run_sstabledump(keyspace='ks', column_families=['cf']) |
| debug(out) |
| debug(error) |
| |
| # Load the json output and check that it contains the inserted key=1 |
| s = json.loads(out) |
| debug(s) |
| self.assertEqual(len(s), 2) |
| |
| # order the rows so that we have key=1 first, then key=2 |
| row0, row1 = s |
| (row0, row1) = (row0, row1) if row0['partition']['key'] == ['1'] else (row1, row0) |
| |
| self.assertEqual(row0['partition']['key'], ['1']) |
| |
| self.assertEqual(row1['partition']['key'], ['2']) |
| self.assertIsNotNone(row1['partition'].get('deletion_info')) |
| self.assertIsNotNone(row1.get('rows')) |
| |
| # Check that we only get the key back using the enumerate option |
| [(out, error, rc)] = node1.run_sstabledump(keyspace='ks', column_families=['cf'], enumerate_keys=True) |
| debug(out) |
| debug(error) |
| s = json.loads(out) |
| debug(s) |
| self.assertEqual(len(s), 2) |
| dumped_keys = set(row[0] for row in s) |
| self.assertEqual(set(['1', '2']), dumped_keys) |
| |
| def _check_stderr_error(self, error): |
| acceptable = ["Max sstable size of", "Consider adding more capacity", "JNA link failure", "Class JavaLaunchHelper is implemented in both"] |
| |
| if len(error) > 0: |
| for line in error.splitlines(): |
| self.assertTrue(any([msg in line for msg in acceptable]), |
| 'Found line \n\n"{line}"\n\n in error\n\n{error}'.format(line=line, error=error)) |
| |
| def _get_final_sstables(self, node, ks, table): |
| """ |
| Return the node final sstable data files, excluding the temporary tables. |
| If sstableutil exists (>= 3.0) then we rely on this tool since the table |
| file names no longer contain tmp in their names (CASSANDRA-7066). |
| """ |
| # Get all sstable data files |
| allsstables = map(os.path.normcase, node.get_sstables(ks, table)) |
| |
| # Remove any temporary files |
| tool_bin = node.get_tool('sstableutil') |
| if os.path.isfile(tool_bin): |
| args = [tool_bin, '--type', 'tmp', ks, table] |
| env = common.make_cassandra_env(node.get_install_cassandra_root(), node.get_node_cassandra_root()) |
| p = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE) |
| (stdout, stderr) = p.communicate() |
| tmpsstables = map(os.path.normcase, stdout.splitlines()) |
| |
| ret = list(set(allsstables) - set(tmpsstables)) |
| else: |
| ret = [sstable for sstable in allsstables if "tmp" not in sstable[50:]] |
| |
| return ret |
