Merge pull request #712 from fomars/develop

neuploader cleanup, null check

setup.py

@@ -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.3.0',
+        'pyyaml>=3.12', 'cerberus==1.2', 'influxdb>=5.0.0', 'netort>=0.3.2',
         'retrying==1.3.3', 'pytest-benchmark==3.2.2'
     ],
     setup_requires=[

yandextank/plugins/NeUploader/plugin.py

@@ -21,6 +21,7 @@ def configure(self):
 
     def start_test(self):
         self.data_session = DataSession({'clients': self.clients_cfg})
+        self.add_cleanup(self.data_session.close)
         self.data_session.update_job({'name': self.cfg.get('test_name')})
         col_map_aggr = {name: 'metric %s' % name for name in
                         ['interval_real', 'connect_time', 'send_time', 'latency',
@@ -29,7 +30,9 @@ def start_test(self):
         self.reader = self.core.job.generator_plugin.get_reader(parser=string_to_df_microsec)
 
     def is_test_finished(self):
-        self.uploader(next(self.reader))
+        df = next(self.reader)
+        if df is not None:
+            self.uploader(df)
         return -1
 
     def monitoring_data(self, data_list):
@@ -38,7 +41,6 @@ def monitoring_data(self, data_list):
     def post_process(self, retcode):
         for chunk in self.reader:
             self.uploader(chunk)
-        self.data_session.close()
         return retcode
 
     @property