<commit_before>""" Utility classes for phantom module """
# TODO: use separate answ log per benchmark
import copy
import logging
import multiprocessing
import string

from pkg_resources import resource_string
from yandextank.common.util import AddressWizard

from ...stepper import StepperWrapper
from ...stepper.util import parse_duration

logger = logging.getLogger(__name__)


class PhantomConfig:
    """ config file generator """
    OPTION_PHOUT = "phout_file"

    def __init__(self, core, cfg, stat_log):
        self.core = core
        self.cfg = cfg
        self.streams = []

        # common
        self.timeout = 11000
        self.answ_log = None
        self.answ_log_level = None
        self._phout_file = None
        self.stat_log = stat_log
        self.phantom_log = None
        self.phantom_start_time = None
        self.phantom_modules_path = None
        self.threads = None
        self.additional_libs = None
        self.enum_ammo = False
        self._config_file = None

    @staticmethod
    def get_available_options():
        opts = [
            "threads",
            "phantom_modules_path",
            "additional_libs",
            "writelog",
            "enum_ammo",
            "timeout",
        ]
        opts += StreamConfig.get_available_options()
        return opts

    @property
    def phout_file(self):
        if not self._phout_file:
            self._phout_file = self.cfg['phout_file'] or self.core.mkstemp(".log", "phout_")
        return self._phout_file

    def read_config(self):
        """        Read phantom tool specific options        """
        self.threads = self.cfg["threads"] or str(int(multiprocessing.cpu_count() / 2) + 1)
        self.phantom_modules_path = self.cfg["phantom_modules_path"]
        self.additional_libs = self.cfg["additional_libs"]
        self.answ_log_level = self.cfg["writelog"]
        if self.answ_log_level.lower() in ['0', 'false']:
            self.answ_log_level = 'none'
        elif self.answ_log_level.lower() in ['1', 'true']:
            self.answ_log_level = 'all'
        self.timeout = parse_duration(self.cfg["timeout"])
        if self.timeout > 120000:
            logger.warning(
                "You've set timeout over 2 minutes."
                " Are you a functional tester?")
        self.answ_log = self.core.mkstemp(".log", "answ_")
        self.core.add_artifact_file(self.answ_log)
        self.core.add_artifact_file(self.phout_file)
        self.core.add_artifact_file(self.stat_log)
        self.phantom_log = self.core.mkstemp(".log", "phantom_")
        self.core.add_artifact_file(self.phantom_log)

        main_stream = StreamConfig(
            self.core,
            len(self.streams), self.phout_file, self.answ_log,
            self.answ_log_level, self.timeout, self.cfg, True)
        self.streams.append(main_stream)

        for section in self.multi():
            self.streams.append(
                StreamConfig(
                    self.core,
                    len(self.streams), self.phout_file, self.answ_log,
                    self.answ_log_level, self.timeout, section))

        for stream in self.streams:
            stream.read_config()

        if any(stream.ssl for stream in self.streams):
            self.additional_libs += ' ssl io_benchmark_method_stream_transport_ssl'

    def multi(self):
        return (dict(self.cfg, **section) for section in self.cfg['multi'])

    @property
    def config_file(self):
        if self._config_file is None:
            self._config_file = self.compose_config()
        return self._config_file

    def compose_config(self):
        """        Generate phantom tool run config        """
        streams_config = ''
        stat_benchmarks = ''
        for stream in self.streams:
            streams_config += stream.compose_config()
            if not stream.is_main:
                stat_benchmarks += " " + "benchmark_io%s" % stream.sequence_no

        kwargs = {}
        kwargs['threads'] = self.threads
        kwargs['phantom_log'] = self.phantom_log
        kwargs['stat_log'] = self.stat_log
        kwargs['benchmarks_block'] = streams_config
        kwargs['stat_benchmarks'] = stat_benchmarks
        kwargs['additional_libs'] = self.additional_libs
        kwargs['phantom_modules_path'] = self.phantom_modules_path
        filename = self.core.mkstemp(".conf", "phantom_")
        self.core.add_artifact_file(filename)
        logger.debug("Generating phantom config: %s", filename)
        template_str = resource_string(__name__, "config/phantom.conf.tpl")
        tpl = string.Template(template_str)
        config = tpl.substitute(kwargs)

        handle = open(filename, 'w')
        handle.write(config)
        handle.close()
        return filename

    def set_timeout(self, timeout):
        """ pass timeout to all streams """
        for stream in self.streams:
            stream.timeout = timeout

    def get_info(self):
        """ get merged info about phantom conf """
        result = copy.copy(self.streams[0])
        result.stat_log = self.stat_log
        result.steps = []
        result.ammo_file = ''
        result.rps_schedule = None
        result.ammo_count = 0
        result.duration = 0

        result.instances = 0
        result.loadscheme = []
        result.loop_count = 0

        for stream in self.streams:
            sec_no = 0
            logger.debug("Steps: %s", stream.stepper_wrapper.steps)
            for item in stream.stepper_wrapper.steps:
                for x in range(0, item[1]):
                    if len(result.steps) > sec_no:
                        result.steps[sec_no][0] += item[0]
                    else:
                        result.steps.append([item[0], 1])
                    sec_no += 1

            if result.rps_schedule:
                result.rps_schedule = []
            else:
                result.rps_schedule = stream.stepper_wrapper.loadscheme
            if result.loadscheme:
                result.loadscheme = ''
            else:
                # FIXME: add formatted load scheme for server:
                # <step_size,step_type,first_rps,last_rps,original_step_params>
                # as a string
                result.loadscheme = ''

            if result.loop_count:
                result.loop_count = u'0'
            else:
                result.loop_count = stream.stepper_wrapper.loop_count

            result.ammo_file += stream.stepper_wrapper.ammo_file + ' '<commit_msg>fix phantom info ammo=null<commit_after>""" Utility classes for phantom module """
