sqlite_metadata.c · cryptotaxi247/netdata-1

Code Commits Issues Pulls Sessions CI/CD

master

c 3,135 lines 110 KB

Raw

1	// SPDX-License-Identifier: GPL-3.0-or-later
2
3	#include "sqlite_metadata.h"
4	#include "database/sqlite/vendored/sqlite3recover.h"
5	#include "health/health-alert-entry.h"
6
7	#include "sqlite_db_migration.h"
8
9	#define DB_METADATA_VERSION 18
10
11	#define COMPUTE_DURATION(var_name, unit, start, end) \
12	char var_name[64]; \
13	duration_snprintf(var_name, sizeof(var_name), \
14	(int64_t)((end) - (start)), unit, true)
15
16	#define SHUTDOWN_REQUESTED(config) (__atomic_load_n(&(config)->shutdown_requested, __ATOMIC_RELAXED))
17
18	extern long long def_journal_size_limit;
19
20	const char *database_config[] = {
21	"CREATE TABLE IF NOT EXISTS host(host_id BLOB PRIMARY KEY, hostname TEXT NOT NULL, "
22	"registry_hostname TEXT NOT NULL default 'unknown', update_every INT NOT NULL default 1, "
23	"os TEXT NOT NULL default 'unknown', timezone TEXT NOT NULL default 'unknown', tags TEXT NOT NULL default '',"
24	"hops INT NOT NULL DEFAULT 0,"
25	"memory_mode INT DEFAULT 0, abbrev_timezone TEXT DEFAULT '', utc_offset INT NOT NULL DEFAULT 0,"
26	"program_name TEXT NOT NULL DEFAULT 'unknown', program_version TEXT NOT NULL DEFAULT 'unknown', "
27	"entries INT NOT NULL DEFAULT 0,"
28	"health_enabled INT NOT NULL DEFAULT 0, last_connected INT NOT NULL DEFAULT 0)",
29
30	"CREATE TABLE IF NOT EXISTS chart(chart_id blob PRIMARY KEY, host_id blob, type text, id text, name text, "
31	"family text, context text, title text, unit text, plugin text, module text, priority int, update_every int, "
32	"chart_type int, memory_mode int, history_entries)",
33
34	"CREATE TABLE IF NOT EXISTS dimension(dim_id blob PRIMARY KEY, chart_id blob, id text, name text, "
35	"multiplier int, divisor int , algorithm int, options text)",
36
37	"CREATE TABLE IF NOT EXISTS metadata_migration(filename text, file_size, date_created int)",
38
39	"CREATE TABLE IF NOT EXISTS chart_label(chart_id blob, source_type int, label_key text, "
40	"label_value text, date_created int, PRIMARY KEY (chart_id, label_key))",
41
42	"CREATE TRIGGER IF NOT EXISTS del_chart_label AFTER DELETE ON chart "
43	"BEGIN DELETE FROM chart_label WHERE chart_id = old.chart_id; END",
44
45	"CREATE TRIGGER IF NOT EXISTS del_chart "
46	"AFTER DELETE ON dimension "
47	"FOR EACH ROW "
48	"BEGIN"
49	" DELETE FROM chart WHERE chart_id = OLD.chart_id "
50	" AND NOT EXISTS (SELECT 1 FROM dimension WHERE chart_id = OLD.chart_id);"
51	"END",
52
53	"CREATE TABLE IF NOT EXISTS node_instance (host_id blob PRIMARY KEY, claim_id, node_id, date_created)",
54
55	"CREATE TABLE IF NOT EXISTS alert_hash(hash_id blob PRIMARY KEY, date_updated int, alarm text, template text, "
56	"on_key text, class text, component text, type text, os text, hosts text, lookup text, "
57	"every text, units text, calc text, families text, plugin text, module text, charts text, green text, "
58	"red text, warn text, crit text, exec text, to_key text, info text, delay text, options text, "
59	"repeat text, host_labels text, p_db_lookup_dimensions text, p_db_lookup_method text, p_db_lookup_options int, "
60	"p_db_lookup_after int, p_db_lookup_before int, p_update_every int, source text, chart_labels text, "
61	"summary text, time_group_condition INT, time_group_value DOUBLE, dims_group INT, data_source INT)",
62
63	"CREATE TABLE IF NOT EXISTS host_info(host_id blob, system_key text NOT NULL, system_value text NOT NULL, "
64	"date_created INT, PRIMARY KEY(host_id, system_key))",
65
66	"CREATE TABLE IF NOT EXISTS host_label(host_id blob, source_type int, label_key text NOT NULL, "
67	"label_value text NOT NULL, date_created INT, PRIMARY KEY (host_id, label_key))",
68
69	"CREATE TRIGGER IF NOT EXISTS ins_host AFTER INSERT ON host BEGIN INSERT INTO node_instance (host_id, date_created)"
70	" SELECT new.host_id, unixepoch() WHERE new.host_id NOT IN (SELECT host_id FROM node_instance); END",
71
72	"CREATE TABLE IF NOT EXISTS health_log (health_log_id INTEGER PRIMARY KEY, host_id blob, alarm_id int, "
73	"config_hash_id blob, name text, chart text, family text, recipient text, units text, exec text, "
74	"chart_context text, last_transition_id blob, chart_name text, UNIQUE (host_id, alarm_id))",
75
76	"CREATE TABLE IF NOT EXISTS health_log_detail (health_log_id int, unique_id int, alarm_id int, alarm_event_id int, "
77	"updated_by_id int, updates_id int, when_key int, duration int, non_clear_duration int, "
78	"flags int, exec_run_timestamp int, delay_up_to_timestamp int, "
79	"info text, exec_code int, new_status real, old_status real, delay int, "
80	"new_value double, old_value double, last_repeat int, transition_id blob, global_id int, summary text)",
81
82	"CREATE INDEX IF NOT EXISTS ind_d2 on dimension (chart_id)",
83	"CREATE INDEX IF NOT EXISTS ind_c3 on chart (host_id)",
84	"CREATE INDEX IF NOT EXISTS health_log_ind_1 ON health_log (host_id)",
85	"CREATE INDEX IF NOT EXISTS health_log_d_ind_2 ON health_log_detail (global_id)",
86	"CREATE INDEX IF NOT EXISTS health_log_d_ind_3 ON health_log_detail (transition_id)",
87	"CREATE INDEX IF NOT EXISTS health_log_d_ind_9 ON health_log_detail (unique_id DESC, health_log_id)",
88	"CREATE INDEX IF NOT EXISTS health_log_d_ind_6 on health_log_detail (health_log_id, when_key)",
89	"CREATE INDEX IF NOT EXISTS health_log_d_ind_7 on health_log_detail (alarm_id)",
90	"CREATE INDEX IF NOT EXISTS health_log_d_ind_8 on health_log_detail (new_status, updated_by_id)",
91
92	"CREATE TABLE IF NOT EXISTS agent_event_log (id INTEGER PRIMARY KEY, version TEXT, event_type INT, value, date_created INT)",
93	"CREATE INDEX IF NOT EXISTS idx_agent_event_log1 on agent_event_log (event_type)",
94
95	"CREATE TABLE IF NOT EXISTS alert_queue "
96	" (host_id BLOB, health_log_id INT, unique_id INT, alarm_id INT, status INT, date_scheduled INT, "
97	" UNIQUE(host_id, health_log_id, alarm_id))",
98
99	"CREATE INDEX IF NOT EXISTS ind_alert_queue1 ON alert_queue(host_id, date_scheduled)",
100
101	"CREATE TABLE IF NOT EXISTS alert_version (health_log_id INTEGER PRIMARY KEY, unique_id INT, status INT, "
102	"version INT, date_submitted INT)",
103
104	"CREATE TABLE IF NOT EXISTS aclk_queue (sequence_id INTEGER PRIMARY KEY, host_id blob, health_log_id INT, "
105	"unique_id INT, date_created INT, UNIQUE(host_id, health_log_id))",
106
107	"CREATE TABLE IF NOT EXISTS alert_hash_cloud (hash_id BLOB PRIMARY KEY)",
108
109	"CREATE TABLE IF NOT EXISTS ctx_metadata_cleanup (id INTEGER PRIMARY KEY, host_id BLOB, context TEXT NOT NULL, date_created INT NOT NULL, "
110	"UNIQUE (host_id, context))",
111
112	NULL
113	};
114
115	const char *database_cleanup[] = {
116	"DELETE FROM host WHERE host_id NOT IN (SELECT host_id FROM chart)",
117	"DELETE FROM node_instance WHERE host_id NOT IN (SELECT host_id FROM host)",
118	"DELETE FROM host_info WHERE host_id NOT IN (SELECT host_id FROM host)",
119	"DELETE FROM host_label WHERE host_id NOT IN (SELECT host_id FROM host)",
120	"DELETE FROM ctx_metadata_cleanup WHERE host_id NOT IN (SELECT host_id FROM host)",
121	"DROP TRIGGER IF EXISTS tr_dim_del",
122	"DROP INDEX IF EXISTS ind_d1",
123	"DROP INDEX IF EXISTS ind_c1",
124	"DROP INDEX IF EXISTS ind_c2",
125	"DROP INDEX IF EXISTS alert_hash_index",
126	"DROP INDEX IF EXISTS health_log_d_ind_4",
127	"DROP INDEX IF EXISTS health_log_d_ind_1",
128	"DROP INDEX IF EXISTS health_log_d_ind_5",
129	"DELETE FROM alert_hash_cloud WHERE hash_id NOT IN (SELECT hash_id FROM alert_hash)",
130	NULL
131	};
132
133	sqlite3 *db_meta = NULL;
134
135	// SQL statements
136
137	#define SQL_CLEANUP_AGENT_EVENT_LOG "DELETE FROM agent_event_log WHERE date_created < UNIXEPOCH() - 30 * 86400"
138
139	#define SQL_DELETE_ORPHAN_HEALTH_LOG "DELETE FROM health_log WHERE host_id NOT IN (SELECT host_id FROM host)"
140
141	#define SQL_DELETE_ORPHAN_HEALTH_LOG_DETAIL \
142	"DELETE FROM health_log_detail WHERE health_log_id NOT IN (SELECT health_log_id FROM health_log)"
143
144	#define SQL_DELETE_ORPHAN_ALERT_VERSION \
145	"DELETE FROM alert_version WHERE health_log_id NOT IN (SELECT health_log_id FROM health_log)"
146
147	#define SQL_STORE_CLAIM_ID \
148	"INSERT INTO node_instance " \
149	"(host_id, claim_id, date_created) VALUES (@host_id, @claim_id, UNIXEPOCH()) " \
150	"ON CONFLICT(host_id) DO UPDATE SET claim_id = excluded.claim_id"
151
152	#define SQL_DELETE_HOST_LABELS "DELETE FROM host_label WHERE host_id = @uuid"
153
154	#define SQL_STORE_HOST_LABEL \
155	"INSERT INTO host_label (host_id, source_type, label_key, label_value, date_created) VALUES "
156
157	#define SQL_STORE_HOST_LABEL_CONFLICT " ON CONFLICT (host_id, label_key) " \
158	"DO UPDATE SET source_type = excluded.source_type, label_value = excluded.label_value, date_created = UNIXEPOCH()"
159
160	#define SQL_STORE_CHART_LABEL \
161	"INSERT INTO chart_label (chart_id, source_type, label_key, label_value, date_created) VALUES "
162
163	#define SQL_STORE_CHART_LABEL_CONFLICT " ON CONFLICT (chart_id, label_key) " \
164	"DO UPDATE SET source_type = excluded.source_type, label_value = excluded.label_value, date_created = UNIXEPOCH()"
165
166	#define DELETE_DIMENSION_UUID "DELETE FROM dimension WHERE dim_id = @uuid"
167
168	#define SQL_STORE_HOST_INFO \
169	"INSERT OR REPLACE INTO host (host_id, hostname, registry_hostname, update_every, os, timezone, tags, hops, " \
170	"memory_mode, abbrev_timezone, utc_offset, program_name, program_version, entries, health_enabled, last_connected) " \
171	"VALUES (@host_id, @hostname, @registry_hostname, @update_every, @os, @timezone, @tags, @hops, " \
172	"@memory_mode, @abbrev_tz, @utc_offset, @prog_name, @prog_version, @entries, @health_enabled, @last_connected)"
173
174	#define SQL_STORE_CHART \
175	"INSERT INTO chart (chart_id, host_id, type, id, name, family, context, title, unit, plugin, module, priority, " \
176	"update_every, chart_type, memory_mode, history_entries) " \
177	"values (@chart_id, @host_id, @type, @id, @name, @family, @context, @title, @unit, @plugin, @module, @priority, " \
178	"@update_every, @chart_type, @memory_mode, @history_entries) " \
179	"ON CONFLICT(chart_id) DO UPDATE SET type=excluded.type, id=excluded.id, name=excluded.name, " \
180	"family=excluded.family, context=excluded.context, title=excluded.title, unit=excluded.unit, " \
181	"plugin=excluded.plugin, module=excluded.module, priority=excluded.priority, update_every=excluded.update_every, " \
182	"chart_type=excluded.chart_type, memory_mode = excluded.memory_mode, history_entries = excluded.history_entries"
183
184	#define SQL_STORE_DIMENSION \
185	"INSERT INTO dimension (dim_id, chart_id, id, name, multiplier, divisor , algorithm, options) " \
186	"VALUES (@dim_id, @chart_id, @id, @name, @multiplier, @divisor, @algorithm, @options) " \
187	"ON CONFLICT(dim_id) DO UPDATE SET id=excluded.id, name=excluded.name, multiplier=excluded.multiplier, " \
188	"divisor=excluded.divisor, algorithm=excluded.algorithm, options=excluded.options"
189
190	#define SELECT_DIMENSION_LIST "SELECT dim_id, rowid FROM dimension WHERE rowid > @row_id"
191	#define SELECT_CHART_LIST "SELECT chart_id, rowid FROM chart WHERE rowid > @row_id"
192	#define SELECT_CHART_LABEL_LIST "SELECT chart_id, rowid FROM chart_label WHERE rowid > @row_id"
193
194	#define SQL_STORE_HOST_SYSTEM_INFO_VALUES \
195	"INSERT OR REPLACE INTO host_info (host_id, system_key, system_value, date_created) VALUES " \
196	"(@uuid, @name, @value, UNIXEPOCH())"
197
198	#define CONVERT_EXISTING_LOCALHOST "UPDATE host SET hops = 1 WHERE hops = 0 AND host_id <> @host_id"
199	#define DELETE_MISSING_NODE_INSTANCES "DELETE FROM node_instance WHERE host_id NOT IN (SELECT host_id FROM host)"
200
201	#define METADATA_MAINTENANCE_FIRST_CHECK (1800) // Maintenance first run after agent startup in seconds
202	#define METADATA_MAINTENANCE_REPEAT (60) // Repeat if last run for dimensions, charts, labels needs more work
203	#define METADATA_MAINTENANCE_CTX_CLEAN_REPEAT (300) // Repeat if last run for dimensions, charts, labels needs more work
204	#define METADATA_HEALTH_LOG_INTERVAL (3600) // Repeat maintenance for health
205	#define METADATA_LABEL_CHECK_INTERVAL (3600) // Repeat maintenance for labels
206	#define METADATA_RUNTIME_THRESHOLD (5) // Run time threshold for cleanup task
207
208	#define METADATA_HOST_CHECK_FIRST_CHECK (5) // First check for pending metadata
209	#define METADATA_HOST_CHECK_INTERVAL (5) // Repeat check for pending metadata
210	#define METADATA_MAX_BATCH_SIZE (64) // Maximum commands to execute before running the event loop
211
212	#define DATABASE_VACUUM_FREQUENCY_SECONDS (60)
213	#define DATABASE_FREE_PAGES_THRESHOLD_PC (5) // Percentage of free pages to trigger vacuum
214	#define DATABASE_FREE_PAGES_VACUUM_PC (10) // Percentage of free pages to vacuum
215
216	enum metadata_opcode {
217	METADATA_DATABASE_NOOP = 0,
218	METADATA_DEL_DIMENSION,
219	METADATA_STORE_CLAIM_ID,
220	METADATA_STORE,
221	METADATA_LOAD_HOST_CONTEXT,
222	METADATA_ADD_HOST_AE,
223	METADATA_DEL_HOST_AE,
224	METADATA_ADD_CTX_CLEANUP,
225	METADATA_EXECUTE_STORE_STATEMENT,
226	METADATA_SYNC_SHUTDOWN,
227	METADATA_UNITTEST,
228	// leave this last
229	// we need it to check for worker utilization
230	METADATA_MAX_ENUMERATIONS_DEFINED
231	};
232
233	struct meta_config_s {
234	ND_THREAD *thread;
235	uv_loop_t loop;
236	uv_async_t async;
237	uv_timer_t timer_req;
238	time_t metadata_check_after;
239	Pvoid_t ae_DelJudyL;
240	bool initialized;
241	bool ctx_load_running;
242	bool metadata_running;
243	bool store_metadata;
244	bool shutdown_requested;
245	struct completion start_stop_complete;
246	CmdPool cmd_pool;
247	WorkerPool worker_pool;
248	} meta_config;
249
250	//
251	// For unittest
252	//
253	struct thread_unittest {
254	int join;
255	unsigned added;
256	unsigned processed;
257	unsigned *done;
258	};
259
260	int sql_metadata_cache_stats(int op)
261	{
262	int count, dummy;
263
264	sqlite3_db_status(db_meta, op, &count, &dummy, 0);
265	return count;
266	}
267
268	static inline void set_host_node_id(RRDHOST host, nd_uuid_t node_id)
269	{
270	if (unlikely(!host))
271	return;
272
273	if (unlikely(!node_id)) {
274	host->node_id = UUID_ZERO;
275	return;
276	}
277
278	struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
279
280	uuid_copy(host->node_id.uuid, *node_id);
281
282	if (unlikely(!aclk_host_config))
283	create_aclk_config(host, &host->host_id.uuid, node_id);
284	else
285	uuid_unparse_lower(*node_id, aclk_host_config->node_id);
286
287	stream_receiver_send_node_and_claim_id_to_child(host);
288	stream_path_node_id_updated(host);
289	}
290
291	struct host_ctx_cleanup_s {
292	nd_uuid_t host_uuid;
293	STRING *context;
294	};
295
296	#define CTX_DELETE_CONTEXT_META_CLEANUP_ITEM "DELETE FROM ctx_metadata_cleanup WHERE host_id = @host_id AND context = @context"
297
298	static void ctx_delete_metadata_cleanup_context(sqlite3_stmt *res, nd_uuid_t host_uuid, const char *context)
299	{
300	if (!*res) {
301	if (!PREPARE_STATEMENT(db_meta, CTX_DELETE_CONTEXT_META_CLEANUP_ITEM, res))
302	return;
303	}
304
305	int param = 0;
306	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_uuid, sizeof(host_uuid), SQLITE_STATIC));
307	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, context, -1, SQLITE_STATIC));
308
309	param = 0;
310	int rc = sqlite3_step_monitored(*res);
311	if (rc != SQLITE_DONE)
312	error_report("Failed to delete context check entry, rc = %d", rc);
313
314	done:
315	REPORT_BIND_FAIL(*res, param);
316	SQLITE_RESET(*res);
317	}
318
319	#define CTX_GET_CONTEXT_META_CLEANUP_LIST "SELECT context FROM ctx_metadata_cleanup WHERE host_id = @host_id"
320
321	static void ctx_get_context_list_to_cleanup(nd_uuid_t host_uuid, void (cleanup_cb)(Pvoid_t JudyL, void data), void data)
322	{
323	if (unlikely(!host_uuid))
324	return;
325
326	sqlite3_stmt *res = NULL;
327
328	if (!PREPARE_STATEMENT(db_meta, CTX_GET_CONTEXT_META_CLEANUP_LIST, &res))
329	return;
330
331	int param = 0;
332	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_uuid, sizeof(*host_uuid), SQLITE_STATIC));
333	param = 0;
334
335	const char *context;
336	Pvoid_t CTX_JudyL = NULL;
337	Pvoid_t *Pvalue;
338	while (sqlite3_step_monitored(res) == SQLITE_ROW) {
339	context = (char *) sqlite3_column_text(res, 0);
340	STRING *ctx = string_strdupz(context);
341	Pvalue = JudyLIns(&CTX_JudyL, (Word_t) ctx, PJE0);
342	if (Pvalue == PJERR \|\| *Pvalue)
343	string_freez(ctx);
344	else
345	Pvalue = (void )1;
346	}
347
348	if (CTX_JudyL) {
349	cleanup_cb(CTX_JudyL, data);
350
351	bool first = true;
352	Word_t Index = 0;
353	while ((Pvalue = JudyLFirstThenNext(CTX_JudyL, &Index, &first))) {
354	STRING ctx = (STRING ) Index;
355	string_freez(ctx);
356	}
357	}
358	(void)JudyLFreeArray(&CTX_JudyL, PJE0);
359
360	done:
361	REPORT_BIND_FAIL(res, param);
362	SQLITE_FINALIZE(res);
363	}
364
365	#define SQL_SCHEDULE_HOST_CTX_CLEANUP \
366	"INSERT INTO ctx_metadata_cleanup (host_id, context, date_created) " \
367	"VALUES (@host_id, @context, UNIXEPOCH()) ON CONFLICT DO UPDATE SET date_created = excluded.date_created"
368
369	// Schedule context cleanup for host
370	static void sql_schedule_host_ctx_cleanup(sqlite3_stmt *res, nd_uuid_t host_id, const char *context)
371	{
372	if (!*res) {
373	if (!PREPARE_STATEMENT(db_meta, SQL_SCHEDULE_HOST_CTX_CLEANUP, res))
374	return;
375	}
376
377	int param = 0;
378	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(host_id), SQLITE_STATIC));
379	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, context, -1, SQLITE_STATIC));
380
381	param = 0;
382	int rc = execute_insert(*res);
383	if (rc != SQLITE_DONE)
384	error_report("Failed to host context check data, rc = %d", rc);
385	done:
386	REPORT_BIND_FAIL(*res, param);
387	SQLITE_RESET(*res);
388	}
389
390	#define SQL_SET_HOST_LABEL \
391	"INSERT INTO host_label (host_id, source_type, label_key, label_value, date_created) " \
392	"VALUES (@host_id, @source_type, @label_key, @label_value, UNIXEPOCH()) ON CONFLICT (host_id, label_key) " \
393	" DO UPDATE SET source_type = excluded.source_type, label_value=excluded.label_value, date_created=UNIXEPOCH()"
394
395	bool sql_set_host_label(nd_uuid_t host_id, const char label_key, const char *label_value)
396	{
397	sqlite3_stmt *res = NULL;
398	bool status = false;
399
400	if (!label_key \|\| !label_value \|\| !host_id)
401	return false;
402
403	if (!PREPARE_STATEMENT(db_meta, SQL_SET_HOST_LABEL, &res))
404	return false;
405
406	int param = 0;
407	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
408	SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, RRDLABEL_SRC_AUTO));
409	SQLITE_BIND_FAIL(done, sqlite3_bind_text(res, ++param, label_key, -1, SQLITE_STATIC));
410	SQLITE_BIND_FAIL(done, sqlite3_bind_text(res, ++param, label_value, -1, SQLITE_STATIC));
411
412	param = 0;
413	int rc = sqlite3_step_monitored(res);
414	status = (rc == SQLITE_DONE);
415	if (false == status)
416	error_report("Failed to store node instance information, rc = %d", rc);
417	done:
418	REPORT_BIND_FAIL(res, param);
419	SQLITE_FINALIZE(res);
420	return status;
421	}
422
423	#define SQL_UPDATE_NODE_ID "UPDATE node_instance SET node_id = @node_id WHERE host_id = @host_id"
424
425	void sql_update_node_id(nd_uuid_t host_id, nd_uuid_t node_id)
426	{
427	sqlite3_stmt *res = NULL;
428	RRDHOST *host = NULL;
429	RRDHOST_ACQUIRED *acquired_host = NULL;
430
431	char host_guid[UUID_STR_LEN];
432	uuid_unparse_lower(*host_id, host_guid);
433
434	acquired_host = rrdhost_find_and_acquire(host_guid);
435	if (acquired_host) {
436	if ((host = rrdhost_acquired_to_rrdhost(acquired_host)))
437	set_host_node_id(host, node_id);
438	rrdhost_acquired_release(acquired_host);
439	}
440
441	if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_NODE_ID, &res))
442	return;
443
444	int param = 0;
445	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, node_id, sizeof(*node_id), SQLITE_STATIC));
446	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
447
448	param = 0;
449	int rc = sqlite3_step_monitored(res);
450	if (unlikely(rc != SQLITE_DONE))
451	error_report("Failed to store node instance information, rc = %d", rc);
452
453	done:
454	REPORT_BIND_FAIL(res, param);
455	SQLITE_FINALIZE(res);
456	}
457
458	#define SQL_INVALIDATE_NODE_INSTANCES \
459	"UPDATE node_instance SET node_id = NULL WHERE EXISTS " \
460	"(SELECT host_id FROM node_instance WHERE host_id = @host_id AND (@claim_id IS NULL OR claim_id <> @claim_id))"
461
462	void invalidate_node_instances(nd_uuid_t host_id, nd_uuid_t claim_id)
463	{
464	sqlite3_stmt *res = NULL;
465
466	if (!PREPARE_STATEMENT(db_meta, SQL_INVALIDATE_NODE_INSTANCES, &res))
467	return;
468
469	int param = 0;
470	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
471
472	if (claim_id)
473	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, claim_id, sizeof(*claim_id), SQLITE_STATIC));
474	else
475	SQLITE_BIND_FAIL(done, sqlite3_bind_null(res, ++param));
476
477	param = 0;
478	int rc = sqlite3_step_monitored(res);
479	if (unlikely(rc != SQLITE_DONE))
480	error_report("Failed to invalidate node instance information, rc = %d", rc);
481
482	done:
483	REPORT_BIND_FAIL(res, param);
484	SQLITE_FINALIZE(res);
485	}
486
487	#define SQL_GET_HOST_NODE_ID "SELECT node_id FROM node_instance WHERE host_id = @host_id"
488
489	void sql_load_node_id(RRDHOST *host)
490	{
491	sqlite3_stmt *res = NULL;
492
493	if (!PREPARE_STATEMENT(db_meta, SQL_GET_HOST_NODE_ID, &res))
494	return;
495
496	int param = 0;
497	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
498
499	param = 0;
500	int rc = sqlite3_step_monitored(res);
501	if (likely(rc == SQLITE_ROW)) {
502	nd_uuid_t node_id;
503	if (likely(sqlite3_column_uuid_copy(res, 0, node_id)))
504	set_host_node_id(host, &node_id);
505	else
506	set_host_node_id(host, NULL);
507	}
508
509	done:
510	REPORT_BIND_FAIL(res, param);
511	SQLITE_FINALIZE(res);
512	}
513
514	#define SELECT_HOST_INFO "SELECT system_key, system_value FROM host_info WHERE host_id = @host_id"
515
516	void sql_build_host_system_info(nd_uuid_t host_id, struct rrdhost_system_info system_info)
517	{
518	sqlite3_stmt *res = NULL;
519
520	if (!PREPARE_STATEMENT(db_meta, SELECT_HOST_INFO, &res))
521	return;
522
523	int param = 0;
524	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
525
526	param = 0;
527	while (sqlite3_step_monitored(res) == SQLITE_ROW) {
528	rrdhost_system_info_set_by_name(
529	system_info, (char )sqlite3_column_text(res, 0), (char )sqlite3_column_text(res, 1));
530	}
531
532	done:
533	REPORT_BIND_FAIL(res, param);
534	SQLITE_FINALIZE(res);
535	}
536
537	#define SELECT_HOST_LABELS "SELECT label_key, label_value, source_type FROM host_label WHERE host_id = @host_id " \
538	"AND label_key IS NOT NULL AND label_value IS NOT NULL"
539
540	RRDLABELS sql_load_host_labels(nd_uuid_t host_id)
541	{
542	RRDLABELS *labels = NULL;
543	sqlite3_stmt *res = NULL;
544
545	if (!PREPARE_STATEMENT(db_meta, SELECT_HOST_LABELS, &res))
546	return NULL;
547
548	int param = 0;
549	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
550
551	param = 0;
552	labels = rrdlabels_create();
553
554	while (sqlite3_step_monitored(res) == SQLITE_ROW) {
555	rrdlabels_add(
556	labels,
557	(const char *)sqlite3_column_text(res, 0),
558	(const char *)sqlite3_column_text(res, 1),
559	sqlite3_column_int(res, 2));
560	}
561
562	done:
563	REPORT_BIND_FAIL(res, param);
564	SQLITE_FINALIZE(res);
565	return labels;
566	}
567
568	static int exec_statement_with_uuid(const char sql, nd_uuid_t uuid)
569	{
570	int result = 1;
571	sqlite3_stmt *res = NULL;
572
573	if (!PREPARE_STATEMENT(db_meta, sql, &res)) {
574	error_report("Failed to prepare statement %s", sql);
575	return 1;
576	}
577
578	int param = 0;
579	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, uuid, sizeof(*uuid), SQLITE_STATIC));
580
581	param = 0;
582	int rc = sqlite3_step_monitored(res);
583	if (likely(rc == SQLITE_DONE))
584	result = SQLITE_OK;
585	else
586	error_report("Failed to execute %s, rc = %d", sql, rc);
587
588	done:
589	REPORT_BIND_FAIL(res, param);
590	SQLITE_FINALIZE(res);
591	return result;
592	}
593
594	static void recover_database(const char sqlite_database, const char new_sqlite_database)
595	{
596	sqlite3 *database;
597	int rc = sqlite3_open(sqlite_database, &database);
598	if (rc != SQLITE_OK)
599	return;
600
601	netdata_log_info("Recover %s", sqlite_database);
602	netdata_log_info(" to %s", new_sqlite_database);
603
604	// This will remove the -shm and -wal files when we close the database
605	(void)db_execute(database, "select count(*) from sqlite_master limit 0", NULL);
606
607	sqlite3_recover *recover = sqlite3_recover_init(database, "main", new_sqlite_database);
608	if (recover) {
609
610	rc = sqlite3_recover_run(recover);
611
612	if (rc == SQLITE_OK)
613	netdata_log_info("Recover complete");
614	else
615	netdata_log_error("Recover encountered an error but the database may be usable");
616
617	rc = sqlite3_recover_finish(recover);
618
619	(void) sqlite3_close_v2(database);
620
621	if (rc == SQLITE_OK) {
622	rc = rename(new_sqlite_database, sqlite_database);
623	if (rc == 0) {
624	netdata_log_info("Renamed %s", new_sqlite_database);
625	netdata_log_info(" to %s", sqlite_database);
626	}
627	}
628	else
629	netdata_log_error("Recover failed to free resources");
630	}
631	else
632	(void) sqlite3_close_v2(database);
633	}
634
635
636	static void sqlite_uuid_parse(sqlite3_context context, int argc, sqlite3_value *argv)
637	{
638	nd_uuid_t uuid;
639
640	if ( argc != 1 ){
641	sqlite3_result_null(context);
642	return ;
643	}
644	int rc = uuid_parse((const char *) sqlite3_value_text(argv[0]), uuid);
645	if (rc == -1) {
646	sqlite3_result_null(context);
647	return ;
648	}
649
650	sqlite3_result_blob(context, &uuid, sizeof(nd_uuid_t), SQLITE_TRANSIENT);
651	}
652
653	void sqlite_now_usec(sqlite3_context context, int argc, sqlite3_value *argv)
654	{
655	if (argc != 1 ){
656	sqlite3_result_null(context);
657	return ;
658	}
659
660	if (sqlite3_value_int(argv[0]) != 0) {
661	struct timespec req = {.tv_sec = 0, .tv_nsec = 1};
662	nanosleep(&req, NULL);
663	}
664
665	sqlite3_result_int64(context, (sqlite_int64) now_realtime_usec());
666	}
667
668	void sqlite_uuid_random(sqlite3_context context, int argc, sqlite3_value *argv)
669	{
670	(void)argc;
671	(void)argv;
672
673	nd_uuid_t uuid;
674	uuid_generate_random(uuid);
675	sqlite3_result_blob(context, &uuid, sizeof(nd_uuid_t), SQLITE_TRANSIENT);
676	}
677
678	static int64_t sql_get_wal_size(const char *database_file)
679	{
680	char filename[FILENAME_MAX + 1];
681	snprintfz(filename, sizeof(filename) - 1, "%s/%s-wal", netdata_configured_cache_dir, database_file);
682
683	uv_fs_t req;
684	int result = uv_fs_stat(NULL, &req, filename, NULL);
685	int64_t file_size = result >= 0 ? (int64_t) req.statbuf.st_size : -1;
686
687	uv_fs_req_cleanup(&req);
688	return file_size;
689	}
690
691	#define SQLITE_METADATA_WAL_LIMIT_X (10)
692
693	bool sql_metadata_wal_size_acceptable()
694	{
695	int64_t wal_size = sql_get_wal_size("netdata-meta.db");
696
697	if (wal_size > SQLITE_METADATA_WAL_LIMIT_X * def_journal_size_limit)
698	return false;
699
700	return true;
701	}
702
703	// Init
704	/*
705	* Initialize the SQLite database
706	* Return 0 on success
707	*/
708	int sql_init_meta_database(db_check_action_type_t rebuild, int memory)
709	{
710	char *err_msg = NULL;
711	char sqlite_database[FILENAME_MAX + 1];
712	int rc;
713
714	if (likely(!memory)) {
715	snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/.netdata-meta.db.recover", netdata_configured_cache_dir);
716	rc = unlink(sqlite_database);
717	snprintfz(sqlite_database, FILENAME_MAX, "%s/netdata-meta.db", netdata_configured_cache_dir);
718
719	if (rc == 0 \|\| (rebuild & DB_CHECK_RECOVER)) {
720	char new_sqlite_database[FILENAME_MAX + 1];
721	snprintfz(new_sqlite_database, sizeof(new_sqlite_database) - 1, "%s/netdata-meta-recover.db", netdata_configured_cache_dir);
722	recover_database(sqlite_database, new_sqlite_database);
723	if (rebuild & DB_CHECK_RECOVER)
724	return 0;
725	}
726
727	snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/.netdata-meta.db.delete", netdata_configured_cache_dir);
728	rc = unlink(sqlite_database);
729	snprintfz(sqlite_database, FILENAME_MAX, "%s/netdata-meta.db", netdata_configured_cache_dir);
730	if (rc == 0) {
731	char new_sqlite_database[FILENAME_MAX + 1];
732	snprintfz(new_sqlite_database, sizeof(new_sqlite_database) - 1, "%s/netdata-meta.bad", netdata_configured_cache_dir);
733	rc = rename(sqlite_database, new_sqlite_database);
734	if (rc)
735	error_report("Failed to rename %s to %s", sqlite_database, new_sqlite_database);
736	}
737	// note: sqlite_database contains the right name
738	}
739	else
740	strncpyz(sqlite_database, ":memory:", sizeof(sqlite_database) - 1);
741
742	rc = sqlite3_open(sqlite_database, &db_meta);
743	if (rc != SQLITE_OK) {
744	error_report("Failed to initialize database at %s, due to \"%s\"", sqlite_database, sqlite3_errstr(rc));
745	char *error_str = get_database_extented_error(db_meta, 0, "meta_open");
746	if (error_str)
747	analytics_set_data_str(&analytics_data.netdata_fail_reason, error_str);
748	freez(error_str);
749	goto close_database;
750	}
751
752	if (rebuild & DB_CHECK_RECLAIM_SPACE) {
753	netdata_log_info("Reclaiming space of %s", sqlite_database);
754	rc = sqlite3_exec_monitored(db_meta, "VACUUM", 0, 0, &err_msg);
755	if (rc != SQLITE_OK) {
756	error_report("Failed to execute VACUUM rc = %d (%s)", rc, err_msg);
757	sqlite3_free(err_msg);
758	}
759	else {
760	(void)db_execute(db_meta, "select count(*) from sqlite_master limit 0", NULL);
761	(void) sqlite3_close_v2(db_meta);
762	}
763	return 1;
764	}
765
766	if (rebuild & DB_CHECK_ANALYZE) {
767	errno_clear();
768	netdata_log_info("Running ANALYZE on %s", sqlite_database);
769	rc = sqlite3_exec_monitored(db_meta, "ANALYZE", 0, 0, &err_msg);
770	if (rc != SQLITE_OK) {
771	error_report("Failed to execute ANALYZE rc = %d (%s)", rc, err_msg);
772	sqlite3_free(err_msg);
773	}
774	else {
775	(void)db_execute(db_meta, "select count(*) from sqlite_master limit 0", NULL);
776	(void) sqlite3_close_v2(db_meta);
777	}
778	return 1;
779	}
780
781	errno_clear();
782	netdata_log_info("SQLite database %s initialization", sqlite_database);
783
784	rc = sqlite3_create_function(db_meta, "u2h", 1, SQLITE_ANY \| SQLITE_DETERMINISTIC, 0, sqlite_uuid_parse, 0, 0);
785	if (unlikely(rc != SQLITE_OK))
786	error_report("Failed to register internal u2h function");
787
788	rc = sqlite3_create_function(db_meta, "now_usec", 1, SQLITE_ANY, 0, sqlite_now_usec, 0, 0);
789	if (unlikely(rc != SQLITE_OK))
790	error_report("Failed to register internal now_usec function");
791
792	rc = sqlite3_create_function(db_meta, "uuid_random", 0, SQLITE_ANY, 0, sqlite_uuid_random, 0, 0);
793	if (unlikely(rc != SQLITE_OK))
794	error_report("Failed to register internal uuid_random function");
795
796	int target_version = DB_METADATA_VERSION;
797
798	if (likely(!memory))
799	target_version = perform_database_migration(db_meta, DB_METADATA_VERSION);
800
801	if (configure_sqlite_database(db_meta, target_version, "meta_config"))
802	goto close_database;
803
804	if (init_database_batch(db_meta, &database_config[0], "meta_init"))
805	goto close_database;
806
807	if (init_database_batch(db_meta, &database_cleanup[0], "meta_cleanup"))
808	goto close_database;
809
810	netdata_log_info("SQLite database initialization completed");
811	if (sqlite3_busy_timeout(db_meta, SQLITE_BUSY_DELAY_MS) != SQLITE_OK)
812	nd_log_daemon(NDLP_WARNING, "SQLITE: Failed to set busy timeout to %d ms", SQLITE_BUSY_DELAY_MS);
813
814	return 0;
815
816	close_database:
817	sqlite3_close_v2(db_meta);
818	db_meta = NULL;
819	return 1;
820	}
821
822	// Metadata functions
823
824	// Label storage types
825	typedef enum {
826	STORE_HOST_LABELS,
827	STORE_CHART_LABELS,
828	} label_store_type_t;
829
830	// Structure to hold a single label entry for collection
831	struct label_entry {
832	char *name;
833	char *value;
834	RRDLABEL_SRC ls;
835	};
836
837	// Context for label collection callback (no I/O, just data collection)
838	struct label_collect_ctx {
839	struct label_entry *entries;
840	size_t count;
841	size_t capacity;
842	};
843
844	#define LABEL_COLLECT_INITIAL_CAPACITY 64
845
846	// Max labels per batch: SQLite default SQLITE_MAX_VARIABLE_NUMBER is 32766 in the version we are using
847	#define LABEL_BATCH_SIZE (1024)
848
849	// Callback to collect labels into an array (no SQLite I/O while spinlock is held)
850	static int collect_label_callback(const char name, const char value, RRDLABEL_SRC ls, void *data)
851	{
852	struct label_collect_ctx *ctx = data;
853
854	if (unlikely(!name \|\| !value))
855	return 1;
856
857	// Grow array if needed
858	if (ctx->count >= ctx->capacity) {
859	ctx->capacity *= 2;
860	ctx->entries = reallocz(ctx->entries, ctx->capacity * sizeof(*ctx->entries));
861	}
862
863	// Copy label data
864	ctx->entries[ctx->count].name = strdupz(name);
865	ctx->entries[ctx->count].value = strdupz(value);
866	ctx->entries[ctx->count].ls = ls;
867	ctx->count++;
868
869	return 1;
870	}
871
872	// Execute a batch of label inserts using a single multi-row INSERT statement
873	// Returns number of errors
874	static int store_label_batch(
875	nd_uuid_t *uuid,
876	struct label_entry *entries,
877	size_t count,
878	label_store_type_t type,
879	BUFFER *sql)
880	{
881	if (count == 0)
882	return 0;
883
884	// Build SQL: INSERT INTO table (...) VALUES (?,?,?,?,UNIXEPOCH()), (?,?,?,?,UNIXEPOCH()), ... ON CONFLICT ...
885	buffer_flush(sql);
886
887	if (type == STORE_HOST_LABELS)
888	buffer_strcat(sql, SQL_STORE_HOST_LABEL);
889	else
890	buffer_strcat(sql, SQL_STORE_CHART_LABEL);
891
892	for (size_t i = 0; i < count; i++) {
893	if (i > 0)
894	buffer_strcat(sql, ", ");
895	buffer_strcat(sql, "(?, ?, ?, ?, UNIXEPOCH())");
896	}
897
898	if (type == STORE_HOST_LABELS)
899	buffer_strcat(sql, SQL_STORE_HOST_LABEL_CONFLICT);
900	else
901	buffer_strcat(sql, SQL_STORE_CHART_LABEL_CONFLICT);
902
903	sqlite3_stmt *stmt = NULL;
904
905	if (!PREPARE_STATEMENT(db_meta, buffer_tostring(sql), &stmt))
906	return 1;
907
908	int errors = 0;
909
910	// Bind all parameters: 4 per label (uuid, source, key, value)
911	int param = 0;
912	for (size_t i = 0; i < count; i++) {
913	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_blob(stmt, ++param, uuid, sizeof(*uuid), SQLITE_STATIC));
914	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(stmt, ++param, (int)(entries[i].ls & ~(RRDLABEL_FLAG_INTERNAL))));
915	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_text(stmt, ++param, entries[i].name, -1, SQLITE_STATIC));
916	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_text(stmt, ++param, entries[i].value, -1, SQLITE_STATIC));
917	}
918	param = 0;
919
920	int rc = sqlite3_step_monitored(stmt);
921	if (unlikely(rc != SQLITE_DONE)) {
922	errors = 1;
923	error_report("Failed to store label batch, rc = %d", rc);
924	}
925
926	bind_fail:
927	if (param)
928	errors = 1;
929	REPORT_BIND_FAIL(stmt, param);
930	SQLITE_FINALIZE(stmt);
931
932	return errors;
933	}
934
935	// Store labels for a host or chart using batched prepared statements
936	// Two-phase approach: collect labels first (while spinlock held), then execute SQLite (after spinlock released)
937	static int store_labels(nd_uuid_t uuid, RRDLABELS labels, label_store_type_t type, BUFFER *work_buffer)
938	{
939	if (unlikely(!uuid \|\| !labels))
940	return 0;
941
942	// Phase 1: Collect labels while spinlock is held (fast, no I/O)
943	struct label_collect_ctx collect_ctx = {
944	.entries = mallocz(LABEL_COLLECT_INITIAL_CAPACITY * sizeof(*collect_ctx.entries)),
945	.count = 0,
946	.capacity = LABEL_COLLECT_INITIAL_CAPACITY,
947	};
948
949	rrdlabels_walkthrough_read(labels, collect_label_callback, &collect_ctx);
950	// Spinlock is now released
951
952	// Phase 2: Execute SQLite operations in batches (no spinlock held)
953	int errors = 0;
954
955	if (collect_ctx.count > 0) {
956	bool free_buffer = false;
957	if (!work_buffer) {
958	work_buffer = buffer_create(256 + collect_ctx.count * 30, NULL);
959	free_buffer = true;
960	}
961
962	size_t remaining = collect_ctx.count;
963	size_t offset = 0;
964
965	while (remaining > 0) {
966	size_t batch_count = (remaining > LABEL_BATCH_SIZE) ? LABEL_BATCH_SIZE : remaining;
967	errors += store_label_batch(uuid, &collect_ctx.entries[offset], batch_count, type, work_buffer);
968	offset += batch_count;
969	remaining -= batch_count;
970	}
971
972	if (free_buffer)
973	buffer_free(work_buffer);
974	}
975
976	// Cleanup collected labels
977	for (size_t i = 0; i < collect_ctx.count; i++) {
978	freez(collect_ctx.entries[i].name);
979	freez(collect_ctx.entries[i].value);
980	}
981	freez(collect_ctx.entries);
982
983	return errors ? 1 : 0;
984	}
985
986	static int check_and_update_chart_labels(RRDSET st, BUFFER work_buffer)
987	{
988	uint32_t old_version = st->rrdlabels_last_saved_version;
989	uint32_t new_version = rrdlabels_version(st->rrdlabels);
990
991	if (new_version == old_version)
992	return 0;
993
994	int rc = store_labels(&st->chart_uuid, st->rrdlabels, STORE_CHART_LABELS, work_buffer);
995	if (likely(!rc))
996	st->rrdlabels_last_saved_version = new_version;
997
998	return rc;
999	}
1000
1001	// If the machine guid has changed, then existing one with hops 0 will be marked as hops 1 (child)
1002	void detect_machine_guid_change(nd_uuid_t *host_uuid)
1003	{
1004	int rc;
1005
1006	rc = exec_statement_with_uuid(CONVERT_EXISTING_LOCALHOST, host_uuid);
1007	if (!rc) {
1008	if (unlikely(db_execute(db_meta, DELETE_MISSING_NODE_INSTANCES, NULL)))
1009	error_report("Failed to remove deleted hosts from node instances");
1010	}
1011	}
1012
1013	static int store_claim_id(nd_uuid_t host_id, nd_uuid_t claim_id)
1014	{
1015	sqlite3_stmt *res = NULL;
1016	int rc = 0;
1017
1018	if (!PREPARE_STATEMENT(db_meta, SQL_STORE_CLAIM_ID, &res))
1019	return 1;
1020
1021	int param = 0;
1022	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
1023
1024	if (claim_id)
1025	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param,claim_id, sizeof(*claim_id), SQLITE_STATIC));
1026	else
1027	SQLITE_BIND_FAIL(done, sqlite3_bind_null(res, ++param));
1028
1029	param = 0;
1030	rc = sqlite3_step_monitored(res);
1031	if (unlikely(rc != SQLITE_DONE))
1032	error_report("Failed to store host claim id rc = %d", rc);
1033
1034	done:
1035	REPORT_BIND_FAIL(res, param);
1036	SQLITE_FINALIZE(res);
1037	return rc != SQLITE_DONE;
1038	}
1039
1040	#define SQL_DELETE_DIMENSION_BY_ID "DELETE FROM dimension WHERE rowid = @dimension_row AND dim_id = @uuid"
1041
1042	static void delete_dimension_by_rowid(sqlite3_stmt *res, int64_t dimension_id, nd_uuid_t dim_uuid)
1043	{
1044	if (!*res) {
1045	if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_DIMENSION_BY_ID, res))
1046	return;
1047	}
1048
1049	int param = 0;
1050	SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, dimension_id));
1051	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, dim_uuid, sizeof(dim_uuid), SQLITE_STATIC));
1052
1053	param = 0;
1054	int rc = sqlite3_step_monitored(*res);
1055	if (unlikely(rc != SQLITE_DONE))
1056	error_report("Failed to delete dimension id, rc = %d", rc);
1057
1058	done:
1059	REPORT_BIND_FAIL(*res, param);
1060	SQLITE_RESET(*res);
1061	}
1062
1063	static void delete_dimension_uuid(nd_uuid_t dimension_uuid, sqlite3_stmt *action_res __maybe_unused, bool flag __maybe_unused)
1064	{
1065	if(!dimension_uuid)
1066	return;
1067
1068	sqlite3_stmt *res = NULL;
1069	int rc;
1070
1071	if (!PREPARE_STATEMENT(db_meta, DELETE_DIMENSION_UUID, &res))
1072	return;
1073
1074	int param = 0;
1075	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, dimension_uuid, sizeof(*dimension_uuid), SQLITE_STATIC));
1076
1077	param = 0;
1078	rc = sqlite3_step_monitored(res);
1079	if (unlikely(rc != SQLITE_DONE))
1080	error_report("Failed to delete dimension uuid, rc = %d", rc);
1081
1082	done:
1083	REPORT_BIND_FAIL(res, param);
1084	SQLITE_FINALIZE(res);
1085	}
1086
1087	//
1088	// Store host and host system info information in the database
1089	static int store_host_metadata(RRDHOST *host)
1090	{
1091	sqlite3_stmt *res = NULL;
1092	RRDHOST_TZ host_tz = { 0 };
1093
1094	if (!PREPARE_STATEMENT(db_meta, SQL_STORE_HOST_INFO, &res))
1095	return false;
1096
1097	int param = 0;
1098	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
1099	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_hostname(host), 0));
1100	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_registry_hostname(host), 1));
1101	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host->rrd_update_every));
1102	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_os(host), 1));
1103	host_tz = rrdhost_tz_get(host);
1104	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, host_tz.timezone, 1));
1105	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, "", 1));
1106	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, rrdhost_ingestion_hops(host)));
1107	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host->rrd_memory_mode));
1108	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, host_tz.abbrev_timezone, 1));
1109	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, host_tz.utc_offset));
1110	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_program_name(host), 1));
1111	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, rrdhost_program_version(host), 1));
1112	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int64(res, ++param, host->rrd_history_entries));
1113	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int(res, ++param, (int)host->health.enabled));
1114	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_int64(res, ++param, (sqlite3_int64) host->stream.snd.status.last_connected));
1115
1116	int store_rc = sqlite3_step_monitored(res);
1117
1118	if (unlikely(store_rc != SQLITE_DONE))
1119	error_report("Failed to store host %s, rc = %d", rrdhost_hostname(host), store_rc);
1120
1121	SQLITE_FINALIZE(res);
1122	rrdhost_tz_free(&host_tz);
1123
1124	return store_rc != SQLITE_DONE;
1125
1126	bind_fail:
1127	REPORT_BIND_FAIL(res, param);
1128	SQLITE_FINALIZE(res);
1129	rrdhost_tz_free(&host_tz);
1130	return 1;
1131	}
1132
1133	static int add_host_sysinfo_key_value(const char name, const char value, nd_uuid_t *uuid)
1134	{
1135	sqlite3_stmt *res = NULL;
1136
1137	if (!PREPARE_STATEMENT(db_meta, SQL_STORE_HOST_SYSTEM_INFO_VALUES, &res))
1138	return 0;
1139
1140	int param = 0;
1141	SQLITE_BIND_FAIL(bind_fail, sqlite3_bind_blob(res, ++param, uuid, sizeof(*uuid), SQLITE_STATIC));
1142	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, name, 0));
1143	SQLITE_BIND_FAIL(bind_fail, bind_text_null(res, ++param, value ? value : "unknown", 0));
1144
1145	int store_rc = sqlite3_step_monitored(res);
1146	if (unlikely(store_rc != SQLITE_DONE))
1147	error_report("Failed to store host info value %s, rc = %d", name, store_rc);
1148
1149	SQLITE_FINALIZE(res);
1150
1151	return store_rc == SQLITE_DONE;
1152
1153	bind_fail:
1154	REPORT_BIND_FAIL(res, param);
1155	SQLITE_FINALIZE(res);
1156	return 0;
1157	}
1158
1159	static bool store_host_systeminfo(RRDHOST *host)
1160	{
1161	struct rrdhost_system_info *system_info = host->system_info;
1162
1163	if (unlikely(!system_info))
1164	return false;
1165
1166	return (RRDHOST_SYSTEM_INFO_KEY_COUNT != rrdhost_system_info_foreach(system_info, add_host_sysinfo_key_value, &host->host_id.uuid));
1167	}
1168
1169
1170	/*
1171	* Store a chart in the database
1172	*/
1173
1174	static int store_chart_metadata(RRDSET st, sqlite3_stmt *res)
1175	{
1176	if (!*res) {
1177	if (!PREPARE_STATEMENT(db_meta, SQL_STORE_CHART, res))
1178	return 1;
1179	}
1180
1181	int rc = 1;
1182	int param = 0;
1183	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(*res, ++param, &st->chart_uuid, sizeof(st->chart_uuid), SQLITE_STATIC));
1184	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(*res, ++param, &st->rrdhost->host_id.uuid, sizeof(st->rrdhost->host_id.uuid), SQLITE_STATIC));
1185	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, string2str(st->parts.type), -1, SQLITE_STATIC));
1186	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, string2str(st->parts.id), -1, SQLITE_STATIC));
1187
1188	const char *name = string2str(st->parts.name);
1189	if (name && *name)
1190	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, name, -1, SQLITE_STATIC));
1191	else
1192	SQLITE_BIND_FAIL(done, sqlite3_bind_null(*res, ++param));
1193
1194	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_family(st), -1, SQLITE_STATIC));
1195	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_context(st), -1, SQLITE_STATIC));
1196	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_title(st), -1, SQLITE_STATIC));
1197	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_units(st), -1, SQLITE_STATIC));
1198	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_plugin_name(st), -1, SQLITE_STATIC));
1199	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, rrdset_module_name(st), -1, SQLITE_STATIC));
1200	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, (int) st->priority));
1201	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, st->update_every));
1202	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, st->chart_type));
1203	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, st->rrd_memory_mode));
1204	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, (int) st->db.entries));
1205
1206	param = 0;
1207	rc = sqlite3_step_monitored(*res);
1208	if (unlikely(rc != SQLITE_DONE))
1209	error_report("Failed to store chart, rc = %d", rc);
1210
1211	done:
1212	REPORT_BIND_FAIL(*res, param);
1213	SQLITE_RESET(*res);
1214	return rc != SQLITE_DONE;
1215	}
1216
1217	static bool store_dimension_metadata(RRDDIM rd, sqlite3_stmt *res)
1218	{
1219	if (!*res) {
1220	if (!PREPARE_STATEMENT(db_meta, SQL_STORE_DIMENSION, res))
1221	return 1;
1222	}
1223
1224	int rc = 1;
1225	int param = 0;
1226
1227	nd_uuid_t *rd_uuid = uuidmap_uuid_ptr(rd->uuid);
1228	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, rd_uuid, sizeof(rd_uuid), SQLITE_STATIC));
1229	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(*res, ++param, &rd->rrdset->chart_uuid, sizeof(rd->rrdset->chart_uuid), SQLITE_STATIC));
1230	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, string2str(rd->id), -1, SQLITE_STATIC));
1231	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, string2str(rd->name), -1, SQLITE_STATIC));
1232	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, (int) rd->multiplier));
1233	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, (int ) rd->divisor));
1234	SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, rd->algorithm));
1235	if (rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN))
1236	SQLITE_BIND_FAIL(done, sqlite3_bind_text(*res, ++param, "hidden", -1, SQLITE_STATIC));
1237	else
1238	SQLITE_BIND_FAIL(done, sqlite3_bind_null(*res, ++param));
1239
1240	param = 0;
1241	rc = sqlite3_step_monitored(*res);
1242	if (unlikely(rc != SQLITE_DONE))
1243	error_report("Failed to store dimension, rc = %d", rc);
1244
1245	done:
1246	REPORT_BIND_FAIL(*res, param);
1247	SQLITE_RESET(*res);
1248	return (rc != SQLITE_DONE);
1249	}
1250
1251	static bool dimension_can_be_deleted(nd_uuid_t dim_uuid __maybe_unused, sqlite3_stmt *res __maybe_unused, bool flag __maybe_unused)
1252	{
1253	#ifdef ENABLE_DBENGINE
1254	if(dbengine_enabled && dim_uuid) {
1255	bool no_retention = true;
1256	for (size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {
1257	if (!multidb_ctx[tier])
1258	continue;
1259	time_t first_time_t = 0, last_time_t = 0;
1260	if (rrdeng_metric_retention_by_uuid((void *) multidb_ctx[tier], dim_uuid, &first_time_t, &last_time_t)) {
1261	if (first_time_t > 0) {
1262	no_retention = false;
1263	break;
1264	}
1265	}
1266	}
1267	return no_retention;
1268	}
1269	else
1270	return false;
1271	#else
1272	return false;
1273	#endif
1274	}
1275
1276	static bool run_cleanup_loop(
1277	sqlite3_stmt *res,
1278	struct meta_config_s *config,
1279	bool (check_cb)(nd_uuid_t , sqlite3_stmt **, bool),
1280	void (action_cb)(nd_uuid_t , sqlite3_stmt **, bool),
1281	uint32_t *total_checked,
1282	uint32_t *total_deleted,
1283	uint64_t *row_id,
1284	sqlite3_stmt **check_stmt,
1285	sqlite3_stmt **action_stmt,
1286	bool check_flag,
1287	bool action_flag)
1288	{
1289	if (unlikely(SHUTDOWN_REQUESTED(config)))
1290	return true;
1291
1292	int rc = sqlite3_bind_int64(res, 1, (sqlite3_int64) *row_id);
1293	if (unlikely(rc != SQLITE_OK))
1294	return true;
1295
1296	time_t start_running = now_monotonic_sec();
1297	bool time_expired = false;
1298
1299	uint32_t l_checked = 0;
1300	uint32_t l_deleted = 0;
1301	while (!time_expired && sqlite3_step_monitored(res) == SQLITE_ROW) {
1302	nd_uuid_t uuid = {0};
1303
1304	if (unlikely(SHUTDOWN_REQUESTED(config)))
1305	break;
1306
1307	*row_id = sqlite3_column_int64(res, 1);
1308	if (unlikely(!sqlite3_column_uuid_copy(res, 0, uuid)))
1309	continue;
1310
1311	rc = check_cb(&uuid, check_stmt, check_flag);
1312
1313	if (rc == true) {
1314	action_cb(&uuid, action_stmt, action_flag);
1315	l_deleted++;
1316	}
1317
1318	l_checked++;
1319	time_expired = ((now_monotonic_sec() - start_running) > METADATA_RUNTIME_THRESHOLD);
1320	}
1321
1322	(*total_checked) += l_checked;
1323	(*total_deleted) += l_deleted;
1324	return time_expired;
1325	}
1326
1327
1328	#define SQL_CHECK_CHART_EXISTENCE_IN_DIMENSION "SELECT count(1) FROM dimension WHERE chart_id = @chart_id"
1329	#define SQL_CHECK_CHART_EXISTENCE_IN_CHART "SELECT count(1) FROM chart WHERE chart_id = @chart_id"
1330
1331	static bool chart_can_be_deleted(nd_uuid_t chart_uuid, sqlite3_stmt *check_res, bool check_in_dimension)
1332	{
1333	int rc, result = 1;
1334	sqlite3_stmt res = check_res ? check_res : NULL;
1335
1336	if (!res) {
1337	if (!PREPARE_STATEMENT(
1338	db_meta,
1339	check_in_dimension ? SQL_CHECK_CHART_EXISTENCE_IN_DIMENSION : SQL_CHECK_CHART_EXISTENCE_IN_CHART,
1340	&res))
1341	return 0;
1342
1343	if (check_res)
1344	*check_res = res;
1345	}
1346
1347	rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);
1348	if (unlikely(rc != SQLITE_OK)) {
1349	error_report("Failed to bind chart uuid parameter, rc = %d", rc);
1350	goto skip;
1351	}
1352
1353	rc = sqlite3_step_monitored(res);
1354	if (likely(rc == SQLITE_ROW))
1355	result = sqlite3_column_int(res, 0);
1356
1357	skip:
1358	if (check_res)
1359	SQLITE_RESET(res);
1360	else
1361	SQLITE_FINALIZE(res);
1362
1363	return result == 0;
1364	}
1365
1366	#define SQL_DELETE_CHART_BY_UUID "DELETE FROM chart WHERE chart_id = @chart_id"
1367	#define SQL_DELETE_CHART_LABEL_BY_UUID "DELETE FROM chart_label WHERE chart_id = @chart_id"
1368
1369	static void delete_chart_uuid(nd_uuid_t chart_uuid, sqlite3_stmt *action_res, bool label_only)
1370	{
1371	int rc;
1372	sqlite3_stmt res = action_res ? action_res : NULL;
1373
1374	if (!res) {
1375	if (!PREPARE_STATEMENT(db_meta, label_only ? SQL_DELETE_CHART_LABEL_BY_UUID : SQL_DELETE_CHART_BY_UUID, &res))
1376	return;
1377	if (action_res)
1378	*action_res = res;
1379	}
1380
1381	rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);
1382	if (unlikely(rc != SQLITE_OK)) {
1383	error_report("Failed to bind chart uuid parameter, rc = %d", rc);
1384	goto skip;
1385	}
1386
1387	rc = sqlite3_step_monitored(res);
1388	if (unlikely(rc != SQLITE_DONE))
1389	error_report("Failed to delete a chart uuid from the %s table, rc = %d", label_only ? "labels" : "chart", rc);
1390
1391	skip:
1392	if (action_res)
1393	SQLITE_RESET(res);
1394	else
1395	SQLITE_FINALIZE(res);
1396	}
1397
1398	static uint64_t get_rowid_from_statement(const char *sql)
1399	{
1400	sqlite3_stmt *res = NULL;
1401
1402	if (!PREPARE_STATEMENT(db_meta, sql, &res))
1403	return 0;
1404
1405	uint64_t rowid = 0;
1406
1407	if (sqlite3_step_monitored(res) == SQLITE_ROW) {
1408	rowid = sqlite3_column_int64(res, 0);
1409	}
1410
1411	SQLITE_FINALIZE(res);
1412	return rowid;
1413	}
1414
1415
1416	// Descriptor + state for one cleanup cycle (dimension / chart / chart_label).
1417	// Replaces the three former check_*_metadata() functions with one driver:
1418	// first call -> arm timer; snapshot_pending = true so the snapshot
1419	// is taken at the next entry past the timer (NOT
1420	// now, which would burn a SELECT for nothing — the
1421	// scan can't run yet)
1422	// timer not expired -> return true (yield)
1423	// snapshot_pending true -> snapshot max(rowid) for the upcoming pass and log
1424	// "scheduled to run"; using an explicit boolean
1425	// (instead of max_row_id==0) disambiguates a
1426	// legitimately empty table (MAX returns NULL→0)
1427	// from "snapshot deferred"
1428	// past max(rowid) -> log completion; one-shot cycles (chart, label)
1429	// mark completed and never re-run; cycles with
1430	// complete_repeat_after > 0 (dim) reset last_row_id
1431	// and set snapshot_pending so the next firing takes
1432	// a fresh snapshot
1433	// else -> run one cleanup_loop slice and re-arm short timer
1434	struct cleanup_cycle {
1435	// descriptor (immutable)
1436	const char *select_sql; // SELECT id, rowid FROM <table> WHERE rowid > ?
1437	const char *max_rowid_sql; // SELECT MAX(rowid) FROM <table>
1438	bool (check_cb)(nd_uuid_t , sqlite3_stmt **, bool);
1439	void (action_cb)(nd_uuid_t , sqlite3_stmt **, bool);
1440	bool check_flag;
1441	bool action_flag;
1442	int repeat_after; // seconds to next slice after a partial pass
1443	int complete_repeat_after; // seconds to next pass after full completion;
1444	// 0 means one-shot (don't re-run after first done)
1445	size_t worker_event; // worker_is_busy id
1446	const char *label_singular; // "Dimension" / "Chart" / "Chart label" (INFO logs)
1447	const char *label_plural; // "Dimensions" / "Charts" / "Chart labels" (DEBUG summary)
1448	const char *label_lower; // "dimensions" / "charts" / "chart labels" (DEBUG checking)
1449
1450	// mutable state (per-cycle)
1451	time_t next_execution_t;
1452	uint64_t last_row_id;
1453	uint64_t max_row_id;
1454	bool snapshot_pending; // true => take a fresh max(rowid) snapshot at the next entry past the timer
1455	bool completed; // for one-shot cycles only
1456	};
1457
1458	static bool run_cleanup_cycle(struct cleanup_cycle c, struct meta_config_s wc)
1459	{
1460	if (c->complete_repeat_after == 0 && c->completed)
1461	return true;
1462
1463	time_t now = now_realtime_sec();
1464
1465	if (!c->next_execution_t) {
1466	c->next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
1467	c->snapshot_pending = true;
1468	}
1469
1470	if (c->next_execution_t > now)
1471	return true;
1472
1473	if (c->snapshot_pending) {
1474	c->max_row_id = get_rowid_from_statement(c->max_rowid_sql);
1475	c->snapshot_pending = false;
1476	nd_log(NDLS_DAEMON, NDLP_INFO,
1477	"%s metadata check has been scheduled to run (max id = %" PRIu64 ")",
1478	c->label_singular, c->max_row_id);
1479	}
1480
1481	// No `c->max_row_id &&` guard: a legitimately empty table snapshots to 0,
1482	// and `0 >= 0` correctly takes the completion branch (one-shot cycles
1483	// mark completed; dim re-arms for the next pass).
1484	if (c->last_row_id >= c->max_row_id) {
1485	nd_log(NDLS_DAEMON, NDLP_INFO, "%s metadata check completed", c->label_singular);
1486	if (c->complete_repeat_after) {
1487	// Re-arm for another full pass; the snapshot is deferred to the
1488	// next entry past the timer so it isn't stale by then.
1489	c->next_execution_t = now + c->complete_repeat_after;
1490	c->last_row_id = 0;
1491	c->snapshot_pending = true;
1492	}
1493	else
1494	c->completed = true;
1495	return true;
1496	}
1497
1498	sqlite3_stmt *res = NULL;
1499	if (!PREPARE_STATEMENT(db_meta, c->select_sql, &res))
1500	return true;
1501
1502	uint32_t total_checked = 0;
1503	uint32_t total_deleted = 0;
1504
1505	nd_log(NDLS_DAEMON, NDLP_DEBUG,
1506	"Checking %s starting after row %" PRIu64, c->label_lower, c->last_row_id);
1507
1508	worker_is_busy(c->worker_event);
1509
1510	sqlite3_stmt *check_res = NULL;
1511	sqlite3_stmt *action_res = NULL;
1512
1513	(void) run_cleanup_loop(
1514	res, wc,
1515	c->check_cb, c->action_cb,
1516	&total_checked, &total_deleted,
1517	&c->last_row_id,
1518	&check_res, &action_res,
1519	c->check_flag, c->action_flag);
1520
1521	SQLITE_FINALIZE(check_res);
1522	SQLITE_FINALIZE(action_res);
1523
1524	now = now_realtime_sec();
1525	c->next_execution_t = now + c->repeat_after;
1526
1527	nd_log_daemon(NDLP_DEBUG,
1528	"%s checked %u, deleted %u. Checks will resume in %d seconds",
1529	c->label_plural, total_checked, total_deleted, c->repeat_after);
1530
1531	SQLITE_FINALIZE(res);
1532
1533	worker_is_idle();
1534	return false;
1535	}
1536
1537	static struct cleanup_cycle dim_cleanup_cycle = {
1538	.select_sql = SELECT_DIMENSION_LIST,
1539	.max_rowid_sql = "SELECT MAX(rowid) FROM dimension",
1540	.check_cb = dimension_can_be_deleted,
1541	.action_cb = delete_dimension_uuid,
1542	.check_flag = false,
1543	.action_flag = false,
1544	.repeat_after = METADATA_MAINTENANCE_REPEAT,
1545	.complete_repeat_after = 604800, // re-fire weekly: re-snapshots max(rowid) and rescans dimensions added since the previous pass
1546	.worker_event = UV_EVENT_DIMENSION_CLEANUP,
1547	.label_singular = "Dimension",
1548	.label_plural = "Dimensions",
1549	.label_lower = "dimensions",
1550	};
1551
1552	static struct cleanup_cycle chart_cleanup_cycle = {
1553	.select_sql = SELECT_CHART_LIST,
1554	.max_rowid_sql = "SELECT MAX(rowid) FROM chart",
1555	.check_cb = chart_can_be_deleted,
1556	.action_cb = delete_chart_uuid,
1557	.check_flag = true,
1558	.action_flag = false,
1559	.repeat_after = METADATA_MAINTENANCE_REPEAT,
1560	.complete_repeat_after = 0, // one-shot
1561	.worker_event = UV_EVENT_CHART_CLEANUP,
1562	.label_singular = "Chart",
1563	.label_plural = "Charts",
1564	.label_lower = "charts",
1565	};
1566
1567	static struct cleanup_cycle label_cleanup_cycle = {
1568	.select_sql = SELECT_CHART_LABEL_LIST,
1569	.max_rowid_sql = "SELECT MAX(rowid) FROM chart_label",
1570	.check_cb = chart_can_be_deleted,
1571	.action_cb = delete_chart_uuid,
1572	.check_flag = false,
1573	.action_flag = true,
1574	.repeat_after = METADATA_LABEL_CHECK_INTERVAL,
1575	.complete_repeat_after = 0, // one-shot
1576	.worker_event = UV_EVENT_CHART_LABEL_CLEANUP,
1577	.label_singular = "Chart label",
1578	.label_plural = "Chart labels",
1579	.label_lower = "chart labels",
1580	};
1581
1582	static void cleanup_health_log(struct meta_config_s *config)
1583	{
1584	static time_t next_execution_t = 0;
1585
1586	time_t now = now_realtime_sec();
1587
1588	if (!next_execution_t)
1589	next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
1590
1591	if (next_execution_t && next_execution_t > now)
1592	return;
1593
1594	next_execution_t = now + METADATA_HEALTH_LOG_INTERVAL;
1595
1596	RRDHOST *host;
1597	worker_is_busy(UV_EVENT_HEALTH_LOG_CLEANUP);
1598
1599	dfe_start_reentrant(rrdhost_root_index, host)
1600	{
1601	sql_health_alarm_log_cleanup(host);
1602	if (unlikely(SHUTDOWN_REQUESTED(config)))
1603	break;
1604	}
1605	dfe_done(host);
1606
1607	if (unlikely(SHUTDOWN_REQUESTED(config))) {
1608	worker_is_idle();
1609	return;
1610	}
1611
1612	(void) db_execute(db_meta, SQL_DELETE_ORPHAN_HEALTH_LOG, NULL);
1613	(void) db_execute(db_meta, SQL_DELETE_ORPHAN_HEALTH_LOG_DETAIL, NULL);
1614	(void) db_execute(db_meta, SQL_DELETE_ORPHAN_ALERT_VERSION, NULL);
1615	worker_is_idle();
1616	}
1617
1618	//
1619	// EVENT LOOP STARTS HERE
1620	//
1621
1622
1623	static bool metadata_enq_cmd(cmd_data_t *cmd, bool wait_on_full)
1624	{
1625	if(unlikely(!__atomic_load_n(&meta_config.initialized, __ATOMIC_RELAXED)))
1626	return false;
1627
1628	bool added = push_cmd(&meta_config.cmd_pool, (void *)cmd, wait_on_full);
1629	if (added)
1630	(void) uv_async_send(&meta_config.async);
1631	return added;
1632	}
1633
1634	static cmd_data_t metadata_deq_cmd()
1635	{
1636	cmd_data_t ret;
1637	ret.opcode = METADATA_DATABASE_NOOP;
1638	(void) pop_cmd(&meta_config.cmd_pool, (cmd_data_t *) &ret);
1639	return ret;
1640	}
1641
1642	static void async_cb(uv_async_t *handle __maybe_unused)
1643	{
1644	;
1645	}
1646
1647	#define TIMER_INITIAL_PERIOD_MS (1000)
1648	#define TIMER_REPEAT_PERIOD_MS (1000)
1649
1650	static void timer_cb(uv_timer_t *handle)
1651	{
1652	struct meta_config_s *config = handle->data;
1653	if (config->metadata_check_after < now_realtime_sec())
1654	config->store_metadata = true;
1655	}
1656
1657	void vacuum_database(sqlite3 database, const char db_alias, int threshold, int vacuum_pc, time_t *next_run)
1658	{
1659	time_t now = now_realtime_sec();
1660	if (next_run && *next_run > now)
1661	return;
1662
1663	if (next_run)
1664	*next_run = now + DATABASE_VACUUM_FREQUENCY_SECONDS;
1665
1666	int free_pages = get_free_page_count(database);
1667	int total_pages = get_database_page_count(database);
1668
1669	if (!threshold)
1670	threshold = DATABASE_FREE_PAGES_THRESHOLD_PC;
1671
1672	if (!vacuum_pc)
1673	vacuum_pc = DATABASE_FREE_PAGES_VACUUM_PC;
1674
1675	if (free_pages > (total_pages * threshold / 100)) {
1676	int do_free_pages = (int)(free_pages * vacuum_pc / 100);
1677	nd_log(NDLS_DAEMON, NDLP_DEBUG, "%s: Freeing %d database pages", db_alias, do_free_pages);
1678
1679	char sql[128];
1680	snprintfz(sql, sizeof(sql) - 1, "PRAGMA incremental_vacuum(%d)", do_free_pages);
1681	(void)db_execute(database, sql, NULL);
1682	}
1683	}
1684
1685	#define SQL_SELECT_HOST_CTX_CHART_DIM_LIST \
1686	"SELECT d.dim_id, d.rowid FROM chart c, dimension d WHERE c.chart_id = d.chart_id AND c.rowid = @rowid"
1687
1688	static bool clean_host_chart_dimensions(sqlite3_stmt *res, int64_t chart_row_id, size_t checked, size_t *deleted)
1689	{
1690	struct meta_config_s *config = &meta_config;
1691
1692	bool can_continue = false;
1693
1694	if (!*res) {
1695	if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_HOST_CTX_CHART_DIM_LIST, res))
1696	return false;
1697	}
1698	int param = 0;
1699	SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, chart_row_id));
1700	param = 0;
1701
1702	sqlite3_stmt *dim_del_stmt = NULL;
1703
1704	can_continue = true;
1705	while (can_continue && sqlite3_step_monitored(*res) == SQLITE_ROW) {
1706	nd_uuid_t dim_uuid;
1707
1708	if (!sqlite3_column_uuid_copy(*res, 0, dim_uuid))
1709	continue;
1710
1711	int64_t dimension_id = sqlite3_column_int64(*res, 1);
1712
1713	if (dimension_can_be_deleted(&dim_uuid, NULL, false)) {
1714	delete_dimension_by_rowid(&dim_del_stmt, dimension_id, &dim_uuid);
1715	(*deleted)++;
1716	}
1717	(*checked)++;
1718	can_continue = (!SHUTDOWN_REQUESTED(config)) && sql_metadata_wal_size_acceptable();
1719	}
1720	SQLITE_FINALIZE(dim_del_stmt);
1721
1722	done:
1723	REPORT_BIND_FAIL(*res, param);
1724	SQLITE_RESET(*res);
1725	return can_continue;
1726	}
1727
1728	#define SQL_SELECT_HOST_CTX_CHART_LIST "SELECT rowid, context FROM chart WHERE host_id = @host"
1729
1730	static void cleanup_host_context_metadata(Pvoid_t CTX_JudyL, void *data)
1731	{
1732	if (!CTX_JudyL \|\| !data)
1733	return;
1734
1735	struct meta_config_s *config = &meta_config;
1736
1737	RRDHOST *host = data;
1738
1739	sqlite3_stmt *res = NULL;
1740	sqlite3_stmt *dimension_res = NULL;
1741	sqlite3_stmt *context_res = NULL;
1742
1743	if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_HOST_CTX_CHART_LIST, &res))
1744	return;
1745
1746	Word_t num_of_contexts = JudyLCount(CTX_JudyL, 0, -1, PJE0);
1747
1748	nd_log_daemon(NDLP_DEBUG, "Verifying the retention of %zu contexts for host %s", num_of_contexts, rrdhost_hostname(host));
1749
1750	int param = 0;
1751	SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
1752
1753	param = 0;
1754	Pvoid_t *Pvalue;
1755	int64_t chart_row_id;
1756
1757	size_t deleted = 0;
1758	size_t checked = 0;
1759
1760	bool can_continue = true;
1761	while (can_continue && sqlite3_step_monitored(res) == SQLITE_ROW) {
1762	chart_row_id = sqlite3_column_int64(res, 0);
1763	const char context = (char )sqlite3_column_text(res, 1);
1764	STRING *ctx = string_strdupz(context);
1765	Pvalue = JudyLGet(CTX_JudyL, (Word_t)ctx, PJE0);
1766	if (Pvalue) {
1767	can_continue = clean_host_chart_dimensions(&dimension_res, chart_row_id, &checked, &deleted);
1768	ctx_delete_metadata_cleanup_context(&context_res, &host->host_id.uuid, context);
1769	}
1770	string_freez(ctx);
1771	can_continue = can_continue && (!SHUTDOWN_REQUESTED(config)) && sql_metadata_wal_size_acceptable();
1772	}
1773	SQLITE_FINALIZE(dimension_res);
1774	SQLITE_FINALIZE(context_res);
1775
1776	nd_log_daemon(
1777	NDLP_DEBUG,
1778	"Verified the contexts of host %s (Checked %zu metrics and removed %zu)",
1779	rrdhost_hostname(host),
1780	checked,
1781	deleted);
1782
1783	done:
1784	REPORT_BIND_FAIL(res, param);
1785	SQLITE_FINALIZE(res);
1786	}
1787
1788	void run_metadata_cleanup(struct meta_config_s *config)
1789	{
1790	static time_t next_context_list_cleanup = 0;
1791	static time_t next_vacuum_run = 0;
1792
1793	time_t now = now_realtime_sec();
1794
1795	if (!next_context_list_cleanup)
1796	next_context_list_cleanup = now + 5;
1797
1798	if (next_context_list_cleanup < now && sql_metadata_wal_size_acceptable()) {
1799	RRDHOST *host;
1800	worker_is_busy(UV_EVENT_CTX_CLEANUP);
1801	dfe_start_reentrant(rrdhost_root_index, host) {
1802	ctx_get_context_list_to_cleanup(&host->host_id.uuid, cleanup_host_context_metadata, host);
1803	if (SHUTDOWN_REQUESTED(config) \|\| false == sql_metadata_wal_size_acceptable())
1804	break;
1805	}
1806	dfe_done(host);
1807	worker_is_idle();
1808	next_context_list_cleanup = now_realtime_sec() + METADATA_MAINTENANCE_CTX_CLEAN_REPEAT;
1809	}
1810
1811	if (unlikely(SHUTDOWN_REQUESTED(config)))
1812	return;
1813
1814	if (run_cleanup_cycle(&dim_cleanup_cycle, config))
1815	if (run_cleanup_cycle(&chart_cleanup_cycle, config))
1816	run_cleanup_cycle(&label_cleanup_cycle, config);
1817
1818	cleanup_health_log(config);
1819
1820	if (unlikely(SHUTDOWN_REQUESTED(config)))
1821	return;
1822
1823	vacuum_database(db_meta, "METADATA", DATABASE_FREE_PAGES_THRESHOLD_PC, DATABASE_FREE_PAGES_VACUUM_PC, &next_vacuum_run);
1824
1825	(void) sqlite3_wal_checkpoint(db_meta, NULL);
1826	}
1827
1828	struct host_context_load_thread {
1829	ND_THREAD *thread;
1830	RRDHOST *host;
1831	sqlite3 *db_meta_thread;
1832	sqlite3 *db_context_thread;
1833	bool busy;
1834	bool finished;
1835	};
1836
1837	__thread sqlite3 *db_meta_thread = NULL;
1838	__thread sqlite3 *db_context_thread = NULL;
1839	__thread bool main_context_thread = false;
1840
1841	extern uv_sem_t ctx_sem;
1842	static void restore_host_context(void *arg)
1843	{
1844	struct host_context_load_thread *hclt = arg;
1845	RRDHOST *host = hclt->host;
1846
1847	if (!host)
1848	return;
1849
1850	if (unlikely(exit_initiated_get())) {
1851	__atomic_store_n(&hclt->finished, true, __ATOMIC_RELEASE);
1852	return;
1853	}
1854
1855	if (!db_meta_thread) {
1856	if (hclt->db_meta_thread) {
1857	db_meta_thread = hclt->db_meta_thread;
1858	db_context_thread = hclt->db_context_thread;
1859	} else {
1860	char sqlite_database[FILENAME_MAX + 1];
1861	snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/netdata-meta.db", netdata_configured_cache_dir);
1862	int rc = sqlite3_open_v2(sqlite_database, &db_meta_thread, SQLITE_OPEN_READONLY \| SQLITE_OPEN_NOMUTEX, NULL);
1863	if (rc != SQLITE_OK)
1864	sql_close_thread_db_safe(&db_meta_thread);
1865
1866	snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/context-meta.db", netdata_configured_cache_dir);
1867	rc = sqlite3_open_v2(sqlite_database, &db_context_thread, SQLITE_OPEN_READONLY \| SQLITE_OPEN_NOMUTEX, NULL);
1868	if (rc != SQLITE_OK)
1869	sql_close_thread_db_safe(&db_context_thread);
1870
1871	hclt->db_meta_thread = db_meta_thread;
1872	hclt->db_context_thread = db_context_thread;
1873	}
1874	}
1875
1876	usec_t started_ut = now_monotonic_usec(); (void)started_ut;
1877	rrdhost_load_rrdcontext_data(host);
1878	usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
1879
1880	char load_duration[64];
1881	duration_snprintf(load_duration, sizeof(load_duration), (int64_t)(ended_ut - started_ut), "us", true);
1882	nd_log_daemon(NDLP_DEBUG, "Contexts for host %s loaded in %s", rrdhost_hostname(host), load_duration);
1883
1884	rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
1885	pulse_host_status(host, 0, 0); // this will detect the receiver status
1886
1887	aclk_queue_node_info(host, false);
1888
1889	if (IS_VIRTUAL_HOST_OS(host)) {
1890	uv_sem_post(&ctx_sem);
1891	}
1892
1893	// Check and clear the thread local variables
1894	if (!main_context_thread) {
1895	db_meta_thread = NULL;
1896	db_context_thread = NULL;
1897	}
1898
1899	__atomic_store_n(&hclt->finished, true, __ATOMIC_RELEASE);
1900	}
1901
1902	// Callback after scan of hosts is done
1903	static void after_ctx_hosts_load(uv_work_t *req, int status __maybe_unused)
1904	{
1905	worker_data_t *worker = req->data;
1906	struct meta_config_s *config = worker->config;
1907	config->ctx_load_running = false;
1908	return_worker(&config->worker_pool, worker);
1909	}
1910
1911	static bool cleanup_finished_threads(struct host_context_load_thread hclt, size_t max_thread_slots, bool wait, size_t free_slot)
1912	{
1913	if (!hclt)
1914	return false;
1915
1916	bool found_slot = false;
1917
1918	size_t loop_count = 20;
1919	while (loop_count--) {
1920	for (size_t index = 0; index < max_thread_slots; index++) {
1921	if (free_slot && false == __atomic_load_n(&(hclt[index].busy), __ATOMIC_ACQUIRE)) {
1922	found_slot = true;
1923	*free_slot = index;
1924	break;
1925	}
1926	if (__atomic_load_n(&(hclt[index].finished), __ATOMIC_RELAXED) \|\|
1927	(wait && __atomic_load_n(&(hclt[index].busy), __ATOMIC_ACQUIRE))) {
1928
1929	int rc = nd_thread_join(hclt[index].thread);
1930	if (rc)
1931	nd_log_daemon(NDLP_WARNING, "Failed to join thread, rc = %d", rc);
1932	__atomic_store_n(&(hclt[index].busy), false, __ATOMIC_RELEASE);
1933	__atomic_store_n(&(hclt[index].finished), false, __ATOMIC_RELEASE);
1934	found_slot = true;
1935	if (free_slot) {
1936	*free_slot = index;
1937	break;
1938	}
1939	}
1940	}
1941	if (found_slot \|\| wait)
1942	break;
1943	sleep_usec(10 * USEC_PER_MS);
1944	}
1945	return found_slot \|\| wait;
1946	}
1947
1948	void reset_host_context_load_flag()
1949	{
1950	RRDHOST *host;
1951	dfe_start_reentrant(rrdhost_root_index, host)
1952	{
1953	rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
1954	}
1955	dfe_done(host);
1956	}
1957
1958	static void ctx_hosts_load(uv_work_t *req)
1959	{
1960	register_libuv_worker_jobs();
1961
1962	worker_data_t *worker = req->data;
1963	struct meta_config_s *config = worker->config;
1964
1965	worker_is_busy(UV_EVENT_HOST_CONTEXT_LOAD);
1966	usec_t started_ut = now_monotonic_usec(); (void)started_ut;
1967
1968	RRDHOST *host;
1969
1970	size_t max_threads = netdata_conf_cpus();
1971	if (max_threads < 1)
1972	max_threads = 1;
1973
1974	nd_log(NDLS_DAEMON, NDLP_DEBUG, "Using %zu threads for context loading", max_threads);
1975	struct host_context_load_thread hclt = max_threads > 1 ? callocz(max_threads, sizeof(hclt)) : NULL;
1976
1977	size_t thread_index = 0;
1978	main_context_thread = true;
1979	size_t host_count = 0;
1980	size_t sync_exec = 0;
1981	size_t async_exec = 0;
1982
1983	for (int pass=0 ; pass < 2 ; pass++) {
1984	dfe_start_reentrant(rrdhost_root_index, host) {
1985	// pass 0 will do vnodes (skip the rest)
1986	// pass 1 will do the rest (skip vnodes)
1987	if (pass == IS_VIRTUAL_HOST_OS(host))
1988	continue;
1989
1990	if (!rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
1991	continue;
1992
1993	if (unlikely(SHUTDOWN_REQUESTED(config)))
1994	break;
1995
1996	nd_log_daemon(NDLP_DEBUG, "Loading context for host %s", rrdhost_hostname(host));
1997
1998	int rc = 0;
1999	bool thread_found = cleanup_finished_threads(hclt, max_threads, false, &thread_index);
2000	if (thread_found) {
2001	__atomic_store_n(&hclt[thread_index].busy, true, __ATOMIC_RELAXED);
2002	hclt[thread_index].host = host;
2003	hclt[thread_index].thread = nd_thread_create("CTXLOAD", NETDATA_THREAD_OPTION_DEFAULT, restore_host_context, &hclt[thread_index]);
2004	rc = (hclt[thread_index].thread == NULL);
2005	async_exec += (rc == 0);
2006	// if it failed, mark the thread slot as free
2007	if (rc)
2008	__atomic_store_n(&hclt[thread_index].busy, false, __ATOMIC_RELAXED);
2009	}
2010	// if single thread, thread creation failure or failure tofind slot
2011	if (rc \|\| !thread_found) {
2012	sync_exec++;
2013	struct host_context_load_thread hclt_sync = {.host = host};
2014	restore_host_context(&hclt_sync);
2015	}
2016	host_count++;
2017	}
2018	dfe_done(host);
2019	}
2020
2021	bool should_clean_threads = cleanup_finished_threads(hclt, max_threads, true, NULL);
2022
2023	if (should_clean_threads) {
2024	for (size_t index = 0; index < max_threads; index++) {
2025	sql_close_thread_db_safe(&hclt[index].db_meta_thread);
2026	sql_close_thread_db_safe(&hclt[index].db_context_thread);
2027	}
2028	}
2029
2030	usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
2031	char load_duration[64];
2032	duration_snprintf(load_duration, sizeof(load_duration), (int64_t)(ended_ut - started_ut), "us", true);
2033
2034	nd_log_daemon(
2035	NDLP_INFO,
2036	"Contexts for %zu hosts loaded: %zu delegated to %zu threads, %zu handled directly, in %s.",
2037	host_count,
2038	async_exec,
2039	max_threads,
2040	sync_exec,
2041	load_duration);
2042
2043	sql_close_thread_db_safe(&db_meta_thread);
2044	sql_close_thread_db_safe(&db_context_thread);
2045
2046	freez(hclt);
2047	worker_is_idle();
2048	}
2049
2050	// Callback after scan of hosts is done
2051	static void after_metadata_hosts(uv_work_t *req, int status __maybe_unused)
2052	{
2053	worker_data_t *worker = req->data;
2054	struct meta_config_s *config = worker->config;
2055
2056	bool first = true;
2057	Word_t Index = 0;
2058	Pvoid_t *Pvalue;
2059	while ((Pvalue = JudyLFirstThenNext(config->ae_DelJudyL, &Index, &first))) {
2060	ALARM_ENTRY ae = (ALARM_ENTRY ) Index;
2061	if(!__atomic_load_n(&ae->pending_save_count, __ATOMIC_RELAXED)) {
2062	health_alarm_log_free_one_nochecks_nounlink(ae);
2063	(void) JudyLDel(&config->ae_DelJudyL, Index, PJE0);
2064	first = true;
2065	Index = 0;
2066	}
2067	}
2068
2069	config->metadata_running = false;
2070	return_worker(&config->worker_pool, worker);
2071	}
2072
2073	#ifdef ENABLE_DBENGINE
2074	#define GET_UUID_LIST "SELECT dim_id FROM dimension"
2075	size_t populate_metrics_from_database(void mrg, void (populate_cb)(void mrg, Word_t section, nd_uuid_t uuid))
2076	{
2077	sqlite3_stmt *res = NULL;
2078	sqlite3 *local_meta_db = NULL;
2079
2080	char sqlite_database[FILENAME_MAX + 1];
2081	snprintfz(sqlite_database, sizeof(sqlite_database) - 1, "%s/netdata-meta.db", netdata_configured_cache_dir);
2082	int rc = sqlite3_open_v2(sqlite_database, &local_meta_db, SQLITE_OPEN_READONLY \| SQLITE_OPEN_NOMUTEX, NULL);
2083	if (rc != SQLITE_OK) {
2084	sqlite3_close_v2(local_meta_db);
2085	local_meta_db = NULL;
2086	}
2087
2088	if (local_meta_db)
2089	(void)db_execute(local_meta_db, "PRAGMA cache_size=10000", NULL);
2090
2091	if (!PREPARE_STATEMENT(local_meta_db ? local_meta_db : db_meta, GET_UUID_LIST, &res)) {
2092	sqlite3_close_v2(local_meta_db);
2093	return 0;
2094	}
2095
2096	size_t count = 0;
2097
2098	usec_t started_ut = now_monotonic_usec();
2099	while (sqlite3_step(res) == SQLITE_ROW) {
2100	nd_uuid_t uuid;
2101	if (!sqlite3_column_uuid_copy(res, 0, uuid))
2102	continue;
2103
2104	for (size_t tier = 0; tier < nd_profile.storage_tiers ; tier++) {
2105	if (unlikely(!multidb_ctx[tier]))
2106	continue;
2107
2108	populate_cb(mrg, (Word_t)multidb_ctx[tier], &uuid);
2109	}
2110	count++;
2111	}
2112
2113	SQLITE_FINALIZE(res);
2114	sqlite3_close_v2(local_meta_db);
2115	COMPUTE_DURATION(report_duration, "us", started_ut, now_monotonic_usec());
2116	nd_log_daemon(NDLP_INFO, "MRG: Loaded %zu metrics from database in %s", count, report_duration);
2117	return count;
2118	}
2119	#endif
2120
2121	static void metadata_scan_host(struct meta_config_s config, RRDHOST host, bool is_worker, BUFFER *work_buffer)
2122	{
2123	static bool skip_models = false;
2124	RRDSET *st;
2125	int rc;
2126
2127	sqlite3_stmt *ml_load_stmt = NULL;
2128	sqlite3_stmt *store_dimension = NULL;
2129	sqlite3_stmt *store_chart = NULL;
2130
2131	bool load_ml_models = is_worker;
2132
2133	bool host_need_recheck = false;
2134	(void)db_execute(db_meta, "BEGIN TRANSACTION", NULL);
2135
2136	rrdset_foreach_reentrant(st, host) {
2137
2138	if (SHUTDOWN_REQUESTED(config))
2139	break;
2140
2141	if(rrdset_flag_check(st, RRDSET_FLAG_METADATA_UPDATE)) {
2142
2143	rrdset_flag_clear(st, RRDSET_FLAG_METADATA_UPDATE);
2144
2145	if (is_worker)
2146	worker_is_busy(UV_EVENT_STORE_CHART);
2147
2148	rc = check_and_update_chart_labels(st, work_buffer);
2149	if (unlikely(rc))
2150	error_report("METADATA: 'host:%s': Failed to update labels for chart %s", rrdhost_hostname(host), rrdset_name(st));
2151
2152	rc = store_chart_metadata(st, &store_chart);
2153	if (unlikely(rc)) {
2154	host_need_recheck = true;
2155	rrdset_flag_set(st, RRDSET_FLAG_METADATA_UPDATE);
2156	error_report(
2157	"METADATA: 'host:%s': Failed to store metadata for chart %s",
2158	rrdhost_hostname(host),
2159	rrdset_name(st));
2160	}
2161	if (is_worker)
2162	worker_is_idle();
2163	}
2164
2165	RRDDIM *rd;
2166	rrddim_foreach_read(rd, st) {
2167	if (load_ml_models) {
2168	if (rrddim_flag_check(rd, RRDDIM_FLAG_ML_MODEL_LOAD)) {
2169	rrddim_flag_clear(rd, RRDDIM_FLAG_ML_MODEL_LOAD);
2170	if (likely(!skip_models)) {
2171	if (is_worker)
2172	worker_is_busy(UV_EVENT_METADATA_ML_LOAD);
2173
2174	skip_models = ml_dimension_load_models(rd, &ml_load_stmt);
2175
2176	if (is_worker)
2177	worker_is_idle();
2178	}
2179	}
2180	}
2181
2182	if(likely(!rrddim_flag_check(rd, RRDDIM_FLAG_METADATA_UPDATE)))
2183	continue;
2184
2185	rrddim_flag_clear(rd, RRDDIM_FLAG_METADATA_UPDATE);
2186
2187	if (rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN))
2188	rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
2189	else
2190	rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
2191
2192	if (is_worker)
2193	worker_is_busy(UV_EVENT_STORE_DIMENSION);
2194
2195	rc = store_dimension_metadata(rd, &store_dimension);
2196	if (unlikely(rc)) {
2197	host_need_recheck = true;
2198	rrddim_flag_set(rd, RRDDIM_FLAG_METADATA_UPDATE);
2199	error_report(
2200	"METADATA: 'host:%s': Failed to store dimension metadata for chart %s. dimension %s",
2201	rrdhost_hostname(host),
2202	rrdset_name(st),
2203	rrddim_name(rd));
2204	}
2205
2206	if (is_worker)
2207	worker_is_idle();
2208	}
2209	rrddim_foreach_done(rd);
2210	}
2211	rrdset_foreach_done(st);
2212
2213	(void)db_execute(db_meta, "COMMIT TRANSACTION", NULL);
2214	if (host_need_recheck)
2215	rrdhost_flag_set(host,RRDHOST_FLAG_METADATA_UPDATE);
2216
2217	SQLITE_FINALIZE(ml_load_stmt);
2218	SQLITE_FINALIZE(store_dimension);
2219	SQLITE_FINALIZE(store_chart);
2220	}
2221
2222
2223	static void store_host_and_system_info(RRDHOST *host)
2224	{
2225	rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_INFO);
2226
2227	if (unlikely(store_host_systeminfo(host))) {
2228	error_report("METADATA: 'host:%s': Failed to store host updated system information in the database", rrdhost_hostname(host));
2229	rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_INFO \| RRDHOST_FLAG_METADATA_UPDATE);
2230	}
2231
2232	if (unlikely(store_host_metadata(host))) {
2233	error_report("METADATA: 'host:%s': Failed to store host info in the database", rrdhost_hostname(host));
2234	rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_INFO \| RRDHOST_FLAG_METADATA_UPDATE);
2235	}
2236	}
2237
2238	struct judy_list_t {
2239	Pvoid_t JudyL;
2240	Word_t count;
2241	};
2242
2243	static void do_pending_uuid_deletion(struct meta_config_s config, struct judy_list_t pending_uuid_deletion)
2244	{
2245	if (!pending_uuid_deletion)
2246	return;
2247
2248	worker_is_busy(UV_EVENT_UUID_DELETION);
2249
2250	usec_t started_ut = now_monotonic_usec(); (void)started_ut;
2251
2252	size_t entries = pending_uuid_deletion->count;
2253	Word_t Index = 0;
2254	bool first = true;
2255	Pvoid_t *Pvalue;
2256	while ((Pvalue = JudyLFirstThenNext(pending_uuid_deletion->JudyL, &Index, &first))) {
2257	if (!*Pvalue)
2258	continue;
2259
2260	nd_uuid_t uuid = Pvalue;
2261	if (likely(!SHUTDOWN_REQUESTED(config))) {
2262	if (dimension_can_be_deleted(uuid, NULL, false))
2263	delete_dimension_uuid(uuid, NULL, false);
2264	}
2265
2266	freez(uuid);
2267	}
2268	(void) JudyLFreeArray(&pending_uuid_deletion->JudyL, PJE0);
2269	freez(pending_uuid_deletion);
2270
2271	usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
2272	nd_log_daemon(
2273	NDLP_DEBUG,
2274	"Processed %zu dimension delete items in %0.2f ms",
2275	entries,
2276	(double)(ended_ut - started_ut) / USEC_PER_MS);
2277
2278	worker_is_idle();
2279	}
2280
2281	static void store_ctx_cleanup_list(struct meta_config_s config, struct judy_list_t pending_ctx_cleanup_list)
2282	{
2283	if (!pending_ctx_cleanup_list)
2284	return;
2285
2286	worker_is_busy(UV_EVENT_CTX_CLEANUP_SCHEDULE);
2287
2288	usec_t started_ut = now_monotonic_usec(); (void)started_ut;
2289
2290	size_t entries = pending_ctx_cleanup_list->count;
2291	Word_t Index = 0;
2292	bool first = true;
2293	Pvoid_t *Pvalue;
2294	sqlite3_stmt *res = NULL;
2295	while ((Pvalue = JudyLFirstThenNext(pending_ctx_cleanup_list->JudyL, &Index, &first))) {
2296	if (!*Pvalue)
2297	continue;
2298
2299	struct host_ctx_cleanup_s ctx_cleanup = Pvalue;
2300
2301	if (likely(!SHUTDOWN_REQUESTED(config)))
2302	sql_schedule_host_ctx_cleanup(&res, &ctx_cleanup->host_uuid, string2str(ctx_cleanup->context));
2303
2304	string_freez(ctx_cleanup->context);
2305	freez(ctx_cleanup);
2306	}
2307	(void) JudyLFreeArray(&pending_ctx_cleanup_list->JudyL, PJE0);
2308	freez(pending_ctx_cleanup_list);
2309	SQLITE_FINALIZE(res);
2310
2311	usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
2312	nd_log_daemon(
2313	NDLP_DEBUG,
2314	"Stored %zu host context cleanup items in %0.2f ms",
2315	entries,
2316	(double)(ended_ut - started_ut) / USEC_PER_MS);
2317
2318	worker_is_idle();
2319	}
2320
2321	static void store_alert_transitions(struct judy_list_t *pending_alert_list, bool is_worker, bool cleanup_only)
2322	{
2323	if (!pending_alert_list)
2324	return;
2325
2326	if (cleanup_only)
2327	goto done;
2328
2329	if (is_worker)
2330	worker_is_busy(UV_EVENT_STORE_ALERT_TRANSITIONS);
2331
2332	usec_t started_ut = now_monotonic_usec(); (void)started_ut;
2333
2334	size_t entries = pending_alert_list->count;
2335	Word_t Index = 0;
2336	bool first = true;
2337	Pvoid_t *Pvalue;
2338	while ((Pvalue = JudyLFirstThenNext(pending_alert_list->JudyL, &Index, &first))) {
2339	RRDHOST host = Pvalue;
2340
2341	Pvalue = JudyLGet(pending_alert_list->JudyL, ++Index, PJE0);
2342	ALARM_ENTRY ae = Pvalue;
2343
2344	sql_health_alarm_log_save(host, ae);
2345
2346	__atomic_add_fetch(&ae->pending_save_count, -1, __ATOMIC_RELAXED);
2347	__atomic_add_fetch(&host->health.pending_transitions, -1, __ATOMIC_RELAXED);
2348	}
2349
2350	usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
2351	nd_log(
2352	NDLS_DAEMON,
2353	NDLP_DEBUG,
2354	"Stored and processed %zu alert transitions in %0.2f ms",
2355	entries,
2356	(double)(ended_ut - started_ut) / USEC_PER_MS);
2357
2358	if (is_worker)
2359	worker_is_idle();
2360
2361	done:
2362	(void) JudyLFreeArray(&pending_alert_list->JudyL, PJE0);
2363	freez(pending_alert_list);
2364	}
2365
2366	static int execute_statement(sqlite3_stmt *stmt, bool only_finalize)
2367	{
2368	if (!stmt)
2369	return SQLITE_OK;
2370
2371	int rc = SQLITE_OK;
2372
2373	if (unlikely(only_finalize))
2374	goto done;
2375
2376	rc = sqlite3_step_monitored(stmt);
2377	if (unlikely(rc != SQLITE_DONE))
2378	nd_log_daemon(NDLP_ERR, "Failed to execute sql statement, rc = %d", rc);
2379
2380	done:
2381	SQLITE_FINALIZE(stmt);
2382
2383	return rc == SQLITE_DONE ? SQLITE_OK : rc;
2384	}
2385
2386	static void store_sql_statements(struct judy_list_t *pending_sql_statement, bool is_worker, bool only_finalize)
2387	{
2388	if (!pending_sql_statement)
2389	return;
2390
2391	if (is_worker)
2392	worker_is_busy(METADATA_EXECUTE_STORE_STATEMENT);
2393
2394	usec_t started_ut = now_monotonic_usec();
2395
2396	size_t entries = pending_sql_statement->count;
2397	Word_t Index = 0;
2398	bool first = true;
2399	Pvoid_t *Pvalue;
2400	while ((Pvalue = JudyLFirstThenNext(pending_sql_statement->JudyL, &Index, &first))) {
2401	sqlite3_stmt stmt = Pvalue;
2402	execute_statement(stmt, only_finalize);
2403	}
2404	(void) JudyLFreeArray(&pending_sql_statement->JudyL, PJE0);
2405	freez(pending_sql_statement);
2406
2407	COMPUTE_DURATION(report_duration, "us", started_ut, now_monotonic_usec());
2408	nd_log_daemon(NDLP_DEBUG, "Stored and processed %zu sql statements in %s", entries, report_duration);
2409
2410	if (is_worker)
2411	worker_is_idle();
2412	}
2413
2414	static void meta_store_host_labels(RRDHOST host, BUFFER work_buffer)
2415	{
2416	rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_LABELS);
2417
2418	// Delete existing labels first to handle label removal
2419	int rc = exec_statement_with_uuid(SQL_DELETE_HOST_LABELS, &host->host_id.uuid);
2420	if (unlikely(rc)) {
2421	error_report("METADATA: 'host:%s': failed to delete old host labels", rrdhost_hostname(host));
2422	rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_LABELS \| RRDHOST_FLAG_METADATA_UPDATE);
2423	return;
2424	}
2425
2426	// Store all current labels using prepared statements
2427	rc = store_labels(&host->host_id.uuid, host->rrdlabels, STORE_HOST_LABELS, work_buffer);
2428	if (unlikely(rc)) {
2429	error_report("METADATA: 'host:%s': failed to store host labels", rrdhost_hostname(host));
2430	rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_LABELS \| RRDHOST_FLAG_METADATA_UPDATE);
2431	}
2432	}
2433
2434	static void store_host_claim_id(RRDHOST *host)
2435	{
2436	rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_CLAIMID);
2437	int rc;
2438	ND_UUID uuid = claim_id_get_uuid();
2439	if (!UUIDiszero(uuid))
2440	rc = store_claim_id(&host->host_id.uuid, &uuid.uuid);
2441	else
2442	rc = store_claim_id(&host->host_id.uuid, NULL);
2443
2444	if (unlikely(rc))
2445	rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_CLAIMID \| RRDHOST_FLAG_METADATA_UPDATE);
2446	}
2447
2448	static void store_host_info_and_metadata_with_buffer(RRDHOST host, BUFFER work_buffer)
2449	{
2450	// Store labels (if needed)
2451	if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_LABELS)))
2452	meta_store_host_labels(host, work_buffer);
2453
2454	// Store claim id (if needed)
2455	if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_CLAIMID)))
2456	store_host_claim_id(host);
2457
2458	// Store host and system info (if needed);
2459	if (rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_INFO))
2460	store_host_and_system_info(host);
2461	}
2462
2463	// Public API - creates buffer internally if needed
2464	void store_host_info_and_metadata(RRDHOST *host)
2465	{
2466	store_host_info_and_metadata_with_buffer(host, NULL);
2467	}
2468
2469	static void store_hosts_metadata(struct meta_config_s *config, bool is_worker)
2470	{
2471	RRDHOST *host;
2472	size_t host_count = 0;
2473	usec_t started_ut = now_monotonic_usec();
2474	if (!is_worker) {
2475	dfe_start_reentrant(rrdhost_root_index, host) {
2476	host_count++;
2477	}
2478	dfe_done(host);
2479	if (!host_count)
2480	host_count = 1; // avoid division by zero
2481	}
2482
2483	// Reusable buffer for building SQL statements
2484	BUFFER *work_buffer = buffer_create(1024, NULL);
2485
2486	size_t count = 0;
2487	dfe_start_reentrant(rrdhost_root_index, host)
2488	{
2489	count++;
2490	if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) \|\| !rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_UPDATE))
2491	continue;
2492
2493	rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_UPDATE);
2494
2495	if (SHUTDOWN_REQUESTED(config))
2496	break;
2497
2498	if (is_worker)
2499	worker_is_busy(UV_EVENT_STORE_HOST);
2500
2501	// store labels, claim_id, host and system info (if needed)
2502	store_host_info_and_metadata_with_buffer(host, work_buffer);
2503
2504	if (is_worker)
2505	worker_is_idle();
2506
2507	metadata_scan_host(config, host, is_worker, work_buffer);
2508
2509	if (!is_worker)
2510	nd_log_daemon(NDLP_INFO, "METADATA: Progress of metadata storage: %6.2f%% completed", (100.0 * count / host_count));
2511	}
2512	dfe_done(host);
2513
2514	buffer_free(work_buffer);
2515
2516	if (!is_worker) {
2517	COMPUTE_DURATION(report_duration, "us", started_ut, now_monotonic_usec());
2518	nd_log_daemon(
2519	NDLP_INFO,
2520	"METADATA: Progress of metadata storage: %6.2f%% completed in %s",
2521	(100.0 * count / host_count),
2522	report_duration);
2523	}
2524	}
2525
2526	#define SERVICE_HEARTBEAT 10
2527	void run_maintenace();
2528
2529	// Worker thread to scan hosts for pending metadata to store
2530	static void start_metadata_hosts(uv_work_t *req)
2531	{
2532	static time_t next_maintenance_check = 0;
2533	register_libuv_worker_jobs();
2534
2535	time_t now = now_realtime_sec();
2536	if (now> next_maintenance_check) {
2537	run_maintenace();
2538	next_maintenance_check = now + SERVICE_HEARTBEAT;
2539	}
2540
2541	worker_data_t *worker = req->data;
2542	struct meta_config_s *config = worker->config;
2543
2544	usec_t all_started_ut = now_monotonic_usec();
2545
2546	store_sql_statements((struct judy_list_t *)worker->pending_sql_statement, true, false);
2547
2548	store_alert_transitions((struct judy_list_t *)worker->pending_alert_list, true, false);
2549
2550	// This helper already skips database scheduling once shutdown starts, but it
2551	// still has to release the worker-owned Judy list on that path.
2552	store_ctx_cleanup_list(config, (struct judy_list_t *)worker->pending_ctx_cleanup_list);
2553
2554	worker_is_busy(UV_EVENT_METADATA_STORE);
2555
2556	store_hosts_metadata(config, true);
2557
2558	COMPUTE_DURATION(report_duration, "us", all_started_ut, now_monotonic_usec());
2559	nd_log_daemon(NDLP_DEBUG, "Checking all hosts completed in %s", report_duration);
2560
2561	// This helper already skips dimension deletion once shutdown starts, but it
2562	// still has to release the worker-owned Judy list on that path.
2563	do_pending_uuid_deletion(config, (struct judy_list_t *)worker->pending_uuid_deletion);
2564
2565	if (!SHUTDOWN_REQUESTED(config)) {
2566	run_metadata_cleanup(config);
2567	}
2568
2569	config->metadata_check_after = now_realtime_sec() + METADATA_HOST_CHECK_INTERVAL;
2570	worker_is_idle();
2571	}
2572
2573	#define EVENT_LOOP_NAME "METASYNC"
2574
2575	#define MAX_SHUTDOWN_TIMEOUT_SECONDS (15)
2576	#define SHUTDOWN_SLEEP_INTERVAL_MS (100)
2577	#define CMD_POOL_SIZE (32768)
2578
2579	static void metadata_event_loop(void *arg)
2580	{
2581	struct meta_config_s *config = arg;
2582	uv_thread_set_name_np(EVENT_LOOP_NAME);
2583	service_register(NULL, NULL, NULL);
2584	worker_register(EVENT_LOOP_NAME);
2585
2586	init_cmd_pool(&config->cmd_pool, CMD_POOL_SIZE);
2587
2588	worker_register_job_name(METADATA_DATABASE_NOOP, "noop");
2589	worker_register_job_name(METADATA_DEL_DIMENSION, "delete dimension");
2590	worker_register_job_name(METADATA_STORE_CLAIM_ID, "add claim id");
2591	worker_register_job_name(METADATA_ADD_CTX_CLEANUP, "host ctx cleanup");
2592	worker_register_job_name(METADATA_STORE, "host metadata store");
2593	worker_register_job_name(METADATA_LOAD_HOST_CONTEXT, "host load context");
2594	worker_register_job_name(METADATA_ADD_HOST_AE, "add host alert entry");
2595	worker_register_job_name(METADATA_DEL_HOST_AE, "delete host alert entry");
2596	worker_register_job_name(METADATA_EXECUTE_STORE_STATEMENT, "add sql statement");
2597
2598	uv_loop_t *loop = &config->loop;
2599	fatal_assert(0 == uv_loop_init(loop));
2600	fatal_assert(0 == uv_async_init(loop, &config->async, async_cb));
2601	fatal_assert(0 == uv_timer_init(loop, &config->timer_req));
2602	fatal_assert(0 == uv_timer_start(&config->timer_req, timer_cb, TIMER_INITIAL_PERIOD_MS, TIMER_REPEAT_PERIOD_MS));
2603	loop->data = config;
2604	config->async.data = config;
2605	config->timer_req.data = config;
2606
2607	nd_log(NDLS_DAEMON, NDLP_DEBUG, "Starting metadata sync thread");
2608	config->metadata_check_after = now_realtime_sec() + METADATA_HOST_CHECK_FIRST_CHECK;
2609
2610	worker_data_t *worker;
2611	Pvoid_t *Pvalue;
2612	struct judy_list_t *pending_alert_list = NULL;
2613	struct judy_list_t *pending_ctx_cleanup_list = NULL;
2614	struct judy_list_t *pending_uuid_deletion = NULL;
2615	struct judy_list_t *pending_sql_statement = NULL;
2616
2617	config->initialized = true;
2618	__atomic_store_n(&config->shutdown_requested, false, __ATOMIC_RELAXED);
2619	nd_log_daemon(NDLP_INFO, "METADATA: Synchronization thread is up and running");
2620	completion_mark_complete(&config->start_stop_complete);
2621
2622	while (likely(__atomic_load_n(&config->shutdown_requested, __ATOMIC_RELAXED) == false)) {
2623	nd_uuid_t *uuid;
2624	RRDHOST *host = NULL;
2625	ALARM_ENTRY *ae = NULL;
2626	sqlite3_stmt *stmt;
2627	enum metadata_opcode opcode;
2628
2629	worker_is_idle();
2630	uv_run(loop, UV_RUN_ONCE);
2631
2632	/* wait for commands */
2633	do {
2634	cmd_data_t cmd;
2635	if (config->store_metadata && !config->metadata_running) {
2636	config->store_metadata = false;
2637	opcode = METADATA_STORE;
2638	}
2639	else {
2640	cmd = metadata_deq_cmd();
2641	opcode = cmd.opcode;
2642	}
2643
2644	if (likely(opcode != METADATA_DATABASE_NOOP))
2645	worker_is_busy(opcode);
2646
2647	switch (opcode) {
2648	case METADATA_DATABASE_NOOP:
2649	break;
2650	case METADATA_DEL_DIMENSION:
2651	uuid = (nd_uuid_t *)cmd.param[0];
2652	if (!pending_uuid_deletion)
2653	pending_uuid_deletion = callocz(1, sizeof(*pending_uuid_deletion));
2654
2655	Pvalue = JudyLIns(&pending_uuid_deletion->JudyL, ++pending_uuid_deletion->count, PJE0);
2656	if (unlikely(Pvalue == PJERR)) {
2657	// Failure in Judy, attempt to continue running anyway
2658	// ignore uuid, global cleanup will take care of it
2659	freez(uuid);
2660	}
2661	else
2662	*Pvalue = uuid;
2663	break;
2664	case METADATA_STORE_CLAIM_ID:
2665	store_claim_id((nd_uuid_t )cmd.param[0], (nd_uuid_t )cmd.param[1]);
2666	freez((void *)cmd.param[0]);
2667	freez((void *)cmd.param[1]);
2668	break;
2669
2670	case METADATA_ADD_CTX_CLEANUP:
2671	if (!pending_ctx_cleanup_list)
2672	pending_ctx_cleanup_list = callocz(1, sizeof(*pending_ctx_cleanup_list));
2673
2674	struct host_ctx_cleanup_s ctx_cleanup = (struct host_ctx_cleanup_s )cmd.param[0];
2675	Pvalue = JudyLIns(&pending_ctx_cleanup_list->JudyL, ++pending_ctx_cleanup_list->count, PJE0);
2676	if (unlikely(Pvalue == PJERR)) {
2677	// Failure in Judy, attempt to continue running anyway
2678	// Cleanup structure
2679	string_freez(ctx_cleanup->context);
2680	freez(ctx_cleanup);
2681	}
2682	else
2683	*Pvalue = ctx_cleanup;
2684	break;
2685	case METADATA_STORE:
2686	if (config->metadata_running \|\| unittest_running)
2687	break;
2688
2689	worker = get_worker(&config->worker_pool);
2690	worker->config = config;
2691	worker->pending_alert_list = pending_alert_list;
2692	worker->pending_ctx_cleanup_list = pending_ctx_cleanup_list;
2693	worker->pending_uuid_deletion = pending_uuid_deletion;
2694	worker->pending_sql_statement = pending_sql_statement;
2695	pending_alert_list = NULL;
2696	pending_ctx_cleanup_list = NULL;
2697	pending_uuid_deletion = NULL;
2698	pending_sql_statement = NULL;
2699	config->metadata_running = true;
2700	if (uv_queue_work(loop, &worker->request, start_metadata_hosts, after_metadata_hosts)) {
2701	pending_alert_list = worker->pending_alert_list;
2702	pending_ctx_cleanup_list = worker->pending_ctx_cleanup_list;
2703	pending_uuid_deletion = worker->pending_uuid_deletion;
2704	pending_sql_statement = worker->pending_sql_statement;
2705	config->metadata_running = false;
2706	return_worker(&config->worker_pool, worker);
2707	}
2708	break;
2709	case METADATA_LOAD_HOST_CONTEXT:
2710	if (config->ctx_load_running \|\| unittest_running)
2711	break;
2712
2713	worker = get_worker(&config->worker_pool);
2714	config->ctx_load_running = true;
2715	worker->config = config;
2716	if (uv_queue_work(loop, &worker->request, ctx_hosts_load, after_ctx_hosts_load)) {
2717	config->ctx_load_running = false;
2718	// Fallback reset context so hosts will load on demand
2719	reset_host_context_load_flag();
2720	return_worker(&config->worker_pool, worker);
2721	}
2722	break;
2723	case METADATA_ADD_HOST_AE:
2724	host = (RRDHOST *)cmd.param[0];
2725	ae = (ALARM_ENTRY *)cmd.param[1];
2726
2727	if (!pending_alert_list)
2728	pending_alert_list = callocz(1, sizeof(*pending_alert_list));
2729
2730	Pvalue = JudyLIns(&pending_alert_list->JudyL, ++pending_alert_list->count, PJE0);
2731	if (unlikely(Pvalue == PJERR))
2732	fatal("METASYNC: Failed to insert into pending_alert_list Judy array");
2733	Pvalue = (void )host;
2734
2735	Pvalue = JudyLIns(&pending_alert_list->JudyL, ++pending_alert_list->count, PJE0);
2736	if (unlikely(Pvalue == PJERR))
2737	fatal("METASYNC: Failed to insert into pending_alert_list Judy array");
2738	Pvalue = (void )ae;
2739	break;
2740	case METADATA_DEL_HOST_AE:
2741	Pvalue = JudyLIns(&config->ae_DelJudyL, (Word_t)(void *)cmd.param[0], PJE0);
2742	if (Pvalue == PJERR)
2743	nd_log_daemon(NDLP_ERR, "METADATA: Failed to track alert entry for deletion");
2744	break;
2745	case METADATA_EXECUTE_STORE_STATEMENT:
2746	stmt = (sqlite3_stmt *)cmd.param[0];
2747	if (!pending_sql_statement)
2748	pending_sql_statement = callocz(1, sizeof(*pending_sql_statement));
2749
2750	Pvalue = JudyLIns(&pending_sql_statement->JudyL, ++pending_sql_statement->count, PJE0);
2751	if (unlikely(Pvalue == PJERR)) {
2752	// Fallback execute immediately
2753	execute_statement(stmt, false);
2754	}
2755	else
2756	Pvalue = (void )stmt;
2757	break;
2758	case METADATA_SYNC_SHUTDOWN:
2759	__atomic_store_n(&config->shutdown_requested, true, __ATOMIC_RELAXED);
2760	break;
2761	case METADATA_UNITTEST:;
2762	struct thread_unittest tu = (struct thread_unittest )cmd.param[0];
2763	sleep_usec(1000); // processing takes 1ms
2764	__atomic_fetch_add(&tu->processed, 1, __ATOMIC_SEQ_CST);
2765	break;
2766	default:
2767	break;
2768	}
2769
2770	if (likely(opcode != METADATA_DATABASE_NOOP))
2771	uv_run(loop, UV_RUN_NOWAIT);
2772
2773	} while (opcode != METADATA_DATABASE_NOOP);
2774	}
2775	config->initialized = false;
2776
2777	if (!uv_timer_stop(&config->timer_req))
2778	uv_close((uv_handle_t *)&config->timer_req, NULL);
2779
2780	uv_close((uv_handle_t *)&config->async, NULL);
2781	uv_walk(loop, libuv_close_callback, NULL);
2782
2783	size_t loop_count = (MAX_SHUTDOWN_TIMEOUT_SECONDS * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;
2784
2785	// are we waiting for callbacks?
2786	bool callbacks_pending = (config->metadata_running \|\| config->ctx_load_running);
2787
2788	while (((config->metadata_running \|\| config->ctx_load_running) \|\| uv_loop_alive(loop)) && loop_count > 0) {
2789	callbacks_pending = uv_run(loop, UV_RUN_NOWAIT);
2790	if (!callbacks_pending)
2791	break; // No pending callbacks
2792	sleep_usec(SHUTDOWN_SLEEP_INTERVAL_MS * USEC_PER_MS);
2793	loop_count--;
2794	}
2795
2796	(void)uv_loop_close(loop);
2797
2798	store_alert_transitions(pending_alert_list, false, true);
2799	store_sql_statements(pending_sql_statement, false, true);
2800
2801	if (pending_ctx_cleanup_list) {
2802	Word_t Index = 0;
2803	bool first = true;
2804	while ((Pvalue = JudyLFirstThenNext(pending_ctx_cleanup_list->JudyL, &Index, &first))) {
2805	if (!*Pvalue)
2806	continue;
2807	struct host_ctx_cleanup_s ctx_cleanup = Pvalue;
2808	string_freez(ctx_cleanup->context);
2809	freez(ctx_cleanup);
2810	}
2811	(void)JudyLFreeArray(&pending_ctx_cleanup_list->JudyL, PJE0);
2812	freez(pending_ctx_cleanup_list);
2813	}
2814
2815	if (pending_uuid_deletion) {
2816	Word_t Index = 0;
2817	bool first = true;
2818	Pvoid_t *Pvalue;
2819	while ((Pvalue = JudyLFirstThenNext(pending_uuid_deletion->JudyL, &Index, &first))) {
2820	if (!*Pvalue)
2821	continue;
2822	nd_uuid_t uuid = Pvalue;
2823	freez(uuid);
2824	}
2825	(void)JudyLFreeArray(&pending_uuid_deletion->JudyL, PJE0);
2826	freez(pending_uuid_deletion);
2827	}
2828
2829	release_cmd_pool(&config->cmd_pool);
2830	worker_unregister();
2831	service_exits();
2832	completion_mark_complete(&config->start_stop_complete);
2833	}
2834
2835	void metadata_sync_shutdown(void)
2836	{
2837	cmd_data_t cmd;
2838	memset(&cmd, 0, sizeof(cmd));
2839	cmd.opcode = METADATA_SYNC_SHUTDOWN;
2840
2841	// if we can't sent command return
2842	// This should not happen but if we wait we may not get a completion
2843	// and shutdown will timeout
2844	if (!metadata_enq_cmd(&cmd, true)) {
2845	nd_log_daemon(NDLP_WARNING, "METADATA: Failed to send a shutdown command");
2846	return;
2847	}
2848	nd_log_daemon(NDLP_INFO, "METADATA: Submitted shutdown command, waiting for ACK");
2849
2850	completion_wait_for(&meta_config.start_stop_complete);
2851	completion_destroy(&meta_config.start_stop_complete);
2852
2853	int rc = nd_thread_join(meta_config.thread);
2854	if (rc)
2855	nd_log_daemon(NDLP_ERR, "METADATA: Failed to join synchronization thread");
2856	else
2857	nd_log_daemon(NDLP_INFO, "METADATA: synchronization thread shutdown completed");
2858	}
2859
2860	// -------------------------------------------------------------
2861	// Init function called on agent startup
2862
2863	void metadata_sync_init(void)
2864	{
2865	memset(&meta_config, 0, sizeof(meta_config));
2866	completion_init(&meta_config.start_stop_complete);
2867
2868	init_worker_pool(&meta_config.worker_pool);
2869	meta_config.thread = nd_thread_create("METASYNC", NETDATA_THREAD_OPTION_DEFAULT, metadata_event_loop, &meta_config);
2870	fatal_assert(NULL != meta_config.thread);
2871
2872	// Wait for initialization
2873	completion_wait_for(&meta_config.start_stop_complete);
2874
2875	// Reset the completion, we will use it again during shutdown
2876	completion_reset(&meta_config.start_stop_complete);
2877	}
2878
2879	// Helpers
2880
2881	static inline bool queue_metadata_cmd(enum metadata_opcode opcode, void param0, void param1)
2882	{
2883	cmd_data_t cmd;
2884	cmd.opcode = opcode;
2885	cmd.param[0] = param0;
2886	cmd.param[1] = param1;
2887	return metadata_enq_cmd(&cmd, true);
2888	}
2889
2890	// Public
2891	void metaqueue_delete_dimension_uuid(nd_uuid_t *uuid)
2892	{
2893	if (unlikely(!uuid))
2894	return;
2895
2896	nd_uuid_t use_uuid = mallocz(sizeof(uuid));
2897	uuid_copy(use_uuid, uuid);
2898	if (!queue_metadata_cmd(METADATA_DEL_DIMENSION, use_uuid, NULL))
2899	freez(use_uuid);
2900	}
2901
2902	void metaqueue_store_claim_id(nd_uuid_t host_uuid, nd_uuid_t claim_uuid)
2903	{
2904	if (unlikely(!host_uuid))
2905	return;
2906
2907	nd_uuid_t local_host_uuid = mallocz(sizeof(host_uuid));
2908	nd_uuid_t *local_claim_uuid = NULL;
2909
2910	uuid_copy(local_host_uuid, host_uuid);
2911	if (likely(claim_uuid)) {
2912	local_claim_uuid = mallocz(sizeof(*claim_uuid));
2913	uuid_copy(local_claim_uuid, claim_uuid);
2914	}
2915	if (unlikely(!queue_metadata_cmd(METADATA_STORE_CLAIM_ID, local_host_uuid, local_claim_uuid))) {
2916	freez(local_host_uuid);
2917	freez(local_claim_uuid);
2918	}
2919	}
2920
2921	void metaqueue_ml_load_models(RRDDIM *rd)
2922	{
2923	rrddim_flag_set(rd, RRDDIM_FLAG_ML_MODEL_LOAD);
2924	}
2925
2926	bool metadata_queue_load_host_context()
2927	{
2928	return queue_metadata_cmd(METADATA_LOAD_HOST_CONTEXT, NULL, NULL);
2929	}
2930
2931	void metadata_queue_ctx_host_cleanup(nd_uuid_t host_uuid, const char context)
2932	{
2933	if (unlikely(!host_uuid \|\| !context))
2934	return;
2935
2936	struct host_ctx_cleanup_s ctx_cleanup = mallocz(sizeof(ctx_cleanup));
2937
2938	uuid_copy(ctx_cleanup->host_uuid, *host_uuid);
2939	ctx_cleanup->context = string_strdupz(context);
2940
2941	if (unlikely(!queue_metadata_cmd(METADATA_ADD_CTX_CLEANUP, ctx_cleanup, NULL))) {
2942	string_freez(ctx_cleanup->context);
2943	freez(ctx_cleanup);
2944	}
2945	}
2946
2947	bool metadata_queue_ae_save(RRDHOST host, ALARM_ENTRY ae)
2948	{
2949	if (unlikely(!host \|\| !ae))
2950	return true;
2951
2952	__atomic_add_fetch(&host->health.pending_transitions, 1, __ATOMIC_RELAXED);
2953	__atomic_add_fetch(&ae->pending_save_count, 1, __ATOMIC_RELAXED);
2954
2955	if (unlikely(!queue_metadata_cmd(METADATA_ADD_HOST_AE, host, ae))) {
2956	// Failed to queue, reset counters
2957	__atomic_sub_fetch(&host->health.pending_transitions, 1, __ATOMIC_RELAXED);
2958	__atomic_sub_fetch(&ae->pending_save_count, 1, __ATOMIC_RELAXED);
2959	return false;
2960	}
2961	return true;
2962	}
2963
2964	void metadata_queue_ae_deletion(ALARM_ENTRY *ae)
2965	{
2966	if (unlikely(!ae))
2967	return;
2968
2969	(void) queue_metadata_cmd(METADATA_DEL_HOST_AE, ae, NULL);
2970	}
2971
2972	void metadata_execute_store_statement(sqlite3_stmt *stmt)
2973	{
2974	if (unlikely(!stmt))
2975	return;
2976
2977	(void) queue_metadata_cmd(METADATA_EXECUTE_STORE_STATEMENT, stmt, NULL);
2978	}
2979
2980	void commit_alert_transitions(RRDHOST *host __maybe_unused)
2981	{
2982	(void) queue_metadata_cmd(METADATA_STORE, NULL, NULL);
2983	}
2984
2985	uint64_t sqlite_get_meta_space(void)
2986	{
2987	return sqlite_get_db_space(db_meta);
2988	}
2989
2990	#define SQL_ADD_AGENT_EVENT_LOG \
2991	"INSERT INTO agent_event_log (event_type, version, value, date_created) VALUES " \
2992	" (@event_type, @version, @value, UNIXEPOCH())"
2993
2994	void add_agent_event(event_log_type_t event_id, int64_t value)
2995	{
2996	sqlite3_stmt *res = NULL;
2997
2998	if (!PREPARE_STATEMENT(db_meta, SQL_ADD_AGENT_EVENT_LOG, &res))
2999	return;
3000
3001	int param = 0;
3002	SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, event_id));
3003	SQLITE_BIND_FAIL(done, sqlite3_bind_text(res, ++param, NETDATA_VERSION, -1, SQLITE_STATIC));
3004	SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, value));
3005
3006	param = 0;
3007	int rc = execute_insert(res);
3008	if (rc != SQLITE_DONE)
3009	error_report("Failed to store agent event information, rc = %d", rc);
3010	done:
3011	REPORT_BIND_FAIL(res, param);
3012	SQLITE_FINALIZE(res);
3013	}
3014
3015	void cleanup_agent_event_log(void)
3016	{
3017	(void) db_execute(db_meta, SQL_CLEANUP_AGENT_EVENT_LOG, NULL);
3018	}
3019
3020	#define SQL_GET_AGENT_EVENT_TYPE_MEDIAN \
3021	"SELECT AVG(value) AS median FROM " \
3022	"(SELECT value FROM agent_event_log WHERE event_type = @event ORDER BY value " \
3023	" LIMIT 2 - (SELECT COUNT(*) FROM agent_event_log WHERE event_type = @event) % 2 " \
3024	"OFFSET(SELECT(COUNT(*) - 1) / 2 FROM agent_event_log WHERE event_type = @event)) "
3025
3026	usec_t get_agent_event_time_median(event_log_type_t event_id)
3027	{
3028	static bool initialized[EVENT_AGENT_MAX] = { 0 };
3029	static usec_t median[EVENT_AGENT_MAX] = { 0 };
3030
3031	if(event_id >= EVENT_AGENT_MAX)
3032	return 0;
3033
3034	if(initialized[event_id])
3035	return median[event_id];
3036
3037	sqlite3_stmt *res = NULL;
3038	if (!PREPARE_STATEMENT(db_meta, SQL_GET_AGENT_EVENT_TYPE_MEDIAN, &res))
3039	return 0;
3040
3041	usec_t avg_time = 0;
3042	int param = 0;
3043	SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, event_id));
3044
3045	param = 0;
3046	if (sqlite3_step_monitored(res) == SQLITE_ROW)
3047	avg_time = sqlite3_column_int64(res, 0);
3048
3049	done:
3050	REPORT_BIND_FAIL(res, param);
3051	SQLITE_FINALIZE(res);
3052
3053	median[event_id] = avg_time;
3054	initialized[event_id] = true;
3055	return avg_time;
3056	}
3057
3058	void get_agent_event_time_median_init(void) {
3059	for(event_log_type_t event_id = 1; event_id < EVENT_AGENT_MAX; event_id++)
3060	get_agent_event_time_median(event_id);
3061	}
3062
3063	//
3064	// unitests
3065	//
3066
3067	static void unittest_queue_metadata(void *arg) {
3068	struct thread_unittest *tu = arg;
3069
3070	cmd_data_t cmd;
3071	cmd.opcode = METADATA_UNITTEST;
3072	cmd.param[0] = tu;
3073	cmd.param[1] = NULL;
3074	metadata_enq_cmd(&cmd, true);
3075
3076	do {
3077	__atomic_fetch_add(&tu->added, 1, __ATOMIC_SEQ_CST);
3078	metadata_enq_cmd(&cmd, true);
3079	sleep_usec(10000);
3080	} while (!__atomic_load_n(&tu->join, __ATOMIC_RELAXED));
3081	}
3082
3083	static void *metadata_unittest_threads(void)
3084	{
3085
3086	unsigned done;
3087
3088	struct thread_unittest tu = {
3089	.join = 0,
3090	.added = 0,
3091	.processed = 0,
3092	.done = &done,
3093	};
3094
3095	// Queue messages / Time it
3096	time_t seconds_to_run = 5;
3097	int threads_to_create = 4;
3098	fprintf(
3099	stderr,
3100	"\nChecking metadata queue using %d threads for %lld seconds...\n",
3101	threads_to_create,
3102	(long long)seconds_to_run);
3103
3104	ND_THREAD *threads[threads_to_create];
3105	tu.join = 0;
3106	for (int i = 0; i < threads_to_create; i++) {
3107	char buf[100 + 1];
3108	snprintf(buf, sizeof(buf) - 1, "META[%d]", i);
3109	threads[i] = nd_thread_create(buf, NETDATA_THREAD_OPTION_DONT_LOG, unittest_queue_metadata, &tu);
3110	}
3111	(void) uv_async_send(&meta_config.async);
3112	sleep_usec(seconds_to_run * USEC_PER_SEC);
3113
3114	__atomic_store_n(&tu.join, 1, __ATOMIC_RELAXED);
3115	for (int i = 0; i < threads_to_create; i++) {
3116	nd_thread_join(threads[i]);
3117	}
3118	sleep_usec(5 * USEC_PER_SEC);
3119
3120	fprintf(stderr, "Added %u elements, processed %u\n", tu.added, tu.processed);
3121
3122	return 0;
3123	}
3124
3125	int metadata_unittest(void)
3126	{
3127	metadata_sync_init();
3128
3129	// Queue items for a specific period of time
3130	metadata_unittest_threads();
3131
3132	metadata_sync_shutdown();
3133
3134	return 0;
3135	}