diff options
Diffstat (limited to 'yardstick/benchmark')
| -rw-r--r-- | yardstick/benchmark/contexts/heat.py | 67 | ||||
| -rw-r--r-- | yardstick/benchmark/core/report.py | 135 | ||||
| -rw-r--r-- | yardstick/benchmark/core/task.py | 22 | ||||
| -rwxr-xr-x | yardstick/benchmark/runners/base.py | 49 | ||||
| -rw-r--r-- | yardstick/benchmark/runners/iteration.py | 37 | ||||
| -rw-r--r-- | yardstick/benchmark/runners/iteration_ipc.py | 205 | ||||
| -rw-r--r-- | yardstick/benchmark/runners/sequence.py | 33 | ||||
| -rw-r--r-- | yardstick/benchmark/scenarios/availability/monitor/monitor_multi.py | 27 | ||||
| -rw-r--r-- | yardstick/benchmark/scenarios/availability/monitor/monitor_process.py | 12 | ||||
| -rw-r--r-- | yardstick/benchmark/scenarios/base.py | 4 | ||||
| -rw-r--r-- | yardstick/benchmark/scenarios/networking/vnf_generic.py | 209 |
11 files changed, 411 insertions, 389 deletions
diff --git a/yardstick/benchmark/contexts/heat.py b/yardstick/benchmark/contexts/heat.py index c3c5451bd..917aa9c39 100644 --- a/yardstick/benchmark/contexts/heat.py +++ b/yardstick/benchmark/contexts/heat.py @@ -71,6 +71,7 @@ class HeatContext(Context): self.shade_client = None self.heat_timeout = None self.key_filename = None + self.yardstick_gen_key_file = True self.shade_client = None self.operator_client = None self.nodes = [] @@ -105,6 +106,14 @@ class HeatContext(Context): self.template_file = attrs.get("heat_template") + # try looking for external private key when using external heat template + if self.template_file is not None: + self.key_filename = attrs.get("key_filename", None) + if self.key_filename is not None: + # Disable key file generation if an external private key + # has been provided + self.yardstick_gen_key_file = False + self.shade_client = openstack_utils.get_shade_client() self.operator_client = openstack_utils.get_shade_operator_client() @@ -335,14 +344,16 @@ class HeatContext(Context): """deploys template into a stack using cloud""" LOG.info("Deploying context '%s' START", self.name) - self.key_filename = ''.join( - [consts.YARDSTICK_ROOT_PATH, - 'yardstick/resources/files/yardstick_key-', - self.name]) + # Check if there was no external private key provided + if self.key_filename is None: + self.key_filename = ''.join( + [consts.YARDSTICK_ROOT_PATH, + 'yardstick/resources/files/yardstick_key-', + self.name]) # Permissions may have changed since creation; this can be fixed. If we # overwrite the file, we lose future access to VMs using this key. # As long as the file exists, even if it is unreadable, keep it intact - if not os.path.exists(self.key_filename): + if self.yardstick_gen_key_file and not os.path.exists(self.key_filename): SSH.gen_keys(self.key_filename) heat_template = HeatTemplate( @@ -442,12 +453,14 @@ class HeatContext(Context): } def _delete_key_file(self): - try: - utils.remove_file(self.key_filename) - utils.remove_file(self.key_filename + ".pub") - except OSError: - LOG.exception("There was an error removing the key file %s", - self.key_filename) + # Only remove the key file if it has been generated by yardstick + if self.yardstick_gen_key_file: + try: + utils.remove_file(self.key_filename) + utils.remove_file(self.key_filename + ".pub") + except OSError: + LOG.exception("There was an error removing the key file %s", + self.key_filename) def undeploy(self): """undeploys stack from cloud""" @@ -496,6 +509,14 @@ class HeatContext(Context): server.private_ip = self.stack.outputs.get( attr_name.get("private_ip_attr", object()), None) + + # Try to find interfaces + for key, value in attr_name.get("interfaces", {}).items(): + value["local_ip"] = server.private_ip + for k in ["local_mac", "netmask", "gateway_ip"]: + # Keep explicit None or missing entry as is + value[k] = self.stack.outputs.get(value[k]) + server.interfaces.update({key: value}) else: try: server = self._server_map[attr_name] @@ -505,13 +526,29 @@ class HeatContext(Context): if server is None: return None - pkey = pkg_resources.resource_string( - 'yardstick.resources', - h_join('files/yardstick_key', self.name)).decode('utf-8') - + # Get the pkey + if self.yardstick_gen_key_file: + pkey = pkg_resources.resource_string( + 'yardstick.resources', + h_join('files/yardstick_key', self.name)).decode('utf-8') + key_filename = pkg_resources.resource_filename('yardstick.resources', + h_join('files/yardstick_key', self.name)) + else: + # make sure the file exists before attempting to open it + if not os.path.exists(self.key_filename): + LOG.error("The key_filename provided %s does not exist!", + self.key_filename) + else: + try: + pkey = open(self.key_filename, 'r').read().decode('utf-8') + key_filename = self.key_filename + except IOError: + LOG.error("The key_filename provided (%s) is unreadable.", + self.key_filename) result = { "user": server.context.user, "pkey": pkey, + "key_filename": key_filename, "private_ip": server.private_ip, "interfaces": server.interfaces, "routing_table": self.generate_routing_table(server), diff --git a/yardstick/benchmark/core/report.py b/yardstick/benchmark/core/report.py index 0bc392fe5..b7d2fd02b 100644 --- a/yardstick/benchmark/core/report.py +++ b/yardstick/benchmark/core/report.py @@ -1,6 +1,6 @@ ############################################################################## # Copyright (c) 2017 Rajesh Kudaka <4k.rajesh@gmail.com> -# Copyright (c) 2018 Intel Corporation. +# Copyright (c) 2018-2019 Intel Corporation. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 @@ -10,13 +10,12 @@ """ Handler for yardstick command 'report' """ -import ast import re +import six import uuid import jinja2 from api.utils import influx -from oslo_utils import encodeutils from oslo_utils import uuidutils from yardstick.common import constants as consts from yardstick.common.utils import cliargs @@ -55,11 +54,9 @@ class JSTree(object): def format_for_jstree(self, data): """Format the data into the required format for jsTree. - The data format expected is a list of key-value pairs which represent - the data and label for each metric e.g.: + The data format expected is a list of metric names e.g.: - [{'data': [0, ], 'label': 'tg__0.DropPackets'}, - {'data': [548, ], 'label': 'tg__0.LatencyAvg.5'},] + ['tg__0.DropPackets', 'tg__0.LatencyAvg.5'] This data is converted into the format required for jsTree to group and display the metrics in a hierarchial fashion, including creating a @@ -76,8 +73,8 @@ class JSTree(object): self._created_nodes = ['#'] self.jstree_data = [] - for item in data: - self._create_node(item["label"]) + for metric in data: + self._create_node(metric) return self.jstree_data @@ -115,15 +112,34 @@ class Report(object): else: raise KeyError("Test case not found.") - def _get_tasks(self): - task_cmd = "select * from \"%s\" where task_id= '%s'" - task_query = task_cmd % (self.yaml_name, self.task_id) - query_exec = influx.query(task_query) + def _get_metrics(self): + metrics_cmd = "select * from \"%s\" where task_id = '%s'" + metrics_query = metrics_cmd % (self.yaml_name, self.task_id) + query_exec = influx.query(metrics_query) if query_exec: return query_exec else: raise KeyError("Task ID or Test case not found.") + def _get_trimmed_timestamp(self, metric_time, resolution=4): + if not isinstance(metric_time, str): + metric_time = metric_time.encode('utf8') # PY2: unicode to str + metric_time = metric_time[11:] # skip date, keep time + head, _, tail = metric_time.partition('.') # split HH:MM:SS & nsZ + metric_time = head + '.' + tail[:resolution] # join HH:MM:SS & .us + return metric_time + + def _get_timestamps(self, metrics, resolution=6): + # Extract the timestamps from a list of metrics + timestamps = [] + for metric in metrics: + metric_time = self._get_trimmed_timestamp( + metric['time'], resolution) + timestamps.append(metric_time) # HH:MM:SS.micros + return timestamps + + @cliargs("task_id", type=str, help=" task id", nargs=1) + @cliargs("yaml_name", type=str, help=" Yaml file Name", nargs=1) def _generate_common(self, args): """Actions that are common to both report formats. @@ -132,38 +148,64 @@ class Report(object): """ self._validate(args.yaml_name[0], args.task_id[0]) - self.db_fieldkeys = self._get_fieldkeys() + db_fieldkeys = self._get_fieldkeys() + # list of dicts of: + # - PY2: unicode key and unicode value + # - PY3: str key and str value - self.db_task = self._get_tasks() + db_metrics = self._get_metrics() + # list of dicts of: + # - PY2: unicode key and { None | unicode | float | long | int } value + # - PY3: str key and { None | str | float | int } value - field_keys = [] - datasets = [] - table_vals = {} + # extract fieldKey entries, and convert them to str where needed + field_keys = [key if isinstance(key, str) # PY3: already str + else key.encode('utf8') # PY2: unicode to str + for key in + [field['fieldKey'] + for field in db_fieldkeys]] + + # extract timestamps + self.Timestamp = self._get_timestamps(db_metrics) - field_keys = [encodeutils.to_utf8(field['fieldKey']) - for field in self.db_fieldkeys] + # prepare return values + datasets = [] + table_vals = {'Timestamp': self.Timestamp} + # extract and convert field values for key in field_keys: - self.Timestamp = [] values = [] - for task in self.db_task: - task_time = encodeutils.to_utf8(task['time']) - if not isinstance(task_time, str): - task_time = str(task_time, 'utf8') - if not isinstance(key, str): - key = str(key, 'utf8') - task_time = task_time[11:] - head, _, tail = task_time.partition('.') - task_time = head + "." + tail[:6] - self.Timestamp.append(task_time) - if task[key] is None: - values.append(None) - elif isinstance(task[key], (int, float)): - values.append(task[key]) + for metric in db_metrics: + val = metric.get(key, None) + if val is None: + # keep explicit None or missing entry as is + pass + elif isinstance(val, (int, float)): + # keep plain int or float as is + pass + elif six.PY2 and isinstance(val, + long): # pylint: disable=undefined-variable + # PY2: long value would be rendered with trailing L, + # which JS does not support, so convert it to float + val = float(val) + elif isinstance(val, six.string_types): + s = val + if not isinstance(s, str): + s = s.encode('utf8') # PY2: unicode to str + try: + # convert until failure + val = s + val = float(s) + val = int(s) + if six.PY2 and isinstance(val, + long): # pylint: disable=undefined-variable + val = float(val) # PY2: long to float + except ValueError: + pass else: - values.append(ast.literal_eval(task[key])) + raise ValueError("Cannot convert %r" % val) + values.append(val) datasets.append({'label': key, 'data': values}) - table_vals['Timestamp'] = self.Timestamp table_vals[key] = values return datasets, table_vals @@ -197,8 +239,14 @@ class Report(object): @cliargs("yaml_name", type=str, help=" Yaml file Name", nargs=1) def generate_nsb(self, args): """Start NSB report generation.""" - datasets, table_vals = self._generate_common(args) - jstree_data = JSTree().format_for_jstree(datasets) + _, report_data = self._generate_common(args) + report_time = report_data.pop('Timestamp') + report_keys = sorted(report_data, key=str.lower) + report_tree = JSTree().format_for_jstree(report_keys) + report_meta = { + "testcase": self.yaml_name, + "task_id": self.task_id, + } template_dir = consts.YARDSTICK_ROOT_PATH + "yardstick/common" template_environment = jinja2.Environment( @@ -207,10 +255,11 @@ class Report(object): lstrip_blocks=True) context = { - "Timestamps": self.Timestamp, - "task_id": self.task_id, - "table": table_vals, - "jstree_nodes": jstree_data, + "report_meta": report_meta, + "report_data": report_data, + "report_time": report_time, + "report_keys": report_keys, + "report_tree": report_tree, } template_html = template_environment.get_template("nsb_report.html.j2") diff --git a/yardstick/benchmark/core/task.py b/yardstick/benchmark/core/task.py index 477dbcc57..bcca3558f 100644 --- a/yardstick/benchmark/core/task.py +++ b/yardstick/benchmark/core/task.py @@ -621,9 +621,19 @@ class TaskParser(object): # pragma: no cover scenario: nodes: - tg__0: tg_0.yardstick + tg__0: trafficgen_0.yardstick vnf__0: vnf_0.yardstick + scenario: + nodes: + tg__0: + name: trafficgen_0.yardstick + public_ip_attr: "server1_public_ip" + private_ip_attr: "server1_private_ip" + vnf__0: + name: vnf_0.yardstick + public_ip_attr: "server2_public_ip" + private_ip_attr: "server2_private_ip" NOTE: in Kubernetes context, the separator character between the server name and the context name is "-": scenario: @@ -655,7 +665,15 @@ class TaskParser(object): # pragma: no cover scenario['targets'][idx] = qualified_name(target) if 'nodes' in scenario: for scenario_node, target in scenario['nodes'].items(): - scenario['nodes'][scenario_node] = qualified_name(target) + if isinstance(target, collections.Mapping): + # Update node info on scenario with context info + # Just update the node name with context + # Append context information + target['name'] = qualified_name(target['name']) + # Then update node + scenario['nodes'][scenario_node] = target + else: + scenario['nodes'][scenario_node] = qualified_name(target) def _check_schema(self, cfg_schema, schema_type): """Check if config file is using the correct schema type""" diff --git a/yardstick/benchmark/runners/base.py b/yardstick/benchmark/runners/base.py index af2557441..94de45d1e 100755 --- a/yardstick/benchmark/runners/base.py +++ b/yardstick/benchmark/runners/base.py @@ -25,9 +25,6 @@ import traceback from six import moves from yardstick.benchmark.scenarios import base as base_scenario -from yardstick.common import messaging -from yardstick.common.messaging import payloads -from yardstick.common.messaging import producer from yardstick.common import utils from yardstick.dispatcher.base import Base as DispatcherBase @@ -80,6 +77,33 @@ def _periodic_action(interval, command, queue): queue.put({'periodic-action-data': data}) +class ScenarioOutput(dict): + + QUEUE_PUT_TIMEOUT = 10 + + def __init__(self, queue, **kwargs): + super(ScenarioOutput, self).__init__() + self._queue = queue + self.result_ext = dict() + for key, val in kwargs.items(): + self.result_ext[key] = val + setattr(self, key, val) + + def push(self, data=None, add_timestamp=True): + if data is None: + data = dict(self) + + if add_timestamp: + result = {'timestamp': time.time(), 'data': data} + else: + result = data + + for key in self.result_ext.keys(): + result[key] = getattr(self, key) + + self._queue.put(result, True, self.QUEUE_PUT_TIMEOUT) + + class Runner(object): runners = [] @@ -271,22 +295,3 @@ class Runner(object): 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) - - -class RunnerProducer(producer.MessagingProducer): - """Class implementing the message producer for runners""" - - def __init__(self, _id): - super(RunnerProducer, self).__init__(messaging.TOPIC_RUNNER, _id=_id) - - def start_iteration(self, version=1, data=None): - data = {} if not data else data - self.send_message( - messaging.RUNNER_METHOD_START_ITERATION, - payloads.RunnerPayload(version=version, data=data)) - - def stop_iteration(self, version=1, data=None): - data = {} if not data else data - self.send_message( - messaging.RUNNER_METHOD_STOP_ITERATION, - payloads.RunnerPayload(version=version, data=data)) diff --git a/yardstick/benchmark/runners/iteration.py b/yardstick/benchmark/runners/iteration.py index 58ab06a32..15dad2cd5 100644 --- a/yardstick/benchmark/runners/iteration.py +++ b/yardstick/benchmark/runners/iteration.py @@ -23,7 +23,6 @@ from __future__ import absolute_import import logging import multiprocessing -import time import traceback import os @@ -40,8 +39,6 @@ QUEUE_PUT_TIMEOUT = 10 def _worker_process(queue, cls, method_name, scenario_cfg, context_cfg, aborted, output_queue): - sequence = 1 - runner_cfg = scenario_cfg['runner'] interval = runner_cfg.get("interval", 1) @@ -53,6 +50,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg, runner_cfg['runner_id'] = os.getpid() + scenario_output = base.ScenarioOutput(queue, sequence=1, errors="") benchmark = cls(scenario_cfg, context_cfg) if "setup" in run_step: benchmark.setup() @@ -67,22 +65,21 @@ def _worker_process(queue, cls, method_name, scenario_cfg, LOG.debug("runner=%(runner)s seq=%(sequence)s START", {"runner": runner_cfg["runner_id"], - "sequence": sequence}) - - data = {} - errors = "" + "sequence": scenario_output.sequence}) + scenario_output.clear() + scenario_output.errors = "" benchmark.pre_run_wait_time(interval) try: - result = method(data) + result = method(scenario_output) except y_exc.SLAValidationError as error: # SLA validation failed in scenario, determine what to do now if sla_action == "assert": raise elif sla_action == "monitor": LOG.warning("SLA validation failed: %s", error.args) - errors = error.args + scenario_output.errors = error.args elif sla_action == "rate-control": try: scenario_cfg['options']['rate'] @@ -91,10 +88,10 @@ def _worker_process(queue, cls, method_name, scenario_cfg, scenario_cfg['options']['rate'] = 100 scenario_cfg['options']['rate'] -= delta - sequence = 1 + scenario_output.sequence = 1 continue except Exception: # pylint: disable=broad-except - errors = traceback.format_exc() + scenario_output.errors = traceback.format_exc() LOG.exception("") raise else: @@ -105,23 +102,17 @@ def _worker_process(queue, cls, method_name, scenario_cfg, benchmark.post_run_wait_time(interval) - benchmark_output = { - 'timestamp': time.time(), - 'sequence': sequence, - 'data': data, - 'errors': errors - } - - queue.put(benchmark_output, True, QUEUE_PUT_TIMEOUT) + if scenario_output: + scenario_output.push() LOG.debug("runner=%(runner)s seq=%(sequence)s END", {"runner": runner_cfg["runner_id"], - "sequence": sequence}) + "sequence": scenario_output.sequence}) - sequence += 1 + scenario_output.sequence += 1 - if (errors and sla_action is None) or \ - (sequence > iterations or aborted.is_set()): + if (scenario_output.errors and sla_action is None) or \ + (scenario_output.sequence > iterations or aborted.is_set()): LOG.info("worker END") break if "teardown" in run_step: diff --git a/yardstick/benchmark/runners/iteration_ipc.py b/yardstick/benchmark/runners/iteration_ipc.py deleted file mode 100644 index a0335fdc7..000000000 --- a/yardstick/benchmark/runners/iteration_ipc.py +++ /dev/null @@ -1,205 +0,0 @@ -# Copyright 2018: Intel Corporation -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. - -"""A runner that runs a configurable number of times before it returns. Each - iteration has a configurable timeout. The loop control depends on the - feedback received from the running VNFs. The context PIDs from the VNFs - to listen the messages from are given in the scenario "setup" method. -""" - -import logging -import multiprocessing -import time -import traceback - -import os - -from yardstick.benchmark.runners import base as base_runner -from yardstick.common import exceptions -from yardstick.common import messaging -from yardstick.common import utils -from yardstick.common.messaging import consumer -from yardstick.common.messaging import payloads - - -LOG = logging.getLogger(__name__) - -QUEUE_PUT_TIMEOUT = 10 -ITERATION_TIMEOUT = 180 - - -class RunnerIterationIPCEndpoint(consumer.NotificationHandler): - """Endpoint class for ``RunnerIterationIPCConsumer``""" - - def tg_method_started(self, ctxt, **kwargs): - if ctxt['id'] in self._ctx_ids: - self._queue.put( - {'id': ctxt['id'], - 'action': messaging.TG_METHOD_STARTED, - 'payload': payloads.TrafficGeneratorPayload.dict_to_obj( - kwargs)}, - QUEUE_PUT_TIMEOUT) - - def tg_method_finished(self, ctxt, **kwargs): - if ctxt['id'] in self._ctx_ids: - self._queue.put( - {'id': ctxt['id'], - 'action': messaging.TG_METHOD_FINISHED, - 'payload': payloads.TrafficGeneratorPayload.dict_to_obj( - kwargs)}) - - def tg_method_iteration(self, ctxt, **kwargs): - if ctxt['id'] in self._ctx_ids: - self._queue.put( - {'id': ctxt['id'], - 'action': messaging.TG_METHOD_ITERATION, - 'payload': payloads.TrafficGeneratorPayload.dict_to_obj( - kwargs)}) - - -class RunnerIterationIPCConsumer(consumer.MessagingConsumer): - """MQ consumer for "IterationIPC" runner""" - - def __init__(self, _id, ctx_ids): - self._id = _id - self._queue = multiprocessing.Queue() - endpoints = [RunnerIterationIPCEndpoint(_id, ctx_ids, self._queue)] - super(RunnerIterationIPCConsumer, self).__init__( - messaging.TOPIC_TG, ctx_ids, endpoints) - self._kpi_per_id = {ctx: [] for ctx in ctx_ids} - self.iteration_index = None - - def is_all_kpis_received_in_iteration(self): - """Check if all producers registered have sent the ITERATION msg - - During the present iteration, all producers (traffic generators) must - start and finish the traffic injection, and at the end of the traffic - injection a TG_METHOD_ITERATION must be sent. This function will check - all KPIs in the present iteration are received. E.g.: - self.iteration_index = 2 - - self._kpi_per_id = { - 'ctx1': [kpi0, kpi1, kpi2], - 'ctx2': [kpi0, kpi1]} --> return False - - self._kpi_per_id = { - 'ctx1': [kpi0, kpi1, kpi2], - 'ctx2': [kpi0, kpi1, kpi2]} --> return True - """ - while not self._queue.empty(): - msg = self._queue.get(True, 1) - if msg['action'] == messaging.TG_METHOD_ITERATION: - id_iter_list = self._kpi_per_id[msg['id']] - id_iter_list.append(msg['payload'].kpi) - - return all(len(id_iter_list) == self.iteration_index - for id_iter_list in self._kpi_per_id.values()) - - -def _worker_process(queue, cls, method_name, scenario_cfg, - context_cfg, aborted, output_queue): # pragma: no cover - runner_cfg = scenario_cfg['runner'] - - timeout = runner_cfg.get('timeout', ITERATION_TIMEOUT) - iterations = runner_cfg.get('iterations', 1) - run_step = runner_cfg.get('run_step', 'setup,run,teardown') - LOG.info('Worker START. Iterations %d times, class %s', iterations, cls) - - runner_cfg['runner_id'] = os.getpid() - - benchmark = cls(scenario_cfg, context_cfg) - method = getattr(benchmark, method_name) - - if 'setup' not in run_step: - raise exceptions.RunnerIterationIPCSetupActionNeeded() - benchmark.setup() - producer_ctxs = benchmark.get_mq_ids() - if not producer_ctxs: - raise exceptions.RunnerIterationIPCNoCtxs() - - mq_consumer = RunnerIterationIPCConsumer(os.getpid(), producer_ctxs) - mq_consumer.start_rpc_server() - mq_producer = base_runner.RunnerProducer(scenario_cfg['task_id']) - - iteration_index = 1 - while 'run' in run_step: - LOG.debug('runner=%(runner)s seq=%(sequence)s START', - {'runner': runner_cfg['runner_id'], - 'sequence': iteration_index}) - data = {} - result = None - errors = '' - mq_consumer.iteration_index = iteration_index - mq_producer.start_iteration() - - try: - utils.wait_until_true( - mq_consumer.is_all_kpis_received_in_iteration, - timeout=timeout, sleep=2) - result = method(data) - except Exception: # pylint: disable=broad-except - errors = traceback.format_exc() - LOG.exception(errors) - - mq_producer.stop_iteration() - - if result: - output_queue.put(result, True, QUEUE_PUT_TIMEOUT) - benchmark_output = {'timestamp': time.time(), - 'sequence': iteration_index, - 'data': data, - 'errors': errors} - queue.put(benchmark_output, True, QUEUE_PUT_TIMEOUT) - - LOG.debug('runner=%(runner)s seq=%(sequence)s END', - {'runner': runner_cfg['runner_id'], - 'sequence': iteration_index}) - - iteration_index += 1 - if iteration_index > iterations or aborted.is_set(): - LOG.info('"IterationIPC" worker END') - break - - if 'teardown' in run_step: - try: - benchmark.teardown() - except Exception: - LOG.exception('Exception during teardown process') - mq_consumer.stop_rpc_server() - raise SystemExit(1) - - LOG.debug('Data queue size = %s', queue.qsize()) - LOG.debug('Output queue size = %s', output_queue.qsize()) - mq_consumer.stop_rpc_server() - - -class IterationIPCRunner(base_runner.Runner): - """Run a scenario for a configurable number of times. - - Each iteration has a configurable timeout. The loop control depends on the - feedback received from the running VNFs. The context PIDs from the VNFs to - listen the messages from are given in the scenario "setup" method. - """ - __execution_type__ = 'IterationIPC' - - def _run_benchmark(self, cls, method, scenario_cfg, context_cfg): - name = '{}-{}-{}'.format( - self.__execution_type__, scenario_cfg.get('type'), os.getpid()) - self.process = multiprocessing.Process( - name=name, - target=_worker_process, - args=(self.result_queue, cls, method, scenario_cfg, - context_cfg, self.aborted, self.output_queue)) - self.process.start() diff --git a/yardstick/benchmark/runners/sequence.py b/yardstick/benchmark/runners/sequence.py index 0148a45b2..58ffddd22 100644 --- a/yardstick/benchmark/runners/sequence.py +++ b/yardstick/benchmark/runners/sequence.py @@ -38,8 +38,6 @@ LOG = logging.getLogger(__name__) def _worker_process(queue, cls, method_name, scenario_cfg, context_cfg, aborted, output_queue): - sequence = 1 - runner_cfg = scenario_cfg['runner'] interval = runner_cfg.get("interval", 1) @@ -56,6 +54,7 @@ def _worker_process(queue, cls, method_name, scenario_cfg, LOG.info("worker START, sequence_values(%s, %s), class %s", arg_name, sequence_values, cls) + scenario_output = base.ScenarioOutput(queue, sequence=1, errors="") benchmark = cls(scenario_cfg, context_cfg) benchmark.setup() method = getattr(benchmark, method_name) @@ -68,22 +67,23 @@ def _worker_process(queue, cls, method_name, scenario_cfg, options[arg_name] = value LOG.debug("runner=%(runner)s seq=%(sequence)s START", - {"runner": runner_cfg["runner_id"], "sequence": sequence}) + {"runner": runner_cfg["runner_id"], + "sequence": scenario_output.sequence}) - data = {} - errors = "" + scenario_output.clear() + scenario_output.errors = "" try: - result = method(data) + result = method(scenario_output) except y_exc.SLAValidationError as error: # SLA validation failed in scenario, determine what to do now if sla_action == "assert": raise elif sla_action == "monitor": LOG.warning("SLA validation failed: %s", error.args) - errors = error.args + scenario_output.errors = error.args except Exception as e: # pylint: disable=broad-except - errors = traceback.format_exc() + scenario_output.errors = traceback.format_exc() LOG.exception(e) else: if result: @@ -91,21 +91,16 @@ def _worker_process(queue, cls, method_name, scenario_cfg, time.sleep(interval) - benchmark_output = { - 'timestamp': time.time(), - 'sequence': sequence, - 'data': data, - 'errors': errors - } - - queue.put(benchmark_output) + if scenario_output: + scenario_output.push() LOG.debug("runner=%(runner)s seq=%(sequence)s END", - {"runner": runner_cfg["runner_id"], "sequence": sequence}) + {"runner": runner_cfg["runner_id"], + "sequence": scenario_output.sequence}) - sequence += 1 + scenario_output.sequence += 1 - if (errors and sla_action is None) or aborted.is_set(): + if (scenario_output.errors and sla_action is None) or aborted.is_set(): break try: diff --git a/yardstick/benchmark/scenarios/availability/monitor/monitor_multi.py b/yardstick/benchmark/scenarios/availability/monitor/monitor_multi.py index 971bae1e9..8f1f53cde 100644 --- a/yardstick/benchmark/scenarios/availability/monitor/monitor_multi.py +++ b/yardstick/benchmark/scenarios/availability/monitor/monitor_multi.py @@ -62,20 +62,19 @@ class MultiMonitor(basemonitor.BaseMonitor): outage_time = ( last_outage - first_outage if last_outage > first_outage else 0 ) + self._result = {"outage_time": outage_time} LOG.debug("outage_time is: %f", outage_time) max_outage_time = 0 - if "max_outage_time" in self._config["sla"]: - max_outage_time = self._config["sla"]["max_outage_time"] - elif "max_recover_time" in self._config["sla"]: - max_outage_time = self._config["sla"]["max_recover_time"] - else: - raise RuntimeError("'max_outage_time' or 'max_recover_time' " - "config is not found") - self._result = {"outage_time": outage_time} - - if outage_time > max_outage_time: - LOG.error("SLA failure: %f > %f", outage_time, max_outage_time) - return False - else: - return True + if self._config.get("sla"): + if "max_outage_time" in self._config["sla"]: + max_outage_time = self._config["sla"]["max_outage_time"] + elif "max_recover_time" in self._config["sla"]: + max_outage_time = self._config["sla"]["max_recover_time"] + else: + raise RuntimeError("'max_outage_time' or 'max_recover_time' " + "config is not found") + if outage_time > max_outage_time: + LOG.error("SLA failure: %f > %f", outage_time, max_outage_time) + return False + return True diff --git a/yardstick/benchmark/scenarios/availability/monitor/monitor_process.py b/yardstick/benchmark/scenarios/availability/monitor/monitor_process.py index 8d2f2633c..280e5811d 100644 --- a/yardstick/benchmark/scenarios/availability/monitor/monitor_process.py +++ b/yardstick/benchmark/scenarios/availability/monitor/monitor_process.py @@ -46,12 +46,12 @@ class MonitorProcess(basemonitor.BaseMonitor): def verify_SLA(self): outage_time = self._result.get('outage_time', None) - max_outage_time = self._config["sla"]["max_recover_time"] - if outage_time > max_outage_time: - LOG.info("SLA failure: %f > %f", outage_time, max_outage_time) - return False - else: - return True + if self._config.get("sla"): + max_outage_time = self._config["sla"]["max_recover_time"] + if outage_time > max_outage_time: + LOG.info("SLA failure: %f > %f", outage_time, max_outage_time) + return False + return True def _test(): # pragma: no cover diff --git a/yardstick/benchmark/scenarios/base.py b/yardstick/benchmark/scenarios/base.py index 1737bb942..ae8bfad71 100644 --- a/yardstick/benchmark/scenarios/base.py +++ b/yardstick/benchmark/scenarios/base.py @@ -122,7 +122,3 @@ class Scenario(object): except TypeError: dic[k] = v return dic - - def get_mq_ids(self): # pragma: no cover - """Return stored MQ producer IDs, if defined""" - pass diff --git a/yardstick/benchmark/scenarios/networking/vnf_generic.py b/yardstick/benchmark/scenarios/networking/vnf_generic.py index 5ac51cdfc..c5e75d093 100644 --- a/yardstick/benchmark/scenarios/networking/vnf_generic.py +++ b/yardstick/benchmark/scenarios/networking/vnf_generic.py @@ -1,4 +1,4 @@ -# Copyright (c) 2016-2017 Intel Corporation +# Copyright (c) 2016-2019 Intel Corporation # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -44,14 +44,13 @@ traffic_profile.register_modules() LOG = logging.getLogger(__name__) -class NetworkServiceTestCase(scenario_base.Scenario): - """Class handles Generic framework to do pre-deployment VNF & - Network service testing """ +class NetworkServiceBase(scenario_base.Scenario): + """Base class for Network service testing scenarios""" - __scenario_type__ = "NSPerf" + __scenario_type__ = "" def __init__(self, scenario_cfg, context_cfg): # pragma: no cover - super(NetworkServiceTestCase, self).__init__() + super(NetworkServiceBase, self).__init__() self.scenario_cfg = scenario_cfg self.context_cfg = context_cfg @@ -61,7 +60,32 @@ class NetworkServiceTestCase(scenario_base.Scenario): self.traffic_profile = None self.node_netdevs = {} self.bin_path = get_nsb_option('bin_path', '') - self._mq_ids = [] + + def run(self, *args): + pass + + def teardown(self): + """ Stop the collector and terminate VNF & TG instance + + :return + """ + + try: + try: + self.collector.stop() + for vnf in self.vnfs: + LOG.info("Stopping %s", vnf.name) + vnf.terminate() + LOG.debug("all VNFs terminated: %s", ", ".join(vnf.name for vnf in self.vnfs)) + finally: + terminate_children() + except Exception: + # catch any exception in teardown and convert to simple exception + # never pass exceptions back to multiprocessing, because some exceptions can + # be unpicklable + # https://bugs.python.org/issue9400 + LOG.exception("") + raise RuntimeError("Error in teardown") def is_ended(self): return self.traffic_profile is not None and self.traffic_profile.is_ended() @@ -140,6 +164,13 @@ class NetworkServiceTestCase(scenario_base.Scenario): imix = {} return imix + def _get_ip_priority(self): + try: + priority = self.scenario_cfg['options']['priority'] + except KeyError: + priority = {} + return priority + def _get_traffic_profile(self): profile = self.scenario_cfg["traffic_profile"] path = self.scenario_cfg["task_path"] @@ -177,6 +208,7 @@ class NetworkServiceTestCase(scenario_base.Scenario): tprofile_data = { 'flow': self._get_traffic_flow(), 'imix': self._get_traffic_imix(), + 'priority': self._get_ip_priority(), tprofile_base.TrafficProfile.UPLINK: {}, tprofile_base.TrafficProfile.DOWNLINK: {}, 'extra_args': extra_args, @@ -446,12 +478,30 @@ class NetworkServiceTestCase(scenario_base.Scenario): pass self.create_interfaces_from_node(vnfd, node) vnf_impl = self.get_vnf_impl(vnfd['id']) - vnf_instance = vnf_impl(node_name, vnfd, scenario_cfg['task_id']) + vnf_instance = vnf_impl(node_name, vnfd) vnfs.append(vnf_instance) self.vnfs = vnfs return vnfs + def pre_run_wait_time(self, time_seconds): # pragma: no cover + """Time waited before executing the run method""" + time.sleep(time_seconds) + + def post_run_wait_time(self, time_seconds): # pragma: no cover + """Time waited after executing the run method""" + pass + + +class NetworkServiceTestCase(NetworkServiceBase): + """Class handles Generic framework to do pre-deployment VNF & + Network service testing """ + + __scenario_type__ = "NSPerf" + + def __init__(self, scenario_cfg, context_cfg): # pragma: no cover + super(NetworkServiceTestCase, self).__init__(scenario_cfg, context_cfg) + def setup(self): """Setup infrastructure, provission VNFs & start traffic""" # 1. Verify if infrastructure mapping can meet topology @@ -495,11 +545,6 @@ class NetworkServiceTestCase(scenario_base.Scenario): for traffic_gen in traffic_runners: LOG.info("Starting traffic on %s", traffic_gen.name) traffic_gen.run_traffic(self.traffic_profile) - self._mq_ids.append(traffic_gen.get_mq_producer_id()) - - def get_mq_ids(self): # pragma: no cover - """Return stored MQ producer IDs""" - return self._mq_ids def run(self, result): # yardstick API """ Yardstick calls run() at intervals defined in the yaml and @@ -515,33 +560,125 @@ class NetworkServiceTestCase(scenario_base.Scenario): result.update(self.collector.get_kpi()) - def teardown(self): - """ Stop the collector and terminate VNF & TG instance - :return +class NetworkServiceRFC2544(NetworkServiceBase): + """Class handles RFC2544 Network service testing""" + + __scenario_type__ = "NSPerf-RFC2544" + + def __init__(self, scenario_cfg, context_cfg): # pragma: no cover + super(NetworkServiceRFC2544, self).__init__(scenario_cfg, context_cfg) + + def setup(self): + """Setup infrastructure, provision VNFs""" + self.map_topology_to_infrastructure() + self.load_vnf_models() + + traffic_runners = [vnf for vnf in self.vnfs if vnf.runs_traffic] + non_traffic_runners = [vnf for vnf in self.vnfs if not vnf.runs_traffic] + try: + for vnf in chain(traffic_runners, non_traffic_runners): + LOG.info("Instantiating %s", vnf.name) + vnf.instantiate(self.scenario_cfg, self.context_cfg) + LOG.info("Waiting for %s to instantiate", vnf.name) + vnf.wait_for_instantiate() + except: + LOG.exception("") + for vnf in self.vnfs: + vnf.terminate() + raise + + self._generate_pod_yaml() + + def run(self, output): + """ Run experiment + + :param output: scenario output to push results + :return: None """ + self._fill_traffic_profile() + + traffic_runners = [vnf for vnf in self.vnfs if vnf.runs_traffic] + + for traffic_gen in traffic_runners: + traffic_gen.listen_traffic(self.traffic_profile) + + self.collector = Collector(self.vnfs, + context_base.Context.get_physical_nodes()) + self.collector.start() + + test_completed = False + while not test_completed: + for traffic_gen in traffic_runners: + LOG.info("Run traffic on %s", traffic_gen.name) + traffic_gen.run_traffic_once(self.traffic_profile) + + test_completed = True + for traffic_gen in traffic_runners: + # wait for all tg to complete running traffic + status = traffic_gen.wait_on_traffic() + LOG.info("Run traffic on %s complete status=%s", + traffic_gen.name, status) + if status == 'CONTINUE': + # continue running if at least one tg is running + test_completed = False + + output.push(self.collector.get_kpi()) + + self.collector.stop() + +class NetworkServiceRFC3511(NetworkServiceBase): + """Class handles RFC3511 Network service testing""" + + __scenario_type__ = "NSPerf-RFC3511" + + def __init__(self, scenario_cfg, context_cfg): # pragma: no cover + super(NetworkServiceRFC3511, self).__init__(scenario_cfg, context_cfg) + + def setup(self): + """Setup infrastructure, provision VNFs""" + self.map_topology_to_infrastructure() + self.load_vnf_models() + + traffic_runners = [vnf for vnf in self.vnfs if vnf.runs_traffic] + non_traffic_runners = [vnf for vnf in self.vnfs if not vnf.runs_traffic] try: - try: - self.collector.stop() - for vnf in self.vnfs: - LOG.info("Stopping %s", vnf.name) - vnf.terminate() - LOG.debug("all VNFs terminated: %s", ", ".join(vnf.name for vnf in self.vnfs)) - finally: - terminate_children() - except Exception: - # catch any exception in teardown and convert to simple exception - # never pass exceptions back to multiprocessing, because some exceptions can - # be unpicklable - # https://bugs.python.org/issue9400 + for vnf in chain(traffic_runners, non_traffic_runners): + LOG.info("Instantiating %s", vnf.name) + vnf.instantiate(self.scenario_cfg, self.context_cfg) + LOG.info("Waiting for %s to instantiate", vnf.name) + vnf.wait_for_instantiate() + except: LOG.exception("") - raise RuntimeError("Error in teardown") + for vnf in self.vnfs: + vnf.terminate() + raise - def pre_run_wait_time(self, time_seconds): # pragma: no cover - """Time waited before executing the run method""" - time.sleep(time_seconds) + self._generate_pod_yaml() - def post_run_wait_time(self, time_seconds): # pragma: no cover - """Time waited after executing the run method""" - pass + def run(self, output): + """ Run experiment + + :param output: scenario output to push results + :return: None + """ + + self._fill_traffic_profile() + + traffic_runners = [vnf for vnf in self.vnfs if vnf.runs_traffic] + + for traffic_gen in traffic_runners: + traffic_gen.listen_traffic(self.traffic_profile) + + self.collector = Collector(self.vnfs, + context_base.Context.get_physical_nodes()) + self.collector.start() + + for traffic_gen in traffic_runners: + LOG.info("Run traffic on %s", traffic_gen.name) + traffic_gen.run_traffic(self.traffic_profile) + + output.push(self.collector.get_kpi()) + + self.collector.stop() |