sqlite_metadata.c 68 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "sqlite_metadata.h"
  3. // SQL statements
  4. #define SQL_STORE_CLAIM_ID "INSERT INTO node_instance " \
  5. "(host_id, claim_id, date_created) VALUES (@host_id, @claim_id, unixepoch()) " \
  6. "ON CONFLICT(host_id) DO UPDATE SET claim_id = excluded.claim_id;"
  7. #define SQL_DELETE_HOST_LABELS "DELETE FROM host_label WHERE host_id = @uuid;"
  8. #define STORE_HOST_LABEL \
  9. "INSERT INTO host_label (host_id, source_type, label_key, label_value, date_created) VALUES "
  10. #define STORE_CHART_LABEL \
  11. "INSERT INTO chart_label (chart_id, source_type, label_key, label_value, date_created) VALUES "
  12. #define STORE_HOST_OR_CHART_LABEL_VALUE "(u2h('%s'), %d,'%s','%s', unixepoch())"
  13. #define DELETE_DIMENSION_UUID "DELETE FROM dimension WHERE dim_id = @uuid;"
  14. #define SQL_STORE_HOST_INFO \
  15. "INSERT OR REPLACE INTO host (host_id, hostname, registry_hostname, update_every, os, timezone, tags, hops, " \
  16. "memory_mode, abbrev_timezone, utc_offset, program_name, program_version, entries, health_enabled, last_connected) " \
  17. "VALUES (@host_id, @hostname, @registry_hostname, @update_every, @os, @timezone, @tags, @hops, " \
  18. "@memory_mode, @abbrev_tz, @utc_offset, @prog_name, @prog_version, @entries, @health_enabled, @last_connected);"
  19. #define SQL_STORE_CHART \
  20. "INSERT INTO chart (chart_id, host_id, type, id, name, family, context, title, unit, plugin, module, priority, " \
  21. "update_every, chart_type, memory_mode, history_entries) " \
  22. "values (@chart_id, @host_id, @type, @id, @name, @family, @context, @title, @unit, @plugin, @module, @priority, " \
  23. "@update_every, @chart_type, @memory_mode, @history_entries) " \
  24. "ON CONFLICT(chart_id) DO UPDATE SET type=excluded.type, id=excluded.id, name=excluded.name, " \
  25. "family=excluded.family, context=excluded.context, title=excluded.title, unit=excluded.unit, " \
  26. "plugin=excluded.plugin, module=excluded.module, priority=excluded.priority, update_every=excluded.update_every, " \
  27. "chart_type=excluded.chart_type, memory_mode = excluded.memory_mode, history_entries = excluded.history_entries"
  28. #define SQL_STORE_DIMENSION \
  29. "INSERT INTO dimension (dim_id, chart_id, id, name, multiplier, divisor , algorithm, options) " \
  30. "VALUES (@dim_id, @chart_id, @id, @name, @multiplier, @divisor, @algorithm, @options) " \
  31. "ON CONFLICT(dim_id) DO UPDATE SET id=excluded.id, name=excluded.name, multiplier=excluded.multiplier, " \
  32. "divisor=excluded.divisor, algorithm=excluded.algorithm, options=excluded.options"
  33. #define SELECT_DIMENSION_LIST "SELECT dim_id, rowid FROM dimension WHERE rowid > @row_id"
  34. #define SELECT_CHART_LIST "SELECT chart_id, rowid FROM chart WHERE rowid > @row_id"
  35. #define SELECT_CHART_LABEL_LIST "SELECT chart_id, rowid FROM chart_label WHERE rowid > @row_id"
  36. #define SQL_STORE_HOST_SYSTEM_INFO_VALUES \
  37. "INSERT OR REPLACE INTO host_info (host_id, system_key, system_value, date_created) VALUES " \
  38. "(@uuid, @name, @value, UNIXEPOCH())"
  39. #define MIGRATE_LOCALHOST_TO_NEW_MACHINE_GUID \
  40. "UPDATE chart SET host_id = @host_id WHERE host_id in (SELECT host_id FROM host where host_id <> @host_id and hops = 0);"
  41. #define DELETE_NON_EXISTING_LOCALHOST "DELETE FROM host WHERE hops = 0 AND host_id <> @host_id;"
  42. #define DELETE_MISSING_NODE_INSTANCES "DELETE FROM node_instance WHERE host_id NOT IN (SELECT host_id FROM host);"
  43. #define METADATA_CMD_Q_MAX_SIZE (2048) // Max queue size; callers will block until there is room
  44. #define METADATA_MAINTENANCE_FIRST_CHECK (1800) // Maintenance first run after agent startup in seconds
  45. #define METADATA_MAINTENANCE_REPEAT (60) // Repeat if last run for dimensions, charts, labels needs more work
  46. #define METADATA_HEALTH_LOG_INTERVAL (3600) // Repeat maintenance for health
  47. #define METADATA_DIM_CHECK_INTERVAL (3600) // Repeat maintenance for dimensions
  48. #define METADATA_CHART_CHECK_INTERVAL (3600) // Repeat maintenance for charts
  49. #define METADATA_LABEL_CHECK_INTERVAL (3600) // Repeat maintenance for labels
  50. #define METADATA_RUNTIME_THRESHOLD (5) // Run time threshold for cleanup task
  51. #define METADATA_HOST_CHECK_FIRST_CHECK (5) // First check for pending metadata
  52. #define METADATA_HOST_CHECK_INTERVAL (30) // Repeat check for pending metadata
  53. #define METADATA_HOST_CHECK_IMMEDIATE (5) // Repeat immediate run because we have more metadata to write
  54. #define MAX_METADATA_CLEANUP (500) // Maximum metadata write operations (e.g deletes before retrying)
  55. #define METADATA_MAX_BATCH_SIZE (512) // Maximum commands to execute before running the event loop
  56. #define DATABASE_FREE_PAGES_THRESHOLD_PC (5) // Percentage of free pages to trigger vacuum
  57. #define DATABASE_FREE_PAGES_VACUUM_PC (10) // Percentage of free pages to vacuum
  58. enum metadata_opcode {
  59. METADATA_DATABASE_NOOP = 0,
  60. METADATA_DATABASE_TIMER,
  61. METADATA_DEL_DIMENSION,
  62. METADATA_STORE_CLAIM_ID,
  63. METADATA_ADD_HOST_INFO,
  64. METADATA_SCAN_HOSTS,
  65. METADATA_LOAD_HOST_CONTEXT,
  66. METADATA_MAINTENANCE,
  67. METADATA_SYNC_SHUTDOWN,
  68. METADATA_UNITTEST,
  69. METADATA_ML_LOAD_MODELS,
  70. // leave this last
  71. // we need it to check for worker utilization
  72. METADATA_MAX_ENUMERATIONS_DEFINED
  73. };
  74. #define MAX_PARAM_LIST (2)
  75. struct metadata_cmd {
  76. enum metadata_opcode opcode;
  77. struct completion *completion;
  78. const void *param[MAX_PARAM_LIST];
  79. struct metadata_cmd *prev, *next;
  80. };
  81. struct metadata_database_cmdqueue {
  82. struct metadata_cmd *cmd_base;
  83. };
  84. typedef enum {
  85. METADATA_FLAG_PROCESSING = (1 << 0), // store or cleanup
  86. METADATA_FLAG_SHUTDOWN = (1 << 1), // Shutting down
  87. METADATA_FLAG_ML_LOADING = (1 << 2), // ML model load in progress
  88. } METADATA_FLAG;
  89. struct metadata_wc {
  90. uv_thread_t thread;
  91. uv_loop_t *loop;
  92. uv_async_t async;
  93. uv_timer_t timer_req;
  94. time_t metadata_check_after;
  95. volatile unsigned queue_size;
  96. METADATA_FLAG flags;
  97. struct completion init_complete;
  98. struct completion *scan_complete;
  99. /* FIFO command queue */
  100. uv_mutex_t cmd_mutex;
  101. struct metadata_database_cmdqueue cmd_queue;
  102. };
  103. #define metadata_flag_check(target_flags, flag) (__atomic_load_n(&((target_flags)->flags), __ATOMIC_SEQ_CST) & (flag))
  104. #define metadata_flag_set(target_flags, flag) __atomic_or_fetch(&((target_flags)->flags), (flag), __ATOMIC_SEQ_CST)
  105. #define metadata_flag_clear(target_flags, flag) __atomic_and_fetch(&((target_flags)->flags), ~(flag), __ATOMIC_SEQ_CST)
  106. //
  107. // For unittest
  108. //
  109. struct thread_unittest {
  110. int join;
  111. unsigned added;
  112. unsigned processed;
  113. unsigned *done;
  114. };
  115. // Metadata functions
  116. struct query_build {
  117. BUFFER *sql;
  118. int count;
  119. char uuid_str[UUID_STR_LEN];
  120. };
  121. static int host_label_store_to_sql_callback(const char *name, const char *value, RRDLABEL_SRC ls, void *data) {
  122. struct query_build *lb = data;
  123. if (unlikely(!lb->count))
  124. buffer_sprintf(lb->sql, STORE_HOST_LABEL);
  125. else
  126. buffer_strcat(lb->sql, ", ");
  127. buffer_sprintf(lb->sql, STORE_HOST_OR_CHART_LABEL_VALUE, lb->uuid_str, (int) (ls & ~(RRDLABEL_FLAG_INTERNAL)), name, value);
  128. lb->count++;
  129. return 1;
  130. }
  131. static int chart_label_store_to_sql_callback(const char *name, const char *value, RRDLABEL_SRC ls, void *data) {
  132. struct query_build *lb = data;
  133. if (unlikely(!lb->count))
  134. buffer_sprintf(lb->sql, STORE_CHART_LABEL);
  135. else
  136. buffer_strcat(lb->sql, ", ");
  137. buffer_sprintf(lb->sql, STORE_HOST_OR_CHART_LABEL_VALUE, lb->uuid_str, (int) (ls & ~(RRDLABEL_FLAG_INTERNAL)), name, value);
  138. lb->count++;
  139. return 1;
  140. }
  141. #define SQL_DELETE_CHART_LABEL "DELETE FROM chart_label WHERE chart_id = @chart_id;"
  142. #define SQL_DELETE_CHART_LABEL_HISTORY "DELETE FROM chart_label WHERE date_created < %ld AND chart_id = @chart_id;"
  143. static void clean_old_chart_labels(RRDSET *st)
  144. {
  145. char sql[512];
  146. time_t first_time_s = rrdset_first_entry_s(st);
  147. if (unlikely(!first_time_s))
  148. snprintfz(sql, 511,SQL_DELETE_CHART_LABEL);
  149. else
  150. snprintfz(sql, 511,SQL_DELETE_CHART_LABEL_HISTORY, first_time_s);
  151. int rc = exec_statement_with_uuid(sql, &st->chart_uuid);
  152. if (unlikely(rc))
  153. error_report("METADATA: 'host:%s' Failed to clean old labels for chart %s", rrdhost_hostname(st->rrdhost), rrdset_name(st));
  154. }
  155. static int check_and_update_chart_labels(RRDSET *st, BUFFER *work_buffer, size_t *query_counter)
  156. {
  157. size_t old_version = st->rrdlabels_last_saved_version;
  158. size_t new_version = rrdlabels_version(st->rrdlabels);
  159. if (new_version == old_version)
  160. return 0;
  161. struct query_build tmp = {.sql = work_buffer, .count = 0};
  162. uuid_unparse_lower(st->chart_uuid, tmp.uuid_str);
  163. rrdlabels_walkthrough_read(st->rrdlabels, chart_label_store_to_sql_callback, &tmp);
  164. buffer_strcat(work_buffer, " ON CONFLICT (chart_id, label_key) DO UPDATE SET source_type = excluded.source_type, label_value=excluded.label_value, date_created=UNIXEPOCH()");
  165. int rc = db_execute(db_meta, buffer_tostring(work_buffer));
  166. if (likely(!rc)) {
  167. st->rrdlabels_last_saved_version = new_version;
  168. (*query_counter)++;
  169. }
  170. clean_old_chart_labels(st);
  171. return rc;
  172. }
  173. // Migrate all hosts with hops zero to this host_uuid
  174. void migrate_localhost(uuid_t *host_uuid)
  175. {
  176. int rc;
  177. rc = exec_statement_with_uuid(MIGRATE_LOCALHOST_TO_NEW_MACHINE_GUID, host_uuid);
  178. if (!rc)
  179. rc = exec_statement_with_uuid(DELETE_NON_EXISTING_LOCALHOST, host_uuid);
  180. if (!rc) {
  181. if (unlikely(db_execute(db_meta, DELETE_MISSING_NODE_INSTANCES)))
  182. error_report("Failed to remove deleted hosts from node instances");
  183. }
  184. }
  185. static int store_claim_id(uuid_t *host_id, uuid_t *claim_id)
  186. {
  187. sqlite3_stmt *res = NULL;
  188. int rc;
  189. if (unlikely(!db_meta)) {
  190. if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
  191. error_report("Database has not been initialized");
  192. return 1;
  193. }
  194. rc = sqlite3_prepare_v2(db_meta, SQL_STORE_CLAIM_ID, -1, &res, 0);
  195. if (unlikely(rc != SQLITE_OK)) {
  196. error_report("Failed to prepare statement to store host claim id");
  197. return 1;
  198. }
  199. rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
  200. if (unlikely(rc != SQLITE_OK)) {
  201. error_report("Failed to bind host_id parameter to store claim id");
  202. goto failed;
  203. }
  204. if (claim_id)
  205. rc = sqlite3_bind_blob(res, 2, claim_id, sizeof(*claim_id), SQLITE_STATIC);
  206. else
  207. rc = sqlite3_bind_null(res, 2);
  208. if (unlikely(rc != SQLITE_OK)) {
  209. error_report("Failed to bind claim_id parameter to host claim id");
  210. goto failed;
  211. }
  212. rc = execute_insert(res);
  213. if (unlikely(rc != SQLITE_DONE))
  214. error_report("Failed to store host claim id rc = %d", rc);
  215. failed:
  216. if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
  217. error_report("Failed to finalize the prepared statement when storing a host claim id");
  218. return rc != SQLITE_DONE;
  219. }
  220. static void delete_dimension_uuid(uuid_t *dimension_uuid, sqlite3_stmt **action_res __maybe_unused, bool flag __maybe_unused)
  221. {
  222. static __thread sqlite3_stmt *res = NULL;
  223. int rc;
  224. if (unlikely(!res)) {
  225. rc = prepare_statement(db_meta, DELETE_DIMENSION_UUID, &res);
  226. if (rc != SQLITE_OK) {
  227. error_report("Failed to prepare statement to delete a dimension uuid");
  228. return;
  229. }
  230. }
  231. rc = sqlite3_bind_blob(res, 1, dimension_uuid, sizeof(*dimension_uuid), SQLITE_STATIC);
  232. if (unlikely(rc != SQLITE_OK))
  233. goto skip_execution;
  234. rc = sqlite3_step_monitored(res);
  235. if (unlikely(rc != SQLITE_DONE))
  236. error_report("Failed to delete dimension uuid, rc = %d", rc);
  237. skip_execution:
  238. rc = sqlite3_reset(res);
  239. if (unlikely(rc != SQLITE_OK))
  240. error_report("Failed to reset statement when deleting dimension UUID, rc = %d", rc);
  241. }
  242. //
  243. // Store host and host system info information in the database
  244. static int store_host_metadata(RRDHOST *host)
  245. {
  246. static __thread sqlite3_stmt *res = NULL;
  247. int rc, param = 0;
  248. if (unlikely((!res))) {
  249. rc = prepare_statement(db_meta, SQL_STORE_HOST_INFO, &res);
  250. if (unlikely(rc != SQLITE_OK)) {
  251. error_report("Failed to prepare statement to store host, rc = %d", rc);
  252. return 1;
  253. }
  254. }
  255. rc = sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);
  256. if (unlikely(rc != SQLITE_OK))
  257. goto bind_fail;
  258. rc = bind_text_null(res, ++param, rrdhost_hostname(host), 0);
  259. if (unlikely(rc != SQLITE_OK))
  260. goto bind_fail;
  261. rc = bind_text_null(res, ++param, rrdhost_registry_hostname(host), 1);
  262. if (unlikely(rc != SQLITE_OK))
  263. goto bind_fail;
  264. rc = sqlite3_bind_int(res, ++param, host->rrd_update_every);
  265. if (unlikely(rc != SQLITE_OK))
  266. goto bind_fail;
  267. rc = bind_text_null(res, ++param, rrdhost_os(host), 1);
  268. if (unlikely(rc != SQLITE_OK))
  269. goto bind_fail;
  270. rc = bind_text_null(res, ++param, rrdhost_timezone(host), 1);
  271. if (unlikely(rc != SQLITE_OK))
  272. goto bind_fail;
  273. rc = bind_text_null(res, ++param, rrdhost_tags(host), 1);
  274. if (unlikely(rc != SQLITE_OK))
  275. goto bind_fail;
  276. rc = sqlite3_bind_int(res, ++param, host->system_info ? host->system_info->hops : 0);
  277. if (unlikely(rc != SQLITE_OK))
  278. goto bind_fail;
  279. rc = sqlite3_bind_int(res, ++param, host->rrd_memory_mode);
  280. if (unlikely(rc != SQLITE_OK))
  281. goto bind_fail;
  282. rc = bind_text_null(res, ++param, rrdhost_abbrev_timezone(host), 1);
  283. if (unlikely(rc != SQLITE_OK))
  284. goto bind_fail;
  285. rc = sqlite3_bind_int(res, ++param, host->utc_offset);
  286. if (unlikely(rc != SQLITE_OK))
  287. goto bind_fail;
  288. rc = bind_text_null(res, ++param, rrdhost_program_name(host), 1);
  289. if (unlikely(rc != SQLITE_OK))
  290. goto bind_fail;
  291. rc = bind_text_null(res, ++param, rrdhost_program_version(host), 1);
  292. if (unlikely(rc != SQLITE_OK))
  293. goto bind_fail;
  294. rc = sqlite3_bind_int64(res, ++param, host->rrd_history_entries);
  295. if (unlikely(rc != SQLITE_OK))
  296. goto bind_fail;
  297. rc = sqlite3_bind_int(res, ++param, (int ) host->health.health_enabled);
  298. if (unlikely(rc != SQLITE_OK))
  299. goto bind_fail;
  300. rc = sqlite3_bind_int64(res, ++param, (sqlite3_int64) host->last_connected);
  301. if (unlikely(rc != SQLITE_OK))
  302. goto bind_fail;
  303. int store_rc = sqlite3_step_monitored(res);
  304. if (unlikely(store_rc != SQLITE_DONE))
  305. error_report("Failed to store host %s, rc = %d", rrdhost_hostname(host), rc);
  306. rc = sqlite3_reset(res);
  307. if (unlikely(rc != SQLITE_OK))
  308. error_report("Failed to reset statement to store host %s, rc = %d", rrdhost_hostname(host), rc);
  309. return store_rc != SQLITE_DONE;
  310. bind_fail:
  311. error_report("Failed to bind %d parameter to store host %s, rc = %d", param, rrdhost_hostname(host), rc);
  312. rc = sqlite3_reset(res);
  313. if (unlikely(rc != SQLITE_OK))
  314. error_report("Failed to reset statement to store host %s, rc = %d", rrdhost_hostname(host), rc);
  315. return 1;
  316. }
  317. static int add_host_sysinfo_key_value(const char *name, const char *value, uuid_t *uuid)
  318. {
  319. static __thread sqlite3_stmt *res = NULL;
  320. int rc, param = 0;
  321. if (unlikely(!db_meta)) {
  322. if (default_rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE)
  323. return 0;
  324. error_report("Database has not been initialized");
  325. return 0;
  326. }
  327. if (unlikely((!res))) {
  328. rc = prepare_statement(db_meta, SQL_STORE_HOST_SYSTEM_INFO_VALUES, &res);
  329. if (unlikely(rc != SQLITE_OK)) {
  330. error_report("Failed to prepare statement to store host info values, rc = %d", rc);
  331. return 0;
  332. }
  333. }
  334. rc = sqlite3_bind_blob(res, ++param, uuid, sizeof(*uuid), SQLITE_STATIC);
  335. if (unlikely(rc != SQLITE_OK))
  336. goto bind_fail;
  337. rc = bind_text_null(res, ++param, name, 0);
  338. if (unlikely(rc != SQLITE_OK))
  339. goto bind_fail;
  340. rc = bind_text_null(res, ++param, value ? value : "unknown", 0);
  341. if (unlikely(rc != SQLITE_OK))
  342. goto bind_fail;
  343. int store_rc = sqlite3_step_monitored(res);
  344. if (unlikely(store_rc != SQLITE_DONE))
  345. error_report("Failed to store host info value %s, rc = %d", name, rc);
  346. rc = sqlite3_reset(res);
  347. if (unlikely(rc != SQLITE_OK))
  348. error_report("Failed to reset statement to store host info value %s, rc = %d", name, rc);
  349. return store_rc == SQLITE_DONE;
  350. bind_fail:
  351. error_report("Failed to bind %d parameter to store host info values %s, rc = %d", param, name, rc);
  352. rc = sqlite3_reset(res);
  353. if (unlikely(rc != SQLITE_OK))
  354. error_report("Failed to reset statement to store host info values %s, rc = %d", name, rc);
  355. return 0;
  356. }
  357. static bool store_host_systeminfo(RRDHOST *host)
  358. {
  359. struct rrdhost_system_info *system_info = host->system_info;
  360. if (unlikely(!system_info))
  361. return false;
  362. int ret = 0;
  363. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_NAME", system_info->container_os_name, &host->host_uuid);
  364. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_ID", system_info->container_os_id, &host->host_uuid);
  365. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_ID_LIKE", system_info->container_os_id_like, &host->host_uuid);
  366. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_VERSION", system_info->container_os_version, &host->host_uuid);
  367. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_VERSION_ID", system_info->container_os_version_id, &host->host_uuid);
  368. ret += add_host_sysinfo_key_value("NETDATA_CONTAINER_OS_DETECTION", system_info->host_os_detection, &host->host_uuid);
  369. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_NAME", system_info->host_os_name, &host->host_uuid);
  370. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_ID", system_info->host_os_id, &host->host_uuid);
  371. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_ID_LIKE", system_info->host_os_id_like, &host->host_uuid);
  372. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_VERSION", system_info->host_os_version, &host->host_uuid);
  373. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_VERSION_ID", system_info->host_os_version_id, &host->host_uuid);
  374. ret += add_host_sysinfo_key_value("NETDATA_HOST_OS_DETECTION", system_info->host_os_detection, &host->host_uuid);
  375. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_KERNEL_NAME", system_info->kernel_name, &host->host_uuid);
  376. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_CPU_LOGICAL_CPU_COUNT", system_info->host_cores, &host->host_uuid);
  377. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_CPU_FREQ", system_info->host_cpu_freq, &host->host_uuid);
  378. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_TOTAL_RAM", system_info->host_ram_total, &host->host_uuid);
  379. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_TOTAL_DISK_SIZE", system_info->host_disk_space, &host->host_uuid);
  380. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_KERNEL_VERSION", system_info->kernel_version, &host->host_uuid);
  381. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_ARCHITECTURE", system_info->architecture, &host->host_uuid);
  382. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_VIRTUALIZATION", system_info->virtualization, &host->host_uuid);
  383. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_VIRT_DETECTION", system_info->virt_detection, &host->host_uuid);
  384. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_CONTAINER", system_info->container, &host->host_uuid);
  385. ret += add_host_sysinfo_key_value("NETDATA_SYSTEM_CONTAINER_DETECTION", system_info->container_detection, &host->host_uuid);
  386. ret += add_host_sysinfo_key_value("NETDATA_HOST_IS_K8S_NODE", system_info->is_k8s_node, &host->host_uuid);
  387. return !(24 == ret);
  388. }
  389. /*
  390. * Store a chart in the database
  391. */
  392. static int store_chart_metadata(RRDSET *st)
  393. {
  394. static __thread sqlite3_stmt *res = NULL;
  395. int rc, param = 0, store_rc = 0;
  396. if (unlikely(!res)) {
  397. rc = prepare_statement(db_meta, SQL_STORE_CHART, &res);
  398. if (unlikely(rc != SQLITE_OK)) {
  399. error_report("Failed to prepare statement to store chart, rc = %d", rc);
  400. return 1;
  401. }
  402. }
  403. rc = sqlite3_bind_blob(res, ++param, &st->chart_uuid, sizeof(st->chart_uuid), SQLITE_STATIC);
  404. if (unlikely(rc != SQLITE_OK))
  405. goto bind_fail;
  406. rc = sqlite3_bind_blob(res, ++param, &st->rrdhost->host_uuid, sizeof(st->rrdhost->host_uuid), SQLITE_STATIC);
  407. if (unlikely(rc != SQLITE_OK))
  408. goto bind_fail;
  409. rc = sqlite3_bind_text(res, ++param, string2str(st->parts.type), -1, SQLITE_STATIC);
  410. if (unlikely(rc != SQLITE_OK))
  411. goto bind_fail;
  412. rc = sqlite3_bind_text(res, ++param, string2str(st->parts.id), -1, SQLITE_STATIC);
  413. if (unlikely(rc != SQLITE_OK))
  414. goto bind_fail;
  415. const char *name = string2str(st->parts.name);
  416. if (name && *name)
  417. rc = sqlite3_bind_text(res, ++param, name, -1, SQLITE_STATIC);
  418. else
  419. rc = sqlite3_bind_null(res, ++param);
  420. if (unlikely(rc != SQLITE_OK))
  421. goto bind_fail;
  422. rc = sqlite3_bind_text(res, ++param, rrdset_family(st), -1, SQLITE_STATIC);
  423. if (unlikely(rc != SQLITE_OK))
  424. goto bind_fail;
  425. rc = sqlite3_bind_text(res, ++param, rrdset_context(st), -1, SQLITE_STATIC);
  426. if (unlikely(rc != SQLITE_OK))
  427. goto bind_fail;
  428. rc = sqlite3_bind_text(res, ++param, rrdset_title(st), -1, SQLITE_STATIC);
  429. if (unlikely(rc != SQLITE_OK))
  430. goto bind_fail;
  431. rc = sqlite3_bind_text(res, ++param, rrdset_units(st), -1, SQLITE_STATIC);
  432. if (unlikely(rc != SQLITE_OK))
  433. goto bind_fail;
  434. rc = sqlite3_bind_text(res, ++param, rrdset_plugin_name(st), -1, SQLITE_STATIC);
  435. if (unlikely(rc != SQLITE_OK))
  436. goto bind_fail;
  437. rc = sqlite3_bind_text(res, ++param, rrdset_module_name(st), -1, SQLITE_STATIC);
  438. if (unlikely(rc != SQLITE_OK))
  439. goto bind_fail;
  440. rc = sqlite3_bind_int(res, ++param, (int) st->priority);
  441. if (unlikely(rc != SQLITE_OK))
  442. goto bind_fail;
  443. rc = sqlite3_bind_int(res, ++param, st->update_every);
  444. if (unlikely(rc != SQLITE_OK))
  445. goto bind_fail;
  446. rc = sqlite3_bind_int(res, ++param, st->chart_type);
  447. if (unlikely(rc != SQLITE_OK))
  448. goto bind_fail;
  449. rc = sqlite3_bind_int(res, ++param, st->rrd_memory_mode);
  450. if (unlikely(rc != SQLITE_OK))
  451. goto bind_fail;
  452. rc = sqlite3_bind_int(res, ++param, (int) st->db.entries);
  453. if (unlikely(rc != SQLITE_OK))
  454. goto bind_fail;
  455. store_rc = execute_insert(res);
  456. if (unlikely(store_rc != SQLITE_DONE))
  457. error_report("Failed to store chart, rc = %d", store_rc);
  458. rc = sqlite3_reset(res);
  459. if (unlikely(rc != SQLITE_OK))
  460. error_report("Failed to reset statement in chart store function, rc = %d", rc);
  461. return store_rc != SQLITE_DONE;
  462. bind_fail:
  463. error_report("Failed to bind parameter %d to store chart, rc = %d", param, rc);
  464. rc = sqlite3_reset(res);
  465. if (unlikely(rc != SQLITE_OK))
  466. error_report("Failed to reset statement in chart store function, rc = %d", rc);
  467. return 1;
  468. }
  469. /*
  470. * Store a dimension
  471. */
  472. static int store_dimension_metadata(RRDDIM *rd)
  473. {
  474. static __thread sqlite3_stmt *res = NULL;
  475. int rc, param = 0;
  476. if (unlikely(!res)) {
  477. rc = prepare_statement(db_meta, SQL_STORE_DIMENSION, &res);
  478. if (unlikely(rc != SQLITE_OK)) {
  479. error_report("Failed to prepare statement to store dimension, rc = %d", rc);
  480. return 1;
  481. }
  482. }
  483. rc = sqlite3_bind_blob(res, ++param, &rd->metric_uuid, sizeof(rd->metric_uuid), SQLITE_STATIC);
  484. if (unlikely(rc != SQLITE_OK))
  485. goto bind_fail;
  486. rc = sqlite3_bind_blob(res, ++param, &rd->rrdset->chart_uuid, sizeof(rd->rrdset->chart_uuid), SQLITE_STATIC);
  487. if (unlikely(rc != SQLITE_OK))
  488. goto bind_fail;
  489. rc = sqlite3_bind_text(res, ++param, string2str(rd->id), -1, SQLITE_STATIC);
  490. if (unlikely(rc != SQLITE_OK))
  491. goto bind_fail;
  492. rc = sqlite3_bind_text(res, ++param, string2str(rd->name), -1, SQLITE_STATIC);
  493. if (unlikely(rc != SQLITE_OK))
  494. goto bind_fail;
  495. rc = sqlite3_bind_int(res, ++param, (int) rd->multiplier);
  496. if (unlikely(rc != SQLITE_OK))
  497. goto bind_fail;
  498. rc = sqlite3_bind_int(res, ++param, (int ) rd->divisor);
  499. if (unlikely(rc != SQLITE_OK))
  500. goto bind_fail;
  501. rc = sqlite3_bind_int(res, ++param, rd->algorithm);
  502. if (unlikely(rc != SQLITE_OK))
  503. goto bind_fail;
  504. if (rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN))
  505. rc = sqlite3_bind_text(res, ++param, "hidden", -1, SQLITE_STATIC);
  506. else
  507. rc = sqlite3_bind_null(res, ++param);
  508. if (unlikely(rc != SQLITE_OK))
  509. goto bind_fail;
  510. rc = execute_insert(res);
  511. if (unlikely(rc != SQLITE_DONE))
  512. error_report("Failed to store dimension, rc = %d", rc);
  513. rc = sqlite3_reset(res);
  514. if (unlikely(rc != SQLITE_OK))
  515. error_report("Failed to reset statement in store dimension, rc = %d", rc);
  516. return 0;
  517. bind_fail:
  518. error_report("Failed to bind parameter %d to store dimension, rc = %d", param, rc);
  519. rc = sqlite3_reset(res);
  520. if (unlikely(rc != SQLITE_OK))
  521. error_report("Failed to reset statement in store dimension, rc = %d", rc);
  522. return 1;
  523. }
  524. static bool dimension_can_be_deleted(uuid_t *dim_uuid __maybe_unused, sqlite3_stmt **res __maybe_unused, bool flag __maybe_unused)
  525. {
  526. #ifdef ENABLE_DBENGINE
  527. if(dbengine_enabled) {
  528. bool no_retention = true;
  529. for (size_t tier = 0; tier < storage_tiers; tier++) {
  530. if (!multidb_ctx[tier])
  531. continue;
  532. time_t first_time_t = 0, last_time_t = 0;
  533. if (rrdeng_metric_retention_by_uuid((void *) multidb_ctx[tier], dim_uuid, &first_time_t, &last_time_t)) {
  534. if (first_time_t > 0) {
  535. no_retention = false;
  536. break;
  537. }
  538. }
  539. }
  540. return no_retention;
  541. }
  542. else
  543. return false;
  544. #else
  545. return false;
  546. #endif
  547. }
  548. int get_pragma_value(sqlite3 *database, const char *sql)
  549. {
  550. sqlite3_stmt *res = NULL;
  551. int rc = sqlite3_prepare_v2(database, sql, -1, &res, 0);
  552. if (unlikely(rc != SQLITE_OK))
  553. return -1;
  554. int result = -1;
  555. rc = sqlite3_step_monitored(res);
  556. if (likely(rc == SQLITE_ROW))
  557. result = sqlite3_column_int(res, 0);
  558. rc = sqlite3_finalize(res);
  559. (void) rc;
  560. return result;
  561. }
  562. int get_free_page_count(sqlite3 *database)
  563. {
  564. return get_pragma_value(database, "PRAGMA freelist_count");
  565. }
  566. int get_database_page_count(sqlite3 *database)
  567. {
  568. return get_pragma_value(database, "PRAGMA page_count");
  569. }
  570. static bool run_cleanup_loop(
  571. sqlite3_stmt *res,
  572. struct metadata_wc *wc,
  573. bool (*check_cb)(uuid_t *, sqlite3_stmt **, bool),
  574. void (*action_cb)(uuid_t *, sqlite3_stmt **, bool),
  575. uint32_t *total_checked,
  576. uint32_t *total_deleted,
  577. uint64_t *row_id,
  578. sqlite3_stmt **check_stmt,
  579. sqlite3_stmt **action_stmt,
  580. bool check_flag,
  581. bool action_flag)
  582. {
  583. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  584. return true;
  585. int rc = sqlite3_bind_int64(res, 1, (sqlite3_int64) *row_id);
  586. if (unlikely(rc != SQLITE_OK))
  587. return true;
  588. time_t start_running = now_monotonic_sec();
  589. bool time_expired = false;
  590. while (!time_expired && sqlite3_step_monitored(res) == SQLITE_ROW &&
  591. (*total_deleted < MAX_METADATA_CLEANUP && *total_checked < MAX_METADATA_CLEANUP)) {
  592. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  593. break;
  594. *row_id = sqlite3_column_int64(res, 1);
  595. rc = check_cb((uuid_t *)sqlite3_column_blob(res, 0), check_stmt, check_flag);
  596. if (rc == true) {
  597. action_cb((uuid_t *)sqlite3_column_blob(res, 0), action_stmt, action_flag);
  598. (*total_deleted)++;
  599. }
  600. (*total_checked)++;
  601. time_expired = ((now_monotonic_sec() - start_running) > METADATA_RUNTIME_THRESHOLD);
  602. }
  603. return time_expired || (*total_checked == MAX_METADATA_CLEANUP) || (*total_deleted == MAX_METADATA_CLEANUP);
  604. }
  605. #define SQL_CHECK_CHART_EXISTENCE_IN_DIMENSION "SELECT count(1) FROM dimension WHERE chart_id = @chart_id"
  606. #define SQL_CHECK_CHART_EXISTENCE_IN_CHART "SELECT count(1) FROM chart WHERE chart_id = @chart_id"
  607. static bool chart_can_be_deleted(uuid_t *chart_uuid, sqlite3_stmt **check_res, bool check_in_dimension)
  608. {
  609. int rc, result = 1;
  610. sqlite3_stmt *res = check_res ? *check_res : NULL;
  611. if (!res) {
  612. if (check_in_dimension)
  613. rc = sqlite3_prepare_v2(db_meta, SQL_CHECK_CHART_EXISTENCE_IN_DIMENSION, -1, &res, 0);
  614. else
  615. rc = sqlite3_prepare_v2(db_meta, SQL_CHECK_CHART_EXISTENCE_IN_CHART, -1, &res, 0);
  616. if (unlikely(rc != SQLITE_OK)) {
  617. error_report("Failed to prepare statement to check for chart existence, rc = %d", rc);
  618. return 0;
  619. }
  620. if (check_res)
  621. *check_res = res;
  622. }
  623. rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);
  624. if (unlikely(rc != SQLITE_OK)) {
  625. error_report("Failed to bind chart uuid parameter, rc = %d", rc);
  626. goto skip;
  627. }
  628. rc = sqlite3_step_monitored(res);
  629. if (likely(rc == SQLITE_ROW))
  630. result = sqlite3_column_int(res, 0);
  631. skip:
  632. if (check_res)
  633. rc = sqlite3_reset(res);
  634. else
  635. rc = sqlite3_finalize(res);
  636. if (unlikely(rc != SQLITE_OK))
  637. error_report("Failed to %s statement that checks chart uuid existence rc = %d", check_res ? "reset" : "finalize", rc);
  638. return result == 0;
  639. }
  640. #define SQL_DELETE_CHART_BY_UUID "DELETE FROM chart WHERE chart_id = @chart_id"
  641. #define SQL_DELETE_CHART_LABEL_BY_UUID "DELETE FROM chart_label WHERE chart_id = @chart_id"
  642. static void delete_chart_uuid(uuid_t *chart_uuid, sqlite3_stmt **action_res, bool label_only)
  643. {
  644. int rc;
  645. sqlite3_stmt *res = action_res ? *action_res : NULL;
  646. if (!res) {
  647. if (label_only)
  648. rc = sqlite3_prepare_v2(db_meta, SQL_DELETE_CHART_LABEL_BY_UUID, -1, &res, 0);
  649. else
  650. rc = sqlite3_prepare_v2(db_meta, SQL_DELETE_CHART_BY_UUID, -1, &res, 0);
  651. if (unlikely(rc != SQLITE_OK)) {
  652. error_report("Failed to prepare statement to check for chart existence, rc = %d", rc);
  653. return;
  654. }
  655. if (action_res)
  656. *action_res = res;
  657. }
  658. rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);
  659. if (unlikely(rc != SQLITE_OK)) {
  660. error_report("Failed to bind chart uuid parameter, rc = %d", rc);
  661. goto skip;
  662. }
  663. rc = sqlite3_step_monitored(res);
  664. if (unlikely(rc != SQLITE_DONE))
  665. error_report("Failed to delete a chart uuid from the %s table, rc = %d", label_only ? "labels" : "chart", rc);
  666. skip:
  667. if (action_res)
  668. rc = sqlite3_reset(res);
  669. else
  670. rc = sqlite3_finalize(res);
  671. if (unlikely(rc != SQLITE_OK))
  672. error_report("Failed to %s statement that deletes a chart uuid rc = %d", action_res ? "reset" : "finalize", rc);
  673. }
  674. static void check_dimension_metadata(struct metadata_wc *wc)
  675. {
  676. static time_t next_execution_t = 0;
  677. static uint64_t last_row_id = 0;
  678. time_t now = now_realtime_sec();
  679. if (!next_execution_t)
  680. next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
  681. if (next_execution_t && next_execution_t > now)
  682. return;
  683. int rc;
  684. sqlite3_stmt *res = NULL;
  685. rc = sqlite3_prepare_v2(db_meta, SELECT_DIMENSION_LIST, -1, &res, 0);
  686. if (unlikely(rc != SQLITE_OK)) {
  687. error_report("Failed to prepare statement to fetch host dimensions");
  688. return;
  689. }
  690. uint32_t total_checked = 0;
  691. uint32_t total_deleted = 0;
  692. internal_error(true, "METADATA: Checking dimensions starting after row %"PRIu64, last_row_id);
  693. bool more_to_do = run_cleanup_loop(
  694. res,
  695. wc,
  696. dimension_can_be_deleted,
  697. delete_dimension_uuid,
  698. &total_checked,
  699. &total_deleted,
  700. &last_row_id,
  701. NULL,
  702. NULL,
  703. false,
  704. false);
  705. now = now_realtime_sec();
  706. if (more_to_do)
  707. next_execution_t = now + METADATA_MAINTENANCE_REPEAT;
  708. else {
  709. last_row_id = 0;
  710. next_execution_t = now + METADATA_DIM_CHECK_INTERVAL;
  711. }
  712. netdata_log_info(
  713. "METADATA: Dimensions checked %u, deleted %u. Checks will %s in %lld seconds",
  714. total_checked,
  715. total_deleted,
  716. last_row_id ? "resume" : "restart",
  717. (long long)(next_execution_t - now));
  718. rc = sqlite3_finalize(res);
  719. if (unlikely(rc != SQLITE_OK))
  720. error_report("Failed to finalize the prepared statement to check dimensions");
  721. }
  722. static void check_chart_metadata(struct metadata_wc *wc)
  723. {
  724. static time_t next_execution_t = 0;
  725. static uint64_t last_row_id = 0;
  726. time_t now = now_realtime_sec();
  727. if (!next_execution_t)
  728. next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
  729. if (next_execution_t && next_execution_t > now)
  730. return;
  731. sqlite3_stmt *res = NULL;
  732. int rc = sqlite3_prepare_v2(db_meta, SELECT_CHART_LIST, -1, &res, 0);
  733. if (unlikely(rc != SQLITE_OK)) {
  734. error_report("Failed to prepare statement to fetch charts");
  735. return;
  736. }
  737. uint32_t total_checked = 0;
  738. uint32_t total_deleted = 0;
  739. internal_error(true, "METADATA: Checking charts starting after row %"PRIu64, last_row_id);
  740. sqlite3_stmt *check_res = NULL;
  741. sqlite3_stmt *action_res = NULL;
  742. bool more_to_do = run_cleanup_loop(
  743. res,
  744. wc,
  745. chart_can_be_deleted,
  746. delete_chart_uuid,
  747. &total_checked,
  748. &total_deleted,
  749. &last_row_id,
  750. &check_res,
  751. &action_res,
  752. true,
  753. false);
  754. if (check_res)
  755. sqlite3_finalize(check_res);
  756. if (action_res)
  757. sqlite3_finalize(action_res);
  758. now = now_realtime_sec();
  759. if (more_to_do)
  760. next_execution_t = now + METADATA_MAINTENANCE_REPEAT;
  761. else {
  762. last_row_id = 0;
  763. next_execution_t = now + METADATA_CHART_CHECK_INTERVAL;
  764. }
  765. netdata_log_info(
  766. "METADATA: Charts checked %u, deleted %u. Checks will %s in %lld seconds",
  767. total_checked,
  768. total_deleted,
  769. last_row_id ? "resume" : "restart",
  770. (long long)(next_execution_t - now));
  771. rc = sqlite3_finalize(res);
  772. if (unlikely(rc != SQLITE_OK))
  773. error_report("Failed to finalize the prepared statement when reading charts");
  774. }
  775. static void check_label_metadata(struct metadata_wc *wc)
  776. {
  777. static time_t next_execution_t = 0;
  778. static uint64_t last_row_id = 0;
  779. time_t now = now_realtime_sec();
  780. if (!next_execution_t)
  781. next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
  782. if (next_execution_t && next_execution_t > now)
  783. return;
  784. int rc;
  785. sqlite3_stmt *res = NULL;
  786. rc = sqlite3_prepare_v2(db_meta, SELECT_CHART_LABEL_LIST, -1, &res, 0);
  787. if (unlikely(rc != SQLITE_OK)) {
  788. error_report("Failed to prepare statement to fetch charts");
  789. return;
  790. }
  791. uint32_t total_checked = 0;
  792. uint32_t total_deleted = 0;
  793. internal_error(true,"METADATA: Checking charts labels starting after row %"PRIu64, last_row_id);
  794. sqlite3_stmt *check_res = NULL;
  795. sqlite3_stmt *action_res = NULL;
  796. bool more_to_do = run_cleanup_loop(
  797. res,
  798. wc,
  799. chart_can_be_deleted,
  800. delete_chart_uuid,
  801. &total_checked,
  802. &total_deleted,
  803. &last_row_id,
  804. &check_res,
  805. &action_res,
  806. false,
  807. true);
  808. if (check_res)
  809. sqlite3_finalize(check_res);
  810. if (action_res)
  811. sqlite3_finalize(action_res);
  812. now = now_realtime_sec();
  813. if (more_to_do)
  814. next_execution_t = now + METADATA_MAINTENANCE_REPEAT;
  815. else {
  816. last_row_id = 0;
  817. next_execution_t = now + METADATA_LABEL_CHECK_INTERVAL;
  818. }
  819. netdata_log_info(
  820. "METADATA: Chart labels checked %u, deleted %u. Checks will %s in %lld seconds",
  821. total_checked,
  822. total_deleted,
  823. last_row_id ? "resume" : "restart",
  824. (long long)(next_execution_t - now));
  825. rc = sqlite3_finalize(res);
  826. if (unlikely(rc != SQLITE_OK))
  827. error_report("Failed to finalize the prepared statement when checking charts");
  828. }
  829. static void cleanup_health_log(struct metadata_wc *wc)
  830. {
  831. static time_t next_execution_t = 0;
  832. time_t now = now_realtime_sec();
  833. if (!next_execution_t)
  834. next_execution_t = now + METADATA_MAINTENANCE_FIRST_CHECK;
  835. if (next_execution_t && next_execution_t > now)
  836. return;
  837. next_execution_t = now + METADATA_HEALTH_LOG_INTERVAL;
  838. RRDHOST *host;
  839. bool is_claimed = claimed();
  840. dfe_start_reentrant(rrdhost_root_index, host){
  841. if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED))
  842. continue;
  843. sql_health_alarm_log_cleanup(host, is_claimed);
  844. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  845. break;
  846. }
  847. dfe_done(host);
  848. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  849. return;
  850. (void) db_execute(db_meta,"DELETE FROM health_log WHERE host_id NOT IN (SELECT host_id FROM host)");
  851. (void) db_execute(db_meta,"DELETE FROM health_log_detail WHERE health_log_id NOT IN (SELECT health_log_id FROM health_log)");
  852. }
  853. //
  854. // EVENT LOOP STARTS HERE
  855. //
  856. static void metadata_init_cmd_queue(struct metadata_wc *wc)
  857. {
  858. wc->cmd_queue.cmd_base = NULL;
  859. fatal_assert(0 == uv_mutex_init(&wc->cmd_mutex));
  860. }
  861. static void metadata_free_cmd_queue(struct metadata_wc *wc)
  862. {
  863. uv_mutex_lock(&wc->cmd_mutex);
  864. while(wc->cmd_queue.cmd_base) {
  865. struct metadata_cmd *t = wc->cmd_queue.cmd_base;
  866. DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);
  867. freez(t);
  868. }
  869. uv_mutex_unlock(&wc->cmd_mutex);
  870. }
  871. static void metadata_enq_cmd(struct metadata_wc *wc, struct metadata_cmd *cmd)
  872. {
  873. if (cmd->opcode == METADATA_SYNC_SHUTDOWN) {
  874. metadata_flag_set(wc, METADATA_FLAG_SHUTDOWN);
  875. goto wakeup_event_loop;
  876. }
  877. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  878. goto wakeup_event_loop;
  879. struct metadata_cmd *t = mallocz(sizeof(*t));
  880. *t = *cmd;
  881. t->prev = t->next = NULL;
  882. uv_mutex_lock(&wc->cmd_mutex);
  883. DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);
  884. uv_mutex_unlock(&wc->cmd_mutex);
  885. wakeup_event_loop:
  886. (void) uv_async_send(&wc->async);
  887. }
  888. static struct metadata_cmd metadata_deq_cmd(struct metadata_wc *wc)
  889. {
  890. struct metadata_cmd ret;
  891. uv_mutex_lock(&wc->cmd_mutex);
  892. if(wc->cmd_queue.cmd_base) {
  893. struct metadata_cmd *t = wc->cmd_queue.cmd_base;
  894. DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wc->cmd_queue.cmd_base, t, prev, next);
  895. ret = *t;
  896. freez(t);
  897. }
  898. else {
  899. ret.opcode = METADATA_DATABASE_NOOP;
  900. ret.completion = NULL;
  901. }
  902. uv_mutex_unlock(&wc->cmd_mutex);
  903. return ret;
  904. }
  905. static void async_cb(uv_async_t *handle)
  906. {
  907. uv_stop(handle->loop);
  908. uv_update_time(handle->loop);
  909. }
  910. #define TIMER_INITIAL_PERIOD_MS (1000)
  911. #define TIMER_REPEAT_PERIOD_MS (1000)
  912. static void timer_cb(uv_timer_t* handle)
  913. {
  914. uv_stop(handle->loop);
  915. uv_update_time(handle->loop);
  916. struct metadata_wc *wc = handle->data;
  917. struct metadata_cmd cmd;
  918. memset(&cmd, 0, sizeof(cmd));
  919. time_t now = now_realtime_sec();
  920. if (wc->metadata_check_after && wc->metadata_check_after < now) {
  921. cmd.opcode = METADATA_SCAN_HOSTS;
  922. metadata_enq_cmd(wc, &cmd);
  923. }
  924. }
  925. void vacuum_database(sqlite3 *database, const char *db_alias, int threshold, int vacuum_pc)
  926. {
  927. int free_pages = get_free_page_count(database);
  928. int total_pages = get_database_page_count(database);
  929. if (!threshold)
  930. threshold = DATABASE_FREE_PAGES_THRESHOLD_PC;
  931. if (!vacuum_pc)
  932. vacuum_pc = DATABASE_FREE_PAGES_VACUUM_PC;
  933. if (free_pages > (total_pages * threshold / 100)) {
  934. int do_free_pages = (int) (free_pages * vacuum_pc / 100);
  935. netdata_log_info("%s: Freeing %d database pages", db_alias, do_free_pages);
  936. char sql[128];
  937. snprintfz(sql, 127, "PRAGMA incremental_vacuum(%d)", do_free_pages);
  938. (void) db_execute(database, sql);
  939. }
  940. }
  941. void run_metadata_cleanup(struct metadata_wc *wc)
  942. {
  943. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  944. return;
  945. check_dimension_metadata(wc);
  946. check_chart_metadata(wc);
  947. check_label_metadata(wc);
  948. cleanup_health_log(wc);
  949. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN)))
  950. return;
  951. vacuum_database(db_meta, "METADATA", DATABASE_FREE_PAGES_THRESHOLD_PC, DATABASE_FREE_PAGES_VACUUM_PC);
  952. (void) sqlite3_wal_checkpoint(db_meta, NULL);
  953. }
  954. struct ml_model_payload {
  955. uv_work_t request;
  956. struct metadata_wc *wc;
  957. Pvoid_t JudyL;
  958. size_t count;
  959. };
  960. struct scan_metadata_payload {
  961. uv_work_t request;
  962. struct metadata_wc *wc;
  963. BUFFER *work_buffer;
  964. uint32_t max_count;
  965. };
  966. struct host_context_load_thread {
  967. uv_thread_t thread;
  968. RRDHOST *host;
  969. bool busy;
  970. bool finished;
  971. };
  972. static void restore_host_context(void *arg)
  973. {
  974. struct host_context_load_thread *hclt = arg;
  975. RRDHOST *host = hclt->host;
  976. usec_t started_ut = now_monotonic_usec(); (void)started_ut;
  977. rrdhost_load_rrdcontext_data(host);
  978. usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
  979. rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD | RRDHOST_FLAG_CONTEXT_LOAD_IN_PROGRESS);
  980. #ifdef ENABLE_ACLK
  981. aclk_queue_node_info(host, false);
  982. #endif
  983. internal_error(true, "METADATA: 'host:%s' context load in %0.2f ms", rrdhost_hostname(host),
  984. (double)(ended_ut - started_ut) / USEC_PER_MS);
  985. __atomic_store_n(&hclt->finished, true, __ATOMIC_RELEASE);
  986. }
  987. // Callback after scan of hosts is done
  988. static void after_start_host_load_context(uv_work_t *req, int status __maybe_unused)
  989. {
  990. struct scan_metadata_payload *data = req->data;
  991. freez(data);
  992. }
  993. #define MAX_FIND_THREAD_RETRIES (10)
  994. static void cleanup_finished_threads(struct host_context_load_thread *hclt, size_t max_thread_slots, bool wait)
  995. {
  996. for (size_t index = 0; index < max_thread_slots; index++) {
  997. if (__atomic_load_n(&(hclt[index].finished), __ATOMIC_RELAXED)
  998. || (wait && __atomic_load_n(&(hclt[index].busy), __ATOMIC_ACQUIRE))) {
  999. int rc = uv_thread_join(&(hclt[index].thread));
  1000. if (rc)
  1001. netdata_log_error("Failed to join thread, rc = %d",rc);
  1002. __atomic_store_n(&(hclt[index].busy), false, __ATOMIC_RELEASE);
  1003. __atomic_store_n(&(hclt[index].finished), false, __ATOMIC_RELEASE);
  1004. }
  1005. }
  1006. }
  1007. static size_t find_available_thread_slot(struct host_context_load_thread *hclt, size_t max_thread_slots, size_t *found_index)
  1008. {
  1009. size_t retries = MAX_FIND_THREAD_RETRIES;
  1010. while (retries--) {
  1011. size_t index = 0;
  1012. while (index < max_thread_slots) {
  1013. if (false == __atomic_load_n(&(hclt[index].busy), __ATOMIC_ACQUIRE)) {
  1014. *found_index = index;
  1015. return true;
  1016. }
  1017. index++;
  1018. }
  1019. sleep_usec(10 * USEC_PER_MS);
  1020. }
  1021. return false;
  1022. }
  1023. static void start_all_host_load_context(uv_work_t *req __maybe_unused)
  1024. {
  1025. register_libuv_worker_jobs();
  1026. struct scan_metadata_payload *data = req->data;
  1027. UNUSED(data);
  1028. worker_is_busy(UV_EVENT_HOST_CONTEXT_LOAD);
  1029. usec_t started_ut = now_monotonic_usec(); (void)started_ut;
  1030. RRDHOST *host;
  1031. size_t max_threads = MIN(get_netdata_cpus() / 2, 6);
  1032. struct host_context_load_thread *hclt = callocz(max_threads, sizeof(*hclt));
  1033. size_t thread_index;
  1034. dfe_start_reentrant(rrdhost_root_index, host) {
  1035. if (rrdhost_flag_check(host, RRDHOST_FLAG_CONTEXT_LOAD_IN_PROGRESS) ||
  1036. !rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
  1037. continue;
  1038. rrdhost_flag_set(host, RRDHOST_FLAG_CONTEXT_LOAD_IN_PROGRESS);
  1039. internal_error(true, "METADATA: 'host:%s' loading context", rrdhost_hostname(host));
  1040. cleanup_finished_threads(hclt, max_threads, false);
  1041. bool found_slot = find_available_thread_slot(hclt, max_threads, &thread_index);
  1042. if (unlikely(!found_slot)) {
  1043. struct host_context_load_thread hclt_sync = {.host = host};
  1044. restore_host_context(&hclt_sync);
  1045. }
  1046. else {
  1047. __atomic_store_n(&hclt[thread_index].busy, true, __ATOMIC_RELAXED);
  1048. hclt[thread_index].host = host;
  1049. assert(0 == uv_thread_create(&hclt[thread_index].thread, restore_host_context, &hclt[thread_index]));
  1050. }
  1051. }
  1052. dfe_done(host);
  1053. cleanup_finished_threads(hclt, max_threads, true);
  1054. freez(hclt);
  1055. usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
  1056. internal_error(true, "METADATA: 'host:ALL' contexts loaded in %0.2f ms", (double)(ended_ut - started_ut) / USEC_PER_MS);
  1057. worker_is_idle();
  1058. }
  1059. // Callback after scan of hosts is done
  1060. static void after_metadata_hosts(uv_work_t *req, int status __maybe_unused)
  1061. {
  1062. struct scan_metadata_payload *data = req->data;
  1063. struct metadata_wc *wc = data->wc;
  1064. metadata_flag_clear(wc, METADATA_FLAG_PROCESSING);
  1065. internal_error(true, "METADATA: scanning hosts complete");
  1066. if (unlikely(wc->scan_complete)) {
  1067. completion_mark_complete(wc->scan_complete);
  1068. internal_error(true, "METADATA: Sending completion done");
  1069. }
  1070. freez(data);
  1071. }
  1072. static bool metadata_scan_host(RRDHOST *host, uint32_t max_count, bool use_transaction, BUFFER *work_buffer, size_t *query_counter) {
  1073. RRDSET *st;
  1074. int rc;
  1075. bool more_to_do = false;
  1076. uint32_t scan_count = 1;
  1077. if (use_transaction)
  1078. (void)db_execute(db_meta, "BEGIN TRANSACTION");
  1079. rrdset_foreach_reentrant(st, host) {
  1080. if (scan_count == max_count) {
  1081. more_to_do = true;
  1082. break;
  1083. }
  1084. if(rrdset_flag_check(st, RRDSET_FLAG_METADATA_UPDATE)) {
  1085. (*query_counter)++;
  1086. rrdset_flag_clear(st, RRDSET_FLAG_METADATA_UPDATE);
  1087. scan_count++;
  1088. buffer_flush(work_buffer);
  1089. rc = check_and_update_chart_labels(st, work_buffer, query_counter);
  1090. if (unlikely(rc))
  1091. error_report("METADATA: 'host:%s': Failed to update labels for chart %s", rrdhost_hostname(host), rrdset_name(st));
  1092. else
  1093. (*query_counter)++;
  1094. rc = store_chart_metadata(st);
  1095. if (unlikely(rc))
  1096. error_report("METADATA: 'host:%s': Failed to store metadata for chart %s", rrdhost_hostname(host), rrdset_name(st));
  1097. }
  1098. RRDDIM *rd;
  1099. rrddim_foreach_read(rd, st) {
  1100. if(rrddim_flag_check(rd, RRDDIM_FLAG_METADATA_UPDATE)) {
  1101. (*query_counter)++;
  1102. rrddim_flag_clear(rd, RRDDIM_FLAG_METADATA_UPDATE);
  1103. if (rrddim_option_check(rd, RRDDIM_OPTION_HIDDEN))
  1104. rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
  1105. else
  1106. rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
  1107. rc = store_dimension_metadata(rd);
  1108. if (unlikely(rc))
  1109. error_report("METADATA: 'host:%s': Failed to dimension metadata for chart %s. dimension %s",
  1110. rrdhost_hostname(host), rrdset_name(st),
  1111. rrddim_name(rd));
  1112. }
  1113. }
  1114. rrddim_foreach_done(rd);
  1115. }
  1116. rrdset_foreach_done(st);
  1117. if (use_transaction)
  1118. (void)db_execute(db_meta, "COMMIT TRANSACTION");
  1119. return more_to_do;
  1120. }
  1121. static void store_host_and_system_info(RRDHOST *host, size_t *query_counter)
  1122. {
  1123. if (unlikely(store_host_systeminfo(host))) {
  1124. error_report("METADATA: 'host:%s': Failed to store host updated system information in the database", rrdhost_hostname(host));
  1125. rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_INFO | RRDHOST_FLAG_METADATA_UPDATE);
  1126. }
  1127. else {
  1128. if (likely(query_counter))
  1129. (*query_counter)++;
  1130. }
  1131. if (unlikely(store_host_metadata(host))) {
  1132. error_report("METADATA: 'host:%s': Failed to store host info in the database", rrdhost_hostname(host));
  1133. rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_INFO | RRDHOST_FLAG_METADATA_UPDATE);
  1134. }
  1135. else {
  1136. if (likely(query_counter))
  1137. (*query_counter)++;
  1138. }
  1139. }
  1140. // Worker thread to scan hosts for pending metadata to store
  1141. static void start_metadata_hosts(uv_work_t *req __maybe_unused)
  1142. {
  1143. register_libuv_worker_jobs();
  1144. RRDHOST *host;
  1145. int transaction_started = 0;
  1146. struct scan_metadata_payload *data = req->data;
  1147. struct metadata_wc *wc = data->wc;
  1148. BUFFER *work_buffer = data->work_buffer;
  1149. usec_t all_started_ut = now_monotonic_usec(); (void)all_started_ut;
  1150. internal_error(true, "METADATA: checking all hosts...");
  1151. usec_t started_ut = now_monotonic_usec(); (void)started_ut;
  1152. bool run_again = false;
  1153. worker_is_busy(UV_EVENT_METADATA_STORE);
  1154. if (!data->max_count)
  1155. transaction_started = !db_execute(db_meta, "BEGIN TRANSACTION;");
  1156. dfe_start_reentrant(rrdhost_root_index, host) {
  1157. if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) || !rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_UPDATE))
  1158. continue;
  1159. size_t query_counter = 0; (void)query_counter;
  1160. rrdhost_flag_clear(host,RRDHOST_FLAG_METADATA_UPDATE);
  1161. if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_LABELS))) {
  1162. rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_LABELS);
  1163. int rc = exec_statement_with_uuid(SQL_DELETE_HOST_LABELS, &host->host_uuid);
  1164. if (likely(!rc)) {
  1165. query_counter++;
  1166. buffer_flush(work_buffer);
  1167. struct query_build tmp = {.sql = work_buffer, .count = 0};
  1168. uuid_unparse_lower(host->host_uuid, tmp.uuid_str);
  1169. rrdlabels_walkthrough_read(host->rrdlabels, host_label_store_to_sql_callback, &tmp);
  1170. buffer_strcat(work_buffer, " ON CONFLICT (host_id, label_key) DO UPDATE SET source_type = excluded.source_type, label_value=excluded.label_value, date_created=UNIXEPOCH()");
  1171. rc = db_execute(db_meta, buffer_tostring(work_buffer));
  1172. if (unlikely(rc)) {
  1173. error_report("METADATA: 'host:%s': failed to update metadata host labels", rrdhost_hostname(host));
  1174. rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_LABELS | RRDHOST_FLAG_METADATA_UPDATE);
  1175. }
  1176. else
  1177. query_counter++;
  1178. } else {
  1179. error_report("METADATA: 'host:%s': failed to delete old host labels", rrdhost_hostname(host));
  1180. rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_LABELS | RRDHOST_FLAG_METADATA_UPDATE);
  1181. }
  1182. }
  1183. if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_CLAIMID))) {
  1184. rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_CLAIMID);
  1185. uuid_t uuid;
  1186. int rc;
  1187. if (likely(host->aclk_state.claimed_id && !uuid_parse(host->aclk_state.claimed_id, uuid)))
  1188. rc = store_claim_id(&host->host_uuid, &uuid);
  1189. else
  1190. rc = store_claim_id(&host->host_uuid, NULL);
  1191. if (unlikely(rc))
  1192. rrdhost_flag_set(host, RRDHOST_FLAG_METADATA_CLAIMID | RRDHOST_FLAG_METADATA_UPDATE);
  1193. else
  1194. query_counter++;
  1195. }
  1196. if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_INFO))) {
  1197. rrdhost_flag_clear(host, RRDHOST_FLAG_METADATA_INFO);
  1198. store_host_and_system_info(host, &query_counter);
  1199. }
  1200. // For clarity
  1201. bool use_transaction = data->max_count;
  1202. if (unlikely(metadata_scan_host(host, data->max_count, use_transaction, work_buffer, &query_counter))) {
  1203. run_again = true;
  1204. rrdhost_flag_set(host,RRDHOST_FLAG_METADATA_UPDATE);
  1205. internal_error(true,"METADATA: 'host:%s': scheduling another run, more charts to store", rrdhost_hostname(host));
  1206. }
  1207. usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
  1208. internal_error(true, "METADATA: 'host:%s': saved metadata with %zu SQL statements, in %0.2f ms",
  1209. rrdhost_hostname(host), query_counter,
  1210. (double)(ended_ut - started_ut) / USEC_PER_MS);
  1211. }
  1212. dfe_done(host);
  1213. if (!data->max_count && transaction_started)
  1214. transaction_started = db_execute(db_meta, "COMMIT TRANSACTION;");
  1215. usec_t all_ended_ut = now_monotonic_usec(); (void)all_ended_ut;
  1216. internal_error(true, "METADATA: checking all hosts completed in %0.2f ms",
  1217. (double)(all_ended_ut - all_started_ut) / USEC_PER_MS);
  1218. if (unlikely(run_again))
  1219. wc->metadata_check_after = now_realtime_sec() + METADATA_HOST_CHECK_IMMEDIATE;
  1220. else {
  1221. wc->metadata_check_after = now_realtime_sec() + METADATA_HOST_CHECK_INTERVAL;
  1222. run_metadata_cleanup(wc);
  1223. }
  1224. worker_is_idle();
  1225. }
  1226. // Callback after scan of hosts is done
  1227. static void after_start_ml_model_load(uv_work_t *req, int status __maybe_unused)
  1228. {
  1229. struct ml_model_payload *ml_data = req->data;
  1230. struct metadata_wc *wc = ml_data->wc;
  1231. metadata_flag_clear(wc, METADATA_FLAG_ML_LOADING);
  1232. JudyLFreeArray(&ml_data->JudyL, PJE0);
  1233. freez(ml_data);
  1234. }
  1235. static void start_ml_model_load(uv_work_t *req __maybe_unused)
  1236. {
  1237. register_libuv_worker_jobs();
  1238. struct ml_model_payload *ml_data = req->data;
  1239. worker_is_busy(UV_EVENT_METADATA_ML_LOAD);
  1240. Pvoid_t *PValue;
  1241. Word_t Index = 0;
  1242. bool first = true;
  1243. RRDDIM *rd;
  1244. RRDDIM_ACQUIRED *rda;
  1245. internal_error(true, "Batch ML load loader, %zu items", ml_data->count);
  1246. while((PValue = JudyLFirstThenNext(ml_data->JudyL, &Index, &first))) {
  1247. UNUSED(PValue);
  1248. rda = (RRDDIM_ACQUIRED *) Index;
  1249. rd = rrddim_acquired_to_rrddim(rda);
  1250. ml_dimension_load_models(rd);
  1251. rrddim_acquired_release(rda);
  1252. }
  1253. worker_is_idle();
  1254. }
  1255. static void metadata_event_loop(void *arg)
  1256. {
  1257. worker_register("METASYNC");
  1258. worker_register_job_name(METADATA_DATABASE_NOOP, "noop");
  1259. worker_register_job_name(METADATA_DATABASE_TIMER, "timer");
  1260. worker_register_job_name(METADATA_DEL_DIMENSION, "delete dimension");
  1261. worker_register_job_name(METADATA_STORE_CLAIM_ID, "add claim id");
  1262. worker_register_job_name(METADATA_ADD_HOST_INFO, "add host info");
  1263. worker_register_job_name(METADATA_MAINTENANCE, "maintenance");
  1264. worker_register_job_name(METADATA_ML_LOAD_MODELS, "ml load models");
  1265. int ret;
  1266. uv_loop_t *loop;
  1267. unsigned cmd_batch_size;
  1268. struct metadata_wc *wc = arg;
  1269. enum metadata_opcode opcode;
  1270. uv_thread_set_name_np(wc->thread, "METASYNC");
  1271. loop = wc->loop = mallocz(sizeof(uv_loop_t));
  1272. ret = uv_loop_init(loop);
  1273. if (ret) {
  1274. netdata_log_error("uv_loop_init(): %s", uv_strerror(ret));
  1275. goto error_after_loop_init;
  1276. }
  1277. loop->data = wc;
  1278. ret = uv_async_init(wc->loop, &wc->async, async_cb);
  1279. if (ret) {
  1280. netdata_log_error("uv_async_init(): %s", uv_strerror(ret));
  1281. goto error_after_async_init;
  1282. }
  1283. wc->async.data = wc;
  1284. ret = uv_timer_init(loop, &wc->timer_req);
  1285. if (ret) {
  1286. netdata_log_error("uv_timer_init(): %s", uv_strerror(ret));
  1287. goto error_after_timer_init;
  1288. }
  1289. wc->timer_req.data = wc;
  1290. fatal_assert(0 == uv_timer_start(&wc->timer_req, timer_cb, TIMER_INITIAL_PERIOD_MS, TIMER_REPEAT_PERIOD_MS));
  1291. netdata_log_info("Starting metadata sync thread with %d entries command queue", METADATA_CMD_Q_MAX_SIZE);
  1292. struct metadata_cmd cmd;
  1293. memset(&cmd, 0, sizeof(cmd));
  1294. metadata_flag_clear(wc, METADATA_FLAG_PROCESSING);
  1295. wc->metadata_check_after = now_realtime_sec() + METADATA_HOST_CHECK_FIRST_CHECK;
  1296. int shutdown = 0;
  1297. completion_mark_complete(&wc->init_complete);
  1298. BUFFER *work_buffer = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
  1299. struct scan_metadata_payload *data;
  1300. struct ml_model_payload *ml_data = NULL;
  1301. while (shutdown == 0 || (wc->flags & METADATA_FLAG_PROCESSING)) {
  1302. uuid_t *uuid;
  1303. RRDHOST *host = NULL;
  1304. worker_is_idle();
  1305. uv_run(loop, UV_RUN_DEFAULT);
  1306. /* wait for commands */
  1307. cmd_batch_size = 0;
  1308. do {
  1309. if (unlikely(cmd_batch_size >= METADATA_MAX_BATCH_SIZE))
  1310. break;
  1311. cmd = metadata_deq_cmd(wc);
  1312. opcode = cmd.opcode;
  1313. if (unlikely(opcode == METADATA_DATABASE_NOOP && metadata_flag_check(wc, METADATA_FLAG_SHUTDOWN))) {
  1314. shutdown = 1;
  1315. continue;
  1316. }
  1317. ++cmd_batch_size;
  1318. if (likely(opcode != METADATA_DATABASE_NOOP))
  1319. worker_is_busy(opcode);
  1320. // Have pending ML models to load?
  1321. if (opcode != METADATA_ML_LOAD_MODELS && ml_data && ml_data->count) {
  1322. static usec_t ml_submit_last = 0;
  1323. usec_t now = now_monotonic_usec();
  1324. if (!ml_submit_last)
  1325. ml_submit_last = now;
  1326. if (!metadata_flag_check(wc, METADATA_FLAG_ML_LOADING) && (now - ml_submit_last > 150 * USEC_PER_MS)) {
  1327. metadata_flag_set(wc, METADATA_FLAG_ML_LOADING);
  1328. if (unlikely(uv_queue_work(loop, &ml_data->request, start_ml_model_load, after_start_ml_model_load)))
  1329. metadata_flag_clear(wc, METADATA_FLAG_ML_LOADING);
  1330. else {
  1331. ml_submit_last = now;
  1332. ml_data = NULL;
  1333. }
  1334. }
  1335. }
  1336. switch (opcode) {
  1337. case METADATA_DATABASE_NOOP:
  1338. case METADATA_DATABASE_TIMER:
  1339. break;
  1340. case METADATA_ML_LOAD_MODELS: {
  1341. RRDDIM *rd = (RRDDIM *) cmd.param[0];
  1342. RRDDIM_ACQUIRED *rda = rrddim_find_and_acquire(rd->rrdset, rrddim_id(rd));
  1343. if (likely(rda)) {
  1344. if (!ml_data) {
  1345. ml_data = callocz(1,sizeof(*ml_data));
  1346. ml_data->request.data = ml_data;
  1347. ml_data->wc = wc;
  1348. }
  1349. JudyLIns(&ml_data->JudyL, (Word_t)rda, PJE0);
  1350. ml_data->count++;
  1351. }
  1352. break;
  1353. }
  1354. case METADATA_DEL_DIMENSION:
  1355. uuid = (uuid_t *) cmd.param[0];
  1356. if (likely(dimension_can_be_deleted(uuid, NULL, false)))
  1357. delete_dimension_uuid(uuid, NULL, false);
  1358. freez(uuid);
  1359. break;
  1360. case METADATA_STORE_CLAIM_ID:
  1361. store_claim_id((uuid_t *) cmd.param[0], (uuid_t *) cmd.param[1]);
  1362. freez((void *) cmd.param[0]);
  1363. freez((void *) cmd.param[1]);
  1364. break;
  1365. case METADATA_ADD_HOST_INFO:
  1366. host = (RRDHOST *) cmd.param[0];
  1367. store_host_and_system_info(host, NULL);
  1368. break;
  1369. case METADATA_SCAN_HOSTS:
  1370. if (unlikely(metadata_flag_check(wc, METADATA_FLAG_PROCESSING)))
  1371. break;
  1372. if (unittest_running)
  1373. break;
  1374. data = mallocz(sizeof(*data));
  1375. data->request.data = data;
  1376. data->wc = wc;
  1377. data->work_buffer = work_buffer;
  1378. if (unlikely(cmd.completion)) {
  1379. data->max_count = 0; // 0 will process all pending updates
  1380. cmd.completion = NULL; // Do not complete after launching worker (worker will do)
  1381. }
  1382. else
  1383. data->max_count = 5000;
  1384. metadata_flag_set(wc, METADATA_FLAG_PROCESSING);
  1385. if (unlikely(
  1386. uv_queue_work(loop,&data->request,
  1387. start_metadata_hosts,
  1388. after_metadata_hosts))) {
  1389. // Failed to launch worker -- let the event loop handle completion
  1390. cmd.completion = wc->scan_complete;
  1391. freez(data);
  1392. metadata_flag_clear(wc, METADATA_FLAG_PROCESSING);
  1393. }
  1394. break;
  1395. case METADATA_LOAD_HOST_CONTEXT:;
  1396. if (unittest_running)
  1397. break;
  1398. data = callocz(1,sizeof(*data));
  1399. data->request.data = data;
  1400. data->wc = wc;
  1401. if (unlikely(
  1402. uv_queue_work(loop,&data->request, start_all_host_load_context,
  1403. after_start_host_load_context))) {
  1404. freez(data);
  1405. }
  1406. break;
  1407. case METADATA_UNITTEST:;
  1408. struct thread_unittest *tu = (struct thread_unittest *) cmd.param[0];
  1409. sleep_usec(1000); // processing takes 1ms
  1410. __atomic_fetch_add(&tu->processed, 1, __ATOMIC_SEQ_CST);
  1411. break;
  1412. default:
  1413. break;
  1414. }
  1415. if (cmd.completion)
  1416. completion_mark_complete(cmd.completion);
  1417. } while (opcode != METADATA_DATABASE_NOOP);
  1418. }
  1419. if (!uv_timer_stop(&wc->timer_req))
  1420. uv_close((uv_handle_t *)&wc->timer_req, NULL);
  1421. uv_close((uv_handle_t *)&wc->async, NULL);
  1422. int rc;
  1423. do {
  1424. rc = uv_loop_close(loop);
  1425. } while (rc != UV_EBUSY);
  1426. buffer_free(work_buffer);
  1427. freez(loop);
  1428. worker_unregister();
  1429. netdata_log_info("METADATA: Shutting down event loop");
  1430. completion_mark_complete(&wc->init_complete);
  1431. completion_destroy(wc->scan_complete);
  1432. freez(wc->scan_complete);
  1433. metadata_free_cmd_queue(wc);
  1434. return;
  1435. error_after_timer_init:
  1436. uv_close((uv_handle_t *)&wc->async, NULL);
  1437. error_after_async_init:
  1438. fatal_assert(0 == uv_loop_close(loop));
  1439. error_after_loop_init:
  1440. freez(loop);
  1441. worker_unregister();
  1442. }
  1443. struct metadata_wc metasync_worker = {.loop = NULL};
  1444. void metadata_sync_shutdown(void)
  1445. {
  1446. completion_init(&metasync_worker.init_complete);
  1447. struct metadata_cmd cmd;
  1448. memset(&cmd, 0, sizeof(cmd));
  1449. netdata_log_info("METADATA: Sending a shutdown command");
  1450. cmd.opcode = METADATA_SYNC_SHUTDOWN;
  1451. metadata_enq_cmd(&metasync_worker, &cmd);
  1452. /* wait for metadata thread to shut down */
  1453. netdata_log_info("METADATA: Waiting for shutdown ACK");
  1454. completion_wait_for(&metasync_worker.init_complete);
  1455. completion_destroy(&metasync_worker.init_complete);
  1456. netdata_log_info("METADATA: Shutdown complete");
  1457. }
  1458. void metadata_sync_shutdown_prepare(void)
  1459. {
  1460. if (unlikely(!metasync_worker.loop))
  1461. return;
  1462. struct metadata_cmd cmd;
  1463. memset(&cmd, 0, sizeof(cmd));
  1464. struct metadata_wc *wc = &metasync_worker;
  1465. struct completion *compl = mallocz(sizeof(*compl));
  1466. completion_init(compl);
  1467. __atomic_store_n(&wc->scan_complete, compl, __ATOMIC_RELAXED);
  1468. netdata_log_info("METADATA: Sending a scan host command");
  1469. uint32_t max_wait_iterations = 2000;
  1470. while (unlikely(metadata_flag_check(&metasync_worker, METADATA_FLAG_PROCESSING)) && max_wait_iterations--) {
  1471. if (max_wait_iterations == 1999)
  1472. netdata_log_info("METADATA: Current worker is running; waiting to finish");
  1473. sleep_usec(1000);
  1474. }
  1475. cmd.opcode = METADATA_SCAN_HOSTS;
  1476. metadata_enq_cmd(&metasync_worker, &cmd);
  1477. netdata_log_info("METADATA: Waiting for host scan completion");
  1478. completion_wait_for(wc->scan_complete);
  1479. netdata_log_info("METADATA: Host scan complete; can continue with shutdown");
  1480. }
  1481. // -------------------------------------------------------------
  1482. // Init function called on agent startup
  1483. void metadata_sync_init(void)
  1484. {
  1485. struct metadata_wc *wc = &metasync_worker;
  1486. memset(wc, 0, sizeof(*wc));
  1487. metadata_init_cmd_queue(wc);
  1488. completion_init(&wc->init_complete);
  1489. fatal_assert(0 == uv_thread_create(&(wc->thread), metadata_event_loop, wc));
  1490. completion_wait_for(&wc->init_complete);
  1491. completion_destroy(&wc->init_complete);
  1492. netdata_log_info("SQLite metadata sync initialization complete");
  1493. }
  1494. // Helpers
  1495. static inline void queue_metadata_cmd(enum metadata_opcode opcode, const void *param0, const void *param1)
  1496. {
  1497. struct metadata_cmd cmd;
  1498. cmd.opcode = opcode;
  1499. cmd.param[0] = param0;
  1500. cmd.param[1] = param1;
  1501. cmd.completion = NULL;
  1502. metadata_enq_cmd(&metasync_worker, &cmd);
  1503. }
  1504. // Public
  1505. void metaqueue_delete_dimension_uuid(uuid_t *uuid)
  1506. {
  1507. if (unlikely(!metasync_worker.loop))
  1508. return;
  1509. uuid_t *use_uuid = mallocz(sizeof(*uuid));
  1510. uuid_copy(*use_uuid, *uuid);
  1511. queue_metadata_cmd(METADATA_DEL_DIMENSION, use_uuid, NULL);
  1512. }
  1513. void metaqueue_store_claim_id(uuid_t *host_uuid, uuid_t *claim_uuid)
  1514. {
  1515. if (unlikely(!host_uuid))
  1516. return;
  1517. uuid_t *local_host_uuid = mallocz(sizeof(*host_uuid));
  1518. uuid_t *local_claim_uuid = NULL;
  1519. uuid_copy(*local_host_uuid, *host_uuid);
  1520. if (likely(claim_uuid)) {
  1521. local_claim_uuid = mallocz(sizeof(*claim_uuid));
  1522. uuid_copy(*local_claim_uuid, *claim_uuid);
  1523. }
  1524. queue_metadata_cmd(METADATA_STORE_CLAIM_ID, local_host_uuid, local_claim_uuid);
  1525. }
  1526. void metaqueue_host_update_info(RRDHOST *host)
  1527. {
  1528. if (unlikely(!metasync_worker.loop))
  1529. return;
  1530. queue_metadata_cmd(METADATA_ADD_HOST_INFO, host, NULL);
  1531. }
  1532. void metaqueue_ml_load_models(RRDDIM *rd)
  1533. {
  1534. if (unlikely(!metasync_worker.loop))
  1535. return;
  1536. queue_metadata_cmd(METADATA_ML_LOAD_MODELS, rd, NULL);
  1537. }
  1538. void metadata_queue_load_host_context(RRDHOST *host)
  1539. {
  1540. if (unlikely(!metasync_worker.loop))
  1541. return;
  1542. queue_metadata_cmd(METADATA_LOAD_HOST_CONTEXT, host, NULL);
  1543. }
  1544. //
  1545. // unitests
  1546. //
  1547. static void *unittest_queue_metadata(void *arg) {
  1548. struct thread_unittest *tu = arg;
  1549. struct metadata_cmd cmd;
  1550. cmd.opcode = METADATA_UNITTEST;
  1551. cmd.param[0] = tu;
  1552. cmd.param[1] = NULL;
  1553. cmd.completion = NULL;
  1554. metadata_enq_cmd(&metasync_worker, &cmd);
  1555. do {
  1556. __atomic_fetch_add(&tu->added, 1, __ATOMIC_SEQ_CST);
  1557. metadata_enq_cmd(&metasync_worker, &cmd);
  1558. sleep_usec(10000);
  1559. } while (!__atomic_load_n(&tu->join, __ATOMIC_RELAXED));
  1560. return arg;
  1561. }
  1562. static void *metadata_unittest_threads(void)
  1563. {
  1564. unsigned done;
  1565. struct thread_unittest tu = {
  1566. .join = 0,
  1567. .added = 0,
  1568. .processed = 0,
  1569. .done = &done,
  1570. };
  1571. // Queue messages / Time it
  1572. time_t seconds_to_run = 5;
  1573. int threads_to_create = 4;
  1574. fprintf(
  1575. stderr,
  1576. "\nChecking metadata queue using %d threads for %lld seconds...\n",
  1577. threads_to_create,
  1578. (long long)seconds_to_run);
  1579. netdata_thread_t threads[threads_to_create];
  1580. tu.join = 0;
  1581. for (int i = 0; i < threads_to_create; i++) {
  1582. char buf[100 + 1];
  1583. snprintf(buf, 100, "META[%d]", i);
  1584. netdata_thread_create(
  1585. &threads[i],
  1586. buf,
  1587. NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
  1588. unittest_queue_metadata,
  1589. &tu);
  1590. }
  1591. (void) uv_async_send(&metasync_worker.async);
  1592. sleep_usec(seconds_to_run * USEC_PER_SEC);
  1593. __atomic_store_n(&tu.join, 1, __ATOMIC_RELAXED);
  1594. for (int i = 0; i < threads_to_create; i++) {
  1595. void *retval;
  1596. netdata_thread_join(threads[i], &retval);
  1597. }
  1598. sleep_usec(5 * USEC_PER_SEC);
  1599. fprintf(stderr, "Added %u elements, processed %u\n", tu.added, tu.processed);
  1600. return 0;
  1601. }
  1602. int metadata_unittest(void)
  1603. {
  1604. metadata_sync_init();
  1605. // Queue items for a specific period of time
  1606. metadata_unittest_threads();
  1607. metadata_sync_shutdown();
  1608. return 0;
  1609. }