Query Engine multi-granularity support (and MC improvements) (#13155)

* set grouping functions

* storage engine should check the validity of timestamps, not the query engine

* calculate and store in RRDR anomaly rates for every query

* anomaly rate used by volume metric correlations

* mc volume should use absolute data, to avoid cancelling effect

* return anomaly-rates in jasonwrap with jw-anomaly-rates option to data queries

* dont return null on anomaly rates

* allow passing group query options from the URL

* added countif to the query engine and used it in metric correlations

* fix configure

* fix countif and anomaly rate percentages

* added group_options to metric correlations; updated swagger

* added newline at the end of yaml file

* always check the time the highlighted window was above/below the highlighted window

* properly track time in memory queries

* error for internal checks only

* moved pack_storage_number() into the storage engines

* moved unpack_storage_number() inside the storage engines

* remove old comment

* pass unit tests

* properly detect zero or subnormal values in pack_storage_number()

* fill nulls before the value, not after

* make sure math.h is included

* workaround for isfinite()

* fix for isfinite()

* faster isfinite() alternative

* fix for faster isfinite() alternative

* next_metric() now returns end_time too

* variable step implemented in a generic way

* remove left-over variables

* ensure we always complete the wanted number of points

* fixes

* ensure no infinite loop

* mc-volume-improvements: Add information about invalid condition

* points should have a duration in the past

* removed unneeded info() line

* Fix unit tests for exporting engine

* new_point should only be checked when it is fetched from the db; better comment about the premature breaking of the main query loop

Co-authored-by: Thiago Marques <thiagoftsm@gmail.com>
Co-authored-by: Vladimir Kobal <vlad@prokk.net>

2 files changed, 51 insertions, 17 deletions

diff --git a/database/ram/rrddim_mem.c b/database/ram/rrddim_mem.c
index b17f03ca50..03b6983a29 100644
--- a/database/ram/rrddim_mem.c
+++ b/database/ram/rrddim_mem.c
@@ -9,9 +9,9 @@ void rrddim_collect_init(RRDDIM *rd) {
     rd->values[rd->rrdset->current_entry] = SN_EMPTY_SLOT;
     rd->state->handle = calloc(1, sizeof(struct mem_collect_handle));
 }
-void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number) {
+void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, calculated_number number, SN_FLAGS flags) {
     (void)point_in_time;
-    rd->values[rd->rrdset->current_entry] = number;
+    rd->values[rd->rrdset->current_entry] = pack_storage_number(number, flags);
 }
 int rrddim_collect_finalize(RRDDIM *rd) {
     free((struct mem_collect_handle*)rd->state->handle);
@@ -28,33 +28,63 @@ void rrddim_query_init(RRDDIM *rd, struct rrddim_query_handle *handle, time_t st
     struct mem_query_handle* h = calloc(1, sizeof(struct mem_query_handle));
     h->slot = rrdset_time2slot(rd->rrdset, start_time);
     h->last_slot = rrdset_time2slot(rd->rrdset, end_time);
-    h->finished = 0;
+    h->dt = rd->update_every;
+
+    h->next_timestamp = start_time;
+    h->slot_timestamp = rrdset_slot2time(rd->rrdset, h->slot);
+    h->last_timestamp = rrdset_slot2time(rd->rrdset, h->last_slot);
+
+    // info("QUERY: start %ld, end %ld, next %ld, first %ld, last %ld", start_time, end_time, h->next_timestamp, h->slot_timestamp, h->last_timestamp);
+
     handle->handle = (STORAGE_QUERY_HANDLE *)h;
 }
 
-storage_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *current_time) {
+// Returns the metric and sets its timestamp into current_time
+// IT IS REQUIRED TO **ALWAYS** SET ALL RETURN VALUES (current_time, end_time, flags)
+// IT IS REQUIRED TO **ALWAYS** KEEP TRACK OF TIME, EVEN OUTSIDE THE DATABASE BOUNDARIES
+calculated_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *start_time, time_t *end_time, SN_FLAGS *flags) {
     RRDDIM *rd = handle->rd;
     struct mem_query_handle* h = (struct mem_query_handle*)handle->handle;
-    long entries = rd->rrdset->entries;
-    long slot = h->slot;
+    size_t entries = rd->rrdset->entries;
+    size_t slot = h->slot;
 
-    (void)current_time;
-    if (unlikely(h->slot == h->last_slot))
-        h->finished = 1;
-    storage_number n = rd->values[slot++];
+    time_t this_timestamp = h->next_timestamp;
+    h->next_timestamp += h->dt;
+
+    // set this timestamp for our caller
+    *start_time = this_timestamp - h->dt;
+    *end_time = this_timestamp;
 
+    if(unlikely(this_timestamp < h->slot_timestamp)) {
+        *flags = SN_EMPTY_SLOT;
+        return NAN;
+    }
+
+    if(unlikely(this_timestamp > h->last_timestamp)) {
+        *flags = SN_EMPTY_SLOT;
+        return NAN;
+    }
+
+    storage_number n = rd->values[slot++];
     if(unlikely(slot >= entries)) slot = 0;
+
     h->slot = slot;
+    h->slot_timestamp += h->dt;
 
-    return n;
+    *flags = (n & SN_ALL_FLAGS);
+    return unpack_storage_number(n);
 }
 
 int rrddim_query_is_finished(struct rrddim_query_handle *handle) {
     struct mem_query_handle* h = (struct mem_query_handle*)handle->handle;
-    return h->finished;
+    return (h->next_timestamp > handle->end_time);
 }
 
 void rrddim_query_finalize(struct rrddim_query_handle *handle) {
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(!rrddim_query_is_finished(handle))
+        error("QUERY: query for chart '%s' dimension '%s' has been stopped unfinished", handle->rd->rrdset->id, handle->rd->name);
+#endif
     freez(handle->handle);
 }
 
diff --git a/database/ram/rrddim_mem.h b/database/ram/rrddim_mem.h
index 9a215387ae..ede8a4e211 100644
--- a/database/ram/rrddim_mem.h
+++ b/database/ram/rrddim_mem.h
@@ -9,18 +9,22 @@ struct mem_collect_handle {
     long slot;
     long entries;
 };
+
 struct mem_query_handle {
-    long slot;
-    long last_slot;
-    uint8_t finished;
+    time_t dt;
+    time_t next_timestamp;
+    time_t last_timestamp;
+    time_t slot_timestamp;
+    size_t slot;
+    size_t last_slot;
 };
 
 extern void rrddim_collect_init(RRDDIM *rd);
-extern void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number);
+extern void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, calculated_number number, SN_FLAGS flags);
 extern int rrddim_collect_finalize(RRDDIM *rd);
 
 extern void rrddim_query_init(RRDDIM *rd, struct rrddim_query_handle *handle, time_t start_time, time_t end_time);
-extern storage_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *current_time);
+extern calculated_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *start_time, time_t *end_time, SN_FLAGS *flags);
 extern int rrddim_query_is_finished(struct rrddim_query_handle *handle);
 extern void rrddim_query_finalize(struct rrddim_query_handle *handle);
 extern time_t rrddim_query_latest_time(RRDDIM *rd);