Merge pull request #785 from fomars/develop

fix duration
v1.12.5

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='yandextank',
-    version='1.12.4',
+    version='1.12.5',
     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.24.2', 'numpy==1.15.4', 'future>=0.16.0',
         'pip>=8.1.2',
-        'pyyaml>=4.2b1', 'cerberus==1.2', 'influxdb>=5.0.0', 'netort>=0.6.8',
+        'pyyaml>=4.2b1', 'cerberus==1.2', 'influxdb>=5.0.0', 'netort>=0.6.9',
         'retrying>=1.3.3', 'pytest-runner'
     ],
     setup_requires=[

yandextank/core/tankcore.py

@@ -267,6 +267,7 @@ def plugins_start_test(self):
                 plugin.start_test()
                 logger.info("Plugin {0:s} required {1:f} seconds to start".format(plugin,
                                                                                   time.time() - start_time))
+            self.publish('generator', 'test_start', self.job.generator_plugin.start_time)
 
     def wait_for_finish(self):
         """
@@ -302,6 +303,7 @@ def plugins_end_test(self, retcode):
         """        Call end_test() on all plugins        """
         logger.info("Finishing test...")
         self.publish("core", "stage", "end")
+        self.publish('generator', 'test_end', time.time())
         logger.info("Stopping load generator and aggregator")
         retcode = self.job.aggregator.end_test(retcode)
         logger.debug("RC after: %s", retcode)

yandextank/plugins/Bfg/plugin.py

@@ -25,7 +25,6 @@ def __init__(self, core, cfg, name):
         self._bfg = None
         self.log = logging.getLogger(__name__)
         self.gun_type = None
-        self.start_time = time.time()
         self.stepper_wrapper = StepperWrapper(core, cfg)
         self.log.info("Initialized BFG")
         self.report_filename = "bfgout.log"

yandextank/plugins/NeUploader/plugin.py

@@ -26,6 +26,8 @@ def __init__(self, core, cfg, name):
                              'db_name': self.cfg.get('db_name')}]
         self.metrics_objs = {}  # map of case names and metric objects
         self.monitoring_metrics = {}
+        self._col_map = None
+        self._data_session = None
 
     def configure(self):
         pass
@@ -37,12 +39,11 @@ def start_test(self):
             logger.error('Generator plugin does not support NeUploader')
             self.is_test_finished = lambda: -1
             self.reader = []
-        else:
-            self.data_session = DataSession({'clients': self.clients_cfg})
-            self.add_cleanup(self._cleanup)
-            self.data_session.update_job({'name': self.cfg.get('test_name'),
-                                          '__type': 'tank'})
-            self.col_map = {
+
+    @property
+    def col_map(self):
+        if self._col_map is None:
+            self._col_map = {
                 'interval_real': self.data_session.new_true_metric,
                 'connect_time': self.data_session.new_true_metric,
                 'send_time': self.data_session.new_true_metric,
@@ -52,11 +53,22 @@ def start_test(self):
                 'net_code': self.data_session.new_event_metric,
                 'proto_code': self.data_session.new_event_metric
             }
+        return self._col_map
+
+    @property
+    def data_session(self):
+        if self._data_session is None:
+            self._data_session = DataSession({'clients': self.clients_cfg},
+                                             test_start=self.core.status['generator']['test_start'] * 10**6)
+            self.add_cleanup(self._cleanup)
+            self._data_session.update_job({'name': self.cfg.get('test_name'),
+                                          '__type': 'tank'})
+        return self._data_session
 
     def _cleanup(self):
         uploader_metainfo = self.map_uploader_tags(self.core.status.get('uploader'))
         self.data_session.update_job(uploader_metainfo)
-        self.data_session.close()
+        self.data_session.close(test_end=self.core.status.get('generator', {}).get('test_end', 0) * 10**6)
 
     def is_test_finished(self):
         df = next(self.reader)

yandextank/plugins/Pandora/plugin.py

@@ -31,7 +31,6 @@ def __init__(self, core, cfg, name):
         super(Plugin, self).__init__(core, cfg, name)
         self.output_finished = Event()
         self.enum_ammo = False
-        self.process_start_time = None
         self.pandora_cmd = None
         self.pandora_config_file = None
         self.config_contents = None
@@ -203,7 +202,7 @@ def start_test(self):
         if self.affinity:
             self.core.__setup_affinity(self.affinity, args=args)
         logger.info("Starting: %s", args)
-        self.process_start_time = time.time()
+        self.start_time = time.time()
         self.process_stderr_file = self.core.mkstemp(".log", "pandora_")
         self.core.add_artifact_file(self.process_stderr_file)
         self.process_stderr = open(self.process_stderr_file, 'w')

yandextank/plugins/Phantom/plugin.py

@@ -144,11 +144,11 @@ def start_test(self):
             logger.info('Enabled cpu affinity %s for phantom', self.affinity)
             args = self.core.__setup_affinity(self.affinity, args=args)
         logger.debug("Starting %s with arguments: %s", self.get_option("phantom_path"), args)
-        self.start_time = time.time()
         phantom_stderr_file = self.core.mkstemp(
             ".log", "phantom_stdout_stderr_")
         self.core.add_artifact_file(phantom_stderr_file)
         self.process_stderr = open(phantom_stderr_file, 'w')
+        self.start_time = time.time()
         self.process = subprocess.Popen(
             args,
             stderr=self.process_stderr,