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 from datetime import datetime
  16 import re
  17 import socket
  18 import struct
  19 import time
  20
  21 from attrdict import AttrDict
  22 import bitmath
  23 from netaddr import IPNetwork
  24 # pylint: disable=import-error
  25 from trex_stl_lib.api import STLError
  26 # pylint: enable=import-error
  27
  28 from log import LOG
  29 from network import Interface
  30 from specs import ChainType
  31 from stats_collector import IntervalCollector
  32 from stats_collector import IterationCollector
  33 import traffic_gen.traffic_utils as utils
  34 from utils import cast_integer
  35
  36
  37 class TrafficClientException(Exception):
  38     """Generic traffic client exception."""
  39
  40     pass
  41
  42
  43 class TrafficRunner(object):
  44     """Serialize various steps required to run traffic."""
  45
  46     def __init__(self, client, duration_sec, interval_sec=0):
  47         self.client = client
  48         self.start_time = None
  49         self.duration_sec = duration_sec
  50         self.interval_sec = interval_sec
  51
  52     def run(self):
  53         LOG.info('Running traffic generator')
  54         self.client.gen.clear_stats()
  55         self.client.gen.start_traffic()
  56         self.start_time = time.time()
  57         return self.poll_stats()
  58
  59     def stop(self):
  60         if self.is_running():
  61             self.start_time = None
  62             self.client.gen.stop_traffic()
  63
  64     def is_running(self):
  65         return self.start_time is not None
  66
  67     def time_elapsed(self):
  68         if self.is_running():
  69             return time.time() - self.start_time
  70         return self.duration_sec
  71
  72     def poll_stats(self):
  73         if not self.is_running():
  74             return None
  75         if self.client.skip_sleep:
  76             self.stop()
  77             return self.client.get_stats()
  78         time_elapsed = self.time_elapsed()
  79         if time_elapsed > self.duration_sec:
  80             self.stop()
  81             return None
  82         time_left = self.duration_sec - time_elapsed
  83         if self.interval_sec > 0.0:
  84             if time_left <= self.interval_sec:
  85                 time.sleep(time_left)
  86                 self.stop()
  87             else:
  88                 time.sleep(self.interval_sec)
  89         else:
  90             time.sleep(self.duration_sec)
  91             self.stop()
  92         return self.client.get_stats()
  93
  94
  95 class IpBlock(object):
  96     """Manage a block of IP addresses."""
  97
  98     def __init__(self, base_ip, step_ip, count_ip):
  99         self.base_ip_int = Device.ip_to_int(base_ip)
 100         self.step = Device.ip_to_int(step_ip)
 101         self.max_available = count_ip
 102         self.next_free = 0
 103
 104     def get_ip(self, index=0):
 105         """Return the IP address at given index."""
 106         if index < 0 or index >= self.max_available:
 107             raise IndexError('Index out of bounds')
 108         return Device.int_to_ip(self.base_ip_int + index * self.step)
 109
 110     def reserve_ip_range(self, count):
 111         """Reserve a range of count consecutive IP addresses spaced by step."""
 112         if self.next_free + count > self.max_available:
 113             raise IndexError('No more IP addresses next free=%d max_available=%d requested=%d' %
 114                              (self.next_free,
 115                               self.max_available,
 116                               count))
 117         first_ip = self.get_ip(self.next_free)
 118         last_ip = self.get_ip(self.next_free + count - 1)
 119         self.next_free += count
 120         return (first_ip, last_ip)
 121
 122     def reset_reservation(self):
 123         self.next_free = 0
 124
 125
 126 class Device(object):
 127     """Represent a port device and all information associated to it."""
 128
 129     def __init__(self, port, pci, switch_port=None, vtep_vlan=None, ip=None, tg_gateway_ip=None,
 130                  gateway_ip=None, ip_addrs_step=None, tg_gateway_ip_addrs_step=None,
 131                  gateway_ip_addrs_step=None, udp_src_port=None, udp_dst_port=None,
 132                  dst_mac=None, chain_count=1, flow_count=1, vlan_tagging=False):
 133         self.chain_count = chain_count
 134         self.flow_count = flow_count
 135         self.dst = None
 136         self.port = port
 137         self.switch_port = switch_port
 138         self.vtep_vlan = vtep_vlan
 139         self.vlan_tag = None
 140         self.vlan_tagging = vlan_tagging
 141         self.pci = pci
 142         self.mac = None
 143         self.dst_mac = dst_mac
 144         self.vm_mac_list = None
 145         subnet = IPNetwork(ip)
 146         self.ip = subnet.ip.format()
 147         self.ip_prefixlen = subnet.prefixlen
 148         self.ip_addrs_step = ip_addrs_step
 149         self.tg_gateway_ip_addrs_step = tg_gateway_ip_addrs_step
 150         self.gateway_ip_addrs_step = gateway_ip_addrs_step
 151         self.gateway_ip = gateway_ip
 152         self.tg_gateway_ip = tg_gateway_ip
 153         self.ip_block = IpBlock(self.ip, ip_addrs_step, flow_count)
 154         self.gw_ip_block = IpBlock(gateway_ip,
 155                                    gateway_ip_addrs_step,
 156                                    chain_count)
 157         self.tg_gw_ip_block = IpBlock(tg_gateway_ip,
 158                                       tg_gateway_ip_addrs_step,
 159                                       chain_count)
 160         self.udp_src_port = udp_src_port
 161         self.udp_dst_port = udp_dst_port
 162
 163     def set_mac(self, mac):
 164         if mac is None:
 165             raise TrafficClientException('Trying to set traffic generator MAC address as None')
 166         self.mac = mac
 167         LOG.info("Port %d: src MAC %s", self.port, self.mac)
 168
 169     def set_destination(self, dst):
 170         self.dst = dst
 171
 172     def set_vm_mac_list(self, vm_mac_list):
 173         self.vm_mac_list = map(str, vm_mac_list)
 174
 175     def set_vlan_tag(self, vlan_tag):
 176         if self.vlan_tagging and vlan_tag is None:
 177             raise TrafficClientException('Trying to set VLAN tag as None')
 178         self.vlan_tag = vlan_tag
 179         LOG.info("Port %d: VLAN %d", self.port, self.vlan_tag)
 180
 181     def get_gw_ip(self, chain_index):
 182         """Retrieve the IP address assigned for the gateway of a given chain."""
 183         return self.gw_ip_block.get_ip(chain_index)
 184
 185     def get_stream_configs(self, service_chain):
 186         configs = []
 187         # exact flow count for each chain is calculated as follows:
 188         # - all chains except the first will have the same flow count
 189         #   calculated as (total_flows + chain_count - 1) / chain_count
 190         # - the first chain will have the remainder
 191         # example 11 flows and 3 chains => 3, 4, 4
 192         flows_per_chain = (self.flow_count + self.chain_count - 1) / self.chain_count
 193         cur_chain_flow_count = self.flow_count - flows_per_chain * (self.chain_count - 1)
 194
 195         self.ip_block.reset_reservation()
 196         self.dst.ip_block.reset_reservation()
 197
 198         for chain_idx in xrange(self.chain_count):
 199             src_ip_first, src_ip_last = self.ip_block.reserve_ip_range(cur_chain_flow_count)
 200             dst_ip_first, dst_ip_last = self.dst.ip_block.reserve_ip_range(cur_chain_flow_count)
 201
 202             dst_mac = self.dst_mac[chain_idx] if self.dst_mac is not None else self.dst.mac
 203             if not re.match("[0-9a-f]{2}([-:])[0-9a-f]{2}(\\1[0-9a-f]{2}){4}$", dst_mac.lower()):
 204                 raise TrafficClientException("Invalid MAC address '{mac}' specified in "
 205                                              "mac_addrs_left/right".format(mac=dst_mac))
 206
 207             configs.append({
 208                 'count': cur_chain_flow_count,
 209                 'mac_src': self.mac,
 210                 'mac_dst': dst_mac if service_chain == ChainType.EXT else self.vm_mac_list[
 211                     chain_idx],
 212                 'ip_src_addr': src_ip_first,
 213                 'ip_src_addr_max': src_ip_last,
 214                 'ip_src_count': cur_chain_flow_count,
 215                 'ip_dst_addr': dst_ip_first,
 216                 'ip_dst_addr_max': dst_ip_last,
 217                 'ip_dst_count': cur_chain_flow_count,
 218                 'ip_addrs_step': self.ip_addrs_step,
 219                 'udp_src_port': self.udp_src_port,
 220                 'udp_dst_port': self.udp_dst_port,
 221                 'mac_discovery_gw': self.get_gw_ip(chain_idx),
 222                 'ip_src_tg_gw': self.tg_gw_ip_block.get_ip(chain_idx),
 223                 'ip_dst_tg_gw': self.dst.tg_gw_ip_block.get_ip(chain_idx),
 224                 'vlan_tag': self.vlan_tag if self.vlan_tagging else None
 225             })
 226             # after first chain, fall back to the flow count for all other chains
 227             cur_chain_flow_count = flows_per_chain
 228
 229         return configs
 230
 231     def ip_range_overlaps(self):
 232         """Check if this device ip range is overlapping with the dst device ip range."""
 233         src_base_ip = Device.ip_to_int(self.ip)
 234         dst_base_ip = Device.ip_to_int(self.dst.ip)
 235         src_last_ip = src_base_ip + self.flow_count - 1
 236         dst_last_ip = dst_base_ip + self.flow_count - 1
 237         return dst_last_ip >= src_base_ip and src_last_ip >= dst_base_ip
 238
 239     @staticmethod
 240     def mac_to_int(mac):
 241         return int(mac.translate(None, ":.- "), 16)
 242
 243     @staticmethod
 244     def int_to_mac(i):
 245         mac = format(i, 'x').zfill(12)
 246         blocks = [mac[x:x + 2] for x in xrange(0, len(mac), 2)]
 247         return ':'.join(blocks)
 248
 249     @staticmethod
 250     def ip_to_int(addr):
 251         return struct.unpack("!I", socket.inet_aton(addr))[0]
 252
 253     @staticmethod
 254     def int_to_ip(nvalue):
 255         return socket.inet_ntoa(struct.pack("!I", nvalue))
 256
 257
 258 class RunningTrafficProfile(object):
 259     """Represents traffic configuration for currently running traffic profile."""
 260
 261     DEFAULT_IP_STEP = '0.0.0.1'
 262     DEFAULT_SRC_DST_IP_STEP = '0.0.0.1'
 263
 264     def __init__(self, config, generator_profile):
 265         generator_config = self.__match_generator_profile(config.traffic_generator,
 266                                                           generator_profile)
 267         self.generator_config = generator_config
 268         self.service_chain = config.service_chain
 269         self.service_chain_count = config.service_chain_count
 270         self.flow_count = config.flow_count
 271         self.host_name = generator_config.host_name
 272         self.name = generator_config.name
 273         self.tool = generator_config.tool
 274         self.cores = generator_config.get('cores', 1)
 275         self.ip_addrs_step = generator_config.ip_addrs_step or self.DEFAULT_SRC_DST_IP_STEP
 276         self.tg_gateway_ip_addrs_step = \
 277             generator_config.tg_gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 278         self.gateway_ip_addrs_step = generator_config.gateway_ip_addrs_step or self.DEFAULT_IP_STEP
 279         self.gateway_ips = generator_config.gateway_ip_addrs
 280         self.ip = generator_config.ip
 281         self.intf_speed = bitmath.parse_string(generator_config.intf_speed.replace('ps', '')).bits
 282         self.vlan_tagging = config.vlan_tagging
 283         self.no_arp = config.no_arp
 284         self.src_device = None
 285         self.dst_device = None
 286         self.vm_mac_list = None
 287         self.mac_addrs_left = generator_config.mac_addrs_left
 288         self.mac_addrs_right = generator_config.mac_addrs_right
 289         self.__prep_interfaces(generator_config)
 290
 291     def to_json(self):
 292         return dict(self.generator_config)
 293
 294     def set_vm_mac_list(self, vm_mac_list):
 295         self.src_device.set_vm_mac_list(vm_mac_list[0])
 296         self.dst_device.set_vm_mac_list(vm_mac_list[1])
 297
 298     @staticmethod
 299     def __match_generator_profile(traffic_generator, generator_profile):
 300         generator_config = AttrDict(traffic_generator)
 301         generator_config.pop('default_profile')
 302         generator_config.pop('generator_profile')
 303         matching_profile = [profile for profile in traffic_generator.generator_profile if
 304                             profile.name == generator_profile]
 305         if len(matching_profile) != 1:
 306             raise Exception('Traffic generator profile not found: ' + generator_profile)
 307
 308         generator_config.update(matching_profile[0])
 309
 310         return generator_config
 311
 312     def __prep_interfaces(self, generator_config):
 313         src_config = {
 314             'chain_count': self.service_chain_count,
 315             'flow_count': self.flow_count / 2,
 316             'ip': generator_config.ip_addrs[0],
 317             'ip_addrs_step': self.ip_addrs_step,
 318             'gateway_ip': self.gateway_ips[0],
 319             'gateway_ip_addrs_step': self.gateway_ip_addrs_step,
 320             'tg_gateway_ip': generator_config.tg_gateway_ip_addrs[0],
 321             'tg_gateway_ip_addrs_step': self.tg_gateway_ip_addrs_step,
 322             'udp_src_port': generator_config.udp_src_port,
 323             'udp_dst_port': generator_config.udp_dst_port,
 324             'vlan_tagging': self.vlan_tagging,
 325             'dst_mac': generator_config.mac_addrs_left
 326         }
 327         dst_config = {
 328             'chain_count': self.service_chain_count,
 329             'flow_count': self.flow_count / 2,
 330             'ip': generator_config.ip_addrs[1],
 331             'ip_addrs_step': self.ip_addrs_step,
 332             'gateway_ip': self.gateway_ips[1],
 333             'gateway_ip_addrs_step': self.gateway_ip_addrs_step,
 334             'tg_gateway_ip': generator_config.tg_gateway_ip_addrs[1],
 335             'tg_gateway_ip_addrs_step': self.tg_gateway_ip_addrs_step,
 336             'udp_src_port': generator_config.udp_src_port,
 337             'udp_dst_port': generator_config.udp_dst_port,
 338             'vlan_tagging': self.vlan_tagging,
 339             'dst_mac': generator_config.mac_addrs_right
 340         }
 341
 342         self.src_device = Device(**dict(src_config, **generator_config.interfaces[0]))
 343         self.dst_device = Device(**dict(dst_config, **generator_config.interfaces[1]))
 344         self.src_device.set_destination(self.dst_device)
 345         self.dst_device.set_destination(self.src_device)
 346
 347         if self.service_chain == ChainType.EXT and not self.no_arp \
 348                 and self.src_device.ip_range_overlaps():
 349             raise Exception('Overlapping IP address ranges src=%s dst=%d flows=%d' %
 350                             self.src_device.ip,
 351                             self.dst_device.ip,
 352                             self.flow_count)
 353
 354     @property
 355     def devices(self):
 356         return [self.src_device, self.dst_device]
 357
 358     @property
 359     def vlans(self):
 360         return [self.src_device.vtep_vlan, self.dst_device.vtep_vlan]
 361
 362     @property
 363     def ports(self):
 364         return [self.src_device.port, self.dst_device.port]
 365
 366     @property
 367     def switch_ports(self):
 368         return [self.src_device.switch_port, self.dst_device.switch_port]
 369
 370     @property
 371     def pcis(self):
 372         return [self.src_device.pci, self.dst_device.pci]
 373
 374
 375 class TrafficGeneratorFactory(object):
 376     """Factory class to generate a traffic generator."""
 377
 378     def __init__(self, config):
 379         self.config = config
 380
 381     def get_tool(self):
 382         return self.config.generator_config.tool
 383
 384     def get_generator_client(self):
 385         tool = self.get_tool().lower()
 386         if tool == 'trex':
 387             from traffic_gen import trex
 388             return trex.TRex(self.config)
 389         elif tool == 'dummy':
 390             from traffic_gen import dummy
 391             return dummy.DummyTG(self.config)
 392         return None
 393
 394     def list_generator_profile(self):
 395         return [profile.name for profile in self.config.traffic_generator.generator_profile]
 396
 397     def get_generator_config(self, generator_profile):
 398         return RunningTrafficProfile(self.config, generator_profile)
 399
 400     def get_matching_profile(self, traffic_profile_name):
 401         matching_profile = [profile for profile in self.config.traffic_profile if
 402                             profile.name == traffic_profile_name]
 403
 404         if len(matching_profile) > 1:
 405             raise Exception('Multiple traffic profiles with the same name found.')
 406         elif not matching_profile:
 407             raise Exception('No traffic profile found.')
 408
 409         return matching_profile[0]
 410
 411     def get_frame_sizes(self, traffic_profile):
 412         matching_profile = self.get_matching_profile(traffic_profile)
 413         return matching_profile.l2frame_size
 414
 415
 416 class TrafficClient(object):
 417     """Traffic generator client."""
 418
 419     PORTS = [0, 1]
 420
 421     def __init__(self, config, notifier=None, skip_sleep=False):
 422         generator_factory = TrafficGeneratorFactory(config)
 423         self.gen = generator_factory.get_generator_client()
 424         self.tool = generator_factory.get_tool()
 425         self.config = config
 426         self.notifier = notifier
 427         self.interval_collector = None
 428         self.iteration_collector = None
 429         self.runner = TrafficRunner(self, self.config.duration_sec, self.config.interval_sec)
 430         if self.gen is None:
 431             raise TrafficClientException('%s is not a supported traffic generator' % self.tool)
 432
 433         self.run_config = {
 434             'l2frame_size': None,
 435             'duration_sec': self.config.duration_sec,
 436             'bidirectional': True,
 437             'rates': []  # to avoid unsbuscriptable-obj warning
 438         }
 439         self.current_total_rate = {'rate_percent': '10'}
 440         if self.config.single_run:
 441             self.current_total_rate = utils.parse_rate_str(self.config.rate)
 442         # UT with dummy TG can bypass all sleeps
 443         self.skip_sleep = skip_sleep
 444
 445     def set_macs(self):
 446         for mac, device in zip(self.gen.get_macs(), self.config.generator_config.devices):
 447             device.set_mac(mac)
 448
 449     def start_traffic_generator(self):
 450         self.gen.init()
 451         self.gen.connect()
 452
 453     def setup(self):
 454         self.gen.set_mode()
 455         self.gen.config_interface()
 456         self.gen.clear_stats()
 457
 458     def get_version(self):
 459         return self.gen.get_version()
 460
 461     def ensure_end_to_end(self):
 462         """Ensure traffic generator receives packets it has transmitted.
 463
 464         This ensures end to end connectivity and also waits until VMs are ready to forward packets.
 465
 466         VMs that are started and in active state may not pass traffic yet. It is imperative to make
 467         sure that all VMs are passing traffic in both directions before starting any benchmarking.
 468         To verify this, we need to send at a low frequency bi-directional packets and make sure
 469         that we receive all packets back from all VMs. The number of flows is equal to 2 times
 470         the number of chains (1 per direction) and we need to make sure we receive packets coming
 471         from exactly 2 x chain count different source MAC addresses.
 472
 473         Example:
 474             PVP chain (1 VM per chain)
 475             N = 10 (number of chains)
 476             Flow count = 20 (number of flows)
 477             If the number of unique source MAC addresses from received packets is 20 then
 478             all 10 VMs 10 VMs are in operational state.
 479         """
 480         LOG.info('Starting traffic generator to ensure end-to-end connectivity')
 481         rate_pps = {'rate_pps': str(self.config.service_chain_count * 1)}
 482         self.gen.create_traffic('64', [rate_pps, rate_pps], bidirectional=True, latency=False)
 483
 484         # ensures enough traffic is coming back
 485         retry_count = (self.config.check_traffic_time_sec +
 486                        self.config.generic_poll_sec - 1) / self.config.generic_poll_sec
 487         mac_addresses = set()
 488         ln = 0
 489         for it in xrange(retry_count):
 490             self.gen.clear_stats()
 491             self.gen.start_traffic()
 492             self.gen.start_capture()
 493             LOG.info('Waiting for packets to be received back... (%d / %d)', it + 1, retry_count)
 494             if not self.skip_sleep:
 495                 time.sleep(self.config.generic_poll_sec)
 496             self.gen.stop_traffic()
 497             self.gen.fetch_capture_packets()
 498             self.gen.stop_capture()
 499
 500             for packet in self.gen.packet_list:
 501                 mac_addresses.add(packet['binary'][6:12])
 502                 if ln != len(mac_addresses):
 503                     ln = len(mac_addresses)
 504                     LOG.info('Flows passing traffic %d / %d', ln,
 505                              self.config.service_chain_count * 2)
 506                 if len(mac_addresses) == self.config.service_chain_count * 2:
 507                     LOG.info('End-to-end connectivity ensured')
 508                     return
 509
 510             if not self.skip_sleep:
 511                 time.sleep(self.config.generic_poll_sec)
 512
 513         raise TrafficClientException('End-to-end connectivity cannot be ensured')
 514
 515     def ensure_arp_successful(self):
 516         if not self.gen.resolve_arp():
 517             raise TrafficClientException('ARP cannot be resolved')
 518
 519     def set_traffic(self, frame_size, bidirectional):
 520         self.run_config['bidirectional'] = bidirectional
 521         self.run_config['l2frame_size'] = frame_size
 522         self.run_config['rates'] = [self.get_per_direction_rate()]
 523         if bidirectional:
 524             self.run_config['rates'].append(self.get_per_direction_rate())
 525         else:
 526             unidir_reverse_pps = int(self.config.unidir_reverse_traffic_pps)
 527             if unidir_reverse_pps > 0:
 528                 self.run_config['rates'].append({'rate_pps': str(unidir_reverse_pps)})
 529         # Fix for [NFVBENCH-67], convert the rate string to PPS
 530         for idx, rate in enumerate(self.run_config['rates']):
 531             if 'rate_pps' not in rate:
 532                 self.run_config['rates'][idx] = {'rate_pps': self.__convert_rates(rate)['rate_pps']}
 533
 534         self.gen.clear_streamblock()
 535         self.gen.create_traffic(frame_size, self.run_config['rates'], bidirectional, latency=True)
 536
 537     def modify_load(self, load):
 538         self.current_total_rate = {'rate_percent': str(load)}
 539         rate_per_direction = self.get_per_direction_rate()
 540
 541         self.gen.modify_rate(rate_per_direction, False)
 542         self.run_config['rates'][0] = rate_per_direction
 543         if self.run_config['bidirectional']:
 544             self.gen.modify_rate(rate_per_direction, True)
 545             self.run_config['rates'][1] = rate_per_direction
 546
 547     def get_ndr_and_pdr(self):
 548         dst = 'Bidirectional' if self.run_config['bidirectional'] else 'Unidirectional'
 549         targets = {}
 550         if self.config.ndr_run:
 551             LOG.info('*** Searching NDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 552             targets['ndr'] = self.config.measurement.NDR
 553         if self.config.pdr_run:
 554             LOG.info('*** Searching PDR for %s (%s)...', self.run_config['l2frame_size'], dst)
 555             targets['pdr'] = self.config.measurement.PDR
 556
 557         self.run_config['start_time'] = time.time()
 558         self.interval_collector = IntervalCollector(self.run_config['start_time'])
 559         self.interval_collector.attach_notifier(self.notifier)
 560         self.iteration_collector = IterationCollector(self.run_config['start_time'])
 561         results = {}
 562         self.__range_search(0.0, 200.0, targets, results)
 563
 564         results['iteration_stats'] = {
 565             'ndr_pdr': self.iteration_collector.get()
 566         }
 567
 568         if self.config.ndr_run:
 569             LOG.info('NDR load: %s', results['ndr']['rate_percent'])
 570             results['ndr']['time_taken_sec'] = \
 571                 results['ndr']['timestamp_sec'] - self.run_config['start_time']
 572             if self.config.pdr_run:
 573                 LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 574                 results['pdr']['time_taken_sec'] = \
 575                     results['pdr']['timestamp_sec'] - results['ndr']['timestamp_sec']
 576         else:
 577             LOG.info('PDR load: %s', results['pdr']['rate_percent'])
 578             results['pdr']['time_taken_sec'] = \
 579                 results['pdr']['timestamp_sec'] - self.run_config['start_time']
 580         return results
 581
 582     def __get_dropped_rate(self, result):
 583         dropped_pkts = result['rx']['dropped_pkts']
 584         total_pkts = result['tx']['total_pkts']
 585         if not total_pkts:
 586             return float('inf')
 587         return float(dropped_pkts) / total_pkts * 100
 588
 589     def get_stats(self):
 590         stats = self.gen.get_stats()
 591         retDict = {'total_tx_rate': stats['total_tx_rate']}
 592         for port in self.PORTS:
 593             retDict[port] = {'tx': {}, 'rx': {}}
 594
 595         tx_keys = ['total_pkts', 'total_pkt_bytes', 'pkt_rate', 'pkt_bit_rate']
 596         rx_keys = tx_keys + ['dropped_pkts']
 597
 598         for port in self.PORTS:
 599             for key in tx_keys:
 600                 retDict[port]['tx'][key] = int(stats[port]['tx'][key])
 601             for key in rx_keys:
 602                 try:
 603                     retDict[port]['rx'][key] = int(stats[port]['rx'][key])
 604                 except ValueError:
 605                     retDict[port]['rx'][key] = 0
 606             retDict[port]['rx']['avg_delay_usec'] = cast_integer(
 607                 stats[port]['rx']['avg_delay_usec'])
 608             retDict[port]['rx']['min_delay_usec'] = cast_integer(
 609                 stats[port]['rx']['min_delay_usec'])
 610             retDict[port]['rx']['max_delay_usec'] = cast_integer(
 611                 stats[port]['rx']['max_delay_usec'])
 612             retDict[port]['drop_rate_percent'] = self.__get_dropped_rate(retDict[port])
 613
 614         ports = sorted(retDict.keys())
 615         if self.run_config['bidirectional']:
 616             retDict['overall'] = {'tx': {}, 'rx': {}}
 617             for key in tx_keys:
 618                 retDict['overall']['tx'][key] = \
 619                     retDict[ports[0]]['tx'][key] + retDict[ports[1]]['tx'][key]
 620             for key in rx_keys:
 621                 retDict['overall']['rx'][key] = \
 622                     retDict[ports[0]]['rx'][key] + retDict[ports[1]]['rx'][key]
 623             total_pkts = [retDict[ports[0]]['rx']['total_pkts'],
 624                           retDict[ports[1]]['rx']['total_pkts']]
 625             avg_delays = [retDict[ports[0]]['rx']['avg_delay_usec'],
 626                           retDict[ports[1]]['rx']['avg_delay_usec']]
 627             max_delays = [retDict[ports[0]]['rx']['max_delay_usec'],
 628                           retDict[ports[1]]['rx']['max_delay_usec']]
 629             min_delays = [retDict[ports[0]]['rx']['min_delay_usec'],
 630                           retDict[ports[1]]['rx']['min_delay_usec']]
 631             retDict['overall']['rx']['avg_delay_usec'] = utils.weighted_avg(total_pkts, avg_delays)
 632             retDict['overall']['rx']['min_delay_usec'] = min(min_delays)
 633             retDict['overall']['rx']['max_delay_usec'] = max(max_delays)
 634             for key in ['pkt_bit_rate', 'pkt_rate']:
 635                 for dirc in ['tx', 'rx']:
 636                     retDict['overall'][dirc][key] /= 2.0
 637         else:
 638             retDict['overall'] = retDict[ports[0]]
 639         retDict['overall']['drop_rate_percent'] = self.__get_dropped_rate(retDict['overall'])
 640         return retDict
 641
 642     def __convert_rates(self, rate):
 643         return utils.convert_rates(self.run_config['l2frame_size'],
 644                                    rate,
 645                                    self.config.generator_config.intf_speed)
 646
 647     def __ndr_pdr_found(self, tag, load):
 648         rates = self.__convert_rates({'rate_percent': load})
 649         self.iteration_collector.add_ndr_pdr(tag, rates['rate_pps'])
 650         last_stats = self.iteration_collector.peek()
 651         self.interval_collector.add_ndr_pdr(tag, last_stats)
 652
 653     def __format_output_stats(self, stats):
 654         for key in self.PORTS + ['overall']:
 655             interface = stats[key]
 656             stats[key] = {
 657                 'tx_pkts': interface['tx']['total_pkts'],
 658                 'rx_pkts': interface['rx']['total_pkts'],
 659                 'drop_percentage': interface['drop_rate_percent'],
 660                 'drop_pct': interface['rx']['dropped_pkts'],
 661                 'avg_delay_usec': interface['rx']['avg_delay_usec'],
 662                 'max_delay_usec': interface['rx']['max_delay_usec'],
 663                 'min_delay_usec': interface['rx']['min_delay_usec'],
 664             }
 665
 666         return stats
 667
 668     def __targets_found(self, rate, targets, results):
 669         for tag, target in targets.iteritems():
 670             LOG.info('Found %s (%s) load: %s', tag, target, rate)
 671             self.__ndr_pdr_found(tag, rate)
 672             results[tag]['timestamp_sec'] = time.time()
 673
 674     def __range_search(self, left, right, targets, results):
 675         """Perform a binary search for a list of targets inside a [left..right] range or rate.
 676
 677         left    the left side of the range to search as a % the line rate (100 = 100% line rate)
 678                 indicating the rate to send on each interface
 679         right   the right side of the range to search as a % of line rate
 680                 indicating the rate to send on each interface
 681         targets a dict of drop rates to search (0.1 = 0.1%), indexed by the DR name or "tag"
 682                 ('ndr', 'pdr')
 683         results a dict to store results
 684         """
 685         if not targets:
 686             return
 687         LOG.info('Range search [%s .. %s] targets: %s', left, right, targets)
 688
 689         # Terminate search when gap is less than load epsilon
 690         if right - left < self.config.measurement.load_epsilon:
 691             self.__targets_found(left, targets, results)
 692             return
 693
 694         # Obtain the average drop rate in for middle load
 695         middle = (left + right) / 2.0
 696         try:
 697             stats, rates = self.__run_search_iteration(middle)
 698         except STLError:
 699             LOG.exception("Got exception from traffic generator during binary search")
 700             self.__targets_found(left, targets, results)
 701             return
 702         # Split target dicts based on the avg drop rate
 703         left_targets = {}
 704         right_targets = {}
 705         for tag, target in targets.iteritems():
 706             if stats['overall']['drop_rate_percent'] <= target:
 707                 # record the best possible rate found for this target
 708                 results[tag] = rates
 709                 results[tag].update({
 710                     'load_percent_per_direction': middle,
 711                     'stats': self.__format_output_stats(dict(stats)),
 712                     'timestamp_sec': None
 713                 })
 714                 right_targets[tag] = target
 715             else:
 716                 # initialize to 0 all fields of result for
 717                 # the worst case scenario of the binary search (if ndr/pdr is not found)
 718                 if tag not in results:
 719                     results[tag] = dict.fromkeys(rates, 0)
 720                     empty_stats = self.__format_output_stats(dict(stats))
 721                     for key in empty_stats:
 722                         if isinstance(empty_stats[key], dict):
 723                             empty_stats[key] = dict.fromkeys(empty_stats[key], 0)
 724                         else:
 725                             empty_stats[key] = 0
 726                     results[tag].update({
 727                         'load_percent_per_direction': 0,
 728                         'stats': empty_stats,
 729                         'timestamp_sec': None
 730                     })
 731                 left_targets[tag] = target
 732
 733         # search lower half
 734         self.__range_search(left, middle, left_targets, results)
 735
 736         # search upper half only if the upper rate does not exceed
 737         # 100%, this only happens when the first search at 100%
 738         # yields a DR that is < target DR
 739         if middle >= 100:
 740             self.__targets_found(100, right_targets, results)
 741         else:
 742             self.__range_search(middle, right, right_targets, results)
 743
 744     def __run_search_iteration(self, rate):
 745         # set load
 746         self.modify_load(rate)
 747
 748         # poll interval stats and collect them
 749         for stats in self.run_traffic():
 750             self.interval_collector.add(stats)
 751             time_elapsed_ratio = self.runner.time_elapsed() / self.run_config['duration_sec']
 752             if time_elapsed_ratio >= 1:
 753                 self.cancel_traffic()
 754         self.interval_collector.reset()
 755
 756         # get stats from the run
 757         stats = self.runner.client.get_stats()
 758         current_traffic_config = self.get_traffic_config()
 759         warning = self.compare_tx_rates(current_traffic_config['direction-total']['rate_pps'],
 760                                         stats['total_tx_rate'])
 761         if warning is not None:
 762             stats['warning'] = warning
 763
 764         # save reliable stats from whole iteration
 765         self.iteration_collector.add(stats, current_traffic_config['direction-total']['rate_pps'])
 766         LOG.info('Average drop rate: %f', stats['overall']['drop_rate_percent'])
 767
 768         return stats, current_traffic_config['direction-total']
 769
 770     @staticmethod
 771     def log_stats(stats):
 772         report = {
 773             'datetime': str(datetime.now()),
 774             'tx_packets': stats['overall']['tx']['total_pkts'],
 775             'rx_packets': stats['overall']['rx']['total_pkts'],
 776             'drop_packets': stats['overall']['rx']['dropped_pkts'],
 777             'drop_rate_percent': stats['overall']['drop_rate_percent']
 778         }
 779         LOG.info('TX: %(tx_packets)d; '
 780                  'RX: %(rx_packets)d; '
 781                  'Dropped: %(drop_packets)d; '
 782                  'Drop rate: %(drop_rate_percent).4f%%',
 783                  report)
 784
 785     def run_traffic(self):
 786         stats = self.runner.run()
 787         while self.runner.is_running:
 788             self.log_stats(stats)
 789             yield stats
 790             stats = self.runner.poll_stats()
 791             if stats is None:
 792                 return
 793         self.log_stats(stats)
 794         LOG.info('Drop rate: %f', stats['overall']['drop_rate_percent'])
 795         yield stats
 796
 797     def cancel_traffic(self):
 798         self.runner.stop()
 799
 800     def get_interface(self, port_index, stats):
 801         port = self.gen.port_handle[port_index]
 802         tx, rx = 0, 0
 803         if stats and port in stats:
 804             tx, rx = int(stats[port]['tx']['total_pkts']), int(stats[port]['rx']['total_pkts'])
 805         return Interface('traffic-generator', self.tool.lower(), tx, rx)
 806
 807     def get_traffic_config(self):
 808         config = {}
 809         load_total = 0.0
 810         bps_total = 0.0
 811         pps_total = 0.0
 812         for idx, rate in enumerate(self.run_config['rates']):
 813             key = 'direction-forward' if idx == 0 else 'direction-reverse'
 814             config[key] = {
 815                 'l2frame_size': self.run_config['l2frame_size'],
 816                 'duration_sec': self.run_config['duration_sec']
 817             }
 818             config[key].update(rate)
 819             config[key].update(self.__convert_rates(rate))
 820             load_total += float(config[key]['rate_percent'])
 821             bps_total += float(config[key]['rate_bps'])
 822             pps_total += float(config[key]['rate_pps'])
 823         config['direction-total'] = dict(config['direction-forward'])
 824         config['direction-total'].update({
 825             'rate_percent': load_total,
 826             'rate_pps': cast_integer(pps_total),
 827             'rate_bps': bps_total
 828         })
 829
 830         return config
 831
 832     def get_run_config(self, results):
 833         """Return configuration which was used for the last run."""
 834         r = {}
 835         # because we want each direction to have the far end RX rates,
 836         # use the far end index (1-idx) to retrieve the RX rates
 837         for idx, key in enumerate(["direction-forward", "direction-reverse"]):
 838             tx_rate = results["stats"][idx]["tx"]["total_pkts"] / self.config.duration_sec
 839             rx_rate = results["stats"][1 - idx]["rx"]["total_pkts"] / self.config.duration_sec
 840             r[key] = {
 841                 "orig": self.__convert_rates(self.run_config['rates'][idx]),
 842                 "tx": self.__convert_rates({'rate_pps': tx_rate}),
 843                 "rx": self.__convert_rates({'rate_pps': rx_rate})
 844             }
 845
 846         total = {}
 847         for direction in ['orig', 'tx', 'rx']:
 848             total[direction] = {}
 849             for unit in ['rate_percent', 'rate_bps', 'rate_pps']:
 850                 total[direction][unit] = sum([float(x[direction][unit]) for x in r.values()])
 851
 852         r['direction-total'] = total
 853         return r
 854
 855     @staticmethod
 856     def compare_tx_rates(required, actual):
 857         threshold = 0.9
 858         are_different = False
 859         try:
 860             if float(actual) / required < threshold:
 861                 are_different = True
 862         except ZeroDivisionError:
 863             are_different = True
 864
 865         if are_different:
 866             msg = "WARNING: There is a significant difference between requested TX rate ({r}) " \
 867                   "and actual TX rate ({a}). The traffic generator may not have sufficient CPU " \
 868                   "to achieve the requested TX rate.".format(r=required, a=actual)
 869             LOG.info(msg)
 870             return msg
 871
 872         return None
 873
 874     def get_per_direction_rate(self):
 875         divisor = 2 if self.run_config['bidirectional'] else 1
 876         if 'rate_percent' in self.current_total_rate:
 877             # don't split rate if it's percentage
 878             divisor = 1
 879
 880         return utils.divide_rate(self.current_total_rate, divisor)
 881
 882     def close(self):
 883         try:
 884             self.gen.stop_traffic()
 885         except Exception:
 886             pass
 887         self.gen.clear_stats()
 888         self.gen.cleanup()