Merge "Pass parameters between scenarios"

6 files changed, 48 insertions, 12 deletions

diff --git a/yardstick/benchmark/core/task.py b/yardstick/benchmark/core/task.py
index 3a151dbba..c44081b73 100644
--- a/yardstick/benchmark/core/task.py
+++ b/yardstick/benchmark/core/task.py
@@ -44,6 +44,7 @@ class Task(object):     # pragma: no cover
     def __init__(self):
         self.config = {}
         self.contexts = []
+        self.outputs = {}
 
     def start(self, args, **kwargs):
         """Start a benchmark scenario."""
@@ -136,6 +137,7 @@ class Task(object):     # pragma: no cover
             # Wait for runners to finish
             for runner in runners:
                 runner_join(runner)
+                self.outputs.update(runner.get_output())
                 print("Runner ended, output in", output_file)
         else:
             # run serially
@@ -143,6 +145,7 @@ class Task(object):     # pragma: no cover
                 if not _is_background_scenario(scenario):
                     runner = self.run_one_scenario(scenario, output_file)
                     runner_join(runner)
+                    self.outputs.update(runner.get_output())
                     print("Runner ended, output in", output_file)
 
         # Abort background runners
@@ -155,6 +158,7 @@ class Task(object):     # pragma: no cover
                 # Nuke if it did not stop nicely
                 base_runner.Runner.terminate(runner)
                 runner_join(runner)
+                self.outputs.update(runner.get_output())
             else:
                 base_runner.Runner.release(runner)
             print("Background task ended")
@@ -168,11 +172,24 @@ class Task(object):     # pragma: no cover
             for context in self.contexts[::-1]:
                 context.undeploy()
 
+    def _parse_options(self, op):
+        if isinstance(op, dict):
+            return {k: self._parse_options(v) for k, v in op.items()}
+        elif isinstance(op, list):
+            return [self._parse_options(v) for v in op]
+        elif isinstance(op, str):
+            return self.outputs.get(op[1:]) if op.startswith('$') else op
+        else:
+            return op
+
     def run_one_scenario(self, scenario_cfg, output_file):
         """run one scenario using context"""
         runner_cfg = scenario_cfg["runner"]
         runner_cfg['output_filename'] = output_file
 
+        options = scenario_cfg.get('options', {})
+        scenario_cfg['options'] = self._parse_options(options)
+
         # TODO support get multi hosts/vms info
         context_cfg = {}
         if "host" in scenario_cfg:
diff --git a/yardstick/benchmark/runners/arithmetic.py b/yardstick/benchmark/runners/arithmetic.py
index 65fdb9d66..7ec593396 100755
--- a/yardstick/benchmark/runners/arithmetic.py
+++ b/yardstick/benchmark/runners/arithmetic.py
@@ -42,7 +42,7 @@ LOG = logging.getLogger(__name__)
 
 
 def _worker_process(queue, cls, method_name, scenario_cfg,
-                    context_cfg, aborted):
+                    context_cfg, aborted, output_queue):
 
     sequence = 1
 
@@ -108,7 +108,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         errors = ""
 
         try:
-            method(data)
+            result = method(data)
         except AssertionError as assertion:
             # SLA validation failed in scenario, determine what to do now
             if sla_action == "assert":
@@ -119,6 +119,9 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         except Exception as e:
             errors = traceback.format_exc()
             LOG.exception(e)
+        else:
+            if result:
+                output_queue.put(result)
 
         time.sleep(interval)
 
@@ -188,5 +191,5 @@ class ArithmeticRunner(base.Runner):
         self.process = multiprocessing.Process(
             target=_worker_process,
             args=(self.result_queue, cls, method, scenario_cfg,
-                  context_cfg, self.aborted))
+                  context_cfg, self.aborted, self.output_queue))
         self.process.start()
diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py
index b48ed973a..ebb9a91b5 100755
--- a/yardstick/benchmark/runners/base.py
+++ b/yardstick/benchmark/runners/base.py
@@ -197,6 +197,7 @@ class Runner(object):
         self.config = config
         self.periodic_action_process = None
         self.result_queue = queue
+        self.output_queue = multiprocessing.Queue()
         self.process = None
         self.aborted = multiprocessing.Event()
         Runner.runners.append(self)
@@ -269,3 +270,9 @@ class Runner(object):
 
         self.run_post_stop_action()
         return self.process.exitcode
+
+    def get_output(self):
+        result = {}
+        while not self.output_queue.empty():
+            result.update(self.output_queue.get())
+        return result
diff --git a/yardstick/benchmark/runners/duration.py b/yardstick/benchmark/runners/duration.py
index 772983cfd..2bf2cd2fe 100644
--- a/yardstick/benchmark/runners/duration.py
+++ b/yardstick/benchmark/runners/duration.py
@@ -32,7 +32,7 @@ LOG = logging.getLogger(__name__)
 
 
 def _worker_process(queue, cls, method_name, scenario_cfg,
-                    context_cfg, aborted):
+                    context_cfg, aborted, output_queue):
 
     sequence = 1
 
@@ -66,7 +66,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         errors = ""
 
         try:
-            method(data)
+            result = method(data)
         except AssertionError as assertion:
             # SLA validation failed in scenario, determine what to do now
             if sla_action == "assert":
@@ -77,6 +77,9 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         except Exception as e:
             errors = traceback.format_exc()
             LOG.exception(e)
+        else:
+            if result:
+                output_queue.put(result)
 
         time.sleep(interval)
 
@@ -126,5 +129,5 @@ If the scenario ends before the time has elapsed, it will be started again.
         self.process = multiprocessing.Process(
             target=_worker_process,
             args=(self.result_queue, cls, method, scenario_cfg,
-                  context_cfg, self.aborted))
+                  context_cfg, self.aborted, self.output_queue))
         self.process.start()
diff --git a/yardstick/benchmark/runners/iteration.py b/yardstick/benchmark/runners/iteration.py
index 3963de871..973bb9ac4 100644
--- a/yardstick/benchmark/runners/iteration.py
+++ b/yardstick/benchmark/runners/iteration.py
@@ -32,7 +32,7 @@ LOG = logging.getLogger(__name__)
 
 
 def _worker_process(queue, cls, method_name, scenario_cfg,
-                    context_cfg, aborted):
+                    context_cfg, aborted, output_queue):
 
     sequence = 1
 
@@ -71,7 +71,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
             errors = ""
 
             try:
-                method(data)
+                result = method(data)
             except AssertionError as assertion:
                 # SLA validation failed in scenario, determine what to do now
                 if sla_action == "assert":
@@ -92,6 +92,9 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
             except Exception as e:
                 errors = traceback.format_exc()
                 LOG.exception(e)
+            else:
+                if result:
+                    output_queue.put(result)
 
             time.sleep(interval)
 
@@ -142,5 +145,5 @@ If the scenario ends before the time has elapsed, it will be started again.
         self.process = multiprocessing.Process(
             target=_worker_process,
             args=(self.result_queue, cls, method, scenario_cfg,
-                  context_cfg, self.aborted))
+                  context_cfg, self.aborted, self.output_queue))
         self.process.start()
diff --git a/yardstick/benchmark/runners/sequence.py b/yardstick/benchmark/runners/sequence.py
index af87c006e..74ff82204 100644
--- a/yardstick/benchmark/runners/sequence.py
+++ b/yardstick/benchmark/runners/sequence.py
@@ -33,7 +33,7 @@ LOG = logging.getLogger(__name__)
 
 
 def _worker_process(queue, cls, method_name, scenario_cfg,
-                    context_cfg, aborted):
+                    context_cfg, aborted, output_queue):
 
     sequence = 1
 
@@ -75,7 +75,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         errors = ""
 
         try:
-            method(data)
+            result = method(data)
         except AssertionError as assertion:
             # SLA validation failed in scenario, determine what to do now
             if sla_action == "assert":
@@ -86,6 +86,9 @@ def _worker_process(queue, cls, method_name, scenario_cfg,
         except Exception as e:
             errors = traceback.format_exc()
             LOG.exception(e)
+        else:
+            if result:
+                output_queue.put(result)
 
         time.sleep(interval)
 
@@ -137,5 +140,5 @@ class SequenceRunner(base.Runner):
         self.process = multiprocessing.Process(
             target=_worker_process,
             args=(self.result_queue, cls, method, scenario_cfg,
-                  context_cfg, self.aborted))
+                  context_cfg, self.aborted, self.output_queue))
         self.process.start()