Merge "Yardstick output format unified"

11 files changed, 220 insertions, 273 deletions

diff --git a/yardstick/benchmark/core/task.py b/yardstick/benchmark/core/task.py
index 5a006f2b2..478a51f9d 100644
--- a/yardstick/benchmark/core/task.py
+++ b/yardstick/benchmark/core/task.py
@@ -24,6 +24,7 @@ from six.moves import filter
 
 from yardstick.benchmark.contexts.base import Context
 from yardstick.benchmark.runners import base as base_runner
+from yardstick.dispatcher.base import Base as DispatcherBase
 from yardstick.common.task_template import TaskTemplate
 from yardstick.common.utils import source_env
 from yardstick.common import utils
@@ -42,7 +43,6 @@ class Task(object):     # pragma: no cover
     """
 
     def __init__(self):
-        self.config = {}
         self.contexts = []
         self.outputs = {}
 
@@ -55,7 +55,14 @@ class Task(object):     # pragma: no cover
 
         check_environment()
 
-        self.config['yardstick'] = utils.parse_ini_file(config_file)
+        output_config = utils.parse_ini_file(config_file)
+        self._init_output_config(output_config)
+        self._set_output_config(output_config, args.output_file)
+        LOG.debug('Output configuration is: %s', output_config)
+
+        if output_config['DEFAULT'].get('dispatcher') == 'file':
+            result = {'status': 0, 'result': {}}
+            utils.write_json_to_file(args.output_file, result)
 
         total_start_time = time.time()
         parser = TaskParser(args.inputfile[0])
@@ -75,6 +82,7 @@ class Task(object):     # pragma: no cover
         if args.parse_only:
             sys.exit(0)
 
+        testcases = {}
         # parse task_files
         for i in range(0, len(task_files)):
             one_task_start_time = time.time()
@@ -90,7 +98,15 @@ class Task(object):     # pragma: no cover
                          meet_precondition)
                 continue
 
-            self._run(scenarios, run_in_parallel, args.output_file)
+            case_name = os.path.splitext(os.path.basename(task_files[i]))[0]
+            try:
+                data = self._run(scenarios, run_in_parallel, args.output_file)
+            except KeyboardInterrupt:
+                raise
+            except Exception:
+                testcases[case_name] = {'criteria': 'FAIL', 'tc_data': []}
+            else:
+                testcases[case_name] = {'criteria': 'PASS', 'tc_data': data}
 
             if args.keep_deploy:
                 # keep deployment, forget about stack
@@ -104,6 +120,10 @@ class Task(object):     # pragma: no cover
             LOG.info("task %s finished in %d secs", task_files[i],
                      one_task_end_time - one_task_start_time)
 
+        result = self._get_format_result(testcases)
+
+        self._do_output(output_config, result)
+
         total_end_time = time.time()
         LOG.info("total finished in %d secs",
                  total_end_time - total_start_time)
@@ -114,6 +134,65 @@ class Task(object):     # pragma: no cover
 
         print("Done, exiting")
 
+    def _init_output_config(self, output_config):
+        output_config.setdefault('DEFAULT', {})
+        output_config.setdefault('dispatcher_http', {})
+        output_config.setdefault('dispatcher_file', {})
+        output_config.setdefault('dispatcher_influxdb', {})
+        output_config.setdefault('nsb', {})
+
+    def _set_output_config(self, output_config, file_path):
+        try:
+            out_type = os.environ['DISPATCHER']
+        except KeyError:
+            output_config['DEFAULT'].setdefault('dispatcher', 'file')
+        else:
+            output_config['DEFAULT']['dispatcher'] = out_type
+
+        output_config['dispatcher_file']['file_path'] = file_path
+
+        try:
+            target = os.environ['TARGET']
+        except KeyError:
+            pass
+        else:
+            k = 'dispatcher_{}'.format(output_config['DEFAULT']['dispatcher'])
+            output_config[k]['target'] = target
+
+    def _get_format_result(self, testcases):
+        criteria = self._get_task_criteria(testcases)
+
+        info = {
+            'deploy_scenario': os.environ.get('DEPLOY_SCENARIO', 'unknown'),
+            'installer': os.environ.get('INSTALLER_TYPE', 'unknown'),
+            'pod_name': os.environ.get('NODE_NAME', 'unknown'),
+            'version': os.environ.get('YARDSTICK_BRANCH', 'unknown')
+        }
+
+        result = {
+            'status': 1,
+            'result': {
+                'criteria': criteria,
+                'task_id': self.task_id,
+                'info': info,
+                'testcases': testcases
+            }
+        }
+
+        return result
+
+    def _get_task_criteria(self, testcases):
+        criteria = any(t.get('criteria') != 'PASS' for t in testcases.values())
+        if criteria:
+            return 'FAIL'
+        else:
+            return 'PASS'
+
+    def _do_output(self, output_config, result):
+
+        dispatcher = DispatcherBase.get(output_config)
+        dispatcher.flush_result_data(result)
+
     def _run(self, scenarios, run_in_parallel, output_file):
         """Deploys context and calls runners"""
         for context in self.contexts:
@@ -121,6 +200,7 @@ class Task(object):     # pragma: no cover
 
         background_runners = []
 
+        result = []
         # Start all background scenarios
         for scenario in filter(_is_background_scenario, scenarios):
             scenario["runner"] = dict(type="Duration", duration=1000000000)
@@ -136,16 +216,23 @@ class Task(object):     # pragma: no cover
 
             # Wait for runners to finish
             for runner in runners:
-                runner_join(runner)
+                status = runner_join(runner)
+                if status != 0:
+                    raise RuntimeError
                 self.outputs.update(runner.get_output())
+                result.extend(runner.get_result())
                 print("Runner ended, output in", output_file)
         else:
             # run serially
             for scenario in scenarios:
                 if not _is_background_scenario(scenario):
                     runner = self.run_one_scenario(scenario, output_file)
-                    runner_join(runner)
+                    status = runner_join(runner)
+                    if status != 0:
+                        LOG.error('Scenario: %s ERROR', scenario.get('type'))
+                        raise RuntimeError
                     self.outputs.update(runner.get_output())
+                    result.extend(runner.get_result())
                     print("Runner ended, output in", output_file)
 
         # Abort background runners
@@ -154,15 +241,21 @@ class Task(object):     # pragma: no cover
 
         # Wait for background runners to finish
         for runner in background_runners:
-            if runner.join(timeout=60) is None:
+            status = runner.join(timeout=60)
+            if status is None:
                 # Nuke if it did not stop nicely
                 base_runner.Runner.terminate(runner)
-                runner_join(runner)
+                status = runner_join(runner)
                 self.outputs.update(runner.get_output())
+                result.extend(runner.get_result())
             else:
                 base_runner.Runner.release(runner)
+            if status != 0:
+                raise RuntimeError
             print("Background task ended")
 
+        return result
+
     def atexit_handler(self):
         """handler for process termination"""
         base_runner.Runner.terminate_all()
@@ -227,7 +320,7 @@ class Task(object):     # pragma: no cover
 
         if "nodes" in scenario_cfg:
             context_cfg["nodes"] = parse_nodes_with_context(scenario_cfg)
-        runner = base_runner.Runner.get(runner_cfg, self.config)
+        runner = base_runner.Runner.get(runner_cfg)
 
         print("Starting runner of type '%s'" % runner_cfg["type"])
         runner.run(scenario_cfg, context_cfg)
@@ -492,8 +585,7 @@ def runner_join(runner):
     """join (wait for) a runner, exit process at runner failure"""
     status = runner.join()
     base_runner.Runner.release(runner)
-    if status != 0:
-        sys.exit("Runner failed")
+    return status
 
 
 def print_invalid_header(source_name, args):
diff --git a/yardstick/benchmark/runners/arithmetic.py b/yardstick/benchmark/runners/arithmetic.py
index 7ec593396..7898ae2bc 100755
--- a/yardstick/benchmark/runners/arithmetic.py
+++ b/yardstick/benchmark/runners/arithmetic.py
@@ -63,10 +63,6 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
     benchmark.setup()
     method = getattr(benchmark, method_name)
 
-    queue.put({'runner_id': runner_cfg['runner_id'],
-               'scenario_cfg': scenario_cfg,
-               'context_cfg': context_cfg})
-
     sla_action = None
     if "sla" in scenario_cfg:
         sla_action = scenario_cfg["sla"].get("action", "assert")
@@ -132,10 +128,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
             'errors': errors
         }
 
-        record = {'runner_id': runner_cfg['runner_id'],
-                  'benchmark': benchmark_output}
-
-        queue.put(record)
+        queue.put(benchmark_output)
 
         LOG.debug("runner=%(runner)s seq=%(sequence)s END",
                   {"runner": runner_cfg["runner_id"], "sequence": sequence})
diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py
index ebb9a91b5..f6816c7ed 100755
--- a/yardstick/benchmark/runners/base.py
+++ b/yardstick/benchmark/runners/base.py
@@ -22,46 +22,13 @@ import logging
 import multiprocessing
 import subprocess
 import time
-import os
 import traceback
 
-from oslo_config import cfg
-
 import yardstick.common.utils as utils
 from yardstick.benchmark.scenarios import base as base_scenario
-from yardstick.dispatcher.base import Base as DispatcherBase
 
 log = logging.getLogger(__name__)
 
-CONF = cfg.CONF
-
-
-def _output_serializer_main(filename, queue, config):
-    """entrypoint for the singleton subprocess writing to outfile
-    Use of this process enables multiple instances of a scenario without
-    messing up the output file.
-    """
-    try:
-        out_type = config['yardstick'].get('DEFAULT', {})['dispatcher']
-    except KeyError:
-        out_type = os.environ.get('DISPATCHER', 'file')
-
-    conf = {
-        'type': out_type.capitalize(),
-        'file_path': filename
-    }
-
-    dispatcher = DispatcherBase.get(conf, config)
-
-    while True:
-        # blocks until data becomes available
-        record = queue.get()
-        if record == '_TERMINATE_':
-            dispatcher.flush_result_data()
-            break
-        else:
-            dispatcher.record_result_data(record)
-
 
 def _execute_shell_command(command):
     """execute shell script with error handling"""
@@ -110,8 +77,6 @@ def _periodic_action(interval, command, queue):
 
 
 class Runner(object):
-    queue = None
-    dump_process = None
     runners = []
 
     @staticmethod
@@ -131,30 +96,10 @@ class Runner(object):
         return types
 
     @staticmethod
-    def get(runner_cfg, config):
+    def get(runner_cfg):
         """Returns instance of a scenario runner for execution type.
         """
-        # if there is no runner, start the output serializer subprocess
-        if not Runner.runners:
-            log.debug("Starting dump process file '%s'",
-                      runner_cfg["output_filename"])
-            Runner.queue = multiprocessing.Queue()
-            Runner.dump_process = multiprocessing.Process(
-                target=_output_serializer_main,
-                name="Dumper",
-                args=(runner_cfg["output_filename"], Runner.queue, config))
-            Runner.dump_process.start()
-
-        return Runner.get_cls(runner_cfg["type"])(runner_cfg, Runner.queue)
-
-    @staticmethod
-    def release_dump_process():
-        """Release the dumper process"""
-        log.debug("Stopping dump process")
-        if Runner.dump_process:
-            Runner.queue.put('_TERMINATE_')
-            Runner.dump_process.join()
-            Runner.dump_process = None
+        return Runner.get_cls(runner_cfg["type"])(runner_cfg)
 
     @staticmethod
     def release(runner):
@@ -162,10 +107,6 @@ class Runner(object):
         if runner in Runner.runners:
             Runner.runners.remove(runner)
 
-        # if this was the last runner, stop the output serializer subprocess
-        if not Runner.runners:
-            Runner.release_dump_process()
-
     @staticmethod
     def terminate(runner):
         """Terminate the runner"""
@@ -179,7 +120,6 @@ class Runner(object):
 
         # release dumper process as some errors before any runner is created
         if not Runner.runners:
-            Runner.release_dump_process()
             return
 
         for runner in Runner.runners:
@@ -193,11 +133,11 @@ class Runner(object):
                 runner.periodic_action_process = None
             Runner.release(runner)
 
-    def __init__(self, config, queue):
+    def __init__(self, config):
         self.config = config
         self.periodic_action_process = None
-        self.result_queue = queue
         self.output_queue = multiprocessing.Queue()
+        self.result_queue = multiprocessing.Queue()
         self.process = None
         self.aborted = multiprocessing.Event()
         Runner.runners.append(self)
@@ -276,3 +216,9 @@ class Runner(object):
         while not self.output_queue.empty():
             result.update(self.output_queue.get())
         return result
+
+    def get_result(self):
+        result = []
+        while not self.result_queue.empty():
+            result.append(self.result_queue.get())
+        return result
diff --git a/yardstick/benchmark/runners/duration.py b/yardstick/benchmark/runners/duration.py
index 2bf2cd2fe..69d744562 100644
--- a/yardstick/benchmark/runners/duration.py
+++ b/yardstick/benchmark/runners/duration.py
@@ -52,10 +52,6 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
     if "sla" in scenario_cfg:
         sla_action = scenario_cfg["sla"].get("action", "assert")
 
-    queue.put({'runner_id': runner_cfg['runner_id'],
-               'scenario_cfg': scenario_cfg,
-               'context_cfg': context_cfg})
-
     start = time.time()
     while True:
 
@@ -90,10 +86,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
             'errors': errors
         }
 
-        record = {'runner_id': runner_cfg['runner_id'],
-                  'benchmark': benchmark_output}
-
-        queue.put(record)
+        queue.put(benchmark_output)
 
         LOG.debug("runner=%(runner)s seq=%(sequence)s END",
                   {"runner": runner_cfg["runner_id"], "sequence": sequence})
diff --git a/yardstick/benchmark/runners/iteration.py b/yardstick/benchmark/runners/iteration.py
index 973bb9ac4..50fe106bd 100644
--- a/yardstick/benchmark/runners/iteration.py
+++ b/yardstick/benchmark/runners/iteration.py
@@ -53,10 +53,6 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
 
     method = getattr(benchmark, method_name)
 
-    queue.put({'runner_id': runner_cfg['runner_id'],
-               'scenario_cfg': scenario_cfg,
-               'context_cfg': context_cfg})
-
     sla_action = None
     if "sla" in scenario_cfg:
         sla_action = scenario_cfg["sla"].get("action", "assert")
@@ -105,10 +101,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
                 'errors': errors
             }
 
-            record = {'runner_id': runner_cfg['runner_id'],
-                      'benchmark': benchmark_output}
-
-            queue.put(record)
+            queue.put(benchmark_output)
 
             LOG.debug("runner=%(runner)s seq=%(sequence)s END",
                       {"runner": runner_cfg["runner_id"],
diff --git a/yardstick/benchmark/runners/sequence.py b/yardstick/benchmark/runners/sequence.py
index 74ff82204..68e272c57 100644
--- a/yardstick/benchmark/runners/sequence.py
+++ b/yardstick/benchmark/runners/sequence.py
@@ -57,10 +57,6 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
     benchmark.setup()
     method = getattr(benchmark, method_name)
 
-    queue.put({'runner_id': runner_cfg['runner_id'],
-               'scenario_cfg': scenario_cfg,
-               'context_cfg': context_cfg})
-
     sla_action = None
     if "sla" in scenario_cfg:
         sla_action = scenario_cfg["sla"].get("action", "assert")
@@ -99,10 +95,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
             'errors': errors
         }
 
-        record = {'runner_id': runner_cfg['runner_id'],
-                  'benchmark': benchmark_output}
-
-        queue.put(record)
+        queue.put(benchmark_output)
 
         LOG.debug("runner=%(runner)s seq=%(sequence)s END",
                   {"runner": runner_cfg["runner_id"], "sequence": sequence})
diff --git a/yardstick/cmd/commands/task.py b/yardstick/cmd/commands/task.py
index 16a4db291..6384e6eb1 100644
--- a/yardstick/cmd/commands/task.py
+++ b/yardstick/cmd/commands/task.py
@@ -14,7 +14,6 @@ from __future__ import absolute_import
 from yardstick.benchmark.core.task import Task
 from yardstick.common.utils import cliargs
 from yardstick.common.utils import write_json_to_file
-from yardstick.common.utils import read_json_from_file
 from yardstick.cmd.commands import change_osloobj_to_paras
 
 output_file_default = "/tmp/yardstick.out"
@@ -46,22 +45,11 @@ class TaskCommands(object):
         param = change_osloobj_to_paras(args)
         self.output_file = param.output_file
 
-        self._init_result_file()
-
         try:
             Task().start(param, **kwargs)
-            self._finish()
         except Exception as e:
             self._write_error_data(e)
-
-    def _init_result_file(self):
-        data = {'status': 0, 'result': []}
-        write_json_to_file(self.output_file, data)
-
-    def _finish(self):
-        result = read_json_from_file(self.output_file).get('result')
-        data = {'status': 1, 'result': result}
-        write_json_to_file(self.output_file, data)
+            raise
 
     def _write_error_data(self, error):
         data = {'status': 2, 'result': str(error)}
diff --git a/yardstick/dispatcher/base.py b/yardstick/dispatcher/base.py
index a1c858297..e77249c54 100644
--- a/yardstick/dispatcher/base.py
+++ b/yardstick/dispatcher/base.py
@@ -38,15 +38,13 @@ class Base(object):
         raise RuntimeError("No such dispatcher_type %s" % dispatcher_type)
 
     @staticmethod
-    def get(conf, config):
+    def get(config):
         """Returns instance of a dispatcher for dispatcher type.
         """
-        return Base.get_cls(conf["type"])(conf, config)
+        out_type = config['DEFAULT']['dispatcher']
 
-    @abc.abstractmethod
-    def record_result_data(self, data):
-        """Recording result data interface."""
+        return Base.get_cls(out_type.capitalize())(config)
 
     @abc.abstractmethod
-    def flush_result_data(self):
+    def flush_result_data(self, data):
         """Flush result data into permanent storage media interface."""
diff --git a/yardstick/dispatcher/file.py b/yardstick/dispatcher/file.py
index 8acd5dfbb..24fc22dd4 100644
--- a/yardstick/dispatcher/file.py
+++ b/yardstick/dispatcher/file.py
@@ -29,18 +29,10 @@ class FileDispatcher(DispatchBase):
 
     __dispatcher_type__ = "File"
 
-    def __init__(self, conf, config):
+    def __init__(self, conf):
         super(FileDispatcher, self).__init__(conf)
-        self.result = []
+        self.target = conf['dispatcher_file'].get('file_path',
+                                                  consts.DEFAULT_OUTPUT_FILE)
 
-    def record_result_data(self, data):
-        self.result.append(data)
-
-    def flush_result_data(self):
-        file_path = self.conf.get('file_path', consts.DEFAULT_OUTPUT_FILE)
-
-        res = utils.read_json_from_file(file_path).get('result')
-        res.extend(self.result)
-
-        data = {'status': 0, 'result': res}
-        utils.write_json_to_file(file_path, data)
+    def flush_result_data(self, data):
+        utils.write_json_to_file(self.target, data)
diff --git a/yardstick/dispatcher/http.py b/yardstick/dispatcher/http.py
index 0d8d2a346..9bf9af33b 100644
--- a/yardstick/dispatcher/http.py
+++ b/yardstick/dispatcher/http.py
@@ -20,30 +20,15 @@ from __future__ import absolute_import
 
 import logging
 import os
+from datetime import datetime
 
 from oslo_serialization import jsonutils
 import requests
-from oslo_config import cfg
 
 from yardstick.dispatcher.base import Base as DispatchBase
 
 LOG = logging.getLogger(__name__)
 
-CONF = cfg.CONF
-http_dispatcher_opts = [
-    cfg.StrOpt('target',
-               default=os.getenv('TARGET', 'http://127.0.0.1:8000/results'),
-               help='The target where the http request will be sent. '
-                    'If this is not set, no data will be posted. For '
-                    'example: target = http://hostname:1234/path'),
-    cfg.IntOpt('timeout',
-               default=5,
-               help='The max time in seconds to wait for a request to '
-                    'timeout.'),
-]
-
-CONF.register_opts(http_dispatcher_opts, group="dispatcher_http")
-
 
 class HttpDispatcher(DispatchBase):
     """Dispatcher class for posting data into a http target.
