yardstick/network_services/vnf_generic/vnf/tg_rfc2544_trex.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 """ Trex traffic generation definitions which implements rfc2544 """
  15
  16 from __future__ import absolute_import
  17 from __future__ import print_function
  18 import multiprocessing
  19 import time
  20 import logging
  21 import os
  22 import yaml
  23
  24 from yardstick import ssh
  25 from yardstick.network_services.vnf_generic.vnf.base import GenericTrafficGen
  26 from yardstick.network_services.utils import get_nsb_option
  27 from stl.trex_stl_lib.trex_stl_client import STLClient
  28 from stl.trex_stl_lib.trex_stl_client import LoggerApi
  29 from stl.trex_stl_lib.trex_stl_exceptions import STLError
  30
  31 LOGGING = logging.getLogger(__name__)
  32
  33 DURATION = 30
  34 WAIT_TIME = 3
  35 TREX_SYNC_PORT = 4500
  36 TREX_ASYNC_PORT = 4501
  37
  38
  39 class TrexTrafficGenRFC(GenericTrafficGen):
  40     """
  41     This class handles mapping traffic profile and generating
  42     traffic for rfc2544 testcase.
  43     """
  44
  45     def __init__(self, vnfd):
  46         super(TrexTrafficGenRFC, self).__init__(vnfd)
  47         self._result = {}
  48         self._terminated = multiprocessing.Value('i', 0)
  49         self._queue = multiprocessing.Queue()
  50         self._terminated = multiprocessing.Value('i', 0)
  51         self._traffic_process = None
  52         self._vpci_ascending = None
  53         self.tc_file_name = None
  54         self.client = None
  55         self.my_ports = None
  56
  57         mgmt_interface = self.vnfd["mgmt-interface"]
  58         ssh_port = mgmt_interface.get("ssh_port", ssh.DEFAULT_PORT)
  59         self.connection = ssh.SSH(mgmt_interface["user"], mgmt_interface["ip"],
  60                                   password=mgmt_interface["password"],
  61                                   port=ssh_port)
  62         self.connection.wait()
  63
  64     @classmethod
  65     def _split_mac_address_into_list(cls, mac):
  66         octets = mac.split(':')
  67         for i, elem in enumerate(octets):
  68             octets[i] = "0x" + str(elem)
  69         return octets
  70
  71     def _generate_trex_cfg(self, vnfd):
  72         """
  73
  74         :param vnfd: vnfd.yaml
  75         :return: trex_cfg.yaml file
  76         """
  77         trex_cfg = dict(
  78             port_limit=0,
  79             version='2',
  80             interfaces=[],
  81             port_info=list(dict(
  82             ))
  83         )
  84         trex_cfg["port_limit"] = len(vnfd["vdu"][0]["external-interface"])
  85         trex_cfg["version"] = '2'
  86
  87         cfg_file = []
  88         vpci = []
  89         port = {}
  90
  91         ext_intf = vnfd["vdu"][0]["external-interface"]
  92         for interface in ext_intf:
  93             virt_intf = interface["virtual-interface"]
  94             vpci.append(virt_intf["vpci"])
  95
  96             port["src_mac"] = \
  97                 self._split_mac_address_into_list(virt_intf["local_mac"])
  98
  99             time.sleep(WAIT_TIME)
 100             port["dest_mac"] = \
 101                 self._split_mac_address_into_list(virt_intf["dst_mac"])
 102             if virt_intf["dst_mac"]:
 103                 trex_cfg["port_info"].append(port.copy())
 104
 105         trex_cfg["interfaces"] = vpci
 106         cfg_file.append(trex_cfg)
 107
 108         with open('/tmp/trex_cfg.yaml', 'w') as outfile:
 109             outfile.write(yaml.safe_dump(cfg_file, default_flow_style=False))
 110         self.connection.put('/tmp/trex_cfg.yaml', '/etc')
 111
 112         self._vpci_ascending = sorted(vpci)
 113
 114     def scale(self, flavor=""):
 115         ''' scale vnfbased on flavor input '''
 116         super(TrexTrafficGenRFC, self).scale(flavor)
 117
 118     def instantiate(self, scenario_cfg, context_cfg):
 119         self._generate_trex_cfg(self.vnfd)
 120         self.tc_file_name = '{0}.yaml'.format(scenario_cfg['tc'])
 121         trex = os.path.join(self.bin_path, "trex")
 122         err, _, _ = \
 123             self.connection.execute("ls {} >/dev/null 2>&1".format(trex))
 124         if err != 0:
 125             self.connection.put(trex, trex, True)
 126
 127         LOGGING.debug("Starting TRex server...")
 128         _tg_server = \
 129             multiprocessing.Process(target=self._start_server)
 130         _tg_server.start()
 131         while True:
 132             LOGGING.info("Waiting for TG Server to start.. ")
 133             time.sleep(WAIT_TIME)
 134
 135             status = \
 136                 self.connection.execute("lsof -i:%s" % TREX_SYNC_PORT)[0]
 137             if status == 0:
 138                 LOGGING.info("TG server is up and running.")
 139                 return _tg_server.exitcode
 140             if not _tg_server.is_alive():
 141                 raise RuntimeError("Traffic Generator process died.")
 142
 143     def listen_traffic(self, traffic_profile):
 144         pass
 145
 146     def _get_logical_if_name(self, vpci):
 147         ext_intf = self.vnfd["vdu"][0]["external-interface"]
 148         for interface in range(len(self.vnfd["vdu"][0]["external-interface"])):
 149             virtual_intf = ext_intf[interface]["virtual-interface"]
 150             if virtual_intf["vpci"] == vpci:
 151                 return ext_intf[interface]["name"]
 152
 153     def run_traffic(self, traffic_profile,
 154                     client_started=multiprocessing.Value('i', 0)):
 155
 156         self._traffic_process = \
 157             multiprocessing.Process(target=self._traffic_runner,
 158                                     args=(traffic_profile, self._queue,
 159                                           client_started, self._terminated))
 160         self._traffic_process.start()
 161         # Wait for traffic process to start
 162         while client_started.value == 0:
 163             time.sleep(1)
 164
 165         return self._traffic_process.is_alive()
 166
 167     def _start_server(self):
 168         mgmt_interface = self.vnfd["mgmt-interface"]
 169         ssh_port = mgmt_interface.get("ssh_port", ssh.DEFAULT_PORT)
 170         _server = ssh.SSH(mgmt_interface["user"], mgmt_interface["ip"],
 171                           password=mgmt_interface["password"],
 172                           port=ssh_port)
 173         _server.wait()
 174
 175         _server.execute("fuser -n tcp %s %s -k > /dev/null 2>&1" %
 176                         (TREX_SYNC_PORT, TREX_ASYNC_PORT))
 177         _server.execute("pkill -9 rex > /dev/null 2>&1")
 178
 179         trex_path = os.path.join(self.bin_path, "trex/scripts")
 180         path = get_nsb_option("trex_path", trex_path)
 181         trex_cmd = "cd " + path + "; sudo ./t-rex-64 -i > /dev/null 2>&1"
 182
 183         _server.execute(trex_cmd)
 184
 185     def _connect_client(self, client=None):
 186         if client is None:
 187             client = STLClient(username=self.vnfd["mgmt-interface"]["user"],
 188                                server=self.vnfd["mgmt-interface"]["ip"],
 189                                verbose_level=LoggerApi.VERBOSE_QUIET)
 190         for idx in range(6):
 191             try:
 192                 client.connect()
 193                 break
 194             except STLError:
 195                 LOGGING.info("Unable to connect to Trex. Attempt %s", idx)
 196                 time.sleep(WAIT_TIME)
 197         return client
 198
 199     @classmethod
 200     def _get_rfc_tolerance(cls, tc_yaml):
 201         tolerance = '0.8 - 1.0'
 202         if 'tc_options' in tc_yaml['scenarios'][0]:
 203             tc_options = tc_yaml['scenarios'][0]['tc_options']
 204             if 'rfc2544' in tc_options:
 205                 tolerance = \
 206                     tc_options['rfc2544'].get('allowed_drop_rate', '0.8 - 1.0')
 207
 208         tolerance = tolerance.split('-')
 209         min_tol = float(tolerance[0])
 210         if len(tolerance) == 2:
 211             max_tol = float(tolerance[1])
 212         else:
 213             max_tol = float(tolerance[0])
 214
 215         return [min_tol, max_tol]
 216
 217     def _traffic_runner(self, traffic_profile, queue,
 218                         client_started, terminated):
 219         LOGGING.info("Starting TRex client...")
 220         tc_yaml = {}
 221
 222         with open(self.tc_file_name) as tc_file:
 223             tc_yaml = yaml.load(tc_file.read())
 224
 225         tolerance = self._get_rfc_tolerance(tc_yaml)
 226
 227         # fixme: fix passing correct trex config file,
 228         # instead of searching the default path
 229         self.my_ports = [0, 1]
 230         self.client = self._connect_client()
 231         self.client.reset(ports=self.my_ports)
 232         self.client.remove_all_streams(self.my_ports)  # remove all streams
 233         while not terminated.value:
 234             traffic_profile.execute(self)
 235             client_started.value = 1
 236             time.sleep(DURATION)
 237             self.client.stop(self.my_ports)
 238             time.sleep(WAIT_TIME)
 239             last_res = self.client.get_stats(self.my_ports)
 240             samples = {}
 241             for vpci_idx in range(len(self._vpci_ascending)):
 242                 name = \
 243                     self._get_logical_if_name(self._vpci_ascending[vpci_idx])
 244                 # fixme: VNFDs KPIs values needs to be mapped to TRex structure
 245                 if not isinstance(last_res, dict):
 246                     terminated.value = 1
 247                     last_res = {}
 248
 249                 samples[name] = \
 250                     {"rx_throughput_fps":
 251                      float(last_res.get(vpci_idx, {}).get("rx_pps", 0.0)),
 252                      "tx_throughput_fps":
 253                      float(last_res.get(vpci_idx, {}).get("tx_pps", 0.0)),
 254                      "rx_throughput_mbps":
 255                      float(last_res.get(vpci_idx, {}).get("rx_bps", 0.0)),
 256                      "tx_throughput_mbps":
 257                      float(last_res.get(vpci_idx, {}).get("tx_bps", 0.0)),
 258                      "in_packets":
 259                      last_res.get(vpci_idx, {}).get("ipackets", 0),
 260                      "out_packets":
 261                      last_res.get(vpci_idx, {}).get("opackets", 0)}
 262
 263             samples = \
 264                 traffic_profile.get_drop_percentage(self, samples,
 265                                                     tolerance[0], tolerance[1])
 266             queue.put(samples)
 267         self.client.stop(self.my_ports)
 268         self.client.disconnect()
 269         queue.put(samples)
 270
 271     def collect_kpi(self):
 272         if not self._queue.empty():
 273             result = self._queue.get()
 274             self._result.update(result)
 275         LOGGING.debug("trex collect Kpis %s", self._result)
 276         return self._result
 277
 278     def terminate(self):
 279         self._terminated.value = 1  # stop Trex clinet
 280
 281         self.connection.execute("fuser -n tcp %s %s -k > /dev/null 2>&1" %
 282                                 (TREX_SYNC_PORT, TREX_ASYNC_PORT))
 283
 284         if self._traffic_process:
 285             self._traffic_process.terminate()