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