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):
  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     config = {}
  44     config["type"] = CONF.dispatcher.capitalize()
  45     config["file_path"] = filename
  46     dispatcher = DispatcherBase.get(config)
  47
  48     while True:
  49         # blocks until data becomes available
  50         record = queue.get()
  51         if record == '_TERMINATE_':
  52             dispatcher.flush_result_data()
  53             break
  54         else:
  55             dispatcher.record_result_data(record)
  56
  57
  58 def _execute_shell_command(command):
  59     """execute shell script with error handling"""
  60     exitcode = 0
  61     output = []
  62     try:
  63         output = subprocess.check_output(command, shell=True)
  64     except Exception:
  65         exitcode = -1
  66         output = traceback.format_exc()
  67         log.error("exec command '%s' error:\n ", command)
  68         log.error(traceback.format_exc())
  69
  70     return exitcode, output
  71
  72
  73 def _single_action(seconds, command, queue):
  74     """entrypoint for the single action process"""
  75     log.debug("single action, fires after %d seconds (from now)", seconds)
  76     time.sleep(seconds)
  77     log.debug("single action: executing command: '%s'", command)
  78     ret_code, data = _execute_shell_command(command)
  79     if ret_code < 0:
  80         log.error("single action error! command:%s", command)
  81         queue.put({'single-action-data': data})
  82         return
  83     log.debug("single action data: \n%s", data)
  84     queue.put({'single-action-data': data})
  85
  86
  87 def _periodic_action(interval, command, queue):
  88     """entrypoint for the periodic action process"""
  89     log.debug("periodic action, fires every: %d seconds", interval)
  90     time_spent = 0
  91     while True:
  92         time.sleep(interval)
  93         time_spent += interval
  94         log.debug("periodic action, executing command: '%s'", command)
  95         ret_code, data = _execute_shell_command(command)
  96         if ret_code < 0:
  97             log.error("periodic action error! command:%s", command)
  98             queue.put({'periodic-action-data': data})
  99             break
 100         log.debug("periodic action data: \n%s", data)
 101         queue.put({'periodic-action-data': data})
 102
 103
 104 class Runner(object):
 105     queue = None
 106     dump_process = None
 107     runners = []
 108
 109     @staticmethod
 110     def get_cls(runner_type):
 111         """return class of specified type"""
 112         for runner in utils.itersubclasses(Runner):
 113             if runner_type == runner.__execution_type__:
 114                 return runner
 115         raise RuntimeError("No such runner_type %s" % runner_type)
 116
 117     @staticmethod
 118     def get_types():
 119         """return a list of known runner type (class) names"""
 120         types = []
 121         for runner in utils.itersubclasses(Runner):
 122             types.append(runner)
 123         return types
 124
 125     @staticmethod
 126     def get(config):
 127         """Returns instance of a scenario runner for execution type.
 128         """
 129         # if there is no runner, start the output serializer subprocess
 130         if len(Runner.runners) == 0:
 131             log.debug("Starting dump process file '%s'",
 132                       config["output_filename"])
 133             Runner.queue = multiprocessing.Queue()
 134             Runner.dump_process = multiprocessing.Process(
 135                 target=_output_serializer_main,
 136                 name="Dumper",
 137                 args=(config["output_filename"], Runner.queue))
 138             Runner.dump_process.start()
 139
 140         return Runner.get_cls(config["type"])(config, Runner.queue)
 141
 142     @staticmethod
 143     def release_dump_process():
 144         """Release the dumper process"""
 145         log.debug("Stopping dump process")
 146         if Runner.dump_process:
 147             Runner.queue.put('_TERMINATE_')
 148             Runner.dump_process.join()
 149             Runner.dump_process = None
 150
 151     @staticmethod
 152     def release(runner):
 153         """Release the runner"""
 154         if runner in Runner.runners:
 155             Runner.runners.remove(runner)
 156
 157         # if this was the last runner, stop the output serializer subprocess
 158         if len(Runner.runners) == 0:
 159             Runner.release_dump_process()
 160
 161     @staticmethod
 162     def terminate(runner):
 163         """Terminate the runner"""
 164         if runner.process and runner.process.is_alive():
 165             runner.process.terminate()
 166
 167     @staticmethod
 168     def terminate_all():
 169         """Terminate all runners (subprocesses)"""
 170         log.debug("Terminating all runners")
 171
 172         # release dumper process as some errors before any runner is created
 173         if len(Runner.runners) == 0:
 174             Runner.release_dump_process()
 175             return
 176
 177         for runner in Runner.runners:
 178             log.debug("Terminating runner: %s", runner)
 179             if runner.process:
 180                 runner.process.terminate()
 181                 runner.process.join()
 182             if runner.periodic_action_process:
 183                 log.debug("Terminating periodic action process")
 184                 runner.periodic_action_process.terminate()
 185                 runner.periodic_action_process = None
 186             Runner.release(runner)
 187
 188     def __init__(self, config, queue):
 189         self.config = config
 190         self.periodic_action_process = None
 191         self.result_queue = queue
 192         self.process = None
 193         self.aborted = multiprocessing.Event()
 194         Runner.runners.append(self)
 195
 196     def run_post_stop_action(self):
 197         """run a potentially configured post-stop action"""
 198         if "post-stop-action" in self.config:
 199             command = self.config["post-stop-action"]["command"]
 200             log.debug("post stop action: command: '%s'", command)
 201             ret_code, data = _execute_shell_command(command)
 202             if ret_code < 0:
 203                 log.error("post action error! command:%s", command)
 204                 self.result_queue.put({'post-stop-action-data': data})
 205                 return
 206             log.debug("post-stop data: \n%s", data)
 207             self.result_queue.put({'post-stop-action-data': data})
 208
 209     def run(self, scenario_cfg, context_cfg):
 210         scenario_type = scenario_cfg["type"]
 211         class_name = base_scenario.Scenario.get(scenario_type)
 212         path_split = class_name.split(".")
 213         module_path = ".".join(path_split[:-1])
 214         module = importlib.import_module(module_path)
 215         cls = getattr(module, path_split[-1])
 216
 217         self.config['object'] = class_name
 218         self.aborted.clear()
 219
 220         # run a potentially configured pre-start action
 221         if "pre-start-action" in self.config:
 222             command = self.config["pre-start-action"]["command"]
 223             log.debug("pre start action: command: '%s'", command)
 224             ret_code, data = _execute_shell_command(command)
 225             if ret_code < 0:
 226                 log.error("pre-start action error! command:%s", command)
 227                 self.result_queue.put({'pre-start-action-data': data})
 228                 return
 229             log.debug("pre-start data: \n%s", data)
 230             self.result_queue.put({'pre-start-action-data': data})
 231
 232         if "single-shot-action" in self.config:
 233             single_action_process = multiprocessing.Process(
 234                 target=_single_action,
 235                 name="single-shot-action",
 236                 args=(self.config["single-shot-action"]["after"],
 237                       self.config["single-shot-action"]["command"],
 238                       self.result_queue))
 239             single_action_process.start()
 240
 241         if "periodic-action" in self.config:
 242             self.periodic_action_process = multiprocessing.Process(
 243                 target=_periodic_action,
 244                 name="periodic-action",
 245                 args=(self.config["periodic-action"]["interval"],
 246                       self.config["periodic-action"]["command"],
 247                       self.result_queue))
 248             self.periodic_action_process.start()
 249
 250         self._run_benchmark(cls, "run", scenario_cfg, context_cfg)
 251
 252     def abort(self):
 253         """Abort the execution of a scenario"""
 254         self.aborted.set()
 255
 256     def join(self, timeout=None):
 257         self.process.join(timeout)
 258         if self.periodic_action_process:
 259             self.periodic_action_process.terminate()
 260             self.periodic_action_process = None
 261
 262         self.run_post_stop_action()
 263         return self.process.exitcode