Reuse the SN_EXISTS bit to track anomaly status. (#11154)

* Replace all usages of SN_EXISTS with SN_DEFAULT_FLAGS.

* Remove references to SN_NOT_EXISTS in comments.

* Replace raw zero constant with SN_EMPTY_SLOT.

* Use get_storage_number_flags only in storage_number.{c,h}

* Compare against SN_EMPTY_SLOT to check if a storage_number exists.

This is safe because:

  1. rrdset_done_interpolate() is the only place where we call store_metric(),
  2. All store_metric() calls, except for one, store an SN_EMPTY_SLOT value.
  3. When we are not storing an SN_EMPTY_SLOT value, the flags that we pass to
     pack_storage_number() can be either SN_EXISTS *or* SN_EXISTS_RESET.

* Compare only the SN_EXISTS_RESET bit to find reset values.

* Remove get_storage_number_flags from storage_number.h

* Do not set storage_number flags outside of rrdset_done_interpolate().

This is a NFC intended to limit the scope of storage_number flags
processing to just one function.

* Set reset bit without overwriting the rest of the flags.

* Rename SN_EXISTS to SN_ANOMALY_BIT.

* Use GOTOs in pack_storage_number to return from a single place.

* Teach pack_storage_number how to handle anomalous zero values.

Up until now, a storage_number had always either the SN_EXISTS or
SN_EXISTS_RESET bit set. This meant that it was not possible for any
packed storage_number to compare equal to the SN_EMPTY_SLOT.

However, the SN_ANOMALY_BIT can be set to zero. This is fine for every
value other than the anomalous 0 value, because it would compare equal to
SN_EMPTY_SLOT. We address this issue by mapping the anomalous zero value
to SN_EXISTS_100 (a number which was not possible to generate with the
previous versions of the agent, ie. it won't exist in older dbengine files).

This change was tested manually by intentionally flipping the anomaly
bit for odd/even iterations in rrdset_done_interpolate. Prior to this
change, charts whose dimensions had 0 values, where showing up in the
dashboard as gaps (SN_EMPTY_SLOT), whereas with this commit the values
are displayed correctly.

3 files changed, 18 insertions, 14 deletions

diff --git a/database/engine/rrdengine.c b/database/engine/rrdengine.c
index f48181d29a..54a9cdf8dc 100644
--- a/database/engine/rrdengine.c
+++ b/database/engine/rrdengine.c
@@ -336,7 +336,7 @@ after_crc_check:
         /* care, we don't hold the descriptor mutex */
         if (have_read_error) {
             /* Applications should make sure NULL values match 0 as does SN_EMPTY_SLOT */
-            memset(page, 0, descr->page_length);
+            memset(page, SN_EMPTY_SLOT, descr->page_length);
         } else if (RRD_NO_COMPRESSION == header->compression_algorithm) {
             (void) memcpy(page, xt_io_descr->buf + payload_offset + page_offset, descr->page_length);
         } else {
diff --git a/database/rrddim.c b/database/rrddim.c
index 51c4428c72..5479ade7aa 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -119,7 +119,7 @@ inline int rrddim_set_divisor(RRDSET *st, RRDDIM *rd, collected_number divisor)
 // RRDDIM legacy data collection functions
 
 static void rrddim_collect_init(RRDDIM *rd) {
-    rd->values[rd->rrdset->current_entry] = SN_EMPTY_SLOT; // pack_storage_number(0, SN_NOT_EXISTS);
+    rd->values[rd->rrdset->current_entry] = SN_EMPTY_SLOT;
 }
 static void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number) {
     (void)point_in_time;
diff --git a/database/rrdset.c b/database/rrdset.c
index e4701e3b9e..282b0e5ba1 100644
--- a/database/rrdset.c
+++ b/database/rrdset.c
@@ -1120,7 +1120,7 @@ static inline size_t rrdset_done_interpolate(
         , usec_t last_collect_ut
         , usec_t now_collect_ut
         , char store_this_entry
-        , uint32_t storage_flags
+        , uint32_t has_reset_value
 ) {
     RRDDIM *rd;
 
@@ -1135,6 +1135,11 @@ static inline size_t rrdset_done_interpolate(
     size_t counter = st->counter;
     long current_entry = st->current_entry;
 
+    uint32_t storage_flags = SN_DEFAULT_FLAGS;
+
+    if (has_reset_value)
+        storage_flags |= SN_EXISTS_RESET;
+
     for( ; next_store_ut <= now_collect_ut ; last_collect_ut = next_store_ut, next_store_ut += update_every_ut, iterations-- ) {
 
         #ifdef NETDATA_INTERNAL_CHECKS
@@ -1232,8 +1237,8 @@ static inline size_t rrdset_done_interpolate(
             }
 
             if(unlikely(!store_this_entry)) {
-                rd->state->collect_ops.store_metric(rd, next_store_ut, SN_EMPTY_SLOT); //pack_storage_number(0, SN_NOT_EXISTS)
-//                rd->values[current_entry] = SN_EMPTY_SLOT; //pack_storage_number(0, SN_NOT_EXISTS);
+                rd->state->collect_ops.store_metric(rd, next_store_ut, SN_EMPTY_SLOT);
+//                rd->values[current_entry] = SN_EMPTY_SLOT;
                 continue;
             }
 
@@ -1261,8 +1266,8 @@ static inline size_t rrdset_done_interpolate(
                 );
                 #endif
 
-//                rd->values[current_entry] = SN_EMPTY_SLOT; // pack_storage_number(0, SN_NOT_EXISTS);
-                rd->state->collect_ops.store_metric(rd, next_store_ut, SN_EMPTY_SLOT); //pack_storage_number(0, SN_NOT_EXISTS)
+//                rd->values[current_entry] = SN_EMPTY_SLOT;
+                rd->state->collect_ops.store_metric(rd, next_store_ut, SN_EMPTY_SLOT);
                 rd->last_stored_value = NAN;
             }
 
@@ -1274,11 +1279,10 @@ static inline size_t rrdset_done_interpolate(
                 calculated_number t2 = unpack_storage_number(rd->values[current_entry]);
 
                 calculated_number accuracy = accuracy_loss(t1, t2);
-                debug(D_RRD_STATS, "%s/%s: UNPACK[%ld] = " CALCULATED_NUMBER_FORMAT " FLAGS=0x%08x (original = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s)"
+                debug(D_RRD_STATS, "%s/%s: UNPACK[%ld] = " CALCULATED_NUMBER_FORMAT " (original = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s)"
                       , st->id, rd->name
                       , current_entry
                       , t2
-                      , get_storage_number_flags(rd->values[current_entry])
                       , t1
                       , accuracy
                       , (accuracy > ACCURACY_LOSS_ACCEPTED_PERCENT) ? " **TOO BIG** " : ""
@@ -1300,7 +1304,7 @@ static inline size_t rrdset_done_interpolate(
             #endif
         }
         // reset the storage flags for the next point, if any;
-        storage_flags = SN_EXISTS;
+        storage_flags = SN_DEFAULT_FLAGS;
 
         st->counter = ++counter;
         st->current_entry = current_entry = ((current_entry + 1) >= st->entries) ? 0 : current_entry + 1;
@@ -1540,7 +1544,7 @@ after_first_database_work:
             st->collected_total += rd->collected_value;
     }
 
-    uint32_t storage_flags = SN_EXISTS;
+    uint32_t has_reset_value = 0;
 
     // process all dimensions to calculate their values
     // based on the collected figures only
@@ -1637,7 +1641,7 @@ after_first_database_work:
                           , rd->collected_value);
 
                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
-                        storage_flags = SN_EXISTS_RESET;
+                        has_reset_value = 1;
 
                     uint64_t last = (uint64_t)rd->last_collected_value;
                     uint64_t new = (uint64_t)rd->collected_value;
@@ -1708,7 +1712,7 @@ after_first_database_work:
                     );
 
                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
-                        storage_flags = SN_EXISTS_RESET;
+                        has_reset_value = 1;
 
                     rd->last_collected_value = rd->collected_value;
                 }
@@ -1787,7 +1791,7 @@ after_first_database_work:
             , last_collect_ut
             , now_collect_ut
             , store_this_entry
-            , storage_flags
+            , has_reset_value
     );
 
 after_second_database_work: