summaryrefslogtreecommitdiffstats
path: root/tools/pkt_gen/ixnet/ixnet.py
diff options
context:
space:
mode:
Diffstat (limited to 'tools/pkt_gen/ixnet/ixnet.py')
-rwxr-xr-xtools/pkt_gen/ixnet/ixnet.py516
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())