Merge "Add new scenario NSPerf-RFC2544"

2 files changed, 41 insertions, 23 deletions

diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py
index 3878f20aa..94de45d1e 100755
--- a/yardstick/benchmark/runners/base.py
+++ b/yardstick/benchmark/runners/base.py
@@ -77,6 +77,33 @@ def _periodic_action(interval, command, queue):
         queue.put({'periodic-action-data': data})
 
 
+class ScenarioOutput(dict):
+
+    QUEUE_PUT_TIMEOUT = 10
+
+    def __init__(self, queue, **kwargs):
+        super(ScenarioOutput, self).__init__()
+        self._queue = queue
+        self.result_ext = dict()
+        for key, val in kwargs.items():
+            self.result_ext[key] = val
+            setattr(self, key, val)
+
+    def push(self, data=None, add_timestamp=True):
+        if data is None:
+            data = dict(self)
+
+        if add_timestamp:
+            result = {'timestamp': time.time(), 'data': data}
+        else:
+            result = data
+
+        for key in self.result_ext.keys():
+            result[key] = getattr(self, key)
+
+        self._queue.put(result, True, self.QUEUE_PUT_TIMEOUT)
+
+
 class Runner(object):
     runners = []
 
diff --git a/yardstick/benchmark/runners/iteration.py b/yardstick/benchmark/runners/iteration.py
index 58ab06a32..15dad2cd5 100644
--- a/yardstick/benchmark/runners/iteration.py
+++ b/yardstick/benchmark/runners/iteration.py
@@ -23,7 +23,6 @@ from __future__ import absolute_import
 
 import logging
 import multiprocessing
-import time
 import traceback
 
 import os
@@ -40,8 +39,6 @@ QUEUE_PUT_TIMEOUT = 10
 def _worker_process(queue, cls, method_name, scenario_cfg,
                     context_cfg, aborted, output_queue):
 
-    sequence = 1
-
     runner_cfg = scenario_cfg['runner']
 
     interval = runner_cfg.get("interval", 1)
@@ -53,6 +50,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
 
     runner_cfg['runner_id'] = os.getpid()
 
+    scenario_output = base.ScenarioOutput(queue, sequence=1, errors="")
     benchmark = cls(scenario_cfg, context_cfg)
     if "setup" in run_step:
         benchmark.setup()
@@ -67,22 +65,21 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
 
             LOG.debug("runner=%(runner)s seq=%(sequence)s START",
                       {"runner": runner_cfg["runner_id"],
-                       "sequence": sequence})
-
-            data = {}
-            errors = ""
+                       "sequence": scenario_output.sequence})
 
+            scenario_output.clear()
+            scenario_output.errors = ""
             benchmark.pre_run_wait_time(interval)
 
             try:
-                result = method(data)
+                result = method(scenario_output)
             except y_exc.SLAValidationError as error:
                 # SLA validation failed in scenario, determine what to do now
                 if sla_action == "assert":
                     raise
                 elif sla_action == "monitor":
                     LOG.warning("SLA validation failed: %s", error.args)
-                    errors = error.args
+                    scenario_output.errors = error.args
                 elif sla_action == "rate-control":
                     try:
                         scenario_cfg['options']['rate']
@@ -91,10 +88,10 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
                         scenario_cfg['options']['rate'] = 100
 
                     scenario_cfg['options']['rate'] -= delta
-                    sequence = 1
+                    scenario_output.sequence = 1
                     continue
             except Exception:  # pylint: disable=broad-except
-                errors = traceback.format_exc()
+                scenario_output.errors = traceback.format_exc()
                 LOG.exception("")
                 raise
             else:
@@ -105,23 +102,17 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
 
             benchmark.post_run_wait_time(interval)
 
-            benchmark_output = {
-                'timestamp': time.time(),
-                'sequence': sequence,
-                'data': data,
-                'errors': errors
-            }
-
-            queue.put(benchmark_output, True, QUEUE_PUT_TIMEOUT)
+            if scenario_output:
+                scenario_output.push()
 
             LOG.debug("runner=%(runner)s seq=%(sequence)s END",
                       {"runner": runner_cfg["runner_id"],
-                       "sequence": sequence})
+                       "sequence": scenario_output.sequence})
 
-            sequence += 1
+            scenario_output.sequence += 1
 
-            if (errors and sla_action is None) or \
-                    (sequence > iterations or aborted.is_set()):
+            if (scenario_output.errors and sla_action is None) or \
+                    (scenario_output.sequence > iterations or aborted.is_set()):
                 LOG.info("worker END")
                 break
     if "teardown" in run_step: