kxdpgun: -j/--json output option

4 files changed, 227 insertions, 39 deletions

diff --git a/src/utils/kxdpgun/main.c b/src/utils/kxdpgun/main.c
index 93960179d..56edd7db5 100644
--- a/src/utils/kxdpgun/main.c
+++ b/src/utils/kxdpgun/main.c
@@ -75,8 +75,23 @@ const static xdp_gun_ctx_t ctx_defaults = {
 	.target_port = 0,
 	.flags = KNOT_XDP_FILTER_UDP | KNOT_XDP_FILTER_PASS,
 	.xdp_config = { .ring_size = 2048 },
+	.jw = NULL,
 };
 
+static uint64_t us_timestamp(void)
+{
+	struct timespec ts;
+	clock_gettime(CLOCK_REALTIME, &ts);
+	return ((uint64_t)ts.tv_sec * 1000000) + (ts.tv_nsec / 1000);
+}
+
+static uint64_t ns_timestamp(void)
+{
+	struct timespec ts;
+	clock_gettime(CLOCK_REALTIME, &ts);
+	return ((uint64_t)ts.tv_sec * 1000000000) + ts.tv_nsec;
+}
+
 static void sigterm_handler(int signo)
 {
 	assert(signo == SIGTERM || signo == SIGINT);
@@ -91,20 +106,6 @@ static void sigusr_handler(int signo)
 	}
 }
 
-inline static void timer_start(struct timespec *timesp)
-{
-	clock_gettime(CLOCK_MONOTONIC, timesp);
-}
-
-inline static uint64_t timer_end(struct timespec *timesp)
-{
-	struct timespec end;
-	clock_gettime(CLOCK_MONOTONIC, &end);
-	uint64_t res = (end.tv_sec - timesp->tv_sec) * (uint64_t)1000000;
-	res += ((int64_t)end.tv_nsec - timesp->tv_nsec) / 1000;
-	return res;
-}
-
 static unsigned addr_bits(bool ipv6)
 {
 	return ipv6 ? 128 : 32;
@@ -323,10 +324,9 @@ void *xdp_gun_thread(void *_ctx)
 {
 	xdp_gun_ctx_t *ctx = _ctx;
 	struct knot_xdp_socket *xsk = NULL;
-	struct timespec timer;
 	knot_xdp_msg_t pkts[ctx->at_once];
 	uint64_t duration = 0;
-	kxdpgun_stats_t local_stats = { 0 };
+	kxdpgun_stats_t local_stats = { 0 }; // cumulative stats of past periods excluding the current
 	unsigned stats_triggered = 0;
 	knot_tcp_table_t *tcp_table = NULL;
 #ifdef ENABLE_QUIC
@@ -382,7 +382,7 @@ void *xdp_gun_thread(void *_ctx)
 	}
 
 	if (ctx->thread_id == 0) {
-		print_stats_header(ctx);
+		STATS_HDR(ctx);
 	}
 
 	struct pollfd pfd = { knot_xdp_socket_fd(xsk), POLLIN, 0 };
@@ -428,7 +428,8 @@ void *xdp_gun_thread(void *_ctx)
 	size_t local_ports_it = 0;
 #endif // ENABLE_QUIC
 
-	timer_start(&timer);
+	local_stats.since = ns_timestamp();
+	ctx->stats_start_us = local_stats.since / 1000;
 
 	while (duration < ctx->duration + extra_wait) {
 		// sending part
@@ -728,7 +729,8 @@ void *xdp_gun_thread(void *_ctx)
 
 		// speed and signal part
 		uint64_t dura_exp = (local_stats.qry_sent * 1000000) / ctx->qps;
-		duration = timer_end(&timer);
+		uint64_t now_ns = ns_timestamp();
+		duration = (now_ns - local_stats.since) / 1000;
 		if (xdp_trigger == KXDPGUN_STOP && ctx->duration > duration) {
 			ctx->duration = duration;
 		}
@@ -736,11 +738,12 @@ void *xdp_gun_thread(void *_ctx)
 		if (tmp_stats_trigger > stats_triggered) {
 			stats_triggered = tmp_stats_trigger;
 
-			local_stats.duration = duration;
+			local_stats.until = now_ns;
 			size_t collected = collect_stats(&global_stats, &local_stats);
+
 			assert(collected <= ctx->n_threads);
 			if (collected == ctx->n_threads) {
-				print_stats(&global_stats, ctx);
+				STATS_FMT(ctx, &global_stats, STATS_SUM);
 				clear_stats(&global_stats);
 			}
 		}
@@ -752,10 +755,10 @@ void *xdp_gun_thread(void *_ctx)
 		}
 		tick++;
 	}
