1 # Copyright (c) 2016-2018 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """ Base class implementation for generic vnf implementation """
16 from collections import Mapping
18 from multiprocessing import Queue, Value, Process
28 from trex_stl_lib.trex_stl_client import LoggerApi
29 from trex_stl_lib.trex_stl_client import STLClient
30 from trex_stl_lib.trex_stl_exceptions import STLError
31 from yardstick.benchmark.contexts.base import Context
32 from yardstick.common import exceptions as y_exceptions
33 from yardstick.common.process import check_if_process_failed
34 from yardstick.common import utils
35 from yardstick.network_services import constants
36 from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper, DpdkNode
37 from yardstick.network_services.helpers.samplevnf_helper import MultiPortConfig
38 from yardstick.network_services.nfvi.resource import ResourceProfile
39 from yardstick.network_services.utils import get_nsb_option
40 from yardstick.network_services.vnf_generic.vnf.base import GenericTrafficGen
41 from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
42 from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper
43 from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper
46 LOG = logging.getLogger(__name__)
49 class SetupEnvHelper(object):
51 CFG_CONFIG = os.path.join(constants.REMOTE_TMP, "sample_config")
52 CFG_SCRIPT = os.path.join(constants.REMOTE_TMP, "sample_script")
53 DEFAULT_CONFIG_TPL_CFG = "sample.cfg"
57 def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
58 super(SetupEnvHelper, self).__init__()
59 self.vnfd_helper = vnfd_helper
60 self.ssh_helper = ssh_helper
61 self.scenario_helper = scenario_helper
63 def build_config(self):
64 raise NotImplementedError
66 def setup_vnf_environment(self):
73 raise NotImplementedError
76 class DpdkVnfSetupEnvHelper(SetupEnvHelper):
79 FIND_NET_CMD = "find /sys/class/net -lname '*{}*' -printf '%f'"
80 NR_HUGEPAGES_PATH = '/proc/sys/vm/nr_hugepages'
83 def _update_packet_type(ip_pipeline_cfg, traffic_options):
84 match_str = 'pkt_type = ipv4'
85 replace_str = 'pkt_type = {0}'.format(traffic_options['pkt_type'])
86 pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
87 return pipeline_config_str
90 def _update_traffic_type(cls, ip_pipeline_cfg, traffic_options):
91 traffic_type = traffic_options['traffic_type']
93 if traffic_options['vnf_type'] is not cls.APP_NAME:
94 match_str = 'traffic_type = 4'
95 replace_str = 'traffic_type = {0}'.format(traffic_type)
97 elif traffic_type == 4:
98 match_str = 'pkt_type = ipv4'
99 replace_str = 'pkt_type = ipv4'
102 match_str = 'pkt_type = ipv4'
103 replace_str = 'pkt_type = ipv6'
105 pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
106 return pipeline_config_str
108 def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
109 super(DpdkVnfSetupEnvHelper, self).__init__(vnfd_helper, ssh_helper, scenario_helper)
110 self.all_ports = None
111 self.bound_pci = None
113 self.used_drivers = None
114 self.dpdk_bind_helper = DpdkBindHelper(ssh_helper)
116 def _setup_hugepages(self):
117 meminfo = utils.read_meminfo(self.ssh_helper)
118 hp_size_kb = int(meminfo['Hugepagesize'])
119 hugepages_gb = self.scenario_helper.all_options.get('hugepages_gb', 16)
120 nr_hugepages = int(abs(hugepages_gb * 1024 * 1024 / hp_size_kb))
121 self.ssh_helper.execute('echo %s | sudo tee %s' %
122 (nr_hugepages, self.NR_HUGEPAGES_PATH))
124 self.ssh_helper.get_file_obj(self.NR_HUGEPAGES_PATH, hp)
125 nr_hugepages_set = int(hp.getvalue().decode('utf-8').splitlines()[0])
126 LOG.info('Hugepages size (kB): %s, number claimed: %s, number set: %s',
127 hp_size_kb, nr_hugepages, nr_hugepages_set)
129 def build_config(self):
130 vnf_cfg = self.scenario_helper.vnf_cfg
131 task_path = self.scenario_helper.task_path
133 config_file = vnf_cfg.get('file')
134 lb_count = vnf_cfg.get('lb_count', 3)
135 lb_config = vnf_cfg.get('lb_config', 'SW')
136 worker_config = vnf_cfg.get('worker_config', '1C/1T')
137 worker_threads = vnf_cfg.get('worker_threads', 3)
139 traffic_type = self.scenario_helper.all_options.get('traffic_type', 4)
141 'traffic_type': traffic_type,
142 'pkt_type': 'ipv%s' % traffic_type,
143 'vnf_type': self.VNF_TYPE,
146 config_tpl_cfg = utils.find_relative_file(self.DEFAULT_CONFIG_TPL_CFG,
148 config_basename = posixpath.basename(self.CFG_CONFIG)
149 script_basename = posixpath.basename(self.CFG_SCRIPT)
150 multiport = MultiPortConfig(self.scenario_helper.topology,
161 multiport.generate_config()
163 with utils.open_relative_file(config_file, task_path) as infile:
164 new_config = ['[EAL]']
166 for port in self.vnfd_helper.port_pairs.all_ports:
167 interface = self.vnfd_helper.find_interface(name=port)
168 vpci.append(interface['virtual-interface']["vpci"])
169 new_config.extend('w = {0}'.format(item) for item in vpci)
170 new_config = '\n'.join(new_config) + '\n' + infile.read()
172 with open(self.CFG_CONFIG) as handle:
173 new_config = handle.read()
174 new_config = self._update_traffic_type(new_config, traffic_options)
175 new_config = self._update_packet_type(new_config, traffic_options)
176 self.ssh_helper.upload_config_file(config_basename, new_config)
177 self.ssh_helper.upload_config_file(script_basename,
178 multiport.generate_script(self.vnfd_helper))
180 LOG.info("Provision and start the %s", self.APP_NAME)
181 self._build_pipeline_kwargs()
182 return self.PIPELINE_COMMAND.format(**self.pipeline_kwargs)
184 def _build_pipeline_kwargs(self):
185 tool_path = self.ssh_helper.provision_tool(tool_file=self.APP_NAME)
186 # count the number of actual ports in the list of pairs
187 # remove duplicate ports
188 # this is really a mapping from LINK ID to DPDK PMD ID
189 # e.g. 0x110 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_2
190 # 0x1010 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_3
191 ports = self.vnfd_helper.port_pairs.all_ports
192 port_nums = self.vnfd_helper.port_nums(ports)
193 # create mask from all the dpdk port numbers
194 ports_mask_hex = hex(sum(2 ** num for num in port_nums))
195 self.pipeline_kwargs = {
196 'cfg_file': self.CFG_CONFIG,
197 'script': self.CFG_SCRIPT,
198 'port_mask_hex': ports_mask_hex,
199 'tool_path': tool_path,
202 def setup_vnf_environment(self):
205 # bind before _setup_resources so we can use dpdk_port_num
206 self._detect_and_bind_drivers()
207 resource = self._setup_resources()
211 # pkill is not matching, debug with pgrep
212 self.ssh_helper.execute("sudo pgrep -lax %s" % self.APP_NAME)
213 self.ssh_helper.execute("sudo ps aux | grep -i %s" % self.APP_NAME)
214 # have to use exact match
215 # try using killall to match
216 self.ssh_helper.execute("sudo killall %s" % self.APP_NAME)
218 def _setup_dpdk(self):
219 """Setup DPDK environment needed for VNF to run"""
220 self._setup_hugepages()
221 self.dpdk_bind_helper.load_dpdk_driver()
223 exit_status = self.dpdk_bind_helper.check_dpdk_driver()
227 def get_collectd_options(self):
228 options = self.scenario_helper.all_options.get("collectd", {})
229 # override with specific node settings
230 options.update(self.scenario_helper.options.get("collectd", {}))
233 def _setup_resources(self):
234 # what is this magic? how do we know which socket is for which port?
235 # what about quad-socket?
236 if any(v[5] == "0" for v in self.bound_pci):
241 # implicit ordering, presumably by DPDK port num, so pre-sort by port_num
242 # this won't work because we don't have DPDK port numbers yet
243 ports = sorted(self.vnfd_helper.interfaces, key=self.vnfd_helper.port_num)
244 port_names = (intf["name"] for intf in ports)
245 collectd_options = self.get_collectd_options()
246 plugins = collectd_options.get("plugins", {})
247 # we must set timeout to be the same as the VNF otherwise KPIs will die before VNF
248 return ResourceProfile(self.vnfd_helper.mgmt_interface, port_names=port_names,
249 plugins=plugins, interval=collectd_options.get("interval"),
250 timeout=self.scenario_helper.timeout)
252 def _check_interface_fields(self):
253 num_nodes = len(self.scenario_helper.nodes)
254 # OpenStack instance creation time is probably proportional to the number
256 timeout = 120 * num_nodes
257 dpdk_node = DpdkNode(self.scenario_helper.name, self.vnfd_helper.interfaces,
258 self.ssh_helper, timeout)
261 def _detect_and_bind_drivers(self):
262 interfaces = self.vnfd_helper.interfaces
264 self._check_interface_fields()
265 # check for bound after probe
266 self.bound_pci = [v['virtual-interface']["vpci"] for v in interfaces]
268 self.dpdk_bind_helper.read_status()
269 self.dpdk_bind_helper.save_used_drivers()
271 self.dpdk_bind_helper.bind(self.bound_pci, 'igb_uio')
273 sorted_dpdk_pci_addresses = sorted(self.dpdk_bind_helper.dpdk_bound_pci_addresses)
274 for dpdk_port_num, vpci in enumerate(sorted_dpdk_pci_addresses):
276 intf = next(v for v in interfaces
277 if vpci == v['virtual-interface']['vpci'])
279 intf['virtual-interface']['dpdk_port_num'] = int(dpdk_port_num)
280 except: # pylint: disable=bare-except
284 def get_local_iface_name_by_vpci(self, vpci):
285 find_net_cmd = self.FIND_NET_CMD.format(vpci)
286 exit_status, stdout, _ = self.ssh_helper.execute(find_net_cmd)
292 self.dpdk_bind_helper.rebind_drivers()
295 class ResourceHelper(object):
298 MAKE_INSTALL = 'cd {0} && make && sudo make install'
299 RESOURCE_WORD = 'sample'
303 def __init__(self, setup_helper):
304 super(ResourceHelper, self).__init__()
306 self.setup_helper = setup_helper
307 self.ssh_helper = setup_helper.ssh_helper
310 self.resource = self.setup_helper.setup_vnf_environment()
312 def generate_cfg(self):
315 def _collect_resource_kpi(self):
317 status = self.resource.check_if_system_agent_running("collectd")[0]
319 result = self.resource.amqp_collect_nfvi_kpi()
321 result = {"core": result}
324 def start_collect(self):
325 self.resource.initiate_systemagent(self.ssh_helper.bin_path)
326 self.resource.start()
327 self.resource.amqp_process_for_nfvi_kpi()
329 def stop_collect(self):
333 def collect_kpi(self):
334 return self._collect_resource_kpi()
337 class ClientResourceHelper(ResourceHelper):
344 def __init__(self, setup_helper):
345 super(ClientResourceHelper, self).__init__(setup_helper)
346 self.vnfd_helper = setup_helper.vnfd_helper
347 self.scenario_helper = setup_helper.scenario_helper
350 self.client_started = Value('i', 0)
351 self.all_ports = None
352 self._queue = Queue()
354 self._terminated = Value('i', 0)
356 def _build_ports(self):
357 self.networks = self.vnfd_helper.port_pairs.networks
358 self.uplink_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.uplink_ports)
359 self.downlink_ports = \
360 self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.downlink_ports)
361 self.all_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.all_ports)
363 def port_num(self, intf):
364 # by default return port num
365 return self.vnfd_helper.port_num(intf)
367 def get_stats(self, *args, **kwargs):
369 return self.client.get_stats(*args, **kwargs)
371 LOG.error('TRex client not connected')
374 def generate_samples(self, ports, key=None, default=None):
375 # needs to be used ports
376 last_result = self.get_stats(ports)
377 key_value = last_result.get(key, default)
379 if not isinstance(last_result, Mapping): # added for mock unit test
380 self._terminated.value = 1
384 # recalculate port for interface and see if it matches ports provided
385 for intf in self.vnfd_helper.interfaces:
387 port = self.vnfd_helper.port_num(name)
389 xe_value = last_result.get(port, {})
391 "rx_throughput_fps": float(xe_value.get("rx_pps", 0.0)),
392 "tx_throughput_fps": float(xe_value.get("tx_pps", 0.0)),
393 "rx_throughput_mbps": float(xe_value.get("rx_bps", 0.0)),
394 "tx_throughput_mbps": float(xe_value.get("tx_bps", 0.0)),
395 "in_packets": int(xe_value.get("ipackets", 0)),
396 "out_packets": int(xe_value.get("opackets", 0)),
399 samples[name][key] = key_value
402 def _run_traffic_once(self, traffic_profile):
403 traffic_profile.execute_traffic(self)
404 self.client_started.value = 1
405 time.sleep(self.RUN_DURATION)
406 samples = self.generate_samples(traffic_profile.ports)
407 time.sleep(self.QUEUE_WAIT_TIME)
408 self._queue.put(samples)
410 def run_traffic(self, traffic_profile):
411 # if we don't do this we can hang waiting for the queue to drain
412 # have to do this in the subprocess
413 self._queue.cancel_join_thread()
414 # fixme: fix passing correct trex config file,
415 # instead of searching the default path
418 self.client = self._connect()
419 self.client.reset(ports=self.all_ports)
420 self.client.remove_all_streams(self.all_ports) # remove all streams
421 traffic_profile.register_generator(self)
423 while self._terminated.value == 0:
424 self._run_traffic_once(traffic_profile)
426 self.client.stop(self.all_ports)
427 self.client.disconnect()
428 self._terminated.value = 0
430 if self._terminated.value:
431 LOG.debug("traffic generator is stopped")
432 return # return if trex/tg server is stopped.
436 self._terminated.value = 1 # stop client
438 def clear_stats(self, ports=None):
440 ports = self.all_ports
441 self.client.clear_stats(ports=ports)
443 def start(self, ports=None, *args, **kwargs):
444 # pylint: disable=keyword-arg-before-vararg
445 # NOTE(ralonsoh): defining keyworded arguments before variable
446 # positional arguments is a bug. This function definition doesn't work
447 # in Python 2, although it works in Python 3. Reference:
448 # https://www.python.org/dev/peps/pep-3102/
450 ports = self.all_ports
451 self.client.start(ports=ports, *args, **kwargs)
453 def collect_kpi(self):
454 if not self._queue.empty():
455 kpi = self._queue.get()
456 self._result.update(kpi)
457 LOG.debug('Got KPIs from _queue for %s %s',
458 self.scenario_helper.name, self.RESOURCE_WORD)
461 def _connect(self, client=None):
463 client = STLClient(username=self.vnfd_helper.mgmt_interface["user"],
464 server=self.vnfd_helper.mgmt_interface["ip"],
465 verbose_level=LoggerApi.VERBOSE_QUIET)
467 # try to connect with 5s intervals, 30s max
473 LOG.info("Unable to connect to Trex Server.. Attempt %s", idx)
478 class Rfc2544ResourceHelper(object):
480 DEFAULT_CORRELATED_TRAFFIC = False
481 DEFAULT_LATENCY = False
482 DEFAULT_TOLERANCE = '0.0001 - 0.0001'
484 def __init__(self, scenario_helper):
485 super(Rfc2544ResourceHelper, self).__init__()
486 self.scenario_helper = scenario_helper
487 self._correlated_traffic = None
488 self.iteration = Value('i', 0)
491 self._tolerance_low = None
492 self._tolerance_high = None
496 if self._rfc2544 is None:
497 self._rfc2544 = self.scenario_helper.all_options['rfc2544']
501 def tolerance_low(self):
502 if self._tolerance_low is None:
503 self.get_rfc_tolerance()
504 return self._tolerance_low
507 def tolerance_high(self):
508 if self._tolerance_high is None:
509 self.get_rfc_tolerance()
510 return self._tolerance_high
513 def correlated_traffic(self):
514 if self._correlated_traffic is None:
515 self._correlated_traffic = \
516 self.get_rfc2544('correlated_traffic', self.DEFAULT_CORRELATED_TRAFFIC)
518 return self._correlated_traffic
522 if self._latency is None:
523 self._latency = self.get_rfc2544('latency', self.DEFAULT_LATENCY)
526 def get_rfc2544(self, name, default=None):
527 return self.rfc2544.get(name, default)
529 def get_rfc_tolerance(self):
530 tolerance_str = self.get_rfc2544('allowed_drop_rate', self.DEFAULT_TOLERANCE)
531 tolerance_iter = iter(sorted(float(t.strip()) for t in tolerance_str.split('-')))
532 self._tolerance_low = next(tolerance_iter)
533 self._tolerance_high = next(tolerance_iter, self.tolerance_low)
536 class SampleVNFDeployHelper(object):
538 SAMPLE_VNF_REPO = 'https://gerrit.opnfv.org/gerrit/samplevnf'
539 REPO_NAME = posixpath.basename(SAMPLE_VNF_REPO)
540 SAMPLE_REPO_DIR = os.path.join('~/', REPO_NAME)
542 def __init__(self, vnfd_helper, ssh_helper):
543 super(SampleVNFDeployHelper, self).__init__()
544 self.ssh_helper = ssh_helper
545 self.vnfd_helper = vnfd_helper
547 def deploy_vnfs(self, app_name):
548 vnf_bin = self.ssh_helper.join_bin_path(app_name)
549 exit_status = self.ssh_helper.execute("which %s" % vnf_bin)[0]
553 subprocess.check_output(["rm", "-rf", self.REPO_NAME])
554 subprocess.check_output(["git", "clone", self.SAMPLE_VNF_REPO])
556 self.ssh_helper.execute("rm -rf %s" % self.SAMPLE_REPO_DIR)
557 self.ssh_helper.put(self.REPO_NAME, self.SAMPLE_REPO_DIR, True)
559 build_script = os.path.join(self.SAMPLE_REPO_DIR, 'tools/vnf_build.sh')
561 http_proxy = os.environ.get('http_proxy', '')
562 cmd = "sudo -E %s -s -p='%s'" % (build_script, http_proxy)
564 self.ssh_helper.execute(cmd)
565 vnf_bin_loc = os.path.join(self.SAMPLE_REPO_DIR, "VNFs", app_name, "build", app_name)
566 self.ssh_helper.execute("sudo mkdir -p %s" % self.ssh_helper.bin_path)
567 self.ssh_helper.execute("sudo cp %s %s" % (vnf_bin_loc, vnf_bin))
570 class ScenarioHelper(object):
575 'worker_config': '1C/1T',
579 def __init__(self, name):
581 self.scenario_cfg = None
585 return self.scenario_cfg['task_path']
589 return self.scenario_cfg.get('nodes')
592 def all_options(self):
593 return self.scenario_cfg.get('options', {})
597 return self.all_options.get(self.name, {})
601 return self.options.get('vnf_config', self.DEFAULT_VNF_CFG)
605 return self.scenario_cfg['topology']
609 test_duration = self.scenario_cfg.get('runner', {}).get('duration',
610 self.options.get('timeout', constants.DEFAULT_VNF_TIMEOUT))
611 test_timeout = self.options.get('timeout', constants.DEFAULT_VNF_TIMEOUT)
612 return test_duration if test_duration > test_timeout else test_timeout
614 class SampleVNF(GenericVNF):
615 """ Class providing file-like API for generic VNF implementation """
617 VNF_PROMPT = "pipeline>"
619 WAIT_TIME_FOR_SCRIPT = 10
620 APP_NAME = "SampleVNF"
621 # we run the VNF interactively, so the ssh command will timeout after this long
623 def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
624 super(SampleVNF, self).__init__(name, vnfd)
625 self.bin_path = get_nsb_option('bin_path', '')
627 self.scenario_helper = ScenarioHelper(self.name)
628 self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path)
630 if setup_env_helper_type is None:
631 setup_env_helper_type = SetupEnvHelper
633 self.setup_helper = setup_env_helper_type(self.vnfd_helper,
635 self.scenario_helper)
637 self.deploy_helper = SampleVNFDeployHelper(vnfd, self.ssh_helper)
639 if resource_helper_type is None:
640 resource_helper_type = ResourceHelper
642 self.resource_helper = resource_helper_type(self.setup_helper)
644 self.context_cfg = None
645 self.nfvi_context = None
646 self.pipeline_kwargs = {}
647 self.uplink_ports = None
648 self.downlink_ports = None
649 # NOTE(esm): make QueueFileWrapper invert-able so that we
650 # never have to manage the queues
653 self.queue_wrapper = None
655 self.used_drivers = {}
656 self.vnf_port_pairs = None
657 self._vnf_process = None
659 def _start_vnf(self):
660 self.queue_wrapper = QueueFileWrapper(self.q_in, self.q_out, self.VNF_PROMPT)
661 name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
662 self._vnf_process = Process(name=name, target=self._run)
663 self._vnf_process.start()
665 def _vnf_up_post(self):
668 def instantiate(self, scenario_cfg, context_cfg):
669 self.scenario_helper.scenario_cfg = scenario_cfg
670 self.context_cfg = context_cfg
671 self.nfvi_context = Context.get_context_from_server(self.scenario_helper.nodes[self.name])
672 # self.nfvi_context = None
674 # vnf deploy is unsupported, use ansible playbooks
675 if self.scenario_helper.options.get("vnf_deploy", False):
676 self.deploy_helper.deploy_vnfs(self.APP_NAME)
677 self.resource_helper.setup()
680 def wait_for_instantiate(self):
682 time.sleep(self.WAIT_TIME) # Give some time for config to load
684 if not self._vnf_process.is_alive():
685 raise RuntimeError("%s VNF process died." % self.APP_NAME)
687 # NOTE(esm): move to QueueFileWrapper
688 while self.q_out.qsize() > 0:
689 buf.append(self.q_out.get())
690 message = ''.join(buf)
691 if self.VNF_PROMPT in message:
692 LOG.info("%s VNF is up and running.", self.APP_NAME)
694 self.queue_wrapper.clear()
695 self.resource_helper.start_collect()
696 return self._vnf_process.exitcode
698 if "PANIC" in message:
699 raise RuntimeError("Error starting %s VNF." %
702 LOG.info("Waiting for %s VNF to start.. ", self.APP_NAME)
703 time.sleep(self.WAIT_TIME_FOR_SCRIPT)
704 # Send ENTER to display a new prompt in case the prompt text was corrupted
705 # by other VNF output
706 self.q_in.put('\r\n')
708 def _build_run_kwargs(self):
710 'stdin': self.queue_wrapper,
711 'stdout': self.queue_wrapper,
712 'keep_stdin_open': True,
714 'timeout': self.scenario_helper.timeout,
717 def _build_config(self):
718 return self.setup_helper.build_config()
721 # we can't share ssh paramiko objects to force new connection
722 self.ssh_helper.drop_connection()
723 cmd = self._build_config()
724 # kill before starting
725 self.setup_helper.kill_vnf()
728 self._build_run_kwargs()
729 self.ssh_helper.run(cmd, **self.run_kwargs)
731 def vnf_execute(self, cmd, wait_time=2):
732 """ send cmd to vnf process """
734 LOG.info("%s command: %s", self.APP_NAME, cmd)
735 self.q_in.put("{}\r\n".format(cmd))
736 time.sleep(wait_time)
738 while self.q_out.qsize() > 0:
739 output.append(self.q_out.get())
740 return "".join(output)
742 def _tear_down(self):
746 self.vnf_execute("quit")
747 self.setup_helper.kill_vnf()
749 self.resource_helper.stop_collect()
750 if self._vnf_process is not None:
751 # be proper and join first before we kill
752 LOG.debug("joining before terminate %s", self._vnf_process.name)
753 self._vnf_process.join(constants.PROCESS_JOIN_TIMEOUT)
754 self._vnf_process.terminate()
755 # no terminate children here because we share processes with tg
757 def get_stats(self, *args, **kwargs): # pylint: disable=unused-argument
758 """Method for checking the statistics
760 This method could be overridden in children classes.
762 :return: VNF statistics
764 cmd = 'p {0} stats'.format(self.APP_WORD)
765 out = self.vnf_execute(cmd)
768 def collect_kpi(self):
769 # we can't get KPIs if the VNF is down
770 check_if_process_failed(self._vnf_process)
771 stats = self.get_stats()
772 m = re.search(self.COLLECT_KPI, stats, re.MULTILINE)
774 result = {k: int(m.group(v)) for k, v in self.COLLECT_MAP.items()}
775 result["collect_stats"] = self.resource_helper.collect_kpi()
780 "packets_dropped": 0,
782 LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
785 def scale(self, flavor=""):
786 """The SampleVNF base class doesn't provide the 'scale' feature"""
787 raise y_exceptions.FunctionNotImplemented(
788 function_name='scale', class_name='SampleVNFTrafficGen')
791 class SampleVNFTrafficGen(GenericTrafficGen):
792 """ Class providing file-like API for generic traffic generator """
797 def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
798 super(SampleVNFTrafficGen, self).__init__(name, vnfd)
799 self.bin_path = get_nsb_option('bin_path', '')
801 self.scenario_helper = ScenarioHelper(self.name)
802 self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path, wait=True)
804 if setup_env_helper_type is None:
805 setup_env_helper_type = SetupEnvHelper
807 self.setup_helper = setup_env_helper_type(self.vnfd_helper,
809 self.scenario_helper)
811 if resource_helper_type is None:
812 resource_helper_type = ClientResourceHelper
814 self.resource_helper = resource_helper_type(self.setup_helper)
816 self.runs_traffic = True
817 self.traffic_finished = False
818 self._tg_process = None
819 self._traffic_process = None
821 def _start_server(self):
822 # we can't share ssh paramiko objects to force new connection
823 self.ssh_helper.drop_connection()
825 def instantiate(self, scenario_cfg, context_cfg):
826 self.scenario_helper.scenario_cfg = scenario_cfg
827 self.resource_helper.setup()
828 # must generate_cfg after DPDK bind because we need port number
829 self.resource_helper.generate_cfg()
831 LOG.info("Starting %s server...", self.APP_NAME)
832 name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
833 self._tg_process = Process(name=name, target=self._start_server)
834 self._tg_process.start()
836 def _check_status(self):
837 raise NotImplementedError
839 def _wait_for_process(self):
841 if not self._tg_process.is_alive():
842 raise RuntimeError("%s traffic generator process died." % self.APP_NAME)
843 LOG.info("Waiting for %s TG Server to start.. ", self.APP_NAME)
845 status = self._check_status()
847 LOG.info("%s TG Server is up and running.", self.APP_NAME)
848 return self._tg_process.exitcode
850 def _traffic_runner(self, traffic_profile):
851 # always drop connections first thing in new processes
852 # so we don't get paramiko errors
853 self.ssh_helper.drop_connection()
854 LOG.info("Starting %s client...", self.APP_NAME)
855 self.resource_helper.run_traffic(traffic_profile)
857 def run_traffic(self, traffic_profile):
858 """ Generate traffic on the wire according to the given params.
859 Method is non-blocking, returns immediately when traffic process
860 is running. Mandatory.
862 :param traffic_profile:
865 name = "{}-{}-{}-{}".format(self.name, self.APP_NAME, traffic_profile.__class__.__name__,
867 self._traffic_process = Process(name=name, target=self._traffic_runner,
868 args=(traffic_profile,))
869 self._traffic_process.start()
870 # Wait for traffic process to start
871 while self.resource_helper.client_started.value == 0:
872 time.sleep(self.RUN_WAIT)
873 # what if traffic process takes a few seconds to start?
874 if not self._traffic_process.is_alive():
877 return self._traffic_process.is_alive()
879 def collect_kpi(self):
880 # check if the tg processes have exited
881 for proc in (self._tg_process, self._traffic_process):
882 check_if_process_failed(proc)
883 result = self.resource_helper.collect_kpi()
884 LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
888 """ After this method finishes, all traffic processes should stop. Mandatory.
892 self.traffic_finished = True
893 # we must kill client before we kill the server, or the client will raise exception
894 if self._traffic_process is not None:
895 # be proper and try to join before terminating
896 LOG.debug("joining before terminate %s", self._traffic_process.name)
897 self._traffic_process.join(constants.PROCESS_JOIN_TIMEOUT)
898 self._traffic_process.terminate()
899 if self._tg_process is not None:
900 # be proper and try to join before terminating
901 LOG.debug("joining before terminate %s", self._tg_process.name)
902 self._tg_process.join(constants.PROCESS_JOIN_TIMEOUT)
903 self._tg_process.terminate()
904 # no terminate children here because we share processes with vnf
906 def scale(self, flavor=""):
907 """A traffic generator VFN doesn't provide the 'scale' feature"""
908 raise y_exceptions.FunctionNotImplemented(
909 function_name='scale', class_name='SampleVNFTrafficGen')