<commit_before>import json
import os
import random
import re
import subprocess

from ccmlib import common

from dtest import Tester, debug
from tools import known_failure, 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"]

    @known_failure(failure_source='test',
                   jira_url='https://issues.apache.org/jira/browse/CASSANDRA-11235',
                   flaky=False,
                   notes='windows')
    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)
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        self.assertIn("ColumnFamily not found: keyspace1/standard1", error)
        # this should return exit code 1
        self.assertEqual(rc, 1, msg=str(rc))

        # 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", output=True)
        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", output=True)
        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"]))
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        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", capture_output=True)
            if pattern.search(output):
                break

    @known_failure(failure_source='test',
                   jira_url='https://issues.apache.org/jira/browse/CASSANDRA-12275',
                   flaky=False,
                   notes='windows')
    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)

        (output, error, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)

        self.assertIn("No sstables to relevel for keyspace1.standard1", output)
        self.assertEqual(rc, 1, msg=str(rc))

        # 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, error, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)
        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(1024)',
                      '-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 = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"]))
        debug("Running sstablelevelreset")
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        debug("Getting final levels")
        final_levels = self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"]))

        # 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, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)
        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<commit_msg>improve error reporting for empty levels<commit_after>import json
import os
import random
import re
import subprocess

from ccmlib import common

from dtest import Tester, debug
from tools import known_failure, 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"]

    @known_failure(failure_source='test',
                   jira_url='https://issues.apache.org/jira/browse/CASSANDRA-11235',
                   flaky=False,
                   notes='windows')
    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)
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        self.assertIn("ColumnFamily not found: keyspace1/standard1", error)
        # this should return exit code 1
        self.assertEqual(rc, 1, msg=str(rc))

        # 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", output=True)
        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", output=True)
        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"]))
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        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", capture_output=True)
            if pattern.search(output):
                break

    @known_failure(failure_source='test',
                   jira_url='https://issues.apache.org/jira/browse/CASSANDRA-12275',
                   flaky=False,
                   notes='windows')
    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)

        (output, error, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)

        self.assertIn("No sstables to relevel for keyspace1.standard1", output)
        self.assertEqual(rc, 1, msg=str(rc))

        # 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, error, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)
        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(1024)',
                      '-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("Running sstablelevelreset")
        (output, error, rc) = node1.run_sstablelevelreset("keyspace1", "standard1", output=True)
        debug("Getting final levels")
        final_levels = list(self.get_levels(node1.run_sstablemetadata(keyspace="keyspace1", column_families=["standard1"])))
        self.assertNotEqual([], 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, rc) = node1.run_sstableofflinerelevel("keyspace1", "standard1", output=True)
        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
{'fraction_non_alphanumeric': 0.11197678792112192, 'fraction_numerical': 0.019235935736929773, 'mean_word_length': 2.847044233154196, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6305267', 'n_tokens_mistral': 5726, 'n_tokens_neox': 5367, 'n_words': 1308}