+	local_stats.until = ns_timestamp() - extra_wait * 1000;
 
-	print_thrd_summary(ctx, &local_stats);
+	STATS_THRD(ctx, &local_stats);
 
-	local_stats.duration = ctx->duration;
 	collect_stats(&global_stats, &local_stats);
 
 cleanup:
@@ -974,6 +977,7 @@ static void print_help(void)
 	       " -e, --edns-size <size>   "SPACE"EDNS UDP payload size, range 512-4096 (default 1232)\n"
 	       " -m, --mode <mode>        "SPACE"Set XDP mode (auto, copy, generic).\n"
 	       " -G, --qlog <path>        "SPACE"Output directory for qlog (useful for QUIC only).\n"
+	       " -j, --json               "SPACE"Output statistics in json.\n"
 	       " -h, --help               "SPACE"Print the program help.\n"
 	       " -V, --version            "SPACE"Print the program version.\n"
 	       "\n"
@@ -1074,7 +1078,7 @@ static int set_mode(const char *arg, knot_xdp_config_t *config)
 
 static bool get_opts(int argc, char *argv[], xdp_gun_ctx_t *ctx)
 {
-	const char *opts_str = "hV::t:Q:b:rp:T::U::F:I:i:Bl:L:R:v:e:m:G:";
+	const char *opts_str = "hV::t:Q:b:rp:T::U::F:I:i:Bl:L:R:v:e:m:G:j";
 	struct option opts[] = {
 		{ "help",       no_argument,       NULL, 'h' },
 		{ "version",    optional_argument, NULL, 'V' },
@@ -1096,6 +1100,7 @@ static bool get_opts(int argc, char *argv[], xdp_gun_ctx_t *ctx)
 		{ "edns-size",  required_argument, NULL, 'e' },
 		{ "mode",       required_argument, NULL, 'm' },
 		{ "qlog",       required_argument, NULL, 'G' },
+		{ "json",       no_argument,       NULL, 'j' },
 		{ 0 }
 	};
 
@@ -1255,6 +1260,12 @@ static bool get_opts(int argc, char *argv[], xdp_gun_ctx_t *ctx)
 		case 'G':
 			ctx->qlog_dir = optarg;
 			break;
+		case 'j':
+			if ((ctx->jw = jsonw_new(stdout, JSON_INDENT)) == NULL) {
+				ERR2("failed to use JSON");
+				return false;
+			}
+			break;
 		default:
 			print_help();
 			return false;
@@ -1296,12 +1307,18 @@ int main(int argc, char *argv[])
 
 	xdp_gun_ctx_t ctx = ctx_defaults, *thread_ctxs = NULL;
 	ctx.msgid = time(NULL) % UINT16_MAX;
+	ctx.runid = us_timestamp();
+	ctx.argv = argv;
 	pthread_t *threads = NULL;
 
 	if (!get_opts(argc, argv, &ctx)) {
 		goto err;
 	}
 
+	if (JSON_MODE(ctx)) {
+		jsonw_list(ctx.jw, NULL); // wrap the json in a list, for syntactic correctness
+	}
+
 	thread_ctxs = calloc(ctx.n_threads, sizeof(*thread_ctxs));
 	threads = calloc(ctx.n_threads, sizeof(*threads));
 	if (thread_ctxs == NULL || threads == NULL) {
@@ -1346,15 +1363,14 @@ int main(int argc, char *argv[])
 		usleep(20000);
 	}
 	usleep(1000000);
-
 	xdp_trigger = KXDPGUN_START;
 	usleep(1000000);
 
 	for (size_t i = 0; i < ctx.n_threads; i++) {
 		pthread_join(threads[i], NULL);
 	}
-	if (global_stats.duration > 0 && global_stats.qry_sent > 0) {
-		print_stats(&global_stats, &ctx);
+	if (DURATION_US(global_stats) > 0 && global_stats.qry_sent > 0) {
+		STATS_FMT(&ctx, &global_stats, STATS_SUM);
 	}
 	pthread_mutex_destroy(&global_stats.mutex);
 
@@ -1365,5 +1381,9 @@ err:
 	free(thread_ctxs);
 	free(threads);
 	free_global_payloads();
+	if (JSON_MODE(ctx)) {
+		jsonw_end(ctx.jw);
+		jsonw_free(&ctx.jw);
+	}
 	return ecode;
 }
diff --git a/src/utils/kxdpgun/main.h b/src/utils/kxdpgun/main.h
index a738b161d..d87aee880 100644
--- a/src/utils/kxdpgun/main.h
+++ b/src/utils/kxdpgun/main.h
@@ -20,6 +20,7 @@
 #include <netinet/in.h>
 #include <stdbool.h>
 
+#include "contrib/json.h"
 #include "libknot/xdp/eth.h"
 #include "libknot/xdp/tcp.h"
 
@@ -59,6 +60,9 @@ typedef struct xdp_gun_ctx {
 	};
 	char                   dev[IFNAMSIZ];
 	uint64_t               qps, duration;
+	uint64_t               runid;
+	uint64_t               stats_start_us;
+	uint32_t               stats_period; // 0 means no periodic stats
 	unsigned               at_once;
 	uint16_t               msgid;
 	uint16_t               edns_size;
@@ -77,5 +81,7 @@ typedef struct xdp_gun_ctx {
 	knot_xdp_filter_flag_t flags;
 	unsigned               n_threads, thread_id;
 	knot_eth_rss_conf_t    *rss_conf;
+	jsonw_t                *jw;
+	char                   **argv;
 	knot_xdp_config_t      xdp_config;
 } xdp_gun_ctx_t;
diff --git a/src/utils/kxdpgun/stats.c b/src/utils/kxdpgun/stats.c
index 1f217360a..61be48e32 100644
--- a/src/utils/kxdpgun/stats.c
+++ b/src/utils/kxdpgun/stats.c
@@ -27,10 +27,13 @@
 #include "utils/kxdpgun/main.h"
 #include "utils/kxdpgun/stats.h"
 
+pthread_mutex_t stdout_mtx = PTHREAD_MUTEX_INITIALIZER;
+
 void clear_stats(kxdpgun_stats_t *st)
 {
 	pthread_mutex_lock(&st->mutex);
-	st->duration    = 0;
+	st->since       = 0;
+	st->until       = 0;
 	st->qry_sent    = 0;
 	st->synack_recv = 0;
 	st->ans_recv    = 0;
@@ -48,7 +51,8 @@ void clear_stats(kxdpgun_stats_t *st)
 size_t collect_stats(kxdpgun_stats_t *into, const kxdpgun_stats_t *what)
 {
 	pthread_mutex_lock(&into->mutex);
-	into->duration     = MAX(into->duration, what->duration);
+	into->since        = MAX(into->since, what->since);
+	into->until        = MAX(into->until, what->until);
 	into->qry_sent    += what->qry_sent;
 	into->synack_recv += what->synack_recv;
 	into->ans_recv    += what->ans_recv;
@@ -66,7 +70,7 @@ size_t collect_stats(kxdpgun_stats_t *into, const kxdpgun_stats_t *what)
 	return res;
 }
 
-void print_stats_header(const xdp_gun_ctx_t *ctx)
+void plain_stats_header(const xdp_gun_ctx_t *ctx)
 {
 	INFO2("using interface %s, XDP threads %u, IPv%c/%s%s%s, %s mode", ctx->dev, ctx->n_threads,
 	      (ctx->ipv6 ? '6' : '4'),
@@ -76,7 +80,60 @@ void print_stats_header(const xdp_gun_ctx_t *ctx)
 	      (knot_eth_xdp_mode(if_nametoindex(ctx->dev)) == KNOT_XDP_MODE_FULL ? "native" : "emulated"));
 }
 
-void print_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st)
+/* see:
+ * - https://github.com/DNS-OARC/dns-metrics/blob/main/dns-metrics.schema.json
+ * - https://github.com/DNS-OARC/dns-metrics/issues/16#issuecomment-2139462920
+ */
+void json_stats_header(const xdp_gun_ctx_t *ctx)
+{
+	jsonw_t *w = ctx->jw;
+
+	jsonw_object(w, NULL);
+	{
+		jsonw_ulong(w, "runid", ctx->runid);
+		jsonw_str(w, "type", "header");
+		jsonw_int(w, "schema_version", STATS_SCHEMA_VERSION);
+		jsonw_str(w, "generator", PROGRAM_NAME);
+		jsonw_str(w, "generator_version", PACKAGE_VERSION);
+
+		jsonw_list(w, "generator_params");
+		{
+			for (char **it = ctx->argv; *it != NULL; ++it) {
+				jsonw_str(w, NULL, *it);
+			}
+		}
+		jsonw_end(w);
+
+		jsonw_ulong(w, "time_units_per_sec", 1000000000);
+		if (ctx->stats_period > 0) {
+			jsonw_double(w, "stats_interval", ctx->stats_period / 1000.0);
+		}
+		// TODO: timeout
+
+		// mirror the info given by the plaintext printout
+		jsonw_object(w, "additional_info");
+		{
+			jsonw_str(w, "interface", ctx->dev);
+			jsonw_int(w, "xdp_threads", ctx->n_threads);
+			jsonw_int(w, "ip_version", ctx->ipv6 ? 6 : 4);
+			jsonw_str(w, "transport_layer_proto", ctx->tcp ? "TCP" : (ctx->quic ? "QUIC" : "UDP"));
+			jsonw_object(w, "mode_info");
+			{
+				if (ctx->sending_mode[0] != '\0') {
+					jsonw_str(w, "debug", ctx->sending_mode);
+				}
+				jsonw_str(w, "mode", knot_eth_xdp_mode(if_nametoindex(ctx->dev)) == KNOT_XDP_MODE_FULL
+							? "native"
+							: "emulated");
+			}
+			jsonw_end(w);
+		}
+		jsonw_end(w);
+	}
+	jsonw_end(w);
+}
+
+void plain_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st)
 {
 	char recv_str[40] = "", lost_str[40] = "", err_str[40] = "";
 	if (!(ctx->flags & KNOT_XDP_FILTER_DROP)) {
@@ -92,18 +149,41 @@ void print_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st)
 	      ctx->thread_id, st->qry_sent, recv_str, lost_str, err_str);
 }
 
