yardstick/benchmark/runners/base.py

   1 # Copyright 2014: Mirantis Inc.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 # yardstick comment: this is a modified copy of
  17 # rally/rally/benchmark/runners/base.py
  18
  19 from __future__ import absolute_import
  20 import importlib
  21 import logging
  22 import multiprocessing
  23 import subprocess
  24 import time
  25 import traceback
  26
  27 from oslo_config import cfg
  28
  29 import yardstick.common.utils as utils
  30 from yardstick.benchmark.scenarios import base as base_scenario
  31 from yardstick.dispatcher.base import Base as DispatcherBase
  32
  33 log = logging.getLogger(__name__)
  34
  35 CONF = cfg.CONF
  36
  37
  38 def _output_serializer_main(filename, queue, config):
  39     """entrypoint for the singleton subprocess writing to outfile
  40     Use of this process enables multiple instances of a scenario without
  41     messing up the output file.
  42     """
  43     out_type = config['yardstick'].get('DEFAULT', {}).get('dispatcher', 'file')
  44     conf = {
  45         'type': out_type.capitalize(),
  46         'file_path': filename
  47     }
  48
  49     dispatcher = DispatcherBase.get(conf, config)
  50
  51     while True:
  52         # blocks until data becomes available
  53         record = queue.get()
  54         if record == '_TERMINATE_':
  55             dispatcher.flush_result_data()
  56             break
  57         else:
  58             dispatcher.record_result_data(record)
  59
  60
  61 def _execute_shell_command(command):
  62     """execute shell script with error handling"""
  63     exitcode = 0
  64     output = []
  65     try:
  66         output = subprocess.check_output(command, shell=True)
  67     except Exception:
  68         exitcode = -1
  69         output = traceback.format_exc()
  70         log.error("exec command '%s' error:\n ", command)
  71         log.error(traceback.format_exc())
  72
  73     return exitcode, output
  74
  75
  76 def _single_action(seconds, command, queue):
  77     """entrypoint for the single action process"""
  78     log.debug("single action, fires after %d seconds (from now)", seconds)
  79     time.sleep(seconds)
  80     log.debug("single action: executing command: '%s'", command)
  81     ret_code, data = _execute_shell_command(command)
  82     if ret_code < 0:
  83         log.error("single action error! command:%s", command)
  84         queue.put({'single-action-data': data})
  85         return
  86     log.debug("single action data: \n%s", data)
  87     queue.put({'single-action-data': data})
  88
  89
  90 def _periodic_action(interval, command, queue):
  91     """entrypoint for the periodic action process"""
  92     log.debug("periodic action, fires every: %d seconds", interval)
  93     time_spent = 0
  94     while True:
  95         time.sleep(interval)
  96         time_spent += interval
  97         log.debug("periodic action, executing command: '%s'", command)
  98         ret_code, data = _execute_shell_command(command)
  99         if ret_code < 0:
 100             log.error("periodic action error! command:%s", command)
 101             queue.put({'periodic-action-data': data})
 102             break
 103         log.debug("periodic action data: \n%s", data)
 104         queue.put({'periodic-action-data': data})
 105
 106
 107 class Runner(object):
 108     queue = None
 109     dump_process = None
 110     runners = []
 111
 112     @staticmethod
 113     def get_cls(runner_type):
 114         """return class of specified type"""
 115         for runner in utils.itersubclasses(Runner):
 116             if runner_type == runner.__execution_type__:
 117                 return runner
 118         raise RuntimeError("No such runner_type %s" % runner_type)
 119
 120     @staticmethod
 121     def get_types():
 122         """return a list of known runner type (class) names"""
 123         types = []
 124         for runner in utils.itersubclasses(Runner):
 125             types.append(runner)
 126         return types
 127
 128     @staticmethod
 129     def get(runner_cfg, config):
 130         """Returns instance of a scenario runner for execution type.
 131         """
 132         # if there is no runner, start the output serializer subprocess
 133         if not Runner.runners:
 134             log.debug("Starting dump process file '%s'",
 135                       runner_cfg["output_filename"])
 136             Runner.queue = multiprocessing.Queue()
 137             Runner.dump_process = multiprocessing.Process(
 138                 target=_output_serializer_main,
 139                 name="Dumper",
 140                 args=(runner_cfg["output_filename"], Runner.queue, config))
 141             Runner.dump_process.start()
 142
 143         return Runner.get_cls(runner_cfg["type"])(runner_cfg, Runner.queue)
 144
 145     @staticmethod
 146     def release_dump_process():
 147         """Release the dumper process"""
 148         log.debug("Stopping dump process")
 149         if Runner.dump_process:
 150             Runner.queue.put('_TERMINATE_')
 151             Runner.dump_process.join()
 152             Runner.dump_process = None
 153
 154     @staticmethod
 155     def release(runner):
 156         """Release the runner"""
 157         if runner in Runner.runners:
 158             Runner.runners.remove(runner)
 159
 160         # if this was the last runner, stop the output serializer subprocess
 161         if not Runner.runners:
 162             Runner.release_dump_process()
 163
 164     @staticmethod
 165     def terminate(runner):
 166         """Terminate the runner"""
 167         if runner.process and runner.process.is_alive():
 168             runner.process.terminate()
 169
 170     @staticmethod
 171     def terminate_all():
 172         """Terminate all runners (subprocesses)"""
 173         log.debug("Terminating all runners")
 174
 175         # release dumper process as some errors before any runner is created
 176         if not Runner.runners:
 177             Runner.release_dump_process()
 178             return
 179
 180         for runner in Runner.runners:
 181             log.debug("Terminating runner: %s", runner)
 182             if runner.process:
 183                 runner.process.terminate()
 184                 runner.process.join()
 185             if runner.periodic_action_process:
 186                 log.debug("Terminating periodic action process")
 187                 runner.periodic_action_process.terminate()
 188                 runner.periodic_action_process = None
 189             Runner.release(runner)
 190
 191     def __init__(self, config, queue):
 192         self.config = config
 193         self.periodic_action_process = None
 194         self.result_queue = queue
 195         self.process = None
 196         self.aborted = multiprocessing.Event()
 197         Runner.runners.append(self)
 198
 199     def run_post_stop_action(self):
 200         """run a potentially configured post-stop action"""
 201         if "post-stop-action" in self.config:
 202             command = self.config["post-stop-action"]["command"]
 203             log.debug("post stop action: command: '%s'", command)
 204             ret_code, data = _execute_shell_command(command)
 205             if ret_code < 0:
 206                 log.error("post action error! command:%s", command)
 207                 self.result_queue.put({'post-stop-action-data': data})
 208                 return
 209             log.debug("post-stop data: \n%s", data)
 210             self.result_queue.put({'post-stop-action-data': data})
 211
 212     def run(self, scenario_cfg, context_cfg):
 213         scenario_type = scenario_cfg["type"]
 214         class_name = base_scenario.Scenario.get(scenario_type)
 215         path_split = class_name.split(".")
 216         module_path = ".".join(path_split[:-1])
 217         module = importlib.import_module(module_path)
 218         cls = getattr(module, path_split[-1])
 219
 220         self.config['object'] = class_name
 221         self.aborted.clear()
 222
 223         # run a potentially configured pre-start action
 224         if "pre-start-action" in self.config:
 225             command = self.config["pre-start-action"]["command"]
 226             log.debug("pre start action: command: '%s'", command)
 227             ret_code, data = _execute_shell_command(command)
 228             if ret_code < 0:
 229                 log.error("pre-start action error! command:%s", command)
 230                 self.result_queue.put({'pre-start-action-data': data})
 231                 return
 232             log.debug("pre-start data: \n%s", data)
 233             self.result_queue.put({'pre-start-action-data': data})
 234
 235         if "single-shot-action" in self.config:
 236             single_action_process = multiprocessing.Process(
 237                 target=_single_action,
 238                 name="single-shot-action",
 239                 args=(self.config["single-shot-action"]["after"],
 240                       self.config["single-shot-action"]["command"],
 241                       self.result_queue))
 242             single_action_process.start()
 243
 244         if "periodic-action" in self.config:
 245             self.periodic_action_process = multiprocessing.Process(
 246                 target=_periodic_action,
 247                 name="periodic-action",
 248                 args=(self.config["periodic-action"]["interval"],
 249                       self.config["periodic-action"]["command"],
 250                       self.result_queue))
 251             self.periodic_action_process.start()
 252
 253         self._run_benchmark(cls, "run", scenario_cfg, context_cfg)
 254
 255     def abort(self):
 256         """Abort the execution of a scenario"""
 257         self.aborted.set()
 258
 259     def join(self, timeout=None):
 260         self.process.join(timeout)
 261         if self.periodic_action_process:
 262             self.periodic_action_process.terminate()
 263             self.periodic_action_process = None
 264
 265         self.run_post_stop_action()
 266         return self.process.exitcode