1 ##############################################################################
2 # Copyright (c) 2015 Ericsson AB and others.
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
12 import multiprocessing
17 log = logging.getLogger(__name__)
19 from oslo_config import cfg
21 import yardstick.common.utils as utils
22 from yardstick.benchmark.scenarios import base as base_scenario
23 from yardstick.dispatcher.base import Base as DispatcherBase
28 def _output_serializer_main(filename, queue):
29 '''entrypoint for the singleton subprocess writing to outfile
30 Use of this process enables multiple instances of a scenario without
31 messing up the output file.
34 config["type"] = CONF.dispatcher.capitalize()
35 config["file_path"] = filename
36 dispatcher = DispatcherBase.get(config)
39 # blocks until data becomes available
41 if record == '_TERMINATE_':
42 dispatcher.flush_result_data()
45 dispatcher.record_result_data(record)
48 def _execute_shell_command(command):
49 '''execute shell script with error handling'''
53 output = subprocess.check_output(command, shell=True)
56 output = traceback.format_exc()
57 log.error("exec command '%s' error:\n " % command)
58 log.error(traceback.format_exc())
60 return exitcode, output
63 def _single_action(seconds, command, queue):
64 '''entrypoint for the single action process'''
65 log.debug("single action, fires after %d seconds (from now)", seconds)
67 log.debug("single action: executing command: '%s'", command)
68 ret_code, data = _execute_shell_command(command)
70 log.error("single action error! command:%s" % command)
71 queue.put({'single-action-data': data})
73 log.debug("single action data: \n%s" % data)
74 queue.put({'single-action-data': data})
77 def _periodic_action(interval, command, queue):
78 '''entrypoint for the periodic action process'''
79 log.debug("periodic action, fires every: %d seconds", interval)
83 time_spent += interval
84 log.debug("periodic action, executing command: '%s'", command)
85 ret_code, data = _execute_shell_command(command)
87 log.error("periodic action error! command:%s", command)
88 queue.put({'periodic-action-data': data})
90 log.debug("periodic action data: \n%s" % data)
91 queue.put({'periodic-action-data': data})
100 def get_cls(runner_type):
101 '''return class of specified type'''
102 for runner in utils.itersubclasses(Runner):
103 if runner_type == runner.__execution_type__:
105 raise RuntimeError("No such runner_type %s" % runner_type)
109 '''return a list of known runner type (class) names'''
111 for runner in utils.itersubclasses(Runner):
117 """Returns instance of a scenario runner for execution type.
119 # if there is no runner, start the output serializer subprocess
120 if len(Runner.runners) == 0:
121 log.debug("Starting dump process file '%s'" %
122 config["output_filename"])
123 Runner.queue = multiprocessing.Queue()
124 Runner.dump_process = multiprocessing.Process(
125 target=_output_serializer_main,
127 args=(config["output_filename"], Runner.queue))
128 Runner.dump_process.start()
130 return Runner.get_cls(config["type"])(config, Runner.queue)
133 def release_dump_process():
134 '''Release the dumper process'''
135 log.debug("Stopping dump process")
136 if Runner.dump_process:
137 Runner.queue.put('_TERMINATE_')
138 Runner.dump_process.join()
139 Runner.dump_process = None
143 '''Release the runner'''
144 Runner.runners.remove(runner)
146 # if this was the last runner, stop the output serializer subprocess
147 if len(Runner.runners) == 0:
148 Runner.release_dump_process()
152 '''Terminate all runners (subprocesses)'''
153 log.debug("Terminating all runners")
155 # release dumper process as some errors before any runner is created
156 if len(Runner.runners) == 0:
157 Runner.release_dump_process()
160 for runner in Runner.runners:
161 log.debug("Terminating runner: %s", runner)
163 runner.process.terminate()
164 runner.process.join()
165 if runner.periodic_action_process:
166 log.debug("Terminating periodic action process")
167 runner.periodic_action_process.terminate()
168 runner.periodic_action_process = None
169 Runner.release(runner)
171 def __init__(self, config, queue):
173 self.periodic_action_process = None
174 self.result_queue = queue
176 Runner.runners.append(self)
178 def run_post_stop_action(self):
179 '''run a potentially configured post-stop action'''
180 if "post-stop-action" in self.config:
181 command = self.config["post-stop-action"]["command"]
182 log.debug("post stop action: command: '%s'" % command)
183 ret_code, data = _execute_shell_command(command)
185 log.error("post action error! command:%s", command)
186 self.result_queue.put({'post-stop-action-data': data})
188 log.debug("post-stop data: \n%s" % data)
189 self.result_queue.put({'post-stop-action-data': data})
191 def run(self, scenario_cfg, context_cfg):
192 scenario_type = scenario_cfg["type"]
193 class_name = base_scenario.Scenario.get(scenario_type)
194 path_split = class_name.split(".")
195 module_path = ".".join(path_split[:-1])
196 module = importlib.import_module(module_path)
197 cls = getattr(module, path_split[-1])
199 self.config['object'] = class_name
201 # run a potentially configured pre-start action
202 if "pre-start-action" in self.config:
203 command = self.config["pre-start-action"]["command"]
204 log.debug("pre start action: command: '%s'" % command)
205 ret_code, data = _execute_shell_command(command)
207 log.error("pre-start action error! command:%s", command)
208 self.result_queue.put({'pre-start-action-data': data})
210 log.debug("pre-start data: \n%s" % data)
211 self.result_queue.put({'pre-start-action-data': data})
213 if "single-shot-action" in self.config:
214 single_action_process = multiprocessing.Process(
215 target=_single_action,
216 name="single-shot-action",
217 args=(self.config["single-shot-action"]["after"],
218 self.config["single-shot-action"]["command"],
220 single_action_process.start()
222 if "periodic-action" in self.config:
223 self.periodic_action_process = multiprocessing.Process(
224 target=_periodic_action,
225 name="periodic-action",
226 args=(self.config["periodic-action"]["interval"],
227 self.config["periodic-action"]["command"],
229 self.periodic_action_process.start()
231 self._run_benchmark(cls, "run", scenario_cfg, context_cfg)
235 if self.periodic_action_process:
236 self.periodic_action_process.terminate()
237 self.periodic_action_process = None
239 self.run_post_stop_action()
240 return self.process.exitcode