Detect stored metric size by page type (#13334)

* Report unknown page only once
Get metric storage size by the page type
Verify validity of the page and skip problematic ones

* Change PAGE_SIZE to PAGE_POINT_SIZE_BYTES

* Add bitmap256 and unittests

* Fix unit test
tier_page_type array
page_type_size arrays

* Add another counter to not rely on uint8_t overflow to stop the test loop

11 files changed, 184 insertions, 27 deletions

diff --git a/daemon/main.c b/daemon/main.c
index eec5473808..ece125bcb4 100644
--- a/daemon/main.c
+++ b/daemon/main.c
@@ -973,7 +973,8 @@ int main(int argc, char **argv) {
                                 return 1;
                             if (unit_test_str2ld())
                                 return 1;
-
+                            if (unit_test_bitmap256())
+                                return 1;
                             // No call to load the config file on this code-path
                             post_conf_load(&user);
                             get_netdata_configured_variables();
diff --git a/daemon/unit_test.c b/daemon/unit_test.c
index 0cfe5abb5f..f7f4bb24d0 100644
--- a/daemon/unit_test.c
+++ b/daemon/unit_test.c
@@ -1606,6 +1606,113 @@ error:
     return 1;
 }
 
+int unit_test_bitmap256(void) {
+    fprintf(stderr, "%s() running...\n", __FUNCTION__ );
+
+    BITMAP256 test_bitmap = {0};
+
+    bitmap256_set_bit(&test_bitmap, 0, 1);
+    bitmap256_set_bit(&test_bitmap, 64, 1);
+    bitmap256_set_bit(&test_bitmap, 128, 1);
+    bitmap256_set_bit(&test_bitmap, 192, 1);
+    if (test_bitmap.data[0] == 1)
+        fprintf(stderr, "%s() INDEX 1 is OK\n", __FUNCTION__ );
+    if (test_bitmap.data[1] == 1)
+        fprintf(stderr, "%s() INDEX 65 is OK\n", __FUNCTION__ );
+    if (test_bitmap.data[2] == 1)
+        fprintf(stderr, "%s() INDEX 129 is OK\n", __FUNCTION__ );
+    if (test_bitmap.data[3] == 1)
+        fprintf(stderr, "%s() INDEX 192 is OK\n", __FUNCTION__ );
+
+    uint8_t i=0;
+    int j = 0;
+    do {
+        bitmap256_set_bit(&test_bitmap, i++, 1);
+        j++;
+    } while (j < 256);
+
+    if (test_bitmap.data[0] == 0xffffffff)
+        fprintf(stderr, "%s() INDEX 0 is fully set OK\n", __FUNCTION__);
+    if (test_bitmap.data[1] == 0xffffffff)
+        fprintf(stderr, "%s() INDEX 1 is fully set OK\n", __FUNCTION__);
+    if (test_bitmap.data[2] == 0xffffffff)
+        fprintf(stderr, "%s() INDEX 2 is fully set OK\n", __FUNCTION__);
+    if (test_bitmap.data[3] == 0xffffffff)
+        fprintf(stderr, "%s() INDEX 3 is fully set OK\n", __FUNCTION__);
+
+    i = 0;
+    j = 0;
+    do {
+        bitmap256_set_bit(&test_bitmap, i++, 0);
+        j++;
+    } while (j < 256);
+
+    if (test_bitmap.data[0] == 0)
+        fprintf(stderr, "%s() INDEX 0 is reset OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 0 is not reset FAILED\n", __FUNCTION__);
+        return 1;
+    }
+    if (test_bitmap.data[1] == 0)
+        fprintf(stderr, "%s() INDEX 1 is reset OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 1 is not reset FAILED\n", __FUNCTION__);
+        return 1;
+    }
+
+    if (test_bitmap.data[2] == 0)
+        fprintf(stderr, "%s() INDEX 2 is reset OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 2 is not reset FAILED\n", __FUNCTION__);
+        return 1;
+    }
+
+    if (test_bitmap.data[3] == 0)
+        fprintf(stderr, "%s() INDEX 3 is reset OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 3 is not reset FAILED\n", __FUNCTION__);
+        return 1;
+    }
+
+    i=0;
+    j = 0;
+    do {
+        bitmap256_set_bit(&test_bitmap, i, 1);
+        i += 4;
+        j += 4;
+    } while (j < 256);
+
+    if (test_bitmap.data[0] == 0x11111111)
+        fprintf(stderr, "%s() INDEX 0 is 0x11111111 set OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 0 is %lx expected 0x11111111\n", __FUNCTION__, test_bitmap.data[0]);
+        return 1;
+    }
+
+    if (test_bitmap.data[1] == 0x11111111)
+        fprintf(stderr, "%s() INDEX 1 is 0x11111111 set OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 1 is %lx expected 0x11111111\n", __FUNCTION__, test_bitmap.data[1]);
+        return 1;
+    }
+
+    if (test_bitmap.data[2] == 0x11111111)
+        fprintf(stderr, "%s() INDEX 2 is 0x11111111 set OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 2 is %lx expected 0x11111111\n", __FUNCTION__, test_bitmap.data[2]);
+        return 1;
+    }
+
+    if (test_bitmap.data[3] == 0x11111111)
+        fprintf(stderr, "%s() INDEX 3 is 0x11111111 set OK\n", __FUNCTION__);
+    else {
+        fprintf(stderr, "%s() INDEX 3 is %lx expected 0x11111111\n", __FUNCTION__, test_bitmap.data[3]);
+        return 1;
+    }
+
+    fprintf(stderr, "%s() tests passed\n", __FUNCTION__);
+    return 0;
+}
 
 #ifdef ENABLE_DBENGINE
 static inline void rrddim_set_by_pointer_fake_time(RRDDIM *rd, collected_number value, time_t now)
