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