ACLK statistics on bytes recvd and sent (#13091)

3 files changed, 33 insertions, 0 deletions

diff --git a/aclk/aclk.c b/aclk/aclk.c
index 6426c5b5e7..9593ff5564 100644
--- a/aclk/aclk.c
+++ b/aclk/aclk.c
@@ -822,6 +822,7 @@ void *aclk_main(void *ptr)
         stats_thread = callocz(1, sizeof(struct aclk_stats_thread));
         stats_thread->thread = mallocz(sizeof(netdata_thread_t));
         stats_thread->query_thread_count = query_threads.count;
+        stats_thread->client = mqttwss_client;
         aclk_stats_thread_prepare(query_threads.count, proto_hdl_cnt);
         netdata_thread_create(
             stats_thread->thread, ACLK_STATS_THREAD_NAME, NETDATA_THREAD_OPTION_JOINABLE, aclk_stats_main_thread,
diff --git a/aclk/aclk_stats.c b/aclk/aclk_stats.c
index ca05326386..b6e8a673eb 100644
--- a/aclk/aclk_stats.c
+++ b/aclk/aclk_stats.c
@@ -261,6 +261,33 @@ static void aclk_stats_newproto_rx(uint32_t *rx_msgs_sample)
 }
 #endif
 
+static void aclk_stats_mqtt_wss(struct mqtt_wss_stats *stats)
+{
+    static RRDSET *st = NULL;
+    static RRDDIM *rd_sent = NULL;
+    static RRDDIM *rd_recvd = NULL;
+    static uint64_t sent = 0;
+    static uint64_t recvd = 0;
+
+    sent += stats->bytes_tx;
+    recvd += stats->bytes_rx;
+
+    if (unlikely(!st)) {
+        st = rrdset_create_localhost(
+            "netdata", "aclk_openssl_bytes", NULL, "aclk", NULL, "Received and Sent bytes.", "B/s",
+            "netdata", "stats", 200011, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
+
+        rd_sent  = rrddim_add(st, "sent", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
+        rd_recvd = rrddim_add(st, "received", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
+    } else
+        rrdset_next(st);
+
+    rrddim_set_by_pointer(st, rd_sent, sent);
+    rrddim_set_by_pointer(st, rd_recvd, recvd);
+
+    rrdset_done(st);
+}
+
 void aclk_stats_thread_prepare(int query_thread_count, unsigned int proto_hdl_cnt)
 {
 #ifndef ENABLE_NEW_CLOUD_PROTOCOL
@@ -343,6 +370,9 @@ void *aclk_stats_main_thread(void *ptr)
 
         aclk_stats_query_time(&per_sample);
 
+        struct mqtt_wss_stats mqtt_wss_stats = mqtt_wss_get_stats(args->client);
+        aclk_stats_mqtt_wss(&mqtt_wss_stats);
+
 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
         aclk_stats_newproto_rx(aclk_stats_cfg.aclk_proto_rx_msgs_sample);
 #endif
diff --git a/aclk/aclk_stats.h b/aclk/aclk_stats.h
index 4f2894798f..aec13e212d 100644
--- a/aclk/aclk_stats.h
+++ b/aclk/aclk_stats.h
@@ -6,6 +6,7 @@
 #include "daemon/common.h"
 #include "libnetdata/libnetdata.h"
 #include "aclk_query_queue.h"
+#include "mqtt_wss_client.h"
 
 #define ACLK_STATS_THREAD_NAME "ACLK_Stats"
 
@@ -22,6 +23,7 @@ int aclk_cloud_req_http_type_to_idx(const char *name);
 struct aclk_stats_thread {
     netdata_thread_t *thread;
     int query_thread_count;
+    mqtt_wss_client client;
 };
 
 // preserve between samples