@cryptotaxi247 / netdata-1 / commits / 100a12c6c

Configurable storage engine for Netdata agents: step 3 (#12892)

* storage engine: add host context API Add a new API to allow storage engines to manage host contexts. * Replace single global context with per-engine global context * Context is full managed by storage engines: a storage engine can use no context, a global engine context, per host contexts, or a mix of these. * Currently, only dbengine uses contexts. Following the current logic, legacy hosts use their own context, while non-legacy hosts share the global context. * storage engine: use empty function instead of null for context ops * rrdhost: don't check return value for void call * rrdhost: create context with host * storage engine: move rrddim ops to rrddim_mem.{c,h} * storage engine: don't use NULL for end-of-list marker * storage engine: fallback to default engine

Adrien Béraud committed Jun 16, 2022 at 09:53 UTC 100a12c6cc01222b1518e5e50d2147f592d8a111
19 files changed +320 -217
daemon/analytics.c
+3
@@ -2,6 +2,9 @@
2
3 #include "common.h"
4 #include "buildinfo.h"
5 +#ifdef ENABLE_DBENGINE
6 +#include "database/engine/rrdengineapi.h"
7 +#endif
8
9 struct analytics_data analytics_data;
10 extern void analytics_exporting_connectors (BUFFER *b);
daemon/global_statistics.c
+4 -1
@@ -1,6 +1,9 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "common.h"
4 +#ifdef ENABLE_DBENGINE
5 +#include "database/engine/rrdengineapi.h"
6 +#endif
7
8 #define GLOBAL_STATS_RESET_WEB_USEC_MAX 0x01
9
@@ -456,7 +459,7 @@ static void dbengine_statistics_charts(void) {
459
460 rrdhost_foreach_read(host) {
461 if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && !rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {
459 - if (&multidb_ctx == host->rrdeng_ctx) {
462 + if (host->rrdeng_ctx == host->rrdeng_ctx->engine->context) {
463 if (counted_multihost_db)
464 continue; /* Only count multi-host DB once */
465 counted_multihost_db = 1;
daemon/main.c
+15 -6
@@ -3,6 +3,10 @@
3 #include "common.h"
4 #include "buildinfo.h"
5 #include "static_threads.h"
6 +#include "database/storage_engine.h"
7 +#ifdef ENABLE_DBENGINE
8 +#include "database/engine/rrdengineapi.h"
9 +#endif
10
11 int netdata_zero_metrics_enabled;
12 int netdata_anonymous_statistics_enabled;
@@ -54,13 +58,18 @@ void netdata_cleanup_and_exit(int ret) {
58
59 // free the database
60 info("EXIT: freeing database memory...");
57 -#ifdef ENABLE_DBENGINE
58 - rrdeng_prepare_exit(&multidb_ctx);
59 -#endif
61 + for (STORAGE_ENGINE* eng = storage_engine_foreach_init(); eng; eng = storage_engine_foreach_next(eng)) {
62 + if (eng->context)
63 + eng->api.engine_ops.exit(eng->context);
64 + }
65 +
66 rrdhost_free_all();
61 -#ifdef ENABLE_DBENGINE
62 - rrdeng_exit(&multidb_ctx);
63 -#endif
67 + for (STORAGE_ENGINE* eng = storage_engine_foreach_init(); eng; eng = storage_engine_foreach_next(eng)) {
68 + if (eng->context) {
69 + eng->api.engine_ops.destroy(eng->context);
70 + eng->context = NULL;
71 + }
72 + }
73 }
74 sql_close_database();
75
daemon/unit_test.c
+5 -2
@@ -1,6 +1,9 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "common.h"
4 +#ifdef ENABLE_DBENGINE
5 +#include "database/engine/rrdengineapi.h"
6 +#endif
7
8 static int check_number_printing(void) {
9 struct {
@@ -1157,7 +1160,7 @@ int run_test(struct test *test)
1160 RRDSET *st = rrdset_create_localhost("netdata", name, name, "netdata", NULL, "Unit Testing", "a value", "unittest", NULL, 1
1161 , test->update_every, RRDSET_TYPE_LINE);
1162 RRDDIM *rd = rrddim_add(st, "dim1", NULL, test->multiplier, test->divisor, test->algorithm);
1160 -
1163 +
1164 RRDDIM *rd2 = NULL;
1165 if(test->feed2)
1166 rd2 = rrddim_add(st, "dim2", NULL, test->multiplier, test->divisor, test->algorithm);
@@ -1173,7 +1176,7 @@ int run_test(struct test *test)
1176
1177 if(c) {
1178 time_now += test->feed[c].microseconds;
1176 - fprintf(stderr, " > %s: feeding position %lu, after %0.3f seconds (%0.3f seconds from start), delta " CALCULATED_NUMBER_FORMAT ", rate " CALCULATED_NUMBER_FORMAT "\n",
1179 + fprintf(stderr, " > %s: feeding position %lu, after %0.3f seconds (%0.3f seconds from start), delta " CALCULATED_NUMBER_FORMAT ", rate " CALCULATED_NUMBER_FORMAT "\n",
1180 test->name, c+1,
1181 (float)test->feed[c].microseconds / 1000000.0,
1182 (float)time_now / 1000000.0,
database/engine/rrdengine.c
+4 -5
@@ -1339,13 +1339,12 @@ error_after_loop_init:
1339 */
1340 void rrdengine_main(void)
1341 {
1342 - int ret;
1343 - struct rrdengine_instance *ctx;
1342 + STORAGE_ENGINE_INSTANCE* ctx;
1343
1344 sanity_check();
1346 - ret = rrdeng_init(NULL, &ctx, "/tmp", RRDENG_MIN_PAGE_CACHE_SIZE_MB, RRDENG_MIN_DISK_SPACE_MB);
1347 - if (ret) {
1348 - exit(ret);
1345 + ctx = rrdeng_init(storage_engine_get(RRD_MEMORY_MODE_DBENGINE), NULL);
1346 + if (!ctx) {
1347 + exit(1);
1348 }
1349 rrdeng_exit(ctx);
1350 fprintf(stderr, "Hello world!");
database/engine/rrdengine.h
+2
@@ -13,6 +13,7 @@
13 #include <openssl/evp.h>
14 #include "daemon/common.h"
15 #include "../rrd.h"
16 +#include "../storage_engine.h"
17 #include "rrddiskprotocol.h"
18 #include "rrdenginelib.h"
19 #include "datafile.h"
@@ -226,6 +227,7 @@ extern rrdeng_stats_t global_flushing_pressure_page_deletions; /* number of dele
227 #define QUIESCED (2) /* is set after all threads have finished running */
228
229 struct rrdengine_instance {
230 + STORAGE_ENGINE_INSTANCE parent;
231 struct metalog_instance *metalog_ctx;
232 struct rrdengine_worker_config worker_config;
233 struct completion rrdengine_completion;
database/engine/rrdengineapi.c
+57 -42
@@ -1,8 +1,6 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2 #include "rrdengine.h"
3 -
4 -/* Default global database instance */
5 -struct rrdengine_instance multidb_ctx;
3 +#include "../storage_engine.h"
4
5 int db_engine_use_malloc = 0;
6 int default_rrdeng_page_cache_mb = 32;
@@ -11,9 +9,16 @@ int default_multidb_disk_quota_mb = 256;
9 /* Default behaviour is to unblock data collection if the page cache is full of dirty pages by dropping metrics */
10 uint8_t rrdeng_drop_metrics_under_page_cache_pressure = 1;
11
12 +void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrdeng_page_descr **ret_descr);
13 +void rrdeng_commit_page(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr,
14 + Word_t page_correlation_id);
15 +void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **handle);
16 +void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle);
17 +void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle);
18 +
19 static inline struct rrdengine_instance *get_rrdeng_ctx_from_host(RRDHOST *host)
20 {
16 - return host->rrdeng_ctx;
21 + return (struct rrdengine_instance*) host->rrdeng_ctx;
22 }
23
24 /* This UUID is not unique across hosts */
@@ -69,7 +74,7 @@ void rrdeng_metric_init(RRDDIM *rd)
74 pg_cache = &ctx->pg_cache;
75
76 rrdeng_generate_legacy_uuid(rd->id, rd->rrdset->id, &legacy_uuid);
72 - if (host != localhost && host->rrdeng_ctx == &multidb_ctx)
77 + if (host != localhost && host->rrdeng_ctx->engine && host->rrdeng_ctx == host->rrdeng_ctx->engine->context)
78 is_multihost_child = 1;
79
80 uv_rwlock_rdlock(&pg_cache->metrics_index.lock);
@@ -196,16 +201,12 @@ void rrdeng_store_metric_flush_current_page(RRDDIM *rd)
201 void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number number)
202 {
203 struct rrdeng_collect_handle *handle = (struct rrdeng_collect_handle *)rd->state->handle;
199 - struct rrdengine_instance *ctx;
200 - struct page_cache *pg_cache;
201 - struct rrdeng_page_descr *descr;
204 + struct rrdengine_instance *ctx = handle->ctx;
205 + struct page_cache *pg_cache = &ctx->pg_cache;
206 + struct rrdeng_page_descr *descr = handle->descr;
207 storage_number *page;
208 uint8_t must_flush_unaligned_page = 0, perfect_page_alignment = 0;
209
205 - ctx = handle->ctx;
206 - pg_cache = &ctx->pg_cache;
207 - descr = handle->descr;
208 -
210 if (descr) {
211 /* Make alignment decisions */
212
@@ -820,10 +821,11 @@ void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_i
821 * You must not change the indices of the statistics or user code will break.
822 * You must not exceed RRDENG_NR_STATS or it will crash.
823 */
823 -void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array)
824 +void rrdeng_get_37_statistics(STORAGE_ENGINE_INSTANCE* context, unsigned long long *array)
825 {
825 - if (ctx == NULL)
826 + if (context == NULL)
827 return;
828 + struct rrdengine_instance* ctx = (struct rrdengine_instance*) context;
829
830 struct page_cache *pg_cache = &ctx->pg_cache;
831
@@ -874,19 +876,23 @@ void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle)
876 pg_cache_put(ctx, (struct rrdeng_page_descr *)handle);
877 }
878
877 -/*
878 - * Returns 0 on success, negative on error
879 - */
880 -int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
881 - unsigned disk_space_mb)
879 +
880 +STORAGE_ENGINE_INSTANCE*
881 +rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host)
882 {
883 struct rrdengine_instance *ctx;
884 int error;
885 - uint32_t max_open_files;
885
887 - max_open_files = rlimit_nofile.rlim_cur / 4;
886 + bool is_legacy = is_legacy_child(host->machine_guid);
887 + if (!is_legacy && eng->context) {
888 + if (host->rrd_memory_mode == eng->id && host->rrdeng_ctx == NULL) {
889 + host->rrdeng_ctx = eng->context;
890 + }
891 + return eng->context;
892 + }
893
894 /* reserve RRDENG_FD_BUDGET_PER_INSTANCE file descriptors for this instance */
895 + uint32_t max_open_files = rlimit_nofile.rlim_cur / 4;
896 rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, RRDENG_FD_BUDGET_PER_INSTANCE);
897 if (rrdeng_reserved_file_descriptors > max_open_files) {
898 error(
@@ -895,15 +901,18 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
901
902 rrd_stat_atomic_add(&global_fs_errors, 1);
903 rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
898 - return UV_EMFILE;
904 + return NULL;//UV_EMFILE;
905 }
906 + char dbfiles_path[FILENAME_MAX + 1];
907
901 - if (NULL == ctxp) {
902 - ctx = &multidb_ctx;
903 - memset(ctx, 0, sizeof(*ctx));
904 - } else {
905 - *ctxp = ctx = callocz(1, sizeof(*ctx));
906 - }
908 + snprintfz(dbfiles_path, FILENAME_MAX, "%s/dbengine", host->cache_dir);
909 + mkdir(dbfiles_path, 0775);
910 +
911 + int page_cache_mb = default_rrdeng_page_cache_mb;
912 + int disk_space_mb = default_rrdeng_disk_quota_mb;
913 +
914 + ctx = callocz(1, sizeof(*ctx));
915 + ctx->parent.engine = eng;
916 ctx->global_compress_alg = RRD_LZ4;
917 if (page_cache_mb < RRDENG_MIN_PAGE_CACHE_SIZE_MB)
918 page_cache_mb = RRDENG_MIN_PAGE_CACHE_SIZE_MB;
@@ -926,6 +935,15 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
935 ctx->metalog_ctx = NULL; /* only set this after the metadata log has finished initializing */
936 ctx->host = host;
937
938 + // Attach context as the global context
939 + if (!is_legacy && !eng->context) {
940 + eng->context = (STORAGE_ENGINE_INSTANCE *)ctx;
941 + }
942 + // Attach context as the host context
943 + if (host->rrd_memory_mode == eng->id && host->rrdeng_ctx == NULL) {
944 + host->rrdeng_ctx = eng->context;
945 + }
946 +
947 memset(&ctx->worker_config, 0, sizeof(ctx->worker_config));
948 ctx->worker_config.ctx = ctx;
949 init_page_cache(ctx);
@@ -950,30 +968,30 @@ int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_p
968 goto error_after_rrdeng_worker;
969 }
970
953 - return 0;
971 + return (STORAGE_ENGINE_INSTANCE *)ctx;
972
973 error_after_rrdeng_worker:
974 finalize_rrd_files(ctx);
975 error_after_init_rrd_files:
976 free_page_cache(ctx);
959 - if (ctx != &multidb_ctx) {
977 + if ((STORAGE_ENGINE_INSTANCE *)ctx != eng->context) {
978 freez(ctx);
961 - *ctxp = NULL;
979 }
980 rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
964 - return UV_EIO;
981 + return NULL;//UV_EIO;
982 }
983
984 /*
985 * Returns 0 on success, 1 on error
986 */
970 -int rrdeng_exit(struct rrdengine_instance *ctx)
987 +void rrdeng_exit(STORAGE_ENGINE_INSTANCE* context)
988 {
989 struct rrdeng_cmd cmd;
990
974 - if (NULL == ctx) {
975 - return 1;
991 + if (NULL == context) {
992 + return;
993 }
994 + struct rrdengine_instance* ctx = (struct rrdengine_instance*)context;
995
996 /* TODO: add page to page cache */
997 cmd.opcode = RRDENG_SHUTDOWN;
@@ -984,21 +1002,18 @@ int rrdeng_exit(struct rrdengine_instance *ctx)
1002 finalize_rrd_files(ctx);
1003 //metalog_exit(ctx->metalog_ctx);
1004 free_page_cache(ctx);
987 -
988 - if (ctx != &multidb_ctx) {
989 - freez(ctx);
990 - }
1005 + freez(ctx);
1006 rrd_stat_atomic_add(&rrdeng_reserved_file_descriptors, -RRDENG_FD_BUDGET_PER_INSTANCE);
992 - return 0;
1007 }
1008
995 -void rrdeng_prepare_exit(struct rrdengine_instance *ctx)
1009 +void rrdeng_prepare_exit(STORAGE_ENGINE_INSTANCE* context)
1010 {
1011 struct rrdeng_cmd cmd;
1012
999 - if (NULL == ctx) {
1013 + if (NULL == context) {
1014 return;
1015 }
1016 + struct rrdengine_instance* ctx = (struct rrdengine_instance*)context;
1017
1018 completion_init(&ctx->rrdengine_completion);
1019 cmd.opcode = RRDENG_QUIESCE;
database/engine/rrdengineapi.h
+4 -13
@@ -17,7 +17,6 @@ extern int default_rrdeng_page_cache_mb;
17 extern int default_rrdeng_disk_quota_mb;
18 extern int default_multidb_disk_quota_mb;
19 extern uint8_t rrdeng_drop_metrics_under_page_cache_pressure;
20 -extern struct rrdengine_instance multidb_ctx;
20
21 struct rrdeng_region_info {
22 time_t start_time;
@@ -25,13 +24,6 @@ struct rrdeng_region_info {
24 unsigned points;
25 };
26
28 -extern void *rrdeng_create_page(struct rrdengine_instance *ctx, uuid_t *id, struct rrdeng_page_descr **ret_descr);
29 -extern void rrdeng_commit_page(struct rrdengine_instance *ctx, struct rrdeng_page_descr *descr,
30 - Word_t page_correlation_id);
31 -extern void *rrdeng_get_latest_page(struct rrdengine_instance *ctx, uuid_t *id, void **handle);
32 -extern void *rrdeng_get_page(struct rrdengine_instance *ctx, uuid_t *id, usec_t point_in_time, void **handle);
33 -extern void rrdeng_put_page(struct rrdengine_instance *ctx, void *handle);
34 -
27 extern void rrdeng_generate_legacy_uuid(const char *dim_id, char *chart_id, uuid_t *ret_uuid);
28 extern void rrdeng_convert_legacy_uuid_to_multihost(char machine_guid[GUID_LEN + 1], uuid_t *legacy_uuid,
29 uuid_t *ret_uuid);
@@ -52,14 +44,13 @@ extern int rrdeng_load_metric_is_finished(struct rrddim_query_handle *rrdimm_han
44 extern void rrdeng_load_metric_finalize(struct rrddim_query_handle *rrdimm_handle);
45 extern time_t rrdeng_metric_latest_time(RRDDIM *rd);
46 extern time_t rrdeng_metric_oldest_time(RRDDIM *rd);
55 -extern void rrdeng_get_37_statistics(struct rrdengine_instance *ctx, unsigned long long *array);
47 +extern void rrdeng_get_37_statistics(STORAGE_ENGINE_INSTANCE *ctx, unsigned long long *array);
48
49 /* must call once before using anything */
58 -extern int rrdeng_init(RRDHOST *host, struct rrdengine_instance **ctxp, char *dbfiles_path, unsigned page_cache_mb,
59 - unsigned disk_space_mb);
50 +extern STORAGE_ENGINE_INSTANCE* rrdeng_init(STORAGE_ENGINE* eng, RRDHOST *host);
51
61 -extern int rrdeng_exit(struct rrdengine_instance *ctx);
62 -extern void rrdeng_prepare_exit(struct rrdengine_instance *ctx);
52 +extern void rrdeng_exit(STORAGE_ENGINE_INSTANCE *ctx);
53 +extern void rrdeng_prepare_exit(STORAGE_ENGINE_INSTANCE *ctx);
54 extern int rrdeng_metric_latest_time_by_uuid(uuid_t *dim_uuid, time_t *first_entry_t, time_t *last_entry_t);
55
56 #endif /* NETDATA_RRDENGINEAPI_H */
database/ram/rrddim_mem.c
+14
@@ -1,6 +1,20 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "rrddim_mem.h"
4 +#include "../storage_engine.h"
5 +
6 +STORAGE_ENGINE_INSTANCE* rrddim_storage_engine_instance_new(STORAGE_ENGINE* engine, RRDHOST *host) {
7 + (void)engine; (void)host;
8 + return NULL;
9 +}
10 +
11 +void rrddim_storage_engine_instance_exit(STORAGE_ENGINE_INSTANCE* context) {
12 + (void)context;
13 +}
14 +
15 +void rrddim_storage_engine_instance_destroy(STORAGE_ENGINE_INSTANCE* context) {
16 + (void)context;
17 +}
18
19 // ----------------------------------------------------------------------------
20 // RRDDIM legacy data collection functions
database/ram/rrddim_mem.h
+4
@@ -15,6 +15,10 @@ struct mem_query_handle {
15 uint8_t finished;
16 };
17
18 +STORAGE_ENGINE_INSTANCE* rrddim_storage_engine_instance_new(STORAGE_ENGINE* engine, RRDHOST *host);
19 +void rrddim_storage_engine_instance_exit(STORAGE_ENGINE_INSTANCE* context);
20 +void rrddim_storage_engine_instance_destroy(STORAGE_ENGINE_INSTANCE* context);
21 +
22 extern void rrddim_collect_init(RRDDIM *rd);
23 extern void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number);
24 extern int rrddim_collect_finalize(RRDDIM *rd);
database/rrd.c
-21
@@ -28,31 +28,10 @@ int gap_when_lost_iterations_above = 1;
28 // RRD - memory modes
29
30 inline const char *rrd_memory_mode_name(RRD_MEMORY_MODE id) {
31 - switch(id) {
32 - case RRD_MEMORY_MODE_RAM:
33 - return RRD_MEMORY_MODE_RAM_NAME;
34 -
35 - case RRD_MEMORY_MODE_MAP:
36 - return RRD_MEMORY_MODE_MAP_NAME;
37 -
38 - case RRD_MEMORY_MODE_NONE:
39 - return RRD_MEMORY_MODE_NONE_NAME;
40 -
41 - case RRD_MEMORY_MODE_SAVE:
42 - return RRD_MEMORY_MODE_SAVE_NAME;
43 -
44 - case RRD_MEMORY_MODE_ALLOC:
45 - return RRD_MEMORY_MODE_ALLOC_NAME;
46 -
47 - case RRD_MEMORY_MODE_DBENGINE:
48 - return RRD_MEMORY_MODE_DBENGINE_NAME;
49 - }
50 -
31 STORAGE_ENGINE* eng = storage_engine_get(id);
32 if (eng) {
33 return eng->name;
34 }
55 -
35 return RRD_MEMORY_MODE_SAVE_NAME;
36 }
37
database/rrd.h
+3 -6
@@ -18,6 +18,8 @@ typedef struct rrdcalc RRDCALC;
18 typedef struct rrdcalctemplate RRDCALCTEMPLATE;
19 typedef struct alarm_entry ALARM_ENTRY;
20 typedef struct context_param CONTEXT_PARAM;
21 +typedef struct storage_engine_instance STORAGE_ENGINE_INSTANCE;
22 +typedef struct storage_engine STORAGE_ENGINE;
23
24 typedef void *ml_host_t;
25 typedef void *ml_dimension_t;
@@ -854,9 +856,7 @@ struct rrdhost {
856 avl_tree_lock rrdfamily_root_index; // the host's chart families index
857 avl_tree_lock rrdvar_root_index; // the host's chart variables index
858
857 -#ifdef ENABLE_DBENGINE
858 - struct rrdengine_instance *rrdeng_ctx; // DB engine instance for this host
859 -#endif
859 + STORAGE_ENGINE_INSTANCE *rrdeng_ctx; // DB engine instance for this host
860 uuid_t host_uuid; // Global GUID for this host
861 uuid_t *node_id; // Cloud node_id
862
@@ -1324,9 +1324,6 @@ extern void set_host_properties(
1324 // ----------------------------------------------------------------------------
1325 // RRD DB engine declarations
1326
1327 -#ifdef ENABLE_DBENGINE
1328 -#include "database/engine/rrdengineapi.h"
1329 -#endif
1327 #include "sqlite/sqlite_functions.h"
1328 #include "sqlite/sqlite_aclk.h"
1329 #include "sqlite/sqlite_aclk_chart.h"
database/rrdhost.c
+19 -70
@@ -2,6 +2,10 @@
2
3 #define NETDATA_RRD_INTERNALS
4 #include "rrd.h"
5 +#include "storage_engine.h"
6 +#ifdef ENABLE_DBENGINE
7 +#include "engine/rrdenginelib.h"
8 +#endif
9
10 RRDHOST *localhost = NULL;
11 size_t rrd_hosts_available = 0;
@@ -333,41 +337,8 @@ RRDHOST *rrdhost_create(const char *hostname,
337 else
338 error_report("Host machine GUID %s is not valid", host->machine_guid);
339
336 - if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
337 -#ifdef ENABLE_DBENGINE
338 - char dbenginepath[FILENAME_MAX + 1];
339 - int ret;
340 -
341 - snprintfz(dbenginepath, FILENAME_MAX, "%s/dbengine", host->cache_dir);
342 - ret = mkdir(dbenginepath, 0775);
343 - if (ret != 0 && errno != EEXIST)
344 - error("Host '%s': cannot create directory '%s'", host->hostname, dbenginepath);
345 - else ret = 0; // succeed
346 - if (is_legacy) // initialize legacy dbengine instance as needed
347 - ret = rrdeng_init(host, &host->rrdeng_ctx, dbenginepath, default_rrdeng_page_cache_mb,
348 - default_rrdeng_disk_quota_mb); // may fail here for legacy dbengine initialization
349 - else
350 - host->rrdeng_ctx = &multidb_ctx;
351 - if (ret) { // check legacy or multihost initialization success
352 - error(
353 - "Host '%s': cannot initialize host with machine guid '%s'. Failed to initialize DB engine at '%s'.",
354 - host->hostname, host->machine_guid, host->cache_dir);
355 - rrdhost_free(host);
356 - host = NULL;
357 - //rrd_hosts_available++; //TODO: maybe we want this?
358 -
359 - return host;
360 - }
361 -
362 -#else
363 - fatal("RRD_MEMORY_MODE_DBENGINE is not supported in this platform.");
364 -#endif
365 - }
366 - else {
367 -#ifdef ENABLE_DBENGINE
368 - host->rrdeng_ctx = &multidb_ctx;
369 -#endif
370 - }
340 + // Create engine
341 + host->rrdeng_ctx = storage_engine_new(storage_engine_get(host->rrd_memory_mode), host);
342
343 // ------------------------------------------------------------------------
344 // link it and add it to the index
@@ -670,10 +641,8 @@ restart_after_removal:
641 info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", host->hostname, host->machine_guid);
642
643 if (rrdhost_flag_check(host, RRDHOST_FLAG_DELETE_ORPHAN_HOST)
673 -#ifdef ENABLE_DBENGINE
644 /* don't delete multi-host DB host files */
675 - && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx == &multidb_ctx)
676 -#endif
645 + && !(host->rrdeng_ctx->engine && host->rrdeng_ctx->engine->context == host->rrdeng_ctx)
646 )
647 rrdhost_delete_charts(host);
648 else
@@ -740,25 +709,6 @@ int rrd_init(char *hostname, struct rrdhost_system_info *system_info) {
709 return 1;
710 }
711
743 -#ifdef ENABLE_DBENGINE
744 - char dbenginepath[FILENAME_MAX + 1];
745 - int ret;
746 - snprintfz(dbenginepath, FILENAME_MAX, "%s/dbengine", localhost->cache_dir);
747 - ret = mkdir(dbenginepath, 0775);
748 - if (ret != 0 && errno != EEXIST)
749 - error("Host '%s': cannot create directory '%s'", localhost->hostname, dbenginepath);
750 - else // Unconditionally create multihost db to support on demand host creation
751 - ret = rrdeng_init(NULL, NULL, dbenginepath, default_rrdeng_page_cache_mb, default_multidb_disk_quota_mb);
752 - if (ret) {
753 - error(
754 - "Host '%s' with machine guid '%s' failed to initialize multi-host DB engine instance at '%s'.",
755 - localhost->hostname, localhost->machine_guid, localhost->cache_dir);
756 - rrdhost_free(localhost);
757 - localhost = NULL;
758 - rrd_unlock();
759 - fatal("Failed to initialize dbengine");
760 - }
761 -#endif
712 sql_aclk_sync_init();
713 rrd_unlock();
714
@@ -903,12 +853,12 @@ void rrdhost_free(RRDHOST *host) {
853 // ------------------------------------------------------------------------
854 // release its children resources
855
906 -#ifdef ENABLE_DBENGINE
907 - if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
908 - if (host->rrdeng_ctx != &multidb_ctx)
909 - rrdeng_prepare_exit(host->rrdeng_ctx);
856 + STORAGE_ENGINE* eng = host->rrdeng_ctx->engine;
857 + if (host->rrdeng_ctx != eng->context) {
858 + if (eng && eng->api.engine_ops.exit)
859 + eng->api.engine_ops.exit(host->rrdeng_ctx);
860 }
911 -#endif
861 +
862 while(host->rrdset_root)
863 rrdset_free(host->rrdset_root);
864
@@ -939,10 +889,11 @@ void rrdhost_free(RRDHOST *host) {
889
890 health_alarm_log_free(host);
891
942 -#ifdef ENABLE_DBENGINE
943 - if (host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx != &multidb_ctx)
944 - rrdeng_exit(host->rrdeng_ctx);
945 -#endif
892 + if (host->rrdeng_ctx != eng->context) {
893 + if (eng)
894 + eng->api.engine_ops.destroy(host->rrdeng_ctx);
895 + host->rrdeng_ctx = NULL;
896 + }
897
898 // ------------------------------------------------------------------------
899 // remove it from the indexes
@@ -1258,10 +1209,8 @@ void rrdhost_cleanup_all(void) {
1209 RRDHOST *host;
1210 rrdhost_foreach_read(host) {
1211 if (host != localhost && rrdhost_flag_check(host, RRDHOST_FLAG_DELETE_ORPHAN_HOST) && !host->receiver
1261 -#ifdef ENABLE_DBENGINE
1212 /* don't delete multi-host DB host files */
1263 - && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && host->rrdeng_ctx == &multidb_ctx)
1264 -#endif
1213 + && !(host->rrdeng_ctx->engine && host->rrdeng_ctx->engine->context == host->rrdeng_ctx)
1214 )
1215 rrdhost_delete_charts(host);
1216 else
@@ -1344,7 +1293,7 @@ restart_after_removal:
1293 rrdvar_free_remaining_variables(host, &st->rrdvar_root_index);
1294
1295 rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
1347 -
1296 +
1297 if (st->dimensions) {
1298 /* If the chart still has dimensions don't delete it from the metadata log */
1299 continue;
database/rrdset.c
+4
@@ -3,6 +3,10 @@
3 #define NETDATA_RRD_INTERNALS
4 #include "rrd.h"
5 #include <sched.h>
6 +#ifdef ENABLE_DBENGINE
7 +#include "database/engine/rrddiskprotocol.h"
8 +#include "database/engine/rrdengineapi.h"
9 +#endif
10
11 void __rrdset_check_rdlock(RRDSET *st, const char *file, const char *function, const unsigned long line) {
12 debug(D_RRD_CALLS, "Checking read lock on chart '%s'", st->id);
database/sqlite/sqlite_aclk_chart.c
+3
@@ -2,6 +2,9 @@
2
3 #include "sqlite_functions.h"
4 #include "sqlite_aclk_chart.h"
5 +#ifdef ENABLE_DBENGINE
6 +#include "../engine/rrdengineapi.h"
7 +#endif
8
9 #if defined(ENABLE_ACLK) && defined(ENABLE_NEW_CLOUD_PROTOCOL)
10 #include "../../aclk/aclk_charts_api.h"
database/sqlite/sqlite_functions.c
+8 -4
@@ -1,6 +1,10 @@
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "sqlite_functions.h"
4 +#ifdef ENABLE_DBENGINE
5 +#include "../engine/rrdengineapi.h"
6 +#endif
7 +#include "database/storage_engine.h"
8
9 #define DB_METADATA_VERSION "1"
10
@@ -423,7 +427,7 @@ int sql_init_database(db_check_action_type_t rebuild, int memory)
427 // PRAGMA temp_store = 0 | DEFAULT | 1 | FILE | 2 | MEMORY;
428 snprintfz(buf, 1024, "PRAGMA temp_store=%s;", config_get(CONFIG_SECTION_SQLITE, "temp store", "MEMORY"));
429 if(init_database_batch(rebuild, 0, list)) return 1;
426 -
430 +
431 // https://www.sqlite.org/pragma.html#pragma_journal_size_limit
432 // PRAGMA schema.journal_size_limit = N ;
433 snprintfz(buf, 1024, "PRAGMA journal_size_limit=%lld;", config_get_number(CONFIG_SECTION_SQLITE, "journal size limit", 16777216));
@@ -1278,9 +1282,9 @@ RRDHOST *sql_create_host_by_uuid(char *hostname)
1282
1283 host->system_info = callocz(1, sizeof(*host->system_info));;
1284 rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED);
1281 -#ifdef ENABLE_DBENGINE
1282 - host->rrdeng_ctx = &multidb_ctx;
1283 -#endif
1285 +
1286 + // Create multidb engine instance if necessary
1287 + host->rrdeng_ctx = storage_engine_new(storage_engine_get(RRD_MEMORY_MODE_DBENGINE), host);
1288
1289 failed:
1290 rc = sqlite3_finalize(res);
database/storage_engine.c
+147 -47
@@ -5,68 +5,144 @@
5 #ifdef ENABLE_DBENGINE
6 #include "engine/rrdengineapi.h"
7 #endif
8 -
9 -#define im_collect_ops { \
10 - .init = rrddim_collect_init,\
11 - .store_metric = rrddim_collect_store_metric,\
12 - .finalize = rrddim_collect_finalize\
13 -}
14 -
15 -#define im_query_ops { \
16 - .init = rrddim_query_init, \
17 - .next_metric = rrddim_query_next_metric, \
18 - .is_finished = rrddim_query_is_finished, \
19 - .finalize = rrddim_query_finalize, \
20 - .latest_time = rrddim_query_latest_time, \
21 - .oldest_time = rrddim_query_oldest_time \
22 -}
8 +#include "libnetdata/libnetdata.h"
9
10 static STORAGE_ENGINE engines[] = {
11 {
12 .id = RRD_MEMORY_MODE_NONE,
13 .name = RRD_MEMORY_MODE_NONE_NAME,
14 .api = {
29 - .collect_ops = im_collect_ops,
30 - .query_ops = im_query_ops
31 - }
15 + .engine_ops = {
16 + .create = rrddim_storage_engine_instance_new,
17 + .exit = rrddim_storage_engine_instance_exit,
18 + .destroy = rrddim_storage_engine_instance_destroy
19 + },
20 + .collect_ops = {
21 + .init = rrddim_collect_init,
22 + .store_metric = rrddim_collect_store_metric,
23 + .finalize = rrddim_collect_finalize
24 + },
25 + .query_ops = {
26 + .init = rrddim_query_init,
27 + .next_metric = rrddim_query_next_metric,
28 + .is_finished = rrddim_query_is_finished,
29 + .finalize = rrddim_query_finalize,
30 + .latest_time = rrddim_query_latest_time,
31 + .oldest_time = rrddim_query_oldest_time
32 + }
33 + },
34 + .context = NULL
35 },
36 {
37 .id = RRD_MEMORY_MODE_RAM,
38 .name = RRD_MEMORY_MODE_RAM_NAME,
39 .api = {
37 - .collect_ops = im_collect_ops,
38 - .query_ops = im_query_ops
39 - }
40 + .engine_ops = {
41 + .create = rrddim_storage_engine_instance_new,
42 + .exit = rrddim_storage_engine_instance_exit,
43 + .destroy = rrddim_storage_engine_instance_destroy
44 + },
45 + .collect_ops = {
46 + .init = rrddim_collect_init,
47 + .store_metric = rrddim_collect_store_metric,
48 + .finalize = rrddim_collect_finalize
49 + },
50 + .query_ops = {
51 + .init = rrddim_query_init,
52 + .next_metric = rrddim_query_next_metric,
53 + .is_finished = rrddim_query_is_finished,
54 + .finalize = rrddim_query_finalize,
55 + .latest_time = rrddim_query_latest_time,
56 + .oldest_time = rrddim_query_oldest_time
57 + }
58 + },
59 + .context = NULL
60 },
61 {
62 .id = RRD_MEMORY_MODE_MAP,
63 .name = RRD_MEMORY_MODE_MAP_NAME,
64 .api = {
45 - .collect_ops = im_collect_ops,
46 - .query_ops = im_query_ops
47 - }
65 + .engine_ops = {
66 + .create = rrddim_storage_engine_instance_new,
67 + .exit = rrddim_storage_engine_instance_exit,
68 + .destroy = rrddim_storage_engine_instance_destroy
69 + },
70 + .collect_ops = {
71 + .init = rrddim_collect_init,
72 + .store_metric = rrddim_collect_store_metric,
73 + .finalize = rrddim_collect_finalize
74 + },
75 + .query_ops = {
76 + .init = rrddim_query_init,
77 + .next_metric = rrddim_query_next_metric,
78 + .is_finished = rrddim_query_is_finished,
79 + .finalize = rrddim_query_finalize,
80 + .latest_time = rrddim_query_latest_time,
81 + .oldest_time = rrddim_query_oldest_time
82 + }
83 + },
84 + .context = NULL
85 },
86 {
87 .id = RRD_MEMORY_MODE_SAVE,
88 .name = RRD_MEMORY_MODE_SAVE_NAME,
89 .api = {
53 - .collect_ops = im_collect_ops,
54 - .query_ops = im_query_ops
55 - }
90 + .engine_ops = {
91 + .create = rrddim_storage_engine_instance_new,
92 + .exit = rrddim_storage_engine_instance_exit,
93 + .destroy = rrddim_storage_engine_instance_destroy
94 + },
95 + .collect_ops = {
96 + .init = rrddim_collect_init,
97 + .store_metric = rrddim_collect_store_metric,
98 + .finalize = rrddim_collect_finalize
99 + },
100 + .query_ops = {
101 + .init = rrddim_query_init,
102 + .next_metric = rrddim_query_next_metric,
103 + .is_finished = rrddim_query_is_finished,
104 + .finalize = rrddim_query_finalize,
105 + .latest_time = rrddim_query_latest_time,
106 + .oldest_time = rrddim_query_oldest_time
107 + }
108 + },
109 + .context = NULL
110 },
111 {
112 .id = RRD_MEMORY_MODE_ALLOC,
113 .name = RRD_MEMORY_MODE_ALLOC_NAME,
114 .api = {
61 - .collect_ops = im_collect_ops,
62 - .query_ops = im_query_ops
63 - }
115 + .engine_ops = {
116 + .create = rrddim_storage_engine_instance_new,
117 + .exit = rrddim_storage_engine_instance_exit,
118 + .destroy = rrddim_storage_engine_instance_destroy
119 + },
120 + .collect_ops = {
121 + .init = rrddim_collect_init,
122 + .store_metric = rrddim_collect_store_metric,
123 + .finalize = rrddim_collect_finalize
124 + },
125 + .query_ops = {
126 + .init = rrddim_query_init,
127 + .next_metric = rrddim_query_next_metric,
128 + .is_finished = rrddim_query_is_finished,
129 + .finalize = rrddim_query_finalize,
130 + .latest_time = rrddim_query_latest_time,
131 + .oldest_time = rrddim_query_oldest_time
132 + }
133 + },
134 + .context = NULL
135 },
136 #ifdef ENABLE_DBENGINE
137 {
138 .id = RRD_MEMORY_MODE_DBENGINE,
139 .name = RRD_MEMORY_MODE_DBENGINE_NAME,
140 .api = {
141 + .engine_ops = {
142 + .create = rrdeng_init,
143 + .exit = rrdeng_prepare_exit,
144 + .destroy = rrdeng_exit
145 + },
146 .collect_ops = {
147 .init = rrdeng_store_metric_init,
148 .store_metric = rrdeng_store_metric_next,
@@ -80,28 +156,34 @@ static STORAGE_ENGINE engines[] = {
156 .latest_time = rrdeng_metric_latest_time,
157 .oldest_time = rrdeng_metric_oldest_time
158 }
83 - }
84 - },
159 + },
160 + .context = NULL
161 + }
162 #endif
86 - { .id = RRD_MEMORY_MODE_NONE, .name = NULL }
163 };
164
165 +const size_t engine_count = sizeof(engines) / sizeof(engines[0]);
166 +
167 STORAGE_ENGINE* storage_engine_find(const char* name)
168 {
91 - for (STORAGE_ENGINE* it = engines; it->name; it++) {
92 - if (strcmp(it->name, name) == 0)
93 - return it;
94 - }
95 - return NULL;
169 + for (size_t i = 0; i != engine_count; i++)
170 + if (strcmp(engines[i].name, name) == 0)
171 + return &engines[i];
172 +
173 + error("No storage engine found for memory mode %s.", name);
174 + return storage_engine_get(default_rrd_memory_mode);
175 }
176
177 STORAGE_ENGINE* storage_engine_get(RRD_MEMORY_MODE mmode)
178 {
100 - for (STORAGE_ENGINE* it = engines; it->name; it++) {
101 - if (it->id == mmode)
102 - return it;
103 - }
104 - return NULL;
179 + for (size_t i = 0; i != engine_count; i++)
180 + if (engines[i].id == mmode)
181 + return &engines[i];
182 + STORAGE_ENGINE* eng = mmode == default_rrd_memory_mode
183 + ? &engines[0] // default engine not available, use NONE
184 + : storage_engine_get(default_rrd_memory_mode);
185 + error("No storage engine for memory mode %u, will use %s (%u) instead.", mmode, eng->name, eng->id);
186 + return eng;
187 }
188
189 STORAGE_ENGINE* storage_engine_foreach_init()
@@ -112,9 +194,27 @@ STORAGE_ENGINE* storage_engine_foreach_init()
194
195 STORAGE_ENGINE* storage_engine_foreach_next(STORAGE_ENGINE* it)
196 {
115 - if (!it || !it->name)
116 - return NULL;
117 -
197 it++;
119 - return it->name ? it : NULL;
198 + return (it >= &engines[engine_count]) ? NULL : it;
199 +}
200 +
201 +STORAGE_ENGINE_INSTANCE* storage_engine_new(STORAGE_ENGINE* eng, RRDHOST *host)
202 +{
203 + STORAGE_ENGINE_INSTANCE* instance = host->rrdeng_ctx;
204 + if (!instance && eng) {
205 + instance = eng->api.engine_ops.create(eng, host);
206 + if (instance) {
207 + instance->engine = eng;
208 + }
209 + }
210 + return instance;
211 +}
212 +
213 +void storage_engine_delete(STORAGE_ENGINE_INSTANCE* instance) {
214 + if (instance) {
215 + STORAGE_ENGINE* eng = instance->engine;
216 + if (eng) {
217 + eng->api.engine_ops.destroy(instance);
218 + }
219 + }
220 }
database/storage_engine.h
+21
@@ -6,10 +6,20 @@
6 #include "rrd.h"
7
8 typedef struct storage_engine STORAGE_ENGINE;
9 +typedef struct storage_engine_instance STORAGE_ENGINE_INSTANCE;
10 +
11 +// ------------------------------------------------------------------------
12 +// function pointers that handle storage engine instance creation and destruction
13 +struct storage_engine_ops {
14 + STORAGE_ENGINE_INSTANCE*(*create)(STORAGE_ENGINE* engine, RRDHOST *host);
15 + void(*exit)(STORAGE_ENGINE_INSTANCE*);
16 + void(*destroy)(STORAGE_ENGINE_INSTANCE*);
17 +};
18
19 // ------------------------------------------------------------------------
20 // function pointers for all APIs provided by a storge engine
21 typedef struct storage_engine_api {
22 + struct storage_engine_ops engine_ops;
23 struct rrddim_collect_ops collect_ops;
24 struct rrddim_query_ops query_ops;
25 } STORAGE_ENGINE_API;
@@ -18,6 +28,12 @@ struct storage_engine {
28 RRD_MEMORY_MODE id;
29 const char* name;
30 STORAGE_ENGINE_API api;
31 + STORAGE_ENGINE_INSTANCE* context;
32 +};
33 +
34 +// Abstract structure to be extended by implementations
35 +struct storage_engine_instance {
36 + STORAGE_ENGINE* engine;
37 };
38
39 extern STORAGE_ENGINE* storage_engine_get(RRD_MEMORY_MODE mmode);
@@ -27,4 +43,9 @@ extern STORAGE_ENGINE* storage_engine_find(const char* name);
43 extern STORAGE_ENGINE* storage_engine_foreach_init();
44 extern STORAGE_ENGINE* storage_engine_foreach_next(STORAGE_ENGINE* it);
45
46 +// ------------------------------------------------------------------------
47 +// Retreive or create a storage engine instance for host
48 +STORAGE_ENGINE_INSTANCE* storage_engine_new(STORAGE_ENGINE* engine, RRDHOST *host);
49 +void storage_engine_delete(STORAGE_ENGINE_INSTANCE* engine);
50 +
51 #endif
web/api/queries/query.c
+3
@@ -4,6 +4,9 @@
4 #include "web/api/formatters/rrd2json.h"
5 #include "rrdr.h"
6 #include "database/ram/rrddim_mem.h"
7 +#ifdef ENABLE_DBENGINE
8 +#include "database/engine/rrdengine.h"
9 +#endif
10
11 #include "average/average.h"
12 #include "incremental_sum/incremental_sum.h"