fix v1.37 dbengine page alignment crashes (#14086)

page alignment is a property of metric collection, so it has been moved to metric collection only

10 files changed, 44 insertions, 69 deletions

diff --git a/database/engine/pagecache.c b/database/engine/pagecache.c
index 5e5766d2f1..4f5da7084d 100644
--- a/database/engine/pagecache.c
+++ b/database/engine/pagecache.c
@@ -1204,7 +1204,6 @@ struct pg_cache_page_index *create_page_index(uuid_t *id, struct rrdengine_insta
     page_index->refcount = 0;
     page_index->writers = 0;
     page_index->ctx = ctx;
-    page_index->alignment = NULL;
     page_index->latest_update_every_s = default_rrd_update_every;
 
     return page_index;
diff --git a/database/engine/pagecache.h b/database/engine/pagecache.h
index 2f4d6b332a..635b02123c 100644
--- a/database/engine/pagecache.h
+++ b/database/engine/pagecache.h
@@ -112,7 +112,6 @@ struct pg_cache_page_index {
     usec_t latest_time_ut;
 
     struct rrdengine_instance *ctx;
-    struct pg_alignment *alignment;
     uint32_t latest_update_every_s;
 
     struct pg_cache_page_index *prev;
diff --git a/database/engine/rrdengine.h b/database/engine/rrdengine.h
index aca8184380..521d2521a2 100644
--- a/database/engine/rrdengine.h
+++ b/database/engine/rrdengine.h
@@ -41,6 +41,7 @@ struct rrdeng_collect_handle {
     unsigned long page_correlation_id;
     // set to 1 when this dimension is not page aligned with the other dimensions in the chart
     uint8_t unaligned_page;
+    struct pg_alignment *alignment;
 };
 
 struct rrdeng_query_handle {
diff --git a/database/engine/rrdengineapi.c b/database/engine/rrdengineapi.c
index 6f2a32777e..4525b041f2 100755
--- a/database/engine/rrdengineapi.c
+++ b/database/engine/rrdengineapi.c
@@ -63,16 +63,11 @@ STORAGE_METRICS_GROUP *rrdeng_metrics_group_get(STORAGE_INSTANCE *db_instance __
 }
 
 // charts call this
-void rrdeng_metrics_group_release(STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg) {
+void rrdeng_metrics_group_release(STORAGE_INSTANCE *db_instance __maybe_unused, STORAGE_METRICS_GROUP *smg) {
     if(unlikely(!smg)) return;
 
-    struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
     struct pg_alignment *pa = (struct pg_alignment *)smg;
-    struct page_cache *pg_cache = &ctx->pg_cache;
-
-    uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
     rrdeng_page_alignment_release(pa);
-    uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
 }
 
 // ----------------------------------------------------------------------------
@@ -112,10 +107,10 @@ void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uu
     memcpy(ret_uuid, hash_value, sizeof(uuid_t));
 }
 
-STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id, STORAGE_METRICS_GROUP *smg) {
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id) {
     uuid_t legacy_uuid;
     rrdeng_generate_legacy_uuid(rd_id, st_id, &legacy_uuid);
-    return rrdeng_metric_get(db_instance, &legacy_uuid, smg);
+    return rrdeng_metric_get(db_instance, &legacy_uuid);
 }
 
 // ----------------------------------------------------------------------------
@@ -124,11 +119,7 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, c
 void rrdeng_metric_release(STORAGE_METRIC_HANDLE *db_metric_handle) {
     struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
 
-    unsigned short refcount = __atomic_sub_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST);
-    if(refcount == 0 && page_index->alignment) {
-        __atomic_sub_fetch(&page_index->alignment->refcount, 1, __ATOMIC_SEQ_CST);
-        page_index->alignment = NULL;
-    }
+    __atomic_sub_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST);
 }
 
 STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle) {
@@ -137,9 +128,8 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle
     return db_metric_handle;
 }
 
-STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg) {
+STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid) {
     struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
-    struct pg_alignment *pa = (struct pg_alignment *)smg;
     struct page_cache *pg_cache = &ctx->pg_cache;
     struct pg_cache_page_index *page_index = NULL;
 
@@ -149,28 +139,16 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *
         page_index = *PValue;
     uv_rwlock_rdunlock(&pg_cache->metrics_index.lock);
 
-    if (likely(page_index)) {
+    if (likely(page_index))
         __atomic_add_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST);
 
-        if(pa) {
-            if(page_index->alignment != pa) {
-                if(!rrdeng_page_alignment_release(page_index->alignment)) // NULL is ok
-                    error("DBENGINE: metric switched alignment, but the previous is still used.");
-
-                rrdeng_page_alignment_acquire(pa);
-                page_index->alignment = pa;
-            }
-        }
-    }
-
     return (STORAGE_METRIC_HANDLE *)page_index;
 }
 
-STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg) {
+STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid) {
     internal_fatal(!db_instance, "DBENGINE: db_instance is NULL");
 
     struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance;
-    struct pg_alignment *pa = (struct pg_alignment *)smg;
     struct pg_cache_page_index *page_index;
     struct page_cache *pg_cache = &ctx->pg_cache;
 
@@ -180,27 +158,25 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_
     *PValue = page_index = create_page_index(uuid, ctx);
     page_index->prev = pg_cache->metrics_index.last_page_index;
     pg_cache->metrics_index.last_page_index = page_index;
-    page_index->alignment = pa;
     page_index->refcount = 1;
-    rrdeng_page_alignment_acquire(pa);
     uv_rwlock_wrunlock(&pg_cache->metrics_index.lock);
 
     return (STORAGE_METRIC_HANDLE *)page_index;
 }
 
-STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg) {
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance) {
     STORAGE_METRIC_HANDLE *db_metric_handle;
 
-    db_metric_handle = rrdeng_metric_get(db_instance, &rd->metric_uuid, smg);
+    db_metric_handle = rrdeng_metric_get(db_instance, &rd->metric_uuid);
     if(!db_metric_handle) {
-        db_metric_handle = rrdeng_metric_get_legacy(db_instance, rrddim_id(rd), rrdset_id(rd->rrdset), smg);
+        db_metric_handle = rrdeng_metric_get_legacy(db_instance, rrddim_id(rd), rrdset_id(rd->rrdset));
         if(db_metric_handle) {
             struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
             uuid_copy(rd->metric_uuid, page_index->id);
         }
     }
     if(!db_metric_handle)
-        db_metric_handle = rrdeng_metric_create(db_instance, &rd->metric_uuid, smg);
+        db_metric_handle = rrdeng_metric_create(db_instance, &rd->metric_uuid);
 
 #ifdef NETDATA_INTERNAL_CHECKS
     struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
@@ -229,19 +205,19 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE
  * Gets a handle for storing metrics to the database.
  * The handle must be released with rrdeng_store_metric_final().
  */
-STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every) {
+STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every, STORAGE_METRICS_GROUP *smg) {
     struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle;
     struct rrdeng_collect_handle *handle;
 
-    if(!page_index->alignment)
-        fatal("DBENGINE: metric group is required for collect operations");
-
     handle = callocz(1, sizeof(struct rrdeng_collect_handle));
     handle->page_index = page_index;
     handle->descr = NULL;
     handle->unaligned_page = 0;
     page_index->latest_update_every_s = update_every;
 
+    handle->alignment = (struct pg_alignment *)smg;
+    rrdeng_page_alignment_acquire(handle->alignment);
+
     uv_rwlock_wrlock(&page_index->lock);
     ++page_index->writers;
     uv_rwlock_wrunlock(&page_index->lock);
@@ -350,18 +326,18 @@ static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection
         }
 #endif
 
