@cryptotaxi247 / netdata-1 / commits / 53a13ab8e

replication fixes No 7 (#14053)

* move global statistics workers to a separate thread; query statistics per query source; query statistics for ML, exporters, backfilling; reset replication point in time every 10 seconds, instead of every 1; fix compilation warnings; optimize the replication queries code; prevent long tail of replication requests (big sleeps); provide query statistics about replication ; optimize replication sender when most senders are full; optimize replication_request_get_first_available(); reset replication completion calculation; * remove workers utilization from global statistics thread

Costa Tsaousis committed Nov 28, 2022 at 12:22 UTC 53a13ab8e110923d097968353a6bc1e22399480f
29 files changed +701 -247
daemon/global_statistics.c
+359 -88
@@ -14,42 +14,66 @@
14 #define WORKER_JOB_MALLOC_TRACE 7
15
16 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 8
17 -#error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 5
17 +#error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 8
18 #endif
19
20 bool global_statistics_enabled = true;
21
22 static struct global_statistics {
23 - volatile uint16_t connected_clients;
24 -
25 - volatile uint64_t web_requests;
26 - volatile uint64_t web_usec;
27 - volatile uint64_t web_usec_max;
28 - volatile uint64_t bytes_received;
29 - volatile uint64_t bytes_sent;
30 - volatile uint64_t content_size;
31 - volatile uint64_t compressed_content_size;
32 -
33 - volatile uint64_t web_client_count;
34 -
35 - volatile uint64_t rrdr_queries_made;
36 - volatile uint64_t rrdr_db_points_read;
37 - volatile uint64_t rrdr_result_points_generated;
38 -
39 - volatile uint64_t sqlite3_queries_made;
40 - volatile uint64_t sqlite3_queries_ok;
41 - volatile uint64_t sqlite3_queries_failed;
42 - volatile uint64_t sqlite3_queries_failed_busy;
43 - volatile uint64_t sqlite3_queries_failed_locked;
44 - volatile uint64_t sqlite3_rows;
45 - volatile uint64_t sqlite3_metadata_cache_hit;
46 - volatile uint64_t sqlite3_context_cache_hit;
47 - volatile uint64_t sqlite3_metadata_cache_miss;
48 - volatile uint64_t sqlite3_context_cache_miss;
49 - volatile uint64_t sqlite3_metadata_cache_spill;
50 - volatile uint64_t sqlite3_context_cache_spill;
51 - volatile uint64_t sqlite3_metadata_cache_write;
52 - volatile uint64_t sqlite3_context_cache_write;
23 + uint16_t connected_clients;
24 +
25 + uint64_t web_requests;
26 + uint64_t web_usec;
27 + uint64_t web_usec_max;
28 + uint64_t bytes_received;
29 + uint64_t bytes_sent;
30 + uint64_t content_size;
31 + uint64_t compressed_content_size;
32 +
33 + uint64_t web_client_count;
34 +
35 + uint64_t api_data_queries_made;
36 + uint64_t api_data_db_points_read;
37 + uint64_t api_data_result_points_generated;
38 +
39 + uint64_t api_weights_queries_made;
40 + uint64_t api_weights_db_points_read;
41 + uint64_t api_weights_result_points_generated;
42 +
43 + uint64_t api_badges_queries_made;
44 + uint64_t api_badges_db_points_read;
45 + uint64_t api_badges_result_points_generated;
46 +
47 + uint64_t health_queries_made;
48 + uint64_t health_db_points_read;
49 + uint64_t health_result_points_generated;
50 +
51 + uint64_t ml_queries_made;
52 + uint64_t ml_db_points_read;
53 + uint64_t ml_result_points_generated;
54 +
55 + uint64_t exporters_queries_made;
56 + uint64_t exporters_db_points_read;
57 +
58 + uint64_t backfill_queries_made;
59 + uint64_t backfill_db_points_read;
60 +
61 + uint64_t db_points_stored_per_tier[RRD_STORAGE_TIERS];
62 +
63 + uint64_t sqlite3_queries_made;
64 + uint64_t sqlite3_queries_ok;
65 + uint64_t sqlite3_queries_failed;
66 + uint64_t sqlite3_queries_failed_busy;
67 + uint64_t sqlite3_queries_failed_locked;
68 + uint64_t sqlite3_rows;
69 + uint64_t sqlite3_metadata_cache_hit;
70 + uint64_t sqlite3_context_cache_hit;
71 + uint64_t sqlite3_metadata_cache_miss;
72 + uint64_t sqlite3_context_cache_miss;
73 + uint64_t sqlite3_metadata_cache_spill;
74 + uint64_t sqlite3_context_cache_spill;
75 + uint64_t sqlite3_metadata_cache_write;
76 + uint64_t sqlite3_context_cache_write;
77
78 } global_statistics = {
79 .connected_clients = 0,
@@ -61,12 +85,34 @@ static struct global_statistics {
85 .compressed_content_size = 0,
86 .web_client_count = 1,
87
64 - .rrdr_queries_made = 0,
65 - .rrdr_db_points_read = 0,
66 - .rrdr_result_points_generated = 0,
88 + .api_data_queries_made = 0,
89 + .api_data_db_points_read = 0,
90 + .api_data_result_points_generated = 0,
91 };
92
69 -void sqlite3_query_completed(bool success, bool busy, bool locked) {
93 +void global_statistics_rrdset_done_chart_collection_completed(size_t *points_read_per_tier_array) {
94 + for(size_t tier = 0; tier < storage_tiers ;tier++) {
95 + __atomic_fetch_add(&global_statistics.db_points_stored_per_tier[tier], points_read_per_tier_array[tier], __ATOMIC_RELAXED);
96 + points_read_per_tier_array[tier] = 0;
97 + }
98 +}
99 +
100 +void global_statistics_ml_query_completed(size_t points_read) {
101 + __atomic_fetch_add(&global_statistics.ml_queries_made, 1, __ATOMIC_RELAXED);
102 + __atomic_fetch_add(&global_statistics.ml_db_points_read, points_read, __ATOMIC_RELAXED);
103 +}
104 +
105 +void global_statistics_exporters_query_completed(size_t points_read) {
106 + __atomic_fetch_add(&global_statistics.exporters_queries_made, 1, __ATOMIC_RELAXED);
107 + __atomic_fetch_add(&global_statistics.exporters_db_points_read, points_read, __ATOMIC_RELAXED);
108 +}
109 +
110 +void global_statistics_backfill_query_completed(size_t points_read) {
111 + __atomic_fetch_add(&global_statistics.backfill_queries_made, 1, __ATOMIC_RELAXED);
112 + __atomic_fetch_add(&global_statistics.backfill_db_points_read, points_read, __ATOMIC_RELAXED);
113 +}
114 +
115 +void global_statistics_sqlite3_query_completed(bool success, bool busy, bool locked) {
116 __atomic_fetch_add(&global_statistics.sqlite3_queries_made, 1, __ATOMIC_RELAXED);
117
118 if(success) {
@@ -83,21 +129,54 @@ void sqlite3_query_completed(bool success, bool busy, bool locked) {
129 }
130 }
131
86 -void sqlite3_row_completed(void) {
132 +void global_statistics_sqlite3_row_completed(void) {
133 __atomic_fetch_add(&global_statistics.sqlite3_rows, 1, __ATOMIC_RELAXED);
134 }
135
90 -void rrdr_query_completed(uint64_t db_points_read, uint64_t result_points_generated) {
91 - __atomic_fetch_add(&global_statistics.rrdr_queries_made, 1, __ATOMIC_RELAXED);
92 - __atomic_fetch_add(&global_statistics.rrdr_db_points_read, db_points_read, __ATOMIC_RELAXED);
93 - __atomic_fetch_add(&global_statistics.rrdr_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
136 +void global_statistics_rrdr_query_completed(size_t queries, uint64_t db_points_read, uint64_t result_points_generated, QUERY_SOURCE query_source) {
137 + switch(query_source) {
138 + case QUERY_SOURCE_API_DATA:
139 + __atomic_fetch_add(&global_statistics.api_data_queries_made, queries, __ATOMIC_RELAXED);
140 + __atomic_fetch_add(&global_statistics.api_data_db_points_read, db_points_read, __ATOMIC_RELAXED);
141 + __atomic_fetch_add(&global_statistics.api_data_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
142 + break;
143 +
144 + case QUERY_SOURCE_ML:
145 + __atomic_fetch_add(&global_statistics.ml_queries_made, queries, __ATOMIC_RELAXED);
146 + __atomic_fetch_add(&global_statistics.ml_db_points_read, db_points_read, __ATOMIC_RELAXED);
147 + __atomic_fetch_add(&global_statistics.ml_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
148 + break;
149 +
150 + case QUERY_SOURCE_API_WEIGHTS:
151 + __atomic_fetch_add(&global_statistics.api_weights_queries_made, queries, __ATOMIC_RELAXED);
152 + __atomic_fetch_add(&global_statistics.api_weights_db_points_read, db_points_read, __ATOMIC_RELAXED);
153 + __atomic_fetch_add(&global_statistics.api_weights_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
154 + break;
155 +
156 + case QUERY_SOURCE_API_BADGE:
157 + __atomic_fetch_add(&global_statistics.api_badges_queries_made, queries, __ATOMIC_RELAXED);
158 + __atomic_fetch_add(&global_statistics.api_badges_db_points_read, db_points_read, __ATOMIC_RELAXED);
159 + __atomic_fetch_add(&global_statistics.api_badges_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
160 + break;
161 +
162 + case QUERY_SOURCE_HEALTH:
163 + __atomic_fetch_add(&global_statistics.health_queries_made, queries, __ATOMIC_RELAXED);
164 + __atomic_fetch_add(&global_statistics.health_db_points_read, db_points_read, __ATOMIC_RELAXED);
165 + __atomic_fetch_add(&global_statistics.health_result_points_generated, result_points_generated, __ATOMIC_RELAXED);
166 + break;
167 +
168 + default:
169 + case QUERY_SOURCE_UNITTEST:
170 + case QUERY_SOURCE_UNKNOWN:
171 + break;
172 + }
173 }
174
96 -void finished_web_request_statistics(uint64_t dt,
97 - uint64_t bytes_received,
98 - uint64_t bytes_sent,
99 - uint64_t content_size,
100 - uint64_t compressed_content_size) {
175 +void global_statistics_web_request_completed(uint64_t dt,
176 + uint64_t bytes_received,
177 + uint64_t bytes_sent,
178 + uint64_t content_size,
179 + uint64_t compressed_content_size) {
180 uint64_t old_web_usec_max = global_statistics.web_usec_max;
181 while(dt > old_web_usec_max)
182 __atomic_compare_exchange(&global_statistics.web_usec_max, &old_web_usec_max, &dt, 1, __ATOMIC_RELAXED, __ATOMIC_RELAXED);
@@ -110,16 +189,15 @@ void finished_web_request_statistics(uint64_t dt,
189 __atomic_fetch_add(&global_statistics.compressed_content_size, compressed_content_size, __ATOMIC_RELAXED);
190 }
191
113 -uint64_t web_client_connected(void) {
192 +uint64_t global_statistics_web_client_connected(void) {
193 __atomic_fetch_add(&global_statistics.connected_clients, 1, __ATOMIC_RELAXED);
194 return __atomic_fetch_add(&global_statistics.web_client_count, 1, __ATOMIC_RELAXED);
195 }
196
118 -void web_client_disconnected(void) {
197 +void global_statistics_web_client_disconnected(void) {
198 __atomic_fetch_sub(&global_statistics.connected_clients, 1, __ATOMIC_RELAXED);
199 }
200
122 -
201 static inline void global_statistics_copy(struct global_statistics *gs, uint8_t options) {
202 gs->connected_clients = __atomic_load_n(&global_statistics.connected_clients, __ATOMIC_RELAXED);
203 gs->web_requests = __atomic_load_n(&global_statistics.web_requests, __ATOMIC_RELAXED);
@@ -131,9 +209,33 @@ static inline void global_statistics_copy(struct global_statistics *gs, uint8_t
209 gs->compressed_content_size = __atomic_load_n(&global_statistics.compressed_content_size, __ATOMIC_RELAXED);
210 gs->web_client_count = __atomic_load_n(&global_statistics.web_client_count, __ATOMIC_RELAXED);
211
134 - gs->rrdr_queries_made = __atomic_load_n(&global_statistics.rrdr_queries_made, __ATOMIC_RELAXED);
135 - gs->rrdr_db_points_read = __atomic_load_n(&global_statistics.rrdr_db_points_read, __ATOMIC_RELAXED);
136 - gs->rrdr_result_points_generated = __atomic_load_n(&global_statistics.rrdr_result_points_generated, __ATOMIC_RELAXED);
212 + gs->api_data_queries_made = __atomic_load_n(&global_statistics.api_data_queries_made, __ATOMIC_RELAXED);
213 + gs->api_data_db_points_read = __atomic_load_n(&global_statistics.api_data_db_points_read, __ATOMIC_RELAXED);
214 + gs->api_data_result_points_generated = __atomic_load_n(&global_statistics.api_data_result_points_generated, __ATOMIC_RELAXED);
215 +
216 + gs->api_weights_queries_made = __atomic_load_n(&global_statistics.api_weights_queries_made, __ATOMIC_RELAXED);
217 + gs->api_weights_db_points_read = __atomic_load_n(&global_statistics.api_weights_db_points_read, __ATOMIC_RELAXED);
218 + gs->api_weights_result_points_generated = __atomic_load_n(&global_statistics.api_weights_result_points_generated, __ATOMIC_RELAXED);
219 +
220 + gs->api_badges_queries_made = __atomic_load_n(&global_statistics.api_badges_queries_made, __ATOMIC_RELAXED);
221 + gs->api_badges_db_points_read = __atomic_load_n(&global_statistics.api_badges_db_points_read, __ATOMIC_RELAXED);
222 + gs->api_badges_result_points_generated = __atomic_load_n(&global_statistics.api_badges_result_points_generated, __ATOMIC_RELAXED);
223 +
224 + gs->health_queries_made = __atomic_load_n(&global_statistics.health_queries_made, __ATOMIC_RELAXED);
225 + gs->health_db_points_read = __atomic_load_n(&global_statistics.health_db_points_read, __ATOMIC_RELAXED);
226 + gs->health_result_points_generated = __atomic_load_n(&global_statistics.health_result_points_generated, __ATOMIC_RELAXED);
227 +
228 + gs->ml_queries_made = __atomic_load_n(&global_statistics.ml_queries_made, __ATOMIC_RELAXED);
229 + gs->ml_db_points_read = __atomic_load_n(&global_statistics.ml_db_points_read, __ATOMIC_RELAXED);
230 + gs->ml_result_points_generated = __atomic_load_n(&global_statistics.ml_result_points_generated, __ATOMIC_RELAXED);
231 +
232 + gs->exporters_queries_made = __atomic_load_n(&global_statistics.exporters_queries_made, __ATOMIC_RELAXED);
233 + gs->exporters_db_points_read = __atomic_load_n(&global_statistics.exporters_db_points_read, __ATOMIC_RELAXED);
234 + gs->backfill_queries_made = __atomic_load_n(&global_statistics.backfill_queries_made, __ATOMIC_RELAXED);
235 + gs->backfill_db_points_read = __atomic_load_n(&global_statistics.backfill_db_points_read, __ATOMIC_RELAXED);
236 +
237 + for(size_t tier = 0; tier < storage_tiers ;tier++)
238 + gs->db_points_stored_per_tier[tier] = __atomic_load_n(&global_statistics.db_points_stored_per_tier[tier], __ATOMIC_RELAXED);
239
240 if(options & GLOBAL_STATS_RESET_WEB_USEC_MAX) {
241 uint64_t n = 0;
@@ -177,6 +279,7 @@ static void global_statistics_charts(void) {
279 struct global_statistics gs;
280 struct rusage me;
281
282 + struct replication_query_statistics replication = replication_get_query_statistics();
283 global_statistics_copy(&gs, GLOBAL_STATS_RESET_WEB_USEC_MAX);
284 getrusage(RUSAGE_SELF, &me);
285
@@ -425,65 +528,184 @@ static void global_statistics_charts(void) {
528
529 // ----------------------------------------------------------------
530
428 - if(gs.rrdr_queries_made) {
429 - static RRDSET *st_rrdr_queries = NULL;
430 - static RRDDIM *rd_queries = NULL;
431 -
432 - if (unlikely(!st_rrdr_queries)) {
433 - st_rrdr_queries = rrdset_create_localhost(
531 + {
532 + static RRDSET *st_queries = NULL;
533 + static RRDDIM *rd_api_data_queries = NULL;
534 + static RRDDIM *rd_api_weights_queries = NULL;
535 + static RRDDIM *rd_api_badges_queries = NULL;
536 + static RRDDIM *rd_health_queries = NULL;
537 + static RRDDIM *rd_ml_queries = NULL;
538 + static RRDDIM *rd_exporters_queries = NULL;
539 + static RRDDIM *rd_backfill_queries = NULL;
540 + static RRDDIM *rd_replication_queries = NULL;
541 +
542 + if (unlikely(!st_queries)) {
543 + st_queries = rrdset_create_localhost(
544 "netdata"
545 , "queries"
546 , NULL
547 , "queries"
548 , NULL
439 - , "Netdata API Queries"
549 + , "Netdata DB Queries"
550 , "queries/s"
551 , "netdata"
552 , "stats"
553 , 131000
554 , localhost->rrd_update_every
445 - , RRDSET_TYPE_LINE
555 + , RRDSET_TYPE_STACKED
556 );
557
448 - rd_queries = rrddim_add(st_rrdr_queries, "queries", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
558 + rd_api_data_queries = rrddim_add(st_queries, "/api/v1/data", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
559 + rd_api_weights_queries = rrddim_add(st_queries, "/api/v1/weights", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
560 + rd_api_badges_queries = rrddim_add(st_queries, "/api/v1/badge", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
561 + rd_health_queries = rrddim_add(st_queries, "health", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
562 + rd_ml_queries = rrddim_add(st_queries, "ml", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
563 + rd_exporters_queries = rrddim_add(st_queries, "exporters", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
564 + rd_backfill_queries = rrddim_add(st_queries, "backfill", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
565 + rd_replication_queries = rrddim_add(st_queries, "replication", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
566 }
567
451 - rrddim_set_by_pointer(st_rrdr_queries, rd_queries, (collected_number)gs.rrdr_queries_made);
568 + rrddim_set_by_pointer(st_queries, rd_api_data_queries, (collected_number)gs.api_data_queries_made);
569 + rrddim_set_by_pointer(st_queries, rd_api_weights_queries, (collected_number)gs.api_weights_queries_made);
570 + rrddim_set_by_pointer(st_queries, rd_api_badges_queries, (collected_number)gs.api_badges_queries_made);
571 + rrddim_set_by_pointer(st_queries, rd_health_queries, (collected_number)gs.health_queries_made);
572 + rrddim_set_by_pointer(st_queries, rd_ml_queries, (collected_number)gs.ml_queries_made);
573 + rrddim_set_by_pointer(st_queries, rd_exporters_queries, (collected_number)gs.exporters_queries_made);
574 + rrddim_set_by_pointer(st_queries, rd_backfill_queries, (collected_number)gs.backfill_queries_made);
575 + rrddim_set_by_pointer(st_queries, rd_replication_queries, (collected_number)replication.queries_finished);
576
453 - rrdset_done(st_rrdr_queries);
577 + rrdset_done(st_queries);
578 }
579
580 // ----------------------------------------------------------------
581
458 - if(gs.rrdr_db_points_read || gs.rrdr_result_points_generated) {
459 - static RRDSET *st_rrdr_points = NULL;
460 - static RRDDIM *rd_points_read = NULL;
461 - static RRDDIM *rd_points_generated = NULL;
462 -
463 - if (unlikely(!st_rrdr_points)) {
464 - st_rrdr_points = rrdset_create_localhost(
582 + {
583 + static RRDSET *st_points_read = NULL;
584 + static RRDDIM *rd_api_data_points_read = NULL;
585 + static RRDDIM *rd_api_weights_points_read = NULL;
586 + static RRDDIM *rd_api_badges_points_read = NULL;
587 + static RRDDIM *rd_health_points_read = NULL;
588 + static RRDDIM *rd_ml_points_read = NULL;
589 + static RRDDIM *rd_exporters_points_read = NULL;
590 + static RRDDIM *rd_backfill_points_read = NULL;
591 + static RRDDIM *rd_replication_points_read = NULL;
592 +
593 + if (unlikely(!st_points_read)) {
594 + st_points_read = rrdset_create_localhost(
595 "netdata"
466 - , "db_points"
596 + , "db_points_read"
597 , NULL
598 , "queries"
599 , NULL
470 - , "Netdata API Points"
600 + , "Netdata DB Points Query Read"
601 , "points/s"
602 , "netdata"
603 , "stats"
604 , 131001
605 , localhost->rrd_update_every
476 - , RRDSET_TYPE_AREA
606 + , RRDSET_TYPE_STACKED
607 + );
608 +
609 + rd_api_data_points_read = rrddim_add(st_points_read, "/api/v1/data", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
610 + rd_api_weights_points_read = rrddim_add(st_points_read, "/api/v1/weights", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
611 + rd_api_badges_points_read = rrddim_add(st_points_read, "/api/v1/badge", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
612 + rd_health_points_read = rrddim_add(st_points_read, "health", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
613 + rd_ml_points_read = rrddim_add(st_points_read, "ml", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
614 + rd_exporters_points_read = rrddim_add(st_points_read, "exporters", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
615 + rd_backfill_points_read = rrddim_add(st_points_read, "backfill", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
616 + rd_replication_points_read = rrddim_add(st_points_read, "replication", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
617 + }
618 +
619 + rrddim_set_by_pointer(st_points_read, rd_api_data_points_read, (collected_number)gs.api_data_db_points_read);
620 + rrddim_set_by_pointer(st_points_read, rd_api_weights_points_read, (collected_number)gs.api_weights_db_points_read);
621 + rrddim_set_by_pointer(st_points_read, rd_api_badges_points_read, (collected_number)gs.api_badges_db_points_read);
622 + rrddim_set_by_pointer(st_points_read, rd_health_points_read, (collected_number)gs.health_db_points_read);
623 + rrddim_set_by_pointer(st_points_read, rd_ml_points_read, (collected_number)gs.ml_db_points_read);
624 + rrddim_set_by_pointer(st_points_read, rd_exporters_points_read, (collected_number)gs.exporters_db_points_read);
625 + rrddim_set_by_pointer(st_points_read, rd_backfill_points_read, (collected_number)gs.backfill_db_points_read);
626 + rrddim_set_by_pointer(st_points_read, rd_replication_points_read, (collected_number)replication.points_read);
627 +
628 + rrdset_done(st_points_read);
629 + }
630 +
631 + // ----------------------------------------------------------------
632 +
633 + if(gs.api_data_result_points_generated || replication.points_generated) {
634 + static RRDSET *st_points_generated = NULL;
635 + static RRDDIM *rd_api_data_points_generated = NULL;
636 + static RRDDIM *rd_api_weights_points_generated = NULL;
637 + static RRDDIM *rd_api_badges_points_generated = NULL;
638 + static RRDDIM *rd_health_points_generated = NULL;
639 + static RRDDIM *rd_ml_points_generated = NULL;
640 + static RRDDIM *rd_replication_points_generated = NULL;
641 +
642 + if (unlikely(!st_points_generated)) {
643 + st_points_generated = rrdset_create_localhost(
644 + "netdata"
645 + , "db_points_results"
646 + , NULL
647 + , "queries"
648 + , NULL
649 + , "Netdata Points in Query Results"
650 + , "points/s"
651 + , "netdata"
652 + , "stats"
653 + , 131002
654 + , localhost->rrd_update_every
655 + , RRDSET_TYPE_STACKED
656 + );
657 +
658 + rd_api_data_points_generated = rrddim_add(st_points_generated, "/api/v1/data", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
659 + rd_api_weights_points_generated = rrddim_add(st_points_generated, "/api/v1/weights", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
660 + rd_api_badges_points_generated = rrddim_add(st_points_generated, "/api/v1/badge", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
661 + rd_health_points_generated = rrddim_add(st_points_generated, "health", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
662 + rd_ml_points_generated = rrddim_add(st_points_generated, "ml", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
663 + rd_replication_points_generated = rrddim_add(st_points_generated, "replication", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
664 + }
665 +
666 + rrddim_set_by_pointer(st_points_generated, rd_api_data_points_generated, (collected_number)gs.api_data_result_points_generated);
667 + rrddim_set_by_pointer(st_points_generated, rd_api_weights_points_generated, (collected_number)gs.api_weights_result_points_generated);
668 + rrddim_set_by_pointer(st_points_generated, rd_api_badges_points_generated, (collected_number)gs.api_badges_result_points_generated);
669 + rrddim_set_by_pointer(st_points_generated, rd_health_points_generated, (collected_number)gs.health_result_points_generated);
670 + rrddim_set_by_pointer(st_points_generated, rd_ml_points_generated, (collected_number)gs.ml_result_points_generated);
671 + rrddim_set_by_pointer(st_points_generated, rd_replication_points_generated, (collected_number)replication.points_generated);
672 +
673 + rrdset_done(st_points_generated);
674 + }
675 +
676 + // ----------------------------------------------------------------
677 +
678 + {
679 + static RRDSET *st_points_stored = NULL;
680 + static RRDDIM *rds[RRD_STORAGE_TIERS] = {};
681 +
682 + if (unlikely(!st_points_stored)) {
683 + st_points_stored = rrdset_create_localhost(
684 + "netdata"
685 + , "db_points_stored"
686 + , NULL
687 + , "queries"
688 + , NULL
689 + , "Netdata DB Points Stored"
690 + , "points/s"
691 + , "netdata"
692 + , "stats"
693 + , 131003
694 + , localhost->rrd_update_every
695 + , RRDSET_TYPE_STACKED
696 );
697
479 - rd_points_read = rrddim_add(st_rrdr_points, "read", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
480 - rd_points_generated = rrddim_add(st_rrdr_points, "generated", NULL, -1, 1, RRD_ALGORITHM_INCREMENTAL);
698 + for(size_t tier = 0; tier < storage_tiers ;tier++) {
699 + char buf[30 + 1];
700 + snprintfz(buf, 30, "tier%zu", tier);
701 + rds[tier] = rrddim_add(st_points_stored, buf, NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
702 + }
703 }
704
483 - rrddim_set_by_pointer(st_rrdr_points, rd_points_read, (collected_number)gs.rrdr_db_points_read);
484 - rrddim_set_by_pointer(st_rrdr_points, rd_points_generated, (collected_number)gs.rrdr_result_points_generated);
705 + for(size_t tier = 0; tier < storage_tiers ;tier++)
706 + rrddim_set_by_pointer(st_points_stored, rds[tier], (collected_number)gs.db_points_stored_per_tier[tier]);
707
486 - rrdset_done(st_rrdr_points);
708 + rrdset_done(st_points_stored);
709 }
710
711 // ----------------------------------------------------------------
@@ -2440,6 +2662,19 @@ static void worker_utilization_finish(void) {
2662 }
2663
2664 // ---------------------------------------------------------------------------------------------------------------------
2665 +// global statistics thread
2666 +
2667 +
2668 +static void global_statistics_register_workers(void) {
2669 + worker_register("STATS");
2670 + worker_register_job_name(WORKER_JOB_GLOBAL, "global");
2671 + worker_register_job_name(WORKER_JOB_REGISTRY, "registry");
2672 + worker_register_job_name(WORKER_JOB_WORKERS, "workers");
2673 + worker_register_job_name(WORKER_JOB_DBENGINE, "dbengine");
2674 + worker_register_job_name(WORKER_JOB_STRINGS, "strings");
2675 + worker_register_job_name(WORKER_JOB_DICTIONARIES, "dictionaries");
2676 + worker_register_job_name(WORKER_JOB_MALLOC_TRACE, "malloc_trace");
2677 +}
2678
2679 static void global_statistics_cleanup(void *ptr)
2680 {
@@ -2457,14 +2692,7 @@ static void global_statistics_cleanup(void *ptr)
2692
2693 void *global_statistics_main(void *ptr)
2694 {
2460 - worker_register("STATS");
2461 - worker_register_job_name(WORKER_JOB_GLOBAL, "global");
2462 - worker_register_job_name(WORKER_JOB_REGISTRY, "registry");
2463 - worker_register_job_name(WORKER_JOB_WORKERS, "workers");
2464 - worker_register_job_name(WORKER_JOB_DBENGINE, "dbengine");
2465 - worker_register_job_name(WORKER_JOB_STRINGS, "strings");
2466 - worker_register_job_name(WORKER_JOB_DICTIONARIES, "dictionaries");
2467 - worker_register_job_name(WORKER_JOB_MALLOC_TRACE, "malloc_trace");
2695 + global_statistics_register_workers();
2696
2697 netdata_thread_cleanup_push(global_statistics_cleanup, ptr);
2698
@@ -2485,9 +2713,6 @@ void *global_statistics_main(void *ptr)
2713 worker_is_idle();
2714 heartbeat_next(&hb, step);
2715
2488 - worker_is_busy(WORKER_JOB_WORKERS);
2489 - worker_utilization_charts();
2490 -
2716 worker_is_busy(WORKER_JOB_GLOBAL);
2717 global_statistics_charts();
2718
@@ -2517,3 +2742,49 @@ void *global_statistics_main(void *ptr)
2742 netdata_thread_cleanup_pop(1);
2743 return NULL;
2744 }
2745 +
2746 +
2747 +// ---------------------------------------------------------------------------------------------------------------------
2748 +// workers thread
2749 +
2750 +static void global_statistics_workers_cleanup(void *ptr)
2751 +{
2752 + worker_unregister();
2753 +
2754 + struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
2755 + static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
2756 +
2757 + info("cleaning up...");
2758 +
2759 + worker_utilization_finish();
2760 +
2761 + static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
2762 +}
2763 +
2764 +void *global_statistics_workers_main(void *ptr)
2765 +{
2766 + global_statistics_register_workers();
2767 +
2768 + netdata_thread_cleanup_push(global_statistics_workers_cleanup, ptr);
2769 +
2770 + int update_every =
2771 + (int)config_get_number(CONFIG_SECTION_GLOBAL_STATISTICS, "update every", localhost->rrd_update_every);
2772 + if (update_every < localhost->rrd_update_every)
2773 + update_every = localhost->rrd_update_every;
2774 +
2775 + usec_t step = update_every * USEC_PER_SEC;
2776 + heartbeat_t hb;
2777 + heartbeat_init(&hb);
2778 +
2779 + while (!netdata_exit) {
2780 + worker_is_idle();
2781 + heartbeat_next(&hb, step);
2782 +
2783 + worker_is_busy(WORKER_JOB_WORKERS);
2784 + worker_utilization_charts();
2785 + }
2786 +
2787 + netdata_thread_cleanup_pop(1);
2788 + return NULL;
2789 +}
2790 +
daemon/global_statistics.h
+17 -13
@@ -3,23 +3,27 @@
3 #ifndef NETDATA_GLOBAL_STATISTICS_H
4 #define NETDATA_GLOBAL_STATISTICS_H 1
5
6 -#include "common.h"
6 +#include "database/rrd.h"
7
8 // ----------------------------------------------------------------------------
9 // global statistics
10
11 -void rrdr_query_completed(uint64_t db_points_read, uint64_t result_points_generated);
12 -void sqlite3_query_completed(bool success, bool busy, bool locked);
13 -void sqlite3_row_completed(void);
14 -
15 -void finished_web_request_statistics(uint64_t dt,
16 - uint64_t bytes_received,
17 - uint64_t bytes_sent,
18 - uint64_t content_size,
19 - uint64_t compressed_content_size);
20 -
21 -uint64_t web_client_connected(void);
22 -void web_client_disconnected(void);
11 +void global_statistics_ml_query_completed(size_t points_read);
12 +void global_statistics_exporters_query_completed(size_t points_read);
13 +void global_statistics_backfill_query_completed(size_t points_read);
14 +void global_statistics_rrdr_query_completed(size_t queries, uint64_t db_points_read, uint64_t result_points_generated, QUERY_SOURCE query_source);
15 +void global_statistics_sqlite3_query_completed(bool success, bool busy, bool locked);
16 +void global_statistics_sqlite3_row_completed(void);
17 +void global_statistics_rrdset_done_chart_collection_completed(size_t *points_read_per_tier_array);
18 +
19 +void global_statistics_web_request_completed(uint64_t dt,
20 + uint64_t bytes_received,
21 + uint64_t bytes_sent,
22 + uint64_t content_size,
23 + uint64_t compressed_content_size);
24 +
25 +uint64_t global_statistics_web_client_connected(void);
26 +void global_statistics_web_client_disconnected(void);
27
28 extern bool global_statistics_enabled;
29
daemon/static_threads.c
+12
@@ -6,6 +6,7 @@ void *aclk_main(void *ptr);
6 void *analytics_main(void *ptr);
7 void *cpuidlejitter_main(void *ptr);
8 void *global_statistics_main(void *ptr);
9 +void *global_statistics_workers_main(void *ptr);
10 void *health_main(void *ptr);
11 void *pluginsd_main(void *ptr);
12 void *service_main(void *ptr);
@@ -54,6 +55,17 @@ const struct netdata_static_thread static_threads_common[] = {
55 .init_routine = NULL,
56 .start_routine = global_statistics_main
57 },
58 + {
59 + .name = "WORKERS_STATS",
60 + .config_section = CONFIG_SECTION_PLUGINS,
61 + .config_name = "netdata monitoring",
62 + .env_name = "NETDATA_INTERNALS_MONITORING",
63 + .global_variable = &global_statistics_enabled,
64 + .enabled = 1,
65 + .thread = NULL,
66 + .init_routine = NULL,
67 + .start_routine = global_statistics_workers_main
68 + },
69 {
70 .name = "PLUGINSD",
71 .config_section = NULL,
daemon/unit_test.c
+2 -2
@@ -1938,7 +1938,7 @@ static int test_dbengine_check_rrdr(RRDSET *st[CHARTS], RRDDIM *rd[CHARTS][DIMS]
1938 ONEWAYALLOC *owa = onewayalloc_create(0);
1939 RRDR *r = rrd2rrdr_legacy(owa, st[i], points, time_start, time_end,
1940 RRDR_GROUPING_AVERAGE, 0, RRDR_OPTION_NATURAL_POINTS,
1941 - NULL, NULL, 0, 0);
1941 + NULL, NULL, 0, 0, QUERY_SOURCE_UNITTEST);
1942 if (!r) {
1943 fprintf(stderr, " DB-engine unittest %s: empty RRDR on region %d ### E R R O R ###\n", rrdset_name(st[i]), current_region);
1944 return ++errors;
@@ -2076,7 +2076,7 @@ int test_dbengine(void)
2076 ONEWAYALLOC *owa = onewayalloc_create(0);
2077 RRDR *r = rrd2rrdr_legacy(owa, st[i], points, time_start[0] + update_every,
2078 time_end[REGIONS - 1], RRDR_GROUPING_AVERAGE, 0,
2079 - RRDR_OPTION_NATURAL_POINTS, NULL, NULL, 0, 0);
2079 + RRDR_OPTION_NATURAL_POINTS, NULL, NULL, 0, 0, QUERY_SOURCE_UNITTEST);
2080
2081 if (!r) {
2082 fprintf(stderr, " DB-engine unittest %s: empty RRDR ### E R R O R ###\n", rrdset_name(st[i]));
database/rrd.h
+10
@@ -33,6 +33,16 @@ typedef struct rrddim_acquired RRDDIM_ACQUIRED;
33 typedef void *ml_host_t;
34 typedef void *ml_dimension_t;
35
36 +typedef enum {
37 + QUERY_SOURCE_UNKNOWN,
38 + QUERY_SOURCE_API_DATA,
39 + QUERY_SOURCE_API_BADGE,
40 + QUERY_SOURCE_API_WEIGHTS,
41 + QUERY_SOURCE_HEALTH,
42 + QUERY_SOURCE_ML,
43 + QUERY_SOURCE_UNITTEST,
44 +} QUERY_SOURCE;
45 +
46 // forward declarations
47 struct rrddim_tier;
48
database/rrdcontext.h
+1
@@ -171,6 +171,7 @@ typedef struct query_target_request {
171 const char *group_options;
172 time_t resampling_time;
173 size_t tier;
174 + QUERY_SOURCE query_source;
175 } QUERY_TARGET_REQUEST;
176
177 typedef struct query_target {
database/rrdset.c
+12 -2
@@ -1046,12 +1046,14 @@ static inline usec_t rrdset_init_last_updated_time(RRDSET *st) {
1046 return last_updated_ut;
1047 }
1048
1049 +static __thread size_t rrdset_done_statistics_points_stored_per_tier[RRD_STORAGE_TIERS];
1050 +
1051 static inline time_t tier_next_point_time(RRDDIM *rd, struct rrddim_tier *t, time_t now) {
1052 time_t loop = (time_t)rd->update_every * (time_t)t->tier_grouping;
1053 return now + loop - ((now + loop) % loop);
1054 }
1055
1054 -void store_metric_at_tier(RRDDIM *rd, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut __maybe_unused) {
1056 +void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut __maybe_unused) {
1057 if (unlikely(!t->next_point_time))
1058 t->next_point_time = tier_next_point_time(rd, t, sp.end_time);
1059
@@ -1079,6 +1081,7 @@ void store_metric_at_tier(RRDDIM *rd, struct rrddim_tier *t, STORAGE_POINT sp, u
1081 0, SN_FLAG_NONE);
1082 }
1083
1084 + rrdset_done_statistics_points_stored_per_tier[tier]++;
1085 t->virtual_point.count = 0; // make the point unset
1086 t->next_point_time = tier_next_point_time(rd, t, sp.end_time);
1087 }
@@ -1141,6 +1144,7 @@ void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n,
1144
1145 // store the metric on tier 0
1146 rd->tiers[0]->collect_ops->store_metric(rd->tiers[0]->db_collection_handle, point_end_time_ut, n, 0, 0, 1, 0, flags);
1147 + rrdset_done_statistics_points_stored_per_tier[0]++;
1148
1149 time_t now = (time_t)(point_end_time_ut / USEC_PER_SEC);
1150
@@ -1167,10 +1171,14 @@ void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n,
1171 rrddim_option_set(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS);
1172 }
1173
1170 - store_metric_at_tier(rd, t, sp, point_end_time_ut);
1174 + store_metric_at_tier(rd, tier, t, sp, point_end_time_ut);
1175 }
1176 }
1177
1178 +void store_metric_collection_completed() {
1179 + global_statistics_rrdset_done_chart_collection_completed(rrdset_done_statistics_points_stored_per_tier);
1180 +}
1181 +
1182 // caching of dimensions rrdset_done() and rrdset_done_interpolate() loop through
1183 struct rda_item {
1184 const DICTIONARY_ITEM *item;
@@ -1934,6 +1942,8 @@ after_second_database_work:
1942 rrdcontext_collected_rrdset(st);
1943
1944 netdata_thread_enable_cancelability();
1945 +
1946 + store_metric_collection_completed();
1947 }
1948
1949 time_t rrdset_set_update_every(RRDSET *st, time_t update_every) {
database/sqlite/sqlite_functions.c
+4 -4
@@ -71,7 +71,7 @@ SQLITE_API int sqlite3_exec_monitored(
71 char **errmsg /* Error msg written here */
72 ) {
73 int rc = sqlite3_exec(db, sql, callback, data, errmsg);
74 - sqlite3_query_completed(rc == SQLITE_OK, rc == SQLITE_BUSY, rc == SQLITE_LOCKED);
74 + global_statistics_sqlite3_query_completed(rc == SQLITE_OK, rc == SQLITE_BUSY, rc == SQLITE_LOCKED);
75 return rc;
76 }
77
@@ -83,14 +83,14 @@ SQLITE_API int sqlite3_step_monitored(sqlite3_stmt *stmt) {
83 rc = sqlite3_step(stmt);
84 switch (rc) {
85 case SQLITE_DONE:
86 - sqlite3_query_completed(1, 0, 0);
86 + global_statistics_sqlite3_query_completed(1, 0, 0);
87 break;
88 case SQLITE_ROW:
89 - sqlite3_row_completed();
89 + global_statistics_sqlite3_row_completed();
90 break;
91 case SQLITE_BUSY:
92 case SQLITE_LOCKED:
93 - sqlite3_query_completed(rc == SQLITE_DONE, rc == SQLITE_BUSY, rc == SQLITE_LOCKED);
93 + global_statistics_sqlite3_query_completed(rc == SQLITE_DONE, rc == SQLITE_BUSY, rc == SQLITE_LOCKED);
94 usleep(SQLITE_INSERT_DELAY * USEC_PER_MS);
95 continue;
96 default:
exporting/process_data.c
+3
@@ -122,11 +122,13 @@ NETDATA_DOUBLE exporting_calculate_value_from_stored_data(
122
123 *last_timestamp = before;
124
125 + size_t points_read = 0;
126 size_t counter = 0;
127 NETDATA_DOUBLE sum = 0;
128
129 for (rd->tiers[0]->query_ops->init(rd->tiers[0]->db_metric_handle, &handle, after, before); !rd->tiers[0]->query_ops->is_finished(&handle);) {
130 STORAGE_POINT sp = rd->tiers[0]->query_ops->next_metric(&handle);
131 + points_read++;
132
133 if (unlikely(storage_point_is_empty(sp))) {
134 // not collected
@@ -137,6 +139,7 @@ NETDATA_DOUBLE exporting_calculate_value_from_stored_data(
139 counter += sp.count;
140 }
141 rd->tiers[0]->query_ops->finalize(&handle);
142 + global_statistics_exporters_query_completed(points_read);
143
144 if (unlikely(!counter)) {
145 debug(
health/health.c
+2 -1
@@ -1039,7 +1039,8 @@ void *health_main(void *ptr) {
1039 0, rc->options,
1040 &rc->db_after,&rc->db_before,
1041 NULL, NULL, NULL,
1042 - &value_is_null, NULL, 0, 0);
1042 + &value_is_null, NULL, 0, 0,
1043 + QUERY_SOURCE_HEALTH);
1044
1045 if (unlikely(ret != 200)) {
1046 // database lookup failed
libnetdata/clocks/clocks.c
+8 -7
@@ -341,20 +341,21 @@ usec_t heartbeat_next(heartbeat_t *hb, usec_t tick) {
341 void sleep_usec(usec_t usec) {
342 // we expect microseconds (1.000.000 per second)
343 // but timespec is nanoseconds (1.000.000.000 per second)
344 - struct timespec rem, req = {
344 + struct timespec rem = { 0, 0 }, req = {
345 .tv_sec = (time_t) (usec / USEC_PER_SEC),
346 .tv_nsec = (suseconds_t) ((usec % USEC_PER_SEC) * NSEC_PER_USEC)
347 };
348
349 #ifdef __linux__
350 - while ((errno = clock_nanosleep(CLOCK_REALTIME, 0, &req, &rem)) != 0) {
350 + while (clock_nanosleep(CLOCK_REALTIME, 0, &req, &rem) != 0) {
351 #else
352 - while ((errno = nanosleep(&req, &rem)) != 0) {
352 + while (nanosleep(&req, &rem) != 0) {
353 #endif
354 - if (likely(errno == EINTR)) {
355 - req.tv_sec = rem.tv_sec;
356 - req.tv_nsec = rem.tv_nsec;
357 - } else {
354 + if (likely(errno == EINTR && (rem.tv_sec || rem.tv_nsec))) {
355 + req = rem;
356 + rem = (struct timespec){ 0, 0 };
357 + }
358 + else {
359 #ifdef __linux__
360 error("Cannot clock_nanosleep(CLOCK_REALTIME) for %llu microseconds.", usec);
361 #else
libnetdata/dictionary/dictionary.c
+9 -6
@@ -199,7 +199,10 @@ static int item_check_and_acquire_advanced(DICTIONARY *dict, DICTIONARY_ITEM *it
199 #define item_is_not_referenced_and_can_be_removed(dict, item) (item_is_not_referenced_and_can_be_removed_advanced(dict, item) == RC_ITEM_OK)
200 static inline int item_is_not_referenced_and_can_be_removed_advanced(DICTIONARY *dict, DICTIONARY_ITEM *item);
201
202 -static inline void pointer_index_init(DICTIONARY *dict) {
202 +// ----------------------------------------------------------------------------
203 +// validate each pointer is indexed once - internal checks only
204 +
205 +static inline void pointer_index_init(DICTIONARY *dict __maybe_unused) {
206 #ifdef NETDATA_INTERNAL_CHECKS
207 netdata_mutex_init(&dict->global_pointer_registry_mutex);
208 #else
@@ -207,7 +210,7 @@ static inline void pointer_index_init(DICTIONARY *dict) {
210 #endif
211 }
212
210 -static inline void pointer_destroy_index(DICTIONARY *dict) {
213 +static inline void pointer_destroy_index(DICTIONARY *dict __maybe_unused) {
214 #ifdef NETDATA_INTERNAL_CHECKS
215 netdata_mutex_lock(&dict->global_pointer_registry_mutex);
216 JudyHSFreeArray(&dict->global_pointer_registry, PJE0);
@@ -216,7 +219,7 @@ static inline void pointer_destroy_index(DICTIONARY *dict) {
219 ;
220 #endif
221 }
219 -static inline void pointer_add(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item) {
222 +static inline void pointer_add(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item __maybe_unused) {
223 #ifdef NETDATA_INTERNAL_CHECKS
224 netdata_mutex_lock(&dict->global_pointer_registry_mutex);
225 Pvoid_t *PValue = JudyHSIns(&dict->global_pointer_registry, &item, sizeof(void *), PJE0);
@@ -229,7 +232,7 @@ static inline void pointer_add(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM
232 #endif
233 }
234
232 -static inline void pointer_check(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item) {
235 +static inline void pointer_check(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item __maybe_unused) {
236 #ifdef NETDATA_INTERNAL_CHECKS
237 netdata_mutex_lock(&dict->global_pointer_registry_mutex);
238 Pvoid_t *PValue = JudyHSGet(dict->global_pointer_registry, &item, sizeof(void *));
@@ -241,7 +244,7 @@ static inline void pointer_check(DICTIONARY *dict __maybe_unused, DICTIONARY_ITE
244 #endif
245 }
246
244 -static inline void pointer_del(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item) {
247 +static inline void pointer_del(DICTIONARY *dict __maybe_unused, DICTIONARY_ITEM *item __maybe_unused) {
248 #ifdef NETDATA_INTERNAL_CHECKS
249 netdata_mutex_lock(&dict->global_pointer_registry_mutex);
250 int ret = JudyHSDel(&dict->global_pointer_registry, &item, sizeof(void *), PJE0);
@@ -413,7 +416,7 @@ static inline void DICTIONARY_ENTRIES_MINUS1(DICTIONARY *dict) {
416 __atomic_fetch_add(&dict->stats->ops.deletes, 1, __ATOMIC_RELAXED);
417 __atomic_fetch_sub(&dict->stats->items.entries, 1, __ATOMIC_RELAXED);
418
416 - size_t entries;
419 + size_t entries; (void)entries;
420 if(unlikely(is_dictionary_single_threaded(dict))) {
421 dict->version++;
422 entries = dict->entries++;
libnetdata/locks/locks.c
+30
@@ -278,6 +278,36 @@ int __netdata_rwlock_trywrlock(netdata_rwlock_t *rwlock) {
278 return ret;
279 }
280
281 +// ----------------------------------------------------------------------------
282 +// spinlock implementation
283 +// https://www.youtube.com/watch?v=rmGJc9PXpuE&t=41s
284 +
285 +void netdata_spinlock_init(SPINLOCK *spinlock) {
286 + *spinlock = NETDATA_SPINLOCK_INITIALIZER;
287 +}
288 +
289 +void netdata_spinlock_lock(SPINLOCK *spinlock) {
290 + static const struct timespec ns = { .tv_sec = 0, .tv_nsec = 1 };
291 + bool expected = false, desired = true;
292 +
293 + for(int i = 1;
294 + __atomic_load_n(&spinlock->locked, __ATOMIC_RELAXED) ||
295 + !__atomic_compare_exchange_n(&spinlock->locked, &expected, desired, false, __ATOMIC_ACQUIRE, __ATOMIC_ACQUIRE)
296 + ; i++
297 + ) {
298 +
299 + if(unlikely(i == 8)) {
300 + i = 0;
301 + nanosleep(&ns, NULL);
302 + }
303 + }
304 + // we have the lock
305 +}
306 +
307 +void netdata_spinlock_unlock(SPINLOCK *spinlock) {
308 + __atomic_store_n(&spinlock->locked, false, __ATOMIC_RELEASE);
309 +}
310 +
311 #ifdef NETDATA_TRACE_RWLOCKS
312
313 // ----------------------------------------------------------------------------
libnetdata/locks/locks.h
+8
@@ -9,6 +9,14 @@
9 typedef pthread_mutex_t netdata_mutex_t;
10 #define NETDATA_MUTEX_INITIALIZER PTHREAD_MUTEX_INITIALIZER
11
12 +typedef struct netdata_spinlock {
13 + bool locked;
14 +} SPINLOCK;
15 +#define NETDATA_SPINLOCK_INITIALIZER (SPINLOCK){ .locked = false }
16 +void netdata_spinlock_init(SPINLOCK *spinlock);
17 +void netdata_spinlock_lock(SPINLOCK *spinlock);
18 +void netdata_spinlock_unlock(SPINLOCK *spinlock);
19 +
20 #ifdef NETDATA_TRACE_RWLOCKS
21 typedef struct netdata_rwlock_locker {
22 pid_t pid;
ml/ADCharts.cc
+2 -1
@@ -140,7 +140,8 @@ void ml::updateHostAndDetectionRateCharts(RRDHOST *RH, collected_number AnomalyR
140 Options, "anomaly_rate",
141 NULL /* group options */,
142 0, /* timeout */
143 - 0 /* tier */
143 + 0, /* tier */
144 + QUERY_SOURCE_ML
145 );
146 if(R) {
147 assert(R->d == 1 && R->n == 1 && R->rows == 1);
ml/Query.h
+7 -1
@@ -22,6 +22,7 @@ public:
22 void init(time_t AfterT, time_t BeforeT) {
23 Ops->init(RD->tiers[0]->db_metric_handle, &Handle, AfterT, BeforeT);
24 Initialized = true;
25 + points_read = 0;
26 }
27
28 bool isFinished() {
@@ -29,11 +30,15 @@ public:
30 }
31
32 ~Query() {
32 - if (Initialized)
33 + if (Initialized) {
34 Ops->finalize(&Handle);
35 + global_statistics_ml_query_completed(points_read);
36 + points_read = 0;
37 + }
38 }
39
40 std::pair<time_t, CalculatedNumber> nextMetric() {
41 + points_read++;
42 STORAGE_POINT sp = Ops->next_metric(&Handle);
43 return { sp.start_time, sp.sum / sp.count };
44 }
@@ -41,6 +46,7 @@ public:
46 private:
47 RRDDIM *RD;
48 bool Initialized;
49 + size_t points_read;
50
51 struct storage_engine_query_ops *Ops;
52 struct storage_engine_query_handle Handle;
streaming/replication.c
+171 -94
@@ -6,23 +6,36 @@
6 #define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 20
7 #define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10
8
9 +static struct replication_query_statistics replication_queries = {
10 + .queries_started = 0,
11 + .queries_finished = 0,
12 + .points_read = 0,
13 + .points_generated = 0,
14 +};
15 +
16 +struct replication_query_statistics replication_get_query_statistics(void) {
17 + return replication_queries;
18 +}
19 +
20 // ----------------------------------------------------------------------------
21 // sending replication replies
22
23 +struct replication_dimension {
24 + STORAGE_POINT sp;
25 + struct storage_engine_query_handle handle;
26 + bool enabled;
27 +
28 + DICTIONARY *dict;
29 + const DICTIONARY_ITEM *rda;
30 + RRDDIM *rd;
31 +};
32 +
33 static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, time_t before, bool enable_streaming, time_t wall_clock_time) {
34 size_t dimensions = rrdset_number_of_dimensions(st);
35 + size_t points_read = 0, points_generated = 0;
36
37 struct storage_engine_query_ops *ops = &st->rrdhost->db[0].eng->api.query_ops;
16 -
17 - struct {
18 - DICTIONARY *dict;
19 - const DICTIONARY_ITEM *rda;
20 - RRDDIM *rd;
21 - struct storage_engine_query_handle handle;
22 - STORAGE_POINT sp;
23 - bool enabled;
24 - } data[dimensions];
25 -
38 + struct replication_dimension data[dimensions];
39 memset(data, 0, sizeof(data));
40
41 if(enable_streaming && st->last_updated.tv_sec > before) {
@@ -38,23 +51,23 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
51 {
52 RRDDIM *rd;
53 rrddim_foreach_read(rd, st) {
41 - if (rd_dfe.counter >= dimensions) {
54 + if(unlikely(!rd || !rd_dfe.item || !rd->exposed))
55 + continue;
56 +
57 + if (unlikely(rd_dfe.counter >= dimensions)) {
58 internal_error(true, "STREAM_SENDER REPLAY ERROR: 'host:%s/chart:%s' has more dimensions than the replicated ones",
59 rrdhost_hostname(st->rrdhost), rrdset_id(st));
60 break;
61 }
62
47 - if(rd->exposed) {
48 - data[rd_dfe.counter].dict = rd_dfe.dict;
49 - data[rd_dfe.counter].rda = dictionary_acquired_item_dup(rd_dfe.dict, rd_dfe.item);
50 - data[rd_dfe.counter].rd = rd;
63 + struct replication_dimension *d = &data[rd_dfe.counter];
64
52 - ops->init(rd->tiers[0]->db_metric_handle, &data[rd_dfe.counter].handle, after, before);
65 + d->dict = rd_dfe.dict;
66 + d->rda = dictionary_acquired_item_dup(rd_dfe.dict, rd_dfe.item);
67 + d->rd = rd;
68
54 - data[rd_dfe.counter].enabled = true;
55 - }
56 - else
57 - data[rd_dfe.counter].enabled = false;
69 + ops->init(rd->tiers[0]->db_metric_handle, &d->handle, after, before);
70 + d->enabled = true;
71 }
72 rrddim_foreach_done(rd);
73 }
@@ -62,32 +75,35 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
75 time_t now = after + 1, actual_after = 0, actual_before = 0; (void)actual_before;
76 while(now <= before) {
77 time_t min_start_time = 0, min_end_time = 0;
65 - for (size_t i = 0; i < dimensions && data[i].rd; i++) {
66 - if(!data[i].enabled) continue;
78 + for (size_t i = 0; i < dimensions ;i++) {
79 + struct replication_dimension *d = &data[i];
80 + if(unlikely(!d->enabled)) continue;
81
82 // fetch the first valid point for the dimension
83 int max_skip = 100;
70 - while(data[i].sp.end_time < now && !ops->is_finished(&data[i].handle) && max_skip-- > 0)
71 - data[i].sp = ops->next_metric(&data[i].handle);
84 + while(d->sp.end_time < now && !ops->is_finished(&d->handle) && max_skip-- > 0) {
85 + d->sp = ops->next_metric(&d->handle);
86 + points_read++;
87 + }
88
89 internal_error(max_skip <= 0,
90 "STREAM_SENDER REPLAY ERROR: 'host:%s/chart:%s/dim:%s': db does not advance the query beyond time %llu",
75 - rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(data[i].rd), (unsigned long long) now);
91 + rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(d->rd), (unsigned long long) now);
92
77 - if(data[i].sp.end_time < now)
93 + if(unlikely(d->sp.end_time < now || storage_point_is_unset(d->sp) || storage_point_is_empty(d->sp)))
94 continue;
95
80 - if(!min_start_time) {
81 - min_start_time = data[i].sp.start_time;
82 - min_end_time = data[i].sp.end_time;
96 + if(unlikely(!min_start_time)) {
97 + min_start_time = d->sp.start_time;
98 + min_end_time = d->sp.end_time;
99 }
100 else {
85 - min_start_time = MIN(min_start_time, data[i].sp.start_time);
86 - min_end_time = MIN(min_end_time, data[i].sp.end_time);
101 + min_start_time = MIN(min_start_time, d->sp.start_time);
102 + min_end_time = MIN(min_end_time, d->sp.end_time);
103 }
104 }
105
90 - if(min_start_time > wall_clock_time + 1 || min_end_time > wall_clock_time + st->update_every + 1) {
106 + if(unlikely(min_start_time > wall_clock_time + 1 || min_end_time > wall_clock_time + st->update_every + 1)) {
107 internal_error(true,
108 "STREAM_SENDER REPLAY ERROR: 'host:%s/chart:%s': db provided future start time %llu or end time %llu (now is %llu)",
109 rrdhost_hostname(st->rrdhost), rrdset_id(st),
@@ -97,7 +113,7 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
113 break;
114 }
115
100 - if(min_end_time < now) {
116 + if(unlikely(min_end_time < now)) {
117 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
118 internal_error(true,
119 "STREAM_SENDER REPLAY: 'host:%s/chart:%s': no data on any dimension beyond time %llu",
@@ -106,10 +122,10 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
122 break;
123 }
124
109 - if(min_end_time <= min_start_time)
125 + if(unlikely(min_end_time <= min_start_time))
126 min_start_time = min_end_time - st->update_every;
127
112 - if(!actual_after) {
128 + if(unlikely(!actual_after)) {
129 actual_after = min_end_time;
130 actual_before = min_end_time;
131 }
@@ -123,15 +139,19 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
139 );
140
141 // output the replay values for this time
126 - for (size_t i = 0; i < dimensions && data[i].rd; i++) {
127 - if(!data[i].enabled) continue;
142 + for (size_t i = 0; i < dimensions ;i++) {
143 + struct replication_dimension *d = &data[i];
144 + if(unlikely(!d->enabled)) continue;
145
129 - if(data[i].sp.start_time <= min_end_time && data[i].sp.end_time >= min_end_time)
146 + if(likely(d->sp.start_time <= min_end_time && d->sp.end_time >= min_end_time))
147 buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_SET " \"%s\" " NETDATA_DOUBLE_FORMAT " \"%s\"\n",
131 - rrddim_id(data[i].rd), data[i].sp.sum, data[i].sp.flags & SN_FLAG_RESET ? "R" : "");
148 + rrddim_id(d->rd), d->sp.sum, d->sp.flags & SN_FLAG_RESET ? "R" : "");
149 +
150 else
151 buffer_sprintf(wb, PLUGINSD_KEYWORD_REPLAY_SET " \"%s\" NAN \"E\"\n",
134 - rrddim_id(data[i].rd));
152 + rrddim_id(d->rd));
153 +
154 + points_generated++;
155 }
156
157 now = min_end_time + 1;
@@ -157,13 +177,22 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
177
178 // release all the dictionary items acquired
179 // finalize the queries
160 - for(size_t i = 0; i < dimensions && data[i].rda ;i++) {
161 - if(!data[i].enabled) continue;
180 + for(size_t i = 0; i < dimensions ;i++) {
181 + struct replication_dimension *d = &data[i];
182 + if(unlikely(!d->enabled)) continue;
183 +
184 + ops->finalize(&d->handle);
185 +
186 + dictionary_acquired_item_release(d->dict, d->rda);
187
163 - ops->finalize(&data[i].handle);
164 - dictionary_acquired_item_release(data[i].dict, data[i].rda);
188 + // update global statistics
189 + replication_queries.queries_started++;
190 + replication_queries.queries_finished++;
191 }
192
193 + replication_queries.points_read += points_read;
194 + replication_queries.points_generated += points_generated;
195 +
196 return before;
197 }
198
@@ -560,6 +589,8 @@ static struct replication_thread {
589 size_t sender_resets;
590 size_t waits;
591
592 + size_t skipped_no_room_last_run;
593 +
594 Pvoid_t JudyL_array;
595 } replication_globals = {
596 .mutex = NETDATA_MUTEX_INITIALIZER,
@@ -570,6 +601,7 @@ static struct replication_thread {
601 .first_time_t = 0,
602 .next_unique_id = 1,
603 .skipped_no_room = 0,
604 + .skipped_no_room_last_run = 0,
605 .skipped_not_connected = 0,
606 .sender_resets = 0,
607 .waits = 0,
@@ -599,6 +631,13 @@ static void replication_recursive_unlock() {
631 #endif
632 }
633
634 +void replication_set_next_point_in_time(time_t after, size_t unique_id) {
635 + replication_recursive_lock();
636 + replication_globals.last_after = after;
637 + replication_globals.last_unique_id = unique_id;
638 + replication_recursive_unlock();
639 +}
640 +
641 // ----------------------------------------------------------------------------
642 // replication sort entry management
643
@@ -626,10 +665,9 @@ static struct replication_sort_entry *replication_sort_entry_add(struct replicat
665
666 struct replication_sort_entry *rse = replication_sort_entry_create(rq);
667
629 - if(rq->after < (time_t)replication_globals.last_after) {
668 + if(rq->after < (time_t)replication_globals.last_after && rq->sender->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && !replication_globals.skipped_no_room_last_run) {
669 // make it find this request first
631 - replication_globals.last_after = rq->after;
632 - replication_globals.last_unique_id = rq->unique_id;
670 + replication_set_next_point_in_time(rq->after, rq->unique_id);
671 }
672
673 replication_globals.added++;
@@ -716,8 +754,9 @@ static struct replication_request replication_request_get_first_available() {
754 Pvoid_t *inner_judy_pptr;
755
756 replication_recursive_lock();
757 + replication_globals.skipped_no_room_last_run = 0;
758
720 - struct replication_request rq = (struct replication_request){ .found = false };
759 + struct replication_request rq_to_return = (struct replication_request){ .found = false };
760
761
762 if(unlikely(!replication_globals.last_after || !replication_globals.last_unique_id)) {
@@ -726,41 +765,50 @@ static struct replication_request replication_request_get_first_available() {
765 }
766
767 bool find_same_after = true;
729 - while(!rq.found && (inner_judy_pptr = JudyLFirstOrNext(replication_globals.JudyL_array, &replication_globals.last_after, find_same_after))) {
768 + while(!rq_to_return.found && (inner_judy_pptr = JudyLFirstOrNext(replication_globals.JudyL_array, &replication_globals.last_after, find_same_after))) {
769 Pvoid_t *our_item_pptr;
770
732 - while(!rq.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.last_unique_id, PJE0))) {
771 + while(!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.last_unique_id, PJE0))) {
772 struct replication_sort_entry *rse = *our_item_pptr;
734 - struct sender_state *s = rse->rq->sender;
773 + struct replication_request *rq = rse->rq;
774 + struct sender_state *s = rq->sender;
775
736 - bool sender_is_connected =
737 - rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
776 + if(likely(s->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)) {
777 + // there is room for this request in the sender buffer
778
739 - bool sender_has_been_flushed_since_this_request =
740 - rse->rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);
779 + bool sender_is_connected =
780 + rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
781
742 - bool sender_has_room_to_spare =
743 - s->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED;
782 + bool sender_has_been_flushed_since_this_request =
783 + rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);
784
745 - if(unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
746 - replication_globals.skipped_not_connected++;
747 - if(replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
748 - break;
749 - }
785 + if (unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
786 + // skip this request, the sender is not connected or it has reconnected
787 +
788 + replication_globals.skipped_not_connected++;
789 + if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
790 + // we removed the item from the outer JudyL
791 + break;
792 + }
793 + else {
794 + // this request is good to execute
795
751 - else if(sender_has_room_to_spare) {
752 - // copy the request to return it
753 - rq = *rse->rq;
754 - rq.chart_id = string_dup(rq.chart_id);
796 + // copy the request to return it
797 + rq_to_return = *rq;
798 + rq_to_return.chart_id = string_dup(rq_to_return.chart_id);
799
756 - // set the return result to found
757 - rq.found = true;
800 + // set the return result to found
801 + rq_to_return.found = true;
802
759 - if(replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
760 - break;
803 + if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
804 + // we removed the item from the outer JudyL
805 + break;
806 + }
807 }
762 - else
808 + else {
809 replication_globals.skipped_no_room++;
810 + replication_globals.skipped_no_room_last_run++;
811 + }
812 }
813
814 // call JudyLNext from now on
@@ -771,7 +819,7 @@ static struct replication_request replication_request_get_first_available() {
819 }
820
821 replication_recursive_unlock();
774 - return rq;
822 + return rq_to_return;
823 }
824
825 // ----------------------------------------------------------------------------
@@ -890,8 +938,7 @@ void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {
938 percentage <= MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED) {
939 s->replication_reached_max = false;
940 replication_recursive_lock();
893 - replication_globals.last_after = 0;
894 - replication_globals.last_unique_id = 0;
941 + replication_set_next_point_in_time(0, 0);
942 replication_globals.sender_resets++;
943 replication_recursive_unlock();
944 }
@@ -929,17 +976,18 @@ static void replication_main_cleanup(void *ptr) {
976 #define WORKER_JOB_CHECK_CONSISTENCY 15
977
978 #define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 10
979 +#define SECONDS_TO_RESET_POINT_IN_TIME 10
980
981 static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {
934 - if(host->sender) {
935 - size_t pending_requests = host->sender->replication_pending_requests;
936 - size_t dict_entries = dictionary_entries(host->sender->replication_requests);
937 -
938 - internal_error(
939 - !pending_requests && dict_entries,
940 - "REPLICATION SUMMARY: 'host:%s' reports %zu pending replication requests, but its chart replication index says there are %zu charts pending replication",
941 - rrdhost_hostname(host), pending_requests, dict_entries);
942 - }
982 + internal_error(
983 + host->sender &&
984 + !host->sender->replication_pending_requests &&
985 + dictionary_entries(host->sender->replication_requests) != 0,
986 + "REPLICATION SUMMARY: 'host:%s' reports %zu pending replication requests, but its chart replication index says there are %zu charts pending replication",
987 + rrdhost_hostname(host),
988 + host->sender->replication_pending_requests,
989 + dictionary_entries(host->sender->replication_requests)
990 + );
991
992 size_t ok = 0;
993 size_t errors = 0;
@@ -983,21 +1031,17 @@ static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {
1031 }
1032
1033 static void verify_all_hosts_charts_are_streaming_now(void) {
986 -#ifdef NETDATA_INTERNAL_CHECKS
1034 worker_is_busy(WORKER_JOB_CHECK_CONSISTENCY);
1035
1036 size_t errors = 0;
1037 RRDHOST *host;
991 - dfe_start_reentrant(rrdhost_root_index, host)
1038 + dfe_start_read(rrdhost_root_index, host)
1039 errors += verify_host_charts_are_streaming_now(host);
1040 dfe_done(host);
1041
1042 size_t executed = replication_globals.executed;
996 - internal_error(true, "REPLICATION SUMMARY: finished, executed %zu replication requests, %zu charts pending replication", executed - replication_globals.last_executed, errors);
1043 + info("REPLICATION SUMMARY: finished, executed %zu replication requests, %zu charts pending replication", executed - replication_globals.last_executed, errors);
1044 replication_globals.last_executed = executed;
998 -#else
999 - ;
1000 -#endif
1045 }
1046
1047 void *replication_thread_main(void *ptr __maybe_unused) {
@@ -1027,7 +1071,9 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1071
1072 time_t latest_first_time_t = 0;
1073 long run_verification_countdown = LONG_MAX; // LONG_MAX to prevent an initial verification when no replication ever took place
1074 + bool slow = true; // control the time we sleep - it has to start with true!
1075 usec_t last_now_mono_ut = now_monotonic_usec();
1076 + time_t replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; // restart from the beginning every 10 seconds
1077
1078 while(!netdata_exit) {
1079
@@ -1036,9 +1082,21 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1082 if(unlikely(now_mono_ut - last_now_mono_ut > default_rrd_update_every * USEC_PER_SEC)) {
1083 last_now_mono_ut = now_mono_ut;
1084
1085 + if(replication_reset_next_point_in_time_countdown-- == 0) {
1086 + // once per second, make it scan all the pending requests next time
1087 + replication_set_next_point_in_time(0, 0);
1088 + replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1089 + }
1090 +
1091 if(!replication_globals.pending && run_verification_countdown-- == 0) {
1040 - replication_globals.first_time_t = 0; // reset the statistics about completion percentage
1092 + // reset the statistics about completion percentage
1093 + replication_globals.first_time_t = 0;
1094 + latest_first_time_t = 0;
1095 +
1096 verify_all_hosts_charts_are_streaming_now();
1097 +
1098 + run_verification_countdown = LONG_MAX;
1099 + slow = true;
1100 }
1101
1102 worker_is_busy(WORKER_JOB_STATISTICS);
@@ -1068,17 +1126,36 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1126
1127 if(unlikely(!rq.found)) {
1128 // make it scan all the pending requests next time
1071 - replication_globals.last_after = 0;
1072 - replication_globals.last_unique_id = 0;
1129 + replication_set_next_point_in_time(0, 0);
1130 + replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1131 +
1132 + // the timeout also defines now frequently we will traverse all the pending requests
1133 + // when the outbound buffers of all senders is full
1134 + usec_t timeout;
1135 + if(slow)
1136 + // no work to be done, wait for a request to come in
1137 + timeout = 1000 * USEC_PER_MS;
1138 +
1139 + else if(replication_globals.pending > 0)
1140 + // there are pending requests waiting to be executed,
1141 + // but none could be executed at this time.
1142 + // try again after this time.
1143 + timeout = 100 * USEC_PER_MS;
1144
1074 - replication_globals.waits++;
1145 + else
1146 + // no pending requests, but there were requests recently (run_verification_countdown)
1147 + // so, try in a short time.
1148 + // if this is big, one chart replicating will be slow to finish (ping - pong just one chart)
1149 + timeout = 10 * USEC_PER_MS;
1150
1151 + replication_globals.waits++;
1152 worker_is_idle();
1077 - sleep_usec(((replication_globals.pending) ? 10 : 1000) * USEC_PER_MS);
1153 + sleep_usec(timeout);
1154 continue;
1155 }
1156
1157 run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;
1158 + slow = false;
1159
1160 // delete the request from the dictionary
1161 worker_is_busy(WORKER_JOB_DELETE_ENTRY);
streaming/replication.h
+9
@@ -5,6 +5,15 @@
5
6 #include "daemon/common.h"
7
8 +struct replication_query_statistics {
9 + size_t queries_started;
10 + size_t queries_finished;
11 + size_t points_read;
12 + size_t points_generated;
13 +};
14 +
15 +struct replication_query_statistics replication_get_query_statistics(void);
16 +
17 bool replicate_chart_response(RRDHOST *rh, RRDSET *rs, bool start_streaming, time_t after, time_t before);
18
19 typedef int (*send_command)(const char *txt, void *data);
web/api/badges/web_buffer_svg.c
+2 -1
@@ -1116,7 +1116,8 @@ int web_client_api_request_v1_badge(RRDHOST *host, struct web_client *w, char *u
1116 points, after, before, group, group_options, 0, options,
1117 NULL, &latest_timestamp,
1118 NULL, NULL, NULL,
1119 - &value_is_null, NULL, 0, 0);
1119 + &value_is_null, NULL, 0, 0,
1120 + QUERY_SOURCE_API_BADGE);
1121
1122 // if the value cannot be calculated, show empty badge
1123 if (ret != HTTP_RESP_OK) {
web/api/formatters/rrd2json.c
+3 -1
@@ -76,6 +76,7 @@ int rrdset2value_api_v1(
76 , NETDATA_DOUBLE *anomaly_rate
77 , time_t timeout
78 , size_t tier
79 + , QUERY_SOURCE query_source
80 ) {
81 int ret = HTTP_RESP_INTERNAL_SERVER_ERROR;
82
@@ -92,7 +93,8 @@ int rrdset2value_api_v1(
93 dimensions,
94 group_options,
95 timeout,
95 - tier);
96 + tier,
97 + query_source);
98
99 if(!r) {
100 if(value_is_null) *value_is_null = 1;
web/api/formatters/rrd2json.h
+1
@@ -78,6 +78,7 @@ int rrdset2value_api_v1(
78 , NETDATA_DOUBLE *anomaly_rate
79 , time_t timeout
80 , size_t tier
81 + , QUERY_SOURCE query_source
82 );
83
84 #endif /* NETDATA_RRD2JSON_H */
web/api/formatters/value/value.c
+2 -1
@@ -106,7 +106,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
106 struct rrdcontext_acquired *rca, struct rrdinstance_acquired *ria, struct rrdmetric_acquired *rma,
107 time_t after, time_t before,
108 RRDR_OPTIONS options, RRDR_GROUPING group_method, const char *group_options,
109 - size_t tier, time_t timeout
109 + size_t tier, time_t timeout, QUERY_SOURCE query_source
110 ) {
111 QUERY_TARGET_REQUEST qtr = {
112 .host = host,
@@ -121,6 +121,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
121 .group_options = group_options,
122 .tier = tier,
123 .timeout = timeout,
124 + .query_source = query_source,
125 };
126
127 ONEWAYALLOC *owa = onewayalloc_create(16 * 1024);
web/api/formatters/value/value.h
+1 -1
@@ -23,7 +23,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
23 struct rrdcontext_acquired *rca, struct rrdinstance_acquired *ria, struct rrdmetric_acquired *rma,
24 time_t after, time_t before,
25 RRDR_OPTIONS options, RRDR_GROUPING group_method, const char *group_options,
26 - size_t tier, time_t timeout
26 + size_t tier, time_t timeout, QUERY_SOURCE query_source
27 );
28
29 NETDATA_DOUBLE rrdr2value(RRDR *r, long i, RRDR_OPTIONS options, int *all_values_are_null, NETDATA_DOUBLE *anomaly_rate);
web/api/queries/query.c
+11 -11
@@ -1473,7 +1473,8 @@ static inline void rrd2rrdr_do_dimension(RRDR *r, size_t dim_id_in_rrdr) {
1473 // ----------------------------------------------------------------------------
1474 // fill the gap of a tier
1475
1476 -extern void store_metric_at_tier(RRDDIM *rd, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut);
1476 +void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut);
1477 +void store_metric_collection_completed(void);
1478
1479 void rrdr_fill_tier_gap_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now) {
1480 if(unlikely(tier >= storage_tiers)) return;
@@ -1494,8 +1495,6 @@ void rrdr_fill_tier_gap_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now)
1495
1496 struct storage_engine_query_handle handle;
1497
1497 - size_t all_points_read = 0;
1498 -
1498 // for each lower tier
1499 for(int tr = (int)tier - 1; tr >= 0 ;tr--){
1500 time_t smaller_tier_first_time = rd->tiers[tr]->query_ops->oldest_time(rd->tiers[tr]->db_metric_handle);
@@ -1508,27 +1507,26 @@ void rrdr_fill_tier_gap_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now)
1507 struct rrddim_tier *tmp = rd->tiers[tr];
1508 tmp->query_ops->init(tmp->db_metric_handle, &handle, after_wanted, before_wanted);
1509
1511 - size_t points = 0;
1510 + size_t points_read = 0;
1511
1512 while(!tmp->query_ops->is_finished(&handle)) {
1513
1514 STORAGE_POINT sp = tmp->query_ops->next_metric(&handle);
1515 + points_read++;
1516
1517 if(sp.end_time > latest_time_t) {
1518 latest_time_t = sp.end_time;
1519 - store_metric_at_tier(rd, t, sp, sp.end_time * USEC_PER_SEC);
1520 - points++;
1519 + store_metric_at_tier(rd, tr, t, sp, sp.end_time * USEC_PER_SEC);
1520 }
1521 }
1522
1524 - all_points_read += points;
1523 tmp->query_ops->finalize(&handle);
1524 + store_metric_collection_completed();
1525 + global_statistics_backfill_query_completed(points_read);
1526
1527 //internal_error(true, "DBENGINE: backfilled chart '%s', dimension '%s', tier %d, from %ld to %ld, with %zu points from tier %d",
1528 // rd->rrdset->name, rd->name, tier, after_wanted, before_wanted, points, tr);
1529 }
1530 -
1531 - rrdr_query_completed(all_points_read, all_points_read);
1530 }
1531
1532 // ----------------------------------------------------------------------------
@@ -1977,7 +1975,7 @@ RRDR *rrd2rrdr_legacy(
1975 ONEWAYALLOC *owa,
1976 RRDSET *st, size_t points, time_t after, time_t before,
1977 RRDR_GROUPING group_method, time_t resampling_time, RRDR_OPTIONS options, const char *dimensions,
1980 - const char *group_options, time_t timeout, size_t tier) {
1978 + const char *group_options, time_t timeout, size_t tier, QUERY_SOURCE query_source) {
1979
1980 QUERY_TARGET_REQUEST qtr = {
1981 .st = st,
@@ -1991,6 +1989,7 @@ RRDR *rrd2rrdr_legacy(
1989 .group_options = group_options,
1990 .timeout = timeout,
1991 .tier = tier,
1992 + .query_source = query_source,
1993 };
1994
1995 return rrd2rrdr(owa, query_target_create(&qtr));
@@ -2170,6 +2169,7 @@ RRDR *rrd2rrdr(ONEWAYALLOC *owa, QUERY_TARGET *qt) {
2169 }
2170 }
2171
2173 - rrdr_query_completed(r->internal.db_points_read, r->internal.result_points_generated);
2172 + global_statistics_rrdr_query_completed(dimensions_used, r->internal.db_points_read,
2173 + r->internal.result_points_generated, qt->request.query_source);
2174 return r;
2175 }
web/api/queries/rrdr.h
+1 -1
@@ -138,7 +138,7 @@ RRDR *rrd2rrdr_legacy(
138 ONEWAYALLOC *owa,
139 RRDSET *st, size_t points, time_t after, time_t before,
140 RRDR_GROUPING group_method, time_t resampling_time, RRDR_OPTIONS options, const char *dimensions,
141 - const char *group_options, time_t timeout, size_t tier);
141 + const char *group_options, time_t timeout, size_t tier, QUERY_SOURCE query_source);
142
143 RRDR *rrd2rrdr(ONEWAYALLOC *owa, struct query_target *qt);
144 bool query_target_calculate_window(struct query_target *qt);
web/api/queries/weights.c
+6 -5
@@ -518,7 +518,8 @@ NETDATA_DOUBLE *rrd2rrdr_ks2(
518 .options = options,
519 .group_method = group_method,
520 .group_options = group_options,
521 - .tier = tier
521 + .tier = tier,
522 + .query_source = QUERY_SOURCE_API_WEIGHTS,
523 };
524
525 RRDR *r = rrd2rrdr(owa, query_target_create(&qtr));
@@ -637,7 +638,7 @@ static void rrdset_metric_correlations_volume(
638
639 options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_ABSOLUTE | RRDR_OPTION_NATURAL_POINTS;
640
640 - QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before, options, group_method, group_options, tier, 0);
641 + QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
642 merge_query_value_to_stats(&baseline_average, stats);
643
644 if(!netdata_double_isnumber(baseline_average.value)) {
@@ -645,7 +646,7 @@ static void rrdset_metric_correlations_volume(
646 baseline_average.value = 0.0;
647 }
648
648 - QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0);
649 + QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
650 merge_query_value_to_stats(&highlight_average, stats);
651
652 if(!netdata_double_isnumber(highlight_average.value))
@@ -658,7 +659,7 @@ static void rrdset_metric_correlations_volume(
659
660 char highlight_countif_options[50 + 1];
661 snprintfz(highlight_countif_options, 50, "%s" NETDATA_DOUBLE_FORMAT, highlight_average.value < baseline_average.value ? "<" : ">", baseline_average.value);
661 - QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before, options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0);
662 + QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before, options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
663 merge_query_value_to_stats(&highlight_countif, stats);
664
665 if(!netdata_double_isnumber(highlight_countif.value)) {
@@ -699,7 +700,7 @@ static void rrdset_weights_anomaly_rate(
700
701 options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_ANOMALY_BIT | RRDR_OPTION_NATURAL_POINTS;
702
702 - QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0);
703 + QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
704 merge_query_value_to_stats(&qv, stats);
705
706 if(netdata_double_isnumber(qv.value))
web/api/web_api_v1.c
+1
@@ -751,6 +751,7 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
751 .tier = tier,
752 .chart_label_key = chart_label_key,
753 .charts_labels_filter = chart_labels_filter,
754 + .query_source = QUERY_SOURCE_API_DATA,
755 };
756 qt = query_target_create(&qtr);
757
web/server/web_client.c
+5 -5
@@ -85,11 +85,11 @@ void web_client_request_done(struct web_client *w) {
85 // --------------------------------------------------------------------
86 // global statistics
87
88 - finished_web_request_statistics(dt_usec(&tv, &w->tv_in),
89 - w->stats_received_bytes,
90 - w->stats_sent_bytes,
91 - size,
92 - sent);
88 + global_statistics_web_request_completed(dt_usec(&tv, &w->tv_in),
89 + w->stats_received_bytes,
90 + w->stats_sent_bytes,
91 + size,
92 + sent);
93
94 w->stats_received_bytes = 0;
95 w->stats_sent_bytes = 0;
web/server/web_client_cache.c
+2 -2
@@ -209,7 +209,7 @@ struct web_client *web_client_get_from_cache_or_allocate() {
209 web_clients_cache.used_count++;
210
211 // initialize it
212 - w->id = web_client_connected();
212 + w->id = global_statistics_web_client_connected();
213 w->mode = WEB_CLIENT_MODE_NORMAL;
214
215 netdata_thread_enable_cancelability();
@@ -230,7 +230,7 @@ void web_client_release(struct web_client *w) {
230
231 web_server_log_connection(w, "DISCONNECTED");
232 web_client_request_done(w);
233 - web_client_disconnected();
233 + global_statistics_web_client_disconnected();
234
235 netdata_thread_disable_cancelability();
236