yardstick/network_services/vnf_generic/vnf/vpe_vnf.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 """ vPE (Power Edge router) VNF model definitions based on IETS Spec """
  15
  16 from __future__ import absolute_import
  17 from __future__ import print_function
  18 import tempfile
  19 import time
  20 import os
  21 import logging
  22 import re
  23 from multiprocessing import Queue
  24 import multiprocessing
  25 import ipaddress
  26 import six
  27
  28 from yardstick import ssh
  29 from yardstick.network_services.vnf_generic.vnf.base import GenericVNF
  30 from yardstick.network_services.utils import provision_tool
  31 from yardstick.network_services.vnf_generic.vnf.base import QueueFileWrapper
  32 from yardstick.network_services.nfvi.resource import ResourceProfile
  33
  34 LOG = logging.getLogger(__name__)
  35 VPE_PIPELINE_COMMAND = '{tool_path} -p 0x3 -f {cfg_file} -s {script}'
  36 CORES = ['0', '1', '2']
  37 WAIT_TIME = 20
  38
  39
  40 class VpeApproxVnf(GenericVNF):
  41     """ This class handles vPE VNF model-driver definitions """
  42
  43     def __init__(self, vnfd):
  44         super(VpeApproxVnf, self).__init__(vnfd)
  45         self.socket = None
  46         self.q_in = Queue()
  47         self.q_out = Queue()
  48         self.vnf_cfg = None
  49         self._vnf_process = None
  50         self.connection = None
  51         self.resource = None
  52
  53     def _resource_collect_start(self):
  54         self.resource.initiate_systemagent(self.bin_path)
  55         self.resource.start()
  56
  57     def _resource_collect_stop(self):
  58         self.resource.stop()
  59
  60     def _collect_resource_kpi(self):
  61         result = {}
  62
  63         status = self.resource.check_if_sa_running("collectd")[0]
  64         if status:
  65             result = self.resource.amqp_collect_nfvi_kpi()
  66
  67         result = {"core": result}
  68
  69         return result
  70
  71     @classmethod
  72     def __setup_hugepages(cls, connection):
  73         hugepages = \
  74             connection.execute(
  75                 "awk '/Hugepagesize/ { print $2$3 }' < /proc/meminfo")[1]
  76         hugepages = hugepages.rstrip()
  77
  78         memory_path = \
  79             '/sys/kernel/mm/hugepages/hugepages-%s/nr_hugepages' % hugepages
  80         connection.execute("awk -F: '{ print $1 }' < %s" % memory_path)
  81
  82         pages = 16384 if hugepages.rstrip() == "2048kB" else 16
  83         connection.execute("echo %s > %s" % (pages, memory_path))
  84
  85     def setup_vnf_environment(self, connection):
  86         ''' setup dpdk environment needed for vnf to run '''
  87
  88         self.__setup_hugepages(connection)
  89         connection.execute("modprobe uio && modprobe igb_uio")
  90
  91         exit_status = connection.execute("lsmod | grep -i igb_uio")[0]
  92         if exit_status == 0:
  93             return
  94
  95         dpdk = os.path.join(self.bin_path, "dpdk-16.07")
  96         dpdk_setup = \
  97             provision_tool(self.connection,
  98                            os.path.join(self.bin_path, "nsb_setup.sh"))
  99         status = connection.execute("ls {} >/dev/null 2>&1".format(dpdk))[0]
 100         if status:
 101             connection.execute("bash %s dpdk >/dev/null 2>&1" % dpdk_setup)
 102
 103     def _get_cpu_sibling_list(self):
 104         cpu_topo = []
 105         for core in CORES:
 106             sys_cmd = \
 107                 "/sys/devices/system/cpu/cpu%s/topology/thread_siblings_list" \
 108                 % core
 109             cpuid = \
 110                 self.connection.execute("awk -F: '{ print $1 }' < %s" %
 111                                         sys_cmd)[1]
 112             cpu_topo += \
 113                 [(idx) if idx.isdigit() else idx for idx in cpuid.split(',')]
 114
 115         return [cpu.strip() for cpu in cpu_topo]
 116
 117     def scale(self, flavor=""):
 118         ''' scale vnfbased on flavor input '''
 119         super(VpeApproxVnf, self).scale(flavor)
 120
 121     def instantiate(self, scenario_cfg, context_cfg):
 122         vnf_cfg = scenario_cfg['vnf_options']['vpe']['cfg']
 123         mgmt_interface = self.vnfd["mgmt-interface"]
 124         ssh_port = mgmt_interface.get("ssh_port", ssh.DEFAULT_PORT)
 125
 126         self.connection = ssh.SSH(mgmt_interface["user"], mgmt_interface["ip"],
 127                                   password=mgmt_interface["password"],
 128                                   port=ssh_port)
 129
 130         self.connection.wait()
 131
 132         self.setup_vnf_environment(self.connection)
 133
 134         cores = self._get_cpu_sibling_list()
 135         self.resource = ResourceProfile(self.vnfd, cores)
 136
 137         self.connection.execute("pkill vPE_vnf")
 138         dpdk_nic_bind = \
 139             provision_tool(self.connection,
 140                            os.path.join(self.bin_path, "dpdk_nic_bind.py"))
 141
 142         interfaces = self.vnfd["vdu"][0]['external-interface']
 143         self.socket = \
 144             next((0 for v in interfaces
 145                   if v['virtual-interface']["vpci"][5] == "0"), 1)
 146
 147         bound_pci = [v['virtual-interface']["vpci"] for v in interfaces]
 148         for vpci in bound_pci:
 149             self.connection.execute(
 150                 "%s --force -b igb_uio %s" % (dpdk_nic_bind, vpci))
 151         queue_wrapper = \
 152             QueueFileWrapper(self.q_in, self.q_out, "pipeline>")
 153         self._vnf_process = multiprocessing.Process(target=self._run_vpe,
 154                                                     args=(queue_wrapper,
 155                                                           vnf_cfg,))
 156         self._vnf_process.start()
 157         buf = []
 158         time.sleep(WAIT_TIME)  # Give some time for config to load
 159         while True:
 160             message = ''
 161             while self.q_out.qsize() > 0:
 162                 buf.append(self.q_out.get())
 163                 message = ''.join(buf)
 164                 if "pipeline>" in message:
 165                     LOG.info("VPE VNF is up and running.")
 166                     queue_wrapper.clear()
 167                     self._resource_collect_start()
 168                     return self._vnf_process.exitcode
 169                 if "PANIC" in message:
 170                     raise RuntimeError("Error starting vPE VNF.")
 171
 172             LOG.info("Waiting for VNF to start.. ")
 173             time.sleep(3)
 174             if not self._vnf_process.is_alive():
 175                 raise RuntimeError("vPE VNF process died.")
 176
 177     def _get_ports_gateway(self, name):
 178         if 'routing_table' in self.vnfd['vdu'][0]:
 179             routing_table = self.vnfd['vdu'][0]['routing_table']
 180
 181             for route in routing_table:
 182                 if name == route['if']:
 183                     return route['gateway']
 184
 185     def terminate(self):
 186         self.execute_command("quit")
 187         if self._vnf_process:
 188             self._vnf_process.terminate()
 189
 190     def _run_vpe(self, filewrapper, vnf_cfg):
 191         mgmt_interface = self.vnfd["mgmt-interface"]
 192         ssh_port = mgmt_interface.get("ssh_port", ssh.DEFAULT_PORT)
 193         self.connection = ssh.SSH(mgmt_interface["user"], mgmt_interface["ip"],
 194                                   password=mgmt_interface["password"],
 195                                   port=ssh_port)
 196         self.connection.wait()
 197         interfaces = self.vnfd["vdu"][0]['external-interface']
 198         port0_ip = ipaddress.ip_interface(six.text_type(
 199             "%s/%s" % (interfaces[0]["virtual-interface"]["local_ip"],
 200                        interfaces[0]["virtual-interface"]["netmask"])))
 201         port1_ip = ipaddress.ip_interface(six.text_type(
 202             "%s/%s" % (interfaces[1]["virtual-interface"]["local_ip"],
 203                        interfaces[1]["virtual-interface"]["netmask"])))
 204         dst_port0_ip = ipaddress.ip_interface(
 205             u"%s/%s" % (interfaces[0]["virtual-interface"]["dst_ip"],
 206                         interfaces[0]["virtual-interface"]["netmask"]))
 207         dst_port1_ip = ipaddress.ip_interface(
 208             u"%s/%s" % (interfaces[1]["virtual-interface"]["dst_ip"],
 209                         interfaces[1]["virtual-interface"]["netmask"]))
 210
 211         vpe_vars = {"port0_local_ip": port0_ip.ip.exploded,
 212                     "port0_dst_ip": dst_port0_ip.ip.exploded,
 213                     "port0_local_ip_hex":
 214                     self._ip_to_hex(port0_ip.ip.exploded),
 215                     "port0_prefixlen": port0_ip.network.prefixlen,
 216                     "port0_netmask": port0_ip.network.netmask.exploded,
 217                     "port0_netmask_hex":
 218                     self._ip_to_hex(port0_ip.network.netmask.exploded),
 219                     "port0_local_mac":
 220                     interfaces[0]["virtual-interface"]["local_mac"],
 221                     "port0_dst_mac":
 222                     interfaces[0]["virtual-interface"]["dst_mac"],
 223                     "port0_gateway":
 224                     self._get_ports_gateway(interfaces[0]["name"]),
 225                     "port0_local_network":
 226                     port0_ip.network.network_address.exploded,
 227                     "port0_prefix": port0_ip.network.prefixlen,
 228                     "port1_local_ip": port1_ip.ip.exploded,
 229                     "port1_dst_ip": dst_port1_ip.ip.exploded,
 230                     "port1_local_ip_hex":
 231                     self._ip_to_hex(port1_ip.ip.exploded),
 232                     "port1_prefixlen": port1_ip.network.prefixlen,
 233                     "port1_netmask": port1_ip.network.netmask.exploded,
 234                     "port1_netmask_hex":
 235                     self._ip_to_hex(port1_ip.network.netmask.exploded),
 236                     "port1_local_mac":
 237                     interfaces[1]["virtual-interface"]["local_mac"],
 238                     "port1_dst_mac":
 239                     interfaces[1]["virtual-interface"]["dst_mac"],
 240                     "port1_gateway":
 241                     self._get_ports_gateway(interfaces[1]["name"]),
 242                     "port1_local_network":
 243                     port1_ip.network.network_address.exploded,
 244                     "port1_prefix": port1_ip.network.prefixlen,
 245                     "port0_local_ip6": self._get_port0localip6(),
 246                     "port1_local_ip6": self._get_port1localip6(),
 247                     "port0_prefixlen6": self._get_port0prefixlen6(),
 248                     "port1_prefixlen6": self._get_port1prefixlen6(),
 249                     "port0_gateway6": self._get_port0gateway6(),
 250                     "port1_gateway6": self._get_port1gateway6(),
 251                     "port0_dst_ip_hex6": self._get_port0localip6(),
 252                     "port1_dst_ip_hex6": self._get_port1localip6(),
 253                     "port0_dst_netmask_hex6": self._get_port0prefixlen6(),
 254                     "port1_dst_netmask_hex6": self._get_port1prefixlen6(),
 255                     "bin_path": self.bin_path,
 256                     "socket": self.socket}
 257
 258         for cfg in os.listdir(vnf_cfg):
 259             vpe_config = ""
 260             with open(os.path.join(vnf_cfg, cfg), 'r') as vpe_cfg:
 261                 vpe_config = vpe_cfg.read()
 262
 263             self._provide_config_file(cfg, vpe_config, vpe_vars)
 264
 265         LOG.info("Provision and start the vPE")
 266         tool_path = provision_tool(self.connection,
 267                                    os.path.join(self.bin_path, "vPE_vnf"))
 268         cmd = VPE_PIPELINE_COMMAND.format(cfg_file="/tmp/vpe_config",
 269                                           script="/tmp/vpe_script",
 270                                           tool_path=tool_path)
 271         self.connection.run(cmd, stdin=filewrapper, stdout=filewrapper,
 272                             keep_stdin_open=True, pty=True)
 273
 274     def _provide_config_file(self, prefix, template, args):
 275         cfg, cfg_content = tempfile.mkstemp()
 276         cfg = os.fdopen(cfg, "w+")
 277         cfg.write(template.format(**args))
 278         cfg.close()
 279         cfg_file = "/tmp/%s" % prefix
 280         self.connection.put(cfg_content, cfg_file)
 281         return cfg_file
 282
 283     def execute_command(self, cmd):
 284         ''' send cmd to vnf process '''
 285         LOG.info("VPE command: %s", cmd)
 286         output = []
 287         if self.q_in:
 288             self.q_in.put(cmd + "\r\n")
 289             time.sleep(3)
 290             while self.q_out.qsize() > 0:
 291                 output.append(self.q_out.get())
 292         return "".join(output)
 293
 294     def collect_kpi(self):
 295         result = self.get_stats_vpe()
 296         collect_stats = self._collect_resource_kpi()
 297         result["collect_stats"] = collect_stats
 298         LOG.debug("vPE collet Kpis: %s", result)
 299         return result
 300
 301     def get_stats_vpe(self):
 302         ''' get vpe statistics '''
 303         result = {'pkt_in_up_stream': 0, 'pkt_drop_up_stream': 0,
 304                   'pkt_in_down_stream': 0, 'pkt_drop_down_stream': 0}
 305         up_stat_commands = ['p 5 stats port in 0', 'p 5 stats port out 0',
 306                             'p 5 stats port out 1']
 307         down_stat_commands = ['p 9 stats port in 0', 'p 9 stats port out 0']
 308         pattern = \
 309             "Pkts in:\\s(\\d+)\\r\\n\\tPkts dropped by " \
 310             "AH:\\s(\\d+)\\r\\n\\tPkts dropped by other:\\s(\\d+)"
 311
 312         for cmd in up_stat_commands:
 313             stats = self.execute_command(cmd)
 314             match = re.search(pattern, stats, re.MULTILINE)
 315             if match:
 316                 result["pkt_in_up_stream"] = \
 317                     result.get("pkt_in_up_stream", 0) + int(match.group(1))
 318                 result["pkt_drop_up_stream"] = \
 319                     result.get("pkt_drop_up_stream", 0) + \
 320                     int(match.group(2)) + int(match.group(3))
 321
 322         for cmd in down_stat_commands:
 323             stats = self.execute_command(cmd)
 324             match = re.search(pattern, stats, re.MULTILINE)
 325             if match:
 326                 result["pkt_in_down_stream"] = \
 327                     result.get("pkt_in_down_stream", 0) + int(match.group(1))
 328                 result["pkt_drop_down_stream"] = \
 329                     result.get("pkt_drop_down_stream", 0) + \
 330                     int(match.group(2)) + int(match.group(3))
 331         return result