// SPDX-License-Identifier: GPL-3.0-or-later

#include "database/engine/rrddiskprotocol.h"

#include "rrdengine.h"

/* Default global database instance */

struct rrdengine_instance multidb_ctx_storage_tier0;

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 != 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)

};

__attribute__((constructor)) void initialize_multidb_ctx(void) {

multidb_ctx[0] = &multidb_ctx_storage_tier0;

multidb_ctx[1] = &multidb_ctx_storage_tier1;

multidb_ctx[2] = &multidb_ctx_storage_tier2;

multidb_ctx[3] = &multidb_ctx_storage_tier3;

multidb_ctx[4] = &multidb_ctx_storage_tier4;

}

int db_engine_journal_check = 0;

int default_rrdeng_disk_quota_mb = 256;

int default_multidb_disk_quota_mb = 256;

#if defined(ENV32BIT)

int default_rrdeng_page_cache_mb = 16;

int default_rrdeng_extent_cache_mb = 0;

#else

int default_rrdeng_page_cache_mb = 32;

int default_rrdeng_extent_cache_mb = 0;

#endif

// ----------------------------------------------------------------------------

// metrics groups

static inline void rrdeng_page_alignment_acquire(struct pg_alignment *pa) {

if(unlikely(!pa)) return;

__atomic_add_fetch(&pa->refcount, 1, __ATOMIC_SEQ_CST);

}

static inline bool rrdeng_page_alignment_release(struct pg_alignment *pa) {

if(unlikely(!pa)) return true;

if(__atomic_sub_fetch(&pa->refcount, 1, __ATOMIC_SEQ_CST) == 0) {

freez(pa);

return true;

}

return false;

}

// charts call this

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 *si __maybe_unused, STORAGE_METRICS_GROUP *smg) {

if(unlikely(!smg)) return;

struct pg_alignment *pa = (struct pg_alignment *)smg;

rrdeng_page_alignment_release(pa);

}

// ----------------------------------------------------------------------------

// metric handle for legacy dbs

/* This UUID is not unique across hosts */

void rrdeng_generate_legacy_uuid(const char *dim_id, const char *chart_id, 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, dim_id, strlen(dim_id));

EVP_DigestUpdate(evpctx, chart_id, strlen(chart_id));

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));

}

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);

}

// ----------------------------------------------------------------------------

// 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 *smh) {

METRIC *metric = (METRIC *)smh;

return (STORAGE_METRIC_HANDLE *) mrg_metric_dup(main_mrg, metric);

}

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 *si, uuid_t *uuid) {

internal_fatal(!si, "DBENGINE: STORAGE_INSTANCE is NULL");

struct rrdengine_instance *ctx = (struct rrdengine_instance *)si;

MRG_ENTRY entry = {

.uuid = uuid,

.section = (Word_t)ctx,

.first_time_s = 0,

.last_time_s = 0,

.latest_update_every_s = 0,

};

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 *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);

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(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(si, &rd->metric_uuid);

}

#ifdef NETDATA_INTERNAL_CHECKS

if(uuid_memcmp(&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(*mrg_metric_uuid(main_mrg, metric), uuid2);

fatal("DBENGINE: uuids do not match, asked for metric '%s', but got metric '%s'", uuid1, uuid2);

}

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 (STORAGE_METRIC_HANDLE *)metric;

}

// ----------------------------------------------------------------------------

// collect ops

static inline void check_and_fix_mrg_update_every(struct rrdeng_collect_handle *handle) {

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, (uint32_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->pgc_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->pgc_page);

time_t end_time_s = pgc_page_end_time_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);

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 *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;

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);

netdata_log_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->common.seb = STORAGE_ENGINE_BACKEND_DBENGINE;

handle->metric = metric;

handle->pgc_page = NULL;

handle->page_data = NULL;

handle->page_data_size = 0;

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);

// 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;

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 *sch) {

struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)sch;

if (unlikely(!handle->pgc_page))

return;

if(pgd_is_empty(handle->page_data))

pgc_page_to_clean_evict_or_release(main_cache, handle->pgc_page);

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);

}

mrg_metric_set_hot_latest_time_s(main_mrg, handle->metric, 0);

handle->pgc_page = NULL;

handle->page_flags = 0;

handle->page_position = 0;

handle->page_entries_max = 0;

handle->page_data = NULL;

handle->page_data_size = 0;

// 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);

timing_step(TIMING_STEP_DBENGINE_FLUSH_PAGE);

}

static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *handle,

struct rrdengine_instance *ctx,

usec_t point_in_time_ut,

PGD *data,

size_t data_size) {

time_t point_in_time_s = (time_t)(point_in_time_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,

.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 *pgc_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

internal_error(true,

#else

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 %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, 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)

);

pgc_page_release(main_cache, pgc_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;

pgc_page = pgc_page_add_and_acquire(main_cache, page_entry, &added);

}

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->pgc_page = pgc_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);

timing_step(TIMING_STEP_DBENGINE_CREATE_NEW_PAGE);

}