@@ -51,55 +36,61 @@ class HttpDispatcher(DispatchBase):
 
     __dispatcher_type__ = "Http"
 
-    def __init__(self, conf, config):
+    def __init__(self, conf):
         super(HttpDispatcher, self).__init__(conf)
+        http_conf = conf['dispatcher_http']
         self.headers = {'Content-type': 'application/json'}
-        self.timeout = CONF.dispatcher_http.timeout
-        self.target = CONF.dispatcher_http.target
-        self.raw_result = []
-        self.result = {
-            "project_name": "yardstick",
-            "description": "yardstick test cases result",
-            "pod_name": os.environ.get('NODE_NAME', 'unknown'),
-            "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
-            "version": os.environ.get('YARDSTICK_VERSION', 'unknown'),
-            "build_tag": os.environ.get('BUILD_TAG')
-        }
-
-    def record_result_data(self, data):
-        self.raw_result.append(data)
+        self.timeout = int(http_conf.get('timeout', 5))
+        self.target = http_conf.get('target', 'http://127.0.0.1:8000/results')
 
-    def flush_result_data(self):
+    def flush_result_data(self, data):
         if self.target == '':
             # if the target was not set, do not do anything
             LOG.error('Dispatcher target was not set, no data will'
                       'be posted.')
             return
 
