archive/netdata_netdata
0
0
Fork 0
mirror of https://github.com/netdata/netdata.git synced 2025-04-17 11:12:42 +00:00
Code Issues Projects Releases Wiki Activity

Revert "Configurable storage engine for Netdata agents: step 3 ()" ()

Browse source 
This reverts commit 100a12c6cc.

A couple parent/child startup/shutdown scenarios can lead to crashes.
This commit is contained in:
vkalintiris 2022-06-17 14:59:35 +03:00 • committed by GitHub
parent 1aab506fcb
commit afae8971f0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 217 additions and 320 deletions

3
daemon/analytics.c
View file

@ -2,9 +2,6 @@
#include "common.h"
#include "buildinfo.h"
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengineapi.h"
#endif
struct analytics_data analytics_data;
extern void analytics_exporting_connectors (BUFFER *b);

5
daemon/global_statistics.c
View file

@ -1,9 +1,6 @@
// SPDX-License-Identifier: GPL-3.0-or-later
#include "common.h"
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengineapi.h"
#endif
#define GLOBAL_STATS_RESET_WEB_USEC_MAX 0x01
@ -459,7 +456,7 @@ static void dbengine_statistics_charts(void) {
rrdhost_foreach_read(host) {
if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && !rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {
if (host->rrdeng_ctx == host->rrdeng_ctx->engine->context) {
if (&multidb_ctx == host->rrdeng_ctx) {
if (counted_multihost_db)
continue; /* Only count multi-host DB once */
counted_multihost_db = 1;

21
daemon/main.c
View file

@ -3,10 +3,6 @@
#include "common.h"
#include "buildinfo.h"
#include "static_threads.h"
#include "database/storage_engine.h"
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengineapi.h"
#endif
int netdata_zero_metrics_enabled;
int netdata_anonymous_statistics_enabled;
@ -58,18 +54,13 @@ void netdata_cleanup_and_exit(int ret) {
// free the database
info("EXIT: freeing database memory...");
for (STORAGE_ENGINE* eng = storage_engine_foreach_init(); eng; eng = storage_engine_foreach_next(eng)) {
if (eng->context)
eng->api.engine_ops.exit(eng->context);
}
#ifdef ENABLE_DBENGINE
rrdeng_prepare_exit(&multidb_ctx);
#endif
rrdhost_free_all();
for (STORAGE_ENGINE* eng = storage_engine_foreach_init(); eng; eng = storage_engine_foreach_next(eng)) {
if (eng->context) {
eng->api.engine_ops.destroy(eng->context);
eng->context = NULL;
}
}
#ifdef ENABLE_DBENGINE
rrdeng_exit(&multidb_ctx);
#endif
}
sql_close_database();

7
daemon/unit_test.c
View file

@ -1,9 +1,6 @@
// SPDX-License-Identifier: GPL-3.0-or-later
#include "common.h"
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengineapi.h"
#endif
static int check_number_printing(void) {
struct {
@ -1160,7 +1157,7 @@ int run_test(struct test *test)
RRDSET *st = rrdset_create_localhost("netdata", name, name, "netdata", NULL, "Unit Testing", "a value", "unittest", NULL, 1
, test->update_every, RRDSET_TYPE_LINE);
RRDDIM *rd = rrddim_add(st, "dim1", NULL, test->multiplier, test->divisor, test->algorithm);
RRDDIM *rd2 = NULL;
if(test->feed2)
rd2 = rrddim_add(st, "dim2", NULL, test->multiplier, test->divisor, test->algorithm);
@ -1176,7 +1173,7 @@ int run_test(struct test *test)
if(c) {
time_now += test->feed[c].microseconds;
fprintf(stderr, " > %s: feeding position %lu, after %0.3f seconds (%0.3f seconds from start), delta " CALCULATED_NUMBER_FORMAT ", rate " CALCULATED_NUMBER_FORMAT "\n",
fprintf(stderr, " > %s: feeding position %lu, after %0.3f seconds (%0.3f seconds from start), delta " CALCULATED_NUMBER_FORMAT ", rate " CALCULATED_NUMBER_FORMAT "\n",
test->name, c+1,
(float)test->feed[c].microseconds / 1000000.0,
(float)time_now / 1000000.0,

9
database/engine/rrdengine.c
View file

@ -1339,12 +1339,13 @@ error_after_loop_init:
*/
void rrdengine_main(void)
{
STORAGE_ENGINE_INSTANCE* ctx;
int ret;
struct rrdengine_instance *ctx;
sanity_check();
ctx = rrdeng_init(storage_engine_get(RRD_MEMORY_MODE_DBENGINE), NULL);
if (!ctx) {
exit(1);
ret = rrdeng_init(NULL, &ctx, "/tmp", RRDENG_MIN_PAGE_CACHE_SIZE_MB, RRDENG_MIN_DISK_SPACE_MB);
if (ret) {
exit(ret);
}
rrdeng_exit(ctx);
fprintf(stderr, "Hello world!");

2
database/engine/rrdengine.h
View file

@ -13,7 +13,6 @@
#include <openssl/evp.h>
#include "daemon/common.h"
#include "../rrd.h"
#include "../storage_engine.h"
#include "rrddiskprotocol.h"
#include "rrdenginelib.h"
#include "datafile.h"
@ -227,7 +226,6 @@ extern rrdeng_stats_t global_flushing_pressure_page_deletions; /* number of dele
#define QUIESCED (2) /* is set after all threads have finished running */
struct rrdengine_instance {
STORAGE_ENGINE_INSTANCE parent;
struct metalog_instance *metalog_ctx;
struct rrdengine_worker_config worker_config;
struct completion rrdengine_completion;

99
database/engine/rrdengineapi.c
View file

@ -1,6 +1,8 @@
// SPDX-License-Identifier: GPL-3.0-or-later
#include "rrdengine.h"
#include "../storage_engine.h"
/* Default global database instance */
struct rrdengine_instance multidb_ctx;
int db_engine_use_malloc = 0;
int default_rrdeng_page_cache_mb = 32;
@ -9,16 +11,9 @@ int default_multidb_disk_quota_mb = 256;
/* Default behaviour is to unblock data collection if the page cache is full of dirty pages by dropping metrics */
uint8_t rrdeng_drop_metrics_under_page_cache_pressure = 1;
void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrdeng_page_descr **ret_descr);
void rrdeng_commit_page(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr,
Word_t page_correlation_id);
void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **handle);
void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle);
void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle);
static inline struct rrdengine_instance *get_rrdeng_ctx_from_host(RRDHOST *host)
{
return (struct rrdengine_instance*) host->rrdeng_ctx;
return host->rrdeng_ctx;
}
/* This UUID is not unique across hosts */
@ -74,7 +69,7 @@ void rrdeng_metric_init(RRDDIM *rd)
pg_cache = &ctx->pg_cache;
rrdeng_generate_legacy_uuid(rd->id, rd->rrdset->id, &legacy_uuid);
if (host != localhost && host->rrdeng_ctx->engine && host->rrdeng_ctx == host->rrdeng_ctx->engine->context)
if (host != localhost && host->rrdeng_ctx == &multidb_ctx)
is_multihost_child = 1;
uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
@ -201,12 +196,16 @@ void rrdeng_store_metric_flush_current_page(RRDDIM *rd)
void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number number)
{
struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)rd->state->handle;
struct rrdengine_instance *ctx = handle->ctx;
struct page_cache *pg_cache = &ctx->pg_cache;
struct rrdeng_page_descr *descr = handle->descr;
struct rrdengine_instance *ctx;
struct page_cache *pg_cache;
struct rrdeng_page_descr *descr;
storage_number *page;
uint8_t must_flush_unaligned_page = 0, perfect_page_alignment = 0;
ctx = handle->ctx;
pg_cache = &ctx->pg_cache;
descr = handle->descr;
if (descr) {
/* Make alignment decisions */
@ -821,11 +820,10 @@ void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_i
* 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(STORAGE_ENGINE_INSTANCE* context, unsigned long long *array)
void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array)
{
if (context == NULL)
if (ctx == NULL)
return;
struct rrdengine_instance* ctx = (struct rrdengine_instance*) context;
struct page_cache *pg_cache = &ctx->pg_cache;
@ -876,23 +874,19 @@ void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle)
pg_cache_put(ctx, (struct rrdeng_page_descr *)handle);
}
STORAGE_ENGINE_INSTANCE*
rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host)
/*
* Returns 0 on success, negative on error
*/
int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
unsigned disk_space_mb)
{
struct rrdengine_instance *ctx;
int error;
uint32_t max_open_files;
bool is_legacy = is_legacy_child(host->machine_guid);
if (!is_legacy && eng->context) {
if (host->rrd_memory_mode == eng->id && host->rrdeng_ctx == NULL) {
host->rrdeng_ctx = eng->context;
}
return eng->context;
}
max_open_files = rlimit_nofile.rlim_cur / 4;
/* reserve RRDENG_FD_BUDGET_PER_INSTANCE file descriptors for this instance */
uint32_t max_open_files = rlimit_nofile.rlim_cur / 4;
rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, RRDENG_FD_BUDGET_PER_INSTANCE);
if (rrdeng_reserved_file_descriptors > max_open_files) {
error(
@ -901,18 +895,15 @@ rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host)
rrd_stat_atomic_add(&global_fs_errors, 1);
rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
return NULL;//UV_EMFILE;
return UV_EMFILE;
}
char dbfiles_path[FILENAME_MAX + 1];
snprintfz(dbfiles_path, FILENAME_MAX, "%s/dbengine", host->cache_dir);
mkdir(dbfiles_path, 0775);
int page_cache_mb = default_rrdeng_page_cache_mb;
int disk_space_mb = default_rrdeng_disk_quota_mb;
ctx = callocz(1, sizeof(*ctx));
ctx->parent.engine = eng;
if (NULL == ctxp) {
ctx = &multidb_ctx;
memset(ctx, 0, sizeof(*ctx));
} else {
*ctxp = ctx = callocz(1, sizeof(*ctx));
}
ctx->global_compress_alg = RRD_LZ4;
if (page_cache_mb < RRDENG_MIN_PAGE_CACHE_SIZE_MB)
page_cache_mb = RRDENG_MIN_PAGE_CACHE_SIZE_MB;
@ -935,15 +926,6 @@ rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host)
ctx->metalog_ctx = NULL; /* only set this after the metadata log has finished initializing */
ctx->host = host;
// Attach context as the global context
if (!is_legacy && !eng->context) {
eng->context = (STORAGE_ENGINE_INSTANCE *)ctx;
}
// Attach context as the host context
if (host->rrd_memory_mode == eng->id && host->rrdeng_ctx == NULL) {
host->rrdeng_ctx = eng->context;
}
memset(&ctx->worker_config, 0, sizeof(ctx->worker_config));
ctx->worker_config.ctx = ctx;
init_page_cache(ctx);
@ -968,30 +950,30 @@ rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host)
goto error_after_rrdeng_worker;
}
return (STORAGE_ENGINE_INSTANCE *)ctx;
return 0;
error_after_rrdeng_worker:
finalize_rrd_files(ctx);
error_after_init_rrd_files:
free_page_cache(ctx);
if ((STORAGE_ENGINE_INSTANCE *)ctx != eng->context) {
if (ctx != &multidb_ctx) {
freez(ctx);
*ctxp = NULL;
}
rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
return NULL;//UV_EIO;
return UV_EIO;
}
/*
* Returns 0 on success, 1 on error
*/
void rrdeng_exit(STORAGE_ENGINE_INSTANCE* context)
int rrdeng_exit(struct rrdengine_instance *ctx)
{
struct rrdeng_cmd cmd;
if (NULL == context) {
return;
if (NULL == ctx) {
return 1;
}
struct rrdengine_instance* ctx = (struct rrdengine_instance*)context;
/* TODO: add page to page cache */
cmd.opcode = RRDENG_SHUTDOWN;
@ -1002,18 +984,21 @@ void rrdeng_exit(STORAGE_ENGINE_INSTANCE* context)
finalize_rrd_files(ctx);
//metalog_exit(ctx->metalog_ctx);
free_page_cache(ctx);
freez(ctx);
if (ctx != &multidb_ctx) {
freez(ctx);
}
rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
return 0;
}
void rrdeng_prepare_exit(STORAGE_ENGINE_INSTANCE* context)
void rrdeng_prepare_exit(struct rrdengine_instance *ctx)
{
struct rrdeng_cmd cmd;
if (NULL == context) {
if (NULL == ctx) {
return;
}
struct rrdengine_instance* ctx = (struct rrdengine_instance*)context;
completion_init(&ctx->rrdengine_completion);
cmd.opcode = RRDENG_QUIESCE;

17
database/engine/rrdengineapi.h
View file

@ -17,6 +17,7 @@ extern int default_rrdeng_page_cache_mb;
extern int default_rrdeng_disk_quota_mb;
extern int default_multidb_disk_quota_mb;
extern uint8_t rrdeng_drop_metrics_under_page_cache_pressure;
extern struct rrdengine_instance multidb_ctx;
struct rrdeng_region_info {
time_t start_time;
@ -24,6 +25,13 @@ struct rrdeng_region_info {
unsigned points;
};
extern void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrdeng_page_descr **ret_descr);
extern void rrdeng_commit_page(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr,
Word_t page_correlation_id);
extern void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **handle);
extern void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle);
extern void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle);
extern void rrdeng_generate_legacy_uuid(const char *dim_id, char *chart_id, uuid_t *ret_uuid);
extern void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uuid_t *legacy_uuid,
uuid_t *ret_uuid);
@ -44,13 +52,14 @@ extern int rrdeng_load_metric_is_finished(struct rrddim_query_handle *rrdimm_han
extern void rrdeng_load_metric_finalize(struct rrddim_query_handle *rrdimm_handle);
extern time_t rrdeng_metric_latest_time(RRDDIM *rd);
extern time_t rrdeng_metric_oldest_time(RRDDIM *rd);
extern void rrdeng_get_37_statistics(STORAGE_ENGINE_INSTANCE *ctx, unsigned long long *array);
extern void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array);
/* must call once before using anything */
extern STORAGE_ENGINE_INSTANCE* rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host);
extern int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
unsigned disk_space_mb);
extern void rrdeng_exit(STORAGE_ENGINE_INSTANCE *ctx);
extern void rrdeng_prepare_exit(STORAGE_ENGINE_INSTANCE *ctx);
extern int rrdeng_exit(struct rrdengine_instance *ctx);
extern void rrdeng_prepare_exit(struct rrdengine_instance *ctx);
extern int rrdeng_metric_latest_time_by_uuid(uuid_t *dim_uuid, time_t *first_entry_t, time_t *last_entry_t);
#endif /* NETDATA_RRDENGINEAPI_H */

