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

11 files changed, 35 insertions, 28 deletions

diff --git a/web/api/badges/web_buffer_svg.c b/web/api/badges/web_buffer_svg.c
index 69f3d4367d..080f2240f0 100644
--- a/web/api/badges/web_buffer_svg.c
+++ b/web/api/badges/web_buffer_svg.c
@@ -1116,7 +1116,8 @@ int web_client_api_request_v1_badge(RRDHOST *host, struct web_client *w, char *u
                                       points, after, before, group, group_options, 0, options,
                                       NULL, &latest_timestamp,
                                       NULL, NULL, NULL,
-                                      &value_is_null, NULL, 0, 0);
+                                      &value_is_null, NULL, 0, 0,
+                                      QUERY_SOURCE_API_BADGE);
 
         // if the value cannot be calculated, show empty badge
         if (ret != HTTP_RESP_OK) {
diff --git a/web/api/formatters/rrd2json.c b/web/api/formatters/rrd2json.c
index 72994048f8..8bf547192c 100644
--- a/web/api/formatters/rrd2json.c
+++ b/web/api/formatters/rrd2json.c
@@ -76,6 +76,7 @@ int rrdset2value_api_v1(
         , NETDATA_DOUBLE *anomaly_rate
         , time_t timeout
         , size_t tier
+        , QUERY_SOURCE query_source
 ) {
     int ret = HTTP_RESP_INTERNAL_SERVER_ERROR;
 
@@ -92,7 +93,8 @@ int rrdset2value_api_v1(
             dimensions,
             group_options,
             timeout,
-            tier);
+            tier,
+            query_source);
 
     if(!r) {
         if(value_is_null) *value_is_null = 1;
diff --git a/web/api/formatters/rrd2json.h b/web/api/formatters/rrd2json.h
index 5a892bfd9a..048281d7e7 100644
--- a/web/api/formatters/rrd2json.h
+++ b/web/api/formatters/rrd2json.h
@@ -78,6 +78,7 @@ int rrdset2value_api_v1(
         , NETDATA_DOUBLE *anomaly_rate
         , time_t timeout
         , size_t tier
+        , QUERY_SOURCE query_source
 );
 
 #endif /* NETDATA_RRD2JSON_H */
diff --git a/web/api/formatters/value/value.c b/web/api/formatters/value/value.c
index 5477e0b0ac..46a71303e6 100644
--- a/web/api/formatters/value/value.c
+++ b/web/api/formatters/value/value.c
@@ -106,7 +106,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
                             struct rrdcontext_acquired *rca, struct rrdinstance_acquired *ria, struct rrdmetric_acquired *rma,
                             time_t after, time_t before,
                             RRDR_OPTIONS options, RRDR_GROUPING group_method, const char *group_options,
-                            size_t tier, time_t timeout
+                            size_t tier, time_t timeout, QUERY_SOURCE query_source
 ) {
     QUERY_TARGET_REQUEST qtr = {
             .host = host,
@@ -121,6 +121,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
             .group_options = group_options,
             .tier = tier,
             .timeout = timeout,
+            .query_source = query_source,
     };
 
     ONEWAYALLOC *owa = onewayalloc_create(16 * 1024);
diff --git a/web/api/formatters/value/value.h b/web/api/formatters/value/value.h
index 08844c6da5..76b1869f36 100644
--- a/web/api/formatters/value/value.h
+++ b/web/api/formatters/value/value.h
@@ -23,7 +23,7 @@ QUERY_VALUE rrdmetric2value(RRDHOST *host,
                             struct rrdcontext_acquired *rca, struct rrdinstance_acquired *ria, struct rrdmetric_acquired *rma,
                             time_t after, time_t before,
                             RRDR_OPTIONS options, RRDR_GROUPING group_method, const char *group_options,
-                            size_t tier, time_t timeout
+                            size_t tier, time_t timeout, QUERY_SOURCE query_source
 );
 
 NETDATA_DOUBLE rrdr2value(RRDR *r, long i, RRDR_OPTIONS options, int *all_values_are_null, NETDATA_DOUBLE *anomaly_rate);
diff --git a/web/api/queries/query.c b/web/api/queries/query.c
index 052ab944d7..ccd1951358 100644
--- a/web/api/queries/query.c
+++ b/web/api/queries/query.c
@@ -1473,7 +1473,8 @@ static inline void rrd2rrdr_do_dimension(RRDR *r, size_t dim_id_in_rrdr) {
 // ----------------------------------------------------------------------------
 // fill the gap of a tier
 
-extern void store_metric_at_tier(RRDDIM *rd, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut);
+void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut);
+void store_metric_collection_completed(void);
 
 void rrdr_fill_tier_gap_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now) {
     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)
 
     struct storage_engine_query_handle handle;
 
-    size_t all_points_read = 0;
-
     // for each lower tier
     for(int tr = (int)tier - 1; tr >= 0 ;tr--){
         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)
         struct rrddim_tier *tmp = rd->tiers[tr];
         tmp->query_ops->init(tmp->db_metric_handle, &handle, after_wanted, before_wanted);
 
-        size_t points = 0;
+        size_t points_read = 0;
 
         while(!tmp->query_ops->is_finished(&handle)) {
 
             STORAGE_POINT sp = tmp->query_ops->next_metric(&handle);
+            points_read++;
 
             if(sp.end_time > latest_time_t) {
                 latest_time_t = sp.end_time;
-                store_metric_at_tier(rd, t, sp, sp.end_time * USEC_PER_SEC);
-                points++;
+                store_metric_at_tier(rd, tr, t, sp, sp.end_time * USEC_PER_SEC);
             }
         }
 
-        all_points_read += points;
         tmp->query_ops->finalize(&handle);
+        store_metric_collection_completed();
+        global_statistics_backfill_query_completed(points_read);
 
         //internal_error(true, "DBENGINE: backfilled chart '%s', dimension '%s', tier %d, from %ld to %ld, with %zu points from tier %d",
         //               rd->rrdset->name, rd->name, tier, after_wanted, before_wanted, points, tr);
     }