-        self.result["details"] = {'results': self.raw_result}
-
-        case_name = ""
-        for v in self.raw_result:
-            if isinstance(v, dict) and "scenario_cfg" in v:
-                case_name = v["scenario_cfg"]["tc"]
-                break
-        if case_name == "":
-            LOG.error('Test result : %s',
-                      jsonutils.dump_as_bytes(self.result))
-            LOG.error('The case_name cannot be found, no data will be posted.')
-            return
+        result = data['result']
+        self.info = result['info']
+        self.task_id = result['task_id']
+        self.criteria = result['criteria']
+        testcases = result['testcases']
+
+        for case, data in testcases.items():
+            self._upload_case_result(case, data)
 
-        self.result["case_name"] = case_name
+    def _upload_case_result(self, case, data):
+        try:
+            scenario_data = data.get('tc_data', [])[0]
+        except IndexError:
+            current_time = datetime.now()
+        else:
+            timestamp = float(scenario_data.get('timestamp', 0.0))
+            current_time = datetime.fromtimestamp(timestamp)
+
+        result = {
+            "project_name": "yardstick",
+            "case_name": case,
+            "description": "yardstick ci scenario status",
+            "scenario": self.info.get('deploy_scenario'),
+            "version": self.info.get('version'),
+            "pod_name": self.info.get('pod_name'),
+            "installer": self.info.get('installer'),
+            "build_tag": os.environ.get('BUILD_TAG'),
+            "criteria": data.get('criteria'),
+            "start_date": current_time.strftime('%Y-%m-%d %H:%M:%S'),
+            "stop_date": current_time.strftime('%Y-%m-%d %H:%M:%S'),
+            "trust_indicator": "",
+            "details": ""
+        }
 
         try:
