yardstick/network_services/vnf_generic/vnf/sample_vnf.py

   1 # Copyright (c) 2016-2018 Intel Corporation
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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 """
  15
  16 from collections import Mapping
  17 import logging
  18 from multiprocessing import Queue, Value, Process
  19
  20 import os
  21 import posixpath
  22 import re
  23 import subprocess
  24 import time
  25
  26 import six
  27
  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.constants import DEFAULT_VNF_TIMEOUT
  36 from yardstick.network_services.constants import PROCESS_JOIN_TIMEOUT
  37 from yardstick.network_services.constants import REMOTE_TMP
  38 from yardstick.network_services.helpers.dpdkbindnic_helper import DpdkBindHelper, DpdkNode
  39 from yardstick.network_services.helpers.samplevnf_helper import MultiPortConfig
  40 from yardstick.network_services.helpers.samplevnf_helper import PortPairs
  41 from yardstick.network_services.nfvi.resource import ResourceProfile
  42 from yardstick.network_services.utils import get_nsb_option
  43 from yardstick.network_services.vnf_generic.vnf.base import GenericTrafficGen
  44 from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
  45 from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper
  46 from yardstick.network_services.vnf_generic.vnf.vnf_ssh_helper import VnfSshHelper
  47
  48
  49 LOG = logging.getLogger(__name__)
  50
  51
  52 class SetupEnvHelper(object):
  53
  54     CFG_CONFIG = os.path.join(REMOTE_TMP, "sample_config")
  55     CFG_SCRIPT = os.path.join(REMOTE_TMP, "sample_script")
  56     DEFAULT_CONFIG_TPL_CFG = "sample.cfg"
  57     PIPELINE_COMMAND = ''
  58     VNF_TYPE = "SAMPLE"
  59
  60     def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
  61         super(SetupEnvHelper, self).__init__()
  62         self.vnfd_helper = vnfd_helper
  63         self.ssh_helper = ssh_helper
  64         self.scenario_helper = scenario_helper
  65
  66     def build_config(self):
  67         raise NotImplementedError
  68
  69     def setup_vnf_environment(self):
  70         pass
  71
  72     def kill_vnf(self):
  73         pass
  74
  75     def tear_down(self):
  76         raise NotImplementedError
  77
  78
  79 class DpdkVnfSetupEnvHelper(SetupEnvHelper):
  80
  81     APP_NAME = 'DpdkVnf'
  82     FIND_NET_CMD = "find /sys/class/net -lname '*{}*' -printf '%f'"
  83     NR_HUGEPAGES_PATH = '/proc/sys/vm/nr_hugepages'
  84     HUGEPAGES_KB = 1024 * 1024 * 16
  85
  86     @staticmethod
  87     def _update_packet_type(ip_pipeline_cfg, traffic_options):
  88         match_str = 'pkt_type = ipv4'
  89         replace_str = 'pkt_type = {0}'.format(traffic_options['pkt_type'])
  90         pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
  91         return pipeline_config_str
  92
  93     @classmethod
  94     def _update_traffic_type(cls, ip_pipeline_cfg, traffic_options):
  95         traffic_type = traffic_options['traffic_type']
  96
  97         if traffic_options['vnf_type'] is not cls.APP_NAME:
  98             match_str = 'traffic_type = 4'
  99             replace_str = 'traffic_type = {0}'.format(traffic_type)
 100
 101         elif traffic_type == 4:
 102             match_str = 'pkt_type = ipv4'
 103             replace_str = 'pkt_type = ipv4'
 104
 105         else:
 106             match_str = 'pkt_type = ipv4'
 107             replace_str = 'pkt_type = ipv6'
 108
 109         pipeline_config_str = ip_pipeline_cfg.replace(match_str, replace_str)
 110         return pipeline_config_str
 111
 112     def __init__(self, vnfd_helper, ssh_helper, scenario_helper):
 113         super(DpdkVnfSetupEnvHelper, self).__init__(vnfd_helper, ssh_helper, scenario_helper)
 114         self.all_ports = None
 115         self.bound_pci = None
 116         self.socket = None
 117         self.used_drivers = None
 118         self.dpdk_bind_helper = DpdkBindHelper(ssh_helper)
 119
 120     def _setup_hugepages(self):
 121         meminfo = utils.read_meminfo(self.ssh_helper)
 122         hp_size_kb = int(meminfo['Hugepagesize'])
 123         nr_hugepages = int(abs(self.HUGEPAGES_KB / hp_size_kb))
 124         self.ssh_helper.execute('echo %s | sudo tee %s' %
 125                                 (nr_hugepages, self.NR_HUGEPAGES_PATH))
 126         hp = six.BytesIO()
 127         self.ssh_helper.get_file_obj(self.NR_HUGEPAGES_PATH, hp)
 128         nr_hugepages_set = int(hp.getvalue().decode('utf-8').splitlines()[0])
 129         LOG.info('Hugepages size (kB): %s, number claimed: %s, number set: %s',
 130                  hp_size_kb, nr_hugepages, nr_hugepages_set)
 131
 132     def build_config(self):
 133         vnf_cfg = self.scenario_helper.vnf_cfg
 134         task_path = self.scenario_helper.task_path
 135
 136         config_file = vnf_cfg.get('file')
 137         lb_count = vnf_cfg.get('lb_count', 3)
 138         lb_config = vnf_cfg.get('lb_config', 'SW')
 139         worker_config = vnf_cfg.get('worker_config', '1C/1T')
 140         worker_threads = vnf_cfg.get('worker_threads', 3)
 141
 142         traffic_type = self.scenario_helper.all_options.get('traffic_type', 4)
 143         traffic_options = {
 144             'traffic_type': traffic_type,
 145             'pkt_type': 'ipv%s' % traffic_type,
 146             'vnf_type': self.VNF_TYPE,
 147         }
 148
 149         config_tpl_cfg = utils.find_relative_file(self.DEFAULT_CONFIG_TPL_CFG,
 150                                                   task_path)
 151         config_basename = posixpath.basename(self.CFG_CONFIG)
 152         script_basename = posixpath.basename(self.CFG_SCRIPT)
 153         multiport = MultiPortConfig(self.scenario_helper.topology,
 154                                     config_tpl_cfg,
 155                                     config_basename,
 156                                     self.vnfd_helper,
 157                                     self.VNF_TYPE,
 158                                     lb_count,
 159                                     worker_threads,
 160                                     worker_config,
 161                                     lb_config,
 162                                     self.socket)
 163
 164         multiport.generate_config()
 165         if config_file:
 166             with utils.open_relative_file(config_file, task_path) as infile:
 167                 new_config = ['[EAL]']
 168                 vpci = []
 169                 for port in self.vnfd_helper.port_pairs.all_ports:
 170                     interface = self.vnfd_helper.find_interface(name=port)
 171                     vpci.append(interface['virtual-interface']["vpci"])
 172                 new_config.extend('w = {0}'.format(item) for item in vpci)
 173                 new_config = '\n'.join(new_config) + '\n' + infile.read()
 174         else:
 175             with open(self.CFG_CONFIG) as handle:
 176                 new_config = handle.read()
 177             new_config = self._update_traffic_type(new_config, traffic_options)
 178             new_config = self._update_packet_type(new_config, traffic_options)
 179         self.ssh_helper.upload_config_file(config_basename, new_config)
 180         self.ssh_helper.upload_config_file(script_basename,
 181                                            multiport.generate_script(self.vnfd_helper))
 182
 183         LOG.info("Provision and start the %s", self.APP_NAME)
 184         self._build_pipeline_kwargs()
 185         return self.PIPELINE_COMMAND.format(**self.pipeline_kwargs)
 186
 187     def _build_pipeline_kwargs(self):
 188         tool_path = self.ssh_helper.provision_tool(tool_file=self.APP_NAME)
 189         # count the number of actual ports in the list of pairs
 190         # remove duplicate ports
 191         # this is really a mapping from LINK ID to DPDK PMD ID
 192         # e.g. 0x110 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_2
 193         #      0x1010 maps LINK0 -> PMD_ID_1, LINK1 -> PMD_ID_3
 194         ports = self.vnfd_helper.port_pairs.all_ports
 195         port_nums = self.vnfd_helper.port_nums(ports)
 196         # create mask from all the dpdk port numbers
 197         ports_mask_hex = hex(sum(2 ** num for num in port_nums))
 198         self.pipeline_kwargs = {
 199             'cfg_file': self.CFG_CONFIG,
 200             'script': self.CFG_SCRIPT,
 201             'port_mask_hex': ports_mask_hex,
 202             'tool_path': tool_path,
 203         }
 204
 205     def setup_vnf_environment(self):
 206         self._setup_dpdk()
 207         self.kill_vnf()
 208         # bind before _setup_resources so we can use dpdk_port_num
 209         self._detect_and_bind_drivers()
 210         resource = self._setup_resources()
 211         return resource
 212
 213     def kill_vnf(self):
 214         # pkill is not matching, debug with pgrep
 215         self.ssh_helper.execute("sudo pgrep -lax  %s" % self.APP_NAME)
 216         self.ssh_helper.execute("sudo ps aux | grep -i %s" % self.APP_NAME)
 217         # have to use exact match
 218         # try using killall to match
 219         self.ssh_helper.execute("sudo killall %s" % self.APP_NAME)
 220
 221     def _setup_dpdk(self):
 222         """Setup DPDK environment needed for VNF to run"""
 223         self._setup_hugepages()
 224         self.dpdk_bind_helper.load_dpdk_driver()
 225
 226         exit_status = self.dpdk_bind_helper.check_dpdk_driver()
 227         if exit_status == 0:
 228             return
 229
 230     def get_collectd_options(self):
 231         options = self.scenario_helper.all_options.get("collectd", {})
 232         # override with specific node settings
 233         options.update(self.scenario_helper.options.get("collectd", {}))
 234         return options
 235
 236     def _setup_resources(self):
 237         # what is this magic?  how do we know which socket is for which port?
 238         # what about quad-socket?
 239         if any(v[5] == "0" for v in self.bound_pci):
 240             self.socket = 0
 241         else:
 242             self.socket = 1
 243
 244         # implicit ordering, presumably by DPDK port num, so pre-sort by port_num
 245         # this won't work because we don't have DPDK port numbers yet
 246         ports = sorted(self.vnfd_helper.interfaces, key=self.vnfd_helper.port_num)
 247         port_names = (intf["name"] for intf in ports)
 248         collectd_options = self.get_collectd_options()
 249         plugins = collectd_options.get("plugins", {})
 250         # we must set timeout to be the same as the VNF otherwise KPIs will die before VNF
 251         return ResourceProfile(self.vnfd_helper.mgmt_interface, port_names=port_names,
 252                                plugins=plugins, interval=collectd_options.get("interval"),
 253                                timeout=self.scenario_helper.timeout)
 254
 255     def _check_interface_fields(self):
 256         num_nodes = len(self.scenario_helper.nodes)
 257         # OpenStack instance creation time is probably proportional to the number
 258         # of instances
 259         timeout = 120 * num_nodes
 260         dpdk_node = DpdkNode(self.scenario_helper.name, self.vnfd_helper.interfaces,
 261                              self.ssh_helper, timeout)
 262         dpdk_node.check()
 263
 264     def _detect_and_bind_drivers(self):
 265         interfaces = self.vnfd_helper.interfaces
 266
 267         self._check_interface_fields()
 268         # check for bound after probe
 269         self.bound_pci = [v['virtual-interface']["vpci"] for v in interfaces]
 270
 271         self.dpdk_bind_helper.read_status()
 272         self.dpdk_bind_helper.save_used_drivers()
 273
 274         self.dpdk_bind_helper.bind(self.bound_pci, 'igb_uio')
 275
 276         sorted_dpdk_pci_addresses = sorted(self.dpdk_bind_helper.dpdk_bound_pci_addresses)
 277         for dpdk_port_num, vpci in enumerate(sorted_dpdk_pci_addresses):
 278             try:
 279                 intf = next(v for v in interfaces
 280                             if vpci == v['virtual-interface']['vpci'])
 281                 # force to int
 282                 intf['virtual-interface']['dpdk_port_num'] = int(dpdk_port_num)
 283             except:  # pylint: disable=bare-except
 284                 pass
 285         time.sleep(2)
 286
 287     def get_local_iface_name_by_vpci(self, vpci):
 288         find_net_cmd = self.FIND_NET_CMD.format(vpci)
 289         exit_status, stdout, _ = self.ssh_helper.execute(find_net_cmd)
 290         if exit_status == 0:
 291             return stdout
 292         return None
 293
 294     def tear_down(self):
 295         self.dpdk_bind_helper.rebind_drivers()
 296
 297
 298 class ResourceHelper(object):
 299
 300     COLLECT_KPI = ''
 301     MAKE_INSTALL = 'cd {0} && make && sudo make install'
 302     RESOURCE_WORD = 'sample'
 303
 304     COLLECT_MAP = {}
 305
 306     def __init__(self, setup_helper):
 307         super(ResourceHelper, self).__init__()
 308         self.resource = None
 309         self.setup_helper = setup_helper
 310         self.ssh_helper = setup_helper.ssh_helper
 311
 312     def setup(self):
 313         self.resource = self.setup_helper.setup_vnf_environment()
 314
 315     def generate_cfg(self):
 316         pass
 317
 318     def _collect_resource_kpi(self):
 319         result = {}
 320         status = self.resource.check_if_system_agent_running("collectd")[0]
 321         if status == 0:
 322             result = self.resource.amqp_collect_nfvi_kpi()
 323
 324         result = {"core": result}
 325         return result
 326
 327     def start_collect(self):
 328         self.resource.initiate_systemagent(self.ssh_helper.bin_path)
 329         self.resource.start()
 330         self.resource.amqp_process_for_nfvi_kpi()
 331
 332     def stop_collect(self):
 333         if self.resource:
 334             self.resource.stop()
 335
 336     def collect_kpi(self):
 337         return self._collect_resource_kpi()
 338
 339
 340 class ClientResourceHelper(ResourceHelper):
 341
 342     RUN_DURATION = 60
 343     QUEUE_WAIT_TIME = 5
 344     SYNC_PORT = 1
 345     ASYNC_PORT = 2
 346
 347     def __init__(self, setup_helper):
 348         super(ClientResourceHelper, self).__init__(setup_helper)
 349         self.vnfd_helper = setup_helper.vnfd_helper
 350         self.scenario_helper = setup_helper.scenario_helper
 351
 352         self.client = None
 353         self.client_started = Value('i', 0)
 354         self.all_ports = None
 355         self._queue = Queue()
 356         self._result = {}
 357         self._terminated = Value('i', 0)
 358
 359     def _build_ports(self):
 360         self.networks = self.vnfd_helper.port_pairs.networks
 361         self.uplink_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.uplink_ports)
 362         self.downlink_ports = \
 363             self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.downlink_ports)
 364         self.all_ports = self.vnfd_helper.port_nums(self.vnfd_helper.port_pairs.all_ports)
 365
 366     def port_num(self, intf):
 367         # by default return port num
 368         return self.vnfd_helper.port_num(intf)
 369
 370     def get_stats(self, *args, **kwargs):
 371         try:
 372             return self.client.get_stats(*args, **kwargs)
 373         except STLError:
 374             LOG.exception("TRex client not connected")
 375             return {}
 376
 377     def generate_samples(self, ports, key=None, default=None):
 378         # needs to be used ports
 379         last_result = self.get_stats(ports)
 380         key_value = last_result.get(key, default)
 381
 382         if not isinstance(last_result, Mapping):  # added for mock unit test
 383             self._terminated.value = 1
 384             return {}
 385
 386         samples = {}
 387         # recalculate port for interface and see if it matches ports provided
 388         for intf in self.vnfd_helper.interfaces:
 389             name = intf["name"]
 390             port = self.vnfd_helper.port_num(name)
 391             if port in ports:
 392                 xe_value = last_result.get(port, {})
 393                 samples[name] = {
 394                     "rx_throughput_fps": float(xe_value.get("rx_pps", 0.0)),
 395                     "tx_throughput_fps": float(xe_value.get("tx_pps", 0.0)),
 396                     "rx_throughput_mbps": float(xe_value.get("rx_bps", 0.0)),
 397                     "tx_throughput_mbps": float(xe_value.get("tx_bps", 0.0)),
 398                     "in_packets": int(xe_value.get("ipackets", 0)),
 399                     "out_packets": int(xe_value.get("opackets", 0)),
 400                 }
 401                 if key:
 402                     samples[name][key] = key_value
 403         return samples
 404
 405     def _run_traffic_once(self, traffic_profile):
 406         traffic_profile.execute_traffic(self)
 407         self.client_started.value = 1
 408         time.sleep(self.RUN_DURATION)
 409         samples = self.generate_samples(traffic_profile.ports)
 410         time.sleep(self.QUEUE_WAIT_TIME)
 411         self._queue.put(samples)
 412
 413     def run_traffic(self, traffic_profile):
 414         # if we don't do this we can hang waiting for the queue to drain
 415         # have to do this in the subprocess
 416         self._queue.cancel_join_thread()
 417         # fixme: fix passing correct trex config file,
 418         # instead of searching the default path
 419         try:
 420             self._build_ports()
 421             self.client = self._connect()
 422             self.client.reset(ports=self.all_ports)
 423             self.client.remove_all_streams(self.all_ports)  # remove all streams
 424             traffic_profile.register_generator(self)
 425
 426             while self._terminated.value == 0:
 427                 self._run_traffic_once(traffic_profile)
 428
 429             self.client.stop(self.all_ports)
 430             self.client.disconnect()
 431             self._terminated.value = 0
 432         except STLError:
 433             if self._terminated.value:
 434                 LOG.debug("traffic generator is stopped")
 435                 return  # return if trex/tg server is stopped.
 436             raise
 437
 438     def terminate(self):
 439         self._terminated.value = 1  # stop client
 440
 441     def clear_stats(self, ports=None):
 442         if ports is None:
 443             ports = self.all_ports
 444         self.client.clear_stats(ports=ports)
 445
 446     def start(self, ports=None, *args, **kwargs):
 447         # pylint: disable=keyword-arg-before-vararg
 448         # NOTE(ralonsoh): defining keyworded arguments before variable
 449         # positional arguments is a bug. This function definition doesn't work
 450         # in Python 2, although it works in Python 3. Reference:
 451         # https://www.python.org/dev/peps/pep-3102/
 452         if ports is None:
 453             ports = self.all_ports
 454         self.client.start(ports=ports, *args, **kwargs)
 455
 456     def collect_kpi(self):
 457         if not self._queue.empty():
 458             kpi = self._queue.get()
 459             self._result.update(kpi)
 460             LOG.debug('Got KPIs from _queue for %s %s',
 461                       self.scenario_helper.name, self.RESOURCE_WORD)
 462         return self._result
 463
 464     def _connect(self, client=None):
 465         if client is None:
 466             client = STLClient(username=self.vnfd_helper.mgmt_interface["user"],
 467                                server=self.vnfd_helper.mgmt_interface["ip"],
 468                                verbose_level=LoggerApi.VERBOSE_QUIET)
 469
 470         # try to connect with 5s intervals, 30s max
 471         for idx in range(6):
 472             try:
 473                 client.connect()
 474                 break
 475             except STLError:
 476                 LOG.info("Unable to connect to Trex Server.. Attempt %s", idx)
 477                 time.sleep(5)
 478         return client
 479
 480
 481 class Rfc2544ResourceHelper(object):
 482
 483     DEFAULT_CORRELATED_TRAFFIC = False
 484     DEFAULT_LATENCY = False
 485     DEFAULT_TOLERANCE = '0.0001 - 0.0001'
 486
 487     def __init__(self, scenario_helper):
 488         super(Rfc2544ResourceHelper, self).__init__()
 489         self.scenario_helper = scenario_helper
 490         self._correlated_traffic = None
 491         self.iteration = Value('i', 0)
 492         self._latency = None
 493         self._rfc2544 = None
 494         self._tolerance_low = None
 495         self._tolerance_high = None
 496
 497     @property
 498     def rfc2544(self):
 499         if self._rfc2544 is None:
 500             self._rfc2544 = self.scenario_helper.all_options['rfc2544']
 501         return self._rfc2544
 502
 503     @property
 504     def tolerance_low(self):
 505         if self._tolerance_low is None:
 506             self.get_rfc_tolerance()
 507         return self._tolerance_low
 508
 509     @property
 510     def tolerance_high(self):
 511         if self._tolerance_high is None:
 512             self.get_rfc_tolerance()
 513         return self._tolerance_high
 514
 515     @property
 516     def correlated_traffic(self):
 517         if self._correlated_traffic is None:
 518             self._correlated_traffic = \
 519                 self.get_rfc2544('correlated_traffic', self.DEFAULT_CORRELATED_TRAFFIC)
 520
 521         return self._correlated_traffic
 522
 523     @property
 524     def latency(self):
 525         if self._latency is None:
 526             self._latency = self.get_rfc2544('latency', self.DEFAULT_LATENCY)
 527         return self._latency
 528
 529     def get_rfc2544(self, name, default=None):
 530         return self.rfc2544.get(name, default)
 531
 532     def get_rfc_tolerance(self):
 533         tolerance_str = self.get_rfc2544('allowed_drop_rate', self.DEFAULT_TOLERANCE)
 534         tolerance_iter = iter(sorted(float(t.strip()) for t in tolerance_str.split('-')))
 535         self._tolerance_low = next(tolerance_iter)
 536         self._tolerance_high = next(tolerance_iter, self.tolerance_low)
 537
 538
 539 class SampleVNFDeployHelper(object):
 540
 541     SAMPLE_VNF_REPO = 'https://gerrit.opnfv.org/gerrit/samplevnf'
 542     REPO_NAME = posixpath.basename(SAMPLE_VNF_REPO)
 543     SAMPLE_REPO_DIR = os.path.join('~/', REPO_NAME)
 544
 545     def __init__(self, vnfd_helper, ssh_helper):
 546         super(SampleVNFDeployHelper, self).__init__()
 547         self.ssh_helper = ssh_helper
 548         self.vnfd_helper = vnfd_helper
 549
 550     def deploy_vnfs(self, app_name):
 551         vnf_bin = self.ssh_helper.join_bin_path(app_name)
 552         exit_status = self.ssh_helper.execute("which %s" % vnf_bin)[0]
 553         if not exit_status:
 554             return
 555
 556         subprocess.check_output(["rm", "-rf", self.REPO_NAME])
 557         subprocess.check_output(["git", "clone", self.SAMPLE_VNF_REPO])
 558         time.sleep(2)
 559         self.ssh_helper.execute("rm -rf %s" % self.SAMPLE_REPO_DIR)
 560         self.ssh_helper.put(self.REPO_NAME, self.SAMPLE_REPO_DIR, True)
 561
 562         build_script = os.path.join(self.SAMPLE_REPO_DIR, 'tools/vnf_build.sh')
 563         time.sleep(2)
 564         http_proxy = os.environ.get('http_proxy', '')
 565         cmd = "sudo -E %s -s -p='%s'" % (build_script, http_proxy)
 566         LOG.debug(cmd)
 567         self.ssh_helper.execute(cmd)
 568         vnf_bin_loc = os.path.join(self.SAMPLE_REPO_DIR, "VNFs", app_name, "build", app_name)
 569         self.ssh_helper.execute("sudo mkdir -p %s" % self.ssh_helper.bin_path)
 570         self.ssh_helper.execute("sudo cp %s %s" % (vnf_bin_loc, vnf_bin))
 571
 572
 573 class ScenarioHelper(object):
 574
 575     DEFAULT_VNF_CFG = {
 576         'lb_config': 'SW',
 577         'lb_count': 1,
 578         'worker_config': '1C/1T',
 579         'worker_threads': 1,
 580     }
 581
 582     def __init__(self, name):
 583         self.name = name
 584         self.scenario_cfg = None
 585
 586     @property
 587     def task_path(self):
 588         return self.scenario_cfg['task_path']
 589
 590     @property
 591     def nodes(self):
 592         return self.scenario_cfg.get('nodes')
 593
 594     @property
 595     def all_options(self):
 596         return self.scenario_cfg.get('options', {})
 597
 598     @property
 599     def options(self):
 600         return self.all_options.get(self.name, {})
 601
 602     @property
 603     def vnf_cfg(self):
 604         return self.options.get('vnf_config', self.DEFAULT_VNF_CFG)
 605
 606     @property
 607     def topology(self):
 608         return self.scenario_cfg['topology']
 609
 610     @property
 611     def timeout(self):
 612         return self.options.get('timeout', DEFAULT_VNF_TIMEOUT)
 613
 614
 615 class SampleVNF(GenericVNF):
 616     """ Class providing file-like API for generic VNF implementation """
 617
 618     VNF_PROMPT = "pipeline>"
 619     WAIT_TIME = 1
 620     WAIT_TIME_FOR_SCRIPT = 10
 621     APP_NAME = "SampleVNF"
 622     # we run the VNF interactively, so the ssh command will timeout after this long
 623
 624     def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
 625         super(SampleVNF, self).__init__(name, vnfd)
 626         self.bin_path = get_nsb_option('bin_path', '')
 627
 628         self.scenario_helper = ScenarioHelper(self.name)
 629         self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path)
 630
 631         if setup_env_helper_type is None:
 632             setup_env_helper_type = SetupEnvHelper
 633
 634         self.setup_helper = setup_env_helper_type(self.vnfd_helper,
 635                                                   self.ssh_helper,
 636                                                   self.scenario_helper)
 637
 638         self.deploy_helper = SampleVNFDeployHelper(vnfd, self.ssh_helper)
 639
 640         if resource_helper_type is None:
 641             resource_helper_type = ResourceHelper
 642
 643         self.resource_helper = resource_helper_type(self.setup_helper)
 644
 645         self.context_cfg = None
 646         self.nfvi_context = None
 647         self.pipeline_kwargs = {}
 648         self.uplink_ports = None
 649         self.downlink_ports = None
 650         # NOTE(esm): make QueueFileWrapper invert-able so that we
 651         #            never have to manage the queues
 652         self.q_in = Queue()
 653         self.q_out = Queue()
 654         self.queue_wrapper = None
 655         self.run_kwargs = {}
 656         self.used_drivers = {}
 657         self.vnf_port_pairs = None
 658         self._vnf_process = None
 659
 660     def _build_ports(self):
 661         self._port_pairs = PortPairs(self.vnfd_helper.interfaces)
 662         self.networks = self._port_pairs.networks
 663         self.uplink_ports = self.vnfd_helper.port_nums(self._port_pairs.uplink_ports)
 664         self.downlink_ports = self.vnfd_helper.port_nums(self._port_pairs.downlink_ports)
 665         self.my_ports = self.vnfd_helper.port_nums(self._port_pairs.all_ports)
 666
 667     def _get_route_data(self, route_index, route_type):
 668         route_iter = iter(self.vnfd_helper.vdu0.get('nd_route_tbl', []))
 669         for _ in range(route_index):
 670             next(route_iter, '')
 671         return next(route_iter, {}).get(route_type, '')
 672
 673     def _get_port0localip6(self):
 674         return_value = self._get_route_data(0, 'network')
 675         LOG.info("_get_port0localip6 : %s", return_value)
 676         return return_value
 677
 678     def _get_port1localip6(self):
 679         return_value = self._get_route_data(1, 'network')
 680         LOG.info("_get_port1localip6 : %s", return_value)
 681         return return_value
 682
 683     def _get_port0prefixlen6(self):
 684         return_value = self._get_route_data(0, 'netmask')
 685         LOG.info("_get_port0prefixlen6 : %s", return_value)
 686         return return_value
 687
 688     def _get_port1prefixlen6(self):
 689         return_value = self._get_route_data(1, 'netmask')
 690         LOG.info("_get_port1prefixlen6 : %s", return_value)
 691         return return_value
 692
 693     def _get_port0gateway6(self):
 694         return_value = self._get_route_data(0, 'network')
 695         LOG.info("_get_port0gateway6 : %s", return_value)
 696         return return_value
 697
 698     def _get_port1gateway6(self):
 699         return_value = self._get_route_data(1, 'network')
 700         LOG.info("_get_port1gateway6 : %s", return_value)
 701         return return_value
 702
 703     def _start_vnf(self):
 704         self.queue_wrapper = QueueFileWrapper(self.q_in, self.q_out, self.VNF_PROMPT)
 705         name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
 706         self._vnf_process = Process(name=name, target=self._run)
 707         self._vnf_process.start()
 708
 709     def _vnf_up_post(self):
 710         pass
 711
 712     def instantiate(self, scenario_cfg, context_cfg):
 713         self.scenario_helper.scenario_cfg = scenario_cfg
 714         self.context_cfg = context_cfg
 715         self.nfvi_context = Context.get_context_from_server(self.scenario_helper.nodes[self.name])
 716         # self.nfvi_context = None
 717
 718         # vnf deploy is unsupported, use ansible playbooks
 719         if self.scenario_helper.options.get("vnf_deploy", False):
 720             self.deploy_helper.deploy_vnfs(self.APP_NAME)
 721         self.resource_helper.setup()
 722         self._start_vnf()
 723
 724     def wait_for_instantiate(self):
 725         buf = []
 726         time.sleep(self.WAIT_TIME)  # Give some time for config to load
 727         while True:
 728             if not self._vnf_process.is_alive():
 729                 raise RuntimeError("%s VNF process died." % self.APP_NAME)
 730
 731             # NOTE(esm): move to QueueFileWrapper
 732             while self.q_out.qsize() > 0:
 733                 buf.append(self.q_out.get())
 734                 message = ''.join(buf)
 735                 if self.VNF_PROMPT in message:
 736                     LOG.info("%s VNF is up and running.", self.APP_NAME)
 737                     self._vnf_up_post()
 738                     self.queue_wrapper.clear()
 739                     self.resource_helper.start_collect()
 740                     return self._vnf_process.exitcode
 741
 742                 if "PANIC" in message:
 743                     raise RuntimeError("Error starting %s VNF." %
 744                                        self.APP_NAME)
 745
 746             LOG.info("Waiting for %s VNF to start.. ", self.APP_NAME)
 747             time.sleep(self.WAIT_TIME_FOR_SCRIPT)
 748             # Send ENTER to display a new prompt in case the prompt text was corrupted
 749             # by other VNF output
 750             self.q_in.put('\r\n')
 751
 752     def _build_run_kwargs(self):
 753         self.run_kwargs = {
 754             'stdin': self.queue_wrapper,
 755             'stdout': self.queue_wrapper,
 756             'keep_stdin_open': True,
 757             'pty': True,
 758             'timeout': self.scenario_helper.timeout,
 759         }
 760
 761     def _build_config(self):
 762         return self.setup_helper.build_config()
 763
 764     def _run(self):
 765         # we can't share ssh paramiko objects to force new connection
 766         self.ssh_helper.drop_connection()
 767         cmd = self._build_config()
 768         # kill before starting
 769         self.setup_helper.kill_vnf()
 770
 771         LOG.debug(cmd)
 772         self._build_run_kwargs()
 773         self.ssh_helper.run(cmd, **self.run_kwargs)
 774
 775     def vnf_execute(self, cmd, wait_time=2):
 776         """ send cmd to vnf process """
 777
 778         LOG.info("%s command: %s", self.APP_NAME, cmd)
 779         self.q_in.put("{}\r\n".format(cmd))
 780         time.sleep(wait_time)
 781         output = []
 782         while self.q_out.qsize() > 0:
 783             output.append(self.q_out.get())
 784         return "".join(output)
 785
 786     def _tear_down(self):
 787         pass
 788
 789     def terminate(self):
 790         self.vnf_execute("quit")
 791         self.setup_helper.kill_vnf()
 792         self._tear_down()
 793         self.resource_helper.stop_collect()
 794         if self._vnf_process is not None:
 795             # be proper and join first before we kill
 796             LOG.debug("joining before terminate %s", self._vnf_process.name)
 797             self._vnf_process.join(PROCESS_JOIN_TIMEOUT)
 798             self._vnf_process.terminate()
 799         # no terminate children here because we share processes with tg
 800
 801     def get_stats(self, *args, **kwargs):  # pylint: disable=unused-argument
 802         """Method for checking the statistics
 803
 804         This method could be overridden in children classes.
 805
 806         :return: VNF statistics
 807         """
 808         cmd = 'p {0} stats'.format(self.APP_WORD)
 809         out = self.vnf_execute(cmd)
 810         return out
 811
 812     def collect_kpi(self):
 813         # we can't get KPIs if the VNF is down
 814         check_if_process_failed(self._vnf_process)
 815         stats = self.get_stats()
 816         m = re.search(self.COLLECT_KPI, stats, re.MULTILINE)
 817         if m:
 818             result = {k: int(m.group(v)) for k, v in self.COLLECT_MAP.items()}
 819             result["collect_stats"] = self.resource_helper.collect_kpi()
 820         else:
 821             result = {
 822                 "packets_in": 0,
 823                 "packets_fwd": 0,
 824                 "packets_dropped": 0,
 825             }
 826         LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
 827         return result
 828
 829     def scale(self, flavor=""):
 830         """The SampleVNF base class doesn't provide the 'scale' feature"""
 831         raise y_exceptions.FunctionNotImplemented(
 832             function_name='scale', class_name='SampleVNFTrafficGen')
 833
 834
 835 class SampleVNFTrafficGen(GenericTrafficGen):
 836     """ Class providing file-like API for generic traffic generator """
 837
 838     APP_NAME = 'Sample'
 839     RUN_WAIT = 1
 840
 841     def __init__(self, name, vnfd, setup_env_helper_type=None, resource_helper_type=None):
 842         super(SampleVNFTrafficGen, self).__init__(name, vnfd)
 843         self.bin_path = get_nsb_option('bin_path', '')
 844
 845         self.scenario_helper = ScenarioHelper(self.name)
 846         self.ssh_helper = VnfSshHelper(self.vnfd_helper.mgmt_interface, self.bin_path, wait=True)
 847
 848         if setup_env_helper_type is None:
 849             setup_env_helper_type = SetupEnvHelper
 850
 851         self.setup_helper = setup_env_helper_type(self.vnfd_helper,
 852                                                   self.ssh_helper,
 853                                                   self.scenario_helper)
 854
 855         if resource_helper_type is None:
 856             resource_helper_type = ClientResourceHelper
 857
 858         self.resource_helper = resource_helper_type(self.setup_helper)
 859
 860         self.runs_traffic = True
 861         self.traffic_finished = False
 862         self._tg_process = None
 863         self._traffic_process = None
 864
 865     def _start_server(self):
 866         # we can't share ssh paramiko objects to force new connection
 867         self.ssh_helper.drop_connection()
 868
 869     def instantiate(self, scenario_cfg, context_cfg):
 870         self.scenario_helper.scenario_cfg = scenario_cfg
 871         self.resource_helper.setup()
 872         # must generate_cfg after DPDK bind because we need port number
 873         self.resource_helper.generate_cfg()
 874
 875         LOG.info("Starting %s server...", self.APP_NAME)
 876         name = "{}-{}-{}".format(self.name, self.APP_NAME, os.getpid())
 877         self._tg_process = Process(name=name, target=self._start_server)
 878         self._tg_process.start()
 879
 880     def _check_status(self):
 881         raise NotImplementedError
 882
 883     def _wait_for_process(self):
 884         while True:
 885             if not self._tg_process.is_alive():
 886                 raise RuntimeError("%s traffic generator process died." % self.APP_NAME)
 887             LOG.info("Waiting for %s TG Server to start.. ", self.APP_NAME)
 888             time.sleep(1)
 889             status = self._check_status()
 890             if status == 0:
 891                 LOG.info("%s TG Server is up and running.", self.APP_NAME)
 892                 return self._tg_process.exitcode
 893
 894     def _traffic_runner(self, traffic_profile):
 895         # always drop connections first thing in new processes
 896         # so we don't get paramiko errors
 897         self.ssh_helper.drop_connection()
 898         LOG.info("Starting %s client...", self.APP_NAME)
 899         self.resource_helper.run_traffic(traffic_profile)
 900
 901     def run_traffic(self, traffic_profile):
 902         """ Generate traffic on the wire according to the given params.
 903         Method is non-blocking, returns immediately when traffic process
 904         is running. Mandatory.
 905
 906         :param traffic_profile:
 907         :return: True/False
 908         """
 909         name = "{}-{}-{}-{}".format(self.name, self.APP_NAME, traffic_profile.__class__.__name__,
 910                                     os.getpid())
 911         self._traffic_process = Process(name=name, target=self._traffic_runner,
 912                                         args=(traffic_profile,))
 913         self._traffic_process.start()
 914         # Wait for traffic process to start
 915         while self.resource_helper.client_started.value == 0:
 916             time.sleep(self.RUN_WAIT)
 917             # what if traffic process takes a few seconds to start?
 918             if not self._traffic_process.is_alive():
 919                 break
 920
 921         return self._traffic_process.is_alive()
 922
 923     def collect_kpi(self):
 924         # check if the tg processes have exited
 925         for proc in (self._tg_process, self._traffic_process):
 926             check_if_process_failed(proc)
 927         result = self.resource_helper.collect_kpi()
 928         LOG.debug("%s collect KPIs %s", self.APP_NAME, result)
 929         return result
 930
 931     def terminate(self):
 932         """ After this method finishes, all traffic processes should stop. Mandatory.
 933
 934         :return: True/False
 935         """
 936         self.traffic_finished = True
 937         # we must kill client before we kill the server, or the client will raise exception
 938         if self._traffic_process is not None:
 939             # be proper and try to join before terminating
 940             LOG.debug("joining before terminate %s", self._traffic_process.name)
 941             self._traffic_process.join(PROCESS_JOIN_TIMEOUT)
 942             self._traffic_process.terminate()
 943         if self._tg_process is not None:
 944             # be proper and try to join before terminating
 945             LOG.debug("joining before terminate %s", self._tg_process.name)
 946             self._tg_process.join(PROCESS_JOIN_TIMEOUT)
 947             self._tg_process.terminate()
 948         # no terminate children here because we share processes with vnf
 949
 950     def scale(self, flavor=""):
 951         """A traffic generator VFN doesn't provide the 'scale' feature"""
 952         raise y_exceptions.FunctionNotImplemented(
 953             function_name='scale', class_name='SampleVNFTrafficGen')