X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=yardstick%2Fbenchmark%2Frunners%2Fbase.py;h=99386a440151ec94c5e8d5b16ad366c5752f4c10;hb=0c64be36cd7820c46dd68bffda47fab6fc56ae94;hp=a69811f8a1e4c90b7c6ad4dbc85746c6655a7e01;hpb=a4d2958ddb727c2aaf46956cbdbf7718bfd0be89;p=yardstick.git diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py index a69811f8a..99386a440 100755 --- a/yardstick/benchmark/runners/base.py +++ b/yardstick/benchmark/runners/base.py @@ -17,15 +17,21 @@ # rally/rally/benchmark/runners/base.py from __future__ import absolute_import -import importlib + import logging import multiprocessing import subprocess import time import traceback +from subprocess import CalledProcessError + +import importlib + +from six.moves.queue import Empty import yardstick.common.utils as utils from yardstick.benchmark.scenarios import base as base_scenario +from yardstick.dispatcher.base import Base as DispatcherBase log = logging.getLogger(__name__) @@ -33,10 +39,9 @@ log = logging.getLogger(__name__) def _execute_shell_command(command): """execute shell script with error handling""" exitcode = 0 - output = [] try: output = subprocess.check_output(command, shell=True) - except Exception: + except CalledProcessError: exitcode = -1 output = traceback.format_exc() log.error("exec command '%s' error:\n ", command) @@ -116,7 +121,7 @@ class Runner(object): @staticmethod def terminate_all(): """Terminate all runners (subprocesses)""" - log.debug("Terminating all runners") + log.debug("Terminating all runners", exc_info=True) # release dumper process as some errors before any runner is created if not Runner.runners: @@ -134,6 +139,8 @@ class Runner(object): Runner.release(runner) def __init__(self, config): + self.task_id = None + self.case_name = None self.config = config self.periodic_action_process = None self.output_queue = multiprocessing.Queue() @@ -167,6 +174,8 @@ class Runner(object): cls = getattr(module, path_split[-1]) self.config['object'] = class_name + self.case_name = scenario_cfg['tc'] + self.task_id = scenario_cfg['task_id'] self.aborted.clear() # run a potentially configured pre-start action @@ -205,9 +214,25 @@ class Runner(object): """Abort the execution of a scenario""" self.aborted.set() - def join(self, timeout=None): + QUEUE_JOIN_INTERVAL = 5 + + def poll(self, timeout=QUEUE_JOIN_INTERVAL): self.process.join(timeout) + return self.process.exitcode + + def join(self, outputs, result, interval=QUEUE_JOIN_INTERVAL): + while self.process.exitcode is None: + # drain the queue while we are running otherwise we won't terminate + outputs.update(self.get_output()) + result.extend(self.get_result()) + self.process.join(interval) + # drain after the process has exited + outputs.update(self.get_output()) + result.extend(self.get_result()) + + self.process.terminate() if self.periodic_action_process: + self.periodic_action_process.join(1) self.periodic_action_process.terminate() self.periodic_action_process = None @@ -217,11 +242,33 @@ class Runner(object): def get_output(self): result = {} while not self.output_queue.empty(): - result.update(self.output_queue.get()) + log.debug("output_queue size %s", self.output_queue.qsize()) + try: + result.update(self.output_queue.get(True, 1)) + except Empty: + pass return result def get_result(self): result = [] + + dispatcher = self.config['output_config']['DEFAULT']['dispatcher'] + output_in_influxdb = 'influxdb' in dispatcher + while not self.result_queue.empty(): - result.append(self.result_queue.get()) + log.debug("result_queue size %s", self.result_queue.qsize()) + try: + one_record = self.result_queue.get(True, 1) + except Empty: + pass + else: + if output_in_influxdb: + self._output_to_influxdb(one_record) + + result.append(one_record) return result + + def _output_to_influxdb(self, record): + dispatchers = DispatcherBase.get(self.config['output_config']) + dispatcher = next((d for d in dispatchers if d.__dispatcher_type__ == 'Influxdb')) + dispatcher.upload_one_record(record, self.case_name, '', task_id=self.task_id)