add ACLK stats per cloud query type (#10602)

* add stats per cloud query type

5 files changed, 117 insertions, 14 deletions

diff --git a/aclk/legacy/aclk_query.c b/aclk/legacy/aclk_query.c
index 1150613a14..efcae58ce6 100644
--- a/aclk/legacy/aclk_query.c
+++ b/aclk/legacy/aclk_query.c
@@ -62,6 +62,7 @@ static void aclk_query_free(struct aclk_query *this_query)
         freez(this_query->query);
     if(this_query->data && this_query->cmd == ACLK_CMD_CLOUD_QUERY_2) {
         struct aclk_cloud_req_v2 *del = (struct aclk_cloud_req_v2 *)this_query->data;
+        freez(del->query_endpoint);
         freez(del->data);
         freez(del);
     }
diff --git a/aclk/legacy/aclk_query.h b/aclk/legacy/aclk_query.h
index a0a6a5c666..026985c8d2 100644
--- a/aclk/legacy/aclk_query.h
+++ b/aclk/legacy/aclk_query.h
@@ -31,6 +31,7 @@ struct aclk_query_threads {
 struct aclk_cloud_req_v2 {
     char *data;
     RRDHOST *host;
+    char *query_endpoint;
 };
 
 void *aclk_query_main_thread(void *ptr);
diff --git a/aclk/legacy/aclk_rx_msgs.c b/aclk/legacy/aclk_rx_msgs.c
index 99fa9d9870..2681445b4d 100644
--- a/aclk/legacy/aclk_rx_msgs.c
+++ b/aclk/legacy/aclk_rx_msgs.c
@@ -25,7 +25,7 @@ static inline int aclk_extract_v2_data(char *payload, char **data)
 #define STRNCMP_CONSTANT_PREFIX(str, const_pref) strncmp(str, const_pref, strlen(const_pref))
 static inline int aclk_v2_payload_get_query(struct aclk_cloud_req_v2 *cloud_req, struct aclk_request *req)
 {
-    const char *start, *end, *ptr;
+    const char *start, *end, *ptr, *query_type;
     char uuid_str[UUID_STR_LEN];
     uuid_t uuid;
 
@@ -66,6 +66,8 @@ static inline int aclk_v2_payload_get_query(struct aclk_cloud_req_v2 *cloud_req,
         error("Only accepting requests that start with \"%s\" from CLOUD.", ACLK_CLOUD_REQ_V2_PREFIX);
         return 1;
     }
+    ptr += strlen(ACLK_CLOUD_REQ_V2_PREFIX);
+    query_type = ptr;
 
     if(!(end = strstr(ptr, " HTTP/1.1\x0D\x0A"))) {
         errno = 0;
@@ -73,6 +75,11 @@ static inline int aclk_v2_payload_get_query(struct aclk_cloud_req_v2 *cloud_req,
         return 1;
     }
 
+    if(!(ptr = strchr(ptr, '?')) || ptr > end) 
+        ptr = end;
+    cloud_req->query_endpoint = mallocz((ptr - query_type) + 1);
+    strncpyz(cloud_req->query_endpoint, query_type, ptr - query_type);
+
     req->payload = mallocz((end - start) + 1);
     strncpyz(req->payload, start, end - start);
 
@@ -122,6 +129,13 @@ static int aclk_handle_cloud_request_v1(struct aclk_request *cloud_to_agent, cha
     if (unlikely(aclk_queue_query(cloud_to_agent->callback_topic, NULL, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0, ACLK_CMD_CLOUD)))
         debug(D_ACLK, "ACLK failed to queue incoming \"http\" message");
 
+    if (aclk_stats_enabled) {
+        ACLK_STATS_LOCK;
+        aclk_metrics_per_sample.cloud_req_v1++;
+        aclk_metrics_per_sample.cloud_req_ok++;
+        ACLK_STATS_UNLOCK;
+    }
+
     return 0;
 }
 
@@ -131,6 +145,7 @@ static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, cha
 
     struct aclk_cloud_req_v2 *cloud_req;
     char *data;
+    int stat_idx;
 
     errno = 0;
     if (cloud_to_agent->version < ACLK_V_COMPRESSION) {
@@ -165,6 +180,10 @@ static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, cha
         goto cleanup;
     }
 
+    // we do this here due to cloud_req being taken over by query thread
+    // which if crazy quick can free it after aclk_queue_query
+    stat_idx = aclk_cloud_req_type_to_idx(cloud_req->query_endpoint);
+
     // aclk_queue_query takes ownership of data pointer
     if (unlikely(aclk_queue_query(
             cloud_to_agent->callback_topic, cloud_req, cloud_to_agent->msg_id, cloud_to_agent->payload, 0, 0,
@@ -173,8 +192,17 @@ static int aclk_handle_cloud_request_v2(struct aclk_request *cloud_to_agent, cha
         goto cleanup;
     }
 
+    if (aclk_stats_enabled) {
+        ACLK_STATS_LOCK;
+        aclk_metrics_per_sample.cloud_req_v2++;
+        aclk_metrics_per_sample.cloud_req_ok++;
+        aclk_metrics_per_sample.cloud_req_by_type[stat_idx]++;
+        ACLK_STATS_UNLOCK;
+    }
+
     return 0;
 cleanup:
+    freez(cloud_req->query_endpoint);
     freez(cloud_req->data);
     freez(cloud_req);
     return 1;
@@ -289,12 +317,6 @@ int aclk_handle_cloud_message(char *payload)
     struct aclk_request cloud_to_agent;
     memset(&cloud_to_agent, 0, sizeof(struct aclk_request));
 
-    if (aclk_stats_enabled) {
-        ACLK_STATS_LOCK;
-        aclk_metrics_per_sample.cloud_req_recvd++;
-        ACLK_STATS_UNLOCK;
-    }
-
     if (unlikely(!payload)) {
         errno = 0;
         error("ACLK incoming message is empty");
diff --git a/aclk/legacy/aclk_stats.c b/aclk/legacy/aclk_stats.c
index da0c82bd7a..7124380a2d 100644
--- a/aclk/legacy/aclk_stats.c
+++ b/aclk/legacy/aclk_stats.c
@@ -162,7 +162,7 @@ static void aclk_stats_read_q(struct aclk_metrics_per_sample *per_sample)
 static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample)
 {
     static RRDSET *st = NULL;
-    static RRDDIM *rd_rq_rcvd = NULL;
+    static RRDDIM *rd_rq_ok = NULL;
     static RRDDIM *rd_rq_err = NULL;
 
     if (unlikely(!st)) {
@@ -170,17 +170,82 @@ static void aclk_stats_cloud_req(struct aclk_metrics_per_sample *per_sample)
             "netdata", "aclk_cloud_req", NULL, "aclk", NULL, "Requests received from cloud", "req/s",
             "netdata", "stats", 200005, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
 
-        rd_rq_rcvd = rrddim_add(st, "received", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
-        rd_rq_err = rrddim_add(st, "malformed", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
+        rd_rq_ok = rrddim_add(st, "accepted", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
+        rd_rq_err = rrddim_add(st, "rejected", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
     } else
         rrdset_next(st);
 
-    rrddim_set_by_pointer(st, rd_rq_rcvd, per_sample->cloud_req_recvd - per_sample->cloud_req_err);
+    rrddim_set_by_pointer(st, rd_rq_ok, per_sample->cloud_req_ok);
     rrddim_set_by_pointer(st, rd_rq_err, per_sample->cloud_req_err);
 
     rrdset_done(st);
 }
 
+static void aclk_stats_cloud_req_version(struct aclk_metrics_per_sample *per_sample)
+{
+    static RRDSET *st = NULL;
+    static RRDDIM *rd_rq_v1 = NULL;
+    static RRDDIM *rd_rq_v2 = NULL;
+
+    if (unlikely(!st)) {
+        st = rrdset_create_localhost(
+            "netdata", "aclk_cloud_req_version", NULL, "aclk", NULL, "Requests received from cloud by their version", "req/s",
+            "netdata", "stats", 200006, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
+
+        rd_rq_v1 = rrddim_add(st, "v1",  NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
+        rd_rq_v2 = rrddim_add(st, "v2+", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
+    } else
+        rrdset_next(st);
+
+    rrddim_set_by_pointer(st, rd_rq_v1, per_sample->cloud_req_v1);
+    rrddim_set_by_pointer(st, rd_rq_v2, per_sample->cloud_req_v2);
+
+    rrdset_done(st);
+}
+
+static char *cloud_req_type_names[ACLK_STATS_CLOUD_REQ_TYPE_CNT] = {
+    "other",   
+    "info",
+    "data",
+    "alarms",
+    "alarm_log",
+    "chart",
+    "charts"
+    // if you change update:
+    // #define ACLK_STATS_CLOUD_REQ_TYPE_CNT 7
+};
+
+int aclk_cloud_req_type_to_idx(const char *name)
+{
+    for (int i = 1; i < ACLK_STATS_CLOUD_REQ_TYPE_CNT; i++)
+        if (!strcmp(cloud_req_type_names[i], name))
+            return i;
+    return 0;
+}
+
+static void aclk_stats_cloud_req_cmd(struct aclk_metrics_per_sample *per_sample)
+{
+    static RRDSET *st;
+    static int initialized = 0;
+    static RRDDIM *rd_rq_types[ACLK_STATS_CLOUD_REQ_TYPE_CNT];
+
+    if (unlikely(!initialized)) {
+        initialized = 1;
+        st = rrdset_create_localhost(
+            "netdata", "aclk_cloud_req_cmd", NULL, "aclk", NULL, "Requests received from cloud by their type (api endpoint queried)", "req/s",
+            "netdata", "stats", 200007, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
+
+        for (int i = 0; i < ACLK_STATS_CLOUD_REQ_TYPE_CNT; i++)
+            rd_rq_types[i] = rrddim_add(st, cloud_req_type_names[i], NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
+    } else
+        rrdset_next(st);
+
+    for (int i = 0; i < ACLK_STATS_CLOUD_REQ_TYPE_CNT; i++)
+        rrddim_set_by_pointer(st, rd_rq_types[i], per_sample->cloud_req_by_type[i]);
+
+    rrdset_done(st);
+}
+
 #define MAX_DIM_NAME 16
 static void aclk_stats_query_threads(uint32_t *queries_per_thread)
 {
@@ -191,7 +256,7 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread)
     if (unlikely(!st)) {
         st = rrdset_create_localhost(
             "netdata", "aclk_query_threads", NULL, "aclk", NULL, "Queries Processed Per Thread", "req/s",
-            "netdata", "stats", 200007, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
+            "netdata", "stats", 200008, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
 
         for (int i = 0; i < query_thread_count; i++) {
             if (snprintf(dim_name, MAX_DIM_NAME, "Query %d", i) < 0)
@@ -244,7 +309,7 @@ static void aclk_stats_cpu_threads(void)
 
             aclk_cpu_data[i].st = rrdset_create_localhost(
                                          "netdata", id, NULL, "aclk", NULL, title, "milliseconds/s",
-                                         "netdata", "stats", 200008 + i, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
+                                         "netdata", "stats", 200020 + i, localhost->rrd_update_every, RRDSET_TYPE_STACKED);
 
             aclk_cpu_data[i].user   = rrddim_add(aclk_cpu_data[i].st, "user",   NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL);
             aclk_cpu_data[i].system = rrddim_add(aclk_cpu_data[i].st, "system", NULL, 1, 1000, RRD_ALGORITHM_INCREMENTAL);
@@ -317,6 +382,10 @@ void *aclk_stats_main_thread(void *ptr)
         aclk_stats_read_q(&per_sample);
 
         aclk_stats_cloud_req(&per_sample);
+        aclk_stats_cloud_req_version(&per_sample);
+
+        aclk_stats_cloud_req_cmd(&per_sample);
+
         aclk_stats_query_threads(aclk_queries_per_thread_sample);
 
         aclk_stats_cpu_threads();
diff --git a/aclk/legacy/aclk_stats.h b/aclk/legacy/aclk_stats.h
index f55c4adac0..5e50a2272a 100644
--- a/aclk/legacy/aclk_stats.h
+++ b/aclk/legacy/aclk_stats.h
@@ -55,6 +55,11 @@ extern struct aclk_mat_metrics {
 
 void aclk_metric_mat_update(struct aclk_metric_mat_data *metric, usec_t measurement);
 
+#define ACLK_STATS_CLOUD_REQ_TYPE_CNT 7
+// if you change update cloud_req_type_names
+
+int aclk_cloud_req_type_to_idx(const char *name);
+
 // reset to 0 on every sample
 extern struct aclk_metrics_per_sample {
     /* in the unlikely event of ACLK disconnecting
@@ -72,9 +77,14 @@ extern struct aclk_metrics_per_sample {
     volatile uint32_t read_q_added;
     volatile uint32_t read_q_consumed;
 
-    volatile uint32_t cloud_req_recvd;
+    volatile uint32_t cloud_req_ok;
     volatile uint32_t cloud_req_err;
 
+    volatile uint16_t cloud_req_v1;
+    volatile uint16_t cloud_req_v2;
+
+    volatile uint16_t cloud_req_by_type[ACLK_STATS_CLOUD_REQ_TYPE_CNT];
+
 #ifdef NETDATA_INTERNAL_CHECKS
     struct aclk_metric_mat_data latency;
 #endif