14
database/ram/rrddim_mem.c
View file

@ -1,20 +1,6 @@
// SPDX-License-Identifier: GPL-3.0-or-later
#include "rrddim_mem.h"
#include "../storage_engine.h"
STORAGE_ENGINE_INSTANCE* rrddim_storage_engine_instance_new(STORAGE_ENGINE* engine, RRDHOST *host) {
(void)engine; (void)host;
return NULL;
}
void rrddim_storage_engine_instance_exit(STORAGE_ENGINE_INSTANCE* context) {
(void)context;
}
void rrddim_storage_engine_instance_destroy(STORAGE_ENGINE_INSTANCE* context) {
(void)context;
}
// ----------------------------------------------------------------------------
// RRDDIM legacy data collection functions

4
database/ram/rrddim_mem.h
View file

@ -15,10 +15,6 @@ struct mem_query_handle {
uint8_t finished;
};
STORAGE_ENGINE_INSTANCE* rrddim_storage_engine_instance_new(STORAGE_ENGINE* engine, RRDHOST *host);
void rrddim_storage_engine_instance_exit(STORAGE_ENGINE_INSTANCE* context);
void rrddim_storage_engine_instance_destroy(STORAGE_ENGINE_INSTANCE* context);
extern void rrddim_collect_init(RRDDIM *rd);
extern void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number);
extern int rrddim_collect_finalize(RRDDIM *rd);

