Merge pull request stevenj#1 from kleinweby/output-json-stream

Implement streaming json output

Author: stevenj

src/iperf.h

@@ -309,6 +309,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 */
     int	      get_server_output;		/* --get-server-output */

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 competely 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.
@@ -153,6 +155,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 ****************************/
@@ -314,6 +315,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)
 {
@@ -622,6 +629,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 )
 {
@@ -827,8 +840,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.
 **
@@ -938,6 +955,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'},
@@ -1084,6 +1102,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());
@@ -2495,6 +2517,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
@@ -3131,6 +3176,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
@@ -3176,11 +3222,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;
@@ -3313,6 +3368,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);
 }
 
 /**
@@ -4443,8 +4503,23 @@ iperf_json_finish(struct iperf_test *test)
     cJSON_free(str);
     if (test->json_output_string == NULL)
         return -1;
-    fprintf(test->outfile, "%s\n", test->json_output_string);
-    iflush(test);
+    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

@@ -87,6 +87,7 @@ typedef uint64_t iperf_size_t;
 #define OPT_IDLE_TIMEOUT 25
 #define OPT_DONT_FRAGMENT 26
 #define OPT_RCV_TIMEOUT 27
+#define OPT_JSON_STREAM 28
 
 /* states */
 #define TEST_START 1
@@ -135,6 +136,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_bind_address ( struct iperf_test* ipt );
@@ -174,6 +176,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

@@ -111,6 +111,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 );