-
-    rrdr_query_completed(all_points_read, all_points_read);
 }
 
 // ----------------------------------------------------------------------------
@@ -1977,7 +1975,7 @@ RRDR *rrd2rrdr_legacy(
         ONEWAYALLOC *owa,
         RRDSET *st, size_t points, time_t after, time_t before,
         RRDR_GROUPING group_method, time_t resampling_time, RRDR_OPTIONS options, const char *dimensions,
-        const char *group_options, time_t timeout, size_t tier) {
+        const char *group_options, time_t timeout, size_t tier, QUERY_SOURCE query_source) {
 
     QUERY_TARGET_REQUEST qtr = {
             .st = st,
@@ -1991,6 +1989,7 @@ RRDR *rrd2rrdr_legacy(
             .group_options = group_options,
             .timeout = timeout,
             .tier = tier,
+            .query_source = query_source,
     };
 
     return rrd2rrdr(owa, query_target_create(&qtr));
@@ -2170,6 +2169,7 @@ RRDR *rrd2rrdr(ONEWAYALLOC *owa, QUERY_TARGET *qt) {
         }
     }
 
-    rrdr_query_completed(r->internal.db_points_read, r->internal.result_points_generated);
+    global_statistics_rrdr_query_completed(dimensions_used, r->internal.db_points_read,
+                                           r->internal.result_points_generated, qt->request.query_source);
     return r;
 }
diff --git a/web/api/queries/rrdr.h b/web/api/queries/rrdr.h
index 397201a72b..6151cddc7d 100644
--- a/web/api/queries/rrdr.h
+++ b/web/api/queries/rrdr.h
@@ -138,7 +138,7 @@ RRDR *rrd2rrdr_legacy(
         ONEWAYALLOC *owa,
         RRDSET *st, size_t points, time_t after, time_t before,
         RRDR_GROUPING group_method, time_t resampling_time, RRDR_OPTIONS options, const char *dimensions,
-        const char *group_options, time_t timeout, size_t tier);
+        const char *group_options, time_t timeout, size_t tier, QUERY_SOURCE query_source);
 
 RRDR *rrd2rrdr(ONEWAYALLOC *owa, struct query_target *qt);
 bool query_target_calculate_window(struct query_target *qt);
diff --git a/web/api/queries/weights.c b/web/api/queries/weights.c
index b249fa02f5..a9555a66be 100644
--- a/web/api/queries/weights.c
+++ b/web/api/queries/weights.c
@@ -518,7 +518,8 @@ NETDATA_DOUBLE *rrd2rrdr_ks2(
             .options = options,
             .group_method = group_method,
             .group_options = group_options,
-            .tier = tier
+            .tier = tier,
+            .query_source = QUERY_SOURCE_API_WEIGHTS,
     };
 
     RRDR *r = rrd2rrdr(owa, query_target_create(&qtr));