diff --git a/daemon/unit_test.h b/daemon/unit_test.h
index 6a7a966c3e..2d2533afed 100644
--- a/daemon/unit_test.h
+++ b/daemon/unit_test.h
@@ -10,6 +10,7 @@ extern int unit_test_str2ld(void);
 extern int unit_test_buffer(void);
 extern int unit_test_static_threads(void);
 extern int test_sqlite(void);
+extern int unit_test_bitmap256(void);
 #ifdef ENABLE_DBENGINE
 extern int test_dbengine(void);
 extern void generate_dbengine_dataset(unsigned history_seconds);
diff --git a/database/engine/journalfile.c b/database/engine/journalfile.c
index 531ed559d7..dc61f569d4 100644
--- a/database/engine/journalfile.c
+++ b/database/engine/journalfile.c
@@ -275,6 +275,7 @@ static int check_journal_file_superblock(uv_file file)
 static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrdengine_journalfile *journalfile,
                                     void *buf, unsigned max_size)
 {
+    static BITMAP256 page_error_map;
     struct page_cache *pg_cache = &ctx->pg_cache;
     unsigned i, count, payload_length, descr_size, valid_pages;
     struct rrdeng_page_descr *descr;
@@ -301,11 +302,31 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
         uuid_t *temp_id;
         Pvoid_t *PValue;
         struct pg_cache_page_index *page_index = NULL;
+        uint8_t page_type = jf_metric_data->descr[i].type;
 
-        if (jf_metric_data->descr[i].type > PAGE_TYPE_MAX) {
-            error("Unknown page type %d encountered.", jf_metric_data->descr[i].type );
+        if (page_type > PAGE_TYPE_MAX) {
+            if (!bitmap256_get_bit(&page_error_map, page_type)) {
+                error("Unknown page type %d encountered.", page_type);
+                bitmap256_set_bit(&page_error_map, page_type, 1);
+            }
             continue;
         }
+        uint64_t start_time = jf_metric_data->descr[i].start_time;
+        uint64_t end_time = jf_metric_data->descr[i].end_time;
+
+        if (unlikely(start_time > end_time)) {
+            error("Invalid page encountered, start time %lu > end time %lu", start_time , end_time );
+            continue;
+        }
+
+        if (unlikely(start_time == end_time)) {
+            size_t entries = jf_metric_data->descr[i].page_length / page_type_size[page_type];
+            if (unlikely(entries > 1)) {
+                error("Invalid page encountered, start time %lu = end time but %zu entries were found", start_time, entries);
+                continue;
+            }
+        }
+
         temp_id = (uuid_t *)jf_metric_data->descr[i].uuid;
 
         uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
@@ -327,11 +348,11 @@ static void restore_extent_metadata(struct rrdengine_instance *ctx, struct rrden
 
         descr = pg_cache_create_descr();
         descr->page_length = jf_metric_data->descr[i].page_length;
-        descr->start_time = jf_metric_data->descr[i].start_time;
-        descr->end_time = jf_metric_data->descr[i].end_time;
+        descr->start_time = start_time;
+        descr->end_time = end_time;
         descr->id = &page_index->id;
         descr->extent = extent;
-        descr->type = jf_metric_data->descr[i].type;
+        descr->type = page_type;
         extent->pages[valid_pages++] = descr;
         pg_cache_insert(ctx, page_index, descr);
     }
diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index ae4a1bffe5..39f7642d0a 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -1334,7 +1334,7 @@ void free_page_cache(struct rrdengine_instance *ctx)
             pages_size_per_type[descr->type] += descr->page_length;
             pages_count_per_type[descr->type]++;
 
-            size_t points_in_page = (descr->page_length / ctx->storage_size);
+            size_t points_in_page = (descr->page_length / PAGE_POINT_SIZE_BYTES(descr));
             size_t page_duration  = ((descr->end_time - descr->start_time) / USEC_PER_SEC);
             size_t update_every = (page_duration == 0) ? 1 : page_duration / (points_in_page - 1);
 
@@ -1351,7 +1351,7 @@ void free_page_cache(struct rrdengine_instance *ctx)
                 page_duration = update_every * points_in_page;
                 metric_duration += page_duration;
                 seconds_in_db += page_duration;
-                points_in_db += descr->page_length / ctx->storage_size;
+                points_in_db += descr->page_length / PAGE_POINT_SIZE_BYTES(descr);
             }
             else
                 metric_single_point_pages++;
diff --git a/database/engine/rrddiskprotocol.h b/database/engine/rrddiskprotocol.h
index 88c6e50696..cb57385a4b 100644
--- a/database/engine/rrddiskprotocol.h
+++ b/database/engine/rrddiskprotocol.h
@@ -36,7 +36,7 @@ struct rrdeng_df_sb {
  */
 #define PAGE_METRICS    (0)
 #define PAGE_TIER       (1)
-#define PAGE_TYPE_MAX   (1)
+#define PAGE_TYPE_MAX   1   // Maximum page type (inclusive)
 
 /*
  * Data file page descriptor
diff --git a/database/engine/rrdengine.h b/database/engine/rrdengine.h
index 7593700dd1..efb25d2f22 100644
--- a/database/engine/rrdengine.h
+++ b/database/engine/rrdengine.h
@@ -243,14 +243,13 @@ struct rrdengine_instance {
     uint64_t disk_space;
     uint64_t max_disk_space;
     int tier;
-    size_t storage_size;
     unsigned last_fileno; /* newest index of datafile and journalfile */
     unsigned long max_cache_pages;
     unsigned long cache_pages_low_watermark;
     unsigned long metric_API_max_producers;
 
     uint8_t quiesce; /* set to SET_QUIESCE before shutdown of the engine */
-    uint8_t page_type; /* set to SET_QUIESCE before shutdown of the engine */
+    uint8_t page_type; /* Default page type for this context */
 
     struct rrdengine_statistics stats;
 };
diff --git a/database/engine/rrdengineapi.c b/database/engine/rrdengineapi.c
index 587f1b12e8..053b686e67 100755
--- a/database/engine/rrdengineapi.c
+++ b/database/engine/rrdengineapi.c
@@ -11,6 +11,12 @@ struct rrdengine_instance multidb_ctx_storage_tier4;
 #error RRD_STORAGE_TIERS is not 5 - you need to add allocations here
 #endif
 struct rrdengine_instance *multidb_ctx[RRD_STORAGE_TIERS];
+uint8_t tier_page_type[RRD_STORAGE_TIERS] = {PAGE_METRICS, PAGE_TIER, PAGE_TIER, PAGE_TIER, PAGE_TIER};
+
+#if PAGE_TYPE_MAX != 1
+#error PAGE_TYPE_MAX is not 1 - you need to add allocations here
+#endif
+size_t page_type_size[256] = {sizeof(storage_number), sizeof(storage_number_tier1_t)};
 
 __attribute__((constructor)) void initialize_multidb_ctx(void) {
     multidb_ctx[0] = &multidb_ctx_storage_tier0;
@@ -169,14 +175,14 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri
 }
 
 /* The page must be populated and referenced */
-static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr, size_t storage_size)
+static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr)
 {
     unsigned i;
     uint8_t has_only_empty_metrics = 1;
     storage_number *page;
 
     page = descr->pg_cache_descr->page;
-    for (i = 0 ; i < descr->page_length / storage_size; ++i) {
+    for (i = 0 ; i < descr->page_length / PAGE_POINT_SIZE_BYTES(descr); ++i) {
         if (SN_EMPTY_SLOT != page[i]) {
             has_only_empty_metrics = 0;
             break;
@@ -199,7 +205,7 @@ void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_h
 
         rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1);
 
-        page_is_empty = page_has_only_empty_metrics(descr, ctx->storage_size);
+        page_is_empty = page_has_only_empty_metrics(descr);
         if (page_is_empty) {
             debug(D_RRDENGINE, "Page has empty metrics only, deleting:");
             if (unlikely(debug_flags & D_RRDENGINE))
@@ -232,7 +238,6 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
 
     void *page;
     uint8_t must_flush_unaligned_page = 0, perfect_page_alignment = 0;
-    size_t storage_size = ctx->storage_size;
 
     if (descr) {
         /* Make alignment decisions */
@@ -242,7 +247,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
             perfect_page_alignment = 1;
         }
         /* is the metric far enough out of alignment with the others? */
-        if (unlikely(descr->page_length + storage_size < rd->rrdset->rrddim_page_alignment)) {
+        if (unlikely(descr->page_length + PAGE_POINT_SIZE_BYTES(descr) < rd->rrdset->rrddim_page_alignment)) {
             handle->unaligned_page = 1;
             debug(D_RRDENGINE, "Metric page is not aligned with chart:");
             if (unlikely(debug_flags & D_RRDENGINE))
@@ -250,14 +255,14 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
         }
         if (unlikely(handle->unaligned_page &&
                      /* did the other metrics change page? */
-                     rd->rrdset->rrddim_page_alignment <= storage_size)) {
+                     rd->rrdset->rrddim_page_alignment <= PAGE_POINT_SIZE_BYTES(descr))) {
             debug(D_RRDENGINE, "Flushing unaligned metric page.");
             must_flush_unaligned_page = 1;
             handle->unaligned_page = 0;
         }
     }
     if (unlikely(NULL == descr ||
-                 descr->page_length + storage_size > RRDENG_BLOCK_SIZE ||
+                 descr->page_length + PAGE_POINT_SIZE_BYTES(descr) > RRDENG_BLOCK_SIZE ||
                  must_flush_unaligned_page)) {
         rrdeng_store_metric_flush_current_page(collection_handle);
 
@@ -278,7 +283,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
 
     switch (descr->type) {
         case PAGE_METRICS: {
-            ((storage_number *)page)[descr->page_length / storage_size] = pack_storage_number(n, flags);
+            ((storage_number *)page)[descr->page_length / PAGE_POINT_SIZE_BYTES(descr)] = pack_storage_number(n, flags);
         }
         break;
 
@@ -289,7 +294,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
             number_tier1.max_value = (float)max_value;
             number_tier1.anomaly_count = anomaly_count;
             number_tier1.count = count;
-            ((storage_number_tier1_t *)page)[descr->page_length / storage_size] = number_tier1;
+            ((storage_number_tier1_t *)page)[descr->page_length / PAGE_POINT_SIZE_BYTES(descr)] = number_tier1;
         }
         break;
 
@@ -303,7 +308,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t
         break;
     }
 
-    pg_cache_atomic_set_pg_info(descr, point_in_time, descr->page_length + storage_size);
+    pg_cache_atomic_set_pg_info(descr, point_in_time, descr->page_length + PAGE_POINT_SIZE_BYTES(descr));
 
     if (perfect_page_alignment)
         rd->rrdset->rrddim_page_alignment = descr->page_length;
@@ -441,7 +446,7 @@ static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle) {
 
     if (unlikely(descr->start_time != page_end_time && next_page_time > descr->start_time)) {
         // we're in the middle of the page somewhere
-        unsigned entries = page_length / ctx->storage_size;
+        unsigned entries = page_length / PAGE_POINT_SIZE_BYTES(descr);
         position = ((uint64_t)(next_page_time - descr->start_time)) * (entries - 1) /
                    (page_end_time - descr->start_time);
     }
@@ -451,7 +456,7 @@ static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle) {
     handle->page_end_time = page_end_time;
     handle->page_length = page_length;
     handle->page = descr->pg_cache_descr->page;
-    usec_t entries = handle->entries = page_length / ctx->storage_size;
+    usec_t entries = handle->entries = page_length / PAGE_POINT_SIZE_BYTES(descr);
     if (likely(entries > 1))
         handle->dt = (page_end_time - descr->start_time) / (entries - 1);
     else {
@@ -815,14 +820,12 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
     if(NULL == ctxp) {
         ctx = multidb_ctx[tier];
         memset(ctx, 0, sizeof(*ctx));
-        ctx->storage_size = (tier == 0) ? sizeof(storage_number) : sizeof(storage_number_tier1_t);
-        ctx->tier = tier;
-        ctx->page_type = !tier ? PAGE_METRICS : PAGE_TIER;  // TODO: In the future it can be different page type per tier
     }
     else {
         *ctxp = ctx = callocz(1, sizeof(*ctx));
-        ctx->storage_size = sizeof(storage_number);
     }
+    ctx->tier = tier;
+    ctx->page_type = tier_page_type[tier];
     ctx->global_compress_alg = RRD_LZ4;
     if (page_cache_mb < RRDENG_MIN_PAGE_CACHE_SIZE_MB)
         page_cache_mb = RRDENG_MIN_PAGE_CACHE_SIZE_MB;
diff --git a/database/engine/rrdengineapi.h b/database/engine/rrdengineapi.h
index 2839129fa7..a8e9522544 100644
--- a/database/engine/rrdengineapi.h
+++ b/database/engine/rrdengineapi.h
@@ -20,6 +20,9 @@ extern int default_rrdeng_disk_quota_mb;
 extern int default_multidb_disk_quota_mb;
 extern uint8_t rrdeng_drop_metrics_under_page_cache_pressure;
 extern struct rrdengine_instance *multidb_ctx[RRD_STORAGE_TIERS];
+extern size_t page_type_size[];
+
+#define PAGE_POINT_SIZE_BYTES(x) page_type_size[(x)->type]
 
 struct rrdeng_region_info {
     time_t start_time;
diff --git a/libnetdata/libnetdata.c b/libnetdata/libnetdata.c
index 2997ce19e4..212eceb824 100644
--- a/libnetdata/libnetdata.c
+++ b/libnetdata/libnetdata.c
@@ -1534,3 +1534,19 @@ char *find_and_replace(const char *src, const char *find, const char *replace, c
 
     return value;
 }
+
+
+bool bitmap256_get_bit(BITMAP256 *ptr, uint8_t idx) {
+    if (unlikely(!ptr))
+        return false;
+    return (ptr->data[idx / 64] & (1 << (idx % 64)));
+}
+
+void bitmap256_set_bit(BITMAP256 *ptr, uint8_t idx, bool value) {
+    if (unlikely(!ptr))
+        return;
+    if (likely(value))
+        ptr->data[idx / 64] |= (1U << (idx % 64));
+    else
+        ptr->data[idx / 64] &= ~(1U << (idx % 64));
+}
diff --git a/libnetdata/libnetdata.h b/libnetdata/libnetdata.h
index 4e0b45426c..e88e5978a8 100644
--- a/libnetdata/libnetdata.h
+++ b/libnetdata/libnetdata.h
@@ -311,6 +311,12 @@ extern char *find_and_replace(const char *src, const char *find, const char *rep
 // Taken from linux kernel
 #define BUILD_BUG_ON(condition) ((void)sizeof(char[1 - 2*!!(condition)]))
 
+typedef struct bitmap256 {
+    uint64_t data[4];
+} BITMAP256;
+
+extern bool bitmap256_get_bit(BITMAP256 *ptr, uint8_t idx);
+extern void bitmap256_set_bit(BITMAP256 *ptr, uint8_t idx, bool value);
 
 extern void netdata_cleanup_and_exit(int ret) NORETURN;
 extern void send_statistics(const char *action, const char *action_result, const char *action_data);