yardstick/benchmark/runners/duration.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/constant.py
  18
  19 """A runner that runs a specific time before it returns
  20 """
  21
  22 from __future__ import absolute_import
  23 import os
  24 import multiprocessing
  25 import logging
  26 import traceback
  27 import time
  28
  29 from yardstick.benchmark.runners import base
  30 from yardstick.common import exceptions as y_exc
  31
  32 LOG = logging.getLogger(__name__)
  33
  34
  35 QUEUE_PUT_TIMEOUT = 10
  36
  37
  38 def _worker_process(queue, cls, method_name, scenario_cfg,
  39                     context_cfg, aborted, output_queue):
  40
  41     sequence = 1
  42
  43     runner_cfg = scenario_cfg['runner']
  44
  45     interval = runner_cfg.get("interval", 1)
  46     duration = runner_cfg.get("duration", 60)
  47     LOG.info("Worker START, duration is %ds", duration)
  48     LOG.debug("class is %s", cls)
  49
  50     runner_cfg['runner_id'] = os.getpid()
  51
  52     benchmark = cls(scenario_cfg, context_cfg)
  53     benchmark.setup()
  54     method = getattr(benchmark, method_name)
  55
  56     sla_action = None
  57     if "sla" in scenario_cfg:
  58         sla_action = scenario_cfg["sla"].get("action", "assert")
  59
  60     start = time.time()
  61     timeout = start + duration
  62     while True:
  63
  64         LOG.debug("runner=%(runner)s seq=%(sequence)s START",
  65                   {"runner": runner_cfg["runner_id"], "sequence": sequence})
  66
  67         data = {}
  68         errors = ""
  69
  70         benchmark.pre_run_wait_time(interval)
  71
  72         try:
  73             result = method(data)
  74         except y_exc.SLAValidationError as error:
  75             # SLA validation failed in scenario, determine what to do now
  76             if sla_action == "assert":
  77                 benchmark.teardown()
  78                 raise
  79             elif sla_action == "monitor":
  80                 LOG.warning("SLA validation failed: %s", error.args)
  81                 errors = error.args
  82         # catch all exceptions because with multiprocessing we can have un-picklable exception
  83         # problems  https://bugs.python.org/issue9400
  84         except Exception:  # pylint: disable=broad-except
  85             errors = traceback.format_exc()
  86             LOG.exception("")
  87         else:
  88             if result:
  89                 # add timeout for put so we don't block test
  90                 # if we do timeout we don't care about dropping individual KPIs
  91                 output_queue.put(result, True, QUEUE_PUT_TIMEOUT)
  92
  93         benchmark.post_run_wait_time(interval)
  94
  95         benchmark_output = {
  96             'timestamp': time.time(),
  97             'sequence': sequence,
  98             'data': data,
  99             'errors': errors
 100         }
 101
 102         queue.put(benchmark_output, True, QUEUE_PUT_TIMEOUT)
 103
 104         LOG.debug("runner=%(runner)s seq=%(sequence)s END",
 105                   {"runner": runner_cfg["runner_id"], "sequence": sequence})
 106
 107         sequence += 1
 108
 109         if (errors and sla_action is None) or time.time() > timeout or aborted.is_set():
 110             LOG.info("Worker END")
 111             break
 112
 113     try:
 114         benchmark.teardown()
 115     except Exception:
 116         # catch any exception in teardown and convert to simple exception
 117         # never pass exceptions back to multiprocessing, because some exceptions can
 118         # be unpicklable
 119         # https://bugs.python.org/issue9400
 120         LOG.exception("")
 121         raise SystemExit(1)
 122
 123     LOG.debug("queue.qsize() = %s", queue.qsize())
 124     LOG.debug("output_queue.qsize() = %s", output_queue.qsize())
 125
 126
 127 class DurationRunner(base.Runner):
 128     """Run a scenario for a certain amount of time
 129
 130 If the scenario ends before the time has elapsed, it will be started again.
 131
 132   Parameters
 133     duration - amount of time the scenario will be run for
 134         type:    int
 135         unit:    seconds
 136         default: 1 sec
 137     interval - time to wait between each scenario invocation
 138         type:    int
 139         unit:    seconds
 140         default: 1 sec
 141     """
 142     __execution_type__ = 'Duration'
 143
 144     def _run_benchmark(self, cls, method, scenario_cfg, context_cfg):
 145         name = "{}-{}-{}".format(self.__execution_type__, scenario_cfg.get("type"), os.getpid())
 146         self.process = multiprocessing.Process(
 147             name=name,
 148             target=_worker_process,
 149             args=(self.result_queue, cls, method, scenario_cfg,
 150                   context_cfg, self.aborted, self.output_queue))
 151         self.process.start()