Merge pull request #706 from yandex/release

Release 1.11.3

.travis.yml

@@ -7,5 +7,5 @@ install:
   - "pip install flake8"
   - "pip install ."
 script:
-  - "flake8 ."
+  - "flake8 --config .flake8 ."
   - "pytest -s"

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='yandextank',
-    version='1.11.2',
+    version='1.11.3',
     description='a performance measurement tool',
     longer_description='''
 Yandex.Tank is a performance measurement and load testing automatization tool.
@@ -20,7 +20,7 @@
         'psutil>=1.2.1', 'requests>=2.5.1', 'paramiko>=1.16.0',
         'pandas>=0.18.0', 'numpy>=1.12.1', 'future>=0.16.0',
         'pip>=8.1.2',
-        'pyyaml>=3.12', 'cerberus==1.2', 'influxdb>=5.0.0', 'netort==0.2.8',
+        'pyyaml>=3.12', 'cerberus==1.2', 'influxdb>=5.0.0', 'netort>=0.3.1',
     ],
     setup_requires=[
     ],

yandextank/aggregator/tank_aggregator.py

@@ -124,15 +124,11 @@ def end_test(self, retcode):
             self.stats_reader.close()
         if self.drain:
             logger.debug('Waiting for gun drain to finish')
-            self.drain.wait()
+            self.drain.join()
             logger.debug('Waiting for stats drain to finish')
-            self.stats_drain.wait()
+            self.stats_drain.join()
         logger.debug('Collecting remaining data')
         self._collect_data(end=True)
-        if self.drain:
-            self.drain.join()
-            self.stats_drain.join()
-
         return retcode
 
     def add_result_listener(self, listener):

yandextank/core/cli.py

@@ -108,7 +108,7 @@ def main():
 
     ammofile = ammofiles[0] if len(ammofiles) > 0 else None
 
-    init_logging(options.error_log, options.verbose, options.quiet)
+    handlers = 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:
@@ -127,6 +127,7 @@ def main():
                             [cli_kwargs],
                             options.no_rc,
                             ammo_file=ammofile if ammofile else None,
+                            log_handlers=handlers
                             )
     except ValidationError as e:
         logging.error('Config validation error:\n{}'.format(e.errors))

yandextank/core/consoleworker.py

@@ -21,7 +21,7 @@
 from ..config_converter.converter import convert_ini, convert_single_option
 
 DEFAULT_CONFIG = 'load.yaml'
-logger = logging.getLogger('yandextank')
+logger = logging.getLogger()
 
 
 class RealConsoleMarkup(object):
@@ -429,6 +429,7 @@ def init_logging(self, debug=False):
         file_handler.setLevel(logging.DEBUG)
         file_handler.setFormatter(logging.Formatter(
             "%(asctime)s [%(levelname)s] %(name)s %(filename)s:%(lineno)d\t%(message)s"))
+        file_handler.addFilter(TankapiLogFilter())
         logger.addHandler(file_handler)
         logger.info("Log file created")
 
@@ -458,6 +459,11 @@ def set_msg(self, msg):
         self.msg = msg
 
 
+class TankapiLogFilter(logging.Filter):
+    def filter(self, record):
+        return record.name != 'tankapi'
+
+
 class DevNullOpts:
     def __init__(self):
         pass

yandextank/core/tankcore.py

@@ -36,6 +36,9 @@
 
 logger = logging.getLogger(__name__)
 
+CONFIGINITIAL = 'configinitial.yaml'
+VALIDATED_CONF = 'validated_conf.yaml'
+
 
 class Job(object):
     def __init__(
@@ -129,14 +132,16 @@ def __init__(self, configs, interrupted_event, artifacts_base_dir=None, artifact
         self.test_id = self.get_option(self.SECTION, 'artifacts_dir',
                                        datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S.%f"))
         self.lock_dir = self.get_option(self.SECTION, 'lock_dir')
-        with open(os.path.join(self.artifacts_dir, 'configinitial.yaml'), 'w') as f:
+        with open(os.path.join(self.artifacts_dir, CONFIGINITIAL), 'w') as f:
             yaml.dump(self.configinitial, f)
         self.add_artifact_file(error_output)
         self.add_artifact_to_send(LPRequisites.CONFIGINITIAL, yaml.dump(self.configinitial))
         configinfo = self.config.validated.copy()
         configinfo.setdefault(self.SECTION, {})
         configinfo[self.SECTION][self.API_JOBNO] = self.test_id
         self.add_artifact_to_send(LPRequisites.CONFIGINFO, yaml.dump(configinfo))
+        with open(os.path.join(self.artifacts_dir, VALIDATED_CONF), 'w') as f:
+            yaml.dump(configinfo, f)
         logger.info('New test id %s' % self.test_id)
 
     @property

yandextank/plugins/DataUploader/cli.py

@@ -17,12 +17,12 @@
 from cerberus import Validator
 
 from yandextank.core import TankCore
+from yandextank.core.tankcore import VALIDATED_CONF
 from yandextank.validator.validator import ValidationError, load_yaml_schema
 from .client import APIClient, OverloadClient, LPRequisites
 from .plugin import LPJob, BackendTypes
 from .plugin import Plugin as DataUploader
 
-CONFIG_FILE = 'saved_conf.yaml'
 DATA_LOG = 'test_data.log'
 MONITORING_LOG = 'monitoring.log'
 SECTION = 'meta'
@@ -43,19 +43,19 @@ def get_logger():
 
 def from_tank_config(test_dir):
     try:
-        config_file = glob.glob(os.path.join(test_dir, CONFIG_FILE))[0]
+        config_file = glob.glob(os.path.join(test_dir, VALIDATED_CONF))[0]
         logger.info('Config file found: %s' % config_file)
     except IndexError:
-        raise OSError('Config file {} not found in {}'.format(CONFIG_FILE, test_dir))
+        raise OSError('Config file {} not found in {}'.format(VALIDATED_CONF, test_dir))
 
     with open(config_file) as f:
         tank_cfg = yaml.load(f)
     try:
-        config = filter(lambda options: 'DataUploader' in options.get('package', ''), tank_cfg.values())[0]
+        section, config = filter(lambda item: 'DataUploader' in item[1].get('package', ''), tank_cfg.items())[0]
     except IndexError:
         logger.warning('DataUploader configuration not found in {}'.format(config_file))
-        config = {}
-    return config
+        section, config = None, {}
+    return section, config
 
 
 def check_log(log_name):
@@ -151,8 +151,9 @@ def post_loader():
     if args.config:
         with open(args.config) as f:
             config = yaml.load(f)
+            section = None
     else:
-        config = from_tank_config(args.test_dir)
+        section, config = from_tank_config(args.test_dir)
     # parse target host and port
     if args.target is not None:
         try:
@@ -180,7 +181,7 @@ def post_loader():
     config = v.normalized(config)
 
     # lunapark or overload?
-    backend_type = BackendTypes.identify_backend(config['api_address'])
+    backend_type = BackendTypes.identify_backend(config['api_address'], section)
     if backend_type == BackendTypes.LUNAPARK:
         client = APIClient
         api_token = None

yandextank/plugins/DataUploader/client.py

@@ -674,3 +674,6 @@ def link_mobile_job(self, lp_key, mobile_key):
 
     def push_events_data(self, number, token, data):
         return
+
+    def set_imbalance_and_dsc(self, **kwargs):
+        return

yandextank/plugins/DataUploader/config/postloader_schema.yaml

@@ -3,14 +3,16 @@ api_address:
   required: true
 target_host:
   type: string
-  required: true
+  default: undefined
 target_port:
   anyof:
     - type: string
     - type: integer
   default: 80
 operator:
   type: string
+  nullable: true
+  default: null
 task:
   type: string
 job_name:

yandextank/plugins/DataUploader/tests/test_postloader.py

@@ -3,24 +3,25 @@
 
 
 @pytest.mark.parametrize('test_dir, expected', [
-    ('yandextank/plugins/DataUploader/tests/test_postloader/test_empty', {}),
+    ('yandextank/plugins/DataUploader/tests/test_postloader/test_empty', (None, {})),
     ('yandextank/plugins/DataUploader/tests/test_postloader/test_full',
-     {'api_address': 'https://lunapark.yandex-team.ru/',
-      'api_attempts': 2,
-      'api_timeout': 5,
-      'enabled': True,
-      'job_dsc': 'hell of a kitty',
-      'job_name': 'Hello kitty',
-      'jobno_file': 'jobno.txt',
-      'lock_targets': 'foo.bar',
-      'maintenance_timeout': 5,
-      'network_attempts': 2,
-      'operator': 'fomars',
-      'package': 'yandextank.plugins.DataUploader',
-      'task': 'LOAD-204'}
+     ('uploader',
+      {'api_address': 'https://lunapark.yandex-team.ru/',
+       'api_attempts': 2,
+       'api_timeout': 5,
+       'enabled': True,
+       'job_dsc': 'hell of a kitty',
+       'job_name': 'Hello kitty',
+       'jobno_file': 'jobno.txt',
+       'lock_targets': 'foo.bar',
+       'maintenance_timeout': 5,
+       'network_attempts': 2,
+       'operator': 'fomars',
+       'package': 'yandextank.plugins.DataUploader',
+       'task': 'LOAD-204'})
      ),
     ('yandextank/plugins/DataUploader/tests/test_postloader/test_disabled',
-     {'enabled': False, 'package': 'yandextank.plugins.DataUploader'}),
+     ('uploader', {'enabled': False, 'package': 'yandextank.plugins.DataUploader'})),
 ])
 def test_from_tank_config(test_dir, expected):
     get_logger()

yandextank/plugins/JMeter/reader.py

@@ -112,8 +112,10 @@ def fix_latency(row):
 
 # timeStamp,elapsed,label,responseCode,success,bytes,grpThreads,allThreads,Latency
 def string_to_df(data):
-    chunk = pd.read_csv(
-        StringIO(data), sep='\t', names=jtl_columns, dtype=jtl_types)
+    chunk = pd.read_csv(StringIO(data),
+                        sep='\t',
+                        names=jtl_columns, dtype=jtl_types,
+                        keep_default_na=False)
     chunk["receive_ts"] = (chunk["send_ts"] + chunk['interval_real']) / 1000.0
     chunk['receive_sec'] = chunk["receive_ts"].astype(np.int64)
     chunk['interval_real'] = chunk["interval_real"] * 1000  # convert to µs
@@ -183,7 +185,6 @@ def _read_jtl_chunk(self, jtl):
                 self.buffer = parts[1]
                 df = string_to_df(ready_chunk)
                 self.stat_queue.put(df)
-
                 return df
             else:
                 self.buffer += parts[0]

yandextank/plugins/Phantom/reader.py

@@ -63,6 +63,22 @@ def string_to_df(data):
     return chunk
 
 
+def string_to_df_microsec(data):
+    # start_time = time.time()
+    try:
+        df = pd.read_csv(StringIO(data), sep='\t', names=phout_columns, dtype=dtypes, quoting=QUOTE_NONE)
+    except CParserError as e:
+        logger.error(e.message)
+        logger.error('Incorrect phout data: {}'.format(data))
+        return
+
+    df['ts'] = (df['send_ts'] * 1e6 + df['interval_real']).astype(int)
+    df['ts'] -= df["ts"][0]
+    df['tag'] = df.tag.str.rsplit('#', 1, expand=True)[0]
+    # logger.debug("Chunk decode time: %.2fms", (time.time() - start_time) * 1000)
+    return df
+
+
 class PhantomReader(object):
     def __init__(self, filename, cache_size=1024 * 1024 * 50, ready_file=False):
         self.buffer = ""

yandextank/plugins/Phantom/tests/test_reader.py

@@ -1,6 +1,6 @@
 import pandas as pd
 
-from yandextank.plugins.Phantom.reader import PhantomReader, PhantomStatsReader
+from yandextank.plugins.Phantom.reader import PhantomReader, PhantomStatsReader, string_to_df_microsec
 
 
 class TestPhantomReader(object):
@@ -24,6 +24,13 @@ def test_reader_closed(self):
         assert len(result) == 200
         assert (result['interval_real'].mean() == 11000714.0)
 
+    def test_reader_us(self):
+        with open('yandextank/plugins/Phantom/tests/phout.dat') as f:
+            chunk = f.read()
+        result = string_to_df_microsec(chunk)
+        expected = pd.read_pickle('yandextank/plugins/Phantom/tests/expected_df.dat')
+        assert result.equals(expected)
+
 
 class MockInfo(object):
     def __init__(self, steps):