Implement streaming json output (esnet#1098)

Currently when enabling json output, the results are only written once the test
concludes. This is done to output one full json document containing all relevant
informations.

To allow status output during the run while using json as output, this patch
adds a newline-delimited JSON output.

In order to achive this multiple event objects are emitted. These are serialized
as json and printed with a newline seperating them.

Each event contains a event name and its data. The following events have been
introduced: start, interval, end, error, server_output_text and
server_output_json. The data contains the relevant portion of the normal JSON
output.

Co-authored-by: swlars <89053414+swlars@users.noreply.github.com>

src/iperf.h

@@ -344,6 +344,7 @@ struct iperf_test
     int       bidirectional;                    /* --bidirectional */
     int	      verbose;                          /* -V option - verbose mode */
     int	      json_output;                      /* -J option - JSON output */
+    int	      json_stream;                      /* --json-stream */
     int	      zerocopy;                         /* -Z option - use sendfile */
     int       debug;				/* -d option - enable debug */
     enum      debug_level debug_level;          /* -d option option - level of debug messages to show */

src/iperf3.1

@@ -96,9 +96,11 @@ test by specifying the --get-server-output flag.
 Either the client or the server can produce its output in a JSON
 structure, useful for integration with other programs, by passing it
 the -J flag.
-Because the contents of the JSON structure are only completely known
+Normally the contents of the JSON structure are only competely known
 after the test has finished, no JSON output will be emitted until the
 end of the test.
+By enabling line-delimited JSON multiple objects will be emitted to
+provide a real-time parsable JSON output.
 .PP
 iperf3 has a (overly) large set of command-line options that can be
 used to set the parameters of a test.
@@ -157,6 +159,9 @@ give more detailed output
 .BR -J ", " --json " "
 output in JSON format
 .TP
+.BR --json-stream " "
+output in line-delimited JSON format
+.TP
 .BR --logfile " \fIfile\fR"
 send output to a log file.
 .TP

src/iperf_api.c

@@ -102,6 +102,7 @@ static int diskfile_recv(struct iperf_stream *sp);
 static int JSON_write(int fd, cJSON *json);
 static void print_interval_results(struct iperf_test *test, struct iperf_stream *sp, cJSON *json_interval_streams);
 static cJSON *JSON_read(int fd);
+static int JSONStream_Output(struct iperf_test *test, const char* event_name, cJSON* obj);
 
 
 /*************************** Print usage functions ****************************/
@@ -326,6 +327,12 @@ iperf_get_test_json_output_string(struct iperf_test *ipt)
     return ipt->json_output_string;
 }
 
+int
+iperf_get_test_json_stream(struct iperf_test *ipt)
+{
+    return ipt->json_stream;
+}
+
 int
 iperf_get_test_zerocopy(struct iperf_test *ipt)
 {
@@ -682,6 +689,12 @@ iperf_set_test_json_output(struct iperf_test *ipt, int json_output)
     ipt->json_output = json_output;
 }
 
+void
+iperf_set_test_json_stream(struct iperf_test *ipt, int json_stream)
+{
+    ipt->json_stream = json_stream;
+}
+
 int
 iperf_has_zerocopy( void )
 {
@@ -892,8 +905,12 @@ iperf_on_test_start(struct iperf_test *test)
 		iperf_printf(test, test_start_time, test->protocol->name, test->num_streams, test->settings->blksize, test->omit, test->duration, test->settings->tos);
 	}
     }
+    if (test->json_stream) {
+        JSONStream_Output(test, "start", test->json_start);
+    }
 }
 
+
 /* This converts an IPv6 string address from IPv4-mapped format into regular
 ** old IPv4 format, which is easier on the eyes of network veterans.
 **
@@ -1058,6 +1075,7 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
         {"one-off", no_argument, NULL, '1'},
         {"verbose", no_argument, NULL, 'V'},
         {"json", no_argument, NULL, 'J'},
+        {"json-stream", no_argument, NULL, OPT_JSON_STREAM},
         {"version", no_argument, NULL, 'v'},
         {"server", no_argument, NULL, 's'},
         {"client", required_argument, NULL, 'c'},
@@ -1207,6 +1225,10 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
             case 'J':
                 test->json_output = 1;
                 break;
+            case OPT_JSON_STREAM:
+                test->json_output = 1;
+                test->json_stream = 1;
+                break;
             case 'v':
                 printf("%s (cJSON %s)\n%s\n%s\n", version, cJSON_Version(), get_system_info(),
 		       get_optional_features());
@@ -2736,6 +2758,29 @@ JSON_read(int fd)
     return json;
 }
 
+/*************************************************************/
+/**
+ * JSONStream_Output - outputs an obj as event without distrubing it
+ */
+
+static int
+JSONStream_Output(struct iperf_test * test, const char * event_name, cJSON * obj)
+{
+    cJSON *event = cJSON_CreateObject();
+    if (!event)
+    return -1;
+    cJSON_AddStringToObject(event, "event", event_name);
+    cJSON_AddItemReferenceToObject(event, "data", obj);
+    char *str = cJSON_PrintUnformatted(event);
+    if (str == NULL)
+    return -1;
+    fprintf(test->outfile, "%s\n", str);
+    iflush(test);
+    cJSON_free(str);
+    cJSON_Delete(event);
+    return 0;
+}
+
 /*************************************************************/
 /**
  * add_to_interval_list -- adds new interval to the interval_list
@@ -3400,6 +3445,7 @@ iperf_print_intermediate(struct iperf_test *test)
 
     int lower_mode, upper_mode;
     int current_mode;
+    int discard_json;
 
     /*
      * Due to timing oddities, there can be cases, especially on the
@@ -3445,11 +3491,20 @@ iperf_print_intermediate(struct iperf_test *test)
 	return;
     }
 
+    /*
+     * When we use streamed json, we don't actually need to keep the interval
+     * results around unless we're the server and the client requested the server output.
+     *
+     * This avoids unneeded memory build up for long sessions.
+     */
+    discard_json = test->json_stream == 1 && !(test->role == 's' && test->get_server_output);
+
     if (test->json_output) {
         json_interval = cJSON_CreateObject();
 	if (json_interval == NULL)
 	    return;
-	cJSON_AddItemToArray(test->json_intervals, json_interval);
+        if (!discard_json)
+	    cJSON_AddItemToArray(test->json_intervals, json_interval);
         json_interval_streams = cJSON_CreateArray();
 	if (json_interval_streams == NULL)
 	    return;
@@ -3600,6 +3655,11 @@ iperf_print_intermediate(struct iperf_test *test)
             }
         }
     }