-void print_stats(kxdpgun_stats_t *st, const xdp_gun_ctx_t *ctx)
+void json_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st)
+{
+	pthread_mutex_lock(&stdout_mtx);
+
+	jsonw_t *w = ctx->jw;
+
+	jsonw_object(ctx->jw, NULL);
+	{
+		jsonw_str(w, "type", "thread_summary");
+		jsonw_ulong(w, "runid", ctx->runid);
+		jsonw_ulong(w, "subid", ctx->thread_id);
+		jsonw_ulong(w, "qry_sent", st->qry_sent);
+		jsonw_ulong(w, "ans_recv", st->ans_recv);
+		jsonw_ulong(w, "lost", st->lost);
+		jsonw_ulong(w, "errors", st->errors);
+	}
+	jsonw_end(ctx->jw);
+	pthread_mutex_unlock(&stdout_mtx);
+}
+
+void plain_stats(const xdp_gun_ctx_t *ctx, kxdpgun_stats_t *st, stats_type_t stt)
 {
 	pthread_mutex_lock(&st->mutex);
 
 	bool recv = !(ctx->flags & KNOT_XDP_FILTER_DROP);
+	uint64_t duration = DURATION_US(*st);
+	double rel_start_us = (st->since / 1000.0) - ctx->stats_start_us ;
+	double rel_end_us = rel_start_us + duration;
 
-#define ps(counter)  ((typeof(counter))((counter) * 1000 / ((float)st->duration / 1000)))
+#define ps(counter)  ((typeof(counter))((counter) * 1000 / ((float)duration / 1000)))
 #define pct(counter) ((counter) * 100.0 / st->qry_sent)
 
 	const char *name = ctx->tcp ? "SYNs:    " : ctx->quic ? "initials:" : "queries: ";
 	printf("total %s    %"PRIu64" (%"PRIu64" pps) (%f%%)\n", name, st->qry_sent,
-	       ps(st->qry_sent), 100.0 * st->qry_sent / (st->duration / 1000000.0 * ctx->qps * ctx->n_threads));
+	       ps(st->qry_sent), 100.0 * st->qry_sent / (duration / 1000000.0 * ctx->qps * ctx->n_threads));
 	if (st->qry_sent > 0 && recv) {
 		if (ctx->tcp || ctx->quic) {
 		name = ctx->tcp ? "established:" : "handshakes: ";
@@ -135,7 +215,65 @@ void print_stats(kxdpgun_stats_t *st, const xdp_gun_ctx_t *ctx)
 			}
 		}
 	}
