yardstick/benchmark/scenarios/networking/vnf_generic.py

   1 # Copyright (c) 2016-2017 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
  15 import copy
  16 import ipaddress
  17 from itertools import chain
  18 import logging
  19 import os
  20 import sys
  21 import time
  22
  23 import six
  24 import yaml
  25
  26 from yardstick.benchmark.contexts import base as context_base
  27 from yardstick.benchmark.scenarios import base as scenario_base
  28 from yardstick.common.constants import LOG_DIR
  29 from yardstick.common import exceptions
  30 from yardstick.common.process import terminate_children
  31 from yardstick.common import utils
  32 from yardstick.network_services.collector.subscriber import Collector
  33 from yardstick.network_services.vnf_generic import vnfdgen
  34 from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
  35 from yardstick.network_services import traffic_profile
  36 from yardstick.network_services.traffic_profile import base as tprofile_base
  37 from yardstick.network_services.utils import get_nsb_option
  38 from yardstick import ssh
  39
  40
  41 traffic_profile.register_modules()
  42
  43
  44 LOG = logging.getLogger(__name__)
  45
  46
  47 class NetworkServiceTestCase(scenario_base.Scenario):
  48     """Class handles Generic framework to do pre-deployment VNF &
  49        Network service testing  """
  50
  51     __scenario_type__ = "NSPerf"
  52
  53     def __init__(self, scenario_cfg, context_cfg):  # pragma: no cover
  54         super(NetworkServiceTestCase, self).__init__()
  55         self.scenario_cfg = scenario_cfg
  56         self.context_cfg = context_cfg
  57
  58         self._render_topology()
  59         self.vnfs = []
  60         self.collector = None
  61         self.traffic_profile = None
  62         self.node_netdevs = {}
  63         self.bin_path = get_nsb_option('bin_path', '')
  64         self._mq_ids = []
  65
  66     def _get_ip_flow_range(self, ip_start_range):
  67         """Retrieve a CIDR first and last viable IPs
  68
  69         :param ip_start_range: could be the IP range itself or a dictionary
  70                with the host name and the port.
  71         :return: (str) IP range (min, max) with this format "x.x.x.x-y.y.y.y"
  72         """
  73         if isinstance(ip_start_range, six.string_types):
  74             return ip_start_range
  75
  76         node_name, range_or_interface = next(iter(ip_start_range.items()),
  77                                              (None, '0.0.0.0'))
  78         if node_name is None:
  79             return range_or_interface
  80
  81         node = self.context_cfg['nodes'].get(node_name, {})
  82         interface = node.get('interfaces', {}).get(range_or_interface)
  83         if interface:
  84             ip = interface['local_ip']
  85             mask = interface['netmask']
  86         else:
  87             ip = '0.0.0.0'
  88             mask = '255.255.255.0'
  89
  90         ipaddr = ipaddress.ip_network(
  91             six.text_type('{}/{}'.format(ip, mask)), strict=False)
  92         if ipaddr.prefixlen + 2 < ipaddr.max_prefixlen:
  93             ip_addr_range = '{}-{}'.format(ipaddr[2], ipaddr[-2])
  94         else:
  95             LOG.warning('Only single IP in range %s', ipaddr)
  96             ip_addr_range = ip
  97         return ip_addr_range
  98
  99     def _get_traffic_flow(self):
 100         flow = {}
 101         try:
 102             # TODO: should be .0  or .1 so we can use list
 103             # but this also roughly matches uplink_0, downlink_0
 104             fflow = self.scenario_cfg["options"]["flow"]
 105             for index, src in enumerate(fflow.get("src_ip", [])):
 106                 flow["src_ip_{}".format(index)] = self._get_ip_flow_range(src)
 107
 108             for index, dst in enumerate(fflow.get("dst_ip", [])):
 109                 flow["dst_ip_{}".format(index)] = self._get_ip_flow_range(dst)
 110
 111             for index, publicip in enumerate(fflow.get("public_ip", [])):
 112                 flow["public_ip_{}".format(index)] = publicip
 113
 114             for index, src_port in enumerate(fflow.get("src_port", [])):
 115                 flow["src_port_{}".format(index)] = src_port
 116
 117             for index, dst_port in enumerate(fflow.get("dst_port", [])):
 118                 flow["dst_port_{}".format(index)] = dst_port
 119
 120             if "count" in fflow:
 121                 flow["count"] = fflow["count"]
 122
 123             if "srcseed" in fflow:
 124                 flow["srcseed"] = fflow["srcseed"]
 125
 126             if "dstseed" in fflow:
 127                 flow["dstseed"] = fflow["dstseed"]
 128
 129         except KeyError:
 130             flow = {}
 131         return {"flow": flow}
 132
 133     def _get_traffic_imix(self):
 134         try:
 135             imix = {"imix": self.scenario_cfg['options']['framesize']}
 136         except KeyError:
 137             imix = {}
 138         return imix
 139
 140     def _get_traffic_profile(self):
 141         profile = self.scenario_cfg["traffic_profile"]
 142         path = self.scenario_cfg["task_path"]
 143         with utils.open_relative_file(profile, path) as infile:
 144             return infile.read()
 145
 146     def _get_duration(self):
 147         options = self.scenario_cfg.get('options', {})
 148         return options.get('duration',
 149                            tprofile_base.TrafficProfileConfig.DEFAULT_DURATION)
 150
 151     def _fill_traffic_profile(self):
 152         tprofile = self._get_traffic_profile()
 153         extra_args = self.scenario_cfg.get('extra_args', {})
 154         tprofile_data = {
 155             'flow': self._get_traffic_flow(),
 156             'imix': self._get_traffic_imix(),
 157             tprofile_base.TrafficProfile.UPLINK: {},
 158             tprofile_base.TrafficProfile.DOWNLINK: {},
 159             'extra_args': extra_args,
 160             'duration': self._get_duration()}
 161         traffic_vnfd = vnfdgen.generate_vnfd(tprofile, tprofile_data)
 162         self.traffic_profile = tprofile_base.TrafficProfile.get(traffic_vnfd)
 163
 164     def _get_topology(self):
 165         topology = self.scenario_cfg["topology"]
 166         path = self.scenario_cfg["task_path"]
 167         with utils.open_relative_file(topology, path) as infile:
 168             return infile.read()
 169
 170     def _render_topology(self):
 171         topology = self._get_topology()
 172         topology_args = self.scenario_cfg.get('extra_args', {})
 173         topolgy_data = {
 174             'extra_args': topology_args
 175         }
 176         topology_yaml = vnfdgen.generate_vnfd(topology, topolgy_data)
 177         self.topology = topology_yaml["nsd:nsd-catalog"]["nsd"][0]
 178
 179     def _find_vnf_name_from_id(self, vnf_id):  # pragma: no cover
 180         return next((vnfd["vnfd-id-ref"]
 181                      for vnfd in self.topology["constituent-vnfd"]
 182                      if vnf_id == vnfd["member-vnf-index"]), None)
 183
 184     def _find_vnfd_from_vnf_idx(self, vnf_id):  # pragma: no cover
 185         return next((vnfd
 186                      for vnfd in self.topology["constituent-vnfd"]
 187                      if vnf_id == vnfd["member-vnf-index"]), None)
 188
 189     @staticmethod
 190     def find_node_if(nodes, name, if_name, vld_id):  # pragma: no cover
 191         try:
 192             # check for xe0, xe1
 193             intf = nodes[name]["interfaces"][if_name]
 194         except KeyError:
 195             # if not xe0, then maybe vld_id,  uplink_0, downlink_0
 196             # pop it and re-insert with the correct name from topology
 197             intf = nodes[name]["interfaces"].pop(vld_id)
 198             nodes[name]["interfaces"][if_name] = intf
 199         return intf
 200
 201     def _resolve_topology(self):
 202         for vld in self.topology["vld"]:
 203             try:
 204                 node0_data, node1_data = vld["vnfd-connection-point-ref"]
 205             except (ValueError, TypeError):
 206                 raise exceptions.IncorrectConfig(
 207                     error_msg='Topology file corrupted, wrong endpoint count '
 208                               'for connection')
 209
 210             node0_name = self._find_vnf_name_from_id(node0_data["member-vnf-index-ref"])
 211             node1_name = self._find_vnf_name_from_id(node1_data["member-vnf-index-ref"])
 212
 213             node0_if_name = node0_data["vnfd-connection-point-ref"]
 214             node1_if_name = node1_data["vnfd-connection-point-ref"]
 215
 216             try:
 217                 nodes = self.context_cfg["nodes"]
 218                 node0_if = self.find_node_if(nodes, node0_name, node0_if_name, vld["id"])
 219                 node1_if = self.find_node_if(nodes, node1_name, node1_if_name, vld["id"])
 220
 221                 # names so we can do reverse lookups
 222                 node0_if["ifname"] = node0_if_name
 223                 node1_if["ifname"] = node1_if_name
 224
 225                 node0_if["node_name"] = node0_name
 226                 node1_if["node_name"] = node1_name
 227
 228                 node0_if["vld_id"] = vld["id"]
 229                 node1_if["vld_id"] = vld["id"]
 230
 231                 # set peer name
 232                 node0_if["peer_name"] = node1_name
 233                 node1_if["peer_name"] = node0_name
 234
 235                 # set peer interface name
 236                 node0_if["peer_ifname"] = node1_if_name
 237                 node1_if["peer_ifname"] = node0_if_name
 238
 239                 # just load the network
 240                 vld_networks = {n.get('vld_id', name): n for name, n in
 241                                 self.context_cfg["networks"].items()}
 242
 243                 node0_if["network"] = vld_networks.get(vld["id"], {})
 244                 node1_if["network"] = vld_networks.get(vld["id"], {})
 245
 246                 node0_if["dst_mac"] = node1_if["local_mac"]
 247                 node0_if["dst_ip"] = node1_if["local_ip"]
 248
 249                 node1_if["dst_mac"] = node0_if["local_mac"]
 250                 node1_if["dst_ip"] = node0_if["local_ip"]
 251
 252             except KeyError:
 253                 LOG.exception("")
 254                 raise exceptions.IncorrectConfig(
 255                     error_msg='Required interface not found, topology file '
 256                               'corrupted')
 257
 258         for vld in self.topology['vld']:
 259             try:
 260                 node0_data, node1_data = vld["vnfd-connection-point-ref"]
 261             except (ValueError, TypeError):
 262                 raise exceptions.IncorrectConfig(
 263                     error_msg='Topology file corrupted, wrong endpoint count '
 264                               'for connection')
 265
 266             node0_name = self._find_vnf_name_from_id(node0_data["member-vnf-index-ref"])
 267             node1_name = self._find_vnf_name_from_id(node1_data["member-vnf-index-ref"])
 268
 269             node0_if_name = node0_data["vnfd-connection-point-ref"]
 270             node1_if_name = node1_data["vnfd-connection-point-ref"]
 271
 272             nodes = self.context_cfg["nodes"]
 273             node0_if = self.find_node_if(nodes, node0_name, node0_if_name, vld["id"])
 274             node1_if = self.find_node_if(nodes, node1_name, node1_if_name, vld["id"])
 275
 276             # add peer interface dict, but remove circular link
 277             # TODO: don't waste memory
 278             node0_copy = node0_if.copy()
 279             node1_copy = node1_if.copy()
 280             node0_if["peer_intf"] = node1_copy
 281             node1_if["peer_intf"] = node0_copy
 282
 283     def _update_context_with_topology(self):  # pragma: no cover
 284         for vnfd in self.topology["constituent-vnfd"]:
 285             vnf_idx = vnfd["member-vnf-index"]
 286             vnf_name = self._find_vnf_name_from_id(vnf_idx)
 287             vnfd = self._find_vnfd_from_vnf_idx(vnf_idx)
 288             self.context_cfg["nodes"][vnf_name].update(vnfd)
 289
 290     def _generate_pod_yaml(self):  # pragma: no cover
 291         context_yaml = os.path.join(LOG_DIR, "pod-{}.yaml".format(self.scenario_cfg['task_id']))
 292         # convert OrderedDict to a list
 293         # pod.yaml nodes is a list
 294         nodes = [self._serialize_node(node) for node in self.context_cfg["nodes"].values()]
 295         pod_dict = {
 296             "nodes": nodes,
 297             "networks": self.context_cfg["networks"]
 298         }
 299         with open(context_yaml, "w") as context_out:
 300             yaml.safe_dump(pod_dict, context_out, default_flow_style=False,
 301                            explicit_start=True)
 302
 303     @staticmethod
 304     def _serialize_node(node):  # pragma: no cover
 305         new_node = copy.deepcopy(node)
 306         # name field is required
 307         # remove context suffix
 308         new_node["name"] = node['name'].split('.')[0]
 309         try:
 310             new_node["pkey"] = ssh.convert_key_to_str(node["pkey"])
 311         except KeyError:
 312             pass
 313         return new_node
 314
 315     def map_topology_to_infrastructure(self):
 316         """ This method should verify if the available resources defined in pod.yaml
 317         match the topology.yaml file.
 318
 319         :return: None. Side effect: context_cfg is updated
 320         """
 321         # 3. Use topology file to find connections & resolve dest address
 322         self._resolve_topology()
 323         self._update_context_with_topology()
 324
 325     @classmethod
 326     def get_vnf_impl(cls, vnf_model_id):  # pragma: no cover
 327         """ Find the implementing class from vnf_model["vnf"]["name"] field
 328
 329         :param vnf_model_id: parsed vnfd model ID field
 330         :return: subclass of GenericVNF
 331         """
 332         utils.import_modules_from_package(
 333             "yardstick.network_services.vnf_generic.vnf")
 334         expected_name = vnf_model_id
 335         classes_found = []
 336
 337         def impl():
 338             for name, class_ in ((c.__name__, c) for c in
 339                                  utils.itersubclasses(GenericVNF)):
 340                 if name == expected_name:
 341                     yield class_
 342                 classes_found.append(name)
 343
 344         try:
 345             return next(impl())
 346         except StopIteration:
 347             pass
 348
 349         message = ('No implementation for %s found in %s'
 350                    % (expected_name, classes_found))
 351         raise exceptions.IncorrectConfig(error_msg=message)
 352
 353     @staticmethod
 354     def create_interfaces_from_node(vnfd, node):  # pragma: no cover
 355         ext_intfs = vnfd["vdu"][0]["external-interface"] = []
 356         # have to sort so xe0 goes first
 357         for intf_name, intf in sorted(node['interfaces'].items()):
 358             # only interfaces with vld_id are added.
 359             # Thus there are two layers of filters, only intefaces with vld_id
 360             # show up in interfaces, and only interfaces with traffic profiles
 361             # are used by the generators
 362             if intf.get('vld_id'):
 363                 # force dpkd_port_num to int so we can do reverse lookup
 364                 try:
 365                     intf['dpdk_port_num'] = int(intf['dpdk_port_num'])
 366                 except KeyError:
 367                     pass
 368                 ext_intf = {
 369                     "name": intf_name,
 370                     "virtual-interface": intf,
 371                     "vnfd-connection-point-ref": intf_name,
 372                 }
 373                 ext_intfs.append(ext_intf)
 374
 375     def load_vnf_models(self, scenario_cfg=None, context_cfg=None):
 376         """ Create VNF objects based on YAML descriptors
 377
 378         :param scenario_cfg:
 379         :type scenario_cfg:
 380         :param context_cfg:
 381         :return:
 382         """
 383         trex_lib_path = get_nsb_option('trex_client_lib')
 384         sys.path[:] = list(chain([trex_lib_path], (x for x in sys.path if x != trex_lib_path)))
 385
 386         if scenario_cfg is None:
 387             scenario_cfg = self.scenario_cfg
 388
 389         if context_cfg is None:
 390             context_cfg = self.context_cfg
 391
 392         vnfs = []
 393         # we assume OrderedDict for consistency in instantiation
 394         for node_name, node in context_cfg["nodes"].items():
 395             LOG.debug(node)
 396             try:
 397                 file_name = node["VNF model"]
 398             except KeyError:
 399                 LOG.debug("no model for %s, skipping", node_name)
 400                 continue
 401             file_path = scenario_cfg['task_path']
 402             with utils.open_relative_file(file_name, file_path) as stream:
 403                 vnf_model = stream.read()
 404             vnfd = vnfdgen.generate_vnfd(vnf_model, node)
 405             # TODO: here add extra context_cfg["nodes"] regardless of template
 406             vnfd = vnfd["vnfd:vnfd-catalog"]["vnfd"][0]
 407             # force inject pkey if it exists
 408             # we want to standardize Heat using pkey as a string so we don't rely
 409             # on the filesystem
 410             try:
 411                 vnfd['mgmt-interface']['pkey'] = node['pkey']
 412             except KeyError:
 413                 pass
 414             self.create_interfaces_from_node(vnfd, node)
 415             vnf_impl = self.get_vnf_impl(vnfd['id'])
 416             vnf_instance = vnf_impl(node_name, vnfd, scenario_cfg['task_id'])
 417             vnfs.append(vnf_instance)
 418
 419         self.vnfs = vnfs
 420         return vnfs
 421
 422     def setup(self):
 423         """Setup infrastructure, provission VNFs & start traffic"""
 424         # 1. Verify if infrastructure mapping can meet topology
 425         self.map_topology_to_infrastructure()
 426         # 1a. Load VNF models
 427         self.load_vnf_models()
 428         # 1b. Fill traffic profile with information from topology
 429         self._fill_traffic_profile()
 430
 431         # 2. Provision VNFs
 432
 433         # link events will cause VNF application to exit
 434         # so we should start traffic runners before VNFs
 435         traffic_runners = [vnf for vnf in self.vnfs if vnf.runs_traffic]
 436         non_traffic_runners = [vnf for vnf in self.vnfs if not vnf.runs_traffic]
 437         try:
 438             for vnf in chain(traffic_runners, non_traffic_runners):
 439                 LOG.info("Instantiating %s", vnf.name)
 440                 vnf.instantiate(self.scenario_cfg, self.context_cfg)
 441                 LOG.info("Waiting for %s to instantiate", vnf.name)
 442                 vnf.wait_for_instantiate()
 443         except:
 444             LOG.exception("")
 445             for vnf in self.vnfs:
 446                 vnf.terminate()
 447             raise
 448
 449         # we have to generate pod.yaml here after VNF has probed so we know vpci and driver
 450         self._generate_pod_yaml()
 451
 452         # 3. Run experiment
 453         # Start listeners first to avoid losing packets
 454         for traffic_gen in traffic_runners:
 455             traffic_gen.listen_traffic(self.traffic_profile)
 456
 457         # register collector with yardstick for KPI collection.
 458         self.collector = Collector(self.vnfs, context_base.Context.get_physical_nodes())
 459         self.collector.start()
 460
 461         # Start the actual traffic
 462         for traffic_gen in traffic_runners:
 463             LOG.info("Starting traffic on %s", traffic_gen.name)
 464             traffic_gen.run_traffic(self.traffic_profile)
 465             self._mq_ids.append(traffic_gen.get_mq_producer_id())
 466
 467     def get_mq_ids(self):  # pragma: no cover
 468         """Return stored MQ producer IDs"""
 469         return self._mq_ids
 470
 471     def run(self, result):  # yardstick API
 472         """ Yardstick calls run() at intervals defined in the yaml and
 473             produces timestamped samples
 474
 475         :param result: dictionary with results to update
 476         :return: None
 477         """
 478
 479         # this is the only method that is check from the runner
 480         # so if we have any fatal error it must be raised via these methods
 481         # otherwise we will not terminate
 482
 483         result.update(self.collector.get_kpi())
 484
 485     def teardown(self):
 486         """ Stop the collector and terminate VNF & TG instance
 487
 488         :return
 489         """
 490
 491         try:
 492             try:
 493                 self.collector.stop()
 494                 for vnf in self.vnfs:
 495                     LOG.info("Stopping %s", vnf.name)
 496                     vnf.terminate()
 497                 LOG.debug("all VNFs terminated: %s", ", ".join(vnf.name for vnf in self.vnfs))
 498             finally:
 499                 terminate_children()
 500         except Exception:
 501             # catch any exception in teardown and convert to simple exception
 502             # never pass exceptions back to multiprocessing, because some exceptions can
 503             # be unpicklable
 504             # https://bugs.python.org/issue9400
 505             LOG.exception("")
 506             raise RuntimeError("Error in teardown")
 507
 508     def pre_run_wait_time(self, time_seconds):  # pragma: no cover
 509         """Time waited before executing the run method"""
 510         time.sleep(time_seconds)
 511
 512     def post_run_wait_time(self, time_seconds):  # pragma: no cover
 513         """Time waited after executing the run method"""
 514         pass