-            LOG.debug('Test result : %s',
-                      jsonutils.dump_as_bytes(self.result))
+            LOG.debug('Test result : %s', result)
             res = requests.post(self.target,
-                                data=jsonutils.dump_as_bytes(self.result),
+                                data=jsonutils.dump_as_bytes(result),
                                 headers=self.headers,
                                 timeout=self.timeout)
             LOG.debug('Test result posting finished with status code'
                       ' %d.' % res.status_code)
         except Exception as err:
-            LOG.exception('Failed to record result data: %s',
-                          err)
+            LOG.exception('Failed to record result data: %s', err)
diff --git a/yardstick/dispatcher/influxdb.py b/yardstick/dispatcher/influxdb.py
index 53af79c71..373aae13a 100644
--- a/yardstick/dispatcher/influxdb.py
+++ b/yardstick/dispatcher/influxdb.py
@@ -10,13 +10,11 @@
 from __future__ import absolute_import
 
 import logging
-import os
 import time
 
 import collections
 import requests
 import six
-from oslo_serialization import jsonutils
 
 from third_party.influxdb.influxdb_line_protocol import make_lines
 from yardstick.dispatcher.base import Base as DispatchBase
@@ -30,28 +28,66 @@ class InfluxdbDispatcher(DispatchBase):
 
     __dispatcher_type__ = "Influxdb"
 
