path: root/tools/pkt_gen/ixia/ixia.py

# Copyright 2015 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""IXIA traffic generator model.

Provides a model for the IXIA traffic generator. In addition, provides
a number of generic "helper" functions that are used to do the "heavy
lifting".

This requires the following settings in your config file:

* TRAFFICGEN_IXIA_LIB_PATH
    IXIA libraries path
* TRAFFICGEN_IXIA_HOST
    IXIA chassis IP address
* TRAFFICGEN_IXIA_CARD
    IXIA card
* TRAFFICGEN_IXIA_PORT1
    IXIA Tx port
* TRAFFICGEN_IXIA_PORT2
    IXIA Rx port

If any of these don't exist, the application will raise an exception
(EAFP).
"""

import tkinter
import logging
import os

from tools.pkt_gen import trafficgen
from conf import settings
from collections import OrderedDict
from core.results.results_constants import ResultsConstants

_ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
_IXIA_ROOT_DIR = settings.getValue('TRAFFICGEN_IXIA_ROOT_DIR')


def configure_env():
    """Configure envionment for TCL.

    """
    os.environ['IXIA_HOME'] = _IXIA_ROOT_DIR

    # USER MAY NEED TO CHANGE THESE IF USING OWN TCL LIBRARY
    os.environ['TCL_HOME'] = _IXIA_ROOT_DIR
    os.environ['TCLver'] = '8.5'

    # USER NORMALLY DOES NOT CHANGE ANY LINES BELOW
    os.environ['IxiaLibPath'] = os.path.expandvars('$IXIA_HOME/lib')
    os.environ['IxiaBinPath'] = os.path.expandvars('$IXIA_HOME/bin')

    os.environ['TCLLibPath'] = os.path.expandvars('$TCL_HOME/lib')
    os.environ['TCLBinPath'] = os.path.expandvars('$TCL_HOME/bin')

    os.environ['TCL_LIBRARY'] = os.path.expandvars('$TCLLibPath/tcl$TCLver')
    os.environ['TK_LIBRARY'] = os.path.expandvars('$TCLLibPath/tk$TCLver')

    os.environ['PATH'] = os.path.expandvars('$IxiaBinPath:.:$TCLBinPath:$PATH')
    os.environ['TCLLIBPATH'] = os.path.expandvars('$IxiaLibPath')
    os.environ['LD_LIBRARY_PATH'] = os.path.expandvars(
        '$IxiaLibPath:$TCLLibPath:$LD_LIBRARY_PATH')

    os.environ['IXIA_RESULTS_DIR'] = '/tmp/Ixia/Results'
    os.environ['IXIA_LOGS_DIR'] = '/tmp/Ixia/Logs'
    os.environ['IXIA_TCL_DIR'] = os.path.expandvars('$IxiaLibPath')
    os.environ['IXIA_SAMPLES'] = os.path.expandvars('$IxiaLibPath/ixTcl1.0')
    os.environ['IXIA_VERSION'] = '6.60.1000.11'


def _build_set_cmds(values, prefix='dict set'):
    """Generate a list of 'dict set' args for Tcl.

    Parse a dictionary and recursively build the arguments for the
    'dict set' Tcl command, given that this is of the format:

        dict set [name...] [key] [value]

    For example, for a non-nested dict (i.e. a non-dict element):

        dict set mydict mykey myvalue

    For a nested dict (i.e. a dict element):

        dict set mydict mysubdict mykey myvalue

    :param values: Dictionary to yield values for
    :param prefix: Prefix to append to output string. Generally the
        already generated part of the command.

    :yields: Output strings to be passed to a `Tcl` instance.
    """
    for key in values:
        value = values[key]

        # Not allowing derived dictionary types for now
        # pylint: disable=unidiomatic-typecheck
        if type(value) == dict:
            _prefix = ' '.join([prefix, key]).strip()
            for subkey in _build_set_cmds(value, _prefix):
                yield subkey
            continue

        # tcl doesn't recognise the strings "True" or "False", only "1"
        # or "0". Special case to convert them
        if type(value) == bool:
            value = str(int(value))
        else:
            value = str(value)

        if prefix:
            yield ' '.join([prefix, key, value]).strip()
        else:
            yield ' '.join([key, value]).strip()


class Ixia(trafficgen.ITrafficGenerator):
    """A wrapper around the IXIA traffic generator.

    Runs different traffic generator tests through an Ixia traffic
    generator chassis by generating TCL scripts from templates.
    """
    _script = os.path.join(os.path.dirname(__file__), 'pass_fail.tcl')
    _tclsh = tkinter.Tcl()
    _logger = logging.getLogger(__name__)

    def run_tcl(self, cmd):
        """Run a TCL script using the TCL interpreter found in ``tkinter``.

        :param cmd: Command to execute

        :returns: Output of command, where applicable.
        """
        self._logger.debug('%s%s', trafficgen.CMD_PREFIX, cmd)

        output = self._tclsh.eval(cmd)

        return output.split()

    def connect(self):
        """Connect to Ixia chassis.
        """
        ixia_cfg = {
            'lib_path': os.path.join(_IXIA_ROOT_DIR, 'lib', 'ixTcl1.0'),
            'host': settings.getValue('TRAFFICGEN_IXIA_HOST'),
            'card': settings.getValue('TRAFFICGEN_IXIA_CARD'),
            'port1': settings.getValue('TRAFFICGEN_IXIA_PORT1'),
            'port2': settings.getValue('TRAFFICGEN_IXIA_PORT2'),
        }

        self._logger.info('Connecting to IXIA...')

        self._logger.debug('IXIA configuration configuration : %s', ixia_cfg)

        configure_env()

        for cmd in _build_set_cmds(ixia_cfg, prefix='set'):
            self.run_tcl(cmd)

        output = self.run_tcl('source {%s}' % self._script)
        if output:
            self._logger.critical(
                'An error occured when connecting to IXIA...')
            raise RuntimeError('Ixia failed to initialise.')

        self._logger.info('Connected to IXIA...')

        return self

    def disconnect(self):
        """Disconnect from Ixia chassis.
        """
        self._logger.info('Disconnecting from IXIA...')

        self.run_tcl('cleanUp')

        self._logger.info('Disconnected from IXIA...')

    def _send_traffic(self, flow, traffic):
        """Send regular traffic.

        :param flow: Flow specification
        :param traffic: Traffic specification

        :returns: Results from IXIA
        """
        params = {}

        params['flow'] = flow
        params['traffic'] = self.traffic_defaults.copy()

        if traffic:
            params['traffic'] = trafficgen.merge_spec(
                params['traffic'], traffic)

        for cmd in _build_set_cmds(params):
            self.run_tcl(cmd)

        result = self.run_tcl('sendTraffic $flow $traffic')

        return result

    def send_burst_traffic(self, traffic=None, numpkts=100, time=20,
                           framerate=100):
        """See ITrafficGenerator for description
        """
        flow = {
            'numpkts': numpkts,
            'time': time,
            'type': 'stopStream',
            'framerate': framerate,
        }

        result = self._send_traffic(flow, traffic)

        assert len(result) == 6  # fail-fast if underlying Tcl code changes

        #TODO - implement Burst results setting via TrafficgenResults.

    def send_cont_traffic(self, traffic=None, time=20, framerate=100):
        """See ITrafficGenerator for description
        """
        flow = {
            'numpkts': 100,
            'time': time,
            'type': 'contPacket',
            'framerate': framerate,
            'multipleStreams': traffic['multistream'],
        }

        result = self._send_traffic(flow, traffic)

        return Ixia._create_result(result)

    def start_cont_traffic(self, traffic=None, time=20, framerate=100):
        """See ITrafficGenerator for description
        """
        return self.send_cont_traffic(traffic, 0, framerate)

    def stop_cont_traffic(self):
        """See ITrafficGenerator for description
        """
        return self.run_tcl('stopTraffic')

    def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
                                lossrate=0.0):
        """See ITrafficGenerator for description
        """
        params = {}

        params['config'] = {
            'trials': trials,
            'duration': duration,
            'lossrate': lossrate,
            'multipleStreams': traffic['multistream'],
        }
        params['traffic'] = self.traffic_defaults.copy()

        if traffic:
            params['traffic'] = trafficgen.merge_spec(
                params['traffic'], traffic)

        for cmd in _build_set_cmds(params):
            self.run_tcl(cmd)

        # this will return a list with one result
        result = self.run_tcl('rfcThroughputTest $config $traffic')

        return Ixia._create_result(result)

    @staticmethod
    def _create_result(result):
        """Create result based on list returned from tcl script.

        :param result: list representing output from tcl script.

        :returns: dictionary strings representing results from
                    traffic generator.
        """
        assert len(result) == 8  # fail-fast if underlying Tcl code changes

        result_dict = OrderedDict()
        # drop the first 4 elements as we don't use/need them. In
        # addition, IxExplorer does not support latency or % line rate
        # metrics so we have to return dummy values for these metrics
        result_dict[ResultsConstants.THROUGHPUT_RX_FPS] = result[4]
        result_dict[ResultsConstants.THROUGHPUT_TX_FPS] = result[5]
        result_dict[ResultsConstants.THROUGHPUT_RX_MBPS] = result[6]
        result_dict[ResultsConstants.THROUGHPUT_TX_MBPS] = result[7]
        result_dict[ResultsConstants.THROUGHPUT_TX_PERCENT] = \
                                            ResultsConstants.UNKNOWN_VALUE
        result_dict[ResultsConstants.THROUGHPUT_RX_PERCENT] = \
                                            ResultsConstants.UNKNOWN_VALUE
        result_dict[ResultsConstants.MIN_LATENCY_NS] = \
                                            ResultsConstants.UNKNOWN_VALUE
        result_dict[ResultsConstants.MAX_LATENCY_NS] = \
                                            ResultsConstants.UNKNOWN_VALUE
        result_dict[ResultsConstants.AVG_LATENCY_NS] = \
                                            ResultsConstants.UNKNOWN_VALUE

        return result_dict

if __name__ == '__main__':
    TRAFFIC = {
        'l3': {
            'proto': 'udp',
            'srcip': '10.1.1.1',
            'dstip': '10.1.1.254',
        },
    }

    with Ixia() as dev:
        print(dev.send_burst_traffic(traffic=TRAFFIC))
        print(dev.send_cont_traffic(traffic=TRAFFIC))
        print(dev.send_rfc2544_throughput(traffic=TRAFFIC))