diff options
Diffstat (limited to 'tools/pkt_gen/ixnet/ixnet.py')
| -rwxr-xr-x | tools/pkt_gen/ixnet/ixnet.py | 516 |
1 files changed, 516 insertions, 0 deletions
diff --git a/tools/pkt_gen/ixnet/ixnet.py b/tools/pkt_gen/ixnet/ixnet.py new file mode 100755 index 00000000..fdea4bfd --- /dev/null +++ b/tools/pkt_gen/ixnet/ixnet.py @@ -0,0 +1,516 @@ +# 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. +"""IxNetwork traffic generator model. + +Provides a model for an IxNetwork machine and appropriate applications. + +This requires the following settings in your config file: + +* TRAFFICGEN_IXNET_LIB_PATH + IxNetwork libraries path +* TRAFFICGEN_IXNET_HOST + IxNetwork host IP address +* TRAFFICGEN_IXNET_PORT + IxNetwork host port number +* TRAFFICGEN_IXNET_USER + IxNetwork host user name +* TRAFFICGEN_IXNET_TESTER_RESULT_DIR + The result directory on the IxNetwork computer +* TRAFFICGEN_IXNET_DUT_RESULT_DIR + The result directory on DUT. This needs to map to the same directory + as the previous one + +The following settings are also required. These can likely be shared +an 'Ixia' traffic generator instance: + +* 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). + +Additional Configuration: +------------------------- + +You will also need to configure the IxNetwork machine to start the IXIA +IxNetworkTclServer. This can be started like so: + +1. Connect to the IxNetwork machine using RDP +2. Go to: + + Start-> + Programs -> + Ixia -> + IxNetwork -> + IxNetwork 7.21.893.14 GA -> + IxNetworkTclServer + + Pin a shortcut to this application to the taskbar. +3. Before running it right click the pinned shortcut and go to + "Properties". Here change the port number to your own port number. + This will be the same value as "TRAFFICGEN_IXNET_PORT" above. +4. You will find this on the shortcut tab under the heading "Target" +5. Finally run it. If you see the following error check that you + followed the above steps exactly: + + ERROR: couldn't open socket : connection refused + +Debugging: +---------- + +This method of automation is quite error prone as the IxNetwork API +does not give any feedback as to the status of tests. As such, it can +be expected that the user have access to the IxNetwork machine should +this trafficgen need to be debugged. +""" + +import tkinter +import logging +import os +import re +import csv + +from collections import OrderedDict +from tools.pkt_gen import trafficgen +from conf import settings +from core.results.results_constants import ResultsConstants + +_ROOT_DIR = os.path.dirname(os.path.realpath(__file__)) + +_RESULT_RE = r'(?:\{kString,result\},\{kString,)(\w+)(?:\})' +_RESULTPATH_RE = r'(?:\{kString,resultPath\},\{kString,)([\\\w\.\-]+)(?:\})' + + +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 + + # pylint: disable=unidiomatic-typecheck + # 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 IxNet(trafficgen.ITrafficGenerator): + """A wrapper around IXIA IxNetwork applications. + + Runs different traffic generator tests through an Ixia traffic + generator chassis by generating TCL scripts from templates. + + Currently only the RFC2544 tests are implemented. + """ + _script = os.path.join(os.path.dirname(__file__), 'ixnetrfc2544.tcl') + _tclsh = tkinter.Tcl() + _cfg = None + _logger = logging.getLogger(__name__) + _params = None + + 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): + """Configure system for IxNetwork. + """ + self._cfg = { + 'lib_path': settings.getValue('TRAFFICGEN_IXNET_LIB_PATH'), + # IxNetwork machine configuration + 'machine': settings.getValue('TRAFFICGEN_IXNET_MACHINE'), + 'port': settings.getValue('TRAFFICGEN_IXNET_PORT'), + 'user': settings.getValue('TRAFFICGEN_IXNET_USER'), + # IXIA chassis configuration + 'chassis': settings.getValue('TRAFFICGEN_IXIA_HOST'), + 'card': settings.getValue('TRAFFICGEN_IXIA_CARD'), + 'port1': settings.getValue('TRAFFICGEN_IXIA_PORT1'), + 'port2': settings.getValue('TRAFFICGEN_IXIA_PORT2'), + 'output_dir': settings.getValue( + 'TRAFFICGEN_IXNET_TESTER_RESULT_DIR'), + } + + self._logger.debug('IXIA configuration configuration : %s', self._cfg) + + return self + + def disconnect(self): + """Disconnect from Ixia chassis. + """ + pass + + def send_cont_traffic(self, traffic=None, time=20, framerate=0, + multistream=False): + """See ITrafficGenerator for description + """ + self.start_cont_traffic(traffic, time, framerate, multistream) + + return self.stop_cont_traffic() + + def start_cont_traffic(self, traffic=None, time=20, framerate=0, + multistream=False): + """Start transmission. + """ + self._params = {} + + self._params['config'] = { + 'binary': False, # don't do binary search and send one stream + 'time': time, + 'framerate': framerate, + 'multipleStreams': multistream, + 'rfc2544TestType': 'throughput', + } + self._params['traffic'] = self.traffic_defaults.copy() + + if traffic: + self._params['traffic'] = trafficgen.merge_spec( + self._params['traffic'], traffic) + + for cmd in _build_set_cmds(self._cfg, prefix='set'): + self.run_tcl(cmd) + + for cmd in _build_set_cmds(self._params): + self.run_tcl(cmd) + + output = self.run_tcl('source {%s}' % self._script) + if output: + self._logger.critical( + 'An error occured when connecting to IxNetwork machine...') + raise RuntimeError('Ixia failed to initialise.') + + self.run_tcl('startRfc2544Test $config $traffic') + if output: + self._logger.critical( + 'Failed to start continuous traffic test') + raise RuntimeError('Continuous traffic test failed to start.') + + def stop_cont_traffic(self): + """See ITrafficGenerator for description + """ + return self._wait_result() + + def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20, + lossrate=0.0, multistream=False): + """See ITrafficGenerator for description + """ + self.start_rfc2544_throughput(traffic, trials, duration, lossrate, + multistream) + + return self.wait_rfc2544_throughput() + + def start_rfc2544_throughput(self, traffic=None, trials=3, duration=20, + lossrate=0.0, multistream=False): + """Start transmission. + """ + self._params = {} + + self._params['config'] = { + 'binary': True, + 'trials': trials, + 'duration': duration, + 'lossrate': lossrate, + 'multipleStreams': multistream, + 'rfc2544TestType': 'throughput', + } + self._params['traffic'] = self.traffic_defaults.copy() + + if traffic: + self._params['traffic'] = trafficgen.merge_spec( + self._params['traffic'], traffic) + + for cmd in _build_set_cmds(self._cfg, prefix='set'): + self.run_tcl(cmd) + + for cmd in _build_set_cmds(self._params): + self.run_tcl(cmd) + + output = self.run_tcl('source {%s}' % self._script) + if output: + self._logger.critical( + 'An error occured when connecting to IxNetwork machine...') + raise RuntimeError('Ixia failed to initialise.') + + self.run_tcl('startRfc2544Test $config $traffic') + if output: + self._logger.critical( + 'Failed to start RFC2544 test') + raise RuntimeError('RFC2544 test failed to start.') + + def wait_rfc2544_throughput(self): + """See ITrafficGenerator for description + """ + return self._wait_result() + + def _wait_result(self): + """Wait for results. + """ + def parse_result_string(results): + """Get path to results file from output + + Check for related errors + + :param results: Text stream from test. + + :returns: Path to results file. + """ + result_status = re.search(_RESULT_RE, results) + result_path = re.search(_RESULTPATH_RE, results) + + if not result_status or not result_path: + self._logger.critical( + 'Could not parse results from IxNetwork machine...') + raise ValueError('Failed to parse output.') + + if result_status.group(1) != 'pass': + self._logger.critical( + 'An error occured when running tests...') + raise RuntimeError('Ixia failed to initialise.') + + # transform path into someting useful + + path = result_path.group(1).replace('\\', '/') + path = os.path.join(path, 'results.csv') + path = path.replace( + settings.getValue('TRAFFICGEN_IXNET_TESTER_RESULT_DIR'), + settings.getValue('TRAFFICGEN_IXNET_DUT_RESULT_DIR')) + return path + + def parse_ixnet_rfc_results(path): + """Parse CSV output of IxNet RFC2544 test run. + + :param path: Input file path + """ + results = OrderedDict() + + with open(path, 'r') as in_file: + reader = csv.reader(in_file, delimiter=',') + next(reader) + for row in reader: + #Replace null entries added by Ixia with 0s. + row = [entry if len(entry) > 0 else '0' for entry in row] + # calculate tx fps by (rx fps * (tx % / rx %)) + tx_fps = float(row[9]) * (float(row[8]) / float(row[7])) + # calculate tx mbps by (rx mbps * (tx % / rx %)) + tx_mbps = float(row[10]) * (float(row[8]) / float(row[7])) + + if bool(results.get(ResultsConstants.THROUGHPUT_RX_FPS)) \ + == False: + prev_percent_rx = 0.0 + else: + prev_percent_rx = \ + float(results.get(ResultsConstants.THROUGHPUT_RX_FPS)) + if float(row[9]) >= prev_percent_rx: + results[ResultsConstants.THROUGHPUT_TX_FPS] = tx_fps + results[ResultsConstants.THROUGHPUT_RX_FPS] = row[9] + results[ResultsConstants.THROUGHPUT_TX_MBPS] = tx_mbps + results[ResultsConstants.THROUGHPUT_RX_MBPS] = row[10] + results[ResultsConstants.THROUGHPUT_TX_PERCENT] = row[7] + results[ResultsConstants.THROUGHPUT_TX_PERCENT] = row[8] + results[ResultsConstants.MIN_LATENCY_NS] = row[15] + results[ResultsConstants.MAX_LATENCY_NS] = row[16] + results[ResultsConstants.AVG_LATENCY_NS] = row[17] + return results + + output = self.run_tcl('waitForRfc2544Test') + + # the run_tcl function will return a list with one element. We extract + # that one element (a string representation of an IXIA-specific Tcl + # datatype), parse it to find the path of the results file then parse + # the results file + return parse_ixnet_rfc_results(parse_result_string(output[0])) + + def send_rfc2544_back2back(self, traffic=None, trials=1, duration=20, + lossrate=0.0, multistream=False): + """See ITrafficGenerator for description + """ + self.start_rfc2544_back2back(traffic, trials, duration, lossrate, + multistream) + + return self.wait_rfc2544_back2back() + + def start_rfc2544_back2back(self, traffic=None, trials=1, duration=20, + lossrate=0.0, multistream=False): + """Start transmission. + """ + self._params = {} + + self._params['config'] = { + 'binary': True, + 'trials': trials, + 'duration': duration, + 'lossrate': lossrate, + 'multipleStreams': multistream, + 'rfc2544TestType': 'back2back', + } + self._params['traffic'] = self.traffic_defaults.copy() + + if traffic: + self._params['traffic'] = trafficgen.merge_spec( + self._params['traffic'], traffic) + + for cmd in _build_set_cmds(self._cfg, prefix='set'): + self.run_tcl(cmd) + + for cmd in _build_set_cmds(self._params): + self.run_tcl(cmd) + + output = self.run_tcl('source {%s}' % self._script) + if output: + self._logger.critical( + 'An error occured when connecting to IxNetwork machine...') + raise RuntimeError('Ixia failed to initialise.') + + self.run_tcl('startRfc2544Test $config $traffic') + if output: + self._logger.critical( + 'Failed to start RFC2544 test') + raise RuntimeError('RFC2544 test failed to start.') + + def wait_rfc2544_back2back(self): + """Wait for results. + """ + def parse_result_string(results): + """Get path to results file from output + + Check for related errors + + :param results: Text stream from test. + + :returns: Path to results file. + """ + result_status = re.search(_RESULT_RE, results) + result_path = re.search(_RESULTPATH_RE, results) + + if not result_status or not result_path: + self._logger.critical( + 'Could not parse results from IxNetwork machine...') + raise ValueError('Failed to parse output.') + + if result_status.group(1) != 'pass': + self._logger.critical( + 'An error occured when running tests...') + raise RuntimeError('Ixia failed to initialise.') + + # transform path into something useful + + path = result_path.group(1).replace('\\', '/') + path = os.path.join(path, 'iteration.csv') + path = path.replace( + settings.getValue('TRAFFICGEN_IXNET_TESTER_RESULT_DIR'), + settings.getValue('TRAFFICGEN_IXNET_DUT_RESULT_DIR')) + + return path + + def parse_ixnet_rfc_results(path): + """Parse CSV output of IxNet RFC2544 Back2Back test run. + + :param path: Input file path + + :returns: Best parsed result from CSV file. + """ + results = [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0] + + with open(path, 'r') as in_file: + reader = csv.reader(in_file, delimiter=',') + next(reader) + for row in reader: + # if back2back count higher than previously found, store it + # Note: row[N] here refers to the Nth column of a row + if float(row[14]) <= self._params['config']['lossrate']: + if int(row[12]) > int(results[5]): + results = [ + float(row[9]), # rx throughput (fps) + float(row[10]), # rx throughput (mbps) + float(row[7]), # tx rate (% linerate) + float(row[8]), # rx rate (% linerate) + int(row[11]), # tx count (frames) + int(row[12]), # back2back count (frames) + int(row[13]), # frame loss (frames) + float(row[14]), # frame loss (%) + ] + + return results + + self.run_tcl('waitForRfc2544Test') + + # the run_tcl function will return a list with one element. We extract + # that one element (a string representation of an IXIA-specific Tcl + # datatype), parse it to find the path of the results file then parse + # the results file + + #TODO implement back2back result via IResult interface. + #return parse_ixnet_rfc_results(parse_result_string(output[0])) + + +if __name__ == '__main__': + TRAFFIC = { + 'l3': { + 'proto': 'udp', + 'srcip': '10.1.1.1', + 'dstip': '10.1.1.254', + }, + } + + with IxNet() as dev: + print(dev.send_cont_traffic()) + print(dev.send_rfc2544_throughput()) |