streaming compression, query planner and replication fixes (#14023)

* streaming compression, query planner and replication fixes

* remove journal v2 stats from global statistics

* disable sql for checking past sql UUIDs

* single threaded replication

* final replication thread using dictionaries and JudyL for sorting the pending requests

* do not timeout the sending socket when there are pending replication requests

* streaming receiver using read() instead of fread()

* remove FILE * from streaming - now using posix read() and write()

* increase timeouts to 10 minutes

* apply sender timeout only when there are metrics that are supposed to be streamed

* error handling in replication

* remove retries on socket read timeout; better error messages

* take into account inbound traffic too to detect that a connection is stale

* remove race conditions from replication thread

* make sure deleted entries are marked as executed, so that even if deletion fails, they will not be executed

* 2 minutes timeout to retry streaming to a parent that already has this node

* remove unecessary condition check

* fix compilation warnings

* include judy in replication

* wrappers to handle retries for SSL_read and SSL_write

* compressed bytes read monitoring

* recursive locks on replication to make it faster during flush or cleanup

* replication completion chart at the receiver side

* simplified recursive mutex

* simplified recursive mutex again

2 files changed, 19 insertions, 8 deletions

diff --git a/libnetdata/worker_utilization/worker_utilization.c b/libnetdata/worker_utilization/worker_utilization.c
index 7df050f386..114c4ad9f8 100644
--- a/libnetdata/worker_utilization/worker_utilization.c
+++ b/libnetdata/worker_utilization/worker_utilization.c
@@ -151,10 +151,17 @@ void worker_set_metric(size_t job_id, NETDATA_DOUBLE value) {
     if(unlikely(job_id >= WORKER_UTILIZATION_MAX_JOB_TYPES))
         return;
 
-    if(worker->per_job_type[job_id].type == WORKER_METRIC_INCREMENTAL)
-        worker->per_job_type[job_id].custom_value += value;
-    else
-        worker->per_job_type[job_id].custom_value = value;
+    switch(worker->per_job_type[job_id].type) {
+        case WORKER_METRIC_INCREMENT:
+            worker->per_job_type[job_id].custom_value += value;
+            break;
+
+        case WORKER_METRIC_INCREMENTAL_TOTAL:
+        case WORKER_METRIC_ABSOLUTE:
+        default:
+            worker->per_job_type[job_id].custom_value = value;
+            break;
+    }
 }
 
 // statistics interface
@@ -200,11 +207,12 @@ void workers_foreach(const char *workname, void (*callback)(
 
             switch(p->per_job_type[i].type) {
                 default:
-                case WORKER_METRIC_EMPTY:
+                case WORKER_METRIC_EMPTY: {
                     per_job_type_jobs_started[i] = 0;
                     per_job_type_busy_time[i] = 0;
                     per_job_custom_values[i] = NAN;
                     break;
+                }
 
                 case WORKER_METRIC_IDLE_BUSY: {
                     size_t tmp_jobs_started = p->per_job_type[i].worker_jobs_started;
@@ -219,14 +227,16 @@ void workers_foreach(const char *workname, void (*callback)(
                     break;
                 }
 
-                case WORKER_METRIC_ABSOLUTE:
+                case WORKER_METRIC_ABSOLUTE: {
                     per_job_type_jobs_started[i] = 0;
                     per_job_type_busy_time[i] = 0;
 
                     per_job_custom_values[i] = p->per_job_type[i].custom_value;
                     break;
+                }
 
-                case WORKER_METRIC_INCREMENTAL: {
+                case WORKER_METRIC_INCREMENTAL_TOTAL:
+                case WORKER_METRIC_INCREMENT: {
                     per_job_type_jobs_started[i] = 0;
                     per_job_type_busy_time[i] = 0;
 
diff --git a/libnetdata/worker_utilization/worker_utilization.h b/libnetdata/worker_utilization/worker_utilization.h
index aed893ed2d..954fcdcba7 100644
--- a/libnetdata/worker_utilization/worker_utilization.h
+++ b/libnetdata/worker_utilization/worker_utilization.h
@@ -11,7 +11,8 @@ typedef enum {
     WORKER_METRIC_EMPTY = 0,
     WORKER_METRIC_IDLE_BUSY = 1,
     WORKER_METRIC_ABSOLUTE = 2,
-    WORKER_METRIC_INCREMENTAL = 3,
+    WORKER_METRIC_INCREMENT = 3,
+    WORKER_METRIC_INCREMENTAL_TOTAL = 4,
 } WORKER_METRIC_TYPE;
 
 void worker_register(const char *workname);