Merge pull request #662 from yandex/refactoring

New tankworker, statuses, locks

.flake8

@@ -25,4 +25,4 @@ exclude =
     *.egg-info/,
     .installed.cfg,
     *.egg
-ignore = E501
+ignore = E501 W503

.gitignore

@@ -92,3 +92,8 @@ ENV/
 tmp/
 yandex-tank.iml
 .idea/
+
+.pytest_cache/
+*.tld.cfg
+logs/
+*.DS_Store

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='yandextank',
-    version='1.10.3',
+    version='1.11.0',
     description='a performance measurement tool',
     longer_description='''
 Yandex.Tank is a performance measurement and load testing automatization tool.
@@ -41,7 +41,6 @@
         'Topic :: Software Development :: Testing',
         'Topic :: Software Development :: Testing :: Traffic Generation',
         'Programming Language :: Python :: 2',
-        'Programming Language :: Python :: 3',
     ],
     entry_points={
         'console_scripts': [

yandextank/aggregator/aggregator.py

@@ -40,14 +40,10 @@ def __init__(self, config, verbose_histogram):
                              np.linspace(10, 499, 490) * 1000)  # 1ms accuracy
             bins = np.append(bins,
                              np.linspace(500, 2995, 500) * 1000)  # 5ms accuracy
-            bins = np.append(bins, np.linspace(3000, 9990, 700) *
-                             1000)  # 10ms accuracy
-            bins = np.append(bins, np.linspace(10000, 29950, 400) *
-                             1000)  # 50ms accuracy
-            bins = np.append(bins, np.linspace(30000, 119900, 900) *
-                             1000)  # 100ms accuracy
-            bins = np.append(bins, np.linspace(120, 300, 181) *
-                             1000000)  # 1s accuracy
+            bins = np.append(bins, np.linspace(3000, 9990, 700) * 1000)  # 10ms accuracy
+            bins = np.append(bins, np.linspace(10000, 29950, 400) * 1000)  # 50ms accuracy
+            bins = np.append(bins, np.linspace(30000, 119900, 900) * 1000)  # 100ms accuracy
+            bins = np.append(bins, np.linspace(120, 300, 181) * 1000000)  # 1s accuracy
         else:
             # yapf: disable
             bins = np.array([

yandextank/api/apiworker.py

@@ -146,8 +146,7 @@ def get_default_configs(self):
                     ]
         except OSError:
             self.log.warn(
-                self.baseconfigs_location +
-                ' is not accessible to get configs list')
+                self.baseconfigs_location + ' is not accessible to get configs list')
 
         configs += [os.path.expanduser('~/.yandex-tank')]
         return configs

yandextank/common/interfaces.py

@@ -20,6 +20,7 @@ def __init__(self, core, cfg):
         @type core: TankCore
         """
         super(AbstractPlugin, self).__init__()
+        self._cleanup_actions = []
         self.log = logging.getLogger(__name__)
         self.core = core
         self.cfg = cfg
@@ -42,10 +43,24 @@ def start_test(self):
     def is_test_finished(self):
         """
         Polling call, if result differs from -1 then test end
-        will be triggeted
+        will be triggered
         """
         return -1
 
+    def add_cleanup(self, action):
+        """
+        :type action: function
+        """
+        assert callable(action)
+        self._cleanup_actions.append(action)
+
+    def cleanup(self):
+        for action in reversed(self._cleanup_actions):
+            try:
+                action()
+            except Exception:
+                logging.error('Exception occurred during plugin cleanup {}'.format(self.__module__), exc_info=True)
+
     def end_test(self, retcode):
         """
         Stop processes launched at 'start_test',

yandextank/common/util.py

@@ -383,7 +383,7 @@ def expand_time(str_time, default_unit='s', multiplier=1):
     """
     helper for above functions
     """
-    parser = re.compile('(\d+)([a-zA-Z]*)')
+    parser = re.compile(r'(\d+)([a-zA-Z]*)')
     parts = parser.findall(str_time)
     result = 0.0
     for value, unit in parts:

yandextank/config_converter/converter.py

@@ -186,12 +186,12 @@ class Option(object):
             'instances_schedule': convert_instances_schedule,
             'stpd_file': convert_stpd_schedule,
             'autocases': TYPE_CASTERS['integer'],
-            'headers': lambda key, value: {key: re.compile('\[(.*?)\]').findall(value)}
+            'headers': lambda key, value: {key: re.compile(r'\[(.*?)\]').findall(value)}
         },
         'Bfg': {
             'rps_schedule': convert_rps_schedule,
             'instances_schedule': convert_instances_schedule,
-            'headers': lambda key, value: {key: re.compile('\[(.*?)\]').findall(value)}
+            'headers': lambda key, value: {key: re.compile(r'\[(.*?)\]').findall(value)}
         },
         'JMeter': {
             'exclude_markers': lambda key, value: {key: value.strip().split(' ')}
@@ -201,7 +201,7 @@ class Option(object):
             'config_content': lambda key, value: {key: yaml.load(value)}  # works for json as well
         },
         'Autostop': {
-            'autostop': lambda k, v: {k: re.findall('\w+\(.+?\)', v)}
+            'autostop': lambda k, v: {k: re.findall(r'\w+\(.+?\)', v)}
         },
         'DataUploader': {
             'lock_targets': lambda k, v: {k: v.strip().split() if v != 'auto' else v}

yandextank/config_converter/tests/test_converter.py

@@ -116,9 +116,9 @@ def test_conflict_opts(ini_file, msgs):
 ])
 def test_validate(ini_file):
     # noinspection PyStatementEffect
-    TankConfig([load_core_base_cfg()] +
-               cfg_folder_loader(os.path.join(os.path.dirname(__file__), 'etc_cfg')) +
-               [load_cfg(os.path.join(os.path.dirname(__file__), ini_file))]).validated
+    TankConfig([load_core_base_cfg()]
+               + cfg_folder_loader(os.path.join(os.path.dirname(__file__), 'etc_cfg'))
+               + [load_cfg(os.path.join(os.path.dirname(__file__), ini_file))]).validated
 
 
 @pytest.mark.parametrize('key, value, expected', [

yandextank/core/cli.py

@@ -1,9 +1,11 @@
 import logging
 import sys
-import traceback
 from optparse import OptionParser
 
-from yandextank.core.consoleworker import ConsoleTank, CompletionHelperOptionParser
+import pkg_resources
+from netort.resource import manager as resource_manager
+from yandextank.core.consoleworker import TankWorker
+from yandextank.validator.validator import ValidationError
 
 
 def main():
@@ -90,22 +92,127 @@ def main():
         and the exact scheme of yaml format config)',
         dest='patches'
     )
-
-    completion_helper = CompletionHelperOptionParser()
-    completion_helper.handle_request(parser)
+    parser.add_option(
+        '--version',
+        action='store_true',
+        dest='version'
+    )
+    # FIXME: restore completion helper
+    # completion_helper = CompletionHelperOptionParser()
+    # completion_helper.handle_request(parser)
 
     options, ammofiles = parser.parse_args()
+    if options.version:
+        print('YandexTank/{}'.format(pkg_resources.require('yandextank')[0].version))
+        return
+
     ammofile = ammofiles[0] if len(ammofiles) > 0 else None
-    worker = ConsoleTank(options, ammofile)
+
+    init_logging(options.error_log, options.verbose, options.quiet)
+
+    cli_kwargs = {'core': {'lock_dir': options.lock_dir}} if options.lock_dir else {}
+    if options.ignore_lock:
+        cli_kwargs.setdefault('core', {})['ignore_lock'] = options.ignore_lock
+
+    if ammofile:
+        logging.debug("Ammofile: %s", ammofile)
+        cli_kwargs['phantom'] = {
+            'use_caching': False,
+            'ammofile': ammofile
+        }
+    try:
+        worker = TankWorker([resource_manager.resource_filename(cfg) for cfg in options.config],
+                            options.option,
+                            options.patches,
+                            [cli_kwargs],
+                            options.no_rc,
+                            ammo_file=ammofile if ammofile else None)
+    except ValidationError as e:
+        logging.error('Config validation error:\n{}'.format(e.errors))
+        return
+    worker.start()
     try:
-        worker.configure()
-        rc = worker.perform_test()
-        sys.exit(rc)
-    except Exception as ex:
-        worker.core._collect_artifacts()
-        logging.error("Exception: %s", ex)
-        logging.debug("Exception: %s", traceback.format_exc(ex))
-        sys.exit(1)
+        worker.join()
+    except KeyboardInterrupt:
+        worker.stop()
+        worker.join()
+    # try:
+    #     worker.configure()
+    #     rc = worker.perform_test()
+    #     sys.exit(rc)
+    # except Exception as ex:
+    #     worker.core._collect_artifacts()
+    #     logging.error("Exception: %s", ex)
+    #     logging.debug("Exception: %s", traceback.format_exc(ex))
+    #     sys.exit(1)
+
+
+def init_logging(events_log_fname, verbose, quiet):
+    """ Set up logging, as it is very important for console tool """
+    logger = logging.getLogger('')
+    logger.setLevel(logging.DEBUG)
+
+    # create file handler which logs error messages
+    if events_log_fname:
+        err_file_handler = logging.FileHandler(events_log_fname)
+        err_file_handler.setLevel(logging.WARNING)
+        err_file_handler.setFormatter(
+            logging.Formatter(
+                "%(asctime)s\t%(message)s"
+            ))
+        logger.addHandler(err_file_handler)
+
+    # create console handler with a higher log level
+    console_handler = logging.StreamHandler(sys.stdout)
+    stderr_hdl = logging.StreamHandler(sys.stderr)
+
+    fmt_verbose = logging.Formatter(
+        "%(asctime)s [%(levelname)s] %(name)s %(filename)s:%(lineno)d\t%(message)s"
+    )
+    fmt_regular = logging.Formatter(
+        "%(asctime)s [%(levelname)s] %(message)s", "%H:%M:%S")
+
+    if verbose:
+        console_handler.setLevel(logging.DEBUG)
+        console_handler.setFormatter(fmt_verbose)
+        stderr_hdl.setFormatter(fmt_verbose)
+    elif quiet:
+        console_handler.setLevel(logging.WARNING)
+        console_handler.setFormatter(fmt_regular)
+        stderr_hdl.setFormatter(fmt_regular)
+    else:
+        console_handler.setLevel(logging.INFO)
+        console_handler.setFormatter(fmt_regular)
+        stderr_hdl.setFormatter(fmt_regular)
+
+    f_err = SingleLevelFilter(logging.ERROR, True)
+    f_warn = SingleLevelFilter(logging.WARNING, True)
+    f_crit = SingleLevelFilter(logging.CRITICAL, True)
+    console_handler.addFilter(f_err)
+    console_handler.addFilter(f_warn)
+    console_handler.addFilter(f_crit)
+    logger.addHandler(console_handler)
+
+    f_info = SingleLevelFilter(logging.INFO, True)
+    f_debug = SingleLevelFilter(logging.DEBUG, True)
+    stderr_hdl.addFilter(f_info)
+    stderr_hdl.addFilter(f_debug)
+    logger.addHandler(stderr_hdl)
+
+
+class SingleLevelFilter(logging.Filter):
+    """Exclude or approve one msg type at a time.    """
+
+    def __init__(self, passlevel, reject):
+        logging.Filter.__init__(self)
+        self.passlevel = passlevel
+        self.reject = reject
+
+    def filter(self, record):
+        if self.reject:
+            return record.levelno != self.passlevel
+        else:
+            return record.levelno == self.passlevel
 
 
 if __name__ == '__main__':

yandextank/core/config/schema.yaml

@@ -16,8 +16,6 @@ core:
     artifacts_dir:
       description: directory inside base directory to store test\'s artifacts, defaults to api_jobno if null
       type: string
-      nullable: true
-      default: null
     cmdline:
       type: string
     exitcode: