parse properly iperf3 output when --json-stream is used

When --json-stream is used, its raw output cannot be simply parsed
with json.loads(raw), but it needs to be parsed line by line.

Change-Id: Ice8caaa8f62bb10feeee6a49d15e7583516c4422

tobiko/podified/_openshift.py

@@ -471,15 +471,7 @@ def _store_iperf3_client_results(
     # in the file when "--logfile" option is used in ipef3
     # So to be able to validate them in the same way, logs from
     # stdout of the Pod needs to be converted
-    iperf3_results_data: dict = {
-        "intervals": []
-    }
-    for log_line in raw_pod_logs.split("\n"):
-        log_line_json = json.loads(log_line)
-        if log_line_json.get('event') != 'interval':
-            continue
-        iperf3_results_data["intervals"].append(
-            log_line_json["data"])
+    iperf3_results_data = iperf3.parse_json_stream_output(raw_pod_logs)
 
     logfile = iperf3.get_iperf3_logs_filepath(address, output_dir)
     with open(logfile, "w") as f:

tobiko/shell/iperf3/__init__.py

@@ -29,6 +29,7 @@ get_iperf3_logs_filepath = _execute.get_iperf3_logs_filepath
 iperf3_client_alive = _execute.iperf3_client_alive
 stop_iperf3_client = _execute.stop_iperf3_client
 start_iperf3_server = _execute.start_iperf3_server
+parse_json_stream_output = _execute.parse_json_stream_output
 
 get_iperf3_client_command = _interface.get_iperf3_client_command
 

tobiko/shell/iperf3/_execute.py

@@ -225,6 +225,25 @@ def _get_iperf3_log_raw(logfile: str,
             tobiko.fail('Failed empty iperf file.')
 
 
+def parse_json_stream_output(iperf_log_raw):
+    # Logs are printed by iperf3 client to the stdout or file in json
+    # format, but the format is different than what is stored
+    # without "--json-stream" option
+    # So to be able to validate them in the same way, logs
+    # need to be converted
+    iperf3_results_data: dict = {
+        "intervals": []
+    }
+    for log_line in iperf_log_raw.splitlines():
+        log_line_json = json.loads(log_line)
+        if log_line_json.get('event') != 'interval':
+            continue
+        iperf3_results_data["intervals"].append(
+            log_line_json["data"])
+
+    return iperf3_results_data
+
+
 def check_iperf3_client_results(address: typing.Union[str, netaddr.IPAddress],
                                 output_dir: str = 'tobiko_iperf_results',
                                 ssh_client: ssh.SSHClientType = None,
@@ -236,7 +255,10 @@ def check_iperf3_client_results(address: typing.Union[str, netaddr.IPAddress],
                   'disabled')
         return
 
-    iperf_log = json.loads(iperf_log_raw)
+    try:
+        iperf_log = json.loads(iperf_log_raw)
+    except json.JSONDecodeError:
+        iperf_log = parse_json_stream_output(iperf_log_raw)
     longest_break = 0  # seconds
     breaks_total = 0  # seconds
     current_break = 0  # seconds