TT#44805 add bookkeeping for output stream stats

Change-Id: I353d3785e79e9e3c53154de93f48d7229cf6643e

daemon/call.c

@@ -550,6 +550,7 @@ static void count_stream_stats_kernel(struct packet_stream *ps) {
 	} while (0)
 
 #define DS(x) DS_io(x, ps, &ke->stats_in, in)
+#define DSo(x) DS_io(x, sink, stats_o, out)
 
 void call_timer(void *ptr) {
 	struct iterator_helper hlp;
@@ -609,6 +610,7 @@ void call_timer(void *ptr) {
 		}
 
 		uint64_t diff_packets_in, diff_bytes_in, diff_errors_in;
+		uint64_t diff_packets_out, diff_bytes_out, diff_errors_out;
 
 		DS(packets);
 		DS(bytes);
@@ -663,6 +665,15 @@ void call_timer(void *ptr) {
 					continue;
 
 				struct rtpengine_output_info *o = &ke->outputs[sh->kernel_output_idx];
+				struct rtpengine_stats *stats_o = &ke->stats_out[sh->kernel_output_idx];
+
+				DSo(bytes);
+				DSo(packets);
+				DSo(errors);
+
+				atomic64_set(&sink->kernel_stats_out.bytes, stats_o->bytes);
+				atomic64_set(&sink->kernel_stats_out.packets, stats_o->packets);
+				atomic64_set(&sink->kernel_stats_out.errors, stats_o->errors);
 
 				mutex_lock(&sink->out_lock);
 				for (unsigned int u = 0; u < G_N_ELEMENTS(ke->target.ssrc); u++) {
@@ -1262,6 +1273,8 @@ enum call_stream_state call_stream_state_machine(struct packet_stream *ps) {
 					"\x00\x00\x00\x00");
 			struct stream_fd *sfd = l->data;
 			socket_sendto(&sfd->socket, fake_rtp.s, fake_rtp.len, &ps->endpoint);
+			atomic64_inc(&ps->stats_out.packets);
+			atomic64_add(&ps->stats_out.bytes, fake_rtp.len);
 		}
 		ret = CSS_PIERCE_NAT;
 	}
@@ -3557,8 +3570,9 @@ void call_destroy(struct call *c) {
 				endpoint_t *local_endpoint = packet_stream_local_addr(ps);
 				char *local_addr = sockaddr_print_buf(&local_endpoint->address);
 
-				ilog(LOG_INFO, "--------- Port %15s:%-5u <> %s%15s:%-5u%s%s, SSRC %s%" PRIx32 "%s, "
-						""UINT64F" p, "UINT64F" b, "UINT64F" e, "UINT64F" ts",
+				ilog(LOG_INFO, "--------- Port %15s:%-5u <> %s%15s:%-5u%s%s, SSRC %s%" PRIx32 "%s, in "
+						UINT64F " p, " UINT64F " b, " UINT64F " e, " UINT64F " ts, "
+						"out " UINT64F " p, " UINT64F " b, " UINT64F " e",
 						local_addr,
 						(unsigned int) local_endpoint->port,
 						FMT_M(addr, ps->endpoint.port),
@@ -3567,7 +3581,10 @@ void call_destroy(struct call *c) {
 						atomic64_get(&ps->stats_in.packets),
 						atomic64_get(&ps->stats_in.bytes),
 						atomic64_get(&ps->stats_in.errors),
-						rtpe_now.tv_sec - atomic64_get(&ps->last_packet));
+						rtpe_now.tv_sec - atomic64_get(&ps->last_packet),
+						atomic64_get(&ps->stats_out.packets),
+						atomic64_get(&ps->stats_out.bytes),
+						atomic64_get(&ps->stats_out.errors));
 			}
 		}
 

daemon/call_interfaces.c

@@ -1982,6 +1982,7 @@ stats:
 	if (!PS_ISSET(ps, RTP))
 		s = &totals->totals[1];
 	ng_stats(bencode_dictionary_add_dictionary(dict, "stats"), &ps->stats_in, s);
+	ng_stats(bencode_dictionary_add_dictionary(dict, "stats_out"), &ps->stats_out, NULL);
 }
 
 #define BF_M(k, f) if (MEDIA_ISSET(m, f)) bencode_list_add_string(flags, k)

daemon/dtls.c

