<commit_before># 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) 2019 ScyllaDB

from __future__ import print_function

import os
import re
import uuid
import time
import logging
import concurrent.futures

from sdcm.loader import CassandraStressExporter
from sdcm.prometheus import nemesis_metrics_obj
from sdcm.cluster import BaseLoaderSet
from sdcm.sct_events import CassandraStressEvent
from sdcm.utils.common import FileFollowerThread, makedirs, generate_random_string, get_profile_content
from sdcm.sct_events import CassandraStressLogEvent, Severity

LOGGER = logging.getLogger(__name__)


def format_stress_cmd_error(exc):
    """
    format nicely the excpetion we get from stress command failures

    :param exc: the exception
    :return: string to add to the event
    """
    if hasattr(exc, 'result') and exc.result.failed:
        stderr = exc.result.stderr
        if len(stderr) > 100:
            stderr = stderr[:100]
        errors_str = f'Stress command completed with bad status {exc.result.exited}: {stderr}'
    else:
        errors_str = f'Stress command execution failed with: {str(exc)}'

    return errors_str


class CassandraStressEventsPublisher(FileFollowerThread):
    def __init__(self, node, cs_log_filename):
        super(CassandraStressEventsPublisher, self).__init__()
        self.cs_log_filename = cs_log_filename
        self.node = str(node)
        self.cs_events = [CassandraStressLogEvent(type='IOException', regex=r'java\.io\.IOException', severity=Severity.ERROR),
                          CassandraStressLogEvent(type='ConsistencyError', regex=r'Cannot achieve consistency level', severity=Severity.ERROR)]

    def run(self):
        patterns = [(event, re.compile(event.regex)) for event in self.cs_events]

        while not self.stopped():
            exists = os.path.isfile(self.cs_log_filename)
            if not exists:
                time.sleep(0.5)
                continue

            for line_number, line in enumerate(self.follow_file(self.cs_log_filename)):
                if self.stopped():
                    break

                for event, pattern in patterns:
                    match = pattern.search(line)
                    if match:
                        event.add_info_and_publish(node=self.node, line=line, line_number=line_number)


class CassandraStressThread():  # pylint: disable=too-many-instance-attributes
    def __init__(self, loader_set, stress_cmd, timeout, stress_num=1, keyspace_num=1, keyspace_name='',  # pylint: disable=too-many-arguments
                 profile=None, node_list=None, round_robin=False, client_encrypt=False):
        if not node_list:
            node_list = []
        self.loader_set = loader_set
        self.stress_cmd = stress_cmd
        self.timeout = timeout
        self.stress_num = stress_num
        self.keyspace_num = keyspace_num
        self.keyspace_name = keyspace_name
        self.profile = profile
        self.node_list = node_list
        self.round_robin = round_robin
        self.client_encrypt = client_encrypt

        self.executor = None
        self.results_futures = []
        self.shell_marker = generate_random_string(20)
        #  This marker is used to mark shell commands, in order to be able to kill them later
        self.max_workers = 0

    def create_stress_cmd(self, node, loader_idx, keyspace_idx):
        stress_cmd = self.stress_cmd
        if node.cassandra_stress_version == "unknown":  # Prior to 3.11, cassandra-stress didn't have version argument
            stress_cmd = stress_cmd.replace("throttle", "limit")  # after 3.11 limit was renamed to throttle

        # When using cassandra-stress with "user profile" the profile yaml should be provided
        if 'profile' in stress_cmd and not self.profile:
            # support of using -profile in sct test-case yaml, assumes they exists data_dir
            # TODO: move those profile to their own directory
            cs_profile, profile = get_profile_content(stress_cmd)
            keyspace_name = profile['keyspace']
            self.profile = cs_profile
            self.keyspace_name = keyspace_name

        if self.keyspace_name:
            stress_cmd = stress_cmd.replace(" -schema ", " -schema keyspace={} ".format(self.keyspace_name))
        elif 'keyspace=' not in stress_cmd:  # if keyspace is defined in the command respect that
            stress_cmd = stress_cmd.replace(" -schema ", " -schema keyspace=keyspace{} ".format(keyspace_idx))

        credentials = self.loader_set.get_db_auth()
        if credentials and 'user=' not in stress_cmd:
            # put the credentials into the right place into -mode section
            stress_cmd = re.sub(r'(-mode.*?)-', r'\1 user={} password={} -'.format(*credentials), stress_cmd)
        if self.client_encrypt and 'transport' not in stress_cmd:
            stress_cmd += ' -transport "truststore=/etc/scylla/ssl_conf/client/cacerts.jks truststore-password=<PASSWORD>"'

        if self.node_list and '-node' not in stress_cmd:
            first_node = [n for n in self.node_list if n.dc_idx == loader_idx %
                          3]  # make sure each loader is targeting on datacenter/region
            first_node = first_node[0] if first_node else self.node_list[0]
            stress_cmd += " -node {}".format(first_node.ip_address)

        stress_cmd += ' -errors skip-unsupported-columns'

        return stress_cmd

    def _run_stress(self, node, loader_idx, cpu_idx, keyspace_idx):
        stress_cmd = self.create_stress_cmd(node, loader_idx, keyspace_idx)

        if self.profile:
            with open(self.profile) as profile_file:
                LOGGER.info('Profile content:\n%s', profile_file.read())
            node.remoter.send_files(self.profile, os.path.join('/tmp', os.path.basename(self.profile)), delete_dst=True)

        stress_cmd_opt = stress_cmd.split()[1]

        LOGGER.info('Stress command:\n%s', stress_cmd)

        if not os.path.exists(node.logdir):
            makedirs(node.logdir)
        log_file_name = os.path.join(node.logdir,
                                     f'cassandra-stress-l{loader_idx}-c{cpu_idx}-k{keyspace_idx}-{uuid.uuid4()}.log')

        LOGGER.debug('cassandra-stress local log: %s', log_file_name)

        # This tag will be output in the header of c-stress result,
        # we parse it to know the loader & cpu info in _parse_cs_summary().
        tag = f'TAG: loader_idx:{loader_idx}-cpu_idx:{cpu_idx}-keyspace_idx:{keyspace_idx}'<commit_msg>fix(sdcm/stress_thread.py): Fix logic of adding errors option<commit_after># 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) 2019 ScyllaDB