-	printf("duration: %"PRIu64" s\n", (st->duration / (1000 * 1000)));
+	if (stt == STATS_SUM) {
+		printf("duration: %.4f s\n", duration / 1000000.0);
+	} else {
+		printf("since: %.4fs   until: %.4fs\n", rel_start_us / 1000000, rel_end_us / 1000000);
+	}
+
+	pthread_mutex_unlock(&st->mutex);
+}
+
+/* see https://github.com/DNS-OARC/dns-metrics/blob/main/dns-metrics.schema.json
+ * and https://github.com/DNS-OARC/dns-metrics/issues/16#issuecomment-2139462920 */
+void json_stats(const xdp_gun_ctx_t *ctx, kxdpgun_stats_t *st, stats_type_t stt)
+{
+	assert(stt == STATS_PERIODIC || stt == STATS_SUM);
+
+	jsonw_t *w = ctx->jw;
+
+	pthread_mutex_lock(&st->mutex);
+
+	jsonw_object(w, NULL);
+	{
+		jsonw_ulong(w, "runid", ctx->runid);
+		jsonw_str(w, "type", (stt == STATS_PERIODIC) ? "stats_periodic" : "stats_sum");
+		jsonw_ulong(w, "since", st->since);
+		jsonw_ulong(w, "until", st->until);
+		jsonw_ulong(w, "queries", st->qry_sent);
+		jsonw_ulong(w, "responses", st->ans_recv);
+
+		jsonw_object(w, "response_rcodes");
+		{
+			for (size_t i = 0; i < RCODE_MAX; ++i) {
+				if (st->rcodes_recv[i] > 0) {
+					const knot_lookup_t *rc = knot_lookup_by_id(knot_rcode_names, i);
+					jsonw_ulong(w, (rc == NULL) ? "unknown" : rc->name, st->rcodes_recv[i]);
+				}
+			}
+		}
+		jsonw_end(w);
+
+		jsonw_object(w, "conn_info");
+		{
+			jsonw_str(w, "type", ctx->tcp ? "tcp" : (ctx->quic ? "quic_conn" : "udp"));
+
+			// TODO:
+			// packets_sent
+			// packets_recieved
+
+			jsonw_ulong(w, "socket_errors", st->errors);
+			if (ctx->tcp || ctx->quic) {
+				jsonw_ulong(w, "handshakes", st->synack_recv);
+				// TODO: handshakes_failed
+				if (ctx->quic) {
+					// TODO: conn_resumption
+				}
+			}
+		}
+		jsonw_end(w);
+	}
+	jsonw_end(w);
 
 	pthread_mutex_unlock(&st->mutex);
 }
