nfvbench/traffic_client.py

   1 # Copyright 2016 Cisco Systems, Inc.  All rights reserved.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    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, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 """Interface to the traffic generator clients including NDR/PDR binary search."""
  16
  17 from datetime import datetime
  18 from math import gcd
  19 import socket
  20 import struct
  21 import time
  22
  23 from attrdict import AttrDict
  24 import bitmath
  25 from netaddr import IPNetwork
  26 # pylint: disable=import-error
  27 from trex.stl.api import Ether
  28 from trex.stl.api import STLError
  29 from trex.stl.api import UDP
  30 # pylint: disable=wrong-import-order
  31 from scapy.contrib.mpls import MPLS  # flake8: noqa
  32 # pylint: enable=wrong-import-order
  33 # pylint: enable=import-error
  34
  35 from .log import LOG
  36 from .packet_stats import InterfaceStats
  37 from .packet_stats import PacketPathStats
  38 from .stats_collector import IntervalCollector
  39 from .stats_collector import IterationCollector
  40 from .traffic_gen import traffic_utils as utils
  41 from .utils import cast_integer
  42
  43 class TrafficClientException(Exception):
  44     """Generic traffic client exception."""
  45
  46 class TrafficRunner(object):
  47     """Serialize various steps required to run traffic."""
  48
  49     def __init__(self, client, duration_sec, interval_sec=0, service_mode=False):
  50         """Create a traffic runner."""
  51         self.client = client
  52         self.start_time = None
  53         self.duration_sec = duration_sec
  54         self.interval_sec = interval_sec
  55         self.service_mode = service_mode
  56
  57     def run(self):
  58         """Clear stats and instruct the traffic generator to start generating traffic."""
  59         if self.is_running():
  60             return None
  61         LOG.info('Running traffic generator')
  62         self.client.gen.clear_stats()
  63         # Debug use only : new '--service-mode' option available for the NFVBench command line.
  64         # A read-only mode TRex console would be able to capture the generated traffic.
  65         self.client.gen.set_service_mode(enabled=self.service_mode)
  66         LOG.info('Service mode is %sabled', 'en' if self.service_mode else 'dis')
  67         self.client.gen.start_traffic()
  68         self.start_time = time.time()
  69         return self.poll_stats()
  70
  71     def stop(self):
  72         """Stop the current run and instruct the traffic generator to stop traffic."""
  73         if self.is_running():
  74             self.start_time = None
  75             self.client.gen.stop_traffic()
  76
  77     def is_running(self):
  78         """Check if a run is still pending."""
  79         return self.start_time is not None
  80
  81     def time_elapsed(self):
  82         """Return time elapsed since start of run."""
  83         if self.is_running():
  84             return time.time() - self.start_time
  85         return self.duration_sec
  86
  87     def poll_stats(self):
  88         """Poll latest stats from the traffic generator at fixed interval - sleeps if necessary.
  89
  90         return: latest stats or None if traffic is stopped
  91         """
  92         if not self.is_running():
  93             return None
  94         if self.client.skip_sleep():
  95             self.stop()
  96             return self.client.get_stats()
  97         time_elapsed = self.time_elapsed()
  98         if time_elapsed > self.duration_sec:
  99             self.stop()
 100             return None
 101         time_left = self.duration_sec - time_elapsed
 102         if self.interval_sec > 0.0:
 103             if time_left <= self.interval_sec:
 104                 time.sleep(time_left)
 105                 self.stop()
 106             else:
 107                 time.sleep(self.interval_sec)
 108         else:
 109             time.sleep(self.duration_sec)
 110             self.stop()
 111         return self.client.get_stats()
 112
 113
 114 class IpBlock(object):
 115     """Manage a block of IP addresses."""
 116
 117     def __init__(self, base_ip, step_ip, count_ip):
 118         """Create an IP block."""
 119         self.base_ip_int = Device.ip_to_int(base_ip)
 120         self.step = Device.ip_to_int(step_ip)
 121         self.max_available = count_ip
 122         self.next_free = 0
 123
 124     def get_ip(self, index=0):
 125         """Return the IP address at given index."""
 126         if index < 0 or index >= self.max_available:
 127             raise IndexError('Index out of bounds: %d (max=%d)' % (index, self.max_available))
 128         return Device.int_to_ip(self.base_ip_int + index * self.step)
 129
 130     def reserve_ip_range(self, count, force_ip_reservation=False):
 131         """Reserve a range of count consecutive IP addresses spaced by step.
 132         force_ip_reservation parameter allows to continue the calculation of IPs when
 133         the 2 sides (ports) have different size and the flow is greater than
 134         the size as well.
 135         """
 136         if self.next_free + count > self.max_available and force_ip_reservation is False:
 137             raise IndexError('No more IP addresses next free=%d max_available=%d requested=%d' %
 138                              (self.next_free,
 139                               self.max_available,
 140                               count))
 141         if self.next_free + count > self.max_available and force_ip_reservation is True:
 142             first_ip = self.get_ip(self.next_free)
 143             last_ip = self.get_ip(self.next_free + self.max_available - 1)
 144             self.next_free += self.max_available
 145         else:
 146             first_ip = self.get_ip(self.next_free)
 147             last_ip = self.get_ip(self.next_free + count - 1)
 148             self.next_free += count
 149         return (first_ip, last_ip)
 150
 151     def reset_reservation(self):
 152         """Reset all reservations and restart with a completely unused IP block."""
 153         self.next_free = 0
 154
 155
 156 class UdpPorts(object):
 157
 158     def __init__(self, src_min, src_max, dst_min, dst_max, step):
 159
 160         self.src_min = src_min
 161         self.src_max = src_max
 162         self.dst_min = dst_min
 163         self.dst_max = dst_max
 164         self.step = step
 165
 166
 167 class Device(object):
 168     """Represent a port device and all information associated to it.
 169
 170     In the curent version we only support 2 port devices for the traffic generator
 171     identified as port 0 or port 1.
 172     """
 173
 174     def __init__(self, port, generator_config):
 175         """Create a new device for a given port."""
 176         self.generator_config = generator_config
 177         self.chain_count = generator_config.service_chain_count
 178         if generator_config.bidirectional:
 179             self.flow_count = generator_config.flow_count / 2
 180         else:
 181             self.flow_count = generator_config.flow_count
 182
 183         self.port = port
 184         self.switch_port = generator_config.interfaces[port].get('switch_port', None)
 185         self.vtep_vlan = None
 186         self.vtep_src_mac = None
 187         self.vxlan = False
 188         self.mpls = False
 189         self.inner_labels = None
 190         self.outer_labels = None
 191         self.pci = generator_config.interfaces[port].pci
 192         self.mac = None
 193         self.dest_macs = None
 194         self.vtep_dst_mac = None
 195         self.vtep_dst_ip = None
 196         if generator_config.vteps is None:
 197             self.vtep_src_ip = None
 198         else:
 199             self.vtep_src_ip = generator_config.vteps[port]
 200         self.vnis = None
 201         self.vlans = None
 202         self.ip_addrs = generator_config.ip_addrs[port]
 203         self.ip_src_static = generator_config.ip_src_static
 204         self.ip_addrs_step = generator_config.ip_addrs_step
 205         if self.ip_addrs_step == 'random':
 206             # Set step to 1 to calculate the IP range size (see check_ip_size below)
 207             step = '0.0.0.1'
 208         else:
 209             step = self.ip_addrs_step
 210         self.ip_size = self.check_ipsize(IPNetwork(self.ip_addrs).size, Device.ip_to_int(step))
 211         self.ip = str(IPNetwork(self.ip_addrs).network)
 212         ip_addrs_left = generator_config.ip_addrs[0]
 213         ip_addrs_right = generator_config.ip_addrs[1]
 214         self.ip_addrs_size = {
 215             'left': self.check_ipsize(IPNetwork(ip_addrs_left).size, Device.ip_to_int(step)),
 216             'right': self.check_ipsize(IPNetwork(ip_addrs_right).size, Device.ip_to_int(step))}
 217         udp_src_port = generator_config.gen_config.udp_src_port
 218         if udp_src_port is None:
 219             udp_src_port = 53
 220         udp_dst_port = generator_config.gen_config.udp_dst_port
 221         if udp_dst_port is None:
 222             udp_dst_port = 53
 223         src_max, src_min = self.define_udp_range(udp_src_port, 'udp_src_port')
 224         dst_max, dst_min = self.define_udp_range(udp_dst_port, 'udp_dst_port')
 225         udp_src_range = int(src_max) - int(src_min) + 1
 226         udp_dst_range = int(dst_max) - int(dst_min) + 1
 227         lcm_port = self.lcm(udp_src_range, udp_dst_range)
 228         if self.ip_src_static is True:
 229             lcm_ip = self.lcm(1, min(self.ip_addrs_size['left'], self.ip_addrs_size['right']))
 230         else:
 231             lcm_ip = self.lcm(self.ip_addrs_size['left'], self.ip_addrs_size['right'])
 232         flow_max = self.lcm(lcm_port, lcm_ip)
 233         if self.flow_count > flow_max:
 234             raise TrafficClientException('Trying to set unachievable traffic (%d > %d)' %
 235                                          (self.flow_count, flow_max))
 236
 237         # manage udp range regarding flow count value
 238         # UDP dst range is greater than FC => range will be limited to min + FC
 239         if self.flow_count <= udp_dst_range:
 240             dst_max = int(dst_min) + self.flow_count - 1
 241         # UDP src range is greater than FC => range will be limited to min + FC
 242         if self.flow_count <= udp_src_range:
 243             src_max = int(src_min) + self.flow_count - 1
 244         # Define IP block limit regarding flow count
 245         if self.flow_count <= self.ip_size:
 246             self.ip_block = IpBlock(self.ip, step, self.flow_count)
 247         else:
 248             self.ip_block = IpBlock(self.ip, step, self.ip_size)
 249
 250         if generator_config.gen_config.udp_port_step == 'random':
 251             step = 1
 252         else:
 253             step = generator_config.gen_config.udp_port_step
 254         self.udp_ports = UdpPorts(src_min, src_max, dst_min, dst_max, step)
 255         self.gw_ip_block = IpBlock(generator_config.gateway_ips[port],
 256                                    generator_config.gateway_ip_addrs_step,
 257                                    self.chain_count)
 258         self.tg_gateway_ip_addrs = generator_config.tg_gateway_ip_addrs[port]
 259         self.tg_gw_ip_block = IpBlock(self.tg_gateway_ip_addrs,
 260                                       generator_config.tg_gateway_ip_addrs_step,
 261                                       self.chain_count)
 262
 263     @staticmethod
 264     def define_udp_range(udp_port, property_name):
 265         if isinstance(udp_port, int):
 266             min = udp_port
 267             max = min
 268         elif isinstance(udp_port, tuple):
 269             min = udp_port[0]
 270             max = udp_port[1]
 271         else:
 272             raise TrafficClientException('Invalid %s property value (53 or [\'53\',\'1024\'])'
 273                                          % property_name)
 274         return max, min
 275
 276     @staticmethod
 277     def lcm(a, b):
 278         """Calculate the maximum possible value for both IP and ports,
 279         eventually for maximum possible flux."""
 280         if a != 0 and b != 0:
 281             lcm_value = a * b // gcd(a, b)
 282             return lcm_value
 283         raise TypeError(" IP size or port range can't be zero !")
 284
 285     @staticmethod
 286     def check_ipsize(ip_size, step):
 287         """Check and set the available IPs, considering the step."""
 288         try:
 289             if ip_size % step == 0:
 290                 value = int(ip_size / step)
 291             else:
 292                 value = int((ip_size / step)) + 1
 293             return value
 294         except ZeroDivisionError:
 295             raise ZeroDivisionError("step can't be zero !")
 296
 297     def set_mac(self, mac):
 298         """Set the local MAC for this port device."""
 299         if mac is None:
 300             raise TrafficClientException('Trying to set traffic generator MAC address as None')
 301         self.mac = mac
 302
 303     def get_peer_device(self):
 304         """Get the peer device (device 0 -> device 1, or device 1 -> device 0)."""
 305         return self.generator_config.devices[1 - self.port]
 306
 307     def set_vtep_dst_mac(self, dest_macs):
 308         """Set the list of dest MACs indexed by the chain id.
 309
 310         This is only called in 2 cases:
 311         - VM macs discovered using openstack API
 312         - dest MACs provisioned in config file
 313         """
 314         self.vtep_dst_mac = list(map(str, dest_macs))
 315
 316     def set_dest_macs(self, dest_macs):
 317         """Set the list of dest MACs indexed by the chain id.
 318
 319         This is only called in 2 cases:
 320         - VM macs discovered using openstack API
 321         - dest MACs provisioned in config file
 322         """
 323         self.dest_macs = list(map(str, dest_macs))
 324
 325     def get_dest_macs(self):
 326         """Get the list of dest macs for this device.
 327
 328         If set_dest_macs was never called, assumes l2-loopback and return
 329         a list of peer mac (as many as chains but normally only 1 chain)
 330         """
 331         if self.dest_macs:
 332             return self.dest_macs
 333         # assume this is l2-loopback
 334         return [self.get_peer_device().mac] * self.chain_count
 335
 336     def set_vlans(self, vlans):
 337         """Set the list of vlans to use indexed by the chain id."""
 338         self.vlans = vlans
 339         LOG.info("Port %d: VLANs %s", self.port, self.vlans)
 340
 341     def set_vtep_vlan(self, vlan):
 342         """Set the vtep vlan to use indexed by specific port."""
 343         self.vtep_vlan = vlan
 344         self.vxlan = True
 345         self.vlan_tagging = None
 346         LOG.info("Port %d: VTEP VLANs %s", self.port, self.vtep_vlan)
 347
 348     def set_vxlan_endpoints(self, src_ip, dst_ip):
 349         self.vtep_dst_ip = dst_ip
 350         self.vtep_src_ip = src_ip
 351         LOG.info("Port %d: src_vtep %s, dst_vtep %s", self.port,
 352                  self.vtep_src_ip, self.vtep_dst_ip)
 353
 354     def set_mpls_peers(self, src_ip, dst_ip):
 355         self.mpls = True
 356         self.vtep_dst_ip = dst_ip
 357         self.vtep_src_ip = src_ip
 358         LOG.info("Port %d: src_mpls_vtep %s, mpls_peer_ip %s", self.port,
 359                  self.vtep_src_ip, self.vtep_dst_ip)
 360
 361     def set_vxlans(self, vnis):
 362         self.vnis = vnis
 363         LOG.info("Port %d: VNIs %s", self.port, self.vnis)
 364
 365     def set_mpls_inner_labels(self, labels):
 366         self.inner_labels = labels
 367         LOG.info("Port %d: MPLS Inner Labels %s", self.port, self.inner_labels)
 368
 369     def set_mpls_outer_labels(self, labels):
 370         self.outer_labels = labels
 371         LOG.info("Port %d: MPLS Outer Labels %s", self.port, self.outer_labels)
 372
 373     def set_gw_ip(self, gateway_ip):
 374         self.gw_ip_block = IpBlock(gateway_ip,
 375                                    self.generator_config.gateway_ip_addrs_step,
 376                                    self.chain_count)
 377
 378     def get_gw_ip(self, chain_index):
 379         """Retrieve the IP address assigned for the gateway of a given chain."""
 380         return self.gw_ip_block.get_ip(chain_index)
 381
 382     def get_stream_configs(self):
 383         """Get the stream config for a given chain on this device.
 384
 385         Called by the traffic generator driver to program the traffic generator properly
 386         before generating traffic
 387         """
 388         configs = []
 389         # exact flow count for each chain is calculated as follows:
 390         # - all chains except the first will have the same flow count
 391         #   calculated as (total_flows + chain_count - 1) / chain_count
 392         # - the first chain will have the remainder
 393         # example 11 flows and 3 chains => 3, 4, 4
 394         flows_per_chain = int((self.flow_count + self.chain_count - 1) / self.chain_count)
 395         cur_chain_flow_count = int(self.flow_count - flows_per_chain * (self.chain_count - 1))
 396         force_ip_reservation = False
 397         # use case example of this parameter:
 398         # - static IP addresses (source & destination), netmask = /30
 399         # - 4 varying UDP source ports | 1 UDP destination port
 400         # - Flow count = 8
 401         # --> parameter 'reserve_ip_range' should have flag 'force_ip_reservation'
 402         # in order to assign the maximum available IP on each iteration
 403         if self.ip_size < cur_chain_flow_count \
 404                 or self.ip_addrs_size['left'] != self.ip_addrs_size['right']:
 405             force_ip_reservation = True
 406
 407         peer = self.get_peer_device()
 408         self.ip_block.reset_reservation()
 409         peer.ip_block.reset_reservation()
 410         dest_macs = self.get_dest_macs()
 411
 412         for chain_idx in range(self.chain_count):
 413             src_ip_first, src_ip_last = self.ip_block.reserve_ip_range\
 414             (cur_chain_flow_count, force_ip_reservation)
 415             dst_ip_first, dst_ip_last = peer.ip_block.reserve_ip_range\
 416             (cur_chain_flow_count, force_ip_reservation)
 417
 418             configs.append({
 419                 'count': cur_chain_flow_count,
 420                 'mac_src': self.mac,
 421                 'mac_dst': dest_macs[chain_idx],
 422                 'ip_src_addr': src_ip_first,
 423                 'ip_src_addr_max': src_ip_last,
 424                 'ip_src_count': cur_chain_flow_count,
 425                 'ip_dst_addr': dst_ip_first,
 426                 'ip_dst_addr_max': dst_ip_last,
 427                 'ip_dst_count': cur_chain_flow_count,
 428                 'ip_addrs_step': self.ip_addrs_step,
 429                 'ip_src_static': self.ip_src_static,
 430                 'udp_src_port': self.udp_ports.src_min,
 431                 'udp_src_port_max': self.udp_ports.src_max,
 432                 'udp_src_count': int(self.udp_ports.src_max) - int(self.udp_ports.src_min) + 1,
 433                 'udp_dst_port': self.udp_ports.dst_min,
 434                 'udp_dst_port_max': self.udp_ports.dst_max,
 435                 'udp_dst_count': int(self.udp_ports.dst_max) - int(self.udp_ports.dst_min) + 1,
 436                 'udp_port_step': self.udp_ports.step,
 437                 'mac_discovery_gw': self.get_gw_ip(chain_idx),
 438                 'ip_src_tg_gw': self.tg_gw_ip_block.get_ip(chain_idx),
 439                 'ip_dst_tg_gw': peer.tg_gw_ip_block.get_ip(chain_idx),
 440                 'vlan_tag': self.vlans[chain_idx] if self.vlans else None,
 441                 'vxlan': self.vxlan,
 442                 'vtep_vlan': self.vtep_vlan if self.vtep_vlan else None,
 443                 'vtep_src_mac': self.mac if (self.vxlan or self.mpls) else None,
 444                 'vtep_dst_mac': self.vtep_dst_mac if (self.vxlan or self.mpls) else None,
 445                 'vtep_dst_ip': self.vtep_dst_ip if self.vxlan is True else None,
 446                 'vtep_src_ip': self.vtep_src_ip if self.vxlan is True else None,
 447                 'net_vni': self.vnis[chain_idx] if self.vxlan is True else None,
 448                 'mpls': self.mpls,
 449                 'mpls_outer_label': self.outer_labels[chain_idx] if self.mpls is True else None,
 450                 'mpls_inner_label': self.inner_labels[chain_idx] if self.mpls is True else None
 451
 452             })
 453             # after first chain, fall back to the flow count for all other chains
 454             cur_chain_flow_count = flows_per_chain
 455         return configs
 456
 457     @staticmethod
 458     def ip_to_int(addr):
 459         """Convert an IP address from string to numeric."""
 460         return struct.unpack("!I", socket.inet_aton(addr))[0]
 461
 462     @staticmethod
 463     def int_to_ip(nvalue):
 464         """Convert an IP address from numeric to string."""
 465         return socket.inet_ntoa(struct.pack("!I", int(nvalue)))
 466
 467
 468 class GeneratorConfig(object):
 469     """Represents traffic configuration for currently running traffic profile."""
 470
 471     DEFAULT_IP_STEP = '0.0.0.1'
 472     DEFAULT_SRC_DST_IP_STEP = '0.0.0.1'
 473
 474     def __init__(self, config):
 475         """Create a generator config."""
 476         self.config = config
 477         # name of the generator profile (normally trex or dummy)
 478         # pick the default one if not specified explicitly from cli options
 479         if not config.generator_profile:
 480             config.generator_profile = config.traffic_generator.default_profile
 481         # pick up the profile dict based on the name
 482         gen_config = self.__match_generator_profile(config.traffic_generator,
 483                                                     config.generator_profile)
 484         self.gen_config = gen_config
 485         # copy over fields from the dict
 486         self.tool = gen_config.tool
 487         self.ip = gen_config.ip
 488         # overrides on config.cores and config.mbuf_factor
 489         if config.cores:
 490             self.cores = config.cores
 491         else:
 492             self.cores = gen_config.get('cores', 1)
 493         self.mbuf_factor = config.mbuf_factor
 494         self.mbuf_64 = config.mbuf_64
 495         self.hdrh = not config.disable_hdrh
 496         if gen_config.intf_speed:
 497             # interface speed is overriden from config
 498             self.intf_speed = bitmath.parse_string(gen_config.intf_speed.replace('ps', '')).bits
 499         else:
 500             # interface speed is discovered/provided by the traffic generator
 501             self.intf_speed = 0
 502         self.name = gen_config.name
 503         self.zmq_pub_port = gen_config.get('zmq_pub_port', 4500)
 504         self.zmq_rpc_port = gen_config.get('zmq_rpc_port', 4501)
 505         self.limit_memory = gen_config.get('limit_memory', 1024)
 506         self.software_mode = gen_config.get('software_mode', False)
 507         self.interfaces = gen_config.interfaces
 508         if self.interfaces[0].port != 0 or self.interfaces[1].port != 1:
 509             raise TrafficClientException('Invalid port order/id in generator_profile.interfaces')
 510         self.service_chain = config.service_chain
 511         self.service_chain_count = config.service_chain_count
 512         self.flow_count = config.flow_count
 513         self.host_name = gen_config.host_name
 514         self.bidirectional = config.traffic.bidirectional
 515         self.tg_gateway_ip_addrs = gen_config.tg_gateway_ip_addrs
 516         self.ip_addrs = gen_config.ip_addrs
 517         self.ip_addrs_step = gen_config.ip_addrs_step or self.DEFAULT_SRC_DST_IP_STEP
 518         self.tg_gateway_ip_addrs_step = \
 519             gen_config.tg_gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 520         self.gateway_ip_addrs_step = gen_config.gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 521         self.gateway_ips = gen_config.gateway_ip_addrs
 522         self.ip_src_static = gen_config.ip_src_static
 523         self.vteps = gen_config.get('vteps')
 524         self.devices = [Device(port, self) for port in [0, 1]]
 525         # This should normally always be [0, 1]
 526         self.ports = [device.port for device in self.devices]
 527
 528         # check that pci is not empty
 529         if not gen_config.interfaces[0].get('pci', None) or \
 530            not gen_config.interfaces[1].get('pci', None):
 531             raise TrafficClientException("configuration interfaces pci fields cannot be empty")
 532
 533         self.pcis = [tgif['pci'] for tgif in gen_config.interfaces]
 534         self.vlan_tagging = config.vlan_tagging
 535
 536         # needed for result/summarizer
 537         config['tg-name'] = gen_config.name
 538         config['tg-tool'] = self.tool
 539
 540     def to_json(self):
 541         """Get json form to display the content into the overall result dict."""
 542         return dict(self.gen_config)
 543
 544     def set_dest_macs(self, port_index, dest_macs):
 545         """Set the list of dest MACs indexed by the chain id on given port.
 546
 547         port_index: the port for which dest macs must be set
 548         dest_macs: a list of dest MACs indexed by chain id
 549         """
 550         if len(dest_macs) < self.config.service_chain_count:
 551             raise TrafficClientException('Dest MAC list %s must have %d entries' %
 552                                          (dest_macs, self.config.service_chain_count))
 553         # only pass the first scc dest MACs
 554         self.devices[port_index].set_dest_macs(dest_macs[:self.config.service_chain_count])
 555         LOG.info('Port %d: dst MAC %s', port_index, [str(mac) for mac in dest_macs])
 556
 557     def set_vtep_dest_macs(self, port_index, dest_macs):
 558         """Set the list of dest MACs indexed by the chain id on given port.
 559
 560         port_index: the port for which dest macs must be set
 561         dest_macs: a list of dest MACs indexed by chain id
 562         """
 563         if len(dest_macs) != self.config.service_chain_count:
 564             raise TrafficClientException('Dest MAC list %s must have %d entries' %
 565                                          (dest_macs, self.config.service_chain_count))
 566         self.devices[port_index].set_vtep_dst_mac(dest_macs)
 567         LOG.info('Port %d: vtep dst MAC %s', port_index, {str(mac) for mac in dest_macs})
 568
 569     def get_dest_macs(self):
 570         """Return the list of dest macs indexed by port."""
 571         return [dev.get_dest_macs() for dev in self.devices]
 572
 573     def set_vlans(self, port_index, vlans):
 574         """Set the list of vlans to use indexed by the chain id on given port.
 575
 576         port_index: the port for which VLANs must be set
 577         vlans: a  list of vlan lists indexed by chain id
 578         """
 579         if len(vlans) != self.config.service_chain_count:
 580             raise TrafficClientException('VLAN list %s must have %d entries' %
 581                                          (vlans, self.config.service_chain_count))
 582         self.devices[port_index].set_vlans(vlans)
 583
 584     def set_vxlans(self, port_index, vxlans):
 585         """Set the list of vxlans (VNIs) to use indexed by the chain id on given port.
 586
 587         port_index: the port for which VXLANs must be set
 588         VXLANs: a  list of VNIs lists indexed by chain id
 589         """
 590         if len(vxlans) != self.config.service_chain_count:
 591             raise TrafficClientException('VXLAN list %s must have %d entries' %
 592                                          (vxlans, self.config.service_chain_count))
 593         self.devices[port_index].set_vxlans(vxlans)
 594
 595     def set_mpls_inner_labels(self, port_index, labels):
 596         """Set the list of MPLS Labels to use indexed by the chain id on given port.
 597
 598         port_index: the port for which Labels must be set
 599         Labels: a list of Labels lists indexed by chain id
 600         """
 601         if len(labels) != self.config.service_chain_count:
 602             raise TrafficClientException('Inner MPLS list %s must have %d entries' %
 603                                          (labels, self.config.service_chain_count))
 604         self.devices[port_index].set_mpls_inner_labels(labels)
 605
 606     def set_mpls_outer_labels(self, port_index, labels):
 607         """Set the list of MPLS Labels to use indexed by the chain id on given port.
 608
 609         port_index: the port for which Labels must be set
 610         Labels: a list of Labels lists indexed by chain id
 611         """
 612         if len(labels) != self.config.service_chain_count:
 613             raise TrafficClientException('Outer MPLS list %s must have %d entries' %
 614                                          (labels, self.config.service_chain_count))
 615         self.devices[port_index].set_mpls_outer_labels(labels)
 616
 617     def set_vtep_vlan(self, port_index, vlan):
 618         """Set the vtep vlan to use indexed by the chain id on given port.
 619         port_index: the port for which VLAN must be set
 620         """
 621         self.devices[port_index].set_vtep_vlan(vlan)
 622
 623     def set_vxlan_endpoints(self, port_index, src_ip, dst_ip):
 624         self.devices[port_index].set_vxlan_endpoints(src_ip, dst_ip)
 625
 626     def set_mpls_peers(self, port_index, src_ip, dst_ip):
 627         self.devices[port_index].set_mpls_peers(src_ip, dst_ip)
 628
 629     @staticmethod
 630     def __match_generator_profile(traffic_generator, generator_profile):
 631         gen_config = AttrDict(traffic_generator)
 632         gen_config.pop('default_profile')
 633         gen_config.pop('generator_profile')
 634         matching_profile = [profile for profile in traffic_generator.generator_profile if
 635                             profile.name == generator_profile]
 636         if len(matching_profile) != 1:
 637             raise Exception('Traffic generator profile not found: ' + generator_profile)
 638
 639         gen_config.update(matching_profile[0])
 640         return gen_config
 641
 642
 643 class TrafficClient(object):
 644     """Traffic generator client with NDR/PDR binary seearch."""
 645
 646     PORTS = [0, 1]
 647
 648     def __init__(self, config, notifier=None):
 649         """Create a new TrafficClient instance.
 650
 651         config: nfvbench config
 652         notifier: notifier (optional)
 653
 654         A new instance is created everytime the nfvbench config may have changed.
 655         """
 656         self.config = config
 657         self.generator_config = GeneratorConfig(config)
 658         self.tool = self.generator_config.tool
 659         self.gen = self._get_generator()
 660         self.notifier = notifier
 661         self.interval_collector = None
 662         self.iteration_collector = None
 663         self.runner = TrafficRunner(self, self.config.duration_sec, self.config.interval_sec,
 664                                     self.config.service_mode)
 665         self.config.frame_sizes = self._get_frame_sizes()
 666         self.run_config = {
 667             'l2frame_size': None,
 668             'duration_sec': self.config.duration_sec,
 669             'bidirectional': True,
 670             'rates': []  # to avoid unsbuscriptable-obj warning
 671         }
 672         self.current_total_rate = {'rate_percent': '10'}
 673         if self.config.single_run:
 674             self.current_total_rate = utils.parse_rate_str(self.config.rate)
 675         self.ifstats = None
 676         # Speed is either discovered when connecting to TG or set from config
 677         # This variable is 0 if not yet discovered from TG or must be the speed of
 678         # each interface in bits per second
 679         self.intf_speed = self.generator_config.intf_speed
 680
 681     def _get_generator(self):
 682         tool = self.tool.lower()
 683         if tool == 'trex':
 684             from .traffic_gen import trex_gen
 685             return trex_gen.TRex(self)
 686         if tool == 'dummy':
 687             from .traffic_gen import dummy
 688             return dummy.DummyTG(self)
 689         raise TrafficClientException('Unsupported generator tool name:' + self.tool)
 690
 691     def skip_sleep(self):
 692         """Skip all sleeps when doing unit testing with dummy TG.
 693
 694         Must be overriden using mock.patch
 695         """
 696         return False
 697
 698     def _get_frame_sizes(self):
 699         traffic_profile_name = self.config.traffic.profile
 700         matching_profiles = [profile for profile in self.config.traffic_profile if
 701                              profile.name == traffic_profile_name]
 702         if len(matching_profiles) > 1:
 703             raise TrafficClientException('Multiple traffic profiles with name: ' +
 704                                          traffic_profile_name)
 705         if not matching_profiles:
 706             raise TrafficClientException('Cannot find traffic profile: ' + traffic_profile_name)
 707         return matching_profiles[0].l2frame_size
 708
 709     def start_traffic_generator(self):
 710         """Start the traffic generator process (traffic not started yet)."""
 711         self.gen.connect()
 712         # pick up the interface speed if it is not set from config
 713         intf_speeds = self.gen.get_port_speed_gbps()
 714         # convert Gbps unit into bps
 715         tg_if_speed = bitmath.parse_string(str(intf_speeds[0]) + 'Gb').bits
 716         if self.intf_speed:
 717             # interface speed is overriden from config
 718             if self.intf_speed != tg_if_speed:
 719                 # Warn the user if the speed in the config is different
 720                 LOG.warning('Interface speed provided is different from actual speed (%d Gbps)',
 721                             intf_speeds[0])
 722         else:
 723             # interface speed not provisioned by config
 724             self.intf_speed = tg_if_speed
 725             # also update the speed in the tg config
 726             self.generator_config.intf_speed = tg_if_speed
 727
 728         # Save the traffic generator local MAC
 729         for mac, device in zip(self.gen.get_macs(), self.generator_config.devices):
 730             device.set_mac(mac)
 731
 732     def setup(self):
 733         """Set up the traffic client."""
 734         self.gen.clear_stats()
 735
 736     def get_version(self):
 737         """Get the traffic generator version."""
 738         return self.gen.get_version()
 739
 740     def ensure_end_to_end(self):
 741         """Ensure traffic generator receives packets it has transmitted.
 742
 743         This ensures end to end connectivity and also waits until VMs are ready to forward packets.
 744
 745         VMs that are started and in active state may not pass traffic yet. It is imperative to make
 746         sure that all VMs are passing traffic in both directions before starting any benchmarking.
 747         To verify this, we need to send at a low frequency bi-directional packets and make sure
 748         that we receive all packets back from all VMs. The number of flows is equal to 2 times
 749         the number of chains (1 per direction) and we need to make sure we receive packets coming
 750         from exactly 2 x chain count different source MAC addresses.
 751
 752         Example:
 753             PVP chain (1 VM per chain)
 754             N = 10 (number of chains)
 755             Flow count = 20 (number of flows)
 756             If the number of unique source MAC addresses from received packets is 20 then
 757             all 10 VMs 10 VMs are in operational state.
 758         """
 759         LOG.info('Starting traffic generator to ensure end-to-end connectivity')
 760         # send 2pps on each chain and each direction
 761         rate_pps = {'rate_pps': str(self.config.service_chain_count * 2)}
 762         self.gen.create_traffic('64', [rate_pps, rate_pps], bidirectional=True, latency=False,
 763                                 e2e=True)
 764         # ensures enough traffic is coming back
 765         retry_count = int((self.config.check_traffic_time_sec +
 766                            self.config.generic_poll_sec - 1) / self.config.generic_poll_sec)
 767
 768         # we expect to see packets coming from 2 unique MAC per chain
 769         # because there can be flooding in the case of shared net
 770         # we must verify that packets from the right VMs are received
 771         # and not just count unique src MAC
 772         # create a dict of (port, chain) tuples indexed by dest mac
 773         mac_map = {}
 774         for port, dest_macs in enumerate(self.generator_config.get_dest_macs()):
 775             for chain, mac in enumerate(dest_macs):
 776                 mac_map[mac] = (port, chain)
 777         unique_src_mac_count = len(mac_map)
 778         if self.config.vxlan and self.config.traffic_generator.vtep_vlan:
 779             get_mac_id = lambda packet: packet['binary'][60:66]
 780         elif self.config.vxlan:
 781             get_mac_id = lambda packet: packet['binary'][56:62]
 782         elif self.config.mpls:
 783             get_mac_id = lambda packet: packet['binary'][24:30]
 784             # mpls_transport_label = lambda packet: packet['binary'][14:18]
 785         else:
 786             get_mac_id = lambda packet: packet['binary'][6:12]
 787         for it in range(retry_count):
 788             self.gen.clear_stats()
 789             self.gen.start_traffic()
 790             self.gen.start_capture()
 791             LOG.info('Captured unique src mac %d/%d, capturing return packets (retry %d/%d)...',
 792                      unique_src_mac_count - len(mac_map), unique_src_mac_count,
 793                      it + 1, retry_count)
 794             if not self.skip_sleep():
 795                 time.sleep(self.config.generic_poll_sec)
 796             self.gen.stop_traffic()
 797             self.gen.fetch_capture_packets()
 798             self.gen.stop_capture()
 799             for packet in self.gen.packet_list:
 800                 mac_id = get_mac_id(packet).decode('latin-1')
 801                 src_mac = ':'.join(["%02x" % ord(x) for x in mac_id])
 802                 if self.config.mpls:
 803                     if src_mac in mac_map and self.is_mpls(packet):
 804                         port, chain = mac_map[src_mac]
 805                         LOG.info('Received mpls packet from mac: %s (chain=%d, port=%d)',
 806                                  src_mac, chain, port)
 807                         mac_map.pop(src_mac, None)
 808                 else:
 809                     if src_mac in mac_map and self.is_udp(packet):
 810                         port, chain = mac_map[src_mac]
 811                         LOG.info('Received udp packet from mac: %s (chain=%d, port=%d)',
 812                                  src_mac, chain, port)
 813                         mac_map.pop(src_mac, None)
 814
 815                 if not mac_map:
 816                     LOG.info('End-to-end connectivity established')
 817                     return
 818             if self.config.l3_router and not self.config.no_arp:
 819                 # In case of L3 traffic mode, routers are not able to route traffic
 820                 # until VM interfaces are up and ARP requests are done
 821                 LOG.info('Waiting for loopback service completely started...')
 822                 LOG.info('Sending ARP request to assure end-to-end connectivity established')
 823                 self.ensure_arp_successful()
 824         raise TrafficClientException('End-to-end connectivity cannot be ensured')
 825
 826     def is_udp(self, packet):
 827         pkt = Ether(packet['binary'])
 828         return UDP in pkt
 829
 830     def is_mpls(self, packet):
 831         pkt = Ether(packet['binary'])
 832         return MPLS in pkt
 833
 834     def ensure_arp_successful(self):
 835         """Resolve all IP using ARP and throw an exception in case of failure."""
 836         dest_macs = self.gen.resolve_arp()
 837         if dest_macs:
 838             # all dest macs are discovered, saved them into the generator config
 839             if self.config.vxlan or self.config.mpls:
 840                 self.generator_config.set_vtep_dest_macs(0, dest_macs[0])
 841                 self.generator_config.set_vtep_dest_macs(1, dest_macs[1])
 842             else:
 843                 self.generator_config.set_dest_macs(0, dest_macs[0])
 844                 self.generator_config.set_dest_macs(1, dest_macs[1])
 845         else:
 846             raise TrafficClientException('ARP cannot be resolved')
 847
 848     def set_traffic(self, frame_size, bidirectional):
 849         """Reconfigure the traffic generator for a new frame size."""
 850         self.run_config['bidirectional'] = bidirectional
 851         self.run_config['l2frame_size'] = frame_size
 852         self.run_config['rates'] = [self.get_per_direction_rate()]
 853         if bidirectional:
 854             self.run_config['rates'].append(self.get_per_direction_rate())
 855         else:
 856             unidir_reverse_pps = int(self.config.unidir_reverse_traffic_pps)
 857             if unidir_reverse_pps > 0:
 858                 self.run_config['rates'].append({'rate_pps': str(unidir_reverse_pps)})
 859         # Fix for [NFVBENCH-67], convert the rate string to PPS
 860         for idx, rate in enumerate(self.run_config['rates']):
 861             if 'rate_pps' not in rate:
 862                 self.run_config['rates'][idx] = {'rate_pps': self.__convert_rates(rate)['rate_pps']}
 863
 864         self.gen.clear_streamblock()
 865
 866         if self.config.no_latency_streams:
 867             LOG.info("Latency streams are disabled")
 868         self.gen.create_traffic(frame_size, self.run_config['rates'], bidirectional,
 869                                 latency=not self.config.no_latency_streams)
 870
 871     def _modify_load(self, load):
 872         self.current_total_rate = {'rate_percent': str(load)}
 873         rate_per_direction = self.get_per_direction_rate()
 874
 875         self.gen.modify_rate(rate_per_direction, False)
 876         self.run_config['rates'][0] = rate_per_direction
 877         if self.run_config['bidirectional']:
 878             self.gen.modify_rate(rate_per_direction, True)
 879             self.run_config['rates'][1] = rate_per_direction
 880
 881     def get_ndr_and_pdr(self):
 882         """Start the NDR/PDR iteration and return the results."""
 883         dst = 'Bidirectional' if self.run_config['bidirectional'] else 'Unidirectional'
 884         targets = {}
 885         if self.config.ndr_run:
 886             LOG.info('*** Searching NDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 887             targets['ndr'] = self.config.measurement.NDR
 888         if self.config.pdr_run:
 889             LOG.info('*** Searching PDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 890             targets['pdr'] = self.config.measurement.PDR
 891
 892         self.run_config['start_time'] = time.time()
 893         self.interval_collector = IntervalCollector(self.run_config['start_time'])
 894         self.interval_collector.attach_notifier(self.notifier)
 895         self.iteration_collector = IterationCollector(self.run_config['start_time'])
 896         results = {}
 897         self.__range_search(0.0, 200.0, targets, results)
 898
 899         results['iteration_stats'] = {
 900             'ndr_pdr': self.iteration_collector.get()
 901         }
 902
 903         if self.config.ndr_run:
 904             LOG.info('NDR load: %s', results['ndr']['rate_percent'])
 905             results['ndr']['time_taken_sec'] = \
 906                 results['ndr']['timestamp_sec'] - self.run_config['start_time']
 907             if self.config.pdr_run:
 908                 LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 909                 results['pdr']['time_taken_sec'] = \
 910                     results['pdr']['timestamp_sec'] - results['ndr']['timestamp_sec']
 911         else:
 912             LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 913             results['pdr']['time_taken_sec'] = \
 914                 results['pdr']['timestamp_sec'] - self.run_config['start_time']
 915         return results
 916
 917     def __get_dropped_rate(self, result):
 918         dropped_pkts = result['rx']['dropped_pkts']
 919         total_pkts = result['tx']['total_pkts']
 920         if not total_pkts:
 921             return float('inf')
 922         return float(dropped_pkts) / total_pkts * 100
 923
 924     def get_stats(self):
 925         """Collect final stats for previous run."""
 926         stats = self.gen.get_stats()
 927         retDict = {'total_tx_rate': stats['total_tx_rate']}
 928
 929         tx_keys = ['total_pkts', 'total_pkt_bytes', 'pkt_rate', 'pkt_bit_rate']
 930         rx_keys = tx_keys + ['dropped_pkts']
 931
 932         for port in self.PORTS:
 933             port_stats = {'tx': {}, 'rx': {}}
 934             for key in tx_keys:
 935                 port_stats['tx'][key] = int(stats[port]['tx'][key])
 936             for key in rx_keys:
 937                 try:
 938                     port_stats['rx'][key] = int(stats[port]['rx'][key])
 939                 except ValueError:
 940                     port_stats['rx'][key] = 0
 941             port_stats['rx']['avg_delay_usec'] = cast_integer(
 942                 stats[port]['rx']['avg_delay_usec'])
 943             port_stats['rx']['min_delay_usec'] = cast_integer(
 944                 stats[port]['rx']['min_delay_usec'])
 945             port_stats['rx']['max_delay_usec'] = cast_integer(
 946                 stats[port]['rx']['max_delay_usec'])
 947             port_stats['drop_rate_percent'] = self.__get_dropped_rate(port_stats)
 948             retDict[str(port)] = port_stats
 949
 950         ports = sorted(list(retDict.keys()), key=str)
 951         if self.run_config['bidirectional']:
 952             retDict['overall'] = {'tx': {}, 'rx': {}}
 953             for key in tx_keys:
 954                 retDict['overall']['tx'][key] = \
 955                     retDict[ports[0]]['tx'][key] + retDict[ports[1]]['tx'][key]
 956             for key in rx_keys:
 957                 retDict['overall']['rx'][key] = \
 958                     retDict[ports[0]]['rx'][key] + retDict[ports[1]]['rx'][key]
 959             total_pkts = [retDict[ports[0]]['rx']['total_pkts'],
 960                           retDict[ports[1]]['rx']['total_pkts']]
 961             avg_delays = [retDict[ports[0]]['rx']['avg_delay_usec'],
 962                           retDict[ports[1]]['rx']['avg_delay_usec']]
 963             max_delays = [retDict[ports[0]]['rx']['max_delay_usec'],
 964                           retDict[ports[1]]['rx']['max_delay_usec']]
 965             min_delays = [retDict[ports[0]]['rx']['min_delay_usec'],
 966                           retDict[ports[1]]['rx']['min_delay_usec']]
 967             retDict['overall']['rx']['avg_delay_usec'] = utils.weighted_avg(total_pkts, avg_delays)
 968             retDict['overall']['rx']['min_delay_usec'] = min(min_delays)
 969             retDict['overall']['rx']['max_delay_usec'] = max(max_delays)
 970             for key in ['pkt_bit_rate', 'pkt_rate']:
 971                 for dirc in ['tx', 'rx']:
 972                     retDict['overall'][dirc][key] /= 2.0
 973         else:
 974             retDict['overall'] = retDict[ports[0]]
 975         retDict['overall']['drop_rate_percent'] = self.__get_dropped_rate(retDict['overall'])
 976         return retDict
 977
 978     def __convert_rates(self, rate):
 979         return utils.convert_rates(self.run_config['l2frame_size'],
 980                                    rate,
 981                                    self.intf_speed)
 982
 983     def __ndr_pdr_found(self, tag, load):
 984         rates = self.__convert_rates({'rate_percent': load})
 985         self.iteration_collector.add_ndr_pdr(tag, rates['rate_pps'])
 986         last_stats = self.iteration_collector.peek()
 987         self.interval_collector.add_ndr_pdr(tag, last_stats)
 988
 989     def __format_output_stats(self, stats):
 990         for key in self.PORTS + ['overall']:
 991             key = str(key)
 992             interface = stats[key]
 993             stats[key] = {
 994                 'tx_pkts': interface['tx']['total_pkts'],
 995                 'rx_pkts': interface['rx']['total_pkts'],
 996                 'drop_percentage': interface['drop_rate_percent'],
 997                 'drop_pct': interface['rx']['dropped_pkts'],
 998                 'avg_delay_usec': interface['rx']['avg_delay_usec'],
 999                 'max_delay_usec': interface['rx']['max_delay_usec'],
1000                 'min_delay_usec': interface['rx']['min_delay_usec'],
1001             }
1002
1003         return stats
1004
1005     def __targets_found(self, rate, targets, results):
1006         for tag, target in list(targets.items()):
1007             LOG.info('Found %s (%s) load: %s', tag, target, rate)
1008             self.__ndr_pdr_found(tag, rate)
1009             results[tag]['timestamp_sec'] = time.time()
1010
1011     def __range_search(self, left, right, targets, results):
1012         """Perform a binary search for a list of targets inside a [left..right] range or rate.
1013
1014         left    the left side of the range to search as a % the line rate (100 = 100% line rate)
1015                 indicating the rate to send on each interface
1016         right   the right side of the range to search as a % of line rate
1017                 indicating the rate to send on each interface
1018         targets a dict of drop rates to search (0.1 = 0.1%), indexed by the DR name or "tag"
1019                 ('ndr', 'pdr')
1020         results a dict to store results
1021         """
1022         if not targets:
1023             return
1024         LOG.info('Range search [%s .. %s] targets: %s', left, right, targets)
1025
1026         # Terminate search when gap is less than load epsilon
1027         if right - left < self.config.measurement.load_epsilon:
1028             self.__targets_found(left, targets, results)
1029             return
1030
1031         # Obtain the average drop rate in for middle load
1032         middle = (left + right) / 2.0
1033         try:
1034             stats, rates = self.__run_search_iteration(middle)
1035         except STLError:
1036             LOG.exception("Got exception from traffic generator during binary search")
1037             self.__targets_found(left, targets, results)
1038             return
1039         # Split target dicts based on the avg drop rate
1040         left_targets = {}
1041         right_targets = {}
1042         for tag, target in list(targets.items()):
1043             if stats['overall']['drop_rate_percent'] <= target:
1044                 # record the best possible rate found for this target
1045                 results[tag] = rates
1046                 results[tag].update({
1047                     'load_percent_per_direction': middle,
1048                     'stats': self.__format_output_stats(dict(stats)),
1049                     'timestamp_sec': None
1050                 })
1051                 right_targets[tag] = target
1052             else:
1053                 # initialize to 0 all fields of result for
1054                 # the worst case scenario of the binary search (if ndr/pdr is not found)
1055                 if tag not in results:
1056                     results[tag] = dict.fromkeys(rates, 0)
1057                     empty_stats = self.__format_output_stats(dict(stats))
1058                     for key in empty_stats:
1059                         if isinstance(empty_stats[key], dict):
1060                             empty_stats[key] = dict.fromkeys(empty_stats[key], 0)
1061                         else:
1062                             empty_stats[key] = 0
1063                     results[tag].update({
1064                         'load_percent_per_direction': 0,
1065                         'stats': empty_stats,
1066                         'timestamp_sec': None
1067                     })
1068                 left_targets[tag] = target
1069
1070         # search lower half
1071         self.__range_search(left, middle, left_targets, results)
1072
1073         # search upper half only if the upper rate does not exceed
1074         # 100%, this only happens when the first search at 100%
1075         # yields a DR that is < target DR
1076         if middle >= 100:
1077             self.__targets_found(100, right_targets, results)
1078         else:
1079             self.__range_search(middle, right, right_targets, results)
1080
1081     def __run_search_iteration(self, rate):
1082         """Run one iteration at the given rate level.
1083
1084         rate: the rate to send on each port in percent (0 to 100)
1085         """
1086         self._modify_load(rate)
1087
1088         # poll interval stats and collect them
1089         for stats in self.run_traffic():
1090             self.interval_collector.add(stats)
1091             time_elapsed_ratio = self.runner.time_elapsed() / self.run_config['duration_sec']
1092             if time_elapsed_ratio >= 1:
1093                 self.cancel_traffic()
1094                 if not self.skip_sleep():
1095                     time.sleep(self.config.pause_sec)
1096         self.interval_collector.reset()
1097
1098         # get stats from the run
1099         stats = self.runner.client.get_stats()
1100         current_traffic_config = self._get_traffic_config()
1101         warning = self.compare_tx_rates(current_traffic_config['direction-total']['rate_pps'],
1102                                         stats['total_tx_rate'])
1103         if warning is not None:
1104             stats['warning'] = warning
1105
1106         # save reliable stats from whole iteration
1107         self.iteration_collector.add(stats, current_traffic_config['direction-total']['rate_pps'])
1108         LOG.info('Average drop rate: %f', stats['overall']['drop_rate_percent'])
1109         return stats, current_traffic_config['direction-total']
1110
1111     @staticmethod
1112     def log_stats(stats):
1113         """Log estimated stats during run."""
1114         report = {
1115             'datetime': str(datetime.now()),
1116             'tx_packets': stats['overall']['tx']['total_pkts'],
1117             'rx_packets': stats['overall']['rx']['total_pkts'],
1118             'drop_packets': stats['overall']['rx']['dropped_pkts'],
1119             'drop_rate_percent': stats['overall']['drop_rate_percent']
1120         }
1121         LOG.info('TX: %(tx_packets)d; '
1122                  'RX: %(rx_packets)d; '
1123                  'Est. Dropped: %(drop_packets)d; '
1124                  'Est. Drop rate: %(drop_rate_percent).4f%%',
1125                  report)
1126
1127     def run_traffic(self):
1128         """Start traffic and return intermediate stats for each interval."""
1129         stats = self.runner.run()
1130         while self.runner.is_running:
1131             self.log_stats(stats)
1132             yield stats
1133             stats = self.runner.poll_stats()
1134             if stats is None:
1135                 return
1136         self.log_stats(stats)
1137         LOG.info('Drop rate: %f', stats['overall']['drop_rate_percent'])
1138         yield stats
1139
1140     def cancel_traffic(self):
1141         """Stop traffic."""
1142         self.runner.stop()
1143
1144     def _get_traffic_config(self):
1145         config = {}
1146         load_total = 0.0
1147         bps_total = 0.0
1148         pps_total = 0.0
1149         for idx, rate in enumerate(self.run_config['rates']):
1150             key = 'direction-forward' if idx == 0 else 'direction-reverse'
1151             config[key] = {
1152                 'l2frame_size': self.run_config['l2frame_size'],
1153                 'duration_sec': self.run_config['duration_sec']
1154             }
1155             config[key].update(rate)
1156             config[key].update(self.__convert_rates(rate))
1157             load_total += float(config[key]['rate_percent'])
1158             bps_total += float(config[key]['rate_bps'])
1159             pps_total += float(config[key]['rate_pps'])
1160         config['direction-total'] = dict(config['direction-forward'])
1161         config['direction-total'].update({
1162             'rate_percent': load_total,
1163             'rate_pps': cast_integer(pps_total),
1164             'rate_bps': bps_total
1165         })
1166
1167         return config
1168
1169     def get_run_config(self, results):
1170         """Return configuration which was used for the last run."""
1171         r = {}
1172         # because we want each direction to have the far end RX rates,
1173         # use the far end index (1-idx) to retrieve the RX rates
1174         for idx, key in enumerate(["direction-forward", "direction-reverse"]):
1175             tx_rate = results["stats"][str(idx)]["tx"]["total_pkts"] / self.config.duration_sec
1176             rx_rate = results["stats"][str(1 - idx)]["rx"]["total_pkts"] / self.config.duration_sec
1177             r[key] = {
1178                 "orig": self.__convert_rates(self.run_config['rates'][idx]),
1179                 "tx": self.__convert_rates({'rate_pps': tx_rate}),
1180                 "rx": self.__convert_rates({'rate_pps': rx_rate})
1181             }
1182
1183         total = {}
1184         for direction in ['orig', 'tx', 'rx']:
1185             total[direction] = {}
1186             for unit in ['rate_percent', 'rate_bps', 'rate_pps']:
1187                 total[direction][unit] = sum([float(x[direction][unit]) for x in list(r.values())])
1188
1189         r['direction-total'] = total
1190         return r
1191
1192     def insert_interface_stats(self, pps_list):
1193         """Insert interface stats to a list of packet path stats.
1194
1195         pps_list: a list of packet path stats instances indexed by chain index
1196
1197         This function will insert the packet path stats for the traffic gen ports 0 and 1
1198         with itemized per chain tx/rx counters.
1199         There will be as many packet path stats as chains.
1200         Each packet path stats will have exactly 2 InterfaceStats for port 0 and port 1
1201         self.pps_list:
1202         [
1203         PacketPathStats(InterfaceStats(chain 0, port 0), InterfaceStats(chain 0, port 1)),
1204         PacketPathStats(InterfaceStats(chain 1, port 0), InterfaceStats(chain 1, port 1)),
1205         ...
1206         ]
1207         """
1208         def get_if_stats(chain_idx):
1209             return [InterfaceStats('p' + str(port), self.tool)
1210                     for port in range(2)]
1211         # keep the list of list of interface stats indexed by the chain id
1212         self.ifstats = [get_if_stats(chain_idx)
1213                         for chain_idx in range(self.config.service_chain_count)]
1214         # note that we need to make a copy of the ifs list so that any modification in the
1215         # list from pps will not change the list saved in self.ifstats
1216         self.pps_list = [PacketPathStats(list(ifs)) for ifs in self.ifstats]
1217         # insert the corresponding pps in the passed list
1218         pps_list.extend(self.pps_list)
1219
1220     def update_interface_stats(self, diff=False):
1221         """Update all interface stats.
1222
1223         diff: if False, simply refresh the interface stats values with latest values
1224               if True, diff the interface stats with the latest values
1225         Make sure that the interface stats inserted in insert_interface_stats() are updated
1226         with proper values.
1227         self.ifstats:
1228         [
1229         [InterfaceStats(chain 0, port 0), InterfaceStats(chain 0, port 1)],
1230         [InterfaceStats(chain 1, port 0), InterfaceStats(chain 1, port 1)],
1231         ...
1232         ]
1233         """
1234         if diff:
1235             stats = self.gen.get_stats()
1236             for chain_idx, ifs in enumerate(self.ifstats):
1237                 # each ifs has exactly 2 InterfaceStats and 2 Latency instances
1238                 # corresponding to the
1239                 # port 0 and port 1 for the given chain_idx
1240                 # Note that we cannot use self.pps_list[chain_idx].if_stats to pick the
1241                 # interface stats for the pps because it could have been modified to contain
1242                 # additional interface stats
1243                 self.gen.get_stream_stats(stats, ifs, self.pps_list[chain_idx].latencies, chain_idx)
1244             # special handling for vxlan
1245             # in case of vxlan, flow stats are not available so all rx counters will be
1246             # zeros when the total rx port counter is non zero.
1247             # in that case,
1248             for port in range(2):
1249                 total_rx = 0
1250                 for ifs in self.ifstats:
1251                     total_rx += ifs[port].rx
1252                 if total_rx == 0:
1253                     # check if the total port rx from Trex is also zero
1254                     port_rx = stats[port]['rx']['total_pkts']
1255                     if port_rx:
1256                         # the total rx for all chains from port level stats is non zero
1257                         # which means that the per-chain stats are not available
1258                         if len(self.ifstats) == 1:
1259                             # only one chain, simply report the port level rx to the chain rx stats
1260                             self.ifstats[0][port].rx = port_rx
1261                         else:
1262                             for ifs in self.ifstats:
1263                                 # mark this data as unavailable
1264                                 ifs[port].rx = None
1265                             # pitch in the total rx only in the last chain pps
1266                             self.ifstats[-1][port].rx_total = port_rx
1267
1268     @staticmethod
1269     def compare_tx_rates(required, actual):
1270         """Compare the actual TX rate to the required TX rate."""
1271         threshold = 0.9
1272         are_different = False
1273         try:
1274             if float(actual) / required < threshold:
1275                 are_different = True
1276         except ZeroDivisionError:
1277             are_different = True
1278
1279         if are_different:
1280             msg = "WARNING: There is a significant difference between requested TX rate ({r}) " \
1281                   "and actual TX rate ({a}). The traffic generator may not have sufficient CPU " \
1282                   "to achieve the requested TX rate.".format(r=required, a=actual)
1283             LOG.info(msg)
1284             return msg
1285
1286         return None
1287
1288     def get_per_direction_rate(self):
1289         """Get the rate for each direction."""
1290         divisor = 2 if self.run_config['bidirectional'] else 1
1291         if 'rate_percent' in self.current_total_rate:
1292             # don't split rate if it's percentage
1293             divisor = 1
1294
1295         return utils.divide_rate(self.current_total_rate, divisor)
1296
1297     def close(self):
1298         """Close this instance."""
1299         try:
1300             self.gen.stop_traffic()
1301         except Exception:
1302             pass
1303         self.gen.clear_stats()
1304         self.gen.cleanup()