@@ -851,6 +851,8 @@ int dtls(struct stream_fd *sfd, const str *s, const endpoint_t *fsin) {
 		if (fsin) {
 			ilogs(srtp, LOG_DEBUG, "Sending DTLS packet");
 			socket_sendto(&sfd->socket, buf, ret, fsin);
+			atomic64_inc(&ps->stats_out.packets);
+			atomic64_add(&ps->stats_out.bytes, ret);
 		}
 	}
 

daemon/media_player.c

@@ -210,6 +210,9 @@ static bool __send_timer_send_1(struct rtp_header *rh, struct packet_stream *sin
 	socket_sendto(&sink_fd->socket,
 			cp->s.s, cp->s.len, &sink->endpoint);
 
+	atomic64_inc(&sink->stats_out.packets);
+	atomic64_add(&sink->stats_out.bytes, cp->s.len);
+
 	log_info_pop();
 
 	return true;

include/call.h

@@ -343,7 +343,9 @@ struct packet_stream {
 	struct jitter_buffer	*jb;		/* RO */
 
 	struct stream_stats	stats_in;
+	struct stream_stats	stats_out;
 	struct stream_stats	kernel_stats_in;
+	struct stream_stats	kernel_stats_out;
 	unsigned char		in_tos_tclass;
 	atomic64		last_packet;
 	GHashTable		*rtp_stats;	/* LOCK: call->master_lock */

kernel-module/xt_RTPENGINE.c

@@ -304,6 +304,7 @@ struct rtpengine_rtp_stats_a {
 struct rtpengine_output {
 	struct rtpengine_output_info	output;
 	struct re_crypto_context	encrypt;
+	struct rtpengine_stats_a	stats_out;
 };
 struct rtpengine_target {
 	atomic_t			refcnt;
@@ -1449,6 +1450,10 @@ static ssize_t proc_blist_read(struct file *f, char __user *b, size_t l, loff_t
 			spin_lock_irqsave(&o->encrypt.lock, flags);
 			opp->outputs[i] = o->output;
 			spin_unlock_irqrestore(&o->encrypt.lock, flags);
+
+			opp->stats_out[i].packets = atomic64_read(&o->stats_out.packets);
+			opp->stats_out[i].bytes = atomic64_read(&o->stats_out.bytes);
+			opp->stats_out[i].errors = atomic64_read(&o->stats_out.errors);
 		}
 	}
 	else
@@ -1685,6 +1690,11 @@ static int proc_list_show(struct seq_file *f, void *v) {
 		proc_list_addr_print(f, "src", &o->output.src_addr);
 		proc_list_addr_print(f, "dst", &o->output.dst_addr);
 
+		seq_printf(f, "      stats: %20llu bytes, %20llu packets, %20llu errors\n",
+			(unsigned long long) atomic64_read(&o->stats_out.bytes),
+			(unsigned long long) atomic64_read(&o->stats_out.packets),
+			(unsigned long long) atomic64_read(&o->stats_out.errors));
+
 		if (o->output.ssrc_subst) {
 			seq_printf(f, " SSRC out:");
 			for (j = 0; j < ARRAY_SIZE(o->output.ssrc_out); j++) {
@@ -4679,8 +4689,14 @@ no_intercept:
 		}
 
 		err = send_proxy_packet(skb2, &o->output.src_addr, &o->output.dst_addr, o->output.tos, par);
-		if (err)
+		if (err) {
 			atomic64_inc(&g->stats_in.errors);
+			atomic64_inc(&o->stats_out.errors);
+		}
+		else {
+			atomic64_inc(&o->stats_out.packets);
+			atomic64_add(rtp2.payload_len + rtp2.header_len, &o->stats_out.bytes);
+		}
 	}
 
 	if (atomic64_read(&g->stats_in.packets)==0)

kernel-module/xt_RTPENGINE.h

@@ -223,6 +223,7 @@ struct rtpengine_list_entry {
 	struct rtpengine_stats		stats_in;
 	struct rtpengine_rtp_stats	rtp_stats[RTPE_NUM_PAYLOAD_TYPES]; // same index as pt_input
 	struct rtpengine_output_info	outputs[RTPE_MAX_FORWARD_DESTINATIONS];
+	struct rtpengine_stats		stats_out[RTPE_MAX_FORWARD_DESTINATIONS];
 };