Improve iperf support

1. Use default value for --len parameter instead of enforcing it into 8k
2. Introduce Shaker parameter `datagram_size` for UDP case. The parameter
   is translated into --len
3. Fix processing of multiple threads for iperf_graph executor

Change-Id: I62e3435e6cf280725f57a060a8192c5834e27946

shaker/engine/executors/iperf.py

@@ -26,11 +26,14 @@ class IperfExecutor(base.BaseExecutor):
         cmd.add('--nodelay')
         if self.test_definition.get('mss'):
             cmd.add('--mss', self.test_definition.get('mss'))
-        cmd.add('--len', self.test_definition.get('buffer_size') or '8k')
+        if self.test_definition.get('buffer_size'):
+            cmd.add('--len', self.test_definition.get('buffer_size'))
         if self.test_definition.get('udp'):
             cmd.add('--udp')
             if self.test_definition.get('bandwidth'):
                 cmd.add('--bandwidth', self.test_definition.get('bandwidth'))
+            if self.test_definition.get('datagram_size'):
+                cmd.add('--len', self.test_definition.get('datagram_size'))
         cmd.add('--time', self.get_expected_duration())
         cmd.add('--parallel', self.test_definition.get('threads') or 1)
         if self.test_definition.get('csv'):
@@ -58,14 +61,15 @@ class IperfGraphExecutor(IperfExecutor):
         for row in csv.reader(result['stdout'].split('\n')):
             if row and len(row) > 8:
                 thread = row[5]
-                if threads_count > 1 and thread != -1:
+                if threads_count > 1 and thread != '-1':
                     # ignore individual per-thread data
                     continue
 
                 start, end = row[6].split('-')
                 samples.append([float(end), int(row[8])])
 
-        samples.pop()  # the last line is summary, remove it
+        if samples:
+            samples.pop()  # the last line is summary, remove it
 
         result['samples'] = samples
         result['meta'] = [['time', 's'], ['bandwidth', 'bps']]

shaker/engine/quorum.py

@@ -15,6 +15,7 @@
 
 import multiprocessing
 import time
+import traceback
 
 from oslo_log import log as logging
 
@@ -46,7 +47,8 @@ class BaseOperation(object):
         return {'status': 'ok'}
 
     def process_error(self, agent_id, exception):
-        return {'status': 'error', 'info': str(exception)}
+        return {'status': 'error', 'info': str(exception),
+                'traceback': traceback.format_exc()}
 
     def process_failure(self, agent_id):
         return {'status': 'lost'}

tests/test_iperf_graph_executor.py

@@ -29,17 +29,18 @@ class TestIperfGraphExecutor(testtools.TestCase):
         executor = iperf.IperfGraphExecutor({}, AGENT)
 
         expected = {'data': ('sudo nice -n -20 iperf --client %s --format m '
-                             '--nodelay --len 8k --time 60 --parallel 1 '
+                             '--nodelay --time 60 --parallel 1 '
                              '--reportstyle C --interval 1') % IP,
                     'type': 'program'}
         self.assertEqual(expected, executor.get_command())
 
     def test_get_command_udp(self):
         executor = iperf.IperfGraphExecutor(
-            {'udp': True, 'bandwidth': '100M', 'time': 30}, AGENT)
+            {'udp': True, 'bandwidth': '100M', 'time': 30,
+             'datagram_size': 1470}, AGENT)
 
         expected = {'data': ('sudo nice -n -20 iperf --client %s --format m '
-                             '--nodelay --len 8k --udp --bandwidth 100M '
+                             '--nodelay --udp --bandwidth 100M --len 1470 '
                              '--time 30 --parallel 1 '
                              '--reportstyle C --interval 1') % IP,
                     'type': 'program'}
@@ -71,6 +72,36 @@ class TestIperfGraphExecutor(testtools.TestCase):
         self.assertEqual(expected['meta'], reply['meta'],
                          message='Metadata')
 
+    def test_process_reply_multiple_threads(self):
+        executor = iperf.IperfGraphExecutor({'threads': 2}, AGENT)
+        message = {
+            'stdout': """
+20150610102341,10.0.0.3,53479,10.0.0.2,5001,4,0.0-1.0,30277632,242221056
+20150610102341,10.0.0.3,53478,10.0.0.2,5001,3,0.0-1.0,23461888,187695104
+20150610102341,10.0.0.3,0,10.0.0.2,5001,-1,0.0-1.0,53739520,429916160
+20150610102342,10.0.0.3,53479,10.0.0.2,5001,4,1.0-2.0,41418752,331350016
+20150610102342,10.0.0.3,53478,10.0.0.2,5001,3,1.0-2.0,22806528,182452224
+20150610102342,10.0.0.3,53478,10.0.0.2,5001,3,0.0-2.0,46268416,370147328
+20150610102342,10.0.0.3,0,10.0.0.2,5001,-1,1.0-2.0,64225280,513802240
+20150610102440,10.0.0.3,53479,10.0.0.2,5001,4,0.0-2.0,71696384,573571072
+20150610102440,10.0.0.3,0,10.0.0.2,5001,-1,0.0-2.0,117964800,479810974\n
+"""
+        }
+        expected = {
+            'samples': [
+                [1.0, 429916160],
+                [2.0, 513802240],
+            ],
+            'meta': [
+                ['time', 's'], ['bandwidth', 'bps']
+            ]
+        }
+        reply = executor.process_reply(message)
+        self.assertEqual(expected['samples'], reply['samples'],
+                         message='Samples data')
+        self.assertEqual(expected['meta'], reply['meta'],
+                         message='Metadata')
+
     def test_process_empty_reply(self):
         executor = iperf.IperfGraphExecutor({}, AGENT)
         message = {

tests/test_operations.py

@@ -62,7 +62,7 @@ class TestOperations(testtools.TestCase):
             'info': 'Error!'
         }
         executor.process_reply.assert_called_once_with(message)
-        self.assertEqual(expected, reply)
+        self.assertDictContainsSubset(expected, reply)
 
     def test_execute_operation_process_reply_with_unhandled_exception(self):
         executor = mock.MagicMock()
@@ -80,4 +80,4 @@ class TestOperations(testtools.TestCase):
             'info': 'Boom!'
         }
         executor.process_reply.assert_called_once_with(message)
-        self.assertEqual(expected, reply)
+        self.assertDictContainsSubset(expected, reply)