diff options
Diffstat (limited to 'tools/pkt_gen/ixia/ixia.py')
| -rwxr-xr-x | tools/pkt_gen/ixia/ixia.py | 328 |
1 files changed, 328 insertions, 0 deletions
diff --git a/tools/pkt_gen/ixia/ixia.py b/tools/pkt_gen/ixia/ixia.py new file mode 100755 index 00000000..92ef5203 --- /dev/null +++ b/tools/pkt_gen/ixia/ixia.py @@ -0,0 +1,328 @@ +# 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, + multistream=False): + """See ITrafficGenerator for description + """ + flow = { + 'numpkts': 100, + 'time': time, + 'type': 'contPacket', + 'framerate': framerate, + 'multipleStreams': multistream, + } + + result = self._send_traffic(flow, traffic) + + return Ixia._create_result(result) + + def start_cont_traffic(self, traffic=None, time=20, framerate=100, + multistream=False): + """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, multistream=False): + """See ITrafficGenerator for description + """ + params = {} + + params['config'] = { + 'trials': trials, + 'duration': duration, + 'lossrate': lossrate, + 'multipleStreams': 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)) |