static size_t aligned_allocation_entries(size_t max_slots, size_t target_slot, time_t now_s) {

size_t slots = target_slot;

size_t pos = (now_s % max_slots);

if(pos > slots)

slots += max_slots - pos;

else if(pos < slots)

slots -= pos;

else

slots = max_slots;

return slots;

}

static PGD *rrdeng_alloc_new_page_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);

PGD *d = NULL;

size_t max_size = tier_page_size[ctx->config.tier];

size_t max_slots = max_size / CTX_POINT_SIZE_BYTES(ctx);

size_t slots = aligned_allocation_entries(

max_slots,

indexing_partition((Word_t) handle->alignment, max_slots),

(time_t) (point_in_time_ut / USEC_PER_SEC)

);

if(slots < max_slots / 3)

slots = max_slots / 3;

if(slots < 3)

slots = 3;

size_t size = slots * CTX_POINT_SIZE_BYTES(ctx);

// internal_error(true, "PAGE ALLOC %zu bytes (%zu max)", size, max_size);

internal_fatal(slots < 3 || slots > max_slots, "ooops! wrong distribution of metrics across time");

internal_fatal(size > tier_page_size[ctx->config.tier] || size < CTX_POINT_SIZE_BYTES(ctx) * 2, "ooops! wrong page size");

*data_size = size;

switch (ctx->config.page_type) {

case PAGE_METRICS:

case PAGE_TIER:

d = pgd_create(ctx->config.page_type, slots);

break;

case PAGE_GORILLA_METRICS:

// 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);

break;

default:

fatal("Unknown page type: %uc\n", ctx->config.page_type);

}

timing_step(TIMING_STEP_DBENGINE_PAGE_ALLOC);

return d;

}

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,

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 *)sch;

struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric);

if(unlikely(!handle->page_data))

handle->page_data = rrdeng_alloc_new_page_data(handle, &handle->page_data_size, point_in_time_ut);

timing_step(TIMING_STEP_DBENGINE_CHECK_DATA);

pgd_append_point(handle->page_data,

point_in_time_ut,

n, min_value, max_value, count, anomaly_count, flags,

handle->page_position);

timing_step(TIMING_STEP_DBENGINE_PACK);

if(unlikely(!handle->pgc_page)) {

rrdeng_store_metric_create_new_page(handle, ctx, point_in_time_ut, handle->page_data, handle->page_data_size);

// handle->position is set to 1 already

}

else {

// update an existing page

pgc_page_hot_set_end_time_s(main_cache, handle->pgc_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(sch);

}

}

timing_step(TIMING_STEP_DBENGINE_PAGE_FIN);

// update the metric information

mrg_metric_set_hot_latest_time_s(main_mrg, handle->metric, (time_t) (point_in_time_ut / USEC_PER_SEC));

timing_step(TIMING_STEP_DBENGINE_MRG_UPDATE);

}

static void store_metric_next_error_log(struct rrdeng_collect_handle *handle __maybe_unused, usec_t point_in_time_ut __maybe_unused, const char *msg __maybe_unused) {

#ifdef NETDATA_INTERNAL_CHECKS

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);

BUFFER *wb = NULL;

if(handle->pgc_page && handle->page_flags) {

wb = buffer_create(0, NULL);

collect_page_flags_to_buffer(wb, handle->page_flags);

}

nd_log_limit_static_global_var(erl, 1, 0);

nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE,

"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->pgc_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);

#else

;

#endif

}

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,

const NETDATA_DOUBLE max_value,

const uint16_t count,

const uint16_t anomaly_count,

const SN_FLAGS flags)

{

timing_step(TIMING_STEP_RRDSET_STORE_METRIC);

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))

handle->page_flags |= RRDENG_PAGE_FUTURE_POINT;

#endif

usec_t delta_ut = point_in_time_ut - handle->page_end_time_ut;

if(likely(delta_ut == handle->update_every_ut)) {

// happy path

;

}

else if(unlikely(point_in_time_ut > handle->page_end_time_ut)) {

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(sch);

}

else if (unlikely(delta_ut % handle->update_every_ut)) {

handle->page_flags |= RRDENG_PAGE_STEP_UNALIGNED;

rrdeng_store_metric_flush_current_page(sch);

}

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(sch);

}

else {

// loop to fill the gap

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(

sch,

this_ut,

NAN, NAN, NAN,

1, 0,

SN_EMPTY_SLOT);

}

}

}

}

}

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;

}

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;

}

timing_step(TIMING_STEP_DBENGINE_FIRST_CHECK);

rrdeng_store_metric_append_point(sch,

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 *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(sch);

rrdeng_page_alignment_release(handle->alignment);

__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);

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;

mrg_metric_get_retention(main_mrg, handle->metric, &first_time_s, &last_time_s, NULL);

mrg_metric_release(main_mrg, handle->metric);

freez(handle);

if(!first_time_s && !last_time_s)

return 1;

return 0;

}

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;

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(sch);

mrg_metric_set_update_every(main_mrg, metric, update_every);