+
+    if (test->json_stream)
+        JSONStream_Output(test, "interval", json_interval);
+    if (discard_json)
+        cJSON_Delete(json_interval);
 }
 
 /**
@@ -4830,7 +4890,35 @@ iperf_json_finish(struct iperf_test *test)
         cJSON_Delete(test->json_top);
         test->json_top = NULL;
     }
-    test->json_start = test->json_connected = test->json_intervals = test->json_server_output = test->json_end = NULL;
+    // Get ASCII rendering of JSON structure.  Then make our
+    // own copy of it and return the storage that cJSON allocated
+    // on our behalf.  We keep our own copy around.
+    char *str = cJSON_Print(test->json_top);
+    if (str == NULL)
+	return -1;
+    test->json_output_string = strdup(str);
+    cJSON_free(str);
+    if (test->json_output_string == NULL)
+        return -1;
+    if (test->json_stream) {
+        cJSON *error = cJSON_GetObjectItem(test->json_top, "error");
+        if (error) {
+            JSONStream_Output(test, "error", error);
+        }
+        if (test->json_server_output) {
+            JSONStream_Output(test, "server_output_json", test->json_server_output);
+        }
+        if (test->server_output_text) {
+            JSONStream_Output(test, "server_output_text", cJSON_CreateString(test->server_output_text));
+        }
+        JSONStream_Output(test, "end", test->json_end);
+    }
+    else {
+        fprintf(test->outfile, "%s\n", test->json_output_string);
+        iflush(test);
+    }
+    cJSON_Delete(test->json_top);
+    test->json_top = test->json_start = test->json_connected = test->json_intervals = test->json_server_output = test->json_end = NULL;
     return 0;
 }
 

src/iperf_api.h

@@ -100,7 +100,8 @@ typedef atomic_uint_fast64_t atomic_iperf_size_t;
 #define OPT_IDLE_TIMEOUT 25
 #define OPT_DONT_FRAGMENT 26
 #define OPT_RCV_TIMEOUT 27
-#define OPT_SND_TIMEOUT 28
+#define OPT_JSON_STREAM 28
+#define OPT_SND_TIMEOUT 29
 
 /* states */
 #define TEST_START 1
@@ -151,6 +152,7 @@ char*	iperf_get_test_template( struct iperf_test* ipt );
 int	iperf_get_test_protocol_id( struct iperf_test* ipt );
 int	iperf_get_test_json_output( struct iperf_test* ipt );
 char*	iperf_get_test_json_output_string ( struct iperf_test* ipt );
+int	iperf_get_test_json_stream( struct iperf_test* ipt );
 int	iperf_get_test_zerocopy( struct iperf_test* ipt );
 int	iperf_get_test_get_server_output( struct iperf_test* ipt );
 char	iperf_get_test_unit_format(struct iperf_test *ipt);
@@ -195,6 +197,7 @@ void	iperf_set_test_server_hostname( struct iperf_test* ipt, const char* server_
 void    iperf_set_test_template( struct iperf_test *ipt, const char *tmp_template );
 void	iperf_set_test_reverse( struct iperf_test* ipt, int reverse );
 void	iperf_set_test_json_output( struct iperf_test* ipt, int json_output );
+void	iperf_set_test_json_stream( struct iperf_test* ipt, int json_stream );
 int	iperf_has_zerocopy( void );
 void	iperf_set_test_zerocopy( struct iperf_test* ipt, int zerocopy );
 void	iperf_set_test_get_server_output( struct iperf_test* ipt, int get_server_output );

src/iperf_locale.c

@@ -126,6 +126,7 @@ const char usage_longstr[] = "Usage: iperf3 [-s|-c host] [options]\n"
 #endif /* HAVE_SO_BINDTODEVICE */
                            "  -V, --verbose             more detailed output\n"
                            "  -J, --json                output in JSON format\n"
+                           "  --json-stream             output in line-delimited JSON format\n"
                            "  --logfile f               send output to a log file\n"
                            "  --forceflush              force flushing output at every interval\n"
                            "  --timestamps<=format>     emit a timestamp at the start of each output line\n"

src/libiperf.3

@@ -32,6 +32,7 @@ Setting test parameters:
     void iperf_set_test_blksize( struct iperf_test *t, int blksize );
     void iperf_set_test_num_streams( struct iperf_test *t, int num_streams );
     void iperf_set_test_json_output( struct iperf_test *t, int json_output );
+    void iperf_set_test_json_stream( struct iperf_test *t, int json_stream );
     int iperf_has_zerocopy( void );
     void iperf_set_test_zerocopy( struct iperf_test* t, int zerocopy );
     void iperf_set_test_tos( struct iperf_test* t, int tos );