snappi-trex

snappi-trex is a snappi plugin that allows executing scripts written using snappi with Cisco's TRex Traffic Generator.

snappi-trex is currently compatible with snappi v0.6.4 and Open Traffic Generator v0.6.4

Design

snappi-trex converts snappi Open Traffic Generator API configuration into the equivalent TRex STL Client configuration. This allows users to use the TRex Traffic Generator and its useful features without having to write complex TRex scripts.

diagram

The above diagram outlines the overall process of how the snappi Open Traffic Generator API is able to interface with TRex and generate traffic over its network interfaces. snappi-trex is essential to convert snappi scripts into the equivalent TRex STL Client instructions.

snappi-trex usage follows the standard usage of snappi with a few modifications outlined in the Usage document.

Demos

Click here for the Quickstart Guide Video Tutorial

This goes over the installation and setup for snappi-trex, and how to run a basic snappi script using snappi-trex

Click here for the snappi-trex P4 PTF Demo

This demonstrates snappi-trex being used with the P4 Packet Testing Framework in a 4 Port Mesh configuration

Quickstart

snappi-trex is a snappi plugin that allows executing scripts written using snappi with Cisco's TRex Traffic Generator

--> Click here for the Quickstart Guide Video Tutorial

Installing and Running TRex

TRex must be installed and running before proceeding

TRex must be installed and configured in order to use snappi-trex. For a quick tutorial on TRex installation, running, and basic usage, check out my TRex Tutorial

Installing snappi-trex

Make sure python-pip3 is installed

sudo apt-get install python3-pip

Install snappi and the snappi-trex extension

pip3 install snappi==0.6.1 snappi[trex]

Start Scripting

Let's run our first script called hello_snappi_trex.py: A basic snappi script that transmits 1000 UDP packets bidirectionally between two ports and verifies that they are received. This file can be found at examples/hello_snappi_trex.py in the snappi-trex Github Repo.

git clone https://github.com/open-traffic-generator/snappi-trex
python3 snappi-trex/examples/hello_snappi_trex.py

You may also just paste the script in from below.

hello_snappi_trex.py

import snappi
import sys, os

# Replace v2.90 with the installed version of TRex. 
# Change '/opt/trex' if you installed TRex in another location
trex_path = '/opt/trex/v2.90/automation/trex_control_plane/interactive'
sys.path.insert(0, os.path.abspath(trex_path))


def hello_snappi_trex():
    """
    This script does following:
    - Send 1000 packets back and forth between the two ports at a rate of
      1000 packets per second.
    - Validate that total packets sent and received on both interfaces is as
      expected using port metrics.
    - Validate that captured UDP packets on both the ports are as expected.
    """
    # create a new API instance where host points to controller
    api = snappi.api(ext='trex')
    # and an empty traffic configuration to be pushed to controller later on
    cfg = api.config()

    # add two ports where location points to traffic-engine (aka ports)
    p1, p2 = (
        cfg.ports
        .port(name='p1')
        .port(name='p2')
    )

    # add layer 1 property to configure same speed on both ports
    ly = cfg.layer1.layer1(name='ly')[-1]
    ly.port_names = [p1.name, p2.name]
    ly.speed = ly.SPEED_1_GBPS

    # enable packet capture on both ports
    cp = cfg.captures.capture(name='cp')[-1]
    cp.port_names = [p1.name, p2.name]

    # add two traffic flows
    f1, f2 = cfg.flows.flow(name='flow p1->p2').flow(name='flow p2->p1')
    # and assign source and destination ports for each
    f1.tx_rx.port.tx_name, f1.tx_rx.port.rx_name = p1.name, p2.name
    f2.tx_rx.port.tx_name, f2.tx_rx.port.rx_name = p2.name, p1.name

    # configure packet size, rate and duration for both flows
    f1.size.fixed, f2.size.fixed = 128, 256
    for f in cfg.flows:
        # send 1000 packets and stop
        f.duration.fixed_packets.packets = 1000
        # send 1000 packets per second
        f.rate.pps = 1000

    # configure packet with Ethernet, IPv4 and UDP headers for both flows
    eth1, ip1, udp1 = f1.packet.ethernet().ipv4().udp()
    eth2, ip2, udp2 = f2.packet.ethernet().ipv4().udp()

    # set source and destination MAC addresses
    eth1.src.value, eth1.dst.value = '00:AA:00:00:04:00', '00:AA:00:00:00:AA'
    eth2.src.value, eth2.dst.value = '00:AA:00:00:00:AA', '00:AA:00:00:04:00'

    # set source and destination IPv4 addresses
    ip1.src.value, ip1.dst.value = '10.0.0.1', '10.0.0.2'
    ip2.src.value, ip2.dst.value = '10.0.0.2', '10.0.0.1'

    # set incrementing port numbers as source UDP ports
    udp1.src_port.increment.start = 5000
    udp1.src_port.increment.step = 2
    udp1.src_port.increment.count = 10

    udp2.src_port.increment.start = 6000
    udp2.src_port.increment.step = 4
    udp2.src_port.increment.count = 10

    # assign list of port numbers as destination UDP ports
    udp1.dst_port.values = [4000, 4044, 4060, 4074]
    udp2.dst_port.values = [8000, 8044, 8060, 8074, 8082, 8084]

    print('Pushing traffic configuration ...')
    api.set_config(cfg)

    print('Starting packet capture on all configured ports ...')
    cs = api.capture_state()
    cs.state = cs.START
    api.set_capture_state(cs)

    print('Starting transmit on all configured flows ...')
    ts = api.transmit_state()
    ts.state = ts.START
    api.set_transmit_state(ts)

    print('Checking metrics on all configured ports ...')
    print('Expected\tTotal Tx\tTotal Rx')
    assert wait_for(lambda: metrics_ok(api, cfg)), 'Metrics validation failed!'

    assert captures_ok(api, cfg), 'Capture validation failed!'

    print('Test passed !')