handle->update_every_ut = update_every_ut;

}

// ----------------------------------------------------------------------------

// query ops

#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();

spinlock_lock(&global_query_handle_spinlock);

DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(global_query_handle_ll, handle, prev, next);

spinlock_unlock(&global_query_handle_spinlock);

}

static void unregister_query_handle(struct rrdeng_query_handle *handle) {

spinlock_lock(&global_query_handle_spinlock);

DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(global_query_handle_ll, handle, prev, next);

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 *smh,

struct storage_engine_query_handle *seqh,

time_t start_time_s,

time_t end_time_s,

STORAGE_PRIORITY priority)

{

usec_t started_ut = now_monotonic_usec();

netdata_thread_disable_cancelability();

METRIC *metric = (METRIC *)smh;

struct rrdengine_instance *ctx = mrg_metric_ctx(metric);

struct rrdeng_query_handle *handle;

handle = rrdeng_query_handle_get();

register_query_handle(handle);

if (unlikely(priority < STORAGE_PRIORITY_HIGH))

priority = STORAGE_PRIORITY_HIGH;

else if (unlikely(priority >= STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE))

priority = STORAGE_PRIORITY_INTERNAL_MAX_DONT_USE - 1;

handle->ctx = ctx;

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;

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) {

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);

}

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);

__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;

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 *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)) {

// we have a page to release

pgc_page_release(main_cache, handle->page);

handle->page = NULL;

pgdc_reset(&handle->pgdc, NULL, UINT32_MAX);

}

if (unlikely(handle->now_s > seqh->end_time_s))

return false;

size_t entries = 0;

handle->page = pg_cache_lookup_next(ctx, handle->pdc, handle->now_s, handle->dt_s, &entries);

internal_fatal(handle->page && (pgc_page_data(handle->page) == PGD_EMPTY || !entries),

"A page was returned, but it is empty - pg_cache_lookup_next() should be handling this case");

if (unlikely(!handle->page || pgc_page_data(handle->page) == PGD_EMPTY || !entries))

return false;

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);

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)) {

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 * (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 * (time_t) page_update_every_s;

}

handle->now_s = point_end_time_s;

}

internal_fatal(position >= entries, "DBENGINE: wrong page position calculation");

}

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->entries = entries;

handle->position = position;

handle->dt_s = page_update_every_s;

pgdc_reset(&handle->pgdc, pgc_page_data(handle->page), handle->position);

return true;

}

// 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 *seqh) {

struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle;

STORAGE_POINT sp;

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;

}

if (unlikely(!handle->page || handle->position >= handle->entries)) {

// We need to get a new page

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;

}

}

sp.start_time_s = handle->now_s - handle->dt_s;

sp.end_time_s = handle->now_s;

pgdc_get_next_point(&handle->pgdc, handle->position, &sp);

prepare_for_next_iteration:

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;

handle->position++;

return sp;

}

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 *seqh)

{

struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)seqh->handle;

if (handle->page) {

pgc_page_release(main_cache, handle->page);

pgdc_reset(&handle->pgdc, NULL, UINT32_MAX);

}

if(!pdc_release_and_destroy_if_unreferenced(handle->pdc, false, false))

__atomic_store_n(&handle->pdc->workers_should_stop, true, __ATOMIC_RELAXED);

unregister_query_handle(handle);

rrdeng_query_handle_release(handle);

seqh->handle = NULL;

netdata_thread_enable_cancelability();

}

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 > seqh->end_time_s)

seqh->end_time_s = handle->pdc->optimal_end_time_s;

}

return seqh->end_time_s;

}

time_t rrdeng_metric_latest_time(STORAGE_METRIC_HANDLE *smh) {

METRIC *metric = (METRIC *)smh;

time_t latest_time_s = 0;

if (metric)

latest_time_s = mrg_metric_get_latest_time_s(main_mrg, metric);

return latest_time_s;

}

time_t rrdeng_metric_oldest_time(STORAGE_METRIC_HANDLE *smh) {

METRIC *metric = (METRIC *)smh;

time_t oldest_time_s = 0;

if (metric)

oldest_time_s = mrg_metric_get_first_time_s(main_mrg, metric);

return oldest_time_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 *)si;

if (unlikely(!ctx)) {

netdata_log_error("DBENGINE: invalid STORAGE INSTANCE to %s()", __FUNCTION__);

return false;

}

METRIC *metric = mrg_metric_get_and_acquire(main_mrg, dim_uuid, (Word_t) ctx);

if (unlikely(!metric))

return false;

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 *si) {

struct rrdengine_instance *ctx = (struct rrdengine_instance *)si;

return ctx->config.max_disk_space;

}

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 *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)

t = 0;

return t;

}

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);

}

/*

* Gathers Database Engine statistics.

* Careful when modifying this function.

* You must not change the indices of the statistics or user code will break.

* You must not exceed RRDENG_NR_STATS or it will crash.

*/

void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array)

{

if (ctx == NULL)

return;

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;