from __future__ import print_function

import os
import re
import uuid
import time
import logging
import concurrent.futures

from sdcm.loader import CassandraStressExporter
from sdcm.prometheus import nemesis_metrics_obj
from sdcm.cluster import BaseLoaderSet
from sdcm.sct_events import CassandraStressEvent
from sdcm.utils.common import FileFollowerThread, makedirs, generate_random_string, get_profile_content
from sdcm.sct_events import CassandraStressLogEvent, Severity

LOGGER = logging.getLogger(__name__)


def format_stress_cmd_error(exc):
    """
    format nicely the excpetion we get from stress command failures

    :param exc: the exception
    :return: string to add to the event
    """
    if hasattr(exc, 'result') and exc.result.failed:
        stderr = exc.result.stderr
        if len(stderr) > 100:
            stderr = stderr[:100]
        errors_str = f'Stress command completed with bad status {exc.result.exited}: {stderr}'
    else:
        errors_str = f'Stress command execution failed with: {str(exc)}'

    return errors_str


class CassandraStressEventsPublisher(FileFollowerThread):
    def __init__(self, node, cs_log_filename):
        super(CassandraStressEventsPublisher, self).__init__()
        self.cs_log_filename = cs_log_filename
        self.node = str(node)
        self.cs_events = [CassandraStressLogEvent(type='IOException', regex=r'java\.io\.IOException', severity=Severity.ERROR),
                          CassandraStressLogEvent(type='ConsistencyError', regex=r'Cannot achieve consistency level', severity=Severity.ERROR)]

    def run(self):
        patterns = [(event, re.compile(event.regex)) for event in self.cs_events]

        while not self.stopped():
            exists = os.path.isfile(self.cs_log_filename)
            if not exists:
                time.sleep(0.5)
                continue

            for line_number, line in enumerate(self.follow_file(self.cs_log_filename)):
                if self.stopped():
                    break

                for event, pattern in patterns:
                    match = pattern.search(line)
                    if match:
                        event.add_info_and_publish(node=self.node, line=line, line_number=line_number)