21
database/rrd.c
View file

@ -28,10 +28,31 @@ int gap_when_lost_iterations_above = 1;
// RRD - memory modes
inline const char *rrd_memory_mode_name(RRD_MEMORY_MODE id) {
switch(id) {
case RRD_MEMORY_MODE_RAM:
return RRD_MEMORY_MODE_RAM_NAME;
case RRD_MEMORY_MODE_MAP:
return RRD_MEMORY_MODE_MAP_NAME;
case RRD_MEMORY_MODE_NONE:
return RRD_MEMORY_MODE_NONE_NAME;
case RRD_MEMORY_MODE_SAVE:
return RRD_MEMORY_MODE_SAVE_NAME;
case RRD_MEMORY_MODE_ALLOC:
return RRD_MEMORY_MODE_ALLOC_NAME;
case RRD_MEMORY_MODE_DBENGINE:
return RRD_MEMORY_MODE_DBENGINE_NAME;
}
STORAGE_ENGINE* eng = storage_engine_get(id);
if (eng) {
return eng->name;
}
return RRD_MEMORY_MODE_SAVE_NAME;
}

9
database/rrd.h
View file

@ -18,8 +18,6 @@ typedef struct rrdcalc RRDCALC;
typedef struct rrdcalctemplate RRDCALCTEMPLATE;
typedef struct alarm_entry ALARM_ENTRY;
typedef struct context_param CONTEXT_PARAM;
typedef struct storage_engine_instance STORAGE_ENGINE_INSTANCE;
typedef struct storage_engine STORAGE_ENGINE;
typedef void *ml_host_t;
typedef void *ml_dimension_t;
@ -856,7 +854,9 @@ struct rrdhost {
avl_tree_lock rrdfamily_root_index; // the host's chart families index
avl_tree_lock rrdvar_root_index; // the host's chart variables index
STORAGE_ENGINE_INSTANCE *rrdeng_ctx; // DB engine instance for this host
#ifdef ENABLE_DBENGINE
struct rrdengine_instance *rrdeng_ctx; // DB engine instance for this host
#endif
uuid_t host_uuid; // Global GUID for this host
uuid_t *node_id; // Cloud node_id
@ -1324,6 +1324,9 @@ extern void set_host_properties(
// ----------------------------------------------------------------------------
// RRD DB engine declarations
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengineapi.h"
#endif
#include "sqlite/sqlite_functions.h"
#include "sqlite/sqlite_aclk.h"
#include "sqlite/sqlite_aclk_chart.h"

89
database/rrdhost.c
View file

@ -2,10 +2,6 @@
#define NETDATA_RRD_INTERNALS
#include "rrd.h"
#include "storage_engine.h"
#ifdef ENABLE_DBENGINE
#include "engine/rrdenginelib.h"
#endif
RRDHOST *localhost = NULL;
size_t rrd_hosts_available = 0;
@ -337,8 +333,41 @@ RRDHOST *rrdhost_create(const char *hostname,
else
error_report("Host machine GUID %s is not valid", host->machine_guid);
// Create engine
host->rrdeng_ctx = storage_engine_new(storage_engine_get(host->rrd_memory_mode), host);
if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
#ifdef ENABLE_DBENGINE
char dbenginepath[FILENAME_MAX + 1];
int ret;
snprintfz(dbenginepath, FILENAME_MAX, "%s/dbengine", host->cache_dir);
ret = mkdir(dbenginepath, 0775);
if (ret != 0 && errno != EEXIST)
error("Host '%s': cannot create directory '%s'", host->hostname, dbenginepath);
else ret = 0; // succeed
if (is_legacy) // initialize legacy dbengine instance as needed
ret = rrdeng_init(host, &host->rrdeng_ctx, dbenginepath, default_rrdeng_page_cache_mb,
default_rrdeng_disk_quota_mb); // may fail here for legacy dbengine initialization
else
host->rrdeng_ctx = &multidb_ctx;
if (ret) { // check legacy or multihost initialization success
error(
"Host '%s': cannot initialize host with machine guid '%s'. Failed to initialize DB engine at '%s'.",
host->hostname, host->machine_guid, host->cache_dir);
rrdhost_free(host);
host = NULL;
//rrd_hosts_available++; //TODO: maybe we want this?
return host;
}
#else
fatal("RRD_MEMORY_MODE_DBENGINE is not supported in this platform.");
#endif
}
else {
#ifdef ENABLE_DBENGINE
host->rrdeng_ctx = &multidb_ctx;
#endif
}
// ------------------------------------------------------------------------
// link it and add it to the index
@ -641,8 +670,10 @@ restart_after_removal:
info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", host->hostname, host->machine_guid);
if (rrdhost_flag_check(host, RRDHOST_FLAG_DELETE_ORPHAN_HOST)
#ifdef ENABLE_DBENGINE
/* don't delete multi-host DB host files */
&& !(host->rrdeng_ctx->engine && host->rrdeng_ctx->engine->context == host->rrdeng_ctx)
&& !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx == &multidb_ctx)
#endif
)
rrdhost_delete_charts(host);
else
@ -709,6 +740,25 @@ int rrd_init(char *hostname, struct rrdhost_system_info *system_info) {
return 1;
}
#ifdef ENABLE_DBENGINE
char dbenginepath[FILENAME_MAX + 1];
int ret;
snprintfz(dbenginepath, FILENAME_MAX, "%s/dbengine", localhost->cache_dir);
ret = mkdir(dbenginepath, 0775);
if (ret != 0 && errno != EEXIST)
error("Host '%s': cannot create directory '%s'", localhost->hostname, dbenginepath);
else // Unconditionally create multihost db to support on demand host creation
ret = rrdeng_init(NULL, NULL, dbenginepath, default_rrdeng_page_cache_mb, default_multidb_disk_quota_mb);
if (ret) {
error(
"Host '%s' with machine guid '%s' failed to initialize multi-host DB engine instance at '%s'.",
localhost->hostname, localhost->machine_guid, localhost->cache_dir);
rrdhost_free(localhost);
localhost = NULL;
rrd_unlock();
fatal("Failed to initialize dbengine");
}
#endif
sql_aclk_sync_init();
rrd_unlock();
@ -853,12 +903,12 @@ void rrdhost_free(RRDHOST *host) {
// ------------------------------------------------------------------------
// release its children resources
STORAGE_ENGINE* eng = host->rrdeng_ctx->engine;
if (host->rrdeng_ctx != eng->context) {
if (eng && eng->api.engine_ops.exit)
eng->api.engine_ops.exit(host->rrdeng_ctx);
#ifdef ENABLE_DBENGINE
if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
if (host->rrdeng_ctx != &multidb_ctx)
rrdeng_prepare_exit(host->rrdeng_ctx);
}
#endif
while(host->rrdset_root)
rrdset_free(host->rrdset_root);
@ -889,11 +939,10 @@ void rrdhost_free(RRDHOST *host) {
health_alarm_log_free(host);
if (host->rrdeng_ctx != eng->context) {
if (eng)
eng->api.engine_ops.destroy(host->rrdeng_ctx);
host->rrdeng_ctx = NULL;
}
#ifdef ENABLE_DBENGINE
if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx != &multidb_ctx)
rrdeng_exit(host->rrdeng_ctx);
#endif
// ------------------------------------------------------------------------
// remove it from the indexes
@ -1209,8 +1258,10 @@ void rrdhost_cleanup_all(void) {
RRDHOST *host;
rrdhost_foreach_read(host) {
if (host != localhost && rrdhost_flag_check(host, RRDHOST_FLAG_DELETE_ORPHAN_HOST) && !host->receiver
#ifdef ENABLE_DBENGINE
/* don't delete multi-host DB host files */
&& !(host->rrdeng_ctx->engine && host->rrdeng_ctx->engine->context == host->rrdeng_ctx)
&& !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx == &multidb_ctx)
#endif
)
rrdhost_delete_charts(host);
else
@ -1293,7 +1344,7 @@ restart_after_removal:
rrdvar_free_remaining_variables(host, &st->rrdvar_root_index);
rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
if (st->dimensions) {
/* If the chart still has dimensions don't delete it from the metadata log */
continue;

4
database/rrdset.c
View file

@ -3,10 +3,6 @@
#define NETDATA_RRD_INTERNALS
#include "rrd.h"
#include <sched.h>
#ifdef ENABLE_DBENGINE
#include "database/engine/rrddiskprotocol.h"
#include "database/engine/rrdengineapi.h"
#endif
void __rrdset_check_rdlock(RRDSET *st, const char *file, const char *function, const unsigned long line) {
debug(D_RRD_CALLS, "Checking read lock on chart '%s'", st->id);

3
database/sqlite/sqlite_aclk_chart.c
View file

@ -2,9 +2,6 @@
#include "sqlite_functions.h"
#include "sqlite_aclk_chart.h"
#ifdef ENABLE_DBENGINE
#include "../engine/rrdengineapi.h"
#endif
#if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
#include "../../aclk/aclk_charts_api.h"

12
database/sqlite/sqlite_functions.c
View file

@ -1,10 +1,6 @@
// SPDX-License-Identifier: GPL-3.0-or-later
#include "sqlite_functions.h"
#ifdef ENABLE_DBENGINE
#include "../engine/rrdengineapi.h"
#endif
#include "database/storage_engine.h"
#define DB_METADATA_VERSION "1"
@ -427,7 +423,7 @@ int sql_init_database(db_check_action_type_t rebuild, int memory)
// PRAGMA temp_store = 0 | DEFAULT | 1 | FILE | 2 | MEMORY;
snprintfz(buf, 1024, "PRAGMA temp_store=%s;", config_get(CONFIG_SECTION_SQLITE, "temp store", "MEMORY"));
if(init_database_batch(rebuild, 0, list)) return 1;
// https://www.sqlite.org/pragma.html#pragma_journal_size_limit
// PRAGMA schema.journal_size_limit = N ;
snprintfz(buf, 1024, "PRAGMA journal_size_limit=%lld;", config_get_number(CONFIG_SECTION_SQLITE, "journal size limit", 16777216));
@ -1282,9 +1278,9 @@ RRDHOST *sql_create_host_by_uuid(char *hostname)
host->system_info = callocz(1, sizeof(*host->system_info));;
rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED);
// Create multidb engine instance if necessary
host->rrdeng_ctx = storage_engine_new(storage_engine_get(RRD_MEMORY_MODE_DBENGINE), host);
#ifdef ENABLE_DBENGINE
host->rrdeng_ctx = &multidb_ctx;
#endif
failed:
rc = sqlite3_finalize(res);

194
database/storage_engine.c
View file

@ -5,144 +5,68 @@
#ifdef ENABLE_DBENGINE
#include "engine/rrdengineapi.h"
#endif
#include "libnetdata/libnetdata.h"
#define im_collect_ops { \
.init = rrddim_collect_init,\
.store_metric = rrddim_collect_store_metric,\
.finalize = rrddim_collect_finalize\
}
#define im_query_ops { \
.init = rrddim_query_init, \
.next_metric = rrddim_query_next_metric, \
.is_finished = rrddim_query_is_finished, \
.finalize = rrddim_query_finalize, \
.latest_time = rrddim_query_latest_time, \
.oldest_time = rrddim_query_oldest_time \
}
static STORAGE_ENGINE engines[] = {
{
.id = RRD_MEMORY_MODE_NONE,
.name = RRD_MEMORY_MODE_NONE_NAME,
.api = {
.engine_ops = {
.create = rrddim_storage_engine_instance_new,
.exit = rrddim_storage_engine_instance_exit,
.destroy = rrddim_storage_engine_instance_destroy
},
.collect_ops = {
.init = rrddim_collect_init,
.store_metric = rrddim_collect_store_metric,
.finalize = rrddim_collect_finalize
},
.query_ops = {
.init = rrddim_query_init,
.next_metric = rrddim_query_next_metric,
.is_finished = rrddim_query_is_finished,
.finalize = rrddim_query_finalize,
.latest_time = rrddim_query_latest_time,
.oldest_time = rrddim_query_oldest_time
}
},
.context = NULL
.collect_ops = im_collect_ops,
.query_ops = im_query_ops
}
},
{
.id = RRD_MEMORY_MODE_RAM,
.name = RRD_MEMORY_MODE_RAM_NAME,
.api = {
.engine_ops = {
.create = rrddim_storage_engine_instance_new,
.exit = rrddim_storage_engine_instance_exit,
.destroy = rrddim_storage_engine_instance_destroy
},
.collect_ops = {
.init = rrddim_collect_init,
.store_metric = rrddim_collect_store_metric,
.finalize = rrddim_collect_finalize
},
.query_ops = {
.init = rrddim_query_init,
.next_metric = rrddim_query_next_metric,
.is_finished = rrddim_query_is_finished,
.finalize = rrddim_query_finalize,
.latest_time = rrddim_query_latest_time,
.oldest_time = rrddim_query_oldest_time
}
},
.context = NULL
.collect_ops = im_collect_ops,
.query_ops = im_query_ops
}
},
{
.id = RRD_MEMORY_MODE_MAP,
.name = RRD_MEMORY_MODE_MAP_NAME,
.api = {
.engine_ops = {
.create = rrddim_storage_engine_instance_new,
.exit = rrddim_storage_engine_instance_exit,
.destroy = rrddim_storage_engine_instance_destroy
},
.collect_ops = {
.init = rrddim_collect_init,
.store_metric = rrddim_collect_store_metric,
.finalize = rrddim_collect_finalize
},
.query_ops = {
.init = rrddim_query_init,
.next_metric = rrddim_query_next_metric,
.is_finished = rrddim_query_is_finished,
.finalize = rrddim_query_finalize,
.latest_time = rrddim_query_latest_time,
.oldest_time = rrddim_query_oldest_time
}
},
.context = NULL
.collect_ops = im_collect_ops,
.query_ops = im_query_ops
}
},
{
.id = RRD_MEMORY_MODE_SAVE,
.name = RRD_MEMORY_MODE_SAVE_NAME,
.api = {
.engine_ops = {
.create = rrddim_storage_engine_instance_new,
.exit = rrddim_storage_engine_instance_exit,
.destroy = rrddim_storage_engine_instance_destroy
},
.collect_ops = {
.init = rrddim_collect_init,
.store_metric = rrddim_collect_store_metric,
.finalize = rrddim_collect_finalize
},
.query_ops = {
.init = rrddim_query_init,
.next_metric = rrddim_query_next_metric,
.is_finished = rrddim_query_is_finished,
.finalize = rrddim_query_finalize,
.latest_time = rrddim_query_latest_time,
.oldest_time = rrddim_query_oldest_time
}
},
.context = NULL
.collect_ops = im_collect_ops,
.query_ops = im_query_ops
}
},
{
.id = RRD_MEMORY_MODE_ALLOC,
.name = RRD_MEMORY_MODE_ALLOC_NAME,
.api = {
.engine_ops = {
.create = rrddim_storage_engine_instance_new,
.exit = rrddim_storage_engine_instance_exit,
.destroy = rrddim_storage_engine_instance_destroy
},
.collect_ops = {
.init = rrddim_collect_init,
.store_metric = rrddim_collect_store_metric,
.finalize = rrddim_collect_finalize
},
.query_ops = {
.init = rrddim_query_init,
.next_metric = rrddim_query_next_metric,
.is_finished = rrddim_query_is_finished,
.finalize = rrddim_query_finalize,
.latest_time = rrddim_query_latest_time,
.oldest_time = rrddim_query_oldest_time
}
},
.context = NULL
.collect_ops = im_collect_ops,
.query_ops = im_query_ops
}
},
#ifdef ENABLE_DBENGINE
{
.id = RRD_MEMORY_MODE_DBENGINE,
.name = RRD_MEMORY_MODE_DBENGINE_NAME,
.api = {
.engine_ops = {
.create = rrdeng_init,
.exit = rrdeng_prepare_exit,
.destroy = rrdeng_exit
},
.collect_ops = {
.init = rrdeng_store_metric_init,
.store_metric = rrdeng_store_metric_next,
@ -156,34 +80,28 @@ static STORAGE_ENGINE engines[] = {
.latest_time = rrdeng_metric_latest_time,
.oldest_time = rrdeng_metric_oldest_time
}
},
.context = NULL
}
}
},
#endif
{ .id = RRD_MEMORY_MODE_NONE, .name = NULL }
};
const size_t engine_count = sizeof(engines) / sizeof(engines[0]);
STORAGE_ENGINE* storage_engine_find(const char* name)
{
for (size_t i = 0; i != engine_count; i++)
if (strcmp(engines[i].name, name) == 0)
return &engines[i];
error("No storage engine found for memory mode %s.", name);
return storage_engine_get(default_rrd_memory_mode);
for (STORAGE_ENGINE* it = engines; it->name; it++) {
if (strcmp(it->name, name) == 0)
return it;
}
return NULL;
}
STORAGE_ENGINE* storage_engine_get(RRD_MEMORY_MODE mmode)
{
for (size_t i = 0; i != engine_count; i++)
if (engines[i].id == mmode)
return &engines[i];
STORAGE_ENGINE* eng = mmode == default_rrd_memory_mode
? &engines[0] // default engine not available, use NONE
: storage_engine_get(default_rrd_memory_mode);
error("No storage engine for memory mode %u, will use %s (%u) instead.", mmode, eng->name, eng->id);
return eng;
for (STORAGE_ENGINE* it = engines; it->name; it++) {
if (it->id == mmode)
return it;
}
return NULL;
}
STORAGE_ENGINE* storage_engine_foreach_init()
@ -194,27 +112,9 @@ STORAGE_ENGINE* storage_engine_foreach_init()
STORAGE_ENGINE* storage_engine_foreach_next(STORAGE_ENGINE* it)
{
if (!it || !it->name)
return NULL;
it++;
return (it >= &engines[engine_count]) ? NULL : it;
}
STORAGE_ENGINE_INSTANCE* storage_engine_new(STORAGE_ENGINE* eng, RRDHOST *host)
{
STORAGE_ENGINE_INSTANCE* instance = host->rrdeng_ctx;
if (!instance && eng) {
instance = eng->api.engine_ops.create(eng, host);
if (instance) {
instance->engine = eng;
}
}
return instance;
}
void storage_engine_delete(STORAGE_ENGINE_INSTANCE* instance) {
if (instance) {
STORAGE_ENGINE* eng = instance->engine;
if (eng) {
eng->api.engine_ops.destroy(instance);
}
}
return it->name ? it : NULL;
}

