1 # Copyright 2016 Red Hat Inc & Xena Networks.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 # Rick Alongi, Red Hat Inc.
17 # Amit Supugade, Red Hat Inc.
18 # Dan Amzulescu, Xena Networks
19 # Christian Trautman, Red Hat Inc.
22 Xena Traffic Generator Model
29 from time import sleep
30 import xml.etree.ElementTree as ET
31 from collections import OrderedDict
34 from conf import settings
35 from core.results.results_constants import ResultsConstants
36 from tools.pkt_gen.trafficgen.trafficgenhelper import (
39 from tools.pkt_gen.trafficgen.trafficgen import ITrafficGenerator
42 from tools.pkt_gen.xena.xena_json import XenaJSON
45 class Xena(ITrafficGenerator):
47 Xena Traffic generator wrapper class
49 _traffic_defaults = TRAFFIC_DEFAULTS.copy()
50 _logger = logging.getLogger(__name__)
58 def traffic_defaults(self):
59 """Default traffic values.
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
66 return self._traffic_defaults
69 def _create_throughput_result(root):
71 Create the results based off the output xml file from the Xena2544.exe
73 :param root: root dictionary from xml import
74 :return: Results Ordered dictionary based off ResultsConstants
76 # get the test type from the report file
77 test_type = root[0][1].get('TestType')
78 # set the version from the report file
79 settings.setValue('XENA_VERSION', root[0][0][1].get('GeneratedBy'))
81 if test_type == 'Throughput':
82 results = OrderedDict()
83 results[ResultsConstants.THROUGHPUT_RX_FPS] = int(
84 root[0][1][0][1].get('PortRxPps'))
85 results[ResultsConstants.THROUGHPUT_RX_MBPS] = int(
86 root[0][1][0][1].get('PortRxBpsL1')) / 1000000
87 results[ResultsConstants.THROUGHPUT_RX_PERCENT] = (
88 100 - int(root[0][1][0].get('TotalLossRatioPcnt'))) * float(
89 root[0][1][0].get('TotalTxRatePcnt'))/100
90 results[ResultsConstants.TX_RATE_FPS] = root[0][1][0].get(
92 results[ResultsConstants.TX_RATE_MBPS] = float(
93 root[0][1][0].get('TotalTxRateBpsL1')) / 1000000
94 results[ResultsConstants.TX_RATE_PERCENT] = root[0][1][0].get(
97 results[ResultsConstants.MIN_LATENCY_NS] = float(
98 root[0][1][0][0].get('MinLatency')) * 1000
100 # Stats for latency returned as N/A so just post them
101 results[ResultsConstants.MIN_LATENCY_NS] = root[0][1][0][0].get(
104 results[ResultsConstants.MAX_LATENCY_NS] = float(
105 root[0][1][0][0].get('MaxLatency')) * 1000
107 # Stats for latency returned as N/A so just post them
108 results[ResultsConstants.MAX_LATENCY_NS] = root[0][1][0][0].get(
111 results[ResultsConstants.AVG_LATENCY_NS] = float(
112 root[0][1][0][0].get('AvgLatency')) * 1000
114 # Stats for latency returned as N/A so just post them
115 results[ResultsConstants.AVG_LATENCY_NS] = root[0][1][0][0].get(
117 elif test_type == 'Back2Back':
118 results = OrderedDict()
120 # Just mimic what Ixia does and only return the b2b frame count.
121 # This may change later once its decided the common results stats
122 # to be returned should be.
123 results[ResultsConstants.B2B_FRAMES] = root[0][1][0][0].get(
124 'TotalTxBurstFrames')
126 raise NotImplementedError('Unknown test type in report file.')
130 def _setup_json_config(self, trials, loss_rate, testtype=None):
132 Create a 2bUsed json file that will be used for xena2544.exe execution.
133 :param trials: Number of trials
134 :param loss_rate: The acceptable loss rate as float
135 :param testtype: Either '2544_b2b' or '2544_throughput' as string
139 j_file = XenaJSON('./tools/pkt_gen/xena/profiles/baseconfig.x2544')
140 j_file.set_chassis_info(
141 settings.getValue('TRAFFICGEN_XENA_IP'),
142 settings.getValue('TRAFFICGEN_XENA_PASSWORD')
144 j_file.set_port(0, settings.getValue('TRAFFICGEN_XENA_MODULE1'),
145 settings.getValue('TRAFFICGEN_XENA_PORT1'))
146 j_file.set_port(1, settings.getValue('TRAFFICGEN_XENA_MODULE2'),
147 settings.getValue('TRAFFICGEN_XENA_PORT2'))
148 j_file.set_port_ip_v4(
149 0, settings.getValue("TRAFFICGEN_XENA_PORT0_IP"),
150 settings.getValue("TRAFFICGEN_XENA_PORT0_CIDR"),
151 settings.getValue("TRAFFICGEN_XENA_PORT0_GATEWAY"))
152 j_file.set_port_ip_v4(
153 1, settings.getValue("TRAFFICGEN_XENA_PORT1_IP"),
154 settings.getValue("TRAFFICGEN_XENA_PORT1_CIDR"),
155 settings.getValue("TRAFFICGEN_XENA_PORT1_GATEWAY"))
156 j_file.set_test_options(
157 packet_sizes=self._params['traffic']['l2']['framesize'],
158 iterations=trials, loss_rate=loss_rate,
159 duration=self._duration, micro_tpld=True if self._params[
160 'traffic']['l2']['framesize'] == 64 else False)
161 if testtype == '2544_throughput':
162 j_file.enable_throughput_test()
163 elif testtype == '2544_b2b':
164 j_file.enable_back2back_test()
166 j_file.set_header_layer2(
167 dst_mac=self._params['traffic']['l2']['dstmac'],
168 src_mac=self._params['traffic']['l2']['srcmac'])
169 j_file.set_header_layer3(
170 src_ip=self._params['traffic']['l3']['srcip'],
171 dst_ip=self._params['traffic']['l3']['dstip'],
172 protocol=self._params['traffic']['l3']['proto'])
173 j_file.set_header_layer4_udp(
174 source_port=self._params['traffic']['l4']['srcport'],
175 destination_port=self._params['traffic']['l4']['dstport'])
176 if self._params['traffic']['vlan']['enabled']:
177 j_file.set_header_vlan(
178 vlan_id=self._params['traffic']['vlan']['id'],
179 id=self._params['traffic']['vlan']['cfi'],
180 prio=self._params['traffic']['vlan']['priority'])
181 j_file.add_header_segments(
182 flows=self._params['traffic']['multistream'],
183 multistream_layer=self._params['traffic']['stream_type'])
185 if self._params['traffic']['bidir'] == "True":
186 j_file.set_topology_mesh()
188 j_file.set_topology_blocks()
190 j_file.write_config('./tools/pkt_gen/xena/profiles/2bUsed.x2544')
191 except Exception as exc:
192 self._logger.exception("Error during Xena JSON setup: %s", exc)
196 """Connect to the traffic generator.
198 This is an optional function, designed for traffic generators
199 which must be "connected to" (i.e. via SSH or an API) before
200 they can be used. If not required, simply do nothing here.
202 Where implemented, this function should raise an exception on
209 def disconnect(self):
210 """Disconnect from the traffic generator.
212 As with :func:`connect`, this function is optional.
214 Where implemented, this function should raise an exception on
221 def send_burst_traffic(self, traffic=None, numpkts=100, duration=20):
222 """Send a burst of traffic.
224 Send a ``numpkts`` packets of traffic, using ``traffic``
225 configuration, with a timeout of ``time``.
228 :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
229 :param numpkts: Number of packets to send
230 :param duration: Time to wait to receive packets
232 :returns: dictionary of strings with following data:
236 - List of List of Rx Bytes,
237 - Payload Errors and Sequence Errors.
239 raise NotImplementedError('Xena burst traffic not implemented')
241 def send_cont_traffic(self, traffic=None, duration=20):
242 """Send a continuous flow of traffic.r
244 Send packets at ``framerate``, using ``traffic`` configuration,
245 until timeout ``time`` occurs.
247 :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
248 :param duration: Time to wait to receive packets (secs)
249 :returns: dictionary of strings with following data:
250 - Tx Throughput (fps),
251 - Rx Throughput (fps),
252 - Tx Throughput (mbps),
253 - Rx Throughput (mbps),
254 - Tx Throughput (% linerate),
255 - Rx Throughput (% linerate),
260 raise NotImplementedError('Xena continuous traffic not implemented')
262 def start_cont_traffic(self, traffic=None, duration=20):
263 """Non-blocking version of 'send_cont_traffic'.
265 Start transmission and immediately return. Do not wait for
267 :param traffic: Detailed "traffic" spec, i.e. IP address, VLAN tags
268 :param duration: Time to wait to receive packets (secs)
270 raise NotImplementedError('Xena continuous traffic not implemented')
272 def stop_cont_traffic(self):
273 """Stop continuous transmission and return results.
275 raise NotImplementedError('Xena continuous traffic not implemented')
277 def send_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
279 """Send traffic per RFC2544 throughput test specifications.
281 See ITrafficGenerator for description
283 self._duration = duration
286 self._params['traffic'] = self.traffic_defaults.copy()
288 self._params['traffic'] = merge_spec(self._params['traffic'],
291 self._setup_json_config(trials, lossrate, '2544_throughput')
293 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
294 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
295 "./tools/pkt_gen/xena", "-u",
296 settings.getValue('TRAFFICGEN_XENA_USER')]
297 self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
298 self.mono_pipe.communicate()
299 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
300 return Xena._create_throughput_result(root)
302 def start_rfc2544_throughput(self, traffic=None, trials=3, duration=20,
304 """Non-blocking version of 'send_rfc2544_throughput'.
306 See ITrafficGenerator for description
308 self._duration = duration
310 self._params['traffic'] = self.traffic_defaults.copy()
312 self._params['traffic'] = merge_spec(self._params['traffic'],
315 self._setup_json_config(trials, lossrate, '2544_throughput')
317 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
318 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
319 "./tools/pkt_gen/xena", "-u",
320 settings.getValue('TRAFFICGEN_XENA_USER')]
321 self.mono_pipe = subprocess.Popen(args, stdout=sys.stdout)
323 def wait_rfc2544_throughput(self):
324 """Wait for and return results of RFC2544 test.
326 See ITrafficGenerator for description
328 self.mono_pipe.communicate()
330 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
331 return Xena._create_throughput_result(root)
333 def send_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
335 """Send traffic per RFC2544 back2back test specifications.
337 See ITrafficGenerator for description
339 self._duration = duration
342 self._params['traffic'] = self.traffic_defaults.copy()
344 self._params['traffic'] = merge_spec(self._params['traffic'],
347 self._setup_json_config(trials, lossrate, '2544_b2b')
349 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
350 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
351 "./tools/pkt_gen/xena", "-u",
352 settings.getValue('TRAFFICGEN_XENA_USER')]
353 self.mono_pipe = subprocess.Popen(
354 args, stdout=sys.stdout)
355 self.mono_pipe.communicate()
356 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
357 return Xena._create_throughput_result(root)
359 def start_rfc2544_back2back(self, traffic=None, trials=1, duration=20,
361 """Non-blocking version of 'send_rfc2544_back2back'.
363 See ITrafficGenerator for description
365 self._duration = duration
368 self._params['traffic'] = self.traffic_defaults.copy()
370 self._params['traffic'] = merge_spec(self._params['traffic'],
373 self._setup_json_config(trials, lossrate, '2544_b2b')
375 args = ["mono", "./tools/pkt_gen/xena/Xena2544.exe", "-c",
376 "./tools/pkt_gen/xena/profiles/2bUsed.x2544", "-e", "-r",
377 "./tools/pkt_gen/xena", "-u",
378 settings.getValue('TRAFFICGEN_XENA_USER')]
379 self.mono_pipe = subprocess.Popen(
380 args, stdout=sys.stdout)
382 def wait_rfc2544_back2back(self):
383 """Wait and set results of RFC2544 test.
385 self.mono_pipe.communicate()
387 root = ET.parse(r'./tools/pkt_gen/xena/xena2544-report.xml').getroot()
388 return Xena._create_throughput_result(root)
391 if __name__ == "__main__":