diff options
Diffstat (limited to 'database/engine/rrdengineapi.c')
| -rwxr-xr-x | database/engine/rrdengineapi.c | 1683 |
1 files changed, 909 insertions, 774 deletions
diff --git a/database/engine/rrdengineapi.c b/database/engine/rrdengineapi.c index 4525b041f..27497bbb8 100755 --- a/database/engine/rrdengineapi.c +++ b/database/engine/rrdengineapi.c @@ -1,6 +1,5 @@ // SPDX-License-Identifier: GPL-3.0-or-later #include "rrdengine.h" -#include "../storage_engine.h" /* Default global database instance */ struct rrdengine_instance multidb_ctx_storage_tier0; @@ -8,12 +7,21 @@ struct rrdengine_instance multidb_ctx_storage_tier1; struct rrdengine_instance multidb_ctx_storage_tier2; struct rrdengine_instance multidb_ctx_storage_tier3; struct rrdengine_instance multidb_ctx_storage_tier4; + +#define mrg_metric_ctx(metric) (struct rrdengine_instance *)mrg_metric_section(main_mrg, metric) + #if RRD_STORAGE_TIERS != 5 #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 defined(ENV32BIT) +size_t tier_page_size[RRD_STORAGE_TIERS] = {2048, 1024, 192, 192, 192}; +#else +size_t tier_page_size[RRD_STORAGE_TIERS] = {4096, 2048, 384, 384, 384}; +#endif + #if PAGE_TYPE_MAX != 1 #error PAGE_TYPE_MAX is not 1 - you need to add allocations here #endif @@ -27,14 +35,17 @@ __attribute__((constructor)) void initialize_multidb_ctx(void) { multidb_ctx[4] = &multidb_ctx_storage_tier4; } -int db_engine_use_malloc = 0; int default_rrdeng_page_fetch_timeout = 3; int default_rrdeng_page_fetch_retries = 3; -int default_rrdeng_page_cache_mb = 32; +int db_engine_journal_check = 0; int default_rrdeng_disk_quota_mb = 256; int default_multidb_disk_quota_mb = 256; -/* Default behaviour is to unblock data collection if the page cache is full of dirty pages by dropping metrics */ -uint8_t rrdeng_drop_metrics_under_page_cache_pressure = 1; + +#if defined(ENV32BIT) +int default_rrdeng_page_cache_mb = 16; +#else +int default_rrdeng_page_cache_mb = 32; +#endif // ---------------------------------------------------------------------------- // metrics groups @@ -90,161 +101,207 @@ void rrdeng_generate_legacy_uuid(const char *dim_id, const char *chart_id, uuid_ memcpy(ret_uuid, hash_value, sizeof(uuid_t)); } -/* Transform legacy UUID to be unique across hosts deterministically */ -void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uuid_t *legacy_uuid, uuid_t *ret_uuid) -{ - EVP_MD_CTX *evpctx; - unsigned char hash_value[EVP_MAX_MD_SIZE]; - unsigned int hash_len; - - evpctx = EVP_MD_CTX_create(); - EVP_DigestInit_ex(evpctx, EVP_sha256(), NULL); - EVP_DigestUpdate(evpctx, machine_guid, GUID_LEN); - EVP_DigestUpdate(evpctx, *legacy_uuid, sizeof(uuid_t)); - EVP_DigestFinal_ex(evpctx, hash_value, &hash_len); - EVP_MD_CTX_destroy(evpctx); - fatal_assert(hash_len > sizeof(uuid_t)); - 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) { +static METRIC *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const char *rd_id, const char *st_id) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; uuid_t legacy_uuid; rrdeng_generate_legacy_uuid(rd_id, st_id, &legacy_uuid); - return rrdeng_metric_get(db_instance, &legacy_uuid); + return mrg_metric_get_and_acquire(main_mrg, &legacy_uuid, (Word_t) ctx); } // ---------------------------------------------------------------------------- // metric handle 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; - - __atomic_sub_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST); + METRIC *metric = (METRIC *)db_metric_handle; + mrg_metric_release(main_mrg, metric); } STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle) { - struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle; - __atomic_add_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST); - return db_metric_handle; + METRIC *metric = (METRIC *)db_metric_handle; + return (STORAGE_METRIC_HANDLE *) mrg_metric_dup(main_mrg, metric); } STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *db_instance, uuid_t *uuid) { struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; - struct page_cache *pg_cache = &ctx->pg_cache; - struct pg_cache_page_index *page_index = NULL; - - uv_rwlock_rdlock(&pg_cache->metrics_index.lock); - Pvoid_t *PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, uuid, sizeof(uuid_t)); - if (likely(NULL != PValue)) - page_index = *PValue; - uv_rwlock_rdunlock(&pg_cache->metrics_index.lock); - - if (likely(page_index)) - __atomic_add_fetch(&page_index->refcount, 1, __ATOMIC_SEQ_CST); - - return (STORAGE_METRIC_HANDLE *)page_index; + return (STORAGE_METRIC_HANDLE *) mrg_metric_get_and_acquire(main_mrg, uuid, (Word_t) ctx); } -STORAGE_METRIC_HANDLE *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid) { +static METRIC *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_cache_page_index *page_index; - struct page_cache *pg_cache = &ctx->pg_cache; - - uv_rwlock_wrlock(&pg_cache->metrics_index.lock); - Pvoid_t *PValue = JudyHSIns(&pg_cache->metrics_index.JudyHS_array, uuid, sizeof(uuid_t), PJE0); - fatal_assert(NULL == *PValue); /* TODO: figure out concurrency model */ - *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->refcount = 1; - uv_rwlock_wrunlock(&pg_cache->metrics_index.lock); - - return (STORAGE_METRIC_HANDLE *)page_index; + MRG_ENTRY entry = { + .section = (Word_t)ctx, + .first_time_s = 0, + .last_time_s = 0, + .latest_update_every_s = 0, + }; + uuid_copy(entry.uuid, *uuid); + + METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, NULL); + return metric; } 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); - if(!db_metric_handle) { - 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); + struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; + METRIC *metric; + + metric = mrg_metric_get_and_acquire(main_mrg, &rd->metric_uuid, (Word_t) ctx); + + if(unlikely(!metric)) { + if(unlikely(ctx->config.legacy)) { + // this is a single host database + // generate uuid from the chart and dimensions ids + // and overwrite the one supplied by rrddim + metric = rrdeng_metric_get_legacy(db_instance, rrddim_id(rd), rrdset_id(rd->rrdset)); + if (metric) + uuid_copy(rd->metric_uuid, *mrg_metric_uuid(main_mrg, metric)); } + + if(likely(!metric)) + metric = rrdeng_metric_create(db_instance, &rd->metric_uuid); } - if(!db_metric_handle) - 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; - if(uuid_compare(rd->metric_uuid, page_index->id) != 0) { + if(uuid_compare(rd->metric_uuid, *mrg_metric_uuid(main_mrg, metric)) != 0) { char uuid1[UUID_STR_LEN + 1]; char uuid2[UUID_STR_LEN + 1]; uuid_unparse(rd->metric_uuid, uuid1); - uuid_unparse(page_index->id, uuid2); - fatal("DBENGINE: uuids do not match, asked for metric '%s', but got page_index of metric '%s'", uuid1, uuid2); + uuid_unparse(*mrg_metric_uuid(main_mrg, metric), uuid2); + fatal("DBENGINE: uuids do not match, asked for metric '%s', but got metric '%s'", uuid1, uuid2); } - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; - if(page_index->ctx != ctx) - fatal("DBENGINE: mixed up rrdengine instances, asked for metric from %p, got from %p", ctx, page_index->ctx); + if(mrg_metric_ctx(metric) != ctx) + fatal("DBENGINE: mixed up db instances, asked for metric from %p, got from %p", + ctx, mrg_metric_ctx(metric)); #endif - return db_metric_handle; + return (STORAGE_METRIC_HANDLE *)metric; } // ---------------------------------------------------------------------------- // collect ops +static inline void check_and_fix_mrg_update_every(struct rrdeng_collect_handle *handle) { + if(unlikely((time_t)(handle->update_every_ut / USEC_PER_SEC) != mrg_metric_get_update_every_s(main_mrg, handle->metric))) { + internal_error(true, "DBENGINE: collection handle has update every %ld, but the metric registry has %ld. Fixing it.", + (time_t)(handle->update_every_ut / USEC_PER_SEC), mrg_metric_get_update_every_s(main_mrg, handle->metric)); + + if(unlikely(!handle->update_every_ut)) + handle->update_every_ut = (usec_t)mrg_metric_get_update_every_s(main_mrg, handle->metric) * USEC_PER_SEC; + else + mrg_metric_set_update_every(main_mrg, handle->metric, (time_t)(handle->update_every_ut / USEC_PER_SEC)); + } +} + +static inline bool check_completed_page_consistency(struct rrdeng_collect_handle *handle __maybe_unused) { +#ifdef NETDATA_INTERNAL_CHECKS + if (unlikely(!handle->page || !handle->page_entries_max || !handle->page_position || !handle->page_end_time_ut)) + return false; + + struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); + + uuid_t *uuid = mrg_metric_uuid(main_mrg, handle->metric); + time_t start_time_s = pgc_page_start_time_s(handle->page); + time_t end_time_s = pgc_page_end_time_s(handle->page); + time_t update_every_s = pgc_page_update_every_s(handle->page); + size_t page_length = handle->page_position * CTX_POINT_SIZE_BYTES(ctx); + size_t entries = handle->page_position; + time_t overwrite_zero_update_every_s = (time_t)(handle->update_every_ut / USEC_PER_SEC); + + if(end_time_s > max_acceptable_collected_time()) + handle->page_flags |= RRDENG_PAGE_COMPLETED_IN_FUTURE; + + VALIDATED_PAGE_DESCRIPTOR vd = validate_page( + uuid, + start_time_s, + end_time_s, + update_every_s, + page_length, + ctx->config.page_type, + entries, + 0, // do not check for future timestamps - we inherit the timestamps of the children + overwrite_zero_update_every_s, + false, + "collected", + handle->page_flags); + + return vd.is_valid; +#else + return true; +#endif +} + /* * 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_METRICS_GROUP *smg) { - struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle; + METRIC *metric = (METRIC *)db_metric_handle; + struct rrdengine_instance *ctx = mrg_metric_ctx(metric); + + bool is_1st_metric_writer = true; + if(!mrg_metric_set_writer(main_mrg, metric)) { + is_1st_metric_writer = false; + char uuid[UUID_STR_LEN + 1]; + uuid_unparse(*mrg_metric_uuid(main_mrg, metric), uuid); + error("DBENGINE: metric '%s' is already collected and should not be collected twice - expect gaps on the charts", uuid); + } + + metric = mrg_metric_dup(main_mrg, metric); + struct rrdeng_collect_handle *handle; 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->metric = metric; + handle->page = NULL; + handle->page_position = 0; + handle->page_entries_max = 0; + handle->update_every_ut = (usec_t)update_every * USEC_PER_SEC; + handle->options = is_1st_metric_writer ? RRDENG_1ST_METRIC_WRITER : 0; + + __atomic_add_fetch(&ctx->atomic.collectors_running, 1, __ATOMIC_RELAXED); + if(!is_1st_metric_writer) + __atomic_add_fetch(&ctx->atomic.collectors_running_duplicate, 1, __ATOMIC_RELAXED); + + mrg_metric_set_update_every(main_mrg, metric, 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); + // this is important! + // if we don't set the page_end_time_ut during the first collection + // data collection may be able to go back in time and during the addition of new pages + // clean pages may be found matching ours! + + time_t db_first_time_s, db_last_time_s, db_update_every_s; + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, &db_update_every_s); + handle->page_end_time_ut = (usec_t)db_last_time_s * USEC_PER_SEC; return (STORAGE_COLLECT_HANDLE *)handle; } /* The page must be populated and referenced */ -static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr) -{ - switch(descr->type) { +static bool page_has_only_empty_metrics(struct rrdeng_collect_handle *handle) { + switch(handle->type) { case PAGE_METRICS: { - size_t slots = descr->page_length / PAGE_POINT_SIZE_BYTES(descr); - storage_number *array = (storage_number *)descr->pg_cache_descr->page; + size_t slots = handle->page_position; + storage_number *array = (storage_number *)pgc_page_data(handle->page); for (size_t i = 0 ; i < slots; ++i) { if(does_storage_number_exist(array[i])) - return 0; + return false; } } break; case PAGE_TIER: { - size_t slots = descr->page_length / PAGE_POINT_SIZE_BYTES(descr); - storage_number_tier1_t *array = (storage_number_tier1_t *)descr->pg_cache_descr->page; + size_t slots = handle->page_position; + storage_number_tier1_t *array = (storage_number_tier1_t *)pgc_page_data(handle->page); for (size_t i = 0 ; i < slots; ++i) { if(fpclassify(array[i].sum_value) != FP_NAN) - return 0; + return false; } } break; @@ -252,422 +309,585 @@ static int page_has_only_empty_metrics(struct rrdeng_page_descr *descr) default: { static bool logged = false; if(!logged) { - error("DBENGINE: cannot check page for nulls on unknown page type id %d", descr->type); + error("DBENGINE: cannot check page for nulls on unknown page type id %d", (mrg_metric_ctx(handle->metric))->config.page_type); logged = true; } - return 0; + return false; } } - return 1; + return true; } void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_handle) { struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; - // struct rrdeng_metric_handle *metric_handle = (struct rrdeng_metric_handle *)handle->metric_handle; - struct rrdengine_instance *ctx = handle->page_index->ctx; - struct rrdeng_page_descr *descr = handle->descr; - - if (unlikely(!ctx)) return; - if (unlikely(!descr)) return; - - if (likely(descr->page_length)) { - int page_is_empty; - - rrd_stat_atomic_add(&ctx->stats.metric_API_producers, -1); - - page_is_empty = page_has_only_empty_metrics(descr); - if (page_is_empty) { - print_page_cache_descr(descr, "Page has empty metrics only, deleting", true); - pg_cache_put(ctx, descr); - pg_cache_punch_hole(ctx, descr, 1, 0, NULL); - } else - rrdeng_commit_page(ctx, descr, handle->page_correlation_id); - } else { - dbengine_page_free(descr->pg_cache_descr->page); - rrdeng_destroy_pg_cache_descr(ctx, descr->pg_cache_descr); - rrdeng_page_descr_freez(descr); + + if (unlikely(!handle->page)) + return; + + if(!handle->page_position || page_has_only_empty_metrics(handle)) + pgc_page_to_clean_evict_or_release(main_cache, handle->page); + + else { + check_completed_page_consistency(handle); + mrg_metric_set_clean_latest_time_s(main_mrg, handle->metric, pgc_page_end_time_s(handle->page)); + pgc_page_hot_to_dirty_and_release(main_cache, handle->page); } - handle->descr = NULL; -} -static void rrdeng_store_metric_next_internal(STORAGE_COLLECT_HANDLE *collection_handle, - usec_t point_in_time_ut, - NETDATA_DOUBLE n, - NETDATA_DOUBLE min_value, - NETDATA_DOUBLE max_value, - uint16_t count, - uint16_t anomaly_count, - SN_FLAGS flags) -{ - struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; - struct pg_cache_page_index *page_index = handle->page_index; - struct rrdengine_instance *ctx = handle->page_index->ctx; - struct page_cache *pg_cache = &ctx->pg_cache; - struct rrdeng_page_descr *descr = handle->descr; + mrg_metric_set_hot_latest_time_s(main_mrg, handle->metric, 0); - void *page; - uint8_t must_flush_unaligned_page = 0, perfect_page_alignment = 0; + handle->page = NULL; + handle->page_flags = 0; + handle->page_position = 0; + handle->page_entries_max = 0; - if (descr) { - /* Make alignment decisions */ + // important! + // we should never zero page end time ut, because this will allow + // collection to go back in time + // handle->page_end_time_ut = 0; + // handle->page_start_time_ut; + + check_and_fix_mrg_update_every(handle); +} + +static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *handle, + struct rrdengine_instance *ctx, + usec_t point_in_time_ut, + void *data, + size_t data_size) { + time_t point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC); + const time_t update_every_s = (time_t)(handle->update_every_ut / USEC_PER_SEC); + + PGC_ENTRY page_entry = { + .section = (Word_t) ctx, + .metric_id = mrg_metric_id(main_mrg, handle->metric), + .start_time_s = point_in_time_s, + .end_time_s = point_in_time_s, + .size = data_size, + .data = data, + .update_every_s = update_every_s, + .hot = true + }; + + size_t conflicts = 0; + bool added = true; + PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, &added); + while (unlikely(!added)) { + conflicts++; + + char uuid[UUID_STR_LEN + 1]; + uuid_unparse(*mrg_metric_uuid(main_mrg, handle->metric), uuid); #ifdef NETDATA_INTERNAL_CHECKS - if(descr->end_time_ut + page_index->latest_update_every_s * USEC_PER_SEC != point_in_time_ut) { - char buffer[200 + 1]; - snprintfz(buffer, 200, - "metrics collected are %s, end_time_ut = %llu, point_in_time_ut = %llu, update_every = %u, delta = %llu", - (point_in_time_ut / USEC_PER_SEC - descr->end_time_ut / USEC_PER_SEC > page_index->latest_update_every_s)?"far apart":"not aligned", - descr->end_time_ut / USEC_PER_SEC, - point_in_time_ut / USEC_PER_SEC, - page_index->latest_update_every_s, - point_in_time_ut / USEC_PER_SEC - descr->end_time_ut / USEC_PER_SEC); - print_page_cache_descr(descr, buffer, false); - } + internal_error(true, +#else + error_limit_static_global_var(erl, 1, 0); + error_limit(&erl, #endif + "DBENGINE: metric '%s' new page from %ld to %ld, update every %ld, has a conflict in main cache " + "with existing %s%s page from %ld to %ld, update every %ld - " + "is it collected more than once?", + uuid, + page_entry.start_time_s, page_entry.end_time_s, (time_t)page_entry.update_every_s, + pgc_is_page_hot(page) ? "hot" : "not-hot", + pgc_page_data(page) == DBENGINE_EMPTY_PAGE ? " gap" : "", + pgc_page_start_time_s(page), pgc_page_end_time_s(page), pgc_page_update_every_s(page) + ); + + pgc_page_release(main_cache, page); + + point_in_time_ut -= handle->update_every_ut; + point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC); + page_entry.start_time_s = point_in_time_s; + page_entry.end_time_s = point_in_time_s; + page = pgc_page_add_and_acquire(main_cache, page_entry, &added); + } - 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) < 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? */ - 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; - } + handle->page_entries_max = data_size / CTX_POINT_SIZE_BYTES(ctx); + handle->page_start_time_ut = point_in_time_ut; + handle->page_end_time_ut = point_in_time_ut; + handle->page_position = 1; // zero is already in our data + handle->page = page; + handle->page_flags = conflicts? RRDENG_PAGE_CONFLICT : 0; + + if(point_in_time_s > max_acceptable_collected_time()) + handle->page_flags |= RRDENG_PAGE_CREATED_IN_FUTURE; + + check_and_fix_mrg_update_every(handle); +} + +static void *rrdeng_alloc_new_metric_data(struct rrdeng_collect_handle *handle, size_t *data_size, usec_t point_in_time_ut) { + struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); + size_t size; + + if(handle->options & RRDENG_FIRST_PAGE_ALLOCATED) { + // any page except the first + size = tier_page_size[ctx->config.tier]; } - if (unlikely(NULL == descr || - descr->page_length + PAGE_POINT_SIZE_BYTES(descr) > RRDENG_BLOCK_SIZE || - must_flush_unaligned_page)) { + else { + size_t final_slots = 0; - if(descr) { - print_page_cache_descr(descr, "flushing metric", true); - rrdeng_store_metric_flush_current_page(collection_handle); + // the first page + handle->options |= RRDENG_FIRST_PAGE_ALLOCATED; + size_t max_size = tier_page_size[ctx->config.tier]; + size_t max_slots = max_size / CTX_POINT_SIZE_BYTES(ctx); + + if(handle->alignment->initial_slots) { + final_slots = handle->alignment->initial_slots; } + else { + max_slots -= 3; - page = rrdeng_create_page(ctx, &page_index->id, &descr); - fatal_assert(page); + size_t smaller_slot = indexing_partition((Word_t)handle->alignment, max_slots); + final_slots = smaller_slot; - descr->update_every_s = page_index->latest_update_every_s; - handle->descr = descr; + time_t now_s = (time_t)(point_in_time_ut / USEC_PER_SEC); + size_t current_pos = (now_s % max_slots); - handle->page_correlation_id = rrd_atomic_fetch_add(&pg_cache->committed_page_index.latest_corr_id, 1); + if(current_pos > final_slots) + final_slots += max_slots - current_pos; - if (0 == handle->alignment->page_length) { - /* this is the leading dimension that defines chart alignment */ - perfect_page_alignment = 1; + else if(current_pos < final_slots) + final_slots -= current_pos; + + if(final_slots < 3) { + final_slots += 3; + smaller_slot += 3; + + if(smaller_slot >= max_slots) + smaller_slot -= max_slots; + } + + max_slots += 3; + handle->alignment->initial_slots = smaller_slot + 3; + + internal_fatal(handle->alignment->initial_slots < 3 || handle->alignment->initial_slots >= max_slots, "ooops! wrong distribution of metrics across time"); + internal_fatal(final_slots < 3 || final_slots >= max_slots, "ooops! wrong distribution of metrics across time"); } + + size = final_slots * CTX_POINT_SIZE_BYTES(ctx); } - page = descr->pg_cache_descr->page; + *data_size = size; + return dbengine_page_alloc(size); +} + +static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_handle, + const usec_t point_in_time_ut, + const NETDATA_DOUBLE n, + const NETDATA_DOUBLE min_value, + const NETDATA_DOUBLE max_value, + const uint16_t count, + const uint16_t anomaly_count, + const SN_FLAGS flags) +{ + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; + struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); - switch (descr->type) { + bool perfect_page_alignment = false; + void *data; + size_t data_size; + + if(likely(handle->page)) { + /* Make alignment decisions */ + if (handle->page_position == handle->alignment->page_position) { + /* this is the leading dimension that defines chart alignment */ + perfect_page_alignment = true; + } + + /* is the metric far enough out of alignment with the others? */ + if (unlikely(handle->page_position + 1 < handle->alignment->page_position)) + handle->options |= RRDENG_CHO_UNALIGNED; + + if (unlikely((handle->options & RRDENG_CHO_UNALIGNED) && + /* did the other metrics change page? */ + handle->alignment->page_position <= 1)) { + handle->options &= ~RRDENG_CHO_UNALIGNED; + handle->page_flags |= RRDENG_PAGE_UNALIGNED; + rrdeng_store_metric_flush_current_page(collection_handle); + + data = rrdeng_alloc_new_metric_data(handle, &data_size, point_in_time_ut); + } + else { + data = pgc_page_data(handle->page); + data_size = pgc_page_data_size(main_cache, handle->page); + } + } + else + data = rrdeng_alloc_new_metric_data(handle, &data_size, point_in_time_ut); + + switch (ctx->config.page_type) { case PAGE_METRICS: { - ((storage_number *)page)[descr->page_length / PAGE_POINT_SIZE_BYTES(descr)] = pack_storage_number(n, flags); + storage_number *tier0_metric_data = data; + tier0_metric_data[handle->page_position] = pack_storage_number(n, flags); } break; case PAGE_TIER: { + storage_number_tier1_t *tier12_metric_data = data; storage_number_tier1_t number_tier1; number_tier1.sum_value = (float)n; number_tier1.min_value = (float)min_value; 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 / PAGE_POINT_SIZE_BYTES(descr)] = number_tier1; + tier12_metric_data[handle->page_position] = number_tier1; } break; default: { static bool logged = false; if(!logged) { - error("DBENGINE: cannot store metric on unknown page type id %d", descr->type); + error("DBENGINE: cannot store metric on unknown page type id %d", ctx->config.page_type); logged = true; } } break; } - pg_cache_atomic_set_pg_info(descr, point_in_time_ut, descr->page_length + PAGE_POINT_SIZE_BYTES(descr)); + if(unlikely(!handle->page)){ + rrdeng_store_metric_create_new_page(handle, ctx, point_in_time_ut, data, data_size); + // handle->position is set to 1 already - if (perfect_page_alignment) - 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; - - new_metric_API_producers = rrd_atomic_add_fetch(&ctx->stats.metric_API_producers, 1); - while (unlikely(new_metric_API_producers > (old_metric_API_max_producers = ctx->metric_API_max_producers))) { - /* Increase ctx->metric_API_max_producers */ - ret_metric_API_max_producers = ulong_compare_and_swap(&ctx->metric_API_max_producers, - old_metric_API_max_producers, - new_metric_API_producers); - if (old_metric_API_max_producers == ret_metric_API_max_producers) { - /* success */ - break; - } + if (0 == handle->alignment->page_position) { + /* this is the leading dimension that defines chart alignment */ + perfect_page_alignment = true; + } + } + else { + // update an existing page + pgc_page_hot_set_end_time_s(main_cache, handle->page, (time_t) (point_in_time_ut / USEC_PER_SEC)); + handle->page_end_time_ut = point_in_time_ut; + + if(unlikely(++handle->page_position >= handle->page_entries_max)) { + internal_fatal(handle->page_position > handle->page_entries_max, "DBENGINE: exceeded page max number of points"); + handle->page_flags |= RRDENG_PAGE_FULL; + rrdeng_store_metric_flush_current_page(collection_handle); } + } + + if (perfect_page_alignment) + handle->alignment->page_position = handle->page_position; + + // update the metric information + mrg_metric_set_hot_latest_time_s(main_mrg, handle->metric, (time_t) (point_in_time_ut / USEC_PER_SEC)); +} + +static void store_metric_next_error_log(struct rrdeng_collect_handle *handle, usec_t point_in_time_ut, const char *msg) { + time_t point_in_time_s = (time_t)(point_in_time_ut / USEC_PER_SEC); + char uuid[UUID_STR_LEN + 1]; + uuid_unparse(*mrg_metric_uuid(main_mrg, handle->metric), uuid); - pg_cache_insert(ctx, page_index, descr); - } else { - pg_cache_add_new_metric_time(page_index, descr); + BUFFER *wb = NULL; + if(handle->page && handle->page_flags) { + wb = buffer_create(0, NULL); + collect_page_flags_to_buffer(wb, handle->page_flags); } -// { -// unsigned char u[16] = { 0x0C, 0x0A, 0x40, 0xD6, 0x2A, 0x43, 0x4A, 0x7C, 0x95, 0xF7, 0xD1, 0x1E, 0x0C, 0x9E, 0x8A, 0xE7 }; -// if(uuid_compare(u, page_index->id) == 0) { -// char buffer[100]; -// snprintfz(buffer, 100, "store system.cpu, collect:%u, page_index first:%u, last:%u", -// (uint32_t)(point_in_time / USEC_PER_SEC), -// (uint32_t)(page_index->oldest_time / USEC_PER_SEC), -// (uint32_t)(page_index->latest_time / USEC_PER_SEC)); -// -// print_page_cache_descr(descr, buffer, false); -// } -// } + error_limit_static_global_var(erl, 1, 0); + error_limit(&erl, + "DBENGINE: metric '%s' collected point at %ld, %s last collection at %ld, " + "update every %ld, %s page from %ld to %ld, position %u (of %u), flags: %s", + uuid, + point_in_time_s, + msg, + (time_t)(handle->page_end_time_ut / USEC_PER_SEC), + (time_t)(handle->update_every_ut / USEC_PER_SEC), + handle->page ? "current" : "*LAST*", + (time_t)(handle->page_start_time_ut / USEC_PER_SEC), + (time_t)(handle->page_end_time_ut / USEC_PER_SEC), + handle->page_position, handle->page_entries_max, + wb ? buffer_tostring(wb) : "" + ); + + buffer_free(wb); } void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, - usec_t point_in_time_ut, - NETDATA_DOUBLE n, - NETDATA_DOUBLE min_value, - NETDATA_DOUBLE max_value, - uint16_t count, - uint16_t anomaly_count, - SN_FLAGS flags) + const usec_t point_in_time_ut, + const NETDATA_DOUBLE n, + const NETDATA_DOUBLE min_value, + const NETDATA_DOUBLE max_value, + const uint16_t count, + const uint16_t anomaly_count, + const SN_FLAGS flags) { struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; - struct pg_cache_page_index *page_index = handle->page_index; - struct rrdeng_page_descr *descr = handle->descr; - - if(likely(descr)) { - usec_t last_point_in_time_ut = descr->end_time_ut; - usec_t update_every_ut = page_index->latest_update_every_s * USEC_PER_SEC; - size_t points_gap = (point_in_time_ut <= last_point_in_time_ut) ? - (size_t)0 : - (size_t)((point_in_time_ut - last_point_in_time_ut) / update_every_ut); - - if(unlikely(points_gap != 1)) { - if (unlikely(points_gap <= 0)) { - time_t now = now_realtime_sec(); - static __thread size_t counter = 0; - static __thread time_t last_time_logged = 0; - counter++; - - if(now - last_time_logged > 600) { - error("DBENGINE: collected point is in the past (repeated %zu times in the last %zu secs). Ignoring these data collection points.", - counter, (size_t)(last_time_logged?(now - last_time_logged):0)); - - last_time_logged = now; - counter = 0; - } - return; - } - size_t point_size = PAGE_POINT_SIZE_BYTES(descr); - size_t page_size_in_points = RRDENG_BLOCK_SIZE / point_size; - size_t used_points = descr->page_length / point_size; - size_t remaining_points_in_page = page_size_in_points - used_points; +#ifdef NETDATA_INTERNAL_CHECKS + if(unlikely(point_in_time_ut > (usec_t)max_acceptable_collected_time() * USEC_PER_SEC)) + handle->page_flags |= RRDENG_PAGE_FUTURE_POINT; +#endif + + if(likely(handle->page_end_time_ut + handle->update_every_ut == point_in_time_ut)) { + // happy path + ; + } + else if(unlikely(point_in_time_ut < handle->page_end_time_ut)) { + handle->page_flags |= RRDENG_PAGE_PAST_COLLECTION; + store_metric_next_error_log(handle, point_in_time_ut, "is older than the"); + return; + } - bool new_point_is_aligned = true; - if(unlikely((point_in_time_ut - last_point_in_time_ut) / points_gap != update_every_ut)) - new_point_is_aligned = false; + else if(unlikely(point_in_time_ut == handle->page_end_time_ut)) { + handle->page_flags |= RRDENG_PAGE_REPEATED_COLLECTION; + store_metric_next_error_log(handle, point_in_time_ut, "is at the same time as the"); + return; + } - if(unlikely(points_gap > remaining_points_in_page || !new_point_is_aligned)) { -// char buffer[200]; -// snprintfz(buffer, 200, "data collection skipped %zu points, last stored point %llu, new point %llu, update every %d. Cutting page.", -// points_gap, last_point_in_time_ut / USEC_PER_SEC, point_in_time_ut / USEC_PER_SEC, page_index->latest_update_every_s); -// print_page_cache_descr(descr, buffer, false); + else if(handle->page) { + usec_t delta_ut = point_in_time_ut - handle->page_end_time_ut; + if(unlikely(delta_ut < handle->update_every_ut)) { + handle->page_flags |= RRDENG_PAGE_STEP_TOO_SMALL; + rrdeng_store_metric_flush_current_page(collection_handle); + } + else if(unlikely(delta_ut % handle->update_every_ut)) { + handle->page_flags |= RRDENG_PAGE_STEP_UNALIGNED; + rrdeng_store_metric_flush_current_page(collection_handle); + } + else { + size_t points_gap = delta_ut / handle->update_every_ut; + size_t page_remaining_points = handle->page_entries_max - handle->page_position; + + if(points_gap >= page_remaining_points) { + handle->page_flags |= RRDENG_PAGE_BIG_GAP; rrdeng_store_metric_flush_current_page(collection_handle); } else { -// char buffer[200]; -// snprintfz(buffer, 200, "data collection skipped %zu points, last stored point %llu, new point %llu, update every %d. Filling the gap.", -// points_gap, last_point_in_time_ut / USEC_PER_SEC, point_in_time_ut / USEC_PER_SEC, page_index->latest_update_every_s); -// print_page_cache_descr(descr, buffer, false); - // loop to fill the gap - usec_t step_ut = page_index->latest_update_every_s * USEC_PER_SEC; - usec_t last_point_filled_ut = last_point_in_time_ut + step_ut; - - while (last_point_filled_ut < point_in_time_ut) { - rrdeng_store_metric_next_internal( - collection_handle, last_point_filled_ut, NAN, NAN, NAN, - 1, 0, SN_EMPTY_SLOT); - - last_point_filled_ut += step_ut; + handle->page_flags |= RRDENG_PAGE_GAP; + + usec_t stop_ut = point_in_time_ut - handle->update_every_ut; + for(usec_t this_ut = handle->page_end_time_ut + handle->update_every_ut; + this_ut <= stop_ut ; + this_ut = handle->page_end_time_ut + handle->update_every_ut) { + rrdeng_store_metric_append_point( + collection_handle, + this_ut, + NAN, NAN, NAN, + 1, 0, + SN_EMPTY_SLOT); } } } } - rrdeng_store_metric_next_internal(collection_handle, point_in_time_ut, n, min_value, max_value, count, anomaly_count, flags); + rrdeng_store_metric_append_point(collection_handle, + point_in_time_ut, + n, min_value, max_value, + count, anomaly_count, + flags); } - /* * Releases the database reference from the handle for storing metrics. * Returns 1 if it's safe to delete the dimension. */ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) { struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; - struct pg_cache_page_index *page_index = handle->page_index; - - uint8_t can_delete_metric = 0; + struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); + handle->page_flags |= RRDENG_PAGE_COLLECT_FINALIZE; rrdeng_store_metric_flush_current_page(collection_handle); - uv_rwlock_wrlock(&page_index->lock); + rrdeng_page_alignment_release(handle->alignment); - if (!--page_index->writers && !page_index->page_count) - can_delete_metric = 1; + __atomic_sub_fetch(&ctx->atomic.collectors_running, 1, __ATOMIC_RELAXED); + if(!(handle->options & RRDENG_1ST_METRIC_WRITER)) + __atomic_sub_fetch(&ctx->atomic.collectors_running_duplicate, 1, __ATOMIC_RELAXED); - uv_rwlock_wrunlock(&page_index->lock); + if((handle->options & RRDENG_1ST_METRIC_WRITER) && !mrg_metric_clear_writer(main_mrg, handle->metric)) + internal_fatal(true, "DBENGINE: metric is already released"); - rrdeng_page_alignment_release(handle->alignment); + time_t first_time_s, last_time_s, update_every_s; + mrg_metric_get_retention(main_mrg, handle->metric, &first_time_s, &last_time_s, &update_every_s); + + mrg_metric_release(main_mrg, handle->metric); freez(handle); - return can_delete_metric; + if(!first_time_s && !last_time_s) + return 1; + + return 0; } void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *collection_handle, int update_every) { struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; - struct pg_cache_page_index *page_index = handle->page_index; + check_and_fix_mrg_update_every(handle); + + METRIC *metric = handle->metric; + usec_t update_every_ut = (usec_t)update_every * USEC_PER_SEC; + + if(update_every_ut == handle->update_every_ut) + return; + + handle->page_flags |= RRDENG_PAGE_UPDATE_EVERY_CHANGE; rrdeng_store_metric_flush_current_page(collection_handle); - uv_rwlock_rdlock(&page_index->lock); - page_index->latest_update_every_s = update_every; - uv_rwlock_rdunlock(&page_index->lock); + mrg_metric_set_update_every(main_mrg, metric, update_every); + handle->update_every_ut = update_every_ut; } // ---------------------------------------------------------------------------- // query ops -//static inline uint32_t *pginfo_to_dt(struct rrdeng_page_info *page_info) -//{ -// return (uint32_t *)&page_info->scratch[0]; -//} -// -//static inline uint32_t *pginfo_to_points(struct rrdeng_page_info *page_info) -//{ -// return (uint32_t *)&page_info->scratch[sizeof(uint32_t)]; -//} -// +#ifdef NETDATA_INTERNAL_CHECKS +SPINLOCK global_query_handle_spinlock = NETDATA_SPINLOCK_INITIALIZER; +static struct rrdeng_query_handle *global_query_handle_ll = NULL; +static void register_query_handle(struct rrdeng_query_handle *handle) { + handle->query_pid = gettid(); + handle->started_time_s = now_realtime_sec(); + + netdata_spinlock_lock(&global_query_handle_spinlock); + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(global_query_handle_ll, handle, prev, next); + netdata_spinlock_unlock(&global_query_handle_spinlock); +} +static void unregister_query_handle(struct rrdeng_query_handle *handle) { + netdata_spinlock_lock(&global_query_handle_spinlock); + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(global_query_handle_ll, handle, prev, next); + netdata_spinlock_unlock(&global_query_handle_spinlock); +} +#else +static void register_query_handle(struct rrdeng_query_handle *handle __maybe_unused) { + ; +} +static void unregister_query_handle(struct rrdeng_query_handle *handle __maybe_unused) { + ; +} +#endif + /* * Gets a handle for loading metrics from the database. * The handle must be released with rrdeng_load_metric_final(). */ -void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct storage_engine_query_handle *rrdimm_handle, time_t start_time_s, time_t end_time_s) +void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, + struct storage_engine_query_handle *rrddim_handle, + time_t start_time_s, + time_t end_time_s, + STORAGE_PRIORITY priority) { - struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle; - struct rrdengine_instance *ctx = page_index->ctx; + usec_t started_ut = now_monotonic_usec(); - // fprintf(stderr, "%s: %s/%s start time %ld, end time %ld\n", __FUNCTION__ , rd->rrdset->name, rd->name, start_time, end_time); + netdata_thread_disable_cancelability(); + METRIC *metric = (METRIC *)db_metric_handle; + struct rrdengine_instance *ctx = mrg_metric_ctx(metric); struct rrdeng_query_handle *handle; - unsigned pages_nr; - if(!page_index->latest_update_every_s) - page_index->latest_update_every_s = default_rrd_update_every; + handle = rrdeng_query_handle_get(); + register_query_handle(handle); - rrdimm_handle->start_time_s = start_time_s; - rrdimm_handle->end_time_s = end_time_s; + if(unlikely(priority < STORAGE_PRIORITY_HIGH)) + priority = STORAGE_PRIORITY_HIGH; + else if(unlikely(priority > STORAGE_PRIORITY_BEST_EFFORT)) + priority = STORAGE_PRIORITY_BEST_EFFORT; - handle = callocz(1, sizeof(struct rrdeng_query_handle)); - handle->wanted_start_time_s = start_time_s; - handle->now_s = start_time_s; - handle->position = 0; handle->ctx = ctx; - handle->descr = NULL; - handle->dt_s = page_index->latest_update_every_s; - rrdimm_handle->handle = (STORAGE_QUERY_HANDLE *)handle; - pages_nr = pg_cache_preload(ctx, &page_index->id, start_time_s * USEC_PER_SEC, end_time_s * USEC_PER_SEC, - NULL, &handle->page_index); - if (unlikely(NULL == handle->page_index || 0 == pages_nr)) - // there are no metrics to load - handle->wanted_start_time_s = INVALID_TIME; -} + handle->metric = metric; + handle->priority = priority; + + // IMPORTANT! + // It is crucial not to exceed the db boundaries, because dbengine + // now has gap caching, so when a gap is detected a negative page + // is inserted into the main cache, to avoid scanning the journals + // again for pages matching the gap. + + time_t db_first_time_s, db_last_time_s, db_update_every_s; + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, &db_update_every_s); + + if(is_page_in_time_range(start_time_s, end_time_s, db_first_time_s, db_last_time_s) == PAGE_IS_IN_RANGE) { + handle->start_time_s = MAX(start_time_s, db_first_time_s); + handle->end_time_s = MIN(end_time_s, db_last_time_s); + handle->now_s = handle->start_time_s; + + handle->dt_s = db_update_every_s; + if (!handle->dt_s) { + handle->dt_s = default_rrd_update_every; + mrg_metric_set_update_every_s_if_zero(main_mrg, metric, default_rrd_update_every); + } -static int rrdeng_load_page_next(struct storage_engine_query_handle *rrdimm_handle, bool debug_this __maybe_unused) { - struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle; + rrddim_handle->handle = (STORAGE_QUERY_HANDLE *) handle; + rrddim_handle->start_time_s = handle->start_time_s; + rrddim_handle->end_time_s = handle->end_time_s; + rrddim_handle->priority = priority; - struct rrdengine_instance *ctx = handle->ctx; - struct rrdeng_page_descr *descr = handle->descr; + pg_cache_preload(handle); - uint32_t page_length; - usec_t page_end_time_ut; - unsigned position; + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.query_time_init, now_monotonic_usec() - started_ut, __ATOMIC_RELAXED); + } + else { + handle->start_time_s = start_time_s; + handle->end_time_s = end_time_s; + handle->now_s = start_time_s; + handle->dt_s = db_update_every_s; + + rrddim_handle->handle = (STORAGE_QUERY_HANDLE *) handle; + rrddim_handle->start_time_s = handle->start_time_s; + rrddim_handle->end_time_s = 0; + rrddim_handle->priority = priority; + } +} - if (likely(descr)) { - // Drop old page's reference +static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_handle, bool debug_this __maybe_unused) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; + struct rrdengine_instance *ctx = handle->ctx; -#ifdef NETDATA_INTERNAL_CHECKS - rrd_stat_atomic_add(&ctx->stats.metric_API_consumers, -1); -#endif + if (likely(handle->page)) { + // we have a page to release + pgc_page_release(main_cache, handle->page); + handle->page = NULL; + } - pg_cache_put(ctx, descr); - handle->descr = NULL; - handle->wanted_start_time_s = (time_t)((handle->page_end_time_ut / USEC_PER_SEC) + handle->dt_s); + if (unlikely(handle->now_s > rrddim_handle->end_time_s)) + return false; - if (unlikely(handle->wanted_start_time_s > rrdimm_handle->end_time_s)) - return 1; - } + size_t entries; + handle->page = pg_cache_lookup_next(ctx, handle->pdc, handle->now_s, handle->dt_s, &entries); + if (unlikely(!handle->page)) + return false; - usec_t wanted_start_time_ut = handle->wanted_start_time_s * USEC_PER_SEC; - descr = pg_cache_lookup_next(ctx, handle->page_index, &handle->page_index->id, - wanted_start_time_ut, rrdimm_handle->end_time_s * USEC_PER_SEC); - if (NULL == descr) - return 1; + internal_fatal(pgc_page_data(handle->page) == DBENGINE_EMPTY_PAGE, "Empty page returned"); -#ifdef NETDATA_INTERNAL_CHECKS - rrd_stat_atomic_add(&ctx->stats.metric_API_consumers, 1); -#endif + time_t page_start_time_s = pgc_page_start_time_s(handle->page); + time_t page_end_time_s = pgc_page_end_time_s(handle->page); + time_t page_update_every_s = pgc_page_update_every_s(handle->page); - handle->descr = descr; - pg_cache_atomic_get_pg_info(descr, &page_end_time_ut, &page_length); - if (unlikely(INVALID_TIME == descr->start_time_ut || INVALID_TIME == page_end_time_ut || 0 == descr->update_every_s)) { - error("DBENGINE: discarding invalid page descriptor (start_time = %llu, end_time = %llu, update_every_s = %d)", - descr->start_time_ut, page_end_time_ut, descr->update_every_s); - return 1; - } + unsigned position; + if(likely(handle->now_s >= page_start_time_s && handle->now_s <= page_end_time_s)) { + + if(unlikely(entries == 1 || page_start_time_s == page_end_time_s || !page_update_every_s)) { + position = 0; + handle->now_s = page_start_time_s; + } + else { + position = (handle->now_s - page_start_time_s) * (entries - 1) / (page_end_time_s - page_start_time_s); + time_t point_end_time_s = page_start_time_s + position * page_update_every_s; + while(point_end_time_s < handle->now_s && position + 1 < entries) { + // https://github.com/netdata/netdata/issues/14411 + // we really need a while() here, because the delta may be + // 2 points at higher tiers + position++; + point_end_time_s = page_start_time_s + position * page_update_every_s; + } + handle->now_s = point_end_time_s; + } - if (unlikely(descr->start_time_ut != page_end_time_ut && wanted_start_time_ut > descr->start_time_ut)) { - // we're in the middle of the page somewhere - unsigned entries = page_length / PAGE_POINT_SIZE_BYTES(descr); - position = ((uint64_t)(wanted_start_time_ut - descr->start_time_ut)) * (entries - 1) / - (page_end_time_ut - descr->start_time_ut); + internal_fatal(position >= entries, "DBENGINE: wrong page position calculation"); } - else + else if(handle->now_s < page_start_time_s) { + handle->now_s = page_start_time_s; position = 0; + } + else { + internal_fatal(true, "DBENGINE: this page is entirely in our past and should not be accepted for this query in the first place"); + handle->now_s = page_end_time_s; + position = entries - 1; + } - handle->page_end_time_ut = page_end_time_ut; - handle->page_length = page_length; - handle->entries = page_length / PAGE_POINT_SIZE_BYTES(descr); - handle->page = descr->pg_cache_descr->page; - handle->dt_s = descr->update_every_s; + handle->entries = entries; handle->position = position; - -// if(debug_this) -// info("DBENGINE: rrdeng_load_page_next(), " -// "position:%d, " -// "start_time_ut:%llu, " -// "page_end_time_ut:%llu, " -// "next_page_time_ut:%llu, " -// "in_out:%s" -// , position -// , descr->start_time_ut -// , page_end_time_ut -// , -// wanted_start_time_ut, in_out?"true":"false" -// ); - - return 0; + handle->metric_data = pgc_page_data((PGC_PAGE *)handle->page); + handle->dt_s = page_update_every_s; + return true; } // Returns the metric and sets its timestamp into current_time @@ -675,75 +895,28 @@ static int rrdeng_load_page_next(struct storage_engine_query_handle *rrdimm_hand // IT IS REQUIRED TO **ALWAYS** KEEP TRACK OF TIME, EVEN OUTSIDE THE DATABASE BOUNDARIES STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim_handle) { struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; - // struct rrdeng_metric_handle *metric_handle = handle->metric_handle; - - struct rrdeng_page_descr *descr = handle->descr; - time_t now = handle->now_s + handle->dt_s; - -// bool debug_this = false; -// { -// unsigned char u[16] = { 0x0C, 0x0A, 0x40, 0xD6, 0x2A, 0x43, 0x4A, 0x7C, 0x95, 0xF7, 0xD1, 0x1E, 0x0C, 0x9E, 0x8A, 0xE7 }; -// if(uuid_compare(u, handle->page_index->id) == 0) { -// char buffer[100]; -// snprintfz(buffer, 100, "load system.cpu, now:%u, dt:%u, position:%u page_index first:%u, last:%u", -// (uint32_t)(now), -// (uint32_t)(handle->dt_s), -// (uint32_t)(handle->position), -// (uint32_t)(handle->page_index->oldest_time / USEC_PER_SEC), -// (uint32_t)(handle->page_index->latest_time / USEC_PER_SEC)); -// -// print_page_cache_descr(descr, buffer, false); -// debug_this = true; -// } -// } - STORAGE_POINT sp; - unsigned position = handle->position + 1; - storage_number_tier1_t tier1_value; - - if (unlikely(INVALID_TIME == handle->wanted_start_time_s)) { - handle->wanted_start_time_s = INVALID_TIME; - handle->now_s = now; - storage_point_empty(sp, now - handle->dt_s, now); - return sp; + + if (unlikely(handle->now_s > rrddim_handle->end_time_s)) { + storage_point_empty(sp, handle->now_s - handle->dt_s, handle->now_s); + goto prepare_for_next_iteration; } - if (unlikely(!descr || position >= handle->entries)) { + if (unlikely(!handle->page || handle->position >= handle->entries)) { // We need to get a new page - if(rrdeng_load_page_next(rrddim_handle, false)) { - // next calls will not load any more metrics - handle->wanted_start_time_s = INVALID_TIME; - handle->now_s = now; - storage_point_empty(sp, now - handle->dt_s, now); - return sp; - } - descr = handle->descr; - position = handle->position; - now = (time_t)((descr->start_time_ut / USEC_PER_SEC) + position * descr->update_every_s); - -// if(debug_this) { -// char buffer[100]; -// snprintfz(buffer, 100, "NEW PAGE system.cpu, now:%u, dt:%u, position:%u page_index first:%u, last:%u", -// (uint32_t)(now), -// (uint32_t)(handle->dt_s), -// (uint32_t)(handle->position), -// (uint32_t)(handle->page_index->oldest_time / USEC_PER_SEC), -// (uint32_t)(handle->page_index->latest_time / USEC_PER_SEC)); -// -// print_page_cache_descr(descr, buffer, false); -// } + if (!rrdeng_load_page_next(rrddim_handle, false)) { + storage_point_empty(sp, handle->now_s - handle->dt_s, handle->now_s); + goto prepare_for_next_iteration; + } } - sp.start_time = now - handle->dt_s; - sp.end_time = now; - - handle->position = position; - handle->now_s = now; + sp.start_time_s = handle->now_s - handle->dt_s; + sp.end_time_s = handle->now_s; - switch(descr->type) { + switch(handle->ctx->config.page_type) { case PAGE_METRICS: { - storage_number n = handle->page[position]; + storage_number n = handle->metric_data[handle->position]; sp.min = sp.max = sp.sum = unpack_storage_number(n); sp.flags = n & SN_USER_FLAGS; sp.count = 1; @@ -752,7 +925,7 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim break; case PAGE_TIER: { - tier1_value = ((storage_number_tier1_t *)handle->page)[position]; + storage_number_tier1_t tier1_value = ((storage_number_tier1_t *)handle->metric_data)[handle->position]; sp.flags = tier1_value.anomaly_count ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS; sp.count = tier1_value.count; sp.anomaly_count = tier1_value.anomaly_count; @@ -766,204 +939,98 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim default: { static bool logged = false; if(!logged) { - error("DBENGINE: unknown page type %d found. Cannot decode it. Ignoring its metrics.", descr->type); + error("DBENGINE: unknown page type %d found. Cannot decode it. Ignoring its metrics.", handle->ctx->config.page_type); logged = true; } - storage_point_empty(sp, sp.start_time, sp.end_time); + storage_point_empty(sp, sp.start_time_s, sp.end_time_s); } break; } - if (unlikely(now >= rrddim_handle->end_time_s)) { - // next calls will not load any more metrics - handle->wanted_start_time_s = INVALID_TIME; - } +prepare_for_next_iteration: + internal_fatal(sp.end_time_s < rrddim_handle->start_time_s, "DBENGINE: this point is too old for this query"); + internal_fatal(sp.end_time_s < handle->now_s, "DBENGINE: this point is too old for this point in time"); -// if(debug_this) -// info("DBENGINE: returning point: " -// "time from %ld to %ld // query from %ld to %ld // wanted_start_time_s %ld" -// , sp.start_time, sp.end_time -// , rrddim_handle->start_time_s, rrddim_handle->end_time_s -// , handle->wanted_start_time_s -// ); + handle->now_s += handle->dt_s; + handle->position++; return sp; } -int rrdeng_load_metric_is_finished(struct storage_engine_query_handle *rrdimm_handle) -{ - struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle; - return (INVALID_TIME == handle->wanted_start_time_s); +int rrdeng_load_metric_is_finished(struct storage_engine_query_handle *rrddim_handle) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; + return (handle->now_s > rrddim_handle->end_time_s); } /* * Releases the database reference from the handle for loading metrics. */ -void rrdeng_load_metric_finalize(struct storage_engine_query_handle *rrdimm_handle) +void rrdeng_load_metric_finalize(struct storage_engine_query_handle *rrddim_handle) { - struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle; - struct rrdengine_instance *ctx = handle->ctx; - struct rrdeng_page_descr *descr = handle->descr; + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; - if (descr) { -#ifdef NETDATA_INTERNAL_CHECKS - rrd_stat_atomic_add(&ctx->stats.metric_API_consumers, -1); -#endif - pg_cache_put(ctx, descr); - } + if (handle->page) + pgc_page_release(main_cache, handle->page); - // whatever is allocated at rrdeng_load_metric_init() should be freed here - freez(handle); - rrdimm_handle->handle = NULL; -} + if(!pdc_release_and_destroy_if_unreferenced(handle->pdc, false, false)) + __atomic_store_n(&handle->pdc->workers_should_stop, true, __ATOMIC_RELAXED); -time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { - struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle; - return (time_t)(page_index->latest_time_ut / USEC_PER_SEC); -} -time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { - struct pg_cache_page_index *page_index = (struct pg_cache_page_index *)db_metric_handle; - return (time_t)(page_index->oldest_time_ut / USEC_PER_SEC); + unregister_query_handle(handle); + rrdeng_query_handle_release(handle); + rrddim_handle->handle = NULL; + netdata_thread_enable_cancelability(); } -int rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, uuid_t *dim_uuid, time_t *first_entry_t, time_t *last_entry_t) -{ - struct page_cache *pg_cache; - struct rrdengine_instance *ctx; - Pvoid_t *PValue; - struct pg_cache_page_index *page_index = NULL; +time_t rrdeng_load_align_to_optimal_before(struct storage_engine_query_handle *rrddim_handle) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; - ctx = (struct rrdengine_instance *)si; - if (unlikely(!ctx)) { - error("DBENGINE: invalid STORAGE INSTANCE to %s()", __FUNCTION__); - return 1; + if(handle->pdc) { + rrdeng_prep_wait(handle->pdc); + if (handle->pdc->optimal_end_time_s > rrddim_handle->end_time_s) + rrddim_handle->end_time_s = handle->pdc->optimal_end_time_s; } - pg_cache = &ctx->pg_cache; - uv_rwlock_rdlock(&pg_cache->metrics_index.lock); - PValue = JudyHSGet(pg_cache->metrics_index.JudyHS_array, dim_uuid, sizeof(uuid_t)); - if (likely(NULL != PValue)) { - page_index = *PValue; - } - uv_rwlock_rdunlock(&pg_cache->metrics_index.lock); + return rrddim_handle->end_time_s; +} - if (likely(page_index)) { - *first_entry_t = page_index->oldest_time_ut / USEC_PER_SEC; - *last_entry_t = page_index->latest_time_ut / USEC_PER_SEC; - return 0; - } +time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { + METRIC *metric = (METRIC *)db_metric_handle; + time_t latest_time_s = 0; - return 1; -} + if (metric) + latest_time_s = mrg_metric_get_latest_time_s(main_mrg, metric); -/* Also gets a reference for the page */ -void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrdeng_page_descr **ret_descr) -{ - struct rrdeng_page_descr *descr; - struct page_cache_descr *pg_cache_descr; - void *page; - /* TODO: check maximum number of pages in page cache limit */ - - descr = pg_cache_create_descr(); - descr->id = id; /* TODO: add page type: metric, log, something? */ - descr->type = ctx->page_type; - page = dbengine_page_alloc(); /*TODO: add page size */ - rrdeng_page_descr_mutex_lock(ctx, descr); - pg_cache_descr = descr->pg_cache_descr; - pg_cache_descr->page = page; - pg_cache_descr->flags = RRD_PAGE_DIRTY /*| RRD_PAGE_LOCKED */ | RRD_PAGE_POPULATED /* | BEING_COLLECTED */; - pg_cache_descr->refcnt = 1; - - debug(D_RRDENGINE, "Created new page:"); - if (unlikely(debug_flags & D_RRDENGINE)) - print_page_cache_descr(descr, "", true); - rrdeng_page_descr_mutex_unlock(ctx, descr); - *ret_descr = descr; - return page; + return latest_time_s; } -/* The page must not be empty */ -void rrdeng_commit_page(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr, - Word_t page_correlation_id) -{ - struct page_cache *pg_cache = &ctx->pg_cache; - Pvoid_t *PValue; - unsigned nr_committed_pages; +time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { + METRIC *metric = (METRIC *)db_metric_handle; - if (unlikely(NULL == descr)) { - debug(D_RRDENGINE, "%s: page descriptor is NULL, page has already been force-committed.", __func__); - return; - } - fatal_assert(descr->page_length); - - uv_rwlock_wrlock(&pg_cache->committed_page_index.lock); - PValue = JudyLIns(&pg_cache->committed_page_index.JudyL_array, page_correlation_id, PJE0); - *PValue = descr; - nr_committed_pages = ++pg_cache->committed_page_index.nr_committed_pages; - uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock); - - if (nr_committed_pages >= pg_cache_hard_limit(ctx) / 2) { - /* over 50% of pages have not been committed yet */ - - if (ctx->drop_metrics_under_page_cache_pressure && - nr_committed_pages >= pg_cache_committed_hard_limit(ctx)) { - /* 100% of pages are dirty */ - struct rrdeng_cmd cmd; - - cmd.opcode = RRDENG_INVALIDATE_OLDEST_MEMORY_PAGE; - rrdeng_enq_cmd(&ctx->worker_config, &cmd); - } else { - if (0 == (unsigned long) ctx->stats.pg_cache_over_half_dirty_events) { - /* only print the first time */ - errno = 0; - error("Failed to flush dirty buffers quickly enough in dbengine instance \"%s\". " - "Metric data at risk of not being stored in the database, " - "please reduce disk load or use a faster disk.", ctx->dbfiles_path); - } - rrd_stat_atomic_add(&ctx->stats.pg_cache_over_half_dirty_events, 1); - rrd_stat_atomic_add(&global_pg_cache_over_half_dirty_events, 1); - } - } + time_t oldest_time_s = 0; + if (metric) + oldest_time_s = mrg_metric_get_first_time_s(main_mrg, metric); - pg_cache_put(ctx, descr); + return oldest_time_s; } -/* Gets a reference for the page */ -void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **handle) +bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *db_instance, uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s) { - struct rrdeng_page_descr *descr; - struct page_cache_descr *pg_cache_descr; - - debug(D_RRDENGINE, "Reading existing page:"); - descr = pg_cache_lookup(ctx, NULL, id, INVALID_TIME); - if (NULL == descr) { - *handle = NULL; - - return NULL; + struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; + if (unlikely(!ctx)) { + error("DBENGINE: invalid STORAGE INSTANCE to %s()", __FUNCTION__); + return false; } - *handle = descr; - pg_cache_descr = descr->pg_cache_descr; - return pg_cache_descr->page; -} + METRIC *metric = mrg_metric_get_and_acquire(main_mrg, dim_uuid, (Word_t) ctx); + if (unlikely(!metric)) + return false; -/* Gets a reference for the page */ -void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time_ut, void **handle) -{ - struct rrdeng_page_descr *descr; - struct page_cache_descr *pg_cache_descr; + time_t update_every_s; + mrg_metric_get_retention(main_mrg, metric, first_entry_s, last_entry_s, &update_every_s); - debug(D_RRDENGINE, "Reading existing page:"); - descr = pg_cache_lookup(ctx, NULL, id, point_in_time_ut); - if (NULL == descr) { - *handle = NULL; - - return NULL; - } - *handle = descr; - pg_cache_descr = descr->pg_cache_descr; + mrg_metric_release(main_mrg, metric); - return pg_cache_descr->page; + return true; } /* @@ -977,62 +1044,126 @@ void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long if (ctx == NULL) return; - struct page_cache *pg_cache = &ctx->pg_cache; - - array[0] = (uint64_t)ctx->stats.metric_API_producers; - array[1] = (uint64_t)ctx->stats.metric_API_consumers; - array[2] = (uint64_t)pg_cache->page_descriptors; - array[3] = (uint64_t)pg_cache->populated_pages; - array[4] = (uint64_t)pg_cache->committed_page_index.nr_committed_pages; - array[5] = (uint64_t)ctx->stats.pg_cache_insertions; - array[6] = (uint64_t)ctx->stats.pg_cache_deletions; - array[7] = (uint64_t)ctx->stats.pg_cache_hits; - array[8] = (uint64_t)ctx->stats.pg_cache_misses; - array[9] = (uint64_t)ctx->stats.pg_cache_backfills; - array[10] = (uint64_t)ctx->stats.pg_cache_evictions; - array[11] = (uint64_t)ctx->stats.before_compress_bytes; - array[12] = (uint64_t)ctx->stats.after_compress_bytes; - array[13] = (uint64_t)ctx->stats.before_decompress_bytes; - array[14] = (uint64_t)ctx->stats.after_decompress_bytes; - array[15] = (uint64_t)ctx->stats.io_write_bytes; - array[16] = (uint64_t)ctx->stats.io_write_requests; - array[17] = (uint64_t)ctx->stats.io_read_bytes; - array[18] = (uint64_t)ctx->stats.io_read_requests; - array[19] = (uint64_t)ctx->stats.io_write_extent_bytes; - array[20] = (uint64_t)ctx->stats.io_write_extents; - array[21] = (uint64_t)ctx->stats.io_read_extent_bytes; - array[22] = (uint64_t)ctx->stats.io_read_extents; - array[23] = (uint64_t)ctx->stats.datafile_creations; - array[24] = (uint64_t)ctx->stats.datafile_deletions; - array[25] = (uint64_t)ctx->stats.journalfile_creations; - array[26] = (uint64_t)ctx->stats.journalfile_deletions; - array[27] = (uint64_t)ctx->stats.page_cache_descriptors; - array[28] = (uint64_t)ctx->stats.io_errors; - array[29] = (uint64_t)ctx->stats.fs_errors; - array[30] = (uint64_t)global_io_errors; - array[31] = (uint64_t)global_fs_errors; - array[32] = (uint64_t)rrdeng_reserved_file_descriptors; - array[33] = (uint64_t)ctx->stats.pg_cache_over_half_dirty_events; - array[34] = (uint64_t)global_pg_cache_over_half_dirty_events; - array[35] = (uint64_t)ctx->stats.flushing_pressure_page_deletions; - array[36] = (uint64_t)global_flushing_pressure_page_deletions; - fatal_assert(RRDENG_NR_STATS == 37); + array[0] = (uint64_t)__atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED); // API producers + array[1] = (uint64_t)__atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED); // API consumers + array[2] = 0; + array[3] = 0; + array[4] = 0; + array[5] = 0; // (uint64_t)ctx->stats.pg_cache_insertions; + array[6] = 0; // (uint64_t)ctx->stats.pg_cache_deletions; + array[7] = 0; // (uint64_t)ctx->stats.pg_cache_hits; + array[8] = 0; // (uint64_t)ctx->stats.pg_cache_misses; + array[9] = 0; // (uint64_t)ctx->stats.pg_cache_backfills; + array[10] = 0; // (uint64_t)ctx->stats.pg_cache_evictions; + array[11] = (uint64_t)__atomic_load_n(&ctx->stats.before_compress_bytes, __ATOMIC_RELAXED); // used + array[12] = (uint64_t)__atomic_load_n(&ctx->stats.after_compress_bytes, __ATOMIC_RELAXED); // used + array[13] = (uint64_t)__atomic_load_n(&ctx->stats.before_decompress_bytes, __ATOMIC_RELAXED); + array[14] = (uint64_t)__atomic_load_n(&ctx->stats.after_decompress_bytes, __ATOMIC_RELAXED); + array[15] = (uint64_t)__atomic_load_n(&ctx->stats.io_write_bytes, __ATOMIC_RELAXED); // used + array[16] = (uint64_t)__atomic_load_n(&ctx->stats.io_write_requests, __ATOMIC_RELAXED); // used + array[17] = (uint64_t)__atomic_load_n(&ctx->stats.io_read_bytes, __ATOMIC_RELAXED); + array[18] = (uint64_t)__atomic_load_n(&ctx->stats.io_read_requests, __ATOMIC_RELAXED); // used + array[19] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.io_write_extent_bytes, __ATOMIC_RELAXED); + array[20] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.io_write_extents, __ATOMIC_RELAXED); + array[21] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.io_read_extent_bytes, __ATOMIC_RELAXED); + array[22] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.io_read_extents, __ATOMIC_RELAXED); + array[23] = (uint64_t)__atomic_load_n(&ctx->stats.datafile_creations, __ATOMIC_RELAXED); + array[24] = (uint64_t)__atomic_load_n(&ctx->stats.datafile_deletions, __ATOMIC_RELAXED); + array[25] = (uint64_t)__atomic_load_n(&ctx->stats.journalfile_creations, __ATOMIC_RELAXED); + array[26] = (uint64_t)__atomic_load_n(&ctx->stats.journalfile_deletions, __ATOMIC_RELAXED); + array[27] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.page_cache_descriptors, __ATOMIC_RELAXED); + array[28] = (uint64_t)__atomic_load_n(&ctx->stats.io_errors, __ATOMIC_RELAXED); + array[29] = (uint64_t)__atomic_load_n(&ctx->stats.fs_errors, __ATOMIC_RELAXED); + array[30] = (uint64_t)__atomic_load_n(&global_io_errors, __ATOMIC_RELAXED); // used + array[31] = (uint64_t)__atomic_load_n(&global_fs_errors, __ATOMIC_RELAXED); // used + array[32] = (uint64_t)__atomic_load_n(&rrdeng_reserved_file_descriptors, __ATOMIC_RELAXED); // used + array[33] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.pg_cache_over_half_dirty_events, __ATOMIC_RELAXED); + array[34] = (uint64_t)__atomic_load_n(&global_pg_cache_over_half_dirty_events, __ATOMIC_RELAXED); // used + array[35] = 0; // (uint64_t)__atomic_load_n(&ctx->stats.flushing_pressure_page_deletions, __ATOMIC_RELAXED); + array[36] = (uint64_t)__atomic_load_n(&global_flushing_pressure_page_deletions, __ATOMIC_RELAXED); // used + array[37] = 0; //(uint64_t)pg_cache->active_descriptors; + + fatal_assert(RRDENG_NR_STATS == 38); } -/* Releases reference to page */ -void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle) -{ - (void)ctx; - pg_cache_put(ctx, (struct rrdeng_page_descr *)handle); +static void rrdeng_populate_mrg(struct rrdengine_instance *ctx) { + uv_rwlock_rdlock(&ctx->datafiles.rwlock); + size_t datafiles = 0; + for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next) + datafiles++; + uv_rwlock_rdunlock(&ctx->datafiles.rwlock); + + size_t cpus = get_netdata_cpus() / storage_tiers; + if(cpus > datafiles) + cpus = datafiles; + + if(cpus < 1) + cpus = 1; + + if(cpus > (size_t)libuv_worker_threads) + cpus = (size_t)libuv_worker_threads; + + if(cpus > MRG_PARTITIONS) + cpus = MRG_PARTITIONS; + + info("DBENGINE: populating retention to MRG from %zu journal files of tier %d, using %zu threads...", datafiles, ctx->config.tier, cpus); + + if(datafiles > 2) { + struct rrdengine_datafile *datafile; + + datafile = ctx->datafiles.first->prev; + if(!(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE)) + datafile = datafile->prev; + + if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) { + journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile); + datafile->populate_mrg.populated = true; + } + + datafile = ctx->datafiles.first; + if(datafile->journalfile->v2.flags & JOURNALFILE_FLAG_IS_AVAILABLE) { + journalfile_v2_populate_retention_to_mrg(ctx, datafile->journalfile); + datafile->populate_mrg.populated = true; + } + } + + ctx->loading.populate_mrg.size = cpus; + ctx->loading.populate_mrg.array = callocz(ctx->loading.populate_mrg.size, sizeof(struct completion)); + + for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) { + completion_init(&ctx->loading.populate_mrg.array[i]); + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_POPULATE_MRG, NULL, &ctx->loading.populate_mrg.array[i], + STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL); + } +} + +void rrdeng_readiness_wait(struct rrdengine_instance *ctx) { + for (size_t i = 0; i < ctx->loading.populate_mrg.size; i++) { + completion_wait_for(&ctx->loading.populate_mrg.array[i]); + completion_destroy(&ctx->loading.populate_mrg.array[i]); + } + + freez(ctx->loading.populate_mrg.array); + ctx->loading.populate_mrg.array = NULL; + ctx->loading.populate_mrg.size = 0; + + info("DBENGINE: tier %d is ready for data collection and queries", ctx->config.tier); +} + +bool rrdeng_is_legacy(STORAGE_INSTANCE *db_instance) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; + return ctx->config.legacy; } +void rrdeng_exit_mode(struct rrdengine_instance *ctx) { + __atomic_store_n(&ctx->quiesce.exit_mode, true, __ATOMIC_RELAXED); +} /* * Returns 0 on success, negative on error */ -int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb, +int rrdeng_init(struct rrdengine_instance **ctxp, const char *dbfiles_path, unsigned disk_space_mb, size_t tier) { struct rrdengine_instance *ctx; - int error; uint32_t max_open_files; max_open_files = rlimit_nofile.rlim_cur / 4; @@ -1053,182 +1184,185 @@ 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->config.legacy = false; } else { *ctxp = ctx = callocz(1, sizeof(*ctx)); + ctx->config.legacy = true; } - 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; - ctx->max_cache_pages = page_cache_mb * (1048576LU / RRDENG_BLOCK_SIZE); - /* try to keep 5% of the page cache free */ - ctx->cache_pages_low_watermark = (ctx->max_cache_pages * 95LLU) / 100; + + ctx->config.tier = (int)tier; + ctx->config.page_type = tier_page_type[tier]; + ctx->config.global_compress_alg = RRD_LZ4; if (disk_space_mb < RRDENG_MIN_DISK_SPACE_MB) disk_space_mb = RRDENG_MIN_DISK_SPACE_MB; - ctx->max_disk_space = disk_space_mb * 1048576LLU; - strncpyz(ctx->dbfiles_path, dbfiles_path, sizeof(ctx->dbfiles_path) - 1); - ctx->dbfiles_path[sizeof(ctx->dbfiles_path) - 1] = '\0'; - if (NULL == host) - strncpyz(ctx->machine_guid, registry_get_this_machine_guid(), GUID_LEN); - else - strncpyz(ctx->machine_guid, host->machine_guid, GUID_LEN); - - ctx->drop_metrics_under_page_cache_pressure = rrdeng_drop_metrics_under_page_cache_pressure; - ctx->metric_API_max_producers = 0; - ctx->quiesce = NO_QUIESCE; - ctx->host = host; - - memset(&ctx->worker_config, 0, sizeof(ctx->worker_config)); - ctx->worker_config.ctx = ctx; - init_page_cache(ctx); - init_commit_log(ctx); - error = init_rrd_files(ctx); - if (error) { - goto error_after_init_rrd_files; - } + ctx->config.max_disk_space = disk_space_mb * 1048576LLU; + strncpyz(ctx->config.dbfiles_path, dbfiles_path, sizeof(ctx->config.dbfiles_path) - 1); + ctx->config.dbfiles_path[sizeof(ctx->config.dbfiles_path) - 1] = '\0'; - completion_init(&ctx->rrdengine_completion); - fatal_assert(0 == uv_thread_create(&ctx->worker_config.thread, rrdeng_worker, &ctx->worker_config)); - /* wait for worker thread to initialize */ - completion_wait_for(&ctx->rrdengine_completion); - completion_destroy(&ctx->rrdengine_completion); - uv_thread_set_name_np(ctx->worker_config.thread, "LIBUV_WORKER"); - if (ctx->worker_config.error) { - goto error_after_rrdeng_worker; - } -// error = metalog_init(ctx); -// if (error) { -// error("Failed to initialize metadata log file event loop."); -// goto error_after_rrdeng_worker; -// } + ctx->atomic.transaction_id = 1; + ctx->quiesce.enabled = false; - return 0; + if (rrdeng_dbengine_spawn(ctx) && !init_rrd_files(ctx)) { + // success - we run this ctx too + rrdeng_populate_mrg(ctx); + return 0; + } -error_after_rrdeng_worker: - finalize_rrd_files(ctx); -error_after_init_rrd_files: - free_page_cache(ctx); - if (!is_storage_engine_shared((STORAGE_INSTANCE *)ctx)) { + if (ctx->config.legacy) { freez(ctx); if (ctxp) *ctxp = NULL; } + rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE); return UV_EIO; } +size_t rrdeng_collectors_running(struct rrdengine_instance *ctx) { + return __atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED); +} + /* * Returns 0 on success, 1 on error */ -int rrdeng_exit(struct rrdengine_instance *ctx) -{ - struct rrdeng_cmd cmd; - - if (NULL == ctx) { +int rrdeng_exit(struct rrdengine_instance *ctx) { + if (NULL == ctx) return 1; + + // FIXME - ktsaou - properly cleanup ctx + // 1. make sure all collectors are stopped + // 2. make new queries will not be accepted (this is quiesce that has already run) + // 3. flush this section of the main cache + // 4. then wait for completion + + bool logged = false; + while(__atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED) && !unittest_running) { + if(!logged) { + info("DBENGINE: waiting for collectors to finish on tier %d...", (ctx->config.legacy) ? -1 : ctx->config.tier); + logged = true; + } + sleep_usec(100 * USEC_PER_MS); } - /* TODO: add page to page cache */ - cmd.opcode = RRDENG_SHUTDOWN; - rrdeng_enq_cmd(&ctx->worker_config, &cmd); + info("DBENGINE: flushing main cache for tier %d", (ctx->config.legacy) ? -1 : ctx->config.tier); + pgc_flush_all_hot_and_dirty_pages(main_cache, (Word_t)ctx); - fatal_assert(0 == uv_thread_join(&ctx->worker_config.thread)); + info("DBENGINE: shutting down tier %d", (ctx->config.legacy) ? -1 : ctx->config.tier); + struct completion completion = {}; + completion_init(&completion); + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_SHUTDOWN, NULL, &completion, STORAGE_PRIORITY_BEST_EFFORT, NULL, NULL); + completion_wait_for(&completion); + completion_destroy(&completion); finalize_rrd_files(ctx); - //metalog_exit(ctx->metalog_ctx); - free_page_cache(ctx); - if(!is_storage_engine_shared((STORAGE_INSTANCE *)ctx)) + if(ctx->config.legacy) freez(ctx); rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE); return 0; } -void rrdeng_prepare_exit(struct rrdengine_instance *ctx) -{ - struct rrdeng_cmd cmd; - - if (NULL == ctx) { +void rrdeng_prepare_exit(struct rrdengine_instance *ctx) { + if (NULL == ctx) return; - } - - completion_init(&ctx->rrdengine_completion); - cmd.opcode = RRDENG_QUIESCE; - rrdeng_enq_cmd(&ctx->worker_config, &cmd); - /* wait for dbengine to quiesce */ - completion_wait_for(&ctx->rrdengine_completion); - completion_destroy(&ctx->rrdengine_completion); + // FIXME - ktsaou - properly cleanup ctx + // 1. make sure all collectors are stopped - //metalog_prepare_exit(ctx->metalog_ctx); + completion_init(&ctx->quiesce.completion); + rrdeng_enq_cmd(ctx, RRDENG_OPCODE_CTX_QUIESCE, NULL, NULL, STORAGE_PRIORITY_INTERNAL_DBENGINE, NULL, NULL); } -RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) { - RRDENG_SIZE_STATS stats = { 0 }; +static void populate_v2_statistics(struct rrdengine_datafile *datafile, RRDENG_SIZE_STATS *stats) +{ + struct journal_v2_header *j2_header = journalfile_v2_data_acquire(datafile->journalfile, NULL, 0, 0); + void *data_start = (void *)j2_header; - for(struct pg_cache_page_index *page_index = ctx->pg_cache.metrics_index.last_page_index; - page_index != NULL ;page_index = page_index->prev) { - stats.metrics++; - stats.metrics_pages += page_index->page_count; + if(unlikely(!j2_header)) + return; + + stats->extents += j2_header->extent_count; + + unsigned entries; + struct journal_extent_list *extent_list = (void *) (data_start + j2_header->extent_offset); + for (entries = 0; entries < j2_header->extent_count; entries++) { + stats->extents_compressed_bytes += extent_list->datafile_size; + stats->extents_pages += extent_list->pages; + extent_list++; } - for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next) { - stats.datafiles++; + struct journal_metric_list *metric = (void *) (data_start + j2_header->metric_offset); + time_t journal_start_time_s = (time_t) (j2_header->start_time_ut / USEC_PER_SEC); - for(struct extent_info *ei = df->extents.first; ei ; ei = ei->next) { - stats.extents++; - stats.extents_compressed_bytes += ei->size; + stats->metrics += j2_header->metric_count; + for (entries = 0; entries < j2_header->metric_count; entries++) { - for(int p = 0; p < ei->number_of_pages ;p++) { - struct rrdeng_page_descr *descr = ei->pages[p]; + struct journal_page_header *metric_list_header = (void *) (data_start + metric->page_offset); + stats->metrics_pages += metric_list_header->entries; + struct journal_page_list *descr = (void *) (data_start + metric->page_offset + sizeof(struct journal_page_header)); + for (uint32_t idx=0; idx < metric_list_header->entries; idx++) { - usec_t update_every_usec; + time_t update_every_s; - size_t points = descr->page_length / PAGE_POINT_SIZE_BYTES(descr); + size_t points = descr->page_length / CTX_POINT_SIZE_BYTES(datafile->ctx); - if(likely(points > 1)) - update_every_usec = (descr->end_time_ut - descr->start_time_ut) / (points - 1); - else { - update_every_usec = default_rrd_update_every * get_tier_grouping(ctx->tier) * USEC_PER_SEC; - stats.single_point_pages++; - } + time_t start_time_s = journal_start_time_s + descr->delta_start_s; + time_t end_time_s = journal_start_time_s + descr->delta_end_s; - time_t duration_secs = (time_t)((descr->end_time_ut - descr->start_time_ut + update_every_usec)/USEC_PER_SEC); + if(likely(points > 1)) + update_every_s = (time_t) ((end_time_s - start_time_s) / (points - 1)); + else { + update_every_s = (time_t) (default_rrd_update_every * get_tier_grouping(datafile->ctx->config.tier)); + stats->single_point_pages++; + } - stats.extents_pages++; - stats.pages_uncompressed_bytes += descr->page_length; - stats.pages_duration_secs += duration_secs; - stats.points += points; + time_t duration_s = (time_t)((end_time_s - start_time_s + update_every_s)); - stats.page_types[descr->type].pages++; - stats.page_types[descr->type].pages_uncompressed_bytes += descr->page_length; - stats.page_types[descr->type].pages_duration_secs += duration_secs; - stats.page_types[descr->type].points += points; + stats->pages_uncompressed_bytes += descr->page_length; + stats->pages_duration_secs += duration_s; + stats->points += points; - if(!stats.first_t || (descr->start_time_ut - update_every_usec) < stats.first_t) - stats.first_t = (descr->start_time_ut - update_every_usec) / USEC_PER_SEC; + stats->page_types[descr->type].pages++; + stats->page_types[descr->type].pages_uncompressed_bytes += descr->page_length; + stats->page_types[descr->type].pages_duration_secs += duration_s; + stats->page_types[descr->type].points += points; - if(!stats.last_t || descr->end_time_ut > stats.last_t) - stats.last_t = descr->end_time_ut / USEC_PER_SEC; - } + if(!stats->first_time_s || (start_time_s - update_every_s) < stats->first_time_s) + stats->first_time_s = (start_time_s - update_every_s); + + if(!stats->last_time_s || end_time_s > stats->last_time_s) + stats->last_time_s = end_time_s; + + descr++; } + metric++; } + journalfile_v2_data_release(datafile->journalfile); +} + +RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) { + RRDENG_SIZE_STATS stats = { 0 }; - stats.currently_collected_metrics = ctx->stats.metric_API_producers; - stats.max_concurrently_collected_metrics = ctx->metric_API_max_producers; + uv_rwlock_rdlock(&ctx->datafiles.rwlock); + for(struct rrdengine_datafile *df = ctx->datafiles.first; df ;df = df->next) { + stats.datafiles++; + populate_v2_statistics(df, &stats); + } + uv_rwlock_rdunlock(&ctx->datafiles.rwlock); + + stats.currently_collected_metrics = __atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED); internal_error(stats.metrics_pages != stats.extents_pages + stats.currently_collected_metrics, "DBENGINE: metrics pages is %zu, but extents pages is %zu and API consumers is %zu", stats.metrics_pages, stats.extents_pages, stats.currently_collected_metrics); - stats.disk_space = ctx->disk_space; - stats.max_disk_space = ctx->max_disk_space; + stats.disk_space = ctx_current_disk_space_get(ctx); + stats.max_disk_space = ctx->config.max_disk_space; - stats.database_retention_secs = (time_t)(stats.last_t - stats.first_t); + stats.database_retention_secs = (time_t)(stats.last_time_s - stats.first_time_s); if(stats.extents_pages) stats.average_page_size_bytes = (double)stats.pages_uncompressed_bytes / (double)stats.extents_pages; @@ -1252,21 +1386,22 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx) { } } - stats.sizeof_metric = struct_natural_alignment(sizeof(struct pg_cache_page_index) + sizeof(struct pg_alignment)); - stats.sizeof_page = struct_natural_alignment(sizeof(struct rrdeng_page_descr)); +// stats.sizeof_metric = 0; stats.sizeof_datafile = struct_natural_alignment(sizeof(struct rrdengine_datafile)) + struct_natural_alignment(sizeof(struct rrdengine_journalfile)); - stats.sizeof_page_in_cache = struct_natural_alignment(sizeof(struct page_cache_descr)); - stats.sizeof_point_data = page_type_size[ctx->page_type]; - stats.sizeof_page_data = RRDENG_BLOCK_SIZE; + stats.sizeof_page_in_cache = 0; // struct_natural_alignment(sizeof(struct page_cache_descr)); + stats.sizeof_point_data = page_type_size[ctx->config.page_type]; + stats.sizeof_page_data = tier_page_size[ctx->config.tier]; stats.pages_per_extent = rrdeng_pages_per_extent; - stats.sizeof_extent = sizeof(struct extent_info); - stats.sizeof_page_in_extent = sizeof(struct rrdeng_page_descr *); - - stats.sizeof_metric_in_index = 40; - stats.sizeof_page_in_index = 24; +// stats.sizeof_metric_in_index = 40; +// stats.sizeof_page_in_index = 24; - stats.default_granularity_secs = (size_t)default_rrd_update_every * get_tier_grouping(ctx->tier); + stats.default_granularity_secs = (size_t)default_rrd_update_every * get_tier_grouping(ctx->config.tier); return stats; } + +struct rrdeng_cache_efficiency_stats rrdeng_get_cache_efficiency_stats(void) { + // FIXME - make cache efficiency stats atomic + return rrdeng_cache_efficiency_stats; +} |