21
database/storage_engine.h
View file

@ -6,20 +6,10 @@
#include "rrd.h"
typedef struct storage_engine STORAGE_ENGINE;
typedef struct storage_engine_instance STORAGE_ENGINE_INSTANCE;
// ------------------------------------------------------------------------
// function pointers that handle storage engine instance creation and destruction
struct storage_engine_ops {
STORAGE_ENGINE_INSTANCE*(*create)(STORAGE_ENGINE* engine, RRDHOST *host);
void(*exit)(STORAGE_ENGINE_INSTANCE*);
void(*destroy)(STORAGE_ENGINE_INSTANCE*);
};
// ------------------------------------------------------------------------
// function pointers for all APIs provided by a storge engine
typedef struct storage_engine_api {
struct storage_engine_ops engine_ops;
struct rrddim_collect_ops collect_ops;
struct rrddim_query_ops query_ops;
} STORAGE_ENGINE_API;
@ -28,12 +18,6 @@ struct storage_engine {
RRD_MEMORY_MODE id;
const char* name;
STORAGE_ENGINE_API api;
STORAGE_ENGINE_INSTANCE* context;
};
// Abstract structure to be extended by implementations
struct storage_engine_instance {
STORAGE_ENGINE* engine;
};
extern STORAGE_ENGINE* storage_engine_get(RRD_MEMORY_MODE mmode);
@ -43,9 +27,4 @@ extern STORAGE_ENGINE* storage_engine_find(const char* name);
extern STORAGE_ENGINE* storage_engine_foreach_init();
extern STORAGE_ENGINE* storage_engine_foreach_next(STORAGE_ENGINE* it);
// ------------------------------------------------------------------------
// Retreive or create a storage engine instance for host
STORAGE_ENGINE_INSTANCE* storage_engine_new(STORAGE_ENGINE* engine, RRDHOST *host);
void storage_engine_delete(STORAGE_ENGINE_INSTANCE* engine);
#endif

3
web/api/queries/query.c
View file

@ -4,9 +4,6 @@
#include "web/api/formatters/rrd2json.h"
#include "rrdr.h"
#include "database/ram/rrddim_mem.h"
#ifdef ENABLE_DBENGINE
#include "database/engine/rrdengine.h"
#endif
#include "average/average.h"
#include "incremental_sum/incremental_sum.h"