X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=yardstick%2Fnetwork_services%2Fvnf_generic%2Fvnf%2Fsample_vnf.py;h=b2c5a98bcf1f7ae745699bbbd92e7de17940bff3;hb=771215553f56993dd68b06dba561dcee17d7fc4f;hp=3976acb7646385bae6a4e76379168c6173c662d2;hpb=1160225135e429ba8581ad162b08d710db4e96a2;p=yardstick.git diff --git a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py index 3976acb76..b2c5a98bc 100644 --- a/yardstick/network_services/vnf_generic/vnf/sample_vnf.py +++ b/yardstick/network_services/vnf_generic/vnf/sample_vnf.py @@ -13,15 +13,16 @@ # limitations under the License. import logging +import decimal from multiprocessing import Queue, Value, Process - import os import posixpath import re -import six +import uuid import subprocess import time + from trex_stl_lib.trex_stl_client import LoggerApi from trex_stl_lib.trex_stl_client import STLClient from trex_stl_lib.trex_stl_exceptions import STLError @@ -39,7 +40,7 @@ from yardstick.network_services.vnf_generic.vnf.base import GenericTrafficGen from yardstick.network_services.vnf_generic.vnf.base import GenericVNF from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper - +from yardstick.benchmark.contexts.node import NodeContext LOG = logging.getLogger(__name__) @@ -112,19 +113,6 @@ class DpdkVnfSetupEnvHelper(SetupEnvHelper): self.used_drivers = None self.dpdk_bind_helper = DpdkBindHelper(ssh_helper) - def _setup_hugepages(self): - meminfo = utils.read_meminfo(self.ssh_helper) - hp_size_kb = int(meminfo['Hugepagesize']) - hugepages_gb = self.scenario_helper.all_options.get('hugepages_gb', 16) - nr_hugepages = int(abs(hugepages_gb * 1024 * 1024 / hp_size_kb)) - self.ssh_helper.execute('echo %s | sudo tee %s' % - (nr_hugepages, self.NR_HUGEPAGES_PATH)) - hp = six.BytesIO() - self.ssh_helper.get_file_obj(self.NR_HUGEPAGES_PATH, hp) - nr_hugepages_set = int(hp.getvalue().decode('utf-8').splitlines()[0]) - LOG.info('Hugepages size (kB): %s, number claimed: %s, number set: %s', - hp_size_kb, nr_hugepages, nr_hugepages_set) - def build_config(self): vnf_cfg = self.scenario_helper.vnf_cfg task_path = self.scenario_helper.task_path @@ -237,7 +225,8 @@ class DpdkVnfSetupEnvHelper(SetupEnvHelper): def _setup_dpdk(self): """Setup DPDK environment needed for VNF to run""" - self._setup_hugepages() + hugepages_gb = self.scenario_helper.all_options.get('hugepages_gb', 16) + utils.setup_hugepages(self.ssh_helper, hugepages_gb * 1024 * 1024) self.dpdk_bind_helper.load_dpdk_driver() exit_status = self.dpdk_bind_helper.check_dpdk_driver() @@ -319,6 +308,7 @@ class ResourceHelper(object): self.resource = None self.setup_helper = setup_helper self.ssh_helper = setup_helper.ssh_helper + self._enable = True def setup(self): self.resource = self.setup_helper.setup_vnf_environment() @@ -326,22 +316,33 @@ class ResourceHelper(object): def generate_cfg(self): pass + def update_from_context(self, context, attr_name): + """Disable resource helper in case of baremetal context. + + And update appropriate node collectd options in context + """ + if isinstance(context, NodeContext): + self._enable = False + context.update_collectd_options_for_node(self.setup_helper.collectd_options, + attr_name) + def _collect_resource_kpi(self): result = {} status = self.resource.check_if_system_agent_running("collectd")[0] - if status == 0: + if status == 0 and self._enable: result = self.resource.amqp_collect_nfvi_kpi() result = {"core": result} return result def start_collect(self): - self.resource.initiate_systemagent(self.ssh_helper.bin_path) - self.resource.start() - self.resource.amqp_process_for_nfvi_kpi() + if self._enable: + self.resource.initiate_systemagent(self.ssh_helper.bin_path) + self.resource.start() + self.resource.amqp_process_for_nfvi_kpi() def stop_collect(self): - if self.resource: + if self.resource and self._enable: self.resource.stop() def collect_kpi(self): @@ -396,12 +397,13 @@ class ClientResourceHelper(ResourceHelper): time.sleep(self.QUEUE_WAIT_TIME) self._queue.put(samples) - def run_traffic(self, traffic_profile): + def run_traffic(self, traffic_profile, mq_producer): # if we don't do this we can hang waiting for the queue to drain # have to do this in the subprocess self._queue.cancel_join_thread() # fixme: fix passing correct trex config file, # instead of searching the default path + mq_producer.tg_method_started() try: self._build_ports() self.client = self._connect() @@ -409,8 +411,12 @@ class ClientResourceHelper(ResourceHelper): self.client.remove_all_streams(self.all_ports) # remove all streams traffic_profile.register_generator(self) + iteration_index = 0 while self._terminated.value == 0: - self._run_traffic_once(traffic_profile) + iteration_index += 1 + if self._run_traffic_once(traffic_profile): + self._terminated.value = 1 + mq_producer.tg_method_iteration(iteration_index) self.client.stop(self.all_ports) self.client.disconnect() @@ -421,6 +427,8 @@ class ClientResourceHelper(ResourceHelper): return # return if trex/tg server is stopped. raise + mq_producer.tg_method_finished() + def terminate(self): self._terminated.value = 1 # stop client @@ -479,6 +487,7 @@ class Rfc2544ResourceHelper(object): self._rfc2544 = None self._tolerance_low = None self._tolerance_high = None + self._tolerance_precision = None @property def rfc2544(self): @@ -498,6 +507,12 @@ class Rfc2544ResourceHelper(object): self.get_rfc_tolerance() return self._tolerance_high + @property + def tolerance_precision(self): + if self._tolerance_precision is None: + self.get_rfc_tolerance() + return self._tolerance_precision + @property def correlated_traffic(self): if self._correlated_traffic is None: @@ -517,9 +532,13 @@ class Rfc2544ResourceHelper(object): def get_rfc_tolerance(self): tolerance_str = self.get_rfc2544('allowed_drop_rate', self.DEFAULT_TOLERANCE) - tolerance_iter = iter(sorted(float(t.strip()) for t in tolerance_str.split('-'))) - self._tolerance_low = next(tolerance_iter) - self._tolerance_high = next(tolerance_iter, self.tolerance_low) + tolerance_iter = iter(sorted( + decimal.Decimal(t.strip()) for t in tolerance_str.split('-'))) + tolerance_low = next(tolerance_iter) + tolerance_high = next(tolerance_iter, tolerance_low) + self._tolerance_precision = abs(tolerance_high.as_tuple().exponent) + self._tolerance_high = float(tolerance_high) + self._tolerance_low = float(tolerance_low) class SampleVNFDeployHelper(object): @@ -600,6 +619,7 @@ class ScenarioHelper(object): test_timeout = self.options.get('timeout', constants.DEFAULT_VNF_TIMEOUT) return test_duration if test_duration > test_timeout else test_timeout + class SampleVNF(GenericVNF): """ Class providing file-like API for generic VNF implementation """ @@ -609,8 +629,9 @@ class SampleVNF(GenericVNF): APP_NAME = "SampleVNF" # we run the VNF interactively, so the ssh command will timeout after this long - def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None): - super(SampleVNF, self).__init__(name, vnfd) + def __init__(self, name, vnfd, task_id, setup_env_helper_type=None, + resource_helper_type=None): + super(SampleVNF, self).__init__(name, vnfd, task_id) self.bin_path = get_nsb_option('bin_path', '') self.scenario_helper = ScenarioHelper(self.name) @@ -631,7 +652,6 @@ class SampleVNF(GenericVNF): self.resource_helper = resource_helper_type(self.setup_helper) self.context_cfg = None - self.nfvi_context = None self.pipeline_kwargs = {} self.uplink_ports = None self.downlink_ports = None @@ -658,8 +678,10 @@ class SampleVNF(GenericVNF): self._update_collectd_options(scenario_cfg, context_cfg) self.scenario_helper.scenario_cfg = scenario_cfg self.context_cfg = context_cfg - self.nfvi_context = Context.get_context_from_server(self.scenario_helper.nodes[self.name]) - # self.nfvi_context = None + self.resource_helper.update_from_context( + Context.get_context_from_server(self.scenario_helper.nodes[self.name]), + self.scenario_helper.nodes[self.name] + ) # vnf deploy is unsupported, use ansible playbooks if self.scenario_helper.options.get("vnf_deploy", False): @@ -813,15 +835,18 @@ class SampleVNF(GenericVNF): check_if_process_failed(self._vnf_process, 0.01) stats = self.get_stats() m = re.search(self.COLLECT_KPI, stats, re.MULTILINE) + physical_node = Context.get_physical_node_from_server( + self.scenario_helper.nodes[self.name]) + + result = {"physical_node": physical_node} if m: - result = {k: int(m.group(v)) for k, v in self.COLLECT_MAP.items()} + result.update({k: int(m.group(v)) for k, v in self.COLLECT_MAP.items()}) result["collect_stats"] = self.resource_helper.collect_kpi() else: - result = { - "packets_in": 0, - "packets_fwd": 0, - "packets_dropped": 0, - } + result.update({"packets_in": 0, + "packets_fwd": 0, + "packets_dropped": 0}) + LOG.debug("%s collect KPIs %s", self.APP_NAME, result) return result @@ -837,8 +862,9 @@ class SampleVNFTrafficGen(GenericTrafficGen): APP_NAME = 'Sample' RUN_WAIT = 1 - def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None): - super(SampleVNFTrafficGen, self).__init__(name, vnfd) + def __init__(self, name, vnfd, task_id, setup_env_helper_type=None, + resource_helper_type=None): + super(SampleVNFTrafficGen, self).__init__(name, vnfd, task_id) self.bin_path = get_nsb_option('bin_path', '') self.scenario_helper = ScenarioHelper(self.name) @@ -867,6 +893,11 @@ class SampleVNFTrafficGen(GenericTrafficGen): def instantiate(self, scenario_cfg, context_cfg): self.scenario_helper.scenario_cfg = scenario_cfg + self.resource_helper.update_from_context( + Context.get_context_from_server(self.scenario_helper.nodes[self.name]), + self.scenario_helper.nodes[self.name] + ) + self.resource_helper.setup() # must generate_cfg after DPDK bind because we need port number self.resource_helper.generate_cfg() @@ -890,12 +921,13 @@ class SampleVNFTrafficGen(GenericTrafficGen): LOG.info("%s TG Server is up and running.", self.APP_NAME) return self._tg_process.exitcode - def _traffic_runner(self, traffic_profile): + def _traffic_runner(self, traffic_profile, mq_id): # always drop connections first thing in new processes # so we don't get paramiko errors self.ssh_helper.drop_connection() LOG.info("Starting %s client...", self.APP_NAME) - self.resource_helper.run_traffic(traffic_profile) + self._mq_producer = self._setup_mq_producer(mq_id) + self.resource_helper.run_traffic(traffic_profile, self._mq_producer) def run_traffic(self, traffic_profile): """ Generate traffic on the wire according to the given params. @@ -905,10 +937,12 @@ class SampleVNFTrafficGen(GenericTrafficGen): :param traffic_profile: :return: True/False """ - name = "{}-{}-{}-{}".format(self.name, self.APP_NAME, traffic_profile.__class__.__name__, + name = '{}-{}-{}-{}'.format(self.name, self.APP_NAME, + traffic_profile.__class__.__name__, os.getpid()) - self._traffic_process = Process(name=name, target=self._traffic_runner, - args=(traffic_profile,)) + self._traffic_process = Process( + name=name, target=self._traffic_runner, + args=(traffic_profile, uuid.uuid1().int)) self._traffic_process.start() # Wait for traffic process to start while self.resource_helper.client_started.value == 0: @@ -917,13 +951,16 @@ class SampleVNFTrafficGen(GenericTrafficGen): if not self._traffic_process.is_alive(): break - return self._traffic_process.is_alive() - def collect_kpi(self): # check if the tg processes have exited + physical_node = Context.get_physical_node_from_server( + self.scenario_helper.nodes[self.name]) + + result = {"physical_node": physical_node} for proc in (self._tg_process, self._traffic_process): check_if_process_failed(proc) - result = self.resource_helper.collect_kpi() + + result["collect_stats"] = self.resource_helper.collect_kpi() LOG.debug("%s collect KPIs %s", self.APP_NAME, result) return result