scrub_test.py

import glob
import os
import re
import subprocess
import time
import uuid
import pytest
import parse
import logging

from ccmlib import common

from tools.misc import ImmutableMapping
from dtest_setup_overrides import DTestSetupOverrides
from dtest import Tester, create_ks, create_cf
from tools.assertions import assert_length_equal, assert_stderr_clean

since = pytest.mark.since
logger = logging.getLogger(__name__)

KEYSPACE = 'ks'


class TestHelper(Tester):

    @pytest.fixture(scope='function', autouse=True)
    def fixture_set_cluster_settings(self, fixture_dtest_setup):
        """
        disable JBOD configuration for scrub tests.
        range-aware JBOD can skip generation in SSTable,
        and some tests rely on generation numbers/
        (see CASSANDRA-11693 and increase_sstable_generations)
        """
        fixture_dtest_setup.cluster.set_datadir_count(1)


    def get_table_paths(self, table):
        """
        Return the path where the table sstables are located
        """
        node1 = self.cluster.nodelist()[0]
        paths = []
        for data_dir in node1.data_directories():
            basepath = os.path.join(data_dir, KEYSPACE)
            for x in os.listdir(basepath):
                if x.startswith(table):
                    paths.append(os.path.join(basepath, x))
                    break
        return paths

    def get_index_paths(self, table, index):
        """
        Return the paths where the index sstables are located
        """
        paths = self.get_table_paths(table)
        index_paths = []
        for path in paths:
            index_paths.append(os.path.join(path, '.' + index))
        return paths

    def get_sstable_files(self, paths):
        """
        Return the sstable files at a specific location
        """
        ret = []
        logger.debug('Checking sstables in {}'.format(paths))

        for ext in ('*.db', '*.txt', '*.adler32', '*.sha1'):
            for path in paths:
                for fname in glob.glob(os.path.join(path, ext)):
                    bname = os.path.basename(fname)
                    logger.debug('Found sstable file {}'.format(bname))
                    ret.append(bname)
        return ret

    def delete_non_essential_sstable_files(self, table):
        """
        Delete all sstable files except for the -Data.db file and the
        -Statistics.db file (only available in >= 3.0)
        """
        for fname in self.get_sstable_files(self.get_table_paths(table)):
            if not fname.endswith("-Data.db") and not fname.endswith("-Statistics.db"):
                paths = self.get_table_paths(table)
                for path in paths:
                    fullname = os.path.join(path, fname)
                    if (os.path.exists(fullname)):
                        logger.debug('Deleting {}'.format(fullname))
                        os.remove(fullname)

    def get_sstables(self, table, indexes):
        """
        Return the sstables for a table and the specified indexes of this table
        """
        sstables = {}
        table_sstables = self.get_sstable_files(self.get_table_paths(table))
        assert len(table_sstables) > 0
        sstables[table] = sorted(table_sstables)

        for index in indexes:
            index_sstables = self.get_sstable_files(self.get_index_paths(table, index))
            assert len(index_sstables) > 0
            sstables[index] = sorted('{}/{}'.format(index, sstable) for sstable in index_sstables)

        return sstables

    def launch_nodetool_cmd(self, cmd):
        """
        Launch a nodetool command and check the result is empty (no error)
        """
        node1 = self.cluster.nodelist()[0]
        response = node1.nodetool(cmd).stdout
        if not common.is_win():  # nodetool always prints out on windows
            assert_length_equal(response, 0)  # nodetool does not print anything unless there is an error

    def launch_standalone_scrub(self, ks, cf, reinsert_overflowed_ttl=False, no_validate=False):
        """
        Launch the standalone scrub
        """
        node1 = self.cluster.nodelist()[0]
        env = common.make_cassandra_env(node1.get_install_cassandra_root(), node1.get_node_cassandra_root())
        scrub_bin = node1.get_tool('sstablescrub')
        logger.debug(scrub_bin)

        args = [scrub_bin]
        if reinsert_overflowed_ttl:
            args += ['--reinsert-overflowed-ttl']
        if no_validate:
            args += ['--no-validate']
        args += [ks, cf] if reinsert_overflowed_ttl else [ks, cf]
        p = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        out, err = p.communicate()
        logger.debug(out.decode("utf-8"))
        # if we have less than 64G free space, we get this warning - ignore it
        if err and "Consider adding more capacity" not in err.decode("utf-8"):
            logger.debug(err.decode("utf-8"))
            assert_stderr_clean(err.decode("utf-8"))

    def perform_node_tool_cmd(self, cmd, table, indexes):
        """
        Perform a nodetool command on a table and the indexes specified
        """
        self.launch_nodetool_cmd('{} {} {}'.format(cmd, KEYSPACE, table))
        for index in indexes:
            self.launch_nodetool_cmd('{} {} {}.{}'.format(cmd, KEYSPACE, table, index))

    def flush(self, table, *indexes):
        """
        Flush table and indexes via nodetool, and then return all sstables
        in a dict keyed by the table or index name.
        """
        self.perform_node_tool_cmd('flush', table, indexes)
        return self.get_sstables(table, indexes)

    def scrub(self, table, *indexes):
        """
        Scrub table and indexes via nodetool, and then return all sstables
        in a dict keyed by the table or index name.
        """
        self.perform_node_tool_cmd('scrub', table, indexes)
        time.sleep(.1)
        return self.get_sstables(table, indexes)

    def standalonescrub(self, table, *indexes):
        """
        Launch standalone scrub on table and indexes, and then return all sstables
        in a dict keyed by the table or index name.
        """
        self.launch_standalone_scrub(KEYSPACE, table)
        for index in indexes:
            self.launch_standalone_scrub(KEYSPACE, '{}.{}'.format(table, index))
        return self.get_sstables(table, indexes)

    def increment_generation_by(self, sstable, generation_increment):
        """
        Set the generation number for an sstable file name
        """
        return re.sub('(\d(?!\d))\-', lambda x: str(int(x.group(1)) + generation_increment) + '-', sstable)

    def increase_sstable_generations(self, sstables):
        """
        After finding the number of existing sstables, increase all of the
        generations by that amount.
        """
        for table_or_index, table_sstables in list(sstables.items()):
            increment_by = len(set(parse.search('{}-{increment_by}-{suffix}.{file_extention}', s).named['increment_by'] for s in table_sstables))
            sstables[table_or_index] = [self.increment_generation_by(s, increment_by) for s in table_sstables]

        logger.debug('sstables after increment {}'.format(str(sstables)))

    def uuid_sstable_identifiers_enabled(self, session):
        return 'true' == session.execute("select value from system_views.settings where name = 'uuid_sstable_identifiers_enabled'")[0][0]


