Streaming interpolated values (#14431)

* first commit - untested

* fix wrong begin command

* added set v2 too

* debug to log stream buffer

* debug to log stream buffer

* faster streaming printing

* mark charts and dimensions as collected

* use stream points even if sender is not enabled

* comment out stream debug log

* parse null as nan

* custom begin v2

* custom set v2; replication now copies the anomalous flag too

* custom end v2

* enabled stream log test

* renamed to BEGIN2, SET2, END2

* dont mix up replay and v2 members in user object

* fix typo

* cleanup

* support to v2 to v1 proxying

* mark updated dimensions as such

* do not log unknown flags

* comment out stream debug log

* send also the chart id on BEGIN2, v2 to v2

* update the data collections counter

* v2 values are transferred in hex

* faster hex parsing

* a little more generic hex and dec printing and parsing

* fix hex parsing

* minor optimization in dbengine api

* turn debugging into info message

* generalized the timings tracking, so that it can be used in more places

* commented out debug info

* renamed conflicting variable with macro

* remove wrong edits

* integrated ML and added cleanup in case parsing is interrupted

* disable data collection locking during v2

* cleanup stale ML locks; send updated chart variables during v2; add info to find stale locks

* inject an END2 between repeated BEGIN2 from rrdset_done()

* test: remove lockless single-threaded logic from dictionary and aral and apply the right acquire/release memory order to reference counters

* more fine grained dictionary atomics

* remove unecessary return values

* pointer validation under NETDATA_DICTIONARY_VALIDATE_POINTERS

* Revert "pointer validation under NETDATA_DICTIONARY_VALIDATE_POINTERS"

This reverts commit 846cdf2713e2a7ee2ff797f38db11714228800e9.

* Revert "remove unecessary return values"

This reverts commit 8c87d30f4d86f0f5d6b4562cf74fe7447138bbff.

* Revert "more fine grained dictionary atomics"

This reverts commit 984aec4234a340d197d45239ff9a10fd479fcf3c.

* Revert "test: remove lockless single-threaded logic from dictionary and aral and apply the right acquire/release memory order to reference counters"

This reverts commit c460b3d0ad497d2641bd0ea1d63cec7c052e74e4.

* Apply again "pointer validation under NETDATA_DICTIONARY_VALIDATE_POINTERS" while keeping the improved atomic operations.

This reverts commit f158d009

* fix last commit

* fix last commit again

* optimizations in dbengine

* do not send anomaly bit on non-supporting agents (send it when the INTERPOLATED capability is available)

* break long empty-points-loops in rrdset_done()

* decide page alignment on new page allocation, not on every point collected

* create max size pages but no smaller than 1/3

* Fix compilation when --disable-ml is specified

* Return false

* fixes for NETDATA_LOG_REPLICATION_REQUESTS

* added compile option NETDATA_WITHOUT_WORKERS_LATENCY

* put timings in BEGIN2, SET2, END2

* isolate begin2 ml

* revert repositioning data collection lock

* fixed multi-threading of statistics

* do not lookup dimensions all the time if they come in the same order

* update used on iteration, not on every points; also do better error handling

---------

Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

1 files changed, 81 insertions, 9 deletions

diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index 256fa82821..3ba9a1da84 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -378,7 +378,72 @@ bool rrdset_push_chart_definition_now(RRDSET *st) {
     return true;
 }
 
-void rrdset_done_push(RRDSET *st) {
+void rrdset_push_metrics_v1(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
+    RRDHOST *host = st->rrdhost;
+    rrdpush_send_chart_metrics(rsb->wb, st, host->sender, rsb->rrdset_flags);
+}
+
+void rrddim_push_metrics_v2(RRDSET_STREAM_BUFFER *rsb, RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags) {
+    if(!rsb->wb || !rsb->v2 || !netdata_double_isnumber(n) || !does_storage_number_exist(flags))
+        return;
+
+    BUFFER *wb = rsb->wb;
+    time_t point_end_time_s = (time_t)(point_end_time_ut / USEC_PER_SEC);
+    if(unlikely(rsb->last_point_end_time_s != point_end_time_s)) {
+
+        if(unlikely(rsb->begin_v2_added))
+            buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
+
+        buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2 " '", sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1 + 2);
+        buffer_fast_strcat(wb, rrdset_id(rd->rrdset), string_strlen(rd->rrdset->id));
+        buffer_fast_strcat(wb, "' ", 2);
+        buffer_print_llu_hex(wb, rd->rrdset->update_every);
+        buffer_fast_strcat(wb, " ", 1);
+        buffer_print_llu_hex(wb, point_end_time_s);
+        buffer_fast_strcat(wb, " ", 1);
+        if(point_end_time_s == rsb->wall_clock_time)
+            buffer_fast_strcat(wb, "#", 1);
+        else
+            buffer_print_llu_hex(wb, rsb->wall_clock_time);
+        buffer_fast_strcat(wb, "\n", 1);
+
+        rsb->last_point_end_time_s = point_end_time_s;
+        rsb->begin_v2_added = true;
+    }
+
+    buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2 " '", sizeof(PLUGINSD_KEYWORD_SET_V2) - 1 + 2);
+    buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id));
+    buffer_fast_strcat(wb, "' ", 2);
+    buffer_print_ll_hex(wb, rd->last_collected_value);
+    buffer_fast_strcat(wb, " ", 1);
+
+    if((NETDATA_DOUBLE)rd->last_collected_value == n)
+        buffer_fast_strcat(wb, "#", 1);
+    else
+        buffer_rrd_value(wb, n);
+
+    buffer_fast_strcat(wb, " ", 1);
+    buffer_print_sn_flags(wb, flags, true);
+    buffer_fast_strcat(wb, "\n", 1);
+}
+
+void rrdset_push_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
+    if(!rsb->wb)
+        return;
+
+    if(rsb->v2 && rsb->begin_v2_added) {
+        if(unlikely(rsb->rrdset_flags & RRDSET_FLAG_UPSTREAM_SEND_VARIABLES))
+            rrdsetvar_print_to_streaming_custom_chart_variables(st, rsb->wb);
+
+        buffer_fast_strcat(rsb->wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
+    }
+
+    sender_commit(st->rrdhost->sender, rsb->wb);
+
+    *rsb = (RRDSET_STREAM_BUFFER){ .wb = NULL, };
+}
+
+RRDSET_STREAM_BUFFER rrdset_push_metric_initialize(RRDSET *st, time_t wall_clock_time) {
     RRDHOST *host = st->rrdhost;
 
     // fetch the flags we need to check with one atomic operation
@@ -395,7 +460,7 @@ void rrdset_done_push(RRDSET *st) {
             error("STREAM %s [send]: not ready - collected metrics are not sent to parent.", rrdhost_hostname(host));
         }
 
-        return;
+        return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
     }
     else if(unlikely(host_flags & RRDHOST_FLAG_RRDPUSH_SENDER_LOGGED_STATUS)) {
         info("STREAM %s [send]: sending metrics to parent...", rrdhost_hostname(host));
@@ -408,17 +473,23 @@ void rrdset_done_push(RRDSET *st) {
 
     if(unlikely((exposed_upstream && replication_in_progress) ||
                 !should_send_chart_matching(st, rrdset_flags)))
-        return;
-
-    BUFFER *wb = sender_start(host->sender);
+        return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
 
-    if(unlikely(!exposed_upstream))
+    if(unlikely(!exposed_upstream)) {
+        BUFFER *wb = sender_start(host->sender);
         replication_in_progress = rrdpush_send_chart_definition(wb, st);
+        sender_commit(host->sender, wb);
+    }
 
-    if (likely(!replication_in_progress))
-        rrdpush_send_chart_metrics(wb, st, host->sender, rrdset_flags);
+    if(replication_in_progress)
+        return (RRDSET_STREAM_BUFFER) { .wb = NULL, };
 
-    sender_commit(host->sender, wb);
+    return (RRDSET_STREAM_BUFFER) {
+        .v2 = stream_has_capability(host->sender, STREAM_CAP_INTERPOLATED),
+        .rrdset_flags = rrdset_flags,
+        .wb = sender_start(host->sender),
+        .wall_clock_time = wall_clock_time,
+    };
 }
 
 // labels
@@ -1077,6 +1148,7 @@ static void stream_capabilities_to_string(BUFFER *wb, STREAM_CAPABILITIES caps)
     if(caps & STREAM_CAP_FUNCTIONS) buffer_strcat(wb, "FUNCTIONS ");
     if(caps & STREAM_CAP_REPLICATION) buffer_strcat(wb, "REPLICATION ");
     if(caps & STREAM_CAP_BINARY) buffer_strcat(wb, "BINARY ");
+    if(caps & STREAM_CAP_INTERPOLATED) buffer_strcat(wb, "INTERPOLATED ");
 }
 
 void log_receiver_capabilities(struct receiver_state *rpt) {