Adds metric showing how long Query spent in Queue (#10016)
* implements new ACLK metric * cleans ACLK_stats code a bit
Timotej S committed
Nov 6, 2020 at 10:16 UTC
144800420e305dfa352aab47ab324af19b320c83
4 files changed
+111
-84
aclk/aclk_query.c
+9
-14
@@ -22,6 +22,7 @@ static netdata_mutex_t queue_mutex = NETDATA_MUTEX_INITIALIZER;
22
23
struct aclk_query {
24
usec_t created;
25
+ usec_t created_boot_time;
26
time_t run_after; // Delay run until after this time
27
ACLK_CMD cmd; // What command is this
28
char *topic; // Topic to respond to
@@ -233,6 +234,7 @@ int aclk_queue_query(char *topic, void *data, char *msg_id, char *query, int run
234
new_query->data = data;
235
new_query->next = NULL;
236
new_query->created = now_realtime_usec();
237
+ new_query->created_boot_time = now_boottime_usec();
238
new_query->run_after = run_after;
239
240
debug(D_ACLK, "Added query (%s) (%s)", topic, query ? query : "");
@@ -319,22 +321,15 @@ static char *aclk_encode_response(char *src, size_t content_size, int keep_newli
321
#pragma region ACLK_QUERY
322
#endif
323
322
-static usec_t aclk_web_api_request_v1(RRDHOST *host, struct web_client *w, char *url)
324
+static usec_t aclk_web_api_request_v1(RRDHOST *host, struct web_client *w, char *url, usec_t q_created)
325
{
324
- usec_t t;
326
+ usec_t t = now_boottime_usec();
327
+ aclk_metric_mat_update(&aclk_metrics_per_sample.cloud_q_recvd_to_processed, t - q_created);
328
326
- t = now_monotonic_high_precision_usec();
329
w->response.code = web_client_api_request_v1(host, w, url);
328
- t = now_monotonic_high_precision_usec() - t;
330
+ t = now_boottime_usec() - t;
331
330
- if (aclk_stats_enabled) {
331
- ACLK_STATS_LOCK;
332
- aclk_metrics_per_sample.cloud_q_process_total += t;
333
- aclk_metrics_per_sample.cloud_q_process_count++;
334
- if (aclk_metrics_per_sample.cloud_q_process_max < t)
335
- aclk_metrics_per_sample.cloud_q_process_max = t;
336
- ACLK_STATS_UNLOCK;
337
- }
332
+ aclk_metric_mat_update(&aclk_metrics_per_sample.cloud_q_db_query_time, t);
333
334
return t;
335
}
@@ -361,7 +356,7 @@ static int aclk_execute_query(struct aclk_query *this_query)
356
mysep = strrchr(this_query->query, '/');
357
358
// TODO: handle bad response perhaps in a different way. For now it does to the payload
364
- aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
359
+ aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop", this_query->created_boot_time);
360
now_realtime_timeval(&w->tv_ready);
361
w->response.data->date = w->tv_ready.tv_sec;
362
web_client_build_http_header(w); // TODO: this function should offset from date, not tv_ready
@@ -427,7 +422,7 @@ static int aclk_execute_query_v2(struct aclk_query *this_query)
422
mysep = strrchr(this_query->query, '/');
423
424
// execute the query
430
- t = aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop");
425
+ t = aclk_web_api_request_v1(localhost, w, mysep ? mysep + 1 : "noop", this_query->created_boot_time);
426
427
#ifdef NETDATA_WITH_ZLIB
428
// check if gzip encoding can and should be used
aclk/aclk_stats.c
+67
-52
@@ -20,6 +20,51 @@ struct aclk_metrics aclk_metrics = {
20
21
struct aclk_metrics_per_sample aclk_metrics_per_sample;
22
23
+struct aclk_mat_metrics aclk_mat_metrics = {
24
+#ifdef NETDATA_INTERNAL_CHECKS
25
+ .latency = { .name = "aclk_latency_mqtt",
26
+ .prio = 200002,
27
+ .st = NULL,
28
+ .rd_avg = NULL,
29
+ .rd_max = NULL,
30
+ .rd_total = NULL,
31
+ .unit = "ms",
32
+ .title = "ACLK Message Publish Latency" },
33
+#endif
34
+
35
+ .cloud_q_db_query_time = { .name = "aclk_db_query_time",
36
+ .prio = 200006,
37
+ .st = NULL,
38
+ .rd_avg = NULL,
39
+ .rd_max = NULL,
40
+ .rd_total = NULL,
41
+ .unit = "us",
42
+ .title = "Time it took to process cloud requested DB queries" },
43
+
44
+ .cloud_q_recvd_to_processed = { .name = "aclk_cloud_q_recvd_to_processed",
45
+ .prio = 200007,
46
+ .st = NULL,
47
+ .rd_avg = NULL,
48
+ .rd_max = NULL,
49
+ .rd_total = NULL,
50
+ .unit = "us",
51
+ .title = "Time from receiving the Cloud Query until it was picked up "
52
+ "by query thread (just before passing to the database)." }
53
+};
54
+
55
+void aclk_metric_mat_update(struct aclk_metric_mat_data *metric, usec_t measurement)
56
+{
57
+ if (aclk_stats_enabled) {
58
+ ACLK_STATS_LOCK;
59
+ if (metric->max < measurement)
60
+ metric->max = measurement;
61
+
62
+ metric->total += measurement;
63
+ metric->count++;
64
+ ACLK_STATS_UNLOCK;
65
+ }
66
+}
67
+
68
static void aclk_stats_collect(struct aclk_metrics_per_sample *per_sample, struct aclk_metrics *permanent)
69
{
70
static RRDSET *st_aclkstats = NULL;
@@ -61,33 +106,6 @@ static void aclk_stats_query_queue(struct aclk_metrics_per_sample *per_sample)
106
rrdset_done(st_query_thread);
107
}
108
64
-#ifdef NETDATA_INTERNAL_CHECKS
65
-static void aclk_stats_latency(struct aclk_metrics_per_sample *per_sample)
66
-{
67
- static RRDSET *st = NULL;
68
- static RRDDIM *rd_avg = NULL;
69
- static RRDDIM *rd_max = NULL;
70
-
71
- if (unlikely(!st)) {
72
- st = rrdset_create_localhost(
73
- "netdata", "aclk_latency_mqtt", NULL, "aclk", NULL, "ACLK Message Publish Latency", "ms",
74
- "netdata", "stats", 200002, localhost->rrd_update_every, RRDSET_TYPE_LINE);
75
-
76
- rd_avg = rrddim_add(st, "avg", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
77
- rd_max = rrddim_add(st, "max", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
78
- } else
79
- rrdset_next(st);
80
- if(per_sample->latency_count)
81
- rrddim_set_by_pointer(st, rd_avg, roundf((float)per_sample->latency_total / per_sample->latency_count));
82
- else
83
- rrddim_set_by_pointer(st, rd_avg, 0);
84
-
85
- rrddim_set_by_pointer(st, rd_max, per_sample->latency_max);
86
-
87
- rrdset_done(st);
88
-}
89
-#endif
90
-
109
static void aclk_stats_write_q(struct aclk_metrics_per_sample *per_sample)
110
{
111
static RRDSET *st = NULL;
@@ -181,32 +199,27 @@ static void aclk_stats_query_threads(uint32_t *queries_per_thread)
199
rrdset_done(st);
200
}
201
184
-static void aclk_stats_query_time(struct aclk_metrics_per_sample *per_sample)
202
+static void aclk_stats_mat_metric_process(struct aclk_metric_mat *metric, struct aclk_metric_mat_data *data)
203
{
186
- static RRDSET *st = NULL;
187
- static RRDDIM *rd_rq_avg = NULL;
188
- static RRDDIM *rd_rq_max = NULL;
189
- static RRDDIM *rd_rq_total = NULL;
190
-
191
- if (unlikely(!st)) {
192
- st = rrdset_create_localhost(
193
- "netdata", "aclk_query_time", NULL, "aclk", NULL, "Time it took to process cloud requested DB queries", "us",
194
- "netdata", "stats", 200006, localhost->rrd_update_every, RRDSET_TYPE_LINE);
195
-
196
- rd_rq_avg = rrddim_add(st, "avg", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
197
- rd_rq_max = rrddim_add(st, "max", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
198
- rd_rq_total = rrddim_add(st, "total", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
204
+ if(unlikely(!metric->st)) {
205
+ metric->st = rrdset_create_localhost(
206
+ "netdata", metric->name, NULL, "aclk", NULL, metric->title, metric->unit, "netdata", "stats", metric->prio,
207
+ localhost->rrd_update_every, RRDSET_TYPE_LINE);
208
+
209
+ metric->rd_avg = rrddim_add(metric->st, "avg", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
210
+ metric->rd_max = rrddim_add(metric->st, "max", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
211
+ metric->rd_total = rrddim_add(metric->st, "total", NULL, 1, localhost->rrd_update_every, RRD_ALGORITHM_ABSOLUTE);
212
} else
200
- rrdset_next(st);
213
+ rrdset_next(metric->st);
214
202
- if(per_sample->cloud_q_process_count)
203
- rrddim_set_by_pointer(st, rd_rq_avg, roundf((float)per_sample->cloud_q_process_total / per_sample->cloud_q_process_count));
215
+ if(data->count)
216
+ rrddim_set_by_pointer(metric->st, metric->rd_avg, roundf((float)data->total / data->count));
217
else
205
- rrddim_set_by_pointer(st, rd_rq_avg, 0);
206
- rrddim_set_by_pointer(st, rd_rq_max, per_sample->cloud_q_process_max);
207
- rrddim_set_by_pointer(st, rd_rq_total, per_sample->cloud_q_process_total);
218
+ rrddim_set_by_pointer(metric->st, metric->rd_avg, 0);
219
+ rrddim_set_by_pointer(metric->st, metric->rd_max, data->max);
220
+ rrddim_set_by_pointer(metric->st, metric->rd_total, data->total);
221
209
- rrdset_done(st);
222
+ rrdset_done(metric->st);
223
}
224
225
void aclk_stats_thread_cleanup()
@@ -255,16 +268,18 @@ void *aclk_stats_main_thread(void *ptr)
268
269
aclk_stats_collect(&per_sample, &permanent);
270
aclk_stats_query_queue(&per_sample);
258
-#ifdef NETDATA_INTERNAL_CHECKS
259
- aclk_stats_latency(&per_sample);
260
-#endif
271
+
272
aclk_stats_write_q(&per_sample);
273
aclk_stats_read_q(&per_sample);
274
275
aclk_stats_cloud_req(&per_sample);
276
aclk_stats_query_threads(aclk_queries_per_thread_sample);
277
267
- aclk_stats_query_time(&per_sample);
278
+#ifdef NETDATA_INTERNAL_CHECKS
279
+ aclk_stats_mat_metric_process(&aclk_mat_metrics.latency, &per_sample.latency);
280
+#endif
281
+ aclk_stats_mat_metric_process(&aclk_mat_metrics.cloud_q_db_query_time, &per_sample.cloud_q_db_query_time);
282
+ aclk_stats_mat_metric_process(&aclk_mat_metrics.cloud_q_recvd_to_processed, &per_sample.cloud_q_recvd_to_processed);
283
}
284
285
return 0;
aclk/aclk_stats.h
+34
-9
@@ -26,6 +26,35 @@ struct aclk_metrics {
26
volatile uint8_t online;
27
};
28
29
+//mat = max average total
30
+struct aclk_metric_mat_data {
31
+ volatile uint32_t total;
32
+ volatile uint32_t count;
33
+ volatile uint32_t max;
34
+};
35
+
36
+//mat = max average total
37
+struct aclk_metric_mat {
38
+ char *name;
39
+ char *title;
40
+ RRDSET *st;
41
+ RRDDIM *rd_avg;
42
+ RRDDIM *rd_max;
43
+ RRDDIM *rd_total;
44
+ long prio;
45
+ char *unit;
46
+};
47
+
48
+extern struct aclk_mat_metrics {
49
+#ifdef NETDATA_INTERNAL_CHECKS
50
+ struct aclk_metric_mat latency;
51
+#endif
52
+ struct aclk_metric_mat cloud_q_db_query_time;
53
+ struct aclk_metric_mat cloud_q_recvd_to_processed;
54
+} aclk_mat_metrics;
55
+
56
+void aclk_metric_mat_update(struct aclk_metric_mat_data *metric, usec_t measurement);
57
+
58
// reset to 0 on every sample
59
extern struct aclk_metrics_per_sample {
60
/* in the unlikely event of ACLK disconnecting
@@ -37,12 +66,6 @@ extern struct aclk_metrics_per_sample {
66
volatile uint32_t queries_queued;
67
volatile uint32_t queries_dispatched;
68
40
-#ifdef NETDATA_INTERNAL_CHECKS
41
- volatile uint32_t latency_max;
42
- volatile uint32_t latency_total;
43
- volatile uint32_t latency_count;
44
-#endif
45
-
69
volatile uint32_t write_q_added;
70
volatile uint32_t write_q_consumed;
71
@@ -52,9 +75,11 @@ extern struct aclk_metrics_per_sample {
75
volatile uint32_t cloud_req_recvd;
76
volatile uint32_t cloud_req_err;
77
55
- volatile uint32_t cloud_q_process_total;
56
- volatile uint32_t cloud_q_process_count;
57
- volatile uint32_t cloud_q_process_max;
78
+#ifdef NETDATA_INTERNAL_CHECKS
79
+ struct aclk_metric_mat_data latency;
80
+#endif
81
+ struct aclk_metric_mat_data cloud_q_db_query_time;
82
+ struct aclk_metric_mat_data cloud_q_recvd_to_processed;
83
} aclk_metrics_per_sample;
84
85
extern uint32_t *aclk_queries_per_thread;
aclk/mqtt.c
+1
-9
@@ -44,15 +44,7 @@ void publish_callback(struct mosquitto *mosq, void *obj, int rc)
44
45
info("Publish_callback: mid=%d latency=%" PRId64 "ms", rc, diff);
46
47
- if (aclk_stats_enabled) {
48
- ACLK_STATS_LOCK;
49
- if (aclk_metrics_per_sample.latency_max < diff)
50
- aclk_metrics_per_sample.latency_max = diff;
51
-
52
- aclk_metrics_per_sample.latency_total += diff;
53
- aclk_metrics_per_sample.latency_count++;
54
- ACLK_STATS_UNLOCK;
55
- }
47
+ aclk_metric_mat_update(&aclk_metrics_per_sample.latency, diff);
48
#endif
49
return;
50
}