@@ -637,7 +638,7 @@ static void rrdset_metric_correlations_volume(
 
     options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_ABSOLUTE | RRDR_OPTION_NATURAL_POINTS;
 
-    QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before, options, group_method, group_options, tier, 0);
+    QUERY_VALUE baseline_average = rrdmetric2value(host, rca, ria, rma, baseline_after, baseline_before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
     merge_query_value_to_stats(&baseline_average, stats);
 
     if(!netdata_double_isnumber(baseline_average.value)) {
@@ -645,7 +646,7 @@ static void rrdset_metric_correlations_volume(
         baseline_average.value = 0.0;
     }
 
-    QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0);
+    QUERY_VALUE highlight_average = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
     merge_query_value_to_stats(&highlight_average, stats);
 
     if(!netdata_double_isnumber(highlight_average.value))
@@ -658,7 +659,7 @@ static void rrdset_metric_correlations_volume(
 
     char highlight_countif_options[50 + 1];
     snprintfz(highlight_countif_options, 50, "%s" NETDATA_DOUBLE_FORMAT, highlight_average.value < baseline_average.value ? "<" : ">", baseline_average.value);
-    QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before, options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0);
+    QUERY_VALUE highlight_countif = rrdmetric2value(host, rca, ria, rma, after, before, options, RRDR_GROUPING_COUNTIF, highlight_countif_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
     merge_query_value_to_stats(&highlight_countif, stats);
 
     if(!netdata_double_isnumber(highlight_countif.value)) {
@@ -699,7 +700,7 @@ static void rrdset_weights_anomaly_rate(
 
     options |= RRDR_OPTION_MATCH_IDS | RRDR_OPTION_ANOMALY_BIT | RRDR_OPTION_NATURAL_POINTS;
 
-    QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0);
+    QUERY_VALUE qv = rrdmetric2value(host, rca, ria, rma, after, before, options, group_method, group_options, tier, 0, QUERY_SOURCE_API_WEIGHTS);
     merge_query_value_to_stats(&qv, stats);
 
     if(netdata_double_isnumber(qv.value))
diff --git a/web/api/web_api_v1.c b/web/api/web_api_v1.c
index c93da27a80..93f501f9ee 100644
--- a/web/api/web_api_v1.c
+++ b/web/api/web_api_v1.c
@@ -751,6 +751,7 @@ inline int web_client_api_request_v1_data(RRDHOST *host, struct web_client *w, c
             .tier = tier,
             .chart_label_key = chart_label_key,
             .charts_labels_filter = chart_labels_filter,
+            .query_source = QUERY_SOURCE_API_DATA,
     };
     qt = query_target_create(&qtr);
 
diff --git a/web/server/web_client.c b/web/server/web_client.c
index 4d6f20052a..b3c5ada7ab 100644
--- a/web/server/web_client.c
+++ b/web/server/web_client.c
@@ -85,11 +85,11 @@ void web_client_request_done(struct web_client *w) {
         // --------------------------------------------------------------------
         // global statistics
 
-        finished_web_request_statistics(dt_usec(&tv, &w->tv_in),
-                                        w->stats_received_bytes,
-                                        w->stats_sent_bytes,
-                                        size,
-                                        sent);
+        global_statistics_web_request_completed(dt_usec(&tv, &w->tv_in),
+                                                w->stats_received_bytes,
+                                                w->stats_sent_bytes,
+                                                size,
+                                                sent);
 
         w->stats_received_bytes = 0;
         w->stats_sent_bytes = 0;
diff --git a/web/server/web_client_cache.c b/web/server/web_client_cache.c
index f7eb527239..1fa5935803 100644
--- a/web/server/web_client_cache.c
+++ b/web/server/web_client_cache.c
@@ -209,7 +209,7 @@ struct web_client *web_client_get_from_cache_or_allocate() {
     web_clients_cache.used_count++;
 
     // initialize it
-    w->id = web_client_connected();
+    w->id = global_statistics_web_client_connected();
     w->mode = WEB_CLIENT_MODE_NORMAL;
 
     netdata_thread_enable_cancelability();
@@ -230,7 +230,7 @@ void web_client_release(struct web_client *w) {
 
     web_server_log_connection(w, "DISCONNECTED");
     web_client_request_done(w);
-    web_client_disconnected();
+    global_statistics_web_client_disconnected();
 
     netdata_thread_disable_cancelability();