def metrics_ok(api, cfg):
    # create a port metrics request and filter based on port names
    req = api.metrics_request()
    req.port.port_names = [p.name for p in cfg.ports]
    # include only sent and received packet counts
    req.port.column_names = [req.port.FRAMES_TX, req.port.FRAMES_RX]

    # fetch port metrics
    res = api.get_metrics(req)
    # calculate total frames sent and received across all configured ports
    total_tx = sum([m.frames_tx for m in res.port_metrics])
    total_rx = sum([m.frames_rx for m in res.port_metrics])
    expected = sum([f.duration.fixed_packets.packets for f in cfg.flows])

    print('%d\t\t%d\t\t%d' % (expected, total_tx, total_rx))

    return expected == total_tx and total_rx >= expected


def captures_ok(api, cfg):
    import dpkt
    print('Checking captured packets on all configured ports ...')
    print('Port Name\tExpected\tUDP packets')

    result = []
    for p in cfg.ports:
        exp, act = 1000, 0
        # create capture request and filter based on port name
        req = api.capture_request()
        req.port_name = p.name
        # fetch captured pcap bytes and feed it to pcap parser dpkt
        pcap = dpkt.pcap.Reader(api.get_capture(req))
        for _, buf in pcap:
            # check if current packet is a valid UDP packet
            eth = dpkt.ethernet.Ethernet(buf)
            if isinstance(eth.data.data, dpkt.udp.UDP):
                act += 1

        print('%s\t\t%d\t\t%d' % (p.name, exp, act))
        result.append(exp == act)

    return all(result)


def wait_for(func, timeout=10, interval=0.2):
    """
    Keeps calling the `func` until it returns true or `timeout` occurs
    every `interval` seconds.
    """
    import time
    start = time.time()

    while time.time() - start <= timeout:
        if func():
            return True
        time.sleep(interval)

    print('Timeout occurred !')
    return False


if __name__ == '__main__':
    hello_snappi_trex()

Output

If everything is working correctly, you should see a similar output as this.

Pushing traffic configuration ...
Starting packet capture on all configured ports ...
Starting transmit on all configured flows ...
Checking metrics on all configured ports ...
Expected        Total Tx        Total Rx
2000            19              17
2000            445             437
2000            881             881
2000            1325            1325
2000            1761            1761
2000           2000            2000
Checking captured packets on all configured ports ...
Port Name       Expected        UDP packets
p1              1000            1000
p2              1000            1000
Test passed !

Metadata

Release files for snappi-trex 0.1.6

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for snappi-trex 0.1.6
File	Size	Uploaded	Provenance
snappi_trex-0.1.6.tar.gz	22.7 kB	Oct 12, 2021	Without provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for snappi-trex 0.1.6
File	Interpreter	ABI	Platform	Reset
snappi_trex-0.1.6-py2.py3-none-any.whl 21.9 kB Oct 12, 2021 Without provenance	Python 2, Python 3	none	any	Details

Total release size: 44.7 kB

Release files / snappi_trex-0.1.6.tar.gz

Download URL	snappi_trex-0.1.6.tar.gz
Size	22.7 kB
Tags	Source
SHA-256 checksum How to use checksums	`9d79e9abc0b8690054f12e6c1076b4c9fbf2eb57eea9a53ab21f0c0d3b866804`
BLAKE2b-256 checksum How to use checksums	`30db1b2d0210c19f2df3977eacb5b6119244edd0fc746b6edd315c6f791f3578`
Upload date	Oct 12, 2021
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/3.4.2 importlib_metadata/4.8.1 pkginfo/1.7.1 requests/2.22.0 requests-toolbelt/0.9.1 tqdm/4.62.3 CPython/3.8.10`

Release files / snappi_trex-0.1.6-py2.py3-none-any.whl

Download URL	snappi_trex-0.1.6-py2.py3-none-any.whl
Size	21.9 kB
Tags	Python 2 Python 3
SHA-256 checksum How to use checksums	`5444c050714488fd786c82ae0e0d3e529354b9d164bdde136c6fbd0254e423c4`
BLAKE2b-256 checksum How to use checksums	`def86beb437c6c7c573a17f4e2923900fad0ebde94f8405ef3599f7a1f7dd1cf`
Upload date	Oct 12, 2021
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/3.4.2 importlib_metadata/4.8.1 pkginfo/1.7.1 requests/2.22.0 requests-toolbelt/0.9.1 tqdm/4.62.3 CPython/3.8.10`

Security for snappi-trex 0.1.6

Report project as malware

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.