-    def __init__(self, conf, config):
+    def __init__(self, conf):
         super(InfluxdbDispatcher, self).__init__(conf)
-        db_conf = config['yardstick'].get('dispatcher_influxdb', {})
+        db_conf = conf['dispatcher_influxdb']
         self.timeout = int(db_conf.get('timeout', 5))
         self.target = db_conf.get('target', 'http://127.0.0.1:8086')
         self.db_name = db_conf.get('db_name', 'yardstick')
         self.username = db_conf.get('username', 'root')
         self.password = db_conf.get('password', 'root')
+
         self.influxdb_url = "%s/write?db=%s" % (self.target, self.db_name)
-        self.raw_result = []
-        self.case_name = ""
-        self.tc = ""
+
         self.task_id = -1
-        self.runners_info = {}
-        self.static_tags = {
-            "pod_name": os.environ.get('NODE_NAME', 'unknown'),
-            "installer": os.environ.get('INSTALLER_TYPE', 'unknown'),
-            "deploy_scenario": os.environ.get('DEPLOY_SCENARIO', 'unknown'),
-            "version": os.path.basename(os.environ.get('YARDSTICK_BRANCH',
-                                                       'unknown'))
 
+    def flush_result_data(self, data):
+        LOG.debug('Test result all : %s', data)
+        if self.target == '':
+            # if the target was not set, do not do anything
+            LOG.error('Dispatcher target was not set, no data will be posted.')
+
+        result = data['result']
+        self.tags = result['info']
+        self.task_id = result['task_id']
+        self.criteria = result['criteria']
+        testcases = result['testcases']
+
+        for case, data in testcases.items():
+            tc_criteria = data['criteria']
+            for record in data['tc_data']:
+                self._upload_one_record(record, case, tc_criteria)
+
+        return 0
+
+    def _upload_one_record(self, data, case, tc_criteria):
+        try:
+            line = self._data_to_line_protocol(data, case, tc_criteria)
+            LOG.debug('Test result line format : %s', line)
+            res = requests.post(self.influxdb_url,
+                                data=line,
+                                auth=(self.username, self.password),
+                                timeout=self.timeout)
+            if res.status_code != 204:
+                LOG.error('Test result posting finished with status code'
+                          ' %d.', res.status_code)
+                LOG.error(res.text)
+
+        except Exception as err:
+            LOG.exception('Failed to record result data: %s', err)
+
+    def _data_to_line_protocol(self, data, case, criteria):
+        msg = {}
+        point = {
+            "measurement": case,
+            "fields": self._dict_key_flatten(data["data"]),
+            "time": self._get_nano_timestamp(data),
+            "tags": self._get_extended_tags(criteria),
         }
+        msg["points"] = [point]
+        msg["tags"] = self.tags
+
+        return make_lines(msg).encode('utf-8')
 
     def _dict_key_flatten(self, data):
         next_data = {}
@@ -76,84 +112,16 @@ class InfluxdbDispatcher(DispatchBase):
 
     def _get_nano_timestamp(self, results):
         try:
-            timestamp = results["benchmark"]["timestamp"]
+            timestamp = results["timestamp"]
         except Exception:
             timestamp = time.time()
 
         return str(int(float(timestamp) * 1000000000))
 
-    def _get_extended_tags(self, data):
-        runner_info = self.runners_info[data["runner_id"]]
+    def _get_extended_tags(self, criteria):
         tags = {
-            "runner_id": data["runner_id"],
             "task_id": self.task_id,
-            "scenarios": runner_info["scenarios"]
+            "criteria": criteria
         }
-        if "host" in runner_info:
-            tags["host"] = runner_info["host"]
-        if "target" in runner_info:
-            tags["target"] = runner_info["target"]
 
         return tags
-
-    def _data_to_line_protocol(self, data):
-        msg = {}
-        point = {
-            "measurement": self.tc,
-            "fields": self._dict_key_flatten(data["benchmark"]["data"]),
-            "time": self._get_nano_timestamp(data),
-            "tags": self._get_extended_tags(data),
-        }
-        msg["points"] = [point]
-        msg["tags"] = self.static_tags
-
-        return make_lines(msg).encode('utf-8')
-
-    def record_result_data(self, data):
-        LOG.debug('Test result : %s', jsonutils.dump_as_bytes(data))
-        self.raw_result.append(data)
-        if self.target == '':
-            # if the target was not set, do not do anything
-            LOG.error('Dispatcher target was not set, no data will'
-                      'be posted.')
-            return -1
-
-        if isinstance(data, dict) and "scenario_cfg" in data:
-            self.tc = data["scenario_cfg"]["tc"]
-            self.task_id = data["scenario_cfg"]["task_id"]
-            scenario_cfg = data["scenario_cfg"]
-            runner_id = data["runner_id"]
-            self.runners_info[runner_id] = {"scenarios": scenario_cfg["type"]}
-            if "host" in scenario_cfg:
-                self.runners_info[runner_id]["host"] = scenario_cfg["host"]
-            if "target" in scenario_cfg:
-                self.runners_info[runner_id]["target"] = scenario_cfg["target"]
-            return 0
-
-        if self.tc == "":
-            LOG.error('Test result : %s', jsonutils.dump_as_bytes(data))
-            LOG.error('The case_name cannot be found, no data will be posted.')
-            return -1
-
-        try:
-            line = self._data_to_line_protocol(data)
-            LOG.debug('Test result line format : %s', line)
-            res = requests.post(self.influxdb_url,
-                                data=line,
-                                auth=(self.username, self.password),
-                                timeout=self.timeout)
-            if res.status_code != 204:
-                LOG.error('Test result posting finished with status code'
-                          ' %d.', res.status_code)
-                LOG.error(res.text)
-
-        except Exception as err:
-            LOG.exception('Failed to record result data: %s',
-                          err)
-            return -1
-        return 0
-
-    def flush_result_data(self):
-        LOG.debug('Test result all : %s',
-                  jsonutils.dump_as_bytes(self.raw_result))
-        return 0