<commit_before>#!/usr/bin/env python

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2020 ScyllaDB

import shutil
import sys
import os
import time
import random
from enum import Enum
from textwrap import dedent
from typing import Optional, Tuple

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement  # pylint: disable=no-name-in-module

from sdcm import cluster
from sdcm.tester import ClusterTester
from sdcm.gemini_thread import GeminiStressThread


class Mode(Enum):
    DELTA = 1
    PREIMAGE = 2
    POSTIMAGE = 3


def mode_str(mode: Mode) -> str:
    return {
        Mode.DELTA: 'delta',
        Mode.PREIMAGE: 'preimage',
        Mode.POSTIMAGE: 'postimage'
    }[mode]


def print_file_to_stdout(path: str) -> None:
    with open(path, "r") as file:
        shutil.copyfileobj(file, sys.stdout)


def write_cql_result(res, path: str):
    """Write a CQL select result to a file.

    :param res: cql results
    :type res: ResultSet
    :param path: path to file
    :type path: str
    """
    with open(path, 'w') as file:
        for row in res:
            file.write(str(row) + '\n')
        file.flush()
        os.fsync(file.fileno())


SCYLLA_MIGRATE_URL = "https://kbr-scylla.s3-eu-west-1.amazonaws.com/scylla-migrate"
REPLICATOR_URL = "https://kbr-scylla.s3-eu-west-1.amazonaws.com/scylla-cdc-replicator-0.0.1-SNAPSHOT-jar-with-dependencies.jar"


class CDCReplicationTest(ClusterTester):
    KS_NAME = 'ks1'
    TABLE_NAME = 'table1'

    def collect_data_for_analysis(self,
                                  migrate_log_path: Optional[str], replicator_log_path: str,
                                  master_node: cluster.BaseNode, replica_node: cluster.BaseNode,
                                  fetch_tables: bool = True) -> None:
        if migrate_log_path:
            self.log.info('scylla-migrate log:')
            print_file_to_stdout(migrate_log_path)
        self.log.info('Replicator log:')
        print_file_to_stdout(replicator_log_path)

        if fetch_tables:
            with self.db_cluster.cql_connection_patient(node=master_node) as sess:
                self.log.info('Fetching master table...')
                res = sess.execute(SimpleStatement(f'select * from {self.KS_NAME}.{self.TABLE_NAME}',
                                                   consistency_level=ConsistencyLevel.QUORUM, fetch_size=1000))
                write_cql_result(res, os.path.join(self.logdir, 'master-table'))

            with self.cs_db_cluster.cql_connection_patient(node=replica_node) as sess:
                self.log.info('Fetching replica table...')
                res = sess.execute(SimpleStatement(f'select * from {self.KS_NAME}.{self.TABLE_NAME}',
                                                   consistency_level=ConsistencyLevel.QUORUM, fetch_size=1000))
                write_cql_result(res, os.path.join(self.logdir, 'replica-table'))

    def test_replication_cs(self) -> None:
        self.log.info('Using cassandra-stress to generate workload.')
        self.test_replication(False, Mode.DELTA)

    def test_replication_gemini(self, mode: Mode) -> None:
        self.log.info('Using gemini to generate workload. Mode: {}'.format(mode.name))
        self.test_replication(True, mode)

    def test_replication_gemini_delta(self) -> None:
        self.test_replication_gemini(Mode.DELTA)

    def test_replication_gemini_preimage(self) -> None:
        self.test_replication_gemini(Mode.PREIMAGE)

    def test_replication_gemini_postimage(self) -> None:
        self.test_replication_gemini(Mode.POSTIMAGE)

    # In this test we run a sequence of ~30 minute rounds of replication;
    # after each round we stop generating workload and check consistency.
    def test_replication_longevity(self) -> None:
        loader_node = self.loaders.nodes[0]
        self.setup_tools(loader_node)

        self.log.info('Waiting for the latest CDC generation to start...')
        # 2 * ring_delay (ring_delay = 30s) + leeway
        time.sleep(70)

        # We'll use the same seed for each round, so that gemini uses the same schema each time.
        # This way we preserve tables from the previous round.
        # The purpose of preserving tables is to have the cluster store more data, which
        # puts strain on the cluster by causing more compactions, for example.
        gemini_seed = random.randint(1, 1000)

        self.log.info('Starting gemini.')
        stress_thread = self.start_gemini(gemini_seed)

        # Wait for gemini to create keyspaces/tables/UTs
        self.log.info('Let gemini run for a while...')
        time.sleep(20)

        self.copy_master_schema_to_replica()

        self.start_replicator(Mode.DELTA)

        self.consistency_ok = True

        no_rounds = 12  # 12 rounds, ~30 minutes each -> ~6 hours
        for rnd in range(no_rounds):
            self.log.info('Starting round {}'.format(rnd))

            self.log.info('Starting nemesis')
            self.db_cluster.add_nemesis(nemesis=self.get_nemesis_class(), tester_obj=self)
            self.db_cluster.start_nemesis()

            self.log.info('Waiting for workload generation to finish (~30 minutes)...')
            stress_results = self.verify_gemini_results(queue=stress_thread)
            self.log.info('gemini results: {}'.format(stress_results))

            self.log.info('Waiting for replicator to finish (sleeping 60s)...')
            time.sleep(60)

            self.log.info('Stopping nemesis...')
            self.db_cluster.stop_nemesis(timeout=1800)
            self.log.info('Nemesis stopped.')

            self.log.info('Fetching replicator logs...')
            replicator_log_path = os.path.join(self.logdir, 'replicator.log')
            loader_node.remoter.receive_files(src='replicatorlog', dst=replicator_log_path)

            migrate_log_path = os.path.join(self.logdir, 'scylla-migrate.log')
            (migrate_ok, consistency_ok) = self.check_consistency(migrate_log_path)
            self.consistency_ok = consistency_ok

            if not (self.consistency_ok and migrate_ok):
                break

            if rnd != no_rounds - 1:
                self.log.info('Starting gemini.')
                stress_thread = self.start_gemini(gemini_seed)

        if not self.consistency_ok:
            self.log.error('Inconsistency detected.')

        if self.consistency_ok and migrate_ok:
            self.log.info('Consistency check successful.')
        else:
            # We don't fetch tables in this test since they are way too large.<commit_msg>improvement(cdc_replication_test): truncate base tables after each round<commit_after>#!/usr/bin/env python

# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See LICENSE for more details.
#
# Copyright (c) 2020 ScyllaDB

import shutil
import sys
import os
import time
import random
from enum import Enum
from textwrap import dedent
from typing import Optional, Tuple

from cassandra import ConsistencyLevel
from cassandra.query import SimpleStatement  # pylint: disable=no-name-in-module

from sdcm import cluster
from sdcm.tester import ClusterTester
from sdcm.gemini_thread import GeminiStressThread


class Mode(Enum):
    DELTA = 1
    PREIMAGE = 2
    POSTIMAGE = 3


def mode_str(mode: Mode) -> str:
    return {
        Mode.DELTA: 'delta',
        Mode.PREIMAGE: 'preimage',
        Mode.POSTIMAGE: 'postimage'
    }[mode]


def print_file_to_stdout(path: str) -> None:
    with open(path, "r") as file:
        shutil.copyfileobj(file, sys.stdout)


def write_cql_result(res, path: str):
    """Write a CQL select result to a file.

    :param res: cql results
    :type res: ResultSet
    :param path: path to file
    :type path: str
    """
    with open(path, 'w') as file:
        for row in res:
            file.write(str(row) + '\n')
        file.flush()
        os.fsync(file.fileno())


SCYLLA_MIGRATE_URL = "https://kbr-scylla.s3-eu-west-1.amazonaws.com/scylla-migrate"
REPLICATOR_URL = "https://kbr-scylla.s3-eu-west-1.amazonaws.com/scylla-cdc-replicator-0.0.1-SNAPSHOT-jar-with-dependencies.jar"


class CDCReplicationTest(ClusterTester):
    KS_NAME = 'ks1'
    TABLE_NAME = 'table1'

    def collect_data_for_analysis(self,
                                  migrate_log_path: Optional[str], replicator_log_path: str,
                                  master_node: cluster.BaseNode, replica_node: cluster.BaseNode,
                                  fetch_tables: bool = True) -> None:
        if migrate_log_path:
            self.log.info('scylla-migrate log:')
            print_file_to_stdout(migrate_log_path)
        self.log.info('Replicator log:')
        print_file_to_stdout(replicator_log_path)

        if fetch_tables:
            with self.db_cluster.cql_connection_patient(node=master_node) as sess:
                self.log.info('Fetching master table...')
                res = sess.execute(SimpleStatement(f'select * from {self.KS_NAME}.{self.TABLE_NAME}',
                                                   consistency_level=ConsistencyLevel.QUORUM, fetch_size=1000))
                write_cql_result(res, os.path.join(self.logdir, 'master-table'))

            with self.cs_db_cluster.cql_connection_patient(node=replica_node) as sess:
                self.log.info('Fetching replica table...')
                res = sess.execute(SimpleStatement(f'select * from {self.KS_NAME}.{self.TABLE_NAME}',
                                                   consistency_level=ConsistencyLevel.QUORUM, fetch_size=1000))
                write_cql_result(res, os.path.join(self.logdir, 'replica-table'))

    def test_replication_cs(self) -> None:
        self.log.info('Using cassandra-stress to generate workload.')
        self.test_replication(False, Mode.DELTA)

    def test_replication_gemini(self, mode: Mode) -> None:
        self.log.info('Using gemini to generate workload. Mode: {}'.format(mode.name))
        self.test_replication(True, mode)

    def test_replication_gemini_delta(self) -> None:
        self.test_replication_gemini(Mode.DELTA)

    def test_replication_gemini_preimage(self) -> None:
        self.test_replication_gemini(Mode.PREIMAGE)

    def test_replication_gemini_postimage(self) -> None:
        self.test_replication_gemini(Mode.POSTIMAGE)

    # In this test we run a sequence of ~30 minute rounds of replication;
    # after each round we stop generating workload and check consistency.
    def test_replication_longevity(self) -> None:
        loader_node = self.loaders.nodes[0]
        self.setup_tools(loader_node)

        self.log.info('Waiting for the latest CDC generation to start...')
        # 2 * ring_delay (ring_delay = 30s) + leeway
        time.sleep(70)

        # We'll use the same seed for each round, so that gemini uses the same schema each time.
        # This way we preserve tables from the previous round.
        # The purpose of preserving tables is to have the cluster store more data, which
        # puts strain on the cluster by causing more compactions, for example.
        gemini_seed = random.randint(1, 1000)

        self.log.info('Starting gemini.')
        stress_thread = self.start_gemini(gemini_seed)

        # Wait for gemini to create keyspaces/tables/UTs
        self.log.info('Let gemini run for a while...')
        time.sleep(20)

        self.copy_master_schema_to_replica()

        self.start_replicator(Mode.DELTA)

        self.consistency_ok = True

        no_rounds = 12  # 12 rounds, ~30 minutes each -> ~6 hours
        for rnd in range(no_rounds):
            self.log.info('Starting round {}'.format(rnd))

            self.log.info('Starting nemesis')
            self.db_cluster.add_nemesis(nemesis=self.get_nemesis_class(), tester_obj=self)
            self.db_cluster.start_nemesis()

            self.log.info('Waiting for workload generation to finish (~30 minutes)...')
            stress_results = self.verify_gemini_results(queue=stress_thread)
            self.log.info('gemini results: {}'.format(stress_results))

            self.log.info('Waiting for replicator to finish (sleeping 60s)...')
            time.sleep(60)

            self.log.info('Stopping nemesis...')
            self.db_cluster.stop_nemesis(timeout=1800)
            self.log.info('Nemesis stopped.')

            self.log.info('Fetching replicator logs...')
            replicator_log_path = os.path.join(self.logdir, 'replicator.log')
            loader_node.remoter.receive_files(src='replicatorlog', dst=replicator_log_path)

            migrate_log_path = os.path.join(self.logdir, 'scylla-migrate.log')
            (migrate_ok, consistency_ok) = self.check_consistency(migrate_log_path)
            self.consistency_ok = consistency_ok

            if not (self.consistency_ok and migrate_ok):
                break

            if rnd != no_rounds - 1:
                self.log.info('Truncating master cluster base table.')
                with self.db_cluster.cql_connection_patient(node=self.db_cluster.nodes[0]) as sess:
                    sess.execute(f"truncate table {self.KS_NAME}.{self.TABLE_NAME}")

                self.log.info('Truncating replica cluster base table.')
                with self.cs_db_cluster.cql_connection_patient(node=self.cs_db_cluster.nodes[0]) as sess:
                    sess.execute(f"truncate table {self.KS_NAME}.{self.TABLE_NAME}")

                self.log.info('Starting gemini.')
                stress_thread = self.start_gemini(gemini_seed)

        if not self.consistency_ok:
            self.log.error('Inconsistency detected.')

        if self.consistency_ok and migrate_ok:
            self.log.info('Consistency check successful.')
        else:
            # We don't fetch tables in this test since they are way too large.
{'fraction_non_alphanumeric': 0.09600383194197346, 'fraction_numerical': 0.008074449158341317, 'mean_word_length': 2.5865030674846627, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3791841', 'n_tokens_mistral': 4607, 'n_tokens_neox': 4348, 'n_words': 1071}