sqlite_aclk.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #include "sqlite_functions.h"
  3. #include "sqlite_aclk.h"
  4. #include "sqlite_aclk_node.h"
  5. struct aclk_sync_config_s {
  6. uv_thread_t thread;
  7. uv_loop_t loop;
  8. uv_timer_t timer_req;
  9. time_t cleanup_after; // Start a cleanup after this timestamp
  10. uv_async_t async;
  11. /* FIFO command queue */
  12. uv_mutex_t cmd_mutex;
  13. uv_cond_t cmd_cond;
  14. bool initialized;
  15. volatile unsigned queue_size;
  16. struct aclk_database_cmdqueue cmd_queue;
  17. } aclk_sync_config = { 0 };
  18. void sanity_check(void) {
  19. // make sure the compiler will stop on misconfigurations
  20. BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
  21. }
  22. int aclk_database_enq_cmd_noblock(struct aclk_database_cmd *cmd)
  23. {
  24. unsigned queue_size;
  25. /* wait for free space in queue */
  26. uv_mutex_lock(&aclk_sync_config.cmd_mutex);
  27. if ((queue_size = aclk_sync_config.queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
  28. uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
  29. return 1;
  30. }
  31. fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
  32. /* enqueue command */
  33. aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.tail] = *cmd;
  34. aclk_sync_config.cmd_queue.tail = aclk_sync_config.cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
  35. aclk_sync_config.cmd_queue.tail + 1 : 0;
  36. aclk_sync_config.queue_size = queue_size + 1;
  37. uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
  38. return 0;
  39. }
  40. static void aclk_database_enq_cmd(struct aclk_database_cmd *cmd)
  41. {
  42. unsigned queue_size;
  43. /* wait for free space in queue */
  44. uv_mutex_lock(&aclk_sync_config.cmd_mutex);
  45. while ((queue_size = aclk_sync_config.queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
  46. uv_cond_wait(&aclk_sync_config.cmd_cond, &aclk_sync_config.cmd_mutex);
  47. }
  48. fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
  49. /* enqueue command */
  50. aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.tail] = *cmd;
  51. aclk_sync_config.cmd_queue.tail = aclk_sync_config.cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
  52. aclk_sync_config.cmd_queue.tail + 1 : 0;
  53. aclk_sync_config.queue_size = queue_size + 1;
  54. uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
  55. /* wake up event loop */
  56. int rc = uv_async_send(&aclk_sync_config.async);
  57. if (unlikely(rc))
  58. debug(D_ACLK_SYNC, "Failed to wake up event loop");
  59. }
  60. enum {
  61. IDX_HOST_ID,
  62. IDX_HOSTNAME,
  63. IDX_REGISTRY,
  64. IDX_UPDATE_EVERY,
  65. IDX_OS,
  66. IDX_TIMEZONE,
  67. IDX_TAGS,
  68. IDX_HOPS,
  69. IDX_MEMORY_MODE,
  70. IDX_ABBREV_TIMEZONE,
  71. IDX_UTC_OFFSET,
  72. IDX_PROGRAM_NAME,
  73. IDX_PROGRAM_VERSION,
  74. IDX_ENTRIES,
  75. IDX_HEALTH_ENABLED,
  76. };
  77. static int create_host_callback(void *data, int argc, char **argv, char **column)
  78. {
  79. int *number_of_chidren = data;
  80. UNUSED(argc);
  81. UNUSED(column);
  82. char guid[UUID_STR_LEN];
  83. uuid_unparse_lower(*(uuid_t *)argv[IDX_HOST_ID], guid);
  84. struct rrdhost_system_info *system_info = callocz(1, sizeof(struct rrdhost_system_info));
  85. __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(struct rrdhost_system_info), __ATOMIC_RELAXED);
  86. system_info->hops = str2i((const char *) argv[IDX_HOPS]);
  87. sql_build_host_system_info((uuid_t *)argv[IDX_HOST_ID], system_info);
  88. RRDHOST *host = rrdhost_find_or_create(
  89. (const char *) argv[IDX_HOSTNAME]
  90. , (const char *) argv[IDX_REGISTRY]
  91. , guid
  92. , (const char *) argv[IDX_OS]
  93. , (const char *) argv[IDX_TIMEZONE]
  94. , (const char *) argv[IDX_ABBREV_TIMEZONE]
  95. , (int32_t) (argv[IDX_UTC_OFFSET] ? str2uint32_t(argv[IDX_UTC_OFFSET], NULL) : 0)
  96. , (const char *) argv[IDX_TAGS]
  97. , (const char *) (argv[IDX_PROGRAM_NAME] ? argv[IDX_PROGRAM_NAME] : "unknown")
  98. , (const char *) (argv[IDX_PROGRAM_VERSION] ? argv[IDX_PROGRAM_VERSION] : "unknown")
  99. , argv[IDX_UPDATE_EVERY] ? str2i(argv[IDX_UPDATE_EVERY]) : 1
  100. , argv[IDX_ENTRIES] ? str2i(argv[IDX_ENTRIES]) : 0
  101. , default_rrd_memory_mode
  102. , 0 // health
  103. , 0 // rrdpush enabled
  104. , NULL //destination
  105. , NULL // api key
  106. , NULL // send charts matching
  107. , false // rrdpush_enable_replication
  108. , 0 // rrdpush_seconds_to_replicate
  109. , 0 // rrdpush_replication_step
  110. , system_info
  111. , 1
  112. );
  113. if (likely(host))
  114. host->rrdlabels = sql_load_host_labels((uuid_t *)argv[IDX_HOST_ID]);
  115. (*number_of_chidren)++;
  116. #ifdef NETDATA_INTERNAL_CHECKS
  117. char node_str[UUID_STR_LEN] = "<none>";
  118. if (likely(host->node_id))
  119. uuid_unparse_lower(*host->node_id, node_str);
  120. internal_error(true, "Adding archived host \"%s\" with GUID \"%s\" node id = \"%s\"", rrdhost_hostname(host), host->machine_guid, node_str);
  121. #endif
  122. return 0;
  123. }
  124. #ifdef ENABLE_ACLK
  125. static struct aclk_database_cmd aclk_database_deq_cmd(void)
  126. {
  127. struct aclk_database_cmd ret;
  128. unsigned queue_size;
  129. uv_mutex_lock(&aclk_sync_config.cmd_mutex);
  130. queue_size = aclk_sync_config.queue_size;
  131. if (queue_size == 0) {
  132. memset(&ret, 0, sizeof(ret));
  133. ret.opcode = ACLK_DATABASE_NOOP;
  134. ret.completion = NULL;
  135. } else {
  136. /* dequeue command */
  137. ret = aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.head];
  138. if (queue_size == 1) {
  139. aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.tail = 0;
  140. } else {
  141. aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.head != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
  142. aclk_sync_config.cmd_queue.head + 1 : 0;
  143. }
  144. aclk_sync_config.queue_size = queue_size - 1;
  145. /* wake up producers */
  146. uv_cond_signal(&aclk_sync_config.cmd_cond);
  147. }
  148. uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
  149. return ret;
  150. }
  151. #define SQL_SELECT_HOST_BY_UUID "SELECT host_id FROM host WHERE host_id = @host_id;"
  152. static int is_host_available(uuid_t *host_id)
  153. {
  154. sqlite3_stmt *res = NULL;
  155. int rc;
  156. if (unlikely(!db_meta)) {
  157. if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
  158. error_report("Database has not been initialized");
  159. return 1;
  160. }
  161. rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_UUID, -1, &res, 0);
  162. if (unlikely(rc != SQLITE_OK)) {
  163. error_report("Failed to prepare statement to select node instance information for a node");
  164. return 1;
  165. }
  166. rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
  167. if (unlikely(rc != SQLITE_OK)) {
  168. error_report("Failed to bind host_id parameter to check host existence");
  169. goto failed;
  170. }
  171. rc = sqlite3_step_monitored(res);
  172. failed:
  173. if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
  174. error_report("Failed to finalize the prepared statement when checking host existence");
  175. return (rc == SQLITE_ROW);
  176. }
  177. // OPCODE: ACLK_DATABASE_DELETE_HOST
  178. static void sql_delete_aclk_table_list(char *host_guid)
  179. {
  180. char uuid_str[UUID_STR_LEN];
  181. char host_str[UUID_STR_LEN];
  182. int rc;
  183. uuid_t host_uuid;
  184. if (unlikely(!host_guid))
  185. return;
  186. rc = uuid_parse(host_guid, host_uuid);
  187. freez(host_guid);
  188. if (rc)
  189. return;
  190. uuid_unparse_lower(host_uuid, host_str);
  191. uuid_unparse_lower_fix(&host_uuid, uuid_str);
  192. debug(D_ACLK_SYNC, "Checking if I should delete aclk tables for node %s", host_str);
  193. if (is_host_available(&host_uuid)) {
  194. debug(D_ACLK_SYNC, "Host %s exists, not deleting aclk sync tables", host_str);
  195. return;
  196. }
  197. debug(D_ACLK_SYNC, "Host %s does NOT exist, can delete aclk sync tables", host_str);
  198. sqlite3_stmt *res = NULL;
  199. BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE, &netdata_buffers_statistics.buffers_sqlite);
  200. buffer_sprintf(sql,"SELECT 'drop '||type||' IF EXISTS '||name||';' FROM sqlite_schema " \
  201. "WHERE name LIKE 'aclk_%%_%s' AND type IN ('table', 'trigger', 'index');", uuid_str);
  202. rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
  203. if (rc != SQLITE_OK) {
  204. error_report("Failed to prepare statement to clean up aclk tables");
  205. goto fail;
  206. }
  207. buffer_flush(sql);
  208. while (sqlite3_step_monitored(res) == SQLITE_ROW)
  209. buffer_strcat(sql, (char *) sqlite3_column_text(res, 0));
  210. rc = sqlite3_finalize(res);
  211. if (unlikely(rc != SQLITE_OK))
  212. error_report("Failed to finalize statement to clean up aclk tables, rc = %d", rc);
  213. rc = db_execute(db_meta, buffer_tostring(sql));
  214. if (unlikely(rc))
  215. error("Failed to drop unused ACLK tables");
  216. fail:
  217. buffer_free(sql);
  218. }
  219. static int sql_check_aclk_table(void *data __maybe_unused, int argc __maybe_unused, char **argv __maybe_unused, char **column __maybe_unused)
  220. {
  221. debug(D_ACLK_SYNC,"Scheduling aclk sync table check for node %s", (char *) argv[0]);
  222. struct aclk_database_cmd cmd;
  223. memset(&cmd, 0, sizeof(cmd));
  224. cmd.opcode = ACLK_DATABASE_DELETE_HOST;
  225. cmd.param[0] = strdupz((char *) argv[0]);
  226. aclk_database_enq_cmd_noblock(&cmd);
  227. return 0;
  228. }
  229. #define SQL_SELECT_ACLK_ACTIVE_LIST "SELECT REPLACE(SUBSTR(name,19),'_','-') FROM sqlite_schema " \
  230. "WHERE name LIKE 'aclk_chart_latest_%' AND type IN ('table');"
  231. static void sql_check_aclk_table_list(void)
  232. {
  233. char *err_msg = NULL;
  234. debug(D_ACLK_SYNC,"Cleaning tables for nodes that do not exist");
  235. int rc = sqlite3_exec_monitored(db_meta, SQL_SELECT_ACLK_ACTIVE_LIST, sql_check_aclk_table, NULL, &err_msg);
  236. if (rc != SQLITE_OK) {
  237. error_report("Query failed when trying to check for obsolete ACLK sync tables, %s", err_msg);
  238. sqlite3_free(err_msg);
  239. }
  240. }
  241. #define SQL_ALERT_CLEANUP "DELETE FROM aclk_alert_%s WHERE date_submitted IS NOT NULL AND CAST(date_cloud_ack AS INT) < unixepoch()-%d;"
  242. static int sql_maint_aclk_sync_database(void *data __maybe_unused, int argc __maybe_unused, char **argv, char **column __maybe_unused)
  243. {
  244. char sql[512];
  245. snprintfz(sql,511, SQL_ALERT_CLEANUP, (char *) argv[0], ACLK_DELETE_ACK_ALERTS_INTERNAL);
  246. if (unlikely(db_execute(db_meta, sql)))
  247. error_report("Failed to clean stale ACLK alert entries");
  248. return 0;
  249. }
  250. #define SQL_SELECT_ACLK_ALERT_LIST "SELECT SUBSTR(name,12) FROM sqlite_schema WHERE name LIKE 'aclk_alert_%' AND type IN ('table');"
  251. static void sql_maint_aclk_sync_database_all(void)
  252. {
  253. char *err_msg = NULL;
  254. debug(D_ACLK_SYNC,"Cleaning tables for nodes that do not exist");
  255. int rc = sqlite3_exec_monitored(db_meta, SQL_SELECT_ACLK_ALERT_LIST, sql_maint_aclk_sync_database, NULL, &err_msg);
  256. if (rc != SQLITE_OK) {
  257. error_report("Query failed when trying to check for obsolete ACLK sync tables, %s", err_msg);
  258. sqlite3_free(err_msg);
  259. }
  260. }
  261. static int aclk_config_parameters(void *data __maybe_unused, int argc __maybe_unused, char **argv, char **column __maybe_unused)
  262. {
  263. char uuid_str[GUID_LEN + 1];
  264. uuid_unparse_lower(*((uuid_t *) argv[0]), uuid_str);
  265. RRDHOST *host = rrdhost_find_by_guid(uuid_str);
  266. if (host == localhost)
  267. return 0;
  268. sql_create_aclk_table(host, (uuid_t *) argv[0], (uuid_t *) argv[1]);
  269. return 0;
  270. }
  271. static void async_cb(uv_async_t *handle)
  272. {
  273. uv_stop(handle->loop);
  274. uv_update_time(handle->loop);
  275. }
  276. #define TIMER_PERIOD_MS (1000)
  277. static void timer_cb(uv_timer_t *handle)
  278. {
  279. uv_stop(handle->loop);
  280. uv_update_time(handle->loop);
  281. struct aclk_sync_config_s *config = handle->data;
  282. struct aclk_database_cmd cmd;
  283. memset(&cmd, 0, sizeof(cmd));
  284. time_t now = now_realtime_sec();
  285. if (config->cleanup_after && config->cleanup_after < now) {
  286. cmd.opcode = ACLK_DATABASE_CLEANUP;
  287. if (!aclk_database_enq_cmd_noblock(&cmd))
  288. config->cleanup_after += ACLK_DATABASE_CLEANUP_INTERVAL;
  289. }
  290. if (aclk_connected) {
  291. cmd.opcode = ACLK_DATABASE_PUSH_ALERT;
  292. aclk_database_enq_cmd_noblock(&cmd);
  293. aclk_check_node_info_and_collectors();
  294. }
  295. }
  296. static void aclk_synchronization(void *arg __maybe_unused)
  297. {
  298. struct aclk_sync_config_s *config = arg;
  299. uv_thread_set_name_np(config->thread, "ACLKSYNC");
  300. worker_register("ACLKSYNC");
  301. service_register(SERVICE_THREAD_TYPE_EVENT_LOOP, NULL, NULL, NULL, true);
  302. worker_register_job_name(ACLK_DATABASE_NOOP, "noop");
  303. worker_register_job_name(ACLK_DATABASE_CLEANUP, "cleanup");
  304. worker_register_job_name(ACLK_DATABASE_DELETE_HOST, "node delete");
  305. worker_register_job_name(ACLK_DATABASE_NODE_STATE, "node state");
  306. worker_register_job_name(ACLK_DATABASE_PUSH_ALERT, "alert push");
  307. worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_CONFIG, "alert conf push");
  308. worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_CHECKPOINT,"alert checkpoint");
  309. worker_register_job_name(ACLK_DATABASE_PUSH_ALERT_SNAPSHOT, "alert snapshot");
  310. worker_register_job_name(ACLK_DATABASE_QUEUE_REMOVED_ALERTS, "alerts check");
  311. worker_register_job_name(ACLK_DATABASE_TIMER, "timer");
  312. uv_loop_t *loop = &config->loop;
  313. fatal_assert(0 == uv_loop_init(loop));
  314. fatal_assert(0 == uv_async_init(loop, &config->async, async_cb));
  315. fatal_assert(0 == uv_timer_init(loop, &config->timer_req));
  316. config->timer_req.data = config;
  317. fatal_assert(0 == uv_timer_start(&config->timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
  318. info("Starting ACLK synchronization thread");
  319. config->cleanup_after = now_realtime_sec() + ACLK_DATABASE_CLEANUP_FIRST;
  320. config->initialized = true;
  321. while (likely(service_running(SERVICE_ACLKSYNC))) {
  322. enum aclk_database_opcode opcode;
  323. worker_is_idle();
  324. uv_run(loop, UV_RUN_DEFAULT);
  325. /* wait for commands */
  326. do {
  327. struct aclk_database_cmd cmd = aclk_database_deq_cmd();
  328. if (unlikely(!service_running(SERVICE_ACLKSYNC)))
  329. break;
  330. opcode = cmd.opcode;
  331. if(likely(opcode != ACLK_DATABASE_NOOP))
  332. worker_is_busy(opcode);
  333. switch (opcode) {
  334. case ACLK_DATABASE_NOOP:
  335. /* the command queue was empty, do nothing */
  336. break;
  337. // MAINTENANCE
  338. case ACLK_DATABASE_CLEANUP:
  339. // Scan all aclk_alert_ tables and cleanup as needed
  340. sql_maint_aclk_sync_database_all();
  341. sql_check_aclk_table_list();
  342. break;
  343. case ACLK_DATABASE_DELETE_HOST:
  344. sql_delete_aclk_table_list(cmd.param[0]);
  345. break;
  346. // NODE STATE
  347. case ACLK_DATABASE_NODE_STATE:;
  348. RRDHOST *host = cmd.param[0];
  349. int live = (host == localhost || host->receiver || !(rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN))) ? 1 : 0;
  350. struct aclk_sync_host_config *ahc = host->aclk_sync_host_config;
  351. if (unlikely(!ahc))
  352. sql_create_aclk_table(host, &host->host_uuid, host->node_id);
  353. aclk_host_state_update(host, live);
  354. break;
  355. // ALERTS
  356. case ACLK_DATABASE_PUSH_ALERT_CONFIG:
  357. aclk_push_alert_config_event(cmd.param[0], cmd.param[1]);
  358. break;
  359. case ACLK_DATABASE_PUSH_ALERT:
  360. aclk_push_alert_events_for_all_hosts();
  361. break;
  362. case ACLK_DATABASE_PUSH_ALERT_SNAPSHOT:;
  363. aclk_push_alert_snapshot_event(cmd.param[0]);
  364. break;
  365. case ACLK_DATABASE_QUEUE_REMOVED_ALERTS:
  366. sql_process_queue_removed_alerts_to_aclk(cmd.param[0]);
  367. break;
  368. default:
  369. debug(D_ACLK_SYNC, "%s: default.", __func__);
  370. break;
  371. }
  372. if (cmd.completion)
  373. completion_mark_complete(cmd.completion);
  374. } while (opcode != ACLK_DATABASE_NOOP);
  375. }
  376. if (!uv_timer_stop(&config->timer_req))
  377. uv_close((uv_handle_t *)&config->timer_req, NULL);
  378. uv_close((uv_handle_t *)&config->async, NULL);
  379. // uv_close((uv_handle_t *)&config->async_exit, NULL);
  380. uv_cond_destroy(&config->cmd_cond);
  381. (void) uv_loop_close(loop);
  382. worker_unregister();
  383. service_exits();
  384. info("ACLK SYNC: Shutting down ACLK synchronization event loop");
  385. }
  386. static void aclk_synchronization_init(void)
  387. {
  388. aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.tail = 0;
  389. aclk_sync_config.queue_size = 0;
  390. fatal_assert(0 == uv_cond_init(&aclk_sync_config.cmd_cond));
  391. fatal_assert(0 == uv_mutex_init(&aclk_sync_config.cmd_mutex));
  392. fatal_assert(0 == uv_thread_create(&aclk_sync_config.thread, aclk_synchronization, &aclk_sync_config));
  393. }
  394. #endif
  395. // -------------------------------------------------------------
  396. void sql_create_aclk_table(RRDHOST *host __maybe_unused, uuid_t *host_uuid __maybe_unused, uuid_t *node_id __maybe_unused)
  397. {
  398. #ifdef ENABLE_ACLK
  399. char uuid_str[GUID_LEN + 1];
  400. char host_guid[GUID_LEN + 1];
  401. int rc;
  402. uuid_unparse_lower_fix(host_uuid, uuid_str);
  403. uuid_unparse_lower(*host_uuid, host_guid);
  404. char sql[ACLK_SYNC_QUERY_SIZE];
  405. snprintfz(sql, ACLK_SYNC_QUERY_SIZE-1, TABLE_ACLK_ALERT, uuid_str);
  406. rc = db_execute(db_meta, sql);
  407. if (unlikely(rc))
  408. error_report("Failed to create ACLK alert table for host %s", host ? rrdhost_hostname(host) : host_guid);
  409. else {
  410. snprintfz(sql, ACLK_SYNC_QUERY_SIZE -1, INDEX_ACLK_ALERT, uuid_str, uuid_str);
  411. rc = db_execute(db_meta, sql);
  412. if (unlikely(rc))
  413. error_report("Failed to create ACLK alert table index for host %s", host ? string2str(host->hostname) : host_guid);
  414. }
  415. if (likely(host) && unlikely(host->aclk_sync_host_config))
  416. return;
  417. if (unlikely(!host))
  418. return;
  419. struct aclk_sync_host_config *wc = callocz(1, sizeof(struct aclk_sync_host_config));
  420. if (node_id && !uuid_is_null(*node_id))
  421. uuid_unparse_lower(*node_id, wc->node_id);
  422. host->aclk_sync_host_config = (void *)wc;
  423. if (node_id && !host->node_id) {
  424. host->node_id = mallocz(sizeof(*host->node_id));
  425. uuid_copy(*host->node_id, *node_id);
  426. }
  427. wc->host = host;
  428. strcpy(wc->uuid_str, uuid_str);
  429. wc->alert_updates = 0;
  430. time_t now = now_realtime_sec();
  431. wc->node_info_send_time = (host == localhost || NULL == localhost) ? now - 25 : now;
  432. #endif
  433. }
  434. #define SQL_FETCH_ALL_HOSTS "SELECT host_id, hostname, registry_hostname, update_every, os, " \
  435. "timezone, tags, hops, memory_mode, abbrev_timezone, utc_offset, program_name, " \
  436. "program_version, entries, health_enabled FROM host WHERE hops >0;"
  437. #define SQL_FETCH_ALL_INSTANCES "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni " \
  438. "WHERE h.host_id = ni.host_id AND ni.node_id IS NOT NULL; "
  439. void sql_aclk_sync_init(void)
  440. {
  441. char *err_msg = NULL;
  442. int rc;
  443. if (unlikely(!db_meta)) {
  444. if (default_rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE) {
  445. return;
  446. }
  447. error_report("Database has not been initialized");
  448. return;
  449. }
  450. info("Creating archived hosts");
  451. int number_of_children = 0;
  452. rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_HOSTS, create_host_callback, &number_of_children, &err_msg);
  453. if (rc != SQLITE_OK) {
  454. error_report("SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg);
  455. sqlite3_free(err_msg);
  456. }
  457. info("Created %d archived hosts", number_of_children);
  458. // Trigger host context load for hosts that have been created
  459. metadata_queue_load_host_context(NULL);
  460. #ifdef ENABLE_ACLK
  461. if (!number_of_children)
  462. aclk_queue_node_info(localhost, true);
  463. rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_INSTANCES,aclk_config_parameters, NULL,&err_msg);
  464. if (rc != SQLITE_OK) {
  465. error_report("SQLite error when configuring host ACLK synchonization parameters, rc = %d (%s)", rc, err_msg);
  466. sqlite3_free(err_msg);
  467. }
  468. aclk_synchronization_init();
  469. info("ACLK sync initialization completed");
  470. #endif
  471. }
  472. // Public
  473. static inline void queue_aclk_sync_cmd(enum aclk_database_opcode opcode, const void *param0, const void *param1)
  474. {
  475. struct aclk_database_cmd cmd;
  476. cmd.opcode = opcode;
  477. cmd.param[0] = (void *) param0;
  478. cmd.param[1] = (void *) param1;
  479. cmd.completion = NULL;
  480. aclk_database_enq_cmd(&cmd);
  481. }
  482. // Public
  483. void aclk_push_alert_config(const char *node_id, const char *config_hash)
  484. {
  485. if (unlikely(!aclk_sync_config.initialized))
  486. return;
  487. queue_aclk_sync_cmd(ACLK_DATABASE_PUSH_ALERT_CONFIG, strdupz(node_id), strdupz(config_hash));
  488. }
  489. void aclk_push_node_alert_snapshot(const char *node_id)
  490. {
  491. if (unlikely(!aclk_sync_config.initialized))
  492. return;
  493. queue_aclk_sync_cmd(ACLK_DATABASE_PUSH_ALERT_SNAPSHOT, strdupz(node_id), NULL);
  494. }
  495. void aclk_push_node_removed_alerts(const char *node_id)
  496. {
  497. if (unlikely(!aclk_sync_config.initialized))
  498. return;
  499. queue_aclk_sync_cmd(ACLK_DATABASE_QUEUE_REMOVED_ALERTS, strdupz(node_id), NULL);
  500. }
  501. void schedule_node_info_update(RRDHOST *host __maybe_unused)
  502. {
  503. #ifdef ENABLE_ACLK
  504. if (unlikely(!host))
  505. return;
  506. struct aclk_database_cmd cmd;
  507. memset(&cmd, 0, sizeof(cmd));
  508. cmd.opcode = ACLK_DATABASE_NODE_STATE;
  509. cmd.param[0] = host;
  510. cmd.completion = NULL;
  511. aclk_database_enq_cmd(&cmd);
  512. #endif
  513. }