diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:03 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:18 +0000 |
commit | 5da14042f70711ea5cf66e034699730335462f66 (patch) | |
tree | 0f6354ccac934ed87a2d555f45be4c831cf92f4a /src/database/engine/rrdengineapi.c | |
parent | Releasing debian version 1.44.3-2. (diff) | |
download | netdata-5da14042f70711ea5cf66e034699730335462f66.tar.xz netdata-5da14042f70711ea5cf66e034699730335462f66.zip |
Merging upstream version 1.45.3+dfsg.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rwxr-xr-x | src/database/engine/rrdengineapi.c (renamed from database/engine/rrdengineapi.c) | 269 |
1 files changed, 150 insertions, 119 deletions
diff --git a/database/engine/rrdengineapi.c b/src/database/engine/rrdengineapi.c index 1ddce5243..43fed492b 100755 --- a/database/engine/rrdengineapi.c +++ b/src/database/engine/rrdengineapi.c @@ -2,6 +2,7 @@ #include "database/engine/rrddiskprotocol.h" #include "rrdengine.h" +#include "dbengine-compression.h" /* Default global database instance */ struct rrdengine_instance multidb_ctx_storage_tier0; @@ -16,7 +17,12 @@ struct rrdengine_instance multidb_ctx_storage_tier4; #error RRD_STORAGE_TIERS is not 5 - you need to add allocations here #endif struct rrdengine_instance *multidb_ctx[RRD_STORAGE_TIERS]; -uint8_t tier_page_type[RRD_STORAGE_TIERS] = {PAGE_METRICS, PAGE_TIER, PAGE_TIER, PAGE_TIER, PAGE_TIER}; +uint8_t tier_page_type[RRD_STORAGE_TIERS] = { + RRDENG_PAGE_TYPE_GORILLA_32BIT, + RRDENG_PAGE_TYPE_ARRAY_TIER1, + RRDENG_PAGE_TYPE_ARRAY_TIER1, + RRDENG_PAGE_TYPE_ARRAY_TIER1, + RRDENG_PAGE_TYPE_ARRAY_TIER1}; #if defined(ENV32BIT) size_t tier_page_size[RRD_STORAGE_TIERS] = {2048, 1024, 192, 192, 192}; @@ -24,14 +30,14 @@ size_t tier_page_size[RRD_STORAGE_TIERS] = {2048, 1024, 192, 192, 192}; size_t tier_page_size[RRD_STORAGE_TIERS] = {4096, 2048, 384, 384, 384}; #endif -#if PAGE_TYPE_MAX != 2 +#if RRDENG_PAGE_TYPE_MAX != 2 #error PAGE_TYPE_MAX is not 2 - you need to add allocations here #endif size_t page_type_size[256] = { - [PAGE_METRICS] = sizeof(storage_number), - [PAGE_TIER] = sizeof(storage_number_tier1_t), - [PAGE_GORILLA_METRICS] = sizeof(storage_number) + [RRDENG_PAGE_TYPE_ARRAY_32BIT] = sizeof(storage_number), + [RRDENG_PAGE_TYPE_ARRAY_TIER1] = sizeof(storage_number_tier1_t), + [RRDENG_PAGE_TYPE_GORILLA_32BIT] = sizeof(storage_number) }; __attribute__((constructor)) void initialize_multidb_ctx(void) { @@ -74,14 +80,14 @@ static inline bool rrdeng_page_alignment_release(struct pg_alignment *pa) { } // charts call this -STORAGE_METRICS_GROUP *rrdeng_metrics_group_get(STORAGE_INSTANCE *db_instance __maybe_unused, uuid_t *uuid __maybe_unused) { +STORAGE_METRICS_GROUP *rrdeng_metrics_group_get(STORAGE_INSTANCE *si __maybe_unused, uuid_t *uuid __maybe_unused) { struct pg_alignment *pa = callocz(1, sizeof(struct pg_alignment)); rrdeng_page_alignment_acquire(pa); return (STORAGE_METRICS_GROUP *)pa; } // charts call this -void rrdeng_metrics_group_release(STORAGE_INSTANCE *db_instance __maybe_unused, STORAGE_METRICS_GROUP *smg) { +void rrdeng_metrics_group_release(STORAGE_INSTANCE *si __maybe_unused, STORAGE_METRICS_GROUP *smg) { if(unlikely(!smg)) return; struct pg_alignment *pa = (struct pg_alignment *)smg; @@ -108,8 +114,8 @@ void rrdeng_generate_legacy_uuid(const char *dim_id, const char *chart_id, uuid_ memcpy(ret_uuid, hash_value, sizeof(uuid_t)); } -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; +static METRIC *rrdeng_metric_get_legacy(STORAGE_INSTANCE *si, const char *rd_id, const char *st_id) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; uuid_t legacy_uuid; rrdeng_generate_legacy_uuid(rd_id, st_id, &legacy_uuid); return mrg_metric_get_and_acquire(main_mrg, &legacy_uuid, (Word_t) ctx); @@ -118,25 +124,25 @@ static METRIC *rrdeng_metric_get_legacy(STORAGE_INSTANCE *db_instance, const cha // ---------------------------------------------------------------------------- // metric handle -void rrdeng_metric_release(STORAGE_METRIC_HANDLE *db_metric_handle) { - METRIC *metric = (METRIC *)db_metric_handle; +void rrdeng_metric_release(STORAGE_METRIC_HANDLE *smh) { + METRIC *metric = (METRIC *)smh; mrg_metric_release(main_mrg, metric); } -STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *db_metric_handle) { - METRIC *metric = (METRIC *)db_metric_handle; +STORAGE_METRIC_HANDLE *rrdeng_metric_dup(STORAGE_METRIC_HANDLE *smh) { + METRIC *metric = (METRIC *)smh; 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; +STORAGE_METRIC_HANDLE *rrdeng_metric_get(STORAGE_INSTANCE *si, uuid_t *uuid) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; return (STORAGE_METRIC_HANDLE *) mrg_metric_get_and_acquire(main_mrg, uuid, (Word_t) ctx); } -static METRIC *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid) { - internal_fatal(!db_instance, "DBENGINE: db_instance is NULL"); +static METRIC *rrdeng_metric_create(STORAGE_INSTANCE *si, uuid_t *uuid) { + internal_fatal(!si, "DBENGINE: STORAGE_INSTANCE is NULL"); - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; MRG_ENTRY entry = { .uuid = uuid, .section = (Word_t)ctx, @@ -145,12 +151,15 @@ static METRIC *rrdeng_metric_create(STORAGE_INSTANCE *db_instance, uuid_t *uuid) .latest_update_every_s = 0, }; - METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, NULL); + bool added; + METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, &added); + if (added) + __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED); return metric; } -STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *db_instance) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; +STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; METRIC *metric; metric = mrg_metric_get_and_acquire(main_mrg, &rd->metric_uuid, (Word_t) ctx); @@ -160,13 +169,13 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE // 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)); + metric = rrdeng_metric_get_legacy(si, 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); + metric = rrdeng_metric_create(si, &rd->metric_uuid); } #ifdef NETDATA_INTERNAL_CHECKS @@ -192,14 +201,14 @@ STORAGE_METRIC_HANDLE *rrdeng_metric_get_or_create(RRDDIM *rd, STORAGE_INSTANCE // 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((uint32_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 %u, but the metric registry has %u. Fixing it.", + (uint32_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)); + mrg_metric_set_update_every(main_mrg, handle->metric, (uint32_t)(handle->update_every_ut / USEC_PER_SEC)); } } @@ -213,7 +222,7 @@ static inline bool check_completed_page_consistency(struct rrdeng_collect_handle uuid_t *uuid = mrg_metric_uuid(main_mrg, handle->metric); time_t start_time_s = pgc_page_start_time_s(handle->pgc_page); time_t end_time_s = pgc_page_end_time_s(handle->pgc_page); - time_t update_every_s = pgc_page_update_every_s(handle->pgc_page); + uint32_t update_every_s = pgc_page_update_every_s(handle->pgc_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); @@ -245,8 +254,8 @@ static inline bool check_completed_page_consistency(struct rrdeng_collect_handle * 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) { - METRIC *metric = (METRIC *)db_metric_handle; +STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *smh, uint32_t update_every, STORAGE_METRICS_GROUP *smg) { + METRIC *metric = (METRIC *)smh; struct rrdengine_instance *ctx = mrg_metric_ctx(metric); bool is_1st_metric_writer = true; @@ -262,7 +271,7 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri struct rrdeng_collect_handle *handle; handle = callocz(1, sizeof(struct rrdeng_collect_handle)); - handle->common.backend = STORAGE_ENGINE_BACKEND_DBENGINE; + handle->common.seb = STORAGE_ENGINE_BACKEND_DBENGINE; handle->metric = metric; handle->pgc_page = NULL; @@ -288,15 +297,15 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri // 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); + time_t db_first_time_s, db_last_time_s; + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, NULL); handle->page_end_time_ut = (usec_t)db_last_time_s * USEC_PER_SEC; return (STORAGE_COLLECT_HANDLE *)handle; } -void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_handle) { - struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; +void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *sch) { + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch; if (unlikely(!handle->pgc_page)) return; @@ -307,7 +316,17 @@ void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *collection_h else { check_completed_page_consistency(handle); mrg_metric_set_clean_latest_time_s(main_mrg, handle->metric, pgc_page_end_time_s(handle->pgc_page)); - pgc_page_hot_to_dirty_and_release(main_cache, handle->pgc_page); + + struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); + time_t start_time_s = pgc_page_start_time_s(handle->pgc_page); + time_t end_time_s = pgc_page_end_time_s(handle->pgc_page); + uint32_t update_every_s = mrg_metric_get_update_every_s(main_mrg, handle->metric); + if (end_time_s && start_time_s && end_time_s > start_time_s && update_every_s) { + uint64_t add_samples = (end_time_s - start_time_s) / update_every_s; + __atomic_add_fetch(&ctx->atomic.samples, add_samples, __ATOMIC_RELAXED); + } + + pgc_page_hot_to_dirty_and_release(main_cache, handle->pgc_page, false); } mrg_metric_set_hot_latest_time_s(main_mrg, handle->metric, 0); @@ -336,7 +355,7 @@ static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *ha PGD *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); + const uint32_t update_every_s = (uint32_t)(handle->update_every_ut / USEC_PER_SEC); PGC_ENTRY page_entry = { .section = (Word_t) ctx, @@ -345,7 +364,7 @@ static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *ha .end_time_s = point_in_time_s, .size = data_size, .data = data, - .update_every_s = (uint32_t) update_every_s, + .update_every_s = update_every_s, .hot = true }; @@ -364,11 +383,11 @@ static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *ha nd_log_limit_static_global_var(erl, 1, 0); nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING, #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 - " + "DBENGINE: metric '%s' new page from %ld to %ld, update every %u, has a conflict in main cache " + "with existing %s%s page from %ld to %ld, update every %u - " "is it collected more than once?", uuid, - page_entry.start_time_s, page_entry.end_time_s, (time_t)page_entry.update_every_s, + page_entry.start_time_s, page_entry.end_time_s, page_entry.update_every_s, pgc_is_page_hot(pgc_page) ? "hot" : "not-hot", pgc_page_data(pgc_page) == PGD_EMPTY ? " gap" : "", pgc_page_start_time_s(pgc_page), pgc_page_end_time_s(pgc_page), pgc_page_update_every_s(pgc_page) @@ -444,14 +463,14 @@ static PGD *rrdeng_alloc_new_page_data(struct rrdeng_collect_handle *handle, siz *data_size = size; switch (ctx->config.page_type) { - case PAGE_METRICS: - case PAGE_TIER: + case RRDENG_PAGE_TYPE_ARRAY_32BIT: + case RRDENG_PAGE_TYPE_ARRAY_TIER1: d = pgd_create(ctx->config.page_type, slots); break; - case PAGE_GORILLA_METRICS: + case RRDENG_PAGE_TYPE_GORILLA_32BIT: // ignore slots, and use the fixed number of slots per gorilla buffer. // gorilla will automatically add more buffers if needed. - d = pgd_create(ctx->config.page_type, GORILLA_BUFFER_SLOTS); + d = pgd_create(ctx->config.page_type, RRDENG_GORILLA_32BIT_BUFFER_SLOTS); break; default: fatal("Unknown page type: %uc\n", ctx->config.page_type); @@ -461,7 +480,7 @@ static PGD *rrdeng_alloc_new_page_data(struct rrdeng_collect_handle *handle, siz return d; } -static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_handle, +static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *sch, const usec_t point_in_time_ut, const NETDATA_DOUBLE n, const NETDATA_DOUBLE min_value, @@ -470,7 +489,7 @@ static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_ const uint16_t anomaly_count, const SN_FLAGS flags) { - struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch; struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); if(unlikely(!handle->page_data)) @@ -497,7 +516,7 @@ static void rrdeng_store_metric_append_point(STORAGE_COLLECT_HANDLE *collection_ 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); + rrdeng_store_metric_flush_current_page(sch); } } @@ -543,7 +562,7 @@ static void store_metric_next_error_log(struct rrdeng_collect_handle *handle __m #endif } -void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, +void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *sch, const usec_t point_in_time_ut, const NETDATA_DOUBLE n, const NETDATA_DOUBLE min_value, @@ -554,7 +573,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, { timing_step(TIMING_STEP_RRDSET_STORE_METRIC); - struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)collection_handle; + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch; #ifdef NETDATA_INTERNAL_CHECKS if(unlikely(point_in_time_ut > (usec_t)max_acceptable_collected_time() * USEC_PER_SEC)) @@ -571,11 +590,11 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, if(handle->pgc_page) { if (unlikely(delta_ut < handle->update_every_ut)) { handle->page_flags |= RRDENG_PAGE_STEP_TOO_SMALL; - rrdeng_store_metric_flush_current_page(collection_handle); + rrdeng_store_metric_flush_current_page(sch); } else if (unlikely(delta_ut % handle->update_every_ut)) { handle->page_flags |= RRDENG_PAGE_STEP_UNALIGNED; - rrdeng_store_metric_flush_current_page(collection_handle); + rrdeng_store_metric_flush_current_page(sch); } else { size_t points_gap = delta_ut / handle->update_every_ut; @@ -583,7 +602,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, if (points_gap >= page_remaining_points) { handle->page_flags |= RRDENG_PAGE_BIG_GAP; - rrdeng_store_metric_flush_current_page(collection_handle); + rrdeng_store_metric_flush_current_page(sch); } else { // loop to fill the gap @@ -594,7 +613,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, this_ut <= stop_ut; this_ut = handle->page_end_time_ut + handle->update_every_ut) { rrdeng_store_metric_append_point( - collection_handle, + sch, this_ut, NAN, NAN, NAN, 1, 0, @@ -618,7 +637,7 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, timing_step(TIMING_STEP_DBENGINE_FIRST_CHECK); - rrdeng_store_metric_append_point(collection_handle, + rrdeng_store_metric_append_point(sch, point_in_time_ut, n, min_value, max_value, count, anomaly_count, @@ -629,12 +648,12 @@ void rrdeng_store_metric_next(STORAGE_COLLECT_HANDLE *collection_handle, * 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; +int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *sch) { + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch; struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); handle->page_flags |= RRDENG_PAGE_COLLECT_FINALIZE; - rrdeng_store_metric_flush_current_page(collection_handle); + rrdeng_store_metric_flush_current_page(sch); rrdeng_page_alignment_release(handle->alignment); __atomic_sub_fetch(&ctx->atomic.collectors_running, 1, __ATOMIC_RELAXED); @@ -644,8 +663,8 @@ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) { if((handle->options & RRDENG_1ST_METRIC_WRITER) && !mrg_metric_clear_writer(main_mrg, handle->metric)) internal_fatal(true, "DBENGINE: metric is already released"); - 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); + time_t first_time_s, last_time_s; + mrg_metric_get_retention(main_mrg, handle->metric, &first_time_s, &last_time_s, NULL); mrg_metric_release(main_mrg, handle->metric); freez(handle); @@ -656,8 +675,8 @@ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) { 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; +void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *sch, int update_every) { + struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch; check_and_fix_mrg_update_every(handle); METRIC *metric = handle->metric; @@ -667,7 +686,7 @@ void rrdeng_store_metric_change_collection_frequency(STORAGE_COLLECT_HANDLE *col return; handle->page_flags |= RRDENG_PAGE_UPDATE_EVERY_CHANGE; - rrdeng_store_metric_flush_current_page(collection_handle); + rrdeng_store_metric_flush_current_page(sch); mrg_metric_set_update_every(main_mrg, metric, update_every); handle->update_every_ut = update_every_ut; } @@ -704,8 +723,8 @@ static void unregister_query_handle(struct rrdeng_query_handle *handle __maybe_u * 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 *rrddim_handle, +void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *smh, + struct storage_engine_query_handle *seqh, time_t start_time_s, time_t end_time_s, STORAGE_PRIORITY priority) @@ -714,7 +733,7 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, netdata_thread_disable_cancelability(); - METRIC *metric = (METRIC *)db_metric_handle; + METRIC *metric = (METRIC *)smh; struct rrdengine_instance *ctx = mrg_metric_ctx(metric); struct rrdeng_query_handle *handle; @@ -736,7 +755,8 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, // 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; + time_t db_first_time_s, db_last_time_s; + uint32_t 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) { @@ -750,11 +770,11 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, mrg_metric_set_update_every_s_if_zero(main_mrg, metric, default_rrd_update_every); } - 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; - rrddim_handle->backend = STORAGE_ENGINE_BACKEND_DBENGINE; + seqh->handle = (STORAGE_QUERY_HANDLE *) handle; + seqh->start_time_s = handle->start_time_s; + seqh->end_time_s = handle->end_time_s; + seqh->priority = priority; + seqh->seb = STORAGE_ENGINE_BACKEND_DBENGINE; pg_cache_preload(handle); @@ -766,16 +786,16 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, 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; - rrddim_handle->backend = STORAGE_ENGINE_BACKEND_DBENGINE; + seqh->handle = (STORAGE_QUERY_HANDLE *) handle; + seqh->start_time_s = handle->start_time_s; + seqh->end_time_s = 0; + seqh->priority = priority; + seqh->seb = STORAGE_ENGINE_BACKEND_DBENGINE; } } -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; +static bool rrdeng_load_page_next(struct storage_engine_query_handle *seqh, bool debug_this __maybe_unused) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle; struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric); if (likely(handle->page)) { @@ -785,7 +805,7 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han pgdc_reset(&handle->pgdc, NULL, UINT32_MAX); } - if (unlikely(handle->now_s > rrddim_handle->end_time_s)) + if (unlikely(handle->now_s > seqh->end_time_s)) return false; size_t entries = 0; @@ -799,7 +819,7 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han 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); + uint32_t page_update_every_s = pgc_page_update_every_s(handle->page); unsigned position; if(likely(handle->now_s >= page_start_time_s && handle->now_s <= page_end_time_s)) { @@ -810,13 +830,13 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han } 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; + time_t point_end_time_s = page_start_time_s + position * (time_t) 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; + point_end_time_s = page_start_time_s + position * (time_t) page_update_every_s; } handle->now_s = point_end_time_s; } @@ -845,11 +865,11 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han // Returns the metric and sets its timestamp into current_time // IT IS REQUIRED TO **ALWAYS** SET ALL RETURN VALUES (current_time, end_time, flags) // IT IS REQUIRED TO **ALWAYS** KEEP TRACK OF TIME, EVEN OUTSIDE THE DATABASE BOUNDARIES -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; +STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *seqh) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle; STORAGE_POINT sp; - if (unlikely(handle->now_s > rrddim_handle->end_time_s)) { + if (unlikely(handle->now_s > seqh->end_time_s)) { storage_point_empty(sp, handle->now_s - handle->dt_s, handle->now_s); goto prepare_for_next_iteration; } @@ -857,8 +877,8 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim if (unlikely(!handle->page || handle->position >= handle->entries)) { // We need to get a new page - if (!rrdeng_load_page_next(rrddim_handle, false)) { - handle->now_s = rrddim_handle->end_time_s; + if (!rrdeng_load_page_next(seqh, false)) { + handle->now_s = seqh->end_time_s; storage_point_empty(sp, handle->now_s - handle->dt_s, handle->now_s); goto prepare_for_next_iteration; } @@ -870,7 +890,7 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim pgdc_get_next_point(&handle->pgdc, handle->position, &sp); 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 < seqh->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"); handle->now_s += handle->dt_s; @@ -879,17 +899,17 @@ prepare_for_next_iteration: return sp; } -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); +int rrdeng_load_metric_is_finished(struct storage_engine_query_handle *seqh) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle; + return (handle->now_s > seqh->end_time_s); } /* * Releases the database reference from the handle for loading metrics. */ -void rrdeng_load_metric_finalize(struct storage_engine_query_handle *rrddim_handle) +void rrdeng_load_metric_finalize(struct storage_engine_query_handle *seqh) { - struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrddim_handle->handle; + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle; if (handle->page) { pgc_page_release(main_cache, handle->page); @@ -901,24 +921,24 @@ void rrdeng_load_metric_finalize(struct storage_engine_query_handle *rrddim_hand unregister_query_handle(handle); rrdeng_query_handle_release(handle); - rrddim_handle->handle = NULL; + seqh->handle = NULL; netdata_thread_enable_cancelability(); } -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; +time_t rrdeng_load_align_to_optimal_before(struct storage_engine_query_handle *seqh) { + struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle; 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; + if (handle->pdc->optimal_end_time_s > seqh->end_time_s) + seqh->end_time_s = handle->pdc->optimal_end_time_s; } - return rrddim_handle->end_time_s; + return seqh->end_time_s; } -time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { - METRIC *metric = (METRIC *)db_metric_handle; +time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *smh) { + METRIC *metric = (METRIC *)smh; time_t latest_time_s = 0; if (metric) @@ -927,8 +947,8 @@ time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { return latest_time_s; } -time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { - METRIC *metric = (METRIC *)db_metric_handle; +time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *smh) { + METRIC *metric = (METRIC *)smh; time_t oldest_time_s = 0; if (metric) @@ -937,9 +957,9 @@ time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *db_metric_handle) { return oldest_time_s; } -bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *db_instance, uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s) +bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *si, uuid_t *dim_uuid, time_t *first_entry_s, time_t *last_entry_s) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; if (unlikely(!ctx)) { netdata_log_error("DBENGINE: invalid STORAGE INSTANCE to %s()", __FUNCTION__); return false; @@ -949,26 +969,35 @@ bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *db_instance, uuid_t *dim_ if (unlikely(!metric)) return false; - time_t update_every_s; - mrg_metric_get_retention(main_mrg, metric, first_entry_s, last_entry_s, &update_every_s); + mrg_metric_get_retention(main_mrg, metric, first_entry_s, last_entry_s, NULL); mrg_metric_release(main_mrg, metric); return true; } -uint64_t rrdeng_disk_space_max(STORAGE_INSTANCE *db_instance) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; +uint64_t rrdeng_disk_space_max(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; return ctx->config.max_disk_space; } -uint64_t rrdeng_disk_space_used(STORAGE_INSTANCE *db_instance) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; +uint64_t rrdeng_disk_space_used(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; return __atomic_load_n(&ctx->atomic.current_disk_space, __ATOMIC_RELAXED); } -time_t rrdeng_global_first_time_s(STORAGE_INSTANCE *db_instance) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; +uint64_t rrdeng_metrics(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; + return __atomic_load_n(&ctx->atomic.metrics, __ATOMIC_RELAXED); +} + +uint64_t rrdeng_samples(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; + return __atomic_load_n(&ctx->atomic.samples, __ATOMIC_RELAXED); +} + +time_t rrdeng_global_first_time_s(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; time_t t = __atomic_load_n(&ctx->atomic.first_time_s, __ATOMIC_RELAXED); if(t == LONG_MAX || t < 0) @@ -977,8 +1006,8 @@ time_t rrdeng_global_first_time_s(STORAGE_INSTANCE *db_instance) { return t; } -size_t rrdeng_currently_collected_metrics(STORAGE_INSTANCE *db_instance) { - struct rrdengine_instance *ctx = (struct rrdengine_instance *)db_instance; +size_t rrdeng_currently_collected_metrics(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; return __atomic_load_n(&ctx->atomic.collectors_running, __ATOMIC_RELAXED); } @@ -1099,8 +1128,8 @@ void rrdeng_readiness_wait(struct rrdengine_instance *ctx) { netdata_log_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; +bool rrdeng_is_legacy(STORAGE_INSTANCE *si) { + struct rrdengine_instance *ctx = (struct rrdengine_instance *)si; return ctx->config.legacy; } @@ -1142,7 +1171,7 @@ int rrdeng_init(struct rrdengine_instance **ctxp, const char *dbfiles_path, ctx->config.tier = (int)tier; ctx->config.page_type = tier_page_type[tier]; - ctx->config.global_compress_alg = RRD_LZ4; + ctx->config.global_compress_alg = dbengine_default_compression(); if (disk_space_mb < RRDENG_MIN_DISK_SPACE_MB) disk_space_mb = RRDENG_MIN_DISK_SPACE_MB; ctx->config.max_disk_space = disk_space_mb * 1048576LLU; @@ -1154,6 +1183,8 @@ int rrdeng_init(struct rrdengine_instance **ctxp, const char *dbfiles_path, rw_spinlock_init(&ctx->njfv2idx.spinlock); ctx->atomic.first_time_s = LONG_MAX; + ctx->atomic.metrics = 0; + ctx->atomic.samples = 0; if (rrdeng_dbengine_spawn(ctx) && !init_rrd_files(ctx)) { // success - we run this ctx too |