# TODO: use separate answ log per benchmark
import copy
import logging
import multiprocessing
import string

from pkg_resources import resource_string
from yandextank.common.util import AddressWizard

from ...stepper import StepperWrapper
from ...stepper.util import parse_duration

logger = logging.getLogger(__name__)


class PhantomConfig:
    """ config file generator """
    OPTION_PHOUT = "phout_file"

    def __init__(self, core, cfg, stat_log):
        self.core = core
        self.cfg = cfg
        self.streams = []

        # common
        self.timeout = 11000
        self.answ_log = None
        self.answ_log_level = None
        self._phout_file = None
        self.stat_log = stat_log
        self.phantom_log = None
        self.phantom_start_time = None
        self.phantom_modules_path = None
        self.threads = None
        self.additional_libs = None
        self.enum_ammo = False
        self._config_file = None

    @staticmethod
    def get_available_options():
        opts = [
            "threads",
            "phantom_modules_path",
            "additional_libs",
            "writelog",
            "enum_ammo",
            "timeout",
        ]
        opts += StreamConfig.get_available_options()
        return opts

    @property
    def phout_file(self):
        if not self._phout_file:
            self._phout_file = self.cfg['phout_file'] or self.core.mkstemp(".log", "phout_")
        return self._phout_file

    def read_config(self):
        """        Read phantom tool specific options        """
        self.threads = self.cfg["threads"] or str(int(multiprocessing.cpu_count() / 2) + 1)
        self.phantom_modules_path = self.cfg["phantom_modules_path"]
        self.additional_libs = self.cfg["additional_libs"]
        self.answ_log_level = self.cfg["writelog"]
        if self.answ_log_level.lower() in ['0', 'false']:
            self.answ_log_level = 'none'
        elif self.answ_log_level.lower() in ['1', 'true']:
            self.answ_log_level = 'all'
        self.timeout = parse_duration(self.cfg["timeout"])
        if self.timeout > 120000:
            logger.warning(
                "You've set timeout over 2 minutes."
                " Are you a functional tester?")
        self.answ_log = self.core.mkstemp(".log", "answ_")
        self.core.add_artifact_file(self.answ_log)
        self.core.add_artifact_file(self.phout_file)
        self.core.add_artifact_file(self.stat_log)
        self.phantom_log = self.core.mkstemp(".log", "phantom_")
        self.core.add_artifact_file(self.phantom_log)

        main_stream = StreamConfig(
            self.core,
            len(self.streams), self.phout_file, self.answ_log,
            self.answ_log_level, self.timeout, self.cfg, True)
        self.streams.append(main_stream)

        for section in self.multi():
            self.streams.append(
                StreamConfig(
                    self.core,
                    len(self.streams), self.phout_file, self.answ_log,
                    self.answ_log_level, self.timeout, section))

        for stream in self.streams:
            stream.read_config()

        if any(stream.ssl for stream in self.streams):
            self.additional_libs += ' ssl io_benchmark_method_stream_transport_ssl'

    def multi(self):
        return (dict(self.cfg, **section) for section in self.cfg['multi'])

    @property
    def config_file(self):
        if self._config_file is None:
            self._config_file = self.compose_config()
        return self._config_file

    def compose_config(self):
        """        Generate phantom tool run config        """
        streams_config = ''
        stat_benchmarks = ''
        for stream in self.streams:
            streams_config += stream.compose_config()
            if not stream.is_main:
                stat_benchmarks += " " + "benchmark_io%s" % stream.sequence_no

        kwargs = {}
        kwargs['threads'] = self.threads
        kwargs['phantom_log'] = self.phantom_log
        kwargs['stat_log'] = self.stat_log
        kwargs['benchmarks_block'] = streams_config
        kwargs['stat_benchmarks'] = stat_benchmarks
        kwargs['additional_libs'] = self.additional_libs
        kwargs['phantom_modules_path'] = self.phantom_modules_path
        filename = self.core.mkstemp(".conf", "phantom_")
        self.core.add_artifact_file(filename)
        logger.debug("Generating phantom config: %s", filename)
        template_str = resource_string(__name__, "config/phantom.conf.tpl")
        tpl = string.Template(template_str)
        config = tpl.substitute(kwargs)

        handle = open(filename, 'w')
        handle.write(config)
        handle.close()
        return filename

    def set_timeout(self, timeout):
        """ pass timeout to all streams """
        for stream in self.streams:
            stream.timeout = timeout

    def get_info(self):
        """ get merged info about phantom conf """
        result = copy.copy(self.streams[0])
        result.stat_log = self.stat_log
        result.steps = []
        result.ammo_file = ''
        result.rps_schedule = None
        result.ammo_count = 0
        result.duration = 0

        result.instances = 0
        result.loadscheme = []
        result.loop_count = 0

        for stream in self.streams:
            sec_no = 0
            logger.debug("Steps: %s", stream.stepper_wrapper.steps)
            for item in stream.stepper_wrapper.steps:
                for x in range(0, item[1]):
                    if len(result.steps) > sec_no:
                        result.steps[sec_no][0] += item[0]
                    else:
                        result.steps.append([item[0], 1])
                    sec_no += 1

            if result.rps_schedule:
                result.rps_schedule = []
            else:
                result.rps_schedule = stream.stepper_wrapper.loadscheme
            if result.loadscheme:
                result.loadscheme = ''
            else:
                # FIXME: add formatted load scheme for server:
                # <step_size,step_type,first_rps,last_rps,original_step_params>
                # as a string
                result.loadscheme = ''

            if result.loop_count:
                result.loop_count = u'0'
            else:
                result.loop_count = stream.stepper_wrapper.loop_count

            result.ammo_file += '{} '.format(stream.stepper_wrapper.ammo_file)
{'fraction_non_alphanumeric': 0.09934790947449175, 'fraction_numerical': 0.004602991944764097, 'mean_word_length': 2.0043788891449643, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, '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': '5919289', 'n_tokens_mistral': 3907, 'n_tokens_neox': 3678, 'n_words': 769}