# under the License.
"""Interface to the traffic generator clients including NDR/PDR binary search."""
-
-from datetime import datetime
import socket
import struct
import time
+import sys
from attrdict import AttrDict
import bitmath
+from hdrh.histogram import HdrHistogram
from netaddr import IPNetwork
# pylint: disable=import-error
from trex.stl.api import Ether
from trex.stl.api import STLError
from trex.stl.api import UDP
+# pylint: disable=wrong-import-order
+from scapy.contrib.mpls import MPLS # flake8: noqa
+# pylint: enable=wrong-import-order
# pylint: enable=import-error
-from log import LOG
-from packet_stats import InterfaceStats
-from packet_stats import PacketPathStats
-from stats_collector import IntervalCollector
-from stats_collector import IterationCollector
-import traffic_gen.traffic_utils as utils
-from utils import cast_integer
-
+from .log import LOG
+from .packet_stats import InterfaceStats
+from .packet_stats import PacketPathStats
+from .stats_collector import IntervalCollector
+from .stats_collector import IterationCollector
+from .traffic_gen import traffic_utils as utils
+from .utils import cast_integer, find_max_size, find_tuples_equal_to_lcm_value, get_divisors, lcm
class TrafficClientException(Exception):
"""Generic traffic client exception."""
- pass
-
-
class TrafficRunner(object):
"""Serialize various steps required to run traffic."""
- def __init__(self, client, duration_sec, interval_sec=0):
+ def __init__(self, client, duration_sec, interval_sec=0, service_mode=False):
"""Create a traffic runner."""
self.client = client
self.start_time = None
self.duration_sec = duration_sec
self.interval_sec = interval_sec
+ self.service_mode = service_mode
def run(self):
"""Clear stats and instruct the traffic generator to start generating traffic."""
return None
LOG.info('Running traffic generator')
self.client.gen.clear_stats()
+ # Debug use only: the service_mode flag may have been set in
+ # the configuration, in order to enable the 'service' mode
+ # in the trex generator, before starting the traffic (run).
+ # From this point, a T-rex console (launched in readonly mode) would
+ # then be able to capture the transmitted and/or received traffic.
+ self.client.gen.set_service_mode(enabled=self.service_mode)
+ LOG.info('Service mode is %sabled', 'en' if self.service_mode else 'dis')
self.client.gen.start_traffic()
self.start_time = time.time()
return self.poll_stats()
def __init__(self, base_ip, step_ip, count_ip):
"""Create an IP block."""
self.base_ip_int = Device.ip_to_int(base_ip)
+ if step_ip == 'random':
+ step_ip = '0.0.0.1'
self.step = Device.ip_to_int(step_ip)
self.max_available = count_ip
self.next_free = 0
raise IndexError('Index out of bounds: %d (max=%d)' % (index, self.max_available))
return Device.int_to_ip(self.base_ip_int + index * self.step)
+ def get_ip_from_chain_first_ip(self, first_ip, index=0):
+ """Return the IP address at given index starting from chain first ip."""
+ if index < 0 or index >= self.max_available:
+ raise IndexError('Index out of bounds: %d (max=%d)' % (index, self.max_available))
+ return Device.int_to_ip(first_ip + index * self.step)
+
def reserve_ip_range(self, count):
- """Reserve a range of count consecutive IP addresses spaced by step."""
+ """Reserve a range of count consecutive IP addresses spaced by step.
+ """
if self.next_free + count > self.max_available:
raise IndexError('No more IP addresses next free=%d max_available=%d requested=%d' %
(self.next_free,
self.next_free = 0
+class UdpPorts(object):
+
+ def __init__(self, src_min, src_max, dst_min, dst_max, udp_src_size, udp_dst_size, step):
+
+ self.src_min = int(src_min)
+ self.src_max = int(src_max)
+ self.dst_min = int(dst_min)
+ self.dst_max = int(dst_max)
+ self.udp_src_size = udp_src_size
+ self.udp_dst_size = udp_dst_size
+ self.step = step
+
+ def get_src_max(self, index=0):
+ """Return the UDP src port at given index."""
+ return int(self.src_min) + index * int(self.step)
+
+ def get_dst_max(self, index=0):
+ """Return the UDP dst port at given index."""
+ return int(self.dst_min) + index * int(self.step)
+
+
class Device(object):
"""Represent a port device and all information associated to it.
"""Create a new device for a given port."""
self.generator_config = generator_config
self.chain_count = generator_config.service_chain_count
- self.flow_count = generator_config.flow_count / 2
+ if generator_config.bidirectional:
+ self.flow_count = generator_config.flow_count / 2
+ else:
+ self.flow_count = generator_config.flow_count
+
self.port = port
self.switch_port = generator_config.interfaces[port].get('switch_port', None)
self.vtep_vlan = None
self.vtep_src_mac = None
self.vxlan = False
+ self.mpls = False
+ self.inner_labels = None
+ self.outer_labels = None
self.pci = generator_config.interfaces[port].pci
self.mac = None
self.dest_macs = None
self.vnis = None
self.vlans = None
self.ip_addrs = generator_config.ip_addrs[port]
- subnet = IPNetwork(self.ip_addrs)
- self.ip = subnet.ip.format()
+ self.ip_src_static = generator_config.ip_src_static
self.ip_addrs_step = generator_config.ip_addrs_step
- self.ip_block = IpBlock(self.ip, self.ip_addrs_step, self.flow_count)
+ if self.ip_addrs_step == 'random':
+ # Set step to 1 to calculate the IP range size (see check_range_size below)
+ step = '0.0.0.1'
+ else:
+ step = self.ip_addrs_step
+ self.ip_size = self.check_range_size(IPNetwork(self.ip_addrs).size, Device.ip_to_int(step))
+ self.ip = str(IPNetwork(self.ip_addrs).network)
+ ip_addrs_left = generator_config.ip_addrs[0]
+ ip_addrs_right = generator_config.ip_addrs[1]
+ self.ip_addrs_size = {
+ 'left': self.check_range_size(IPNetwork(ip_addrs_left).size, Device.ip_to_int(step)),
+ 'right': self.check_range_size(IPNetwork(ip_addrs_right).size, Device.ip_to_int(step))}
+ udp_src_port = generator_config.gen_config.udp_src_port
+ if udp_src_port is None:
+ udp_src_port = 53
+ udp_dst_port = generator_config.gen_config.udp_dst_port
+ if udp_dst_port is None:
+ udp_dst_port = 53
+ src_max, src_min = self.define_udp_range(udp_src_port, 'udp_src_port')
+ dst_max, dst_min = self.define_udp_range(udp_dst_port, 'udp_dst_port')
+ if generator_config.gen_config.udp_port_step == 'random':
+ # Set step to 1 to calculate the UDP range size
+ udp_step = 1
+ else:
+ udp_step = int(generator_config.gen_config.udp_port_step)
+ udp_src_size = self.check_range_size(int(src_max) - int(src_min) + 1, udp_step)
+ udp_dst_size = self.check_range_size(int(dst_max) - int(dst_min) + 1, udp_step)
+ lcm_port = lcm(udp_src_size, udp_dst_size)
+ if self.ip_src_static is True:
+ lcm_ip = lcm(1, min(self.ip_addrs_size['left'], self.ip_addrs_size['right']))
+ else:
+ lcm_ip = lcm(self.ip_addrs_size['left'], self.ip_addrs_size['right'])
+ flow_max = lcm(lcm_port, lcm_ip)
+ if self.flow_count > flow_max:
+ raise TrafficClientException('Trying to set unachievable traffic (%d > %d)' %
+ (self.flow_count, flow_max))
+
+ self.udp_ports = UdpPorts(src_min, src_max, dst_min, dst_max, udp_src_size, udp_dst_size,
+ generator_config.gen_config.udp_port_step)
+
+ self.ip_block = IpBlock(self.ip, step, self.ip_size)
+
self.gw_ip_block = IpBlock(generator_config.gateway_ips[port],
generator_config.gateway_ip_addrs_step,
self.chain_count)
self.tg_gw_ip_block = IpBlock(self.tg_gateway_ip_addrs,
generator_config.tg_gateway_ip_addrs_step,
self.chain_count)
- self.udp_src_port = generator_config.udp_src_port
- self.udp_dst_port = generator_config.udp_dst_port
+
+ def limit_ip_udp_ranges(self, peer_ip_size, cur_chain_flow_count):
+ # init to min value in case of no matching values found with lcm calculation
+ new_src_ip_size = 1
+ new_peer_ip_size = 1
+ new_src_udp_size = 1
+ new_dst_udp_size = 1
+
+ if self.ip_src_static is True:
+ src_ip_size = 1
+ else:
+ src_ip_size = self.ip_size
+ ip_src_divisors = list(get_divisors(src_ip_size))
+ ip_dst_divisors = list(get_divisors(peer_ip_size))
+ udp_src_divisors = list(get_divisors(self.udp_ports.udp_src_size))
+ udp_dst_divisors = list(get_divisors(self.udp_ports.udp_dst_size))
+ fc = int(cur_chain_flow_count)
+ tuples_ip = list(find_tuples_equal_to_lcm_value(ip_src_divisors, ip_dst_divisors, fc))
+ tuples_udp = list(find_tuples_equal_to_lcm_value(udp_src_divisors, udp_dst_divisors, fc))
+
+ if tuples_ip:
+ new_src_ip_size = tuples_ip[-1][0]
+ new_peer_ip_size = tuples_ip[-1][1]
+
+ if tuples_udp:
+ new_src_udp_size = tuples_udp[-1][0]
+ new_dst_udp_size = tuples_udp[-1][1]
+
+ tuples_src = []
+ tuples_dst = []
+ if not tuples_ip and not tuples_udp:
+ # in case of not divisors in common matching LCM value (i.e. requested flow count)
+ # try to find an accurate UDP range to fit requested flow count
+ udp_src_int = range(self.udp_ports.src_min, self.udp_ports.src_max)
+ udp_dst_int = range(self.udp_ports.dst_min, self.udp_ports.dst_max)
+ tuples_src = list(find_tuples_equal_to_lcm_value(ip_src_divisors, udp_src_int, fc))
+ tuples_dst = list(find_tuples_equal_to_lcm_value(ip_dst_divisors, udp_dst_int, fc))
+
+ if not tuples_src and not tuples_dst:
+ # iterate IP and UDP ranges to find a tuple that match flow count values
+ src_ip_range = range(1,src_ip_size)
+ dst_ip_range = range(1, peer_ip_size)
+ tuples_src = list(find_tuples_equal_to_lcm_value(src_ip_range, udp_src_int, fc))
+ tuples_dst = list(find_tuples_equal_to_lcm_value(dst_ip_range, udp_dst_int, fc))
+
+ if tuples_src or tuples_dst:
+ if tuples_src:
+ new_src_ip_size = tuples_src[-1][0]
+ new_src_udp_size = tuples_src[-1][1]
+ if tuples_dst:
+ new_peer_ip_size = tuples_dst[-1][0]
+ new_dst_udp_size = tuples_dst[-1][1]
+ else:
+ if not tuples_ip:
+ if src_ip_size != 1:
+ if src_ip_size > fc:
+ new_src_ip_size = fc
+ else:
+ new_src_ip_size = find_max_size(src_ip_size, tuples_udp, fc)
+ if peer_ip_size != 1:
+ if peer_ip_size > fc:
+ new_peer_ip_size = fc
+ else:
+ new_peer_ip_size = find_max_size(peer_ip_size, tuples_udp, fc)
+
+ if not tuples_udp:
+ if self.udp_ports.udp_src_size != 1:
+ if self.udp_ports.udp_src_size > fc:
+ new_src_udp_size = fc
+ else:
+ new_src_udp_size = find_max_size(self.udp_ports.udp_src_size,
+ tuples_ip, fc)
+ if self.udp_ports.udp_dst_size != 1:
+ if self.udp_ports.udp_dst_size > fc:
+ new_dst_udp_size = fc
+ else:
+ new_dst_udp_size = find_max_size(self.udp_ports.udp_dst_size,
+ tuples_ip, fc)
+ max_possible_flows = lcm(lcm(new_src_ip_size, new_peer_ip_size),
+ lcm(new_src_udp_size, new_dst_udp_size))
+
+ LOG.debug("IP dst size: %d", new_peer_ip_size)
+ LOG.debug("LCM IP: %d", lcm(new_src_ip_size, new_peer_ip_size))
+ LOG.debug("LCM UDP: %d", lcm(new_src_udp_size, new_dst_udp_size))
+ LOG.debug("Global LCM: %d", max_possible_flows)
+ LOG.debug("IP src size: %d, IP dst size: %d, UDP src size: %d, UDP dst size: %d",
+ new_src_ip_size, new_peer_ip_size, self.udp_ports.udp_src_size,
+ self.udp_ports.udp_dst_size)
+ if not max_possible_flows == cur_chain_flow_count:
+ if (self.ip_addrs_step != '0.0.0.1' or self.udp_ports.step != '1') and not (
+ self.ip_addrs_step == 'random' and self.udp_ports.step == 'random'):
+ LOG.warning("Current values of ip_addrs_step and/or udp_port_step properties "
+ "do not allow to control an accurate flow count. "
+ "Values will be overridden as follows:")
+ if self.ip_addrs_step != '0.0.0.1':
+ LOG.info("ip_addrs_step='0.0.0.1' (previous value: ip_addrs_step='%s')",
+ self.ip_addrs_step)
+ self.ip_addrs_step = '0.0.0.1'
+
+ if self.udp_ports.step != '1':
+ LOG.info("udp_port_step='1' (previous value: udp_port_step='%s')",
+ self.udp_ports.step)
+ self.udp_ports.step = '1'
+ # override config for not logging random step warning message in trex_gen.py
+ self.generator_config.gen_config.udp_port_step = self.udp_ports.step
+ else:
+ LOG.error("Current values of ip_addrs_step and udp_port_step properties "
+ "do not allow to control an accurate flow count.")
+ else:
+ src_ip_size = new_src_ip_size
+ peer_ip_size = new_peer_ip_size
+ self.udp_ports.udp_src_size = new_src_udp_size
+ self.udp_ports.udp_dst_size = new_dst_udp_size
+ return src_ip_size, peer_ip_size
+
+ @staticmethod
+ def define_udp_range(udp_port, property_name):
+ if isinstance(udp_port, int):
+ min = udp_port
+ max = min
+ elif isinstance(udp_port, tuple):
+ min = udp_port[0]
+ max = udp_port[1]
+ else:
+ raise TrafficClientException('Invalid %s property value (53 or [\'53\',\'1024\'])'
+ % property_name)
+ return max, min
+
+
+ @staticmethod
+ def check_range_size(range_size, step):
+ """Check and set the available IPs or UDP ports, considering the step."""
+ try:
+ if range_size % step == 0:
+ value = range_size // step
+ else:
+ value = range_size // step + 1
+ return value
+ except ZeroDivisionError:
+ raise ZeroDivisionError("step can't be zero !") from ZeroDivisionError
def set_mac(self, mac):
"""Set the local MAC for this port device."""
- VM macs discovered using openstack API
- dest MACs provisioned in config file
"""
- self.vtep_dst_mac = map(str, dest_macs)
+ self.vtep_dst_mac = list(map(str, dest_macs))
def set_dest_macs(self, dest_macs):
"""Set the list of dest MACs indexed by the chain id.
- VM macs discovered using openstack API
- dest MACs provisioned in config file
"""
- self.dest_macs = map(str, dest_macs)
+ self.dest_macs = list(map(str, dest_macs))
def get_dest_macs(self):
"""Get the list of dest macs for this device.
LOG.info("Port %d: src_vtep %s, dst_vtep %s", self.port,
self.vtep_src_ip, self.vtep_dst_ip)
+ def set_mpls_peers(self, src_ip, dst_ip):
+ self.mpls = True
+ self.vtep_dst_ip = dst_ip
+ self.vtep_src_ip = src_ip
+ LOG.info("Port %d: src_mpls_vtep %s, mpls_peer_ip %s", self.port,
+ self.vtep_src_ip, self.vtep_dst_ip)
+
def set_vxlans(self, vnis):
self.vnis = vnis
LOG.info("Port %d: VNIs %s", self.port, self.vnis)
+ def set_mpls_inner_labels(self, labels):
+ self.inner_labels = labels
+ LOG.info("Port %d: MPLS Inner Labels %s", self.port, self.inner_labels)
+
+ def set_mpls_outer_labels(self, labels):
+ self.outer_labels = labels
+ LOG.info("Port %d: MPLS Outer Labels %s", self.port, self.outer_labels)
+
def set_gw_ip(self, gateway_ip):
self.gw_ip_block = IpBlock(gateway_ip,
self.generator_config.gateway_ip_addrs_step,
# calculated as (total_flows + chain_count - 1) / chain_count
# - the first chain will have the remainder
# example 11 flows and 3 chains => 3, 4, 4
- flows_per_chain = (self.flow_count + self.chain_count - 1) / self.chain_count
- cur_chain_flow_count = self.flow_count - flows_per_chain * (self.chain_count - 1)
+ flows_per_chain = int((self.flow_count + self.chain_count - 1) / self.chain_count)
+ cur_chain_flow_count = int(self.flow_count - flows_per_chain * (self.chain_count - 1))
+
peer = self.get_peer_device()
self.ip_block.reset_reservation()
peer.ip_block.reset_reservation()
dest_macs = self.get_dest_macs()
- for chain_idx in xrange(self.chain_count):
- src_ip_first, src_ip_last = self.ip_block.reserve_ip_range(cur_chain_flow_count)
- dst_ip_first, dst_ip_last = peer.ip_block.reserve_ip_range(cur_chain_flow_count)
+ # limit ranges of UDP ports and IP to avoid overflow of the number of flows
+ peer_size = peer.ip_size // self.chain_count
+
+ for chain_idx in range(self.chain_count):
+ src_ip_size, peer_ip_size = self.limit_ip_udp_ranges(peer_size, cur_chain_flow_count)
+
+ src_ip_first, src_ip_last = self.ip_block.reserve_ip_range \
+ (src_ip_size)
+ dst_ip_first, dst_ip_last = peer.ip_block.reserve_ip_range \
+ (peer_ip_size)
+
+ if self.ip_addrs_step != 'random':
+ src_ip_last = self.ip_block.get_ip_from_chain_first_ip(
+ Device.ip_to_int(src_ip_first), src_ip_size - 1)
+ dst_ip_last = peer.ip_block.get_ip_from_chain_first_ip(
+ Device.ip_to_int(dst_ip_first), peer_ip_size - 1)
+ if self.udp_ports.step != 'random':
+ self.udp_ports.src_max = self.udp_ports.get_src_max(self.udp_ports.udp_src_size - 1)
+ self.udp_ports.dst_max = self.udp_ports.get_dst_max(self.udp_ports.udp_dst_size - 1)
+ if self.ip_src_static:
+ src_ip_last = src_ip_first
+
+ LOG.info("Port %d, chain %d: IP src range [%s,%s]", self.port, chain_idx,
+ src_ip_first, src_ip_last)
+ LOG.info("Port %d, chain %d: IP dst range [%s,%s]", self.port, chain_idx,
+ dst_ip_first, dst_ip_last)
+ LOG.info("Port %d, chain %d: UDP src range [%s,%s]", self.port, chain_idx,
+ self.udp_ports.src_min, self.udp_ports.src_max)
+ LOG.info("Port %d, chain %d: UDP dst range [%s,%s]", self.port, chain_idx,
+ self.udp_ports.dst_min, self.udp_ports.dst_max)
configs.append({
'count': cur_chain_flow_count,
'mac_dst': dest_macs[chain_idx],
'ip_src_addr': src_ip_first,
'ip_src_addr_max': src_ip_last,
- 'ip_src_count': cur_chain_flow_count,
+ 'ip_src_count': src_ip_size,
'ip_dst_addr': dst_ip_first,
'ip_dst_addr_max': dst_ip_last,
- 'ip_dst_count': cur_chain_flow_count,
+ 'ip_dst_count': peer_ip_size,
'ip_addrs_step': self.ip_addrs_step,
- 'udp_src_port': self.udp_src_port,
- 'udp_dst_port': self.udp_dst_port,
+ 'ip_src_static': self.ip_src_static,
+ 'udp_src_port': self.udp_ports.src_min,
+ 'udp_src_port_max': self.udp_ports.src_max,
+ 'udp_src_count': self.udp_ports.udp_src_size,
+ 'udp_dst_port': self.udp_ports.dst_min,
+ 'udp_dst_port_max': self.udp_ports.dst_max,
+ 'udp_dst_count': self.udp_ports.udp_dst_size,
+ 'udp_port_step': self.udp_ports.step,
'mac_discovery_gw': self.get_gw_ip(chain_idx),
'ip_src_tg_gw': self.tg_gw_ip_block.get_ip(chain_idx),
'ip_dst_tg_gw': peer.tg_gw_ip_block.get_ip(chain_idx),
'vlan_tag': self.vlans[chain_idx] if self.vlans else None,
'vxlan': self.vxlan,
'vtep_vlan': self.vtep_vlan if self.vtep_vlan else None,
- 'vtep_src_mac': self.mac if self.vxlan is True else None,
- 'vtep_dst_mac': self.vtep_dst_mac if self.vxlan is True else None,
+ 'vtep_src_mac': self.mac if (self.vxlan or self.mpls) else None,
+ 'vtep_dst_mac': self.vtep_dst_mac if (self.vxlan or self.mpls) else None,
'vtep_dst_ip': self.vtep_dst_ip if self.vxlan is True else None,
'vtep_src_ip': self.vtep_src_ip if self.vxlan is True else None,
- 'net_vni': self.vnis[chain_idx] if self.vxlan is True else None
+ 'net_vni': self.vnis[chain_idx] if self.vxlan is True else None,
+ 'mpls': self.mpls,
+ 'mpls_outer_label': self.outer_labels[chain_idx] if self.mpls is True else None,
+ 'mpls_inner_label': self.inner_labels[chain_idx] if self.mpls is True else None
+
})
# after first chain, fall back to the flow count for all other chains
cur_chain_flow_count = flows_per_chain
@staticmethod
def int_to_ip(nvalue):
"""Convert an IP address from numeric to string."""
- return socket.inet_ntoa(struct.pack("!I", nvalue))
+ return socket.inet_ntoa(struct.pack("!I", int(nvalue)))
class GeneratorConfig(object):
self.cores = config.cores
else:
self.cores = gen_config.get('cores', 1)
+ # let's report the value actually used in the end
+ config.cores_used = self.cores
self.mbuf_factor = config.mbuf_factor
+ self.mbuf_64 = config.mbuf_64
self.hdrh = not config.disable_hdrh
- if gen_config.intf_speed:
- # interface speed is overriden from config
- self.intf_speed = bitmath.parse_string(gen_config.intf_speed.replace('ps', '')).bits
+ if config.intf_speed:
+ # interface speed is overriden from the command line
+ self.intf_speed = config.intf_speed
+ elif gen_config.intf_speed:
+ # interface speed is overriden from the generator config
+ self.intf_speed = gen_config.intf_speed
else:
+ self.intf_speed = "auto"
+ if self.intf_speed in ("auto", "0"):
# interface speed is discovered/provided by the traffic generator
self.intf_speed = 0
+ else:
+ self.intf_speed = bitmath.parse_string(self.intf_speed.replace('ps', '')).bits
self.name = gen_config.name
self.zmq_pub_port = gen_config.get('zmq_pub_port', 4500)
self.zmq_rpc_port = gen_config.get('zmq_rpc_port', 4501)
self.interfaces = gen_config.interfaces
if self.interfaces[0].port != 0 or self.interfaces[1].port != 1:
raise TrafficClientException('Invalid port order/id in generator_profile.interfaces')
- if hasattr(gen_config, 'platform'):
- self.platform = gen_config.platform
self.service_chain = config.service_chain
self.service_chain_count = config.service_chain_count
self.flow_count = config.flow_count
self.host_name = gen_config.host_name
-
+ self.bidirectional = config.traffic.bidirectional
self.tg_gateway_ip_addrs = gen_config.tg_gateway_ip_addrs
self.ip_addrs = gen_config.ip_addrs
self.ip_addrs_step = gen_config.ip_addrs_step or self.DEFAULT_SRC_DST_IP_STEP
gen_config.tg_gateway_ip_addrs_step or self.DEFAULT_IP_STEP
self.gateway_ip_addrs_step = gen_config.gateway_ip_addrs_step or self.DEFAULT_IP_STEP
self.gateway_ips = gen_config.gateway_ip_addrs
- self.udp_src_port = gen_config.udp_src_port
- self.udp_dst_port = gen_config.udp_dst_port
+ self.ip_src_static = gen_config.ip_src_static
self.vteps = gen_config.get('vteps')
self.devices = [Device(port, self) for port in [0, 1]]
# This should normally always be [0, 1]
raise TrafficClientException('Dest MAC list %s must have %d entries' %
(dest_macs, self.config.service_chain_count))
self.devices[port_index].set_vtep_dst_mac(dest_macs)
- LOG.info('Port %d: vtep dst MAC %s', port_index, set([str(mac) for mac in dest_macs]))
+ LOG.info('Port %d: vtep dst MAC %s', port_index, {str(mac) for mac in dest_macs})
def get_dest_macs(self):
"""Return the list of dest macs indexed by port."""
(vxlans, self.config.service_chain_count))
self.devices[port_index].set_vxlans(vxlans)
+ def set_mpls_inner_labels(self, port_index, labels):
+ """Set the list of MPLS Labels to use indexed by the chain id on given port.
+
+ port_index: the port for which Labels must be set
+ Labels: a list of Labels lists indexed by chain id
+ """
+ if len(labels) != self.config.service_chain_count:
+ raise TrafficClientException('Inner MPLS list %s must have %d entries' %
+ (labels, self.config.service_chain_count))
+ self.devices[port_index].set_mpls_inner_labels(labels)
+
+ def set_mpls_outer_labels(self, port_index, labels):
+ """Set the list of MPLS Labels to use indexed by the chain id on given port.
+
+ port_index: the port for which Labels must be set
+ Labels: a list of Labels lists indexed by chain id
+ """
+ if len(labels) != self.config.service_chain_count:
+ raise TrafficClientException('Outer MPLS list %s must have %d entries' %
+ (labels, self.config.service_chain_count))
+ self.devices[port_index].set_mpls_outer_labels(labels)
+
def set_vtep_vlan(self, port_index, vlan):
"""Set the vtep vlan to use indexed by the chain id on given port.
port_index: the port for which VLAN must be set
def set_vxlan_endpoints(self, port_index, src_ip, dst_ip):
self.devices[port_index].set_vxlan_endpoints(src_ip, dst_ip)
+ def set_mpls_peers(self, port_index, src_ip, dst_ip):
+ self.devices[port_index].set_mpls_peers(src_ip, dst_ip)
+
@staticmethod
def __match_generator_profile(traffic_generator, generator_profile):
gen_config = AttrDict(traffic_generator)
self.notifier = notifier
self.interval_collector = None
self.iteration_collector = None
- self.runner = TrafficRunner(self, self.config.duration_sec, self.config.interval_sec)
+ self.runner = TrafficRunner(self, self.config.duration_sec, self.config.interval_sec,
+ self.config.service_mode)
self.config.frame_sizes = self._get_frame_sizes()
self.run_config = {
'l2frame_size': None,
def _get_generator(self):
tool = self.tool.lower()
if tool == 'trex':
- from traffic_gen import trex_gen
+ from .traffic_gen import trex_gen
return trex_gen.TRex(self)
if tool == 'dummy':
- from traffic_gen import dummy
+ from .traffic_gen import dummy
return dummy.DummyTG(self)
raise TrafficClientException('Unsupported generator tool name:' + self.tool)
if len(matching_profiles) > 1:
raise TrafficClientException('Multiple traffic profiles with name: ' +
traffic_profile_name)
- elif not matching_profiles:
+ if not matching_profiles:
raise TrafficClientException('Cannot find traffic profile: ' + traffic_profile_name)
return matching_profiles[0].l2frame_size
# interface speed is overriden from config
if self.intf_speed != tg_if_speed:
# Warn the user if the speed in the config is different
- LOG.warning('Interface speed provided is different from actual speed (%d Gbps)',
- intf_speeds[0])
+ LOG.warning(
+ 'Interface speed provided (%g Gbps) is different from actual speed (%d Gbps)',
+ self.intf_speed / 1000000000.0, intf_speeds[0])
else:
# interface speed not provisioned by config
self.intf_speed = tg_if_speed
# also update the speed in the tg config
self.generator_config.intf_speed = tg_if_speed
+ # let's report detected and actually used interface speed
+ self.config.intf_speed_detected = tg_if_speed
+ self.config.intf_speed_used = self.intf_speed
# Save the traffic generator local MAC
for mac, device in zip(self.gen.get_macs(), self.generator_config.devices):
self.gen.create_traffic('64', [rate_pps, rate_pps], bidirectional=True, latency=False,
e2e=True)
# ensures enough traffic is coming back
- retry_count = (self.config.check_traffic_time_sec +
- self.config.generic_poll_sec - 1) / self.config.generic_poll_sec
+ retry_count = int((self.config.check_traffic_time_sec +
+ self.config.generic_poll_sec - 1) / self.config.generic_poll_sec)
# we expect to see packets coming from 2 unique MAC per chain
# because there can be flooding in the case of shared net
get_mac_id = lambda packet: packet['binary'][60:66]
elif self.config.vxlan:
get_mac_id = lambda packet: packet['binary'][56:62]
+ elif self.config.mpls:
+ get_mac_id = lambda packet: packet['binary'][24:30]
+ # mpls_transport_label = lambda packet: packet['binary'][14:18]
else:
get_mac_id = lambda packet: packet['binary'][6:12]
- for it in xrange(retry_count):
+ for it in range(retry_count):
self.gen.clear_stats()
self.gen.start_traffic()
self.gen.start_capture()
self.gen.fetch_capture_packets()
self.gen.stop_capture()
for packet in self.gen.packet_list:
- mac_id = get_mac_id(packet)
+ mac_id = get_mac_id(packet).decode('latin-1')
src_mac = ':'.join(["%02x" % ord(x) for x in mac_id])
- if src_mac in mac_map and self.is_udp(packet):
- port, chain = mac_map[src_mac]
- LOG.info('Received packet from mac: %s (chain=%d, port=%d)',
- src_mac, chain, port)
- mac_map.pop(src_mac, None)
+ if self.config.mpls:
+ if src_mac in mac_map and self.is_mpls(packet):
+ port, chain = mac_map[src_mac]
+ LOG.info('Received mpls packet from mac: %s (chain=%d, port=%d)',
+ src_mac, chain, port)
+ mac_map.pop(src_mac, None)
+ else:
+ if src_mac in mac_map and self.is_udp(packet):
+ port, chain = mac_map[src_mac]
+ LOG.info('Received udp packet from mac: %s (chain=%d, port=%d)',
+ src_mac, chain, port)
+ mac_map.pop(src_mac, None)
if not mac_map:
LOG.info('End-to-end connectivity established')
pkt = Ether(packet['binary'])
return UDP in pkt
+ def is_mpls(self, packet):
+ pkt = Ether(packet['binary'])
+ return MPLS in pkt
+
def ensure_arp_successful(self):
"""Resolve all IP using ARP and throw an exception in case of failure."""
dest_macs = self.gen.resolve_arp()
if dest_macs:
# all dest macs are discovered, saved them into the generator config
- if self.config.vxlan:
+ if self.config.vxlan or self.config.mpls:
self.generator_config.set_vtep_dest_macs(0, dest_macs[0])
self.generator_config.set_vtep_dest_macs(1, dest_macs[1])
else:
self.run_config['rates'][idx] = {'rate_pps': self.__convert_rates(rate)['rate_pps']}
self.gen.clear_streamblock()
- self.gen.create_traffic(frame_size, self.run_config['rates'], bidirectional, latency=True)
+
+ if self.config.no_latency_streams:
+ LOG.info("Latency streams are disabled")
+ # in service mode, we must disable flow stats (e2e=True)
+ self.gen.create_traffic(frame_size, self.run_config['rates'], bidirectional,
+ latency=not self.config.no_latency_streams,
+ e2e=self.runner.service_mode)
def _modify_load(self, load):
self.current_total_rate = {'rate_percent': str(load)}
def get_stats(self):
"""Collect final stats for previous run."""
- stats = self.gen.get_stats()
- retDict = {'total_tx_rate': stats['total_tx_rate']}
- for port in self.PORTS:
- retDict[port] = {'tx': {}, 'rx': {}}
+ stats = self.gen.get_stats(self.ifstats)
+ retDict = {'total_tx_rate': stats['total_tx_rate'],
+ 'offered_tx_rate_bps': stats['offered_tx_rate_bps'],
+ 'theoretical_tx_rate_bps': stats['theoretical_tx_rate_bps'],
+ 'theoretical_tx_rate_pps': stats['theoretical_tx_rate_pps']}
+
+ if self.config.periodic_gratuitous_arp:
+ retDict['garp_total_tx_rate'] = stats['garp_total_tx_rate']
tx_keys = ['total_pkts', 'total_pkt_bytes', 'pkt_rate', 'pkt_bit_rate']
rx_keys = tx_keys + ['dropped_pkts']
for port in self.PORTS:
+ port_stats = {'tx': {}, 'rx': {}}
for key in tx_keys:
- retDict[port]['tx'][key] = int(stats[port]['tx'][key])
+ port_stats['tx'][key] = int(stats[port]['tx'][key])
for key in rx_keys:
try:
- retDict[port]['rx'][key] = int(stats[port]['rx'][key])
+ port_stats['rx'][key] = int(stats[port]['rx'][key])
except ValueError:
- retDict[port]['rx'][key] = 0
- retDict[port]['rx']['avg_delay_usec'] = cast_integer(
+ port_stats['rx'][key] = 0
+ port_stats['rx']['avg_delay_usec'] = cast_integer(
stats[port]['rx']['avg_delay_usec'])
- retDict[port]['rx']['min_delay_usec'] = cast_integer(
+ port_stats['rx']['min_delay_usec'] = cast_integer(
stats[port]['rx']['min_delay_usec'])
- retDict[port]['rx']['max_delay_usec'] = cast_integer(
+ port_stats['rx']['max_delay_usec'] = cast_integer(
stats[port]['rx']['max_delay_usec'])
- retDict[port]['drop_rate_percent'] = self.__get_dropped_rate(retDict[port])
+ port_stats['drop_rate_percent'] = self.__get_dropped_rate(port_stats)
+ retDict[str(port)] = port_stats
- ports = sorted(retDict.keys())
+ ports = sorted(list(retDict.keys()), key=str)
if self.run_config['bidirectional']:
retDict['overall'] = {'tx': {}, 'rx': {}}
for key in tx_keys:
else:
retDict['overall'] = retDict[ports[0]]
retDict['overall']['drop_rate_percent'] = self.__get_dropped_rate(retDict['overall'])
+
+ if 'overall_hdrh' in stats:
+ retDict['overall']['hdrh'] = stats.get('overall_hdrh', None)
+ decoded_histogram = HdrHistogram.decode(retDict['overall']['hdrh'])
+ retDict['overall']['rx']['lat_percentile'] = {}
+ # override min max and avg from hdrh (only if histogram is valid)
+ if decoded_histogram.get_total_count() != 0:
+ retDict['overall']['rx']['min_delay_usec'] = decoded_histogram.get_min_value()
+ retDict['overall']['rx']['max_delay_usec'] = decoded_histogram.get_max_value()
+ retDict['overall']['rx']['avg_delay_usec'] = decoded_histogram.get_mean_value()
+ for percentile in self.config.lat_percentiles:
+ retDict['overall']['rx']['lat_percentile'][percentile] = \
+ decoded_histogram.get_value_at_percentile(percentile)
+ else:
+ for percentile in self.config.lat_percentiles:
+ retDict['overall']['rx']['lat_percentile'][percentile] = 'n/a'
return retDict
def __convert_rates(self, rate):
def __format_output_stats(self, stats):
for key in self.PORTS + ['overall']:
+ key = str(key)
interface = stats[key]
stats[key] = {
'tx_pkts': interface['tx']['total_pkts'],
'min_delay_usec': interface['rx']['min_delay_usec'],
}
+ if key == 'overall':
+ if 'hdrh' in interface:
+ stats[key]['hdrh'] = interface.get('hdrh', None)
+ decoded_histogram = HdrHistogram.decode(stats[key]['hdrh'])
+ stats[key]['lat_percentile'] = {}
+ # override min max and avg from hdrh (only if histogram is valid)
+ if decoded_histogram.get_total_count() != 0:
+ stats[key]['min_delay_usec'] = decoded_histogram.get_min_value()
+ stats[key]['max_delay_usec'] = decoded_histogram.get_max_value()
+ stats[key]['avg_delay_usec'] = decoded_histogram.get_mean_value()
+ for percentile in self.config.lat_percentiles:
+ stats[key]['lat_percentile'][percentile] = decoded_histogram.\
+ get_value_at_percentile(percentile)
+ else:
+ for percentile in self.config.lat_percentiles:
+ stats[key]['lat_percentile'][percentile] = 'n/a'
return stats
def __targets_found(self, rate, targets, results):
- for tag, target in targets.iteritems():
+ for tag, target in list(targets.items()):
LOG.info('Found %s (%s) load: %s', tag, target, rate)
self.__ndr_pdr_found(tag, rate)
results[tag]['timestamp_sec'] = time.time()
# Split target dicts based on the avg drop rate
left_targets = {}
right_targets = {}
- for tag, target in targets.iteritems():
+ for tag, target in list(targets.items()):
if stats['overall']['drop_rate_percent'] <= target:
# record the best possible rate found for this target
results[tag] = rates
"""
self._modify_load(rate)
+ # There used to be a inconsistency in case of interface speed override.
+ # The emulated 'intf_speed' value is unknown to the T-Rex generator which
+ # refers to the detected line rate for converting relative traffic loads.
+ # Therefore, we need to convert actual rates here, in terms of packets/s.
+
+ for idx, str_rate in enumerate(self.gen.rates):
+ if str_rate.endswith('%'):
+ float_rate = float(str_rate.replace('%', '').strip())
+ pps_rate = self.__convert_rates({'rate_percent': float_rate})['rate_pps']
+ self.gen.rates[idx] = str(pps_rate) + 'pps'
+
# poll interval stats and collect them
for stats in self.run_traffic():
self.interval_collector.add(stats)
LOG.info('Average drop rate: %f', stats['overall']['drop_rate_percent'])
return stats, current_traffic_config['direction-total']
- @staticmethod
- def log_stats(stats):
+ def log_stats(self, stats):
"""Log estimated stats during run."""
- report = {
- 'datetime': str(datetime.now()),
- 'tx_packets': stats['overall']['tx']['total_pkts'],
- 'rx_packets': stats['overall']['rx']['total_pkts'],
- 'drop_packets': stats['overall']['rx']['dropped_pkts'],
- 'drop_rate_percent': stats['overall']['drop_rate_percent']
- }
- LOG.info('TX: %(tx_packets)d; '
- 'RX: %(rx_packets)d; '
- 'Est. Dropped: %(drop_packets)d; '
- 'Est. Drop rate: %(drop_rate_percent).4f%%',
- report)
+ # Calculate a rolling drop rate based on differential to
+ # the previous reading
+ cur_tx = stats['overall']['tx']['total_pkts']
+ cur_rx = stats['overall']['rx']['total_pkts']
+ delta_tx = cur_tx - self.prev_tx
+ delta_rx = cur_rx - self.prev_rx
+ drops = delta_tx - delta_rx
+ if delta_tx == 0:
+ LOG.info("\x1b[1mConfiguration issue!\x1b[0m (no transmission)")
+ sys.exit(0)
+ drop_rate_pct = 100 * (delta_tx - delta_rx)/delta_tx
+ self.prev_tx = cur_tx
+ self.prev_rx = cur_rx
+ LOG.info('TX: %15s; RX: %15s; (Est.) Dropped: %12s; Drop rate: %8.4f%%',
+ format(cur_tx, ',d'),
+ format(cur_rx, ',d'),
+ format(drops, ',d'),
+ drop_rate_pct)
def run_traffic(self):
"""Start traffic and return intermediate stats for each interval."""
stats = self.runner.run()
+ self.prev_tx = 0
+ self.prev_rx = 0
while self.runner.is_running:
self.log_stats(stats)
yield stats
# because we want each direction to have the far end RX rates,
# use the far end index (1-idx) to retrieve the RX rates
for idx, key in enumerate(["direction-forward", "direction-reverse"]):
- tx_rate = results["stats"][idx]["tx"]["total_pkts"] / self.config.duration_sec
- rx_rate = results["stats"][1 - idx]["rx"]["total_pkts"] / self.config.duration_sec
+ tx_rate = results["stats"][str(idx)]["tx"]["total_pkts"] / self.config.duration_sec
+ rx_rate = results["stats"][str(1 - idx)]["rx"]["total_pkts"] / self.config.duration_sec
+
+ orig_rate = self.run_config['rates'][idx]
+ if self.config.periodic_gratuitous_arp:
+ orig_rate['rate_pps'] = float(
+ orig_rate['rate_pps']) - self.config.gratuitous_arp_pps
+
r[key] = {
- "orig": self.__convert_rates(self.run_config['rates'][idx]),
+ "orig": self.__convert_rates(orig_rate),
"tx": self.__convert_rates({'rate_pps': tx_rate}),
"rx": self.__convert_rates({'rate_pps': rx_rate})
}
+ if self.config.periodic_gratuitous_arp:
+ r['garp-direction-total'] = {
+ "orig": self.__convert_rates({'rate_pps': self.config.gratuitous_arp_pps * 2}),
+ "tx": self.__convert_rates({'rate_pps': results["stats"]["garp_total_tx_rate"]}),
+ "rx": self.__convert_rates({'rate_pps': 0})
+ }
+
total = {}
for direction in ['orig', 'tx', 'rx']:
total[direction] = {}
for unit in ['rate_percent', 'rate_bps', 'rate_pps']:
- total[direction][unit] = sum([float(x[direction][unit]) for x in r.values()])
+ total[direction][unit] = sum([float(x[direction][unit]) for x in list(r.values())])
r['direction-total'] = total
+
return r
def insert_interface_stats(self, pps_list):
for chain_idx in range(self.config.service_chain_count)]
# note that we need to make a copy of the ifs list so that any modification in the
# list from pps will not change the list saved in self.ifstats
- self.pps_list = [PacketPathStats(list(ifs)) for ifs in self.ifstats]
+ self.pps_list = [PacketPathStats(self.config, list(ifs)) for ifs in self.ifstats]
# insert the corresponding pps in the passed list
pps_list.extend(self.pps_list)
]
"""
if diff:
- stats = self.gen.get_stats()
+ stats = self.gen.get_stats(self.ifstats)
for chain_idx, ifs in enumerate(self.ifstats):
# each ifs has exactly 2 InterfaceStats and 2 Latency instances
# corresponding to the
Code Review / nfvbench.git / blobdiff