summaryrefslogtreecommitdiffstats
path: root/src/database/engine/rrdengineapi.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rwxr-xr-xsrc/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
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-16 04:18:22 +0000