diff --git a/src/utils/kxdpgun/stats.h b/src/utils/kxdpgun/stats.h
index 06461c869..5d0034254 100644
--- a/src/utils/kxdpgun/stats.h
+++ b/src/utils/kxdpgun/stats.h
@@ -25,9 +25,23 @@
 
 #define RCODE_MAX (0x0F + 1)
 
+#define JSON_INDENT		"  "
+#define STATS_SCHEMA_VERSION	20240530
+
+#define DURATION_US(st) (((st).until - (st).since) / 1000)
+#define DURATION_NS(st) ((st).until - (st).since)
+
+#define JSON_MODE(ctx) ((ctx).jw != NULL)
+
+#define STATS_HDR(ctx) ((JSON_MODE(*(ctx)) ? json_stats_header : plain_stats_header)((ctx)))
+#define STATS_THRD(ctx, stats) \
+	((JSON_MODE(*ctx) ? json_thrd_summary : plain_thrd_summary)((ctx), (stats)))
+#define STATS_FMT(ctx, stats, stats_type) \
+	((JSON_MODE(*(ctx)) ? json_stats : plain_stats)((ctx), (stats), (stats_type)))
+
 typedef struct {
 	size_t		collected;
-	uint64_t	duration;
+	uint64_t	since, until; // nanosecs UNIX
 	uint64_t	qry_sent;
 	uint64_t	synack_recv;
 	uint64_t	ans_recv;
@@ -41,11 +55,21 @@ typedef struct {
 	pthread_mutex_t	mutex;
 } kxdpgun_stats_t;
 
+typedef enum {
+	STATS_PERIODIC,
+	STATS_SUM,
+} stats_type_t;
+
 void clear_stats(kxdpgun_stats_t *st);
 size_t collect_stats(kxdpgun_stats_t *into, const kxdpgun_stats_t *what);
 
-void print_stats_header(const xdp_gun_ctx_t *ctx);
+void plain_stats_header(const xdp_gun_ctx_t *ctx);
+void json_stats_header(const xdp_gun_ctx_t *ctx);
+
+void plain_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st);
+void json_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st);
 
-void print_thrd_summary(const xdp_gun_ctx_t *ctx, const kxdpgun_stats_t *st);
+void plain_stats(const xdp_gun_ctx_t *ctx, kxdpgun_stats_t *st, stats_type_t stt);
+void json_stats(const xdp_gun_ctx_t *ctx, kxdpgun_stats_t *st, stats_type_t stt);
 
-void print_stats(kxdpgun_stats_t *st, const xdp_gun_ctx_t *ctx);
+extern pthread_mutex_t stdout_mtx;