@since('2.2')
class TestScrubIndexes(TestHelper):
    """
    Test that we scrub indexes as well as their parent tables
    Only valid for legacy secondary indexes
    """

    @pytest.fixture(scope='function', autouse=True)
    def fixture_dtest_setup_overrides(self, dtest_config):
        dtest_setup_overrides = DTestSetupOverrides()

        if dtest_config.cassandra_version_from_build >= '5.0':
            dtest_setup_overrides.cluster_options = ImmutableMapping({'default_secondary_index': 'legacy_local_table'})
        return dtest_setup_overrides

    def create_users(self, session):
        columns = {"password": "varchar", "gender": "varchar", "session_token": "varchar", "state": "varchar", "birth_year": "bigint"}
        create_cf(session, 'users', columns=columns)

        session.execute("CREATE INDEX gender_idx ON users (gender)")
        session.execute("CREATE INDEX state_idx ON users (state)")
        session.execute("CREATE INDEX birth_year_idx ON users (birth_year)")

    def update_users(self, session):
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user1', 'ch@ngem3a', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user2', 'ch@ngem3b', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user3', 'ch@ngem3c', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user4', 'ch@ngem3d', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user5', 'ch@ngem3e', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user6', 'ch@ngem3f', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user7', 'ch@ngem3g', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user8', 'ch@ngem3h', 'm', 'CA', 1982)")

        session.execute("DELETE FROM users where KEY = 'user1'")
        session.execute("DELETE FROM users where KEY = 'user5'")
        session.execute("DELETE FROM users where KEY = 'user7'")

    def query_users(self, session):
        ret = list(session.execute("SELECT * FROM users"))
        ret.extend(list(session.execute("SELECT * FROM users WHERE state='TX'")))
        ret.extend(list(session.execute("SELECT * FROM users WHERE gender='f'")))
        ret.extend(list(session.execute("SELECT * FROM users WHERE birth_year=1978")))
        assert_length_equal(ret, 8)
        return ret

    def test_scrub_static_table(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        self.create_users(session)
        self.update_users(session)

        initial_users = self.query_users(session)
        initial_sstables = self.flush('users', 'gender_idx', 'state_idx', 'birth_year_idx')
        scrubbed_sstables = self.scrub('users', 'gender_idx', 'state_idx', 'birth_year_idx')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        users = self.query_users(session)
        assert initial_users == users

        # Scrub and check sstables and data again
        scrubbed_sstables = self.scrub('users', 'gender_idx', 'state_idx', 'birth_year_idx')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        users = self.query_users(session)
        assert initial_users == users

        # Restart and check data again
        cluster.stop()
        cluster.start()

        session = self.patient_cql_connection(node1)
        session.execute('USE {}'.format(KEYSPACE))

        users = self.query_users(session)
        assert initial_users == users

    def test_standalone_scrub(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        self.create_users(session)
        self.update_users(session)

        initial_users = self.query_users(session)
        initial_sstables = self.flush('users', 'gender_idx', 'state_idx', 'birth_year_idx')

        cluster.stop()

        scrubbed_sstables = self.standalonescrub('users', 'gender_idx', 'state_idx', 'birth_year_idx')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        cluster.start()
        session = self.patient_cql_connection(node1)
        session.execute('USE {}'.format(KEYSPACE))

        users = self.query_users(session)
        assert initial_users == users

    def test_scrub_collections_table(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        session.execute("CREATE TABLE users (user_id uuid PRIMARY KEY, email text, uuids list<uuid>)")
        session.execute("CREATE INDEX user_uuids_idx on users (uuids)")

        _id = uuid.uuid4()
        num_users = 100
        for i in range(0, num_users):
            user_uuid = uuid.uuid4()
            session.execute(("INSERT INTO users (user_id, email) values ({user_id}, 'test@example.com')").format(user_id=user_uuid))
            session.execute(("UPDATE users set uuids = [{id}] where user_id = {user_id}").format(id=_id, user_id=user_uuid))

        initial_users = list(session.execute(("SELECT * from users where uuids contains {some_uuid}").format(some_uuid=_id)))
        assert num_users == len(initial_users)

        initial_sstables = self.flush('users', 'user_uuids_idx')
        scrubbed_sstables = self.scrub('users', 'user_uuids_idx')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        users = list(session.execute(("SELECT * from users where uuids contains {some_uuid}").format(some_uuid=_id)))
        assert initial_users == users

        scrubbed_sstables = self.scrub('users', 'user_uuids_idx')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        users = list(session.execute(("SELECT * from users where uuids contains {some_uuid}").format(some_uuid=_id)))

        assert initial_users == users


class TestScrub(TestHelper):
    """
    Generic tests for scrubbing
    """

    def create_users(self, session):
        columns = {"password": "varchar", "gender": "varchar", "session_token": "varchar", "state": "varchar", "birth_year": "bigint"}
        create_cf(session, 'users', columns=columns)

    def update_users(self, session):
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user1', 'ch@ngem3a', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user2', 'ch@ngem3b', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user3', 'ch@ngem3c', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user4', 'ch@ngem3d', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user5', 'ch@ngem3e', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user6', 'ch@ngem3f', 'm', 'CA', 1982)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user7', 'ch@ngem3g', 'f', 'TX', 1978)")
        session.execute("INSERT INTO users (KEY, password, gender, state, birth_year) VALUES ('user8', 'ch@ngem3h', 'm', 'CA', 1982)")

        session.execute("DELETE FROM users where KEY = 'user1'")
        session.execute("DELETE FROM users where KEY = 'user5'")
        session.execute("DELETE FROM users where KEY = 'user7'")

    def query_users(self, session):
        ret = list(session.execute("SELECT * FROM users"))
        assert_length_equal(ret, 5)
        return ret

    def test_nodetool_scrub(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]
        # we don't want automatic minor compaction because we want to block on
        # compactions
        node1.nodetool('disableautocompaction')

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        self.create_users(session)
        self.update_users(session)

        initial_users = self.query_users(session)
        initial_sstables = self.flush('users')
        scrubbed_sstables = self.scrub('users')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        users = self.query_users(session)
        assert initial_users == users

        # Scrub and check sstables and data again
        scrubbed_sstables = self.scrub('users')

        if should_assert_sstables:
           self.increase_sstable_generations(initial_sstables)
           assert initial_sstables == scrubbed_sstables

        users = self.query_users(session)
        assert initial_users == users

        # Restart and check data again
        cluster.stop()
        cluster.start()

        session = self.patient_cql_connection(node1)
        session.execute('USE {}'.format(KEYSPACE))

        users = self.query_users(session)
        assert initial_users == users

    def test_standalone_scrub(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        self.create_users(session)
        self.update_users(session)

        initial_users = self.query_users(session)
        initial_sstables = self.flush('users')

        cluster.stop()

        scrubbed_sstables = self.standalonescrub('users')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        cluster.start()
        session = self.patient_cql_connection(node1)
        session.execute('USE {}'.format(KEYSPACE))

        users = self.query_users(session)
        assert initial_users == users

    def test_standalone_scrub_essential_files_only(self):
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        should_assert_sstables = cluster.version() < '5.0' or (cluster.version() >= '5.0' and not self.uuid_sstable_identifiers_enabled(session))
        create_ks(session, KEYSPACE, 1)

        self.create_users(session)
        self.update_users(session)

        initial_users = self.query_users(session)
        initial_sstables = self.flush('users')

        cluster.stop()

        self.delete_non_essential_sstable_files('users')

        scrubbed_sstables = self.standalonescrub('users')

        if should_assert_sstables:
            self.increase_sstable_generations(initial_sstables)
            assert initial_sstables == scrubbed_sstables

        cluster.start()
        session = self.patient_cql_connection(node1)
        session.execute('USE {}'.format(KEYSPACE))

        users = self.query_users(session)
        assert initial_users == users

    def test_scrub_with_UDT(self):
        """
        @jira_ticket CASSANDRA-7665
        """
        cluster = self.cluster
        cluster.populate(1).start()
        node1 = cluster.nodelist()[0]

        session = self.patient_cql_connection(node1)
        session.execute("CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 };")
        session.execute("use test;")
        session.execute("CREATE TYPE point_t (x double, y double);")

        node1.nodetool("scrub")
        time.sleep(2)
        match = node1.grep_log("org.apache.cassandra.serializers.MarshalException: Not enough bytes to read a set")
        assert len(match) == 0