tools/pkt_gen/xena/xena.py

   1 # Copyright 2016 Red Hat Inc & Xena Networks.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #   http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 #
  15 # Contributors:
  16 #   Rick Alongi, Red Hat Inc.
  17 #   Amit Supugade, Red Hat Inc.
  18 #   Dan Amzulescu, Xena Networks
  19 #   Christian Trautman, Red Hat Inc.
  20
  21 """
  22 Xena Traffic Generator Model
  23 """
  24
  25 # python imports
  26 import logging
  27 import subprocess
  28 import sys
  29 from time import sleep
  30 import xml.etree.ElementTree as ET
  31 from collections import OrderedDict
  32
  33 # VSPerf imports
  34 from conf import settings
  35 from core.results.results_constants import ResultsConstants
  36 from tools.pkt_gen.trafficgen.trafficgenhelper import (
  37     TRAFFIC_DEFAULTS,
  38     merge_spec)
  39 from tools.pkt_gen.trafficgen.trafficgen import ITrafficGenerator
  40
  41 # Xena module imports
  42 from tools.pkt_gen.xena.xena_json import XenaJSON
  43
  44
  45 class Xena(ITrafficGenerator):
  46     """
  47     Xena Traffic generator wrapper class
  48     """
  49     _traffic_defaults = TRAFFIC_DEFAULTS.copy()
  50     _logger = logging.getLogger(__name__)
  51
  52     def __init__(self):
  53         self.mono_pipe = None
  54         self._params = {}
  55         self._duration = None
  56
  57     @property
  58     def traffic_defaults(self):
  59         """Default traffic values.
  60
  61         These can be expected to be constant across traffic generators,
  62         so no setter is provided. Changes to the structure or contents
  63         will likely break traffic generator implementations or tests
  64         respectively.
  65         """
  66         return self._traffic_defaults
  67
  68     @staticmethod
  69     def _create_throughput_result(root):
  70         """
  71         Create the results based off the output xml file from the Xena2544.exe
  72         execution
  73         :param root: root dictionary from xml import
  74         :return: Results Ordered dictionary based off ResultsConstants
  75         """
  76         # get the test type from the report file
  77         test_type = root[0][1].get('TestType')
  78
  79         if test_type == 'Throughput':
  80             results = OrderedDict()
  81             results[ResultsConstants.THROUGHPUT_RX_FPS] = int(
  82                 root[0][1][0][1].get('PortRxPps'))
  83             results[ResultsConstants.THROUGHPUT_RX_MBPS] = int(
  84                 root[0][1][0][1].get('PortRxBpsL1')) / 1000000
  85             results[ResultsConstants.THROUGHPUT_RX_PERCENT] = (
  86                 100 - int(root[0][1][0].get('TotalLossRatioPcnt'))) * float(
  87                     root[0][1][0].get('TotalTxRatePcnt'))/100
  88             results[ResultsConstants.TX_RATE_FPS] = root[0][1][0].get(
  89                 'TotalTxRateFps')
  90             results[ResultsConstants.TX_RATE_MBPS] = float(
  91                 root[0][1][0].get('TotalTxRateBpsL1')) / 1000000
  92             results[ResultsConstants.TX_RATE_PERCENT] = root[0][1][0].get(
  93                 'TotalTxRatePcnt')
  94             try:
  95                 results[ResultsConstants.MIN_LATENCY_NS] = float(
  96                     root[0][1][0][0].get('MinLatency')) * 1000
  97             except ValueError:
  98                 # Stats for latency returned as N/A so just post them
  99                 results[ResultsConstants.MIN_LATENCY_NS] = root[0][1][0][0].get(
 100                     'MinLatency')
 101             try:
 102                 results[ResultsConstants.MAX_LATENCY_NS] = float(
 103                     root[0][1][0][0].get('MaxLatency')) * 1000
 104             except ValueError:
 105                 # Stats for latency returned as N/A so just post them
 106                 results[ResultsConstants.MAX_LATENCY_NS] = root[0][1][0][0].get(
 107                     'MaxLatency')
 108             try:
 109                 results[ResultsConstants.AVG_LATENCY_NS] = float(
 110                     root[0][1][0][0].get('AvgLatency')) * 1000
 111             except ValueError:
 112                 # Stats for latency returned as N/A so just post them
 113                 results[ResultsConstants.AVG_LATENCY_NS] = root[0][1][0][0].get(
 114                     'AvgLatency')
 115         elif test_type == 'Back2Back':
 116             results = OrderedDict()
 117
 118             # Just mimic what Ixia does and only return the b2b frame count.
 119             # This may change later once its decided the common results stats
 120             # to be returned should be.
 121             results[ResultsConstants.B2B_FRAMES] = root[0][1][0][0].get(
 122                 'TotalTxBurstFrames')
 123         else:
 124             raise NotImplementedError('Unknown test type in report file.')
 125
 126         return results
 127
 128     def _setup_json_config(self, trials, loss_rate, testtype=None):
 129         """
 130         Create a 2bUsed json file that will be used for xena2544.exe execution.
 131         :param trials: Number of trials
 132         :param loss_rate: The acceptable loss rate as float
 133         :param testtype: Either '2544_b2b' or '2544_throughput' as string
 134         :return: None
 135         """
 136         try:
 137             j_file = XenaJSON('./tools/pkt_gen/xena/profiles/baseconfig.x2544')
 138             j_file.set_chassis_info(
 139                 settings.getValue('TRAFFICGEN_XENA_IP'),
 140                 settings.getValue('TRAFFICGEN_XENA_PASSWORD')
 141             )
 142             j_file.set_port(0, settings.getValue('TRAFFICGEN_XENA_MODULE1'),
 143                             settings.getValue('TRAFFICGEN_XENA_PORT1')
 144                             )
 145             j_file.set_port(1, settings.getValue('TRAFFICGEN_XENA_MODULE2'),
 146                             settings.getValue('TRAFFICGEN_XENA_PORT2')
 147                             )
 148             j_file.set_test_options(
 149                 packet_sizes=self._params['traffic']['l2']['framesize'],
 150                 iterations=trials, loss_rate=loss_rate,
 151                 duration=self._duration, micro_tpld=True if self._params[
 152                     'traffic']['l2']['framesize'] == 64 else False)
 153             if testtype == '2544_throughput':
 154                 j_file.enable_throughput_test()
 155             elif testtype == '2544_b2b':
 156                 j_file.enable_back2back_test()
 157
 158             j_file.set_header_layer2(
 159                 dst_mac=self._params['traffic']['l2']['dstmac'],
 160                 src_mac=self._params['traffic']['l2']['srcmac'])
 161             j_file.set_header_layer3(
 162                 src_ip=self._params['traffic']['l3']['srcip'],
 163                 dst_ip=self._params['traffic']['l3']['dstip'],
 164                 protocol=self._params['traffic']['l3']['proto'])
 165             j_file.set_header_layer4_udp(
 166                 source_port=self._params['traffic']['l4']['srcport'],
 167                 destination_port=self._params['traffic']['l4']['dstport'])
 168             if self._params['traffic']['vlan']['enabled']:
 169                 j_file.set_header_vlan(
 170                     vlan_id=self._params['traffic']['vlan']['id'],
 171                     id=self._params['traffic']['vlan']['cfi'],
 172                     prio=self._params['traffic']['vlan']['priority'])
 173             j_file.add_header_segments(
 174                 flows=self._params['traffic']['multistream'],
 175                 multistream_layer=self._params['traffic']['stream_type'])
 176             # set duplex mode
 177             if bool(self._params['traffic']['bidir']):
 178                 j_file.set_topology_mesh()
 179             else:
 180                 j_file.set_topology_blocks()
 181
 182             j_file.write_config('./tools/pkt_gen/xena/profiles/2bUsed.x2544')
 183         except Exception as exc:
 184             self._logger.exception("Error during Xena JSON setup: %s", exc)
 185             raise
 186
 187     def connect(self):
 188         """Connect to the traffic generator.
 189
 190         This is an optional function, designed for traffic generators
 191         which must be "connected to" (i.e. via SSH or an API) before
 192         they can be used. If not required, simply do nothing here.
 193
 194         Where implemented, this function should raise an exception on
 195         failure.
 196
 197         :returns: None
 198         """
 199         pass
 200
 201     def disconnect(self):
 202         """Disconnect from the traffic generator.
 203
 204         As with :func:`connect`, this function is optional.
 205
 206         Where implemented, this function should raise an exception on
 207         failure.
 208
 209         :returns: None
 210         """
 211         pass
 212
 213     def send_burst_traffic(self, traffic=None, numpkts=100, duration=20):
 214         """Send a burst of traffic.
 215
 216         Send a ``numpkts`` packets of traffic, using ``traffic``
 217         configuration, with a timeout of ``time``.
 218
 219         Attributes:
 220         :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
 221         :param numpkts: Number of packets to send
 222         :param duration: Time to wait to receive packets
 223
 224         :returns: dictionary of strings with following data:
 225             - List of Tx Frames,
 226             - List of Rx Frames,
 227             - List of Tx Bytes,
 228             - List of List of Rx Bytes,
 229             - Payload Errors and Sequence Errors.
 230         """
 231         raise NotImplementedError('Xena burst traffic not implemented')
 232
 233     def send_cont_traffic(self, traffic=None, duration=20):
 234         """Send a continuous flow of traffic.r
 235
 236         Send packets at ``framerate``, using ``traffic`` configuration,
 237         until timeout ``time`` occurs.
 238
 239         :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
 240         :param duration: Time to wait to receive packets (secs)
 241         :returns: dictionary of strings with following data:
 242             - Tx Throughput (fps),
 243             - Rx Throughput (fps),
 244             - Tx Throughput (mbps),
 245             - Rx Throughput (mbps),
 246             - Tx Throughput (% linerate),
 247             - Rx Throughput (% linerate),
 248             - Min Latency (ns),
 249             - Max Latency (ns),
 250             - Avg Latency (ns)
 251         """
 252         raise NotImplementedError('Xena continuous traffic not implemented')
 253
 254     def start_cont_traffic(self, traffic=None, duration=20):
 255         """Non-blocking version of 'send_cont_traffic'.
 256
 257         Start transmission and immediately return. Do not wait for
 258         results.
 259         :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
 260         :param duration: Time to wait to receive packets (secs)
 261         """
 262         raise NotImplementedError('Xena continuous traffic not implemented')
 263
 264     def stop_cont_traffic(self):
 265         """Stop continuous transmission and return results.
 266         """
 267         raise NotImplementedError('Xena continuous traffic not implemented')
 268
 269     def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
 270                                 lossrate=0.0):
 271         """Send traffic per RFC2544 throughput test specifications.
 272
 273         See ITrafficGenerator for description
 274         """
 275         self._duration = duration
 276
 277         self._params.clear()
 278         self._params['traffic'] = self.traffic_defaults.copy()
 279         if traffic:
 280             self._params['traffic'] = merge_spec(self._params['traffic'],
 281                                                  traffic)
 282
 283         self._setup_json_config(trials, lossrate, '2544_throughput')
 284
 285         args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
 286                 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
 287                 "./tools/pkt_gen/xena", "-u",
 288                 settings.getValue('TRAFFICGEN_XENA_USER')]
 289         self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
 290         self.mono_pipe.communicate()
 291         root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
 292         return Xena._create_throughput_result(root)
 293
 294     def start_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
 295                                  lossrate=0.0):
 296         """Non-blocking version of 'send_rfc2544_throughput'.
 297
 298         See ITrafficGenerator for description
 299         """
 300         self._duration = duration
 301         self._params.clear()
 302         self._params['traffic'] = self.traffic_defaults.copy()
 303         if traffic:
 304             self._params['traffic'] = merge_spec(self._params['traffic'],
 305                                                  traffic)
 306
 307         self._setup_json_config(trials, lossrate, '2544_throughput')
 308
 309         args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
 310                 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
 311                 "./tools/pkt_gen/xena", "-u",
 312                 settings.getValue('TRAFFICGEN_XENA_USER')]
 313         self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
 314
 315     def wait_rfc2544_throughput(self):
 316         """Wait for and return results of RFC2544 test.
 317
 318         See ITrafficGenerator for description
 319         """
 320         self.mono_pipe.communicate()
 321         sleep(2)
 322         root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
 323         return Xena._create_throughput_result(root)
 324
 325     def send_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
 326                                lossrate=0.0):
 327         """Send traffic per RFC2544 back2back test specifications.
 328
 329         See ITrafficGenerator for description
 330         """
 331         self._duration = duration
 332
 333         self._params.clear()
 334         self._params['traffic'] = self.traffic_defaults.copy()
 335         if traffic:
 336             self._params['traffic'] = merge_spec(self._params['traffic'],
 337                                                  traffic)
 338
 339         self._setup_json_config(trials, lossrate, '2544_b2b')
 340
 341         args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
 342                 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
 343                 "./tools/pkt_gen/xena", "-u",
 344                 settings.getValue('TRAFFICGEN_XENA_USER')]
 345         self.mono_pipe = subprocess.Popen(
 346             args, stdout=sys.stdout)
 347         self.mono_pipe.communicate()
 348         root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
 349         return Xena._create_throughput_result(root)
 350
 351     def start_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
 352                                 lossrate=0.0):
 353         """Non-blocking version of 'send_rfc2544_back2back'.
 354
 355         See ITrafficGenerator for description
 356         """
 357         self._duration = duration
 358
 359         self._params.clear()
 360         self._params['traffic'] = self.traffic_defaults.copy()
 361         if traffic:
 362             self._params['traffic'] = merge_spec(self._params['traffic'],
 363                                                  traffic)
 364
 365         self._setup_json_config(trials, lossrate, '2544_b2b')
 366
 367         args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
 368                 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
 369                 "./tools/pkt_gen/xena", "-u",
 370                 settings.getValue('TRAFFICGEN_XENA_USER')]
 371         self.mono_pipe = subprocess.Popen(
 372             args, stdout=sys.stdout)
 373
 374     def wait_rfc2544_back2back(self):
 375         """Wait and set results of RFC2544 test.
 376         """
 377         self.mono_pipe.communicate()
 378         sleep(2)
 379         root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
 380         return Xena._create_throughput_result(root)
 381
 382
 383 if __name__ == "__main__":
 384     pass
 385