-        if (descr->page_length == page_index->alignment->page_length) {
+        if (descr->page_length == handle->alignment->page_length) {
             /* this is the leading dimension that defines chart alignment */
             perfect_page_alignment = 1;
         }
         /* is the metric far enough out of alignment with the others? */
-        if (unlikely(descr->page_length + PAGE_POINT_SIZE_BYTES(descr) < page_index->alignment->page_length)) {
+        if (unlikely(descr->page_length + PAGE_POINT_SIZE_BYTES(descr) < handle->alignment->page_length)) {
             handle->unaligned_page = 1;
             print_page_cache_descr(descr, "Metric page is not aligned with chart", true);
         }
         if (unlikely(handle->unaligned_page &&
                      /* did the other metrics change page? */
-                     page_index->alignment->page_length <= PAGE_POINT_SIZE_BYTES(descr))) {
+                     handle->alignment->page_length <= PAGE_POINT_SIZE_BYTES(descr))) {
             print_page_cache_descr(descr, "must_flush_unaligned_page = 1", true);
             must_flush_unaligned_page = 1;
             handle->unaligned_page = 0;
@@ -384,7 +360,7 @@ static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection
 
         handle->page_correlation_id = rrd_atomic_fetch_add(&pg_cache->committed_page_index.latest_corr_id, 1);
 
-        if (0 == page_index->alignment->page_length) {
+        if (0 == handle->alignment->page_length) {
             /* this is the leading dimension that defines chart alignment */
             perfect_page_alignment = 1;
         }
@@ -422,7 +398,7 @@ static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection
     pg_cache_atomic_set_pg_info(descr, point_in_time_ut, descr->page_length + PAGE_POINT_SIZE_BYTES(descr));
 
     if (perfect_page_alignment)
-        page_index->alignment->page_length = descr->page_length;
+        handle->alignment->page_length = descr->page_length;
     if (unlikely(INVALID_TIME == descr->start_time_ut)) {
         unsigned long new_metric_API_producers, old_metric_API_max_producers, ret_metric_API_max_producers;
         descr->start_time_ut = point_in_time_ut;
@@ -549,13 +525,13 @@ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) {
 
     rrdeng_store_metric_flush_current_page(collection_handle);
     uv_rwlock_wrlock(&page_index->lock);
-    if (!--page_index->writers && !page_index->page_count) {
+
+    if (!--page_index->writers && !page_index->page_count)
         can_delete_metric = 1;
 
-        rrdeng_page_alignment_release(page_index->alignment);
-        page_index->alignment = NULL;
-    }
     uv_rwlock_wrunlock(&page_index->lock);
+
+    rrdeng_page_alignment_release(handle->alignment);
     freez(handle);
 
     return can_delete_metric;
diff --git a/database/engine/rrdengineapi.h b/database/engine/rrdengineapi.h
index 85375044fc..3acee4ec68 100644
--- a/database/engine/rrdengineapi.h
+++ b/database/engine/rrdengineapi.h
@@ -42,14 +42,14 @@ void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uu
                                                     uuid_t *ret_uuid);
 
 
-STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg);
-STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg);
-STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg);
-STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id, STORAGE_METRICS_GROUP *smg);
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance);
+STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid);
+STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid);
+STORAGE_METRIC_HANDLE *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id);
 void rrdeng_metric_release(STORAGE_METRIC_HANDLE *db_metric_handle);
 STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle);
 
-STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every);
+STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every, STORAGE_METRICS_GROUP *smg);
 void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_handle);
 void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *collection_handle, int update_every);
 void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, usec_t point_in_time_ut, NETDATA_DOUBLE n,
