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