1 # Copyright (c) 2016-2017 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 __future__ import absolute_import
24 from collections import Mapping
26 from multiprocessing import Queue, Value, Process
28 from six.moves import cStringIO
30 from yardstick.benchmark.contexts.base import Context
31 from yardstick.benchmark.scenarios.networking.vnf_generic import find_relative_file
32 from yardstick.common.process import check_if_process_failed
33 from yardstick.network_services.helpers.cpu import CpuSysCores
34 from yardstick.network_services.helpers.samplevnf_helper import PortPairs
35 from yardstick.network_services.helpers.samplevnf_helper import MultiPortConfig
36 from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper
37 from yardstick.network_services.nfvi.resource import ResourceProfile
38 from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
39 from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper
40 from yardstick.network_services.vnf_generic.vnf.base import GenericTrafficGen
41 from yardstick.network_services.utils import get_nsb_option
43 from trex_stl_lib.trex_stl_client import STLClient
44 from trex_stl_lib.trex_stl_client import LoggerApi
45 from trex_stl_lib.trex_stl_exceptions import STLError
47 from yardstick.ssh import AutoConnectSSH
49 DPDK_VERSION = "dpdk-16.07"
51 LOG = logging.getLogger(__name__)
55 DEFAULT_VNF_TIMEOUT = 3600
56 PROCESS_JOIN_TIMEOUT = 3
59 class VnfSshHelper(AutoConnectSSH):
61 def __init__(self, node, bin_path, wait=None):
63 kwargs = self.args_from_node(self.node)
65 kwargs.setdefault('wait', wait)
67 super(VnfSshHelper, self).__init__(**kwargs)
68 self.bin_path = bin_path
72 # must return static class name, anything else refers to the calling class
73 # i.e. the subclass, not the superclass
77 # this copy constructor is different from SSH classes, since it uses node
78 return self.get_class()(self.node, self.bin_path)
80 def upload_config_file(self, prefix, content):
81 cfg_file = os.path.join(REMOTE_TMP, prefix)
83 file_obj = cStringIO(content)
84 self.put_file_obj(file_obj, cfg_file)
87 def join_bin_path(self, *args):
88 return os.path.join(self.bin_path, *args)
90 def provision_tool(self, tool_path=None, tool_file=None):
92 tool_path = self.bin_path
93 return super(VnfSshHelper, self).provision_tool(tool_path, tool_file)
96 class SetupEnvHelper(object):
98 CFG_CONFIG = os.path.join(REMOTE_TMP, "sample_config")
99 CFG_SCRIPT = os.path.join(REMOTE_TMP, "sample_script")
101 DEFAULT_CONFIG_TPL_CFG = "sample.cfg"
102 PIPELINE_COMMAND = ''
105 def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
106 super(SetupEnvHelper, self).__init__()
107 self.vnfd_helper = vnfd_helper
108 self.ssh_helper = ssh_helper
109 self.scenario_helper = scenario_helper
111 def _get_ports_gateway(self, name):
112 routing_table = self.vnfd_helper.vdu0.get('routing_table', [])
113 for route in routing_table:
114 if name == route['if']:
115 return route['gateway']
118 def build_config(self):
119 raise NotImplementedError
121 def setup_vnf_environment(self):
128 raise NotImplementedError
131 class DpdkVnfSetupEnvHelper(SetupEnvHelper):
134 FIND_NET_CMD = "find /sys/class/net -lname '*{}*' -printf '%f'"
140 def _update_packet_type(ip_pipeline_cfg, traffic_options):
141 match_str = 'pkt_type = ipv4'
142 replace_str = 'pkt_type = {0}'.format(traffic_options['pkt_type'])
143 pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
144 return pipeline_config_str
147 def _update_traffic_type(cls, ip_pipeline_cfg, traffic_options):
148 traffic_type = traffic_options['traffic_type']
150 if traffic_options['vnf_type'] is not cls.APP_NAME:
151 match_str = 'traffic_type = 4'
152 replace_str = 'traffic_type = {0}'.format(traffic_type)
154 elif traffic_type == 4:
155 match_str = 'pkt_type = ipv4'
156 replace_str = 'pkt_type = ipv4'
159 match_str = 'pkt_type = ipv4'
160 replace_str = 'pkt_type = ipv6'
162 pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
163 return pipeline_config_str
165 def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
166 super(DpdkVnfSetupEnvHelper, self).__init__(vnfd_helper, ssh_helper, scenario_helper)
167 self.all_ports = None
168 self.bound_pci = None
170 self.used_drivers = None
171 self.dpdk_bind_helper = DpdkBindHelper(ssh_helper)
173 def _setup_hugepages(self):
174 cmd = "awk '/Hugepagesize/ { print $2$3 }' < /proc/meminfo"
175 hugepages = self.ssh_helper.execute(cmd)[1].rstrip()
178 '/sys/kernel/mm/hugepages/hugepages-%s/nr_hugepages' % hugepages
179 self.ssh_helper.execute("awk -F: '{ print $1 }' < %s" % memory_path)
181 if hugepages == "2048kB":
186 self.ssh_helper.execute("echo %s | sudo tee %s" % (pages, memory_path))
188 def build_config(self):
189 vnf_cfg = self.scenario_helper.vnf_cfg
190 task_path = self.scenario_helper.task_path
192 lb_count = vnf_cfg.get('lb_count', 3)
193 lb_config = vnf_cfg.get('lb_config', 'SW')
194 worker_config = vnf_cfg.get('worker_config', '1C/1T')
195 worker_threads = vnf_cfg.get('worker_threads', 3)
197 traffic_type = self.scenario_helper.all_options.get('traffic_type', 4)
199 'traffic_type': traffic_type,
200 'pkt_type': 'ipv%s' % traffic_type,
201 'vnf_type': self.VNF_TYPE,
204 config_tpl_cfg = find_relative_file(self.DEFAULT_CONFIG_TPL_CFG, task_path)
205 config_basename = posixpath.basename(self.CFG_CONFIG)
206 script_basename = posixpath.basename(self.CFG_SCRIPT)
207 multiport = MultiPortConfig(self.scenario_helper.topology,
218 multiport.generate_config()
219 with open(self.CFG_CONFIG) as handle:
220 new_config = handle.read()
222 new_config = self._update_traffic_type(new_config, traffic_options)
223 new_config = self._update_packet_type(new_config, traffic_options)
225 self.ssh_helper.upload_config_file(config_basename, new_config)
226 self.ssh_helper.upload_config_file(script_basename,
227 multiport.generate_script(self.vnfd_helper))
229 LOG.info("Provision and start the %s", self.APP_NAME)
230 self._build_pipeline_kwargs()
231 return self.PIPELINE_COMMAND.format(**self.pipeline_kwargs)
233 def _build_pipeline_kwargs(self):
234 tool_path = self.ssh_helper.provision_tool(tool_file=self.APP_NAME)
235 # count the number of actual ports in the list of pairs
236 # remove duplicate ports
237 # this is really a mapping from LINK ID to DPDK PMD ID
238 # e.g. 0x110 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_2
239 # 0x1010 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_3
240 ports = self.vnfd_helper.port_pairs.all_ports
241 port_nums = self.vnfd_helper.port_nums(ports)
242 # create mask from all the dpdk port numbers
243 ports_mask_hex = hex(sum(2 ** num for num in port_nums))
244 self.pipeline_kwargs = {
245 'cfg_file': self.CFG_CONFIG,
246 'script': self.CFG_SCRIPT,
247 'port_mask_hex': ports_mask_hex,
248 'tool_path': tool_path,
251 def _get_app_cpu(self):
255 vnf_cfg = self.scenario_helper.vnf_cfg
256 sys_obj = CpuSysCores(self.ssh_helper)
257 self.sys_cpu = sys_obj.get_core_socket()
258 num_core = int(vnf_cfg["worker_threads"])
259 if vnf_cfg.get("lb_config", "SW") == 'HW':
260 num_core += self.HW_DEFAULT_CORE
262 num_core += self.SW_DEFAULT_CORE
263 app_cpu = self.sys_cpu[str(self.socket)][:num_core]
266 def _get_cpu_sibling_list(self, cores=None):
268 cores = self._get_app_cpu()
269 sys_cmd_template = "%s/cpu%s/topology/thread_siblings_list"
270 awk_template = "awk -F: '{ print $1 }' < %s"
271 sys_path = "/sys/devices/system/cpu/"
275 sys_cmd = sys_cmd_template % (sys_path, core)
276 cpu_id = self.ssh_helper.execute(awk_template % sys_cmd)[1]
277 cpu_topology.extend(cpu.strip() for cpu in cpu_id.split(','))
283 def _validate_cpu_cfg(self):
284 return self._get_cpu_sibling_list()
286 def setup_vnf_environment(self):
288 self.bound_pci = [v['virtual-interface']["vpci"] for v in self.vnfd_helper.interfaces]
290 # bind before _setup_resources so we can use dpdk_port_num
291 self._detect_and_bind_drivers()
292 resource = self._setup_resources()
296 # pkill is not matching, debug with pgrep
297 self.ssh_helper.execute("sudo pgrep -lax %s" % self.APP_NAME)
298 self.ssh_helper.execute("sudo ps aux | grep -i %s" % self.APP_NAME)
299 # have to use exact match
300 # try using killall to match
301 self.ssh_helper.execute("sudo killall %s" % self.APP_NAME)
303 def _setup_dpdk(self):
304 """ setup dpdk environment needed for vnf to run """
306 self._setup_hugepages()
307 self.ssh_helper.execute("sudo modprobe uio && sudo modprobe igb_uio")
309 exit_status = self.ssh_helper.execute("lsmod | grep -i igb_uio")[0]
313 dpdk = self.ssh_helper.join_bin_path(DPDK_VERSION)
314 dpdk_setup = self.ssh_helper.provision_tool(tool_file="nsb_setup.sh")
315 exit_status = self.ssh_helper.execute("which {} >/dev/null 2>&1".format(dpdk))[0]
317 self.ssh_helper.execute("bash %s dpdk >/dev/null 2>&1" % dpdk_setup)
319 def get_collectd_options(self):
320 options = self.scenario_helper.all_options.get("collectd", {})
321 # override with specific node settings
322 options.update(self.scenario_helper.options.get("collectd", {}))
325 def _setup_resources(self):
326 # what is this magic? how do we know which socket is for which port?
327 # what about quad-socket?
328 if any(v[5] == "0" for v in self.bound_pci):
333 cores = self._validate_cpu_cfg()
334 # implicit ordering, presumably by DPDK port num, so pre-sort by port_num
335 # this won't work because we don't have DPDK port numbers yet
336 ports = sorted(self.vnfd_helper.interfaces, key=self.vnfd_helper.port_num)
337 port_names = (intf["name"] for intf in ports)
338 collectd_options = self.get_collectd_options()
339 plugins = collectd_options.get("plugins", {})
340 # we must set timeout to be the same as the VNF otherwise KPIs will die before VNF
341 return ResourceProfile(self.vnfd_helper.mgmt_interface, port_names=port_names, cores=cores,
342 plugins=plugins, interval=collectd_options.get("interval"),
343 timeout=self.scenario_helper.timeout)
345 def _detect_and_bind_drivers(self):
346 interfaces = self.vnfd_helper.interfaces
348 self.dpdk_bind_helper.read_status()
349 self.dpdk_bind_helper.save_used_drivers()
351 self.dpdk_bind_helper.bind(self.bound_pci, 'igb_uio')
353 sorted_dpdk_pci_addresses = sorted(self.dpdk_bind_helper.dpdk_bound_pci_addresses)
354 for dpdk_port_num, vpci in enumerate(sorted_dpdk_pci_addresses):
356 intf = next(v for v in interfaces
357 if vpci == v['virtual-interface']['vpci'])
359 intf['virtual-interface']['dpdk_port_num'] = int(dpdk_port_num)
364 def get_local_iface_name_by_vpci(self, vpci):
365 find_net_cmd = self.FIND_NET_CMD.format(vpci)
366 exit_status, stdout, _ = self.ssh_helper.execute(find_net_cmd)
372 self.dpdk_bind_helper.rebind_drivers()
375 class ResourceHelper(object):
378 MAKE_INSTALL = 'cd {0} && make && sudo make install'
379 RESOURCE_WORD = 'sample'
383 def __init__(self, setup_helper):
384 super(ResourceHelper, self).__init__()
386 self.setup_helper = setup_helper
387 self.ssh_helper = setup_helper.ssh_helper
390 self.resource = self.setup_helper.setup_vnf_environment()
392 def generate_cfg(self):
395 def _collect_resource_kpi(self):
397 status = self.resource.check_if_sa_running("collectd")[0]
399 result = self.resource.amqp_collect_nfvi_kpi()
401 result = {"core": result}
404 def start_collect(self):
405 self.resource.initiate_systemagent(self.ssh_helper.bin_path)
406 self.resource.start()
407 self.resource.amqp_process_for_nfvi_kpi()
409 def stop_collect(self):
413 def collect_kpi(self):
414 return self._collect_resource_kpi()
417 class ClientResourceHelper(ResourceHelper):
424 def __init__(self, setup_helper):
425 super(ClientResourceHelper, self).__init__(setup_helper)
426 self.vnfd_helper = setup_helper.vnfd_helper
427 self.scenario_helper = setup_helper.scenario_helper
430 self.client_started = Value('i', 0)
431 self.all_ports = None
432 self._queue = Queue()
434 self._terminated = Value('i', 0)
436 def _build_ports(self):
437 self.networks = self.vnfd_helper.port_pairs.networks
438 self.uplink_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.uplink_ports)
439 self.downlink_ports = \
440 self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.downlink_ports)
441 self.all_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.all_ports)
443 def get_stats(self, *args, **kwargs):
445 return self.client.get_stats(*args, **kwargs)
447 LOG.exception("TRex client not connected")
450 def generate_samples(self, ports, key=None, default=None):
451 # needs to be used ports
452 last_result = self.get_stats(ports)
453 key_value = last_result.get(key, default)
455 if not isinstance(last_result, Mapping): # added for mock unit test
456 self._terminated.value = 1
460 # recalculate port for interface and see if it matches ports provided
461 for intf in self.vnfd_helper.interfaces:
463 port = self.vnfd_helper.port_num(name)
465 xe_value = last_result.get(port, {})
467 "rx_throughput_fps": float(xe_value.get("rx_pps", 0.0)),
468 "tx_throughput_fps": float(xe_value.get("tx_pps", 0.0)),
469 "rx_throughput_mbps": float(xe_value.get("rx_bps", 0.0)),
470 "tx_throughput_mbps": float(xe_value.get("tx_bps", 0.0)),
471 "in_packets": int(xe_value.get("ipackets", 0)),
472 "out_packets": int(xe_value.get("opackets", 0)),
475 samples[name][key] = key_value
478 def _run_traffic_once(self, traffic_profile):
479 traffic_profile.execute_traffic(self)
480 self.client_started.value = 1
481 time.sleep(self.RUN_DURATION)
482 samples = self.generate_samples(traffic_profile.ports)
483 time.sleep(self.QUEUE_WAIT_TIME)
484 self._queue.put(samples)
486 def run_traffic(self, traffic_profile):
487 # if we don't do this we can hang waiting for the queue to drain
488 # have to do this in the subprocess
489 self._queue.cancel_join_thread()
490 # fixme: fix passing correct trex config file,
491 # instead of searching the default path
494 self.client = self._connect()
495 self.client.reset(ports=self.all_ports)
496 self.client.remove_all_streams(self.all_ports) # remove all streams
497 traffic_profile.register_generator(self)
499 while self._terminated.value == 0:
500 self._run_traffic_once(traffic_profile)
502 self.client.stop(self.all_ports)
503 self.client.disconnect()
504 self._terminated.value = 0
506 if self._terminated.value:
507 LOG.debug("traffic generator is stopped")
508 return # return if trex/tg server is stopped.
512 self._terminated.value = 1 # stop client
514 def clear_stats(self, ports=None):
516 ports = self.all_ports
517 self.client.clear_stats(ports=ports)
519 def start(self, ports=None, *args, **kwargs):
521 ports = self.all_ports
522 self.client.start(ports=ports, *args, **kwargs)
524 def collect_kpi(self):
525 if not self._queue.empty():
526 kpi = self._queue.get()
527 self._result.update(kpi)
528 LOG.debug("Got KPIs from _queue for {0} {1}".format(
529 self.scenario_helper.name, self.RESOURCE_WORD))
532 def _connect(self, client=None):
534 client = STLClient(username=self.vnfd_helper.mgmt_interface["user"],
535 server=self.vnfd_helper.mgmt_interface["ip"],
536 verbose_level=LoggerApi.VERBOSE_QUIET)
538 # try to connect with 5s intervals, 30s max
544 LOG.info("Unable to connect to Trex Server.. Attempt %s", idx)
549 class Rfc2544ResourceHelper(object):
551 DEFAULT_CORRELATED_TRAFFIC = False
552 DEFAULT_LATENCY = False
553 DEFAULT_TOLERANCE = '0.0001 - 0.0001'
555 def __init__(self, scenario_helper):
556 super(Rfc2544ResourceHelper, self).__init__()
557 self.scenario_helper = scenario_helper
558 self._correlated_traffic = None
559 self.iteration = Value('i', 0)
562 self._tolerance_low = None
563 self._tolerance_high = None
567 if self._rfc2544 is None:
568 self._rfc2544 = self.scenario_helper.all_options['rfc2544']
572 def tolerance_low(self):
573 if self._tolerance_low is None:
574 self.get_rfc_tolerance()
575 return self._tolerance_low
578 def tolerance_high(self):
579 if self._tolerance_high is None:
580 self.get_rfc_tolerance()
581 return self._tolerance_high
584 def correlated_traffic(self):
585 if self._correlated_traffic is None:
586 self._correlated_traffic = \
587 self.get_rfc2544('correlated_traffic', self.DEFAULT_CORRELATED_TRAFFIC)
589 return self._correlated_traffic
593 if self._latency is None:
594 self._latency = self.get_rfc2544('latency', self.DEFAULT_LATENCY)
597 def get_rfc2544(self, name, default=None):
598 return self.rfc2544.get(name, default)
600 def get_rfc_tolerance(self):
601 tolerance_str = self.get_rfc2544('allowed_drop_rate', self.DEFAULT_TOLERANCE)
602 tolerance_iter = iter(sorted(float(t.strip()) for t in tolerance_str.split('-')))
603 self._tolerance_low = next(tolerance_iter)
604 self._tolerance_high = next(tolerance_iter, self.tolerance_low)
607 class SampleVNFDeployHelper(object):
609 SAMPLE_VNF_REPO = 'https://gerrit.opnfv.org/gerrit/samplevnf'
610 REPO_NAME = posixpath.basename(SAMPLE_VNF_REPO)
611 SAMPLE_REPO_DIR = os.path.join('~/', REPO_NAME)
613 def __init__(self, vnfd_helper, ssh_helper):
614 super(SampleVNFDeployHelper, self).__init__()
615 self.ssh_helper = ssh_helper
616 self.vnfd_helper = vnfd_helper
618 def deploy_vnfs(self, app_name):
619 vnf_bin = self.ssh_helper.join_bin_path(app_name)
620 exit_status = self.ssh_helper.execute("which %s" % vnf_bin)[0]
624 subprocess.check_output(["rm", "-rf", self.REPO_NAME])
625 subprocess.check_output(["git", "clone", self.SAMPLE_VNF_REPO])
627 self.ssh_helper.execute("rm -rf %s" % self.SAMPLE_REPO_DIR)
628 self.ssh_helper.put(self.REPO_NAME, self.SAMPLE_REPO_DIR, True)
630 build_script = os.path.join(self.SAMPLE_REPO_DIR, 'tools/vnf_build.sh')
632 http_proxy = os.environ.get('http_proxy', '')
633 cmd = "sudo -E %s -s -p='%s'" % (build_script, http_proxy)
635 self.ssh_helper.execute(cmd)
636 vnf_bin_loc = os.path.join(self.SAMPLE_REPO_DIR, "VNFs", app_name, "build", app_name)
637 self.ssh_helper.execute("sudo mkdir -p %s" % self.ssh_helper.bin_path)
638 self.ssh_helper.execute("sudo cp %s %s" % (vnf_bin_loc, vnf_bin))
641 class ScenarioHelper(object):
646 'worker_config': '1C/1T',
650 def __init__(self, name):
652 self.scenario_cfg = None
656 return self.scenario_cfg['task_path']
660 return self.scenario_cfg.get('nodes')
663 def all_options(self):
664 return self.scenario_cfg.get('options', {})
668 return self.all_options.get(self.name, {})
672 return self.options.get('vnf_config', self.DEFAULT_VNF_CFG)
676 return self.scenario_cfg['topology']
680 return self.options.get('timeout', DEFAULT_VNF_TIMEOUT)
683 class SampleVNF(GenericVNF):
684 """ Class providing file-like API for generic VNF implementation """
686 VNF_PROMPT = "pipeline>"
688 WAIT_TIME_FOR_SCRIPT = 10
689 APP_NAME = "SampleVNF"
690 # we run the VNF interactively, so the ssh command will timeout after this long
692 def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
693 super(SampleVNF, self).__init__(name, vnfd)
694 self.bin_path = get_nsb_option('bin_path', '')
696 self.scenario_helper = ScenarioHelper(self.name)
697 self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path)
699 if setup_env_helper_type is None:
700 setup_env_helper_type = SetupEnvHelper
702 self.setup_helper = setup_env_helper_type(self.vnfd_helper,
704 self.scenario_helper)
706 self.deploy_helper = SampleVNFDeployHelper(vnfd, self.ssh_helper)
708 if resource_helper_type is None:
709 resource_helper_type = ResourceHelper
711 self.resource_helper = resource_helper_type(self.setup_helper)
713 self.context_cfg = None
714 self.nfvi_context = None
715 self.pipeline_kwargs = {}
716 self.uplink_ports = None
717 self.downlink_ports = None
718 # TODO(esm): make QueueFileWrapper invert-able so that we
719 # never have to manage the queues
722 self.queue_wrapper = None
724 self.used_drivers = {}
725 self.vnf_port_pairs = None
726 self._vnf_process = None
728 def _build_ports(self):
729 self._port_pairs = PortPairs(self.vnfd_helper.interfaces)
730 self.networks = self._port_pairs.networks
731 self.uplink_ports = self.vnfd_helper.port_nums(self._port_pairs.uplink_ports)
732 self.downlink_ports = self.vnfd_helper.port_nums(self._port_pairs.downlink_ports)
733 self.my_ports = self.vnfd_helper.port_nums(self._port_pairs.all_ports)
735 def _get_route_data(self, route_index, route_type):
736 route_iter = iter(self.vnfd_helper.vdu0.get('nd_route_tbl', []))
737 for _ in range(route_index):
739 return next(route_iter, {}).get(route_type, '')
741 def _get_port0localip6(self):
742 return_value = self._get_route_data(0, 'network')
743 LOG.info("_get_port0localip6 : %s", return_value)
746 def _get_port1localip6(self):
747 return_value = self._get_route_data(1, 'network')
748 LOG.info("_get_port1localip6 : %s", return_value)
751 def _get_port0prefixlen6(self):
752 return_value = self._get_route_data(0, 'netmask')
753 LOG.info("_get_port0prefixlen6 : %s", return_value)
756 def _get_port1prefixlen6(self):
757 return_value = self._get_route_data(1, 'netmask')
758 LOG.info("_get_port1prefixlen6 : %s", return_value)
761 def _get_port0gateway6(self):
762 return_value = self._get_route_data(0, 'network')
763 LOG.info("_get_port0gateway6 : %s", return_value)
766 def _get_port1gateway6(self):
767 return_value = self._get_route_data(1, 'network')
768 LOG.info("_get_port1gateway6 : %s", return_value)
771 def _start_vnf(self):
772 self.queue_wrapper = QueueFileWrapper(self.q_in, self.q_out, self.VNF_PROMPT)
773 name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
774 self._vnf_process = Process(name=name, target=self._run)
775 self._vnf_process.start()
777 def _vnf_up_post(self):
780 def instantiate(self, scenario_cfg, context_cfg):
781 self.scenario_helper.scenario_cfg = scenario_cfg
782 self.context_cfg = context_cfg
783 self.nfvi_context = Context.get_context_from_server(self.scenario_helper.nodes[self.name])
784 # self.nfvi_context = None
786 # vnf deploy is unsupported, use ansible playbooks
787 if self.scenario_helper.options.get("vnf_deploy", False):
788 self.deploy_helper.deploy_vnfs(self.APP_NAME)
789 self.resource_helper.setup()
792 def wait_for_instantiate(self):
794 time.sleep(self.WAIT_TIME) # Give some time for config to load
796 if not self._vnf_process.is_alive():
797 raise RuntimeError("%s VNF process died." % self.APP_NAME)
799 # TODO(esm): move to QueueFileWrapper
800 while self.q_out.qsize() > 0:
801 buf.append(self.q_out.get())
802 message = ''.join(buf)
803 if self.VNF_PROMPT in message:
804 LOG.info("%s VNF is up and running.", self.APP_NAME)
806 self.queue_wrapper.clear()
807 self.resource_helper.start_collect()
808 return self._vnf_process.exitcode
810 if "PANIC" in message:
811 raise RuntimeError("Error starting %s VNF." %
814 LOG.info("Waiting for %s VNF to start.. ", self.APP_NAME)
815 time.sleep(self.WAIT_TIME_FOR_SCRIPT)
816 # Send ENTER to display a new prompt in case the prompt text was corrupted
817 # by other VNF output
818 self.q_in.put('\r\n')
820 def _build_run_kwargs(self):
822 'stdin': self.queue_wrapper,
823 'stdout': self.queue_wrapper,
824 'keep_stdin_open': True,
826 'timeout': self.scenario_helper.timeout,
829 def _build_config(self):
830 return self.setup_helper.build_config()
833 # we can't share ssh paramiko objects to force new connection
834 self.ssh_helper.drop_connection()
835 cmd = self._build_config()
836 # kill before starting
837 self.setup_helper.kill_vnf()
840 self._build_run_kwargs()
841 self.ssh_helper.run(cmd, **self.run_kwargs)
843 def vnf_execute(self, cmd, wait_time=2):
844 """ send cmd to vnf process """
846 LOG.info("%s command: %s", self.APP_NAME, cmd)
847 self.q_in.put("{}\r\n".format(cmd))
848 time.sleep(wait_time)
850 while self.q_out.qsize() > 0:
851 output.append(self.q_out.get())
852 return "".join(output)
854 def _tear_down(self):
858 self.vnf_execute("quit")
859 self.setup_helper.kill_vnf()
861 self.resource_helper.stop_collect()
862 if self._vnf_process is not None:
863 # be proper and join first before we kill
864 LOG.debug("joining before terminate %s", self._vnf_process.name)
865 self._vnf_process.join(PROCESS_JOIN_TIMEOUT)
866 self._vnf_process.terminate()
867 # no terminate children here because we share processes with tg
869 def get_stats(self, *args, **kwargs):
871 Method for checking the statistics
876 cmd = 'p {0} stats'.format(self.APP_WORD)
877 out = self.vnf_execute(cmd)
880 def collect_kpi(self):
881 # we can't get KPIs if the VNF is down
882 check_if_process_failed(self._vnf_process)
883 stats = self.get_stats()
884 m = re.search(self.COLLECT_KPI, stats, re.MULTILINE)
886 result = {k: int(m.group(v)) for k, v in self.COLLECT_MAP.items()}
887 result["collect_stats"] = self.resource_helper.collect_kpi()
892 "packets_dropped": 0,
894 LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
898 class SampleVNFTrafficGen(GenericTrafficGen):
899 """ Class providing file-like API for generic traffic generator """
904 def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
905 super(SampleVNFTrafficGen, self).__init__(name, vnfd)
906 self.bin_path = get_nsb_option('bin_path', '')
908 self.scenario_helper = ScenarioHelper(self.name)
909 self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path, wait=True)
911 if setup_env_helper_type is None:
912 setup_env_helper_type = SetupEnvHelper
914 self.setup_helper = setup_env_helper_type(self.vnfd_helper,
916 self.scenario_helper)
918 if resource_helper_type is None:
919 resource_helper_type = ClientResourceHelper
921 self.resource_helper = resource_helper_type(self.setup_helper)
923 self.runs_traffic = True
924 self.traffic_finished = False
925 self._tg_process = None
926 self._traffic_process = None
928 def _start_server(self):
929 # we can't share ssh paramiko objects to force new connection
930 self.ssh_helper.drop_connection()
932 def instantiate(self, scenario_cfg, context_cfg):
933 self.scenario_helper.scenario_cfg = scenario_cfg
934 self.resource_helper.generate_cfg()
935 self.resource_helper.setup()
937 LOG.info("Starting %s server...", self.APP_NAME)
938 name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
939 self._tg_process = Process(name=name, target=self._start_server)
940 self._tg_process.start()
942 def wait_for_instantiate(self):
943 # overridden by subclasses
944 return self._wait_for_process()
946 def _check_status(self):
947 raise NotImplementedError
949 def _wait_for_process(self):
951 if not self._tg_process.is_alive():
952 raise RuntimeError("%s traffic generator process died." % self.APP_NAME)
953 LOG.info("Waiting for %s TG Server to start.. ", self.APP_NAME)
955 status = self._check_status()
957 LOG.info("%s TG Server is up and running.", self.APP_NAME)
958 return self._tg_process.exitcode
960 def _traffic_runner(self, traffic_profile):
961 # always drop connections first thing in new processes
962 # so we don't get paramiko errors
963 self.ssh_helper.drop_connection()
964 LOG.info("Starting %s client...", self.APP_NAME)
965 self.resource_helper.run_traffic(traffic_profile)
967 def run_traffic(self, traffic_profile):
968 """ Generate traffic on the wire according to the given params.
969 Method is non-blocking, returns immediately when traffic process
970 is running. Mandatory.
972 :param traffic_profile:
975 name = "{}-{}-{}-{}".format(self.name, self.APP_NAME, traffic_profile.__class__.__name__,
977 self._traffic_process = Process(name=name, target=self._traffic_runner,
978 args=(traffic_profile,))
979 self._traffic_process.start()
980 # Wait for traffic process to start
981 while self.resource_helper.client_started.value == 0:
982 time.sleep(self.RUN_WAIT)
983 # what if traffic process takes a few seconds to start?
984 if not self._traffic_process.is_alive():
987 return self._traffic_process.is_alive()
989 def listen_traffic(self, traffic_profile):
990 """ Listen to traffic with the given parameters.
991 Method is non-blocking, returns immediately when traffic process
992 is running. Optional.
994 :param traffic_profile:
999 def verify_traffic(self, traffic_profile):
1000 """ Verify captured traffic after it has ended. Optional.
1002 :param traffic_profile:
1007 def collect_kpi(self):
1008 # check if the tg processes have exited
1009 for proc in (self._tg_process, self._traffic_process):
1010 check_if_process_failed(proc)
1011 result = self.resource_helper.collect_kpi()
1012 LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
1015 def terminate(self):
1016 """ After this method finishes, all traffic processes should stop. Mandatory.
1020 self.traffic_finished = True
1021 # we must kill client before we kill the server, or the client will raise exception
1022 if self._traffic_process is not None:
1023 # be proper and try to join before terminating
1024 LOG.debug("joining before terminate %s", self._traffic_process.name)
1025 self._traffic_process.join(PROCESS_JOIN_TIMEOUT)
1026 self._traffic_process.terminate()
1027 if self._tg_process is not None:
1028 # be proper and try to join before terminating
1029 LOG.debug("joining before terminate %s", self._tg_process.name)
1030 self._tg_process.join(PROCESS_JOIN_TIMEOUT)
1031 self._tg_process.terminate()
1032 # no terminate children here because we share processes with vnf