diff --git a/database/ram/rrddim_mem.c b/database/ram/rrddim_mem.c
index 43f32350b8..299b6557a8 100644
--- a/database/ram/rrddim_mem.c
+++ b/database/ram/rrddim_mem.c
@@ -22,8 +22,8 @@ void rrddim_metrics_group_release(STORAGE_INSTANCE *db_instance __maybe_unused,
 // RRDDIM legacy data collection functions
 
 STORAGE_METRIC_HANDLE *
-rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance __maybe_unused, STORAGE_METRICS_GROUP *smg __maybe_unused) {
-    STORAGE_METRIC_HANDLE *t = rrddim_metric_get(db_instance, &rd->metric_uuid, smg);
+rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance __maybe_unused) {
+    STORAGE_METRIC_HANDLE *t = rrddim_metric_get(db_instance, &rd->metric_uuid);
     if(!t) {
         netdata_rwlock_wrlock(&rrddim_JudyHS_rwlock);
         Pvoid_t *PValue = JudyHSIns(&rrddim_JudyHS_array, &rd->metric_uuid, sizeof(uuid_t), PJE0);
@@ -40,7 +40,7 @@ rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance __maybe_un
 }
 
 STORAGE_METRIC_HANDLE *
-rrddim_metric_get(STORAGE_INSTANCE *db_instance __maybe_unused, uuid_t *uuid, STORAGE_METRICS_GROUP *smg __maybe_unused) {
+rrddim_metric_get(STORAGE_INSTANCE *db_instance __maybe_unused, uuid_t *uuid) {
     RRDDIM *rd = NULL;
     netdata_rwlock_rdlock(&rrddim_JudyHS_rwlock);
     Pvoid_t *PValue = JudyHSGet(rrddim_JudyHS_array, uuid, sizeof(uuid_t));
@@ -67,7 +67,7 @@ void rrddim_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *col
     rrddim_store_metric_flush(collection_handle);
 }
 
-STORAGE_COLLECT_HANDLE *rrddim_collect_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every __maybe_unused) {
+STORAGE_COLLECT_HANDLE *rrddim_collect_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every __maybe_unused, STORAGE_METRICS_GROUP *smg __maybe_unused) {
     RRDDIM *rd = (RRDDIM *)db_metric_handle;
     rd->db[rd->rrdset->current_entry] = pack_storage_number(NAN, SN_FLAG_NONE);
     struct mem_collect_handle *ch = callocz(1, sizeof(struct mem_collect_handle));
diff --git a/database/ram/rrddim_mem.h b/database/ram/rrddim_mem.h
index 297388f51f..79c59f1105 100644
--- a/database/ram/rrddim_mem.h
+++ b/database/ram/rrddim_mem.h
@@ -20,15 +20,15 @@ struct mem_query_handle {
     size_t last_slot;
 };
 
-STORAGE_METRIC_HANDLE *rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg);
-STORAGE_METRIC_HANDLE *rrddim_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg);
+STORAGE_METRIC_HANDLE *rrddim_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance);
+STORAGE_METRIC_HANDLE *rrddim_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid);
 STORAGE_METRIC_HANDLE *rrddim_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle);
 void rrddim_metric_release(STORAGE_METRIC_HANDLE *db_metric_handle);
 
 STORAGE_METRICS_GROUP *rrddim_metrics_group_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid);
 void rrddim_metrics_group_release(STORAGE_INSTANCE *db_instance, STORAGE_METRICS_GROUP *smg);
 