class CassandraStressThread():  # pylint: disable=too-many-instance-attributes
    def __init__(self, loader_set, stress_cmd, timeout, stress_num=1, keyspace_num=1, keyspace_name='',  # pylint: disable=too-many-arguments
                 profile=None, node_list=None, round_robin=False, client_encrypt=False):
        if not node_list:
            node_list = []
        self.loader_set = loader_set
        self.stress_cmd = stress_cmd
        self.timeout = timeout
        self.stress_num = stress_num
        self.keyspace_num = keyspace_num
        self.keyspace_name = keyspace_name
        self.profile = profile
        self.node_list = node_list
        self.round_robin = round_robin
        self.client_encrypt = client_encrypt

        self.executor = None
        self.results_futures = []
        self.shell_marker = generate_random_string(20)
        #  This marker is used to mark shell commands, in order to be able to kill them later
        self.max_workers = 0

    def create_stress_cmd(self, node, loader_idx, keyspace_idx):
        stress_cmd = self.stress_cmd
        if node.cassandra_stress_version == "unknown":  # Prior to 3.11, cassandra-stress didn't have version argument
            stress_cmd = stress_cmd.replace("throttle", "limit")  # after 3.11 limit was renamed to throttle

        # When using cassandra-stress with "user profile" the profile yaml should be provided
        if 'profile' in stress_cmd and not self.profile:
            # support of using -profile in sct test-case yaml, assumes they exists data_dir
            # TODO: move those profile to their own directory
            cs_profile, profile = get_profile_content(stress_cmd)
            keyspace_name = profile['keyspace']
            self.profile = cs_profile
            self.keyspace_name = keyspace_name

        if self.keyspace_name:
            stress_cmd = stress_cmd.replace(" -schema ", " -schema keyspace={} ".format(self.keyspace_name))
        elif 'keyspace=' not in stress_cmd:  # if keyspace is defined in the command respect that
            stress_cmd = stress_cmd.replace(" -schema ", " -schema keyspace=keyspace{} ".format(keyspace_idx))

        credentials = self.loader_set.get_db_auth()
        if credentials and 'user=' not in stress_cmd:
            # put the credentials into the right place into -mode section
            stress_cmd = re.sub(r'(-mode.*?)-', r'\1 user={} password={} -'.format(*credentials), stress_cmd)
        if self.client_encrypt and 'transport' not in stress_cmd:
            stress_cmd += ' -transport "truststore=/etc/scylla/ssl_conf/client/cacerts.jks truststore-password=<PASSWORD>"'

        if self.node_list and '-node' not in stress_cmd:
            first_node = [n for n in self.node_list if n.dc_idx == loader_idx %
                          3]  # make sure each loader is targeting on datacenter/region
            first_node = first_node[0] if first_node else self.node_list[0]
            stress_cmd += " -node {}".format(first_node.ip_address)
        stress_cmd = self._add_errors_option(stress_cmd, ['skip-unsupported-columns'])
        return stress_cmd

    @staticmethod
    def _add_errors_option(stress_cmd: str, to_add: list):
        to_add = (' '.join(to_add))
        if ' -errors' not in stress_cmd:
            return stress_cmd + ' -errors ' + to_add
        return re.sub('-errors([ ]+[^-][a-zA-Z0-9-]+)+([ ]* -[a-z]+|[ ]*)', '-errors\\1 ' + to_add + '\\2', stress_cmd)

    def _run_stress(self, node, loader_idx, cpu_idx, keyspace_idx):
        stress_cmd = self.create_stress_cmd(node, loader_idx, keyspace_idx)

        if self.profile:
            with open(self.profile) as profile_file:
                LOGGER.info('Profile content:\n%s', profile_file.read())
            node.remoter.send_files(self.profile, os.path.join('/tmp', os.path.basename(self.profile)), delete_dst=True)

        stress_cmd_opt = stress_cmd.split()[1]

        LOGGER.info('Stress command:\n%s', stress_cmd)

        if not os.path.exists(node.logdir):
            makedirs(node.logdir)
        log_file_name = os.path.join(node.logdir,
                                     f'cassandra-stress-l{loader_idx}-c{cpu_idx}-k{keyspace_idx}-{uuid.uuid4()}.log')

        LOGGER.debug('cassandra-stress local log: %s', log_file_name)

        # This tag will be output in the header of c-stress result,
        # we parse it to know the loader & cpu info in _parse_cs_summary().
        tag = f'TAG: loader_idx:{loader_idx}-cpu_idx:{cpu_idx}-keyspace_idx:{keyspace_idx}'
{'fraction_non_alphanumeric': 0.09145532033815412, 'fraction_numerical': 0.004471459512331447, 'mean_word_length': 2.840622484572042, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5357016', 'n_tokens_mistral': 4283, 'n_tokens_neox': 4107, 'n_words': 1136}