-STORAGE_COLLECT_HANDLE *rrddim_collect_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every);
+STORAGE_COLLECT_HANDLE *rrddim_collect_init(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every, STORAGE_METRICS_GROUP *smg);
 void rrddim_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *collection_handle, int update_every);
 void rrddim_collect_store_metric(STORAGE_COLLECT_HANDLE *collection_handle, usec_t point_in_time, NETDATA_DOUBLE number,
                                  NETDATA_DOUBLE min_value,
diff --git a/database/rrd.h b/database/rrd.h
index f071ee2548..0796ff9012 100644
--- a/database/rrd.h
+++ b/database/rrd.h
@@ -409,7 +409,7 @@ typedef struct storage_query_handle STORAGE_QUERY_HANDLE;
 // function pointers that handle data collection
 struct storage_engine_collect_ops {
     // an initialization function to run before starting collection
-    STORAGE_COLLECT_HANDLE *(*init)(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every);
+    STORAGE_COLLECT_HANDLE *(*init)(STORAGE_METRIC_HANDLE *db_metric_handle, uint32_t update_every, STORAGE_METRICS_GROUP *smg);
 
     // run this to store each metric into the database
     void (*store_metric)(STORAGE_COLLECT_HANDLE *collection_handle, usec_t point_in_time, NETDATA_DOUBLE number, NETDATA_DOUBLE min_value,
@@ -464,8 +464,8 @@ typedef struct storage_engine STORAGE_ENGINE;
 // function pointers for all APIs provided by a storage engine
 typedef struct storage_engine_api {
     // metric management
-    STORAGE_METRIC_HANDLE *(*metric_get)(STORAGE_INSTANCE *instance, uuid_t *uuid, STORAGE_METRICS_GROUP *smg);
-    STORAGE_METRIC_HANDLE *(*metric_get_or_create)(RRDDIM *rd, STORAGE_INSTANCE *instance, STORAGE_METRICS_GROUP *smg);
+    STORAGE_METRIC_HANDLE *(*metric_get)(STORAGE_INSTANCE *instance, uuid_t *uuid);
+    STORAGE_METRIC_HANDLE *(*metric_get_or_create)(RRDDIM *rd, STORAGE_INSTANCE *instance);
     void (*metric_release)(STORAGE_METRIC_HANDLE *);
     STORAGE_METRIC_HANDLE *(*metric_dup)(STORAGE_METRIC_HANDLE *);
 
diff --git a/database/rrdcontext.c b/database/rrdcontext.c
index 28230c9066..3413d1ea82 100644
--- a/database/rrdcontext.c
+++ b/database/rrdcontext.c
@@ -2436,7 +2436,7 @@ static void query_target_add_metric(QUERY_TARGET_LOCALS *qtl, RRDMETRIC_ACQUIRED
         if(rm->rrddim && rm->rrddim->tiers[tier] && rm->rrddim->tiers[tier]->db_metric_handle)
             tier_retention[tier].db_metric_handle = eng->api.metric_dup(rm->rrddim->tiers[tier]->db_metric_handle);
         else
-            tier_retention[tier].db_metric_handle = eng->api.metric_get(qtl->host->db[tier].instance, &rm->uuid, NULL);
+            tier_retention[tier].db_metric_handle = eng->api.metric_get(qtl->host->db[tier].instance, &rm->uuid);
 
         if(tier_retention[tier].db_metric_handle) {
             tier_retention[tier].db_first_time_t = tier_retention[tier].eng->api.query_ops.oldest_time(tier_retention[tier].db_metric_handle);
diff --git a/database/rrddim.c b/database/rrddim.c
index f3aeb26371..2d909a7015 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -112,7 +112,7 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
             rd->tiers[tier]->tier_grouping = host->db[tier].tier_grouping;
             rd->tiers[tier]->collect_ops = &eng->api.collect_ops;
             rd->tiers[tier]->query_ops = &eng->api.query_ops;
-            rd->tiers[tier]->db_metric_handle = eng->api.metric_get_or_create(rd, host->db[tier].instance, rd->rrdset->storage_metrics_groups[tier]);
+            rd->tiers[tier]->db_metric_handle = eng->api.metric_get_or_create(rd, host->db[tier].instance);
             storage_point_unset(rd->tiers[tier]->virtual_point);
             initialized++;
 
@@ -131,7 +131,7 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
         size_t initialized = 0;
         for (size_t tier = 0; tier < storage_tiers; tier++) {
             if (rd->tiers[tier]) {
-                rd->tiers[tier]->db_collection_handle = rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
+                rd->tiers[tier]->db_collection_handle = rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every, rd->rrdset->storage_metrics_groups[tier]);
                 initialized++;
             }
         }
@@ -261,7 +261,7 @@ static bool rrddim_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused,
     for(size_t tier = 0; tier < storage_tiers ;tier++) {
         if (rd->tiers[tier] && !rd->tiers[tier]->db_collection_handle)
             rd->tiers[tier]->db_collection_handle =
-                rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
+                rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every, rd->rrdset->storage_metrics_groups[tier]);
     }
 
     if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {