@cryptotaxi247 / netdata-1 / commits / 85d436943

Remove queue limit from ACLK sync event loop (#16411)

Code cleanup

Stelios Fragkakis committed Nov 21, 2023 at 12:00 UTC 85d43694355dfd31f1a57fcec68adf27327d5868
9 files changed +241 -293
aclk/aclk.c
+1 -1
@@ -1322,7 +1322,7 @@ void add_aclk_host_labels(void) {
1322
1323 void aclk_queue_node_info(RRDHOST *host, bool immediate)
1324 {
1325 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) host->aclk_sync_host_config;
1325 + struct aclk_sync_cfg_t *wc = host->aclk_config;
1326 if (likely(wc))
1327 wc->node_info_send_time = (host == localhost || immediate) ? 1 : now_realtime_sec();
1328 }
database/rrd.h
+1 -1
@@ -1273,7 +1273,7 @@ struct rrdhost {
1273 struct sender_state *sender;
1274 netdata_thread_t rrdpush_sender_thread; // the sender thread
1275 size_t rrdpush_sender_replicating_charts; // the number of charts currently being replicated to a parent
1276 - void *aclk_sync_host_config;
1276 + struct aclk_sync_cfg_t *aclk_config;
1277
1278 uint32_t rrdpush_receiver_connection_counter; // the number of times this receiver has connected
1279 uint32_t rrdpush_sender_connection_counter; // the number of times this sender has connected
database/sqlite/sqlite_aclk.c
+43 -91
@@ -11,60 +11,46 @@ struct aclk_sync_config_s {
11 uv_timer_t timer_req;
12 time_t cleanup_after; // Start a cleanup after this timestamp
13 uv_async_t async;
14 - /* FIFO command queue */
15 - uv_mutex_t cmd_mutex;
16 - uv_cond_t cmd_cond;
14 bool initialized;
18 - volatile unsigned queue_size;
19 - struct aclk_database_cmdqueue cmd_queue;
15 + SPINLOCK cmd_queue_lock;
16 + struct aclk_database_cmd *cmd_base;
17 } aclk_sync_config = { 0 };
18
22 -
19 void sanity_check(void) {
20 // make sure the compiler will stop on misconfigurations
21 BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
22 }
23
28 -
29 -int aclk_database_enq_cmd_noblock(struct aclk_database_cmd *cmd)
24 +static struct aclk_database_cmd aclk_database_deq_cmd(void)
25 {
31 - unsigned queue_size;
26 + struct aclk_database_cmd ret;
27
33 - /* wait for free space in queue */
34 - uv_mutex_lock(&aclk_sync_config.cmd_mutex);
35 - if ((queue_size = aclk_sync_config.queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
36 - uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
37 - return 1;
28 + spinlock_lock(&aclk_sync_config.cmd_queue_lock);
29 + if(aclk_sync_config.cmd_base) {
30 + struct aclk_database_cmd *t = aclk_sync_config.cmd_base;
31 + DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(aclk_sync_config.cmd_base, t, prev, next);
32 + ret = *t;
33 + freez(t);
34 + }
35 + else {
36 + ret.opcode = ACLK_DATABASE_NOOP;
37 + ret.completion = NULL;
38 }
39 + spinlock_unlock(&aclk_sync_config.cmd_queue_lock);
40
40 - fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
41 - /* enqueue command */
42 - aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.tail] = *cmd;
43 - aclk_sync_config.cmd_queue.tail = aclk_sync_config.cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
44 - aclk_sync_config.cmd_queue.tail + 1 : 0;
45 - aclk_sync_config.queue_size = queue_size + 1;
46 - uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
47 - return 0;
41 + return ret;
42 }
43
44 static void aclk_database_enq_cmd(struct aclk_database_cmd *cmd)
45 {
52 - unsigned queue_size;
46 + struct aclk_database_cmd *t = mallocz(sizeof(*t));
47 + *t = *cmd;
48 + t->prev = t->next = NULL;
49 +
50 + spinlock_lock(&aclk_sync_config.cmd_queue_lock);
51 + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(aclk_sync_config.cmd_base, t, prev, next);
52 + spinlock_unlock(&aclk_sync_config.cmd_queue_lock);
53
54 - /* wait for free space in queue */
55 - uv_mutex_lock(&aclk_sync_config.cmd_mutex);
56 - while ((queue_size = aclk_sync_config.queue_size) == ACLK_DATABASE_CMD_Q_MAX_SIZE) {
57 - uv_cond_wait(&aclk_sync_config.cmd_cond, &aclk_sync_config.cmd_mutex);
58 - }
59 - fatal_assert(queue_size < ACLK_DATABASE_CMD_Q_MAX_SIZE);
60 - /* enqueue command */
61 - aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.tail] = *cmd;
62 - aclk_sync_config.cmd_queue.tail = aclk_sync_config.cmd_queue.tail != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
63 - aclk_sync_config.cmd_queue.tail + 1 : 0;
64 - aclk_sync_config.queue_size = queue_size + 1;
65 - uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
66 -
67 - /* wake up event loop */
54 (void) uv_async_send(&aclk_sync_config.async);
55 }
56
@@ -145,35 +131,6 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
131 }
132
133 #ifdef ENABLE_ACLK
148 -static struct aclk_database_cmd aclk_database_deq_cmd(void)
149 -{
150 - struct aclk_database_cmd ret;
151 - unsigned queue_size;
152 -
153 - uv_mutex_lock(&aclk_sync_config.cmd_mutex);
154 - queue_size = aclk_sync_config.queue_size;
155 - if (queue_size == 0) {
156 - memset(&ret, 0, sizeof(ret));
157 - ret.opcode = ACLK_DATABASE_NOOP;
158 - ret.completion = NULL;
159 -
160 - } else {
161 - /* dequeue command */
162 - ret = aclk_sync_config.cmd_queue.cmd_array[aclk_sync_config.cmd_queue.head];
163 - if (queue_size == 1) {
164 - aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.tail = 0;
165 - } else {
166 - aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.head != ACLK_DATABASE_CMD_Q_MAX_SIZE - 1 ?
167 - aclk_sync_config.cmd_queue.head + 1 : 0;
168 - }
169 - aclk_sync_config.queue_size = queue_size - 1;
170 - /* wake up producers */
171 - uv_cond_signal(&aclk_sync_config.cmd_cond);
172 - }
173 - uv_mutex_unlock(&aclk_sync_config.cmd_mutex);
174 -
175 - return ret;
176 -}
134
135 #define SQL_SELECT_HOST_BY_UUID "SELECT host_id FROM host WHERE host_id = @host_id;"
136 static int is_host_available(uuid_t *host_id)
@@ -264,7 +221,7 @@ static int sql_check_aclk_table(void *data __maybe_unused, int argc __maybe_unus
221 memset(&cmd, 0, sizeof(cmd));
222 cmd.opcode = ACLK_DATABASE_DELETE_HOST;
223 cmd.param[0] = strdupz((char *) argv[0]);
267 - aclk_database_enq_cmd_noblock(&cmd);
224 + aclk_database_enq_cmd(&cmd);
225 return 0;
226 }
227
@@ -292,7 +249,6 @@ static int sql_maint_aclk_sync_database(void *data __maybe_unused, int argc __ma
249 return 0;
250 }
251
295 -
252 #define SQL_SELECT_ACLK_ALERT_LIST "SELECT SUBSTR(name,12) FROM sqlite_schema WHERE name LIKE 'aclk_alert_%' AND type IN ('table');"
253
254 static void sql_maint_aclk_sync_database_all(void)
@@ -307,7 +263,7 @@ static void sql_maint_aclk_sync_database_all(void)
263
264 static int aclk_config_parameters(void *data __maybe_unused, int argc __maybe_unused, char **argv, char **column __maybe_unused)
265 {
310 - char uuid_str[GUID_LEN + 1];
266 + char uuid_str[UUID_STR_LEN];
267 uuid_unparse_lower(*((uuid_t *) argv[0]), uuid_str);
268
269 RRDHOST *host = rrdhost_find_by_guid(uuid_str);
@@ -337,16 +293,15 @@ static void timer_cb(uv_timer_t *handle)
293
294 time_t now = now_realtime_sec();
295
340 - if (config->cleanup_after && config->cleanup_after < now) {
296 + if (config->cleanup_after < now) {
297 cmd.opcode = ACLK_DATABASE_CLEANUP;
342 - if (!aclk_database_enq_cmd_noblock(&cmd))
343 - config->cleanup_after += ACLK_DATABASE_CLEANUP_INTERVAL;
298 + aclk_database_enq_cmd(&cmd);
299 + config->cleanup_after += ACLK_DATABASE_CLEANUP_INTERVAL;
300 }
301
302 if (aclk_connected) {
303 cmd.opcode = ACLK_DATABASE_PUSH_ALERT;
348 - aclk_database_enq_cmd_noblock(&cmd);
349 -
304 + aclk_database_enq_cmd(&cmd);
305 aclk_check_node_info_and_collectors();
306 }
307 }
@@ -417,7 +372,7 @@ static void aclk_synchronization(void *arg __maybe_unused)
372 case ACLK_DATABASE_NODE_STATE:;
373 RRDHOST *host = cmd.param[0];
374 int live = (host == localhost || host->receiver || !(rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN))) ? 1 : 0;
420 - struct aclk_sync_host_config *ahc = host->aclk_sync_host_config;
375 + struct aclk_sync_cfg_t *ahc = host->aclk_config;
376 if (unlikely(!ahc))
377 sql_create_aclk_table(host, &host->host_uuid, host->node_id);
378 aclk_host_state_update(host, live);
@@ -447,8 +402,6 @@ static void aclk_synchronization(void *arg __maybe_unused)
402 uv_close((uv_handle_t *)&config->timer_req, NULL);
403
404 uv_close((uv_handle_t *)&config->async, NULL);
450 -// uv_close((uv_handle_t *)&config->async_exit, NULL);
451 - uv_cond_destroy(&config->cmd_cond);
405 (void) uv_loop_close(loop);
406
407 worker_unregister();
@@ -458,11 +411,7 @@ static void aclk_synchronization(void *arg __maybe_unused)
411
412 static void aclk_synchronization_init(void)
413 {
461 - aclk_sync_config.cmd_queue.head = aclk_sync_config.cmd_queue.tail = 0;
462 - aclk_sync_config.queue_size = 0;
463 - fatal_assert(0 == uv_cond_init(&aclk_sync_config.cmd_cond));
464 - fatal_assert(0 == uv_mutex_init(&aclk_sync_config.cmd_mutex));
465 -
414 + memset(&aclk_sync_config, 0, sizeof(aclk_sync_config));
415 fatal_assert(0 == uv_thread_create(&aclk_sync_config.thread, aclk_synchronization, &aclk_sync_config));
416 }
417 #endif
@@ -472,8 +421,8 @@ static void aclk_synchronization_init(void)
421 void sql_create_aclk_table(RRDHOST *host __maybe_unused, uuid_t *host_uuid __maybe_unused, uuid_t *node_id __maybe_unused)
422 {
423 #ifdef ENABLE_ACLK
475 - char uuid_str[GUID_LEN + 1];
476 - char host_guid[GUID_LEN + 1];
424 + char uuid_str[UUID_STR_LEN];
425 + char host_guid[UUID_STR_LEN];
426 int rc;
427
428 uuid_unparse_lower_fix(host_uuid, uuid_str);
@@ -496,17 +445,17 @@ void sql_create_aclk_table(RRDHOST *host __maybe_unused, uuid_t *host_uuid __may
445 if (unlikely(rc))
446 error_report("Failed to create ACLK alert table index 2 for host %s", host ? string2str(host->hostname) : host_guid);
447 }
499 - if (likely(host) && unlikely(host->aclk_sync_host_config))
448 + if (likely(host) && unlikely(host->aclk_config))
449 return;
450
451 if (unlikely(!host))
452 return;
453
505 - struct aclk_sync_host_config *wc = callocz(1, sizeof(struct aclk_sync_host_config));
454 + struct aclk_sync_cfg_t *wc = callocz(1, sizeof(struct aclk_sync_cfg_t));
455 if (node_id && !uuid_is_null(*node_id))
456 uuid_unparse_lower(*node_id, wc->node_id);
457
509 - host->aclk_sync_host_config = (void *)wc;
458 + host->aclk_config = wc;
459 if (node_id && !host->node_id) {
460 host->node_id = mallocz(sizeof(*host->node_id));
461 uuid_copy(*host->node_id, *node_id);
@@ -520,12 +469,15 @@ void sql_create_aclk_table(RRDHOST *host __maybe_unused, uuid_t *host_uuid __may
469 #endif
470 }
471
523 -#define SQL_FETCH_ALL_HOSTS "SELECT host_id, hostname, registry_hostname, update_every, os, " \
524 - "timezone, tags, hops, memory_mode, abbrev_timezone, utc_offset, program_name, " \
472 +#define SQL_FETCH_ALL_HOSTS \
473 + "SELECT host_id, hostname, registry_hostname, update_every, os, " \
474 + "timezone, tags, hops, memory_mode, abbrev_timezone, utc_offset, program_name, " \
475 "program_version, entries, health_enabled, last_connected FROM host WHERE hops >0;"
476
527 -#define SQL_FETCH_ALL_INSTANCES "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni " \
528 - "WHERE h.host_id = ni.host_id AND ni.node_id IS NOT NULL; "
477 +#define SQL_FETCH_ALL_INSTANCES \
478 + "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni " \
479 + "WHERE h.host_id = ni.host_id AND ni.node_id IS NOT NULL; "
480 +
481 void sql_aclk_sync_init(void)
482 {
483 char *err_msg = NULL;
database/sqlite/sqlite_aclk.h
+7 -18
@@ -5,7 +5,6 @@
5
6 #include "sqlite3.h"
7
8 -
8 #ifndef ACLK_MAX_CHART_BATCH
9 #define ACLK_MAX_CHART_BATCH (200)
10 #endif
@@ -41,11 +40,11 @@ static inline int claimed()
40 return localhost->aclk_state.claimed_id != NULL;
41 }
42
44 -#define TABLE_ACLK_ALERT "CREATE TABLE IF NOT EXISTS aclk_alert_%s (sequence_id INTEGER PRIMARY KEY, " \
45 - "alert_unique_id, date_created, date_submitted, date_cloud_ack, filtered_alert_unique_id NOT NULL, " \
46 - "unique(alert_unique_id));"
43 +#define TABLE_ACLK_ALERT \
44 + "CREATE TABLE IF NOT EXISTS aclk_alert_%s (sequence_id INTEGER PRIMARY KEY, " \
45 + "alert_unique_id, date_created, date_submitted, date_cloud_ack, filtered_alert_unique_id NOT NULL, " \
46 + "UNIQUE(alert_unique_id));"
47
48 -#define INDEX_ACLK_ALERT "CREATE INDEX IF NOT EXISTS aclk_alert_index_%s ON aclk_alert_%s (alert_unique_id);"
48 #define INDEX_ACLK_ALERT1 "CREATE INDEX IF NOT EXISTS aclk_alert_index1_%s ON aclk_alert_%s (filtered_alert_unique_id);"
49 #define INDEX_ACLK_ALERT2 "CREATE INDEX IF NOT EXISTS aclk_alert_index2_%s ON aclk_alert_%s (date_submitted);"
50
@@ -71,16 +70,10 @@ struct aclk_database_cmd {
70 enum aclk_database_opcode opcode;
71 void *param[2];
72 struct completion *completion;
73 + struct aclk_database_cmd *prev, *next;
74 };
75
76 -#define ACLK_DATABASE_CMD_Q_MAX_SIZE (1024)
77 -
78 -struct aclk_database_cmdqueue {
79 - unsigned head, tail;
80 - struct aclk_database_cmd cmd_array[ACLK_DATABASE_CMD_Q_MAX_SIZE];
81 -};
82 -
83 -struct aclk_sync_host_config {
76 +typedef struct aclk_sync_cfg_t {
77 RRDHOST *host;
78 int alert_updates;
79 int alert_checkpoint_req;
@@ -92,16 +85,12 @@ struct aclk_sync_host_config {
85 char *alerts_snapshot_uuid; // will contain the snapshot_uuid value if snapshot was requested
86 uint64_t alerts_log_first_sequence_id;
87 uint64_t alerts_log_last_sequence_id;
95 -};
96 -
97 -extern sqlite3 *db_meta;
88 +} aclk_sync_cfg_t;
89
99 -int aclk_database_enq_cmd_noblock(struct aclk_database_cmd *cmd);
90 void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id);
91 void sql_aclk_sync_init(void);
92 void aclk_push_alert_config(const char *node_id, const char *config_hash);
93 void aclk_push_node_alert_snapshot(const char *node_id);
104 -void aclk_push_node_health_log(const char *node_id);
94 void aclk_push_node_removed_alerts(const char *node_id);
95 void schedule_node_info_update(RRDHOST *host);
96
database/sqlite/sqlite_aclk_alert.c
+179 -171
@@ -13,16 +13,42 @@
13 sqlite3_column_bytes((res), (_param)) ? strdupz((char *)sqlite3_column_text((res), (_param))) : NULL; \
14 })
15
16 -
16 #define SQL_UPDATE_FILTERED_ALERT \
18 - "UPDATE aclk_alert_%s SET filtered_alert_unique_id = %u, date_created = unixepoch() where filtered_alert_unique_id = %u"
17 + "UPDATE aclk_alert_%s SET filtered_alert_unique_id = @new_alert, date_created = UNIXEPOCH() " \
18 + "WHERE filtered_alert_unique_id = @old_alert"
19
20 -static void update_filtered(ALARM_ENTRY *ae, uint32_t unique_id, char *uuid_str)
20 +static void update_filtered(ALARM_ENTRY *ae, int64_t unique_id, char *uuid_str)
21 {
22 + sqlite3_stmt *res = NULL;
23 +
24 char sql[ACLK_SYNC_QUERY_SIZE];
23 - snprintfz(sql, ACLK_SYNC_QUERY_SIZE-1, SQL_UPDATE_FILTERED_ALERT, uuid_str, ae->unique_id, unique_id);
24 - sqlite3_exec_monitored(db_meta, sql, 0, 0, NULL);
25 - ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
25 + snprintfz(sql, ACLK_SYNC_QUERY_SIZE-1, SQL_UPDATE_FILTERED_ALERT, uuid_str);
26 + int rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
27 + if (rc != SQLITE_OK) {
28 + error_report("Failed to prepare statement when trying to check for alert variables.");
29 + return;
30 + }
31 +
32 + rc = sqlite3_bind_int64(res, 1, ae->unique_id);
33 + if (unlikely(rc != SQLITE_OK)) {
34 + error_report("Failed to bind ae unique_id for update_filtered");
35 + goto done;
36 + }
37 +
38 + rc = sqlite3_bind_int64(res, 2, unique_id);
39 + if (unlikely(rc != SQLITE_OK)) {
40 + error_report("Failed to bind unique_id for update_filtered");
41 + goto done;
42 + }
43 +
44 + rc = sqlite3_step_monitored(res);
45 + if (likely(rc == SQLITE_DONE))
46 + ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
47 +
48 +done:
49 + rc = sqlite3_finalize(res);
50 + if (unlikely(rc != SQLITE_OK))
51 + error_report("Failed to finalize statement when trying to update_filtered, rc = %d", rc);
52 }
53
54 #define SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID \
@@ -30,35 +56,35 @@ static void update_filtered(ALARM_ENTRY *ae, uint32_t unique_id, char *uuid_str)
56 "WHERE hld.unique_id = @unique_id AND hl.config_hash_id = ah.hash_id AND hld.health_log_id = hl.health_log_id " \
57 "AND hl.host_id = @host_id AND ah.warn IS NULL AND ah.crit IS NULL"
58
33 -static inline bool is_event_from_alert_variable_config(uint32_t unique_id, uuid_t *host_id)
59 +static inline bool is_event_from_alert_variable_config(int64_t unique_id, uuid_t *host_id)
60 {
61 sqlite3_stmt *res = NULL;
36 - int rc = 0;
37 - bool ret = false;
62
39 - rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, -1, &res, 0);
63 + int rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, -1, &res, 0);
64 if (rc != SQLITE_OK) {
65 error_report("Failed to prepare statement when trying to check for alert variables.");
66 return false;
67 }
68
45 - rc = sqlite3_bind_int(res, 1, (int) unique_id);
69 + bool ret = false;
70 +
71 + rc = sqlite3_bind_int64(res, 1, unique_id);
72 if (unlikely(rc != SQLITE_OK)) {
73 error_report("Failed to bind unique_id for checking alert variable.");
48 - goto fail;
74 + goto done;
75 }
76
77 rc = sqlite3_bind_blob(res, 2, host_id, sizeof(*host_id), SQLITE_STATIC);
78 if (unlikely(rc != SQLITE_OK)) {
79 error_report("Failed to bind host_id for checking alert variable.");
54 - goto fail;
80 + goto done;
81 }
82
83 rc = sqlite3_step_monitored(res);
84 if (likely(rc == SQLITE_ROW))
85 ret = true;
86
61 -fail:
87 +done:
88 rc = sqlite3_finalize(res);
89 if (unlikely(rc != SQLITE_OK))
90 error_report("Failed to finalize statement when trying to check for alert variables, rc = %d", rc);
@@ -78,25 +104,18 @@ fail:
104 static bool should_send_to_cloud(RRDHOST *host, ALARM_ENTRY *ae)
105 {
106 sqlite3_stmt *res = NULL;
81 - char uuid_str[UUID_STR_LEN];
82 - uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
83 -
84 - bool send = false;
107
108 if (ae->new_status == RRDCALC_STATUS_REMOVED || ae->new_status == RRDCALC_STATUS_UNINITIALIZED)
109 return 0;
110
89 - if (unlikely(uuid_is_null(ae->config_hash_id)))
111 + if (unlikely(uuid_is_null(ae->config_hash_id) || !host->aclk_config))
112 return 0;
113
114 char sql[ACLK_SYNC_QUERY_SIZE];
93 - uuid_t config_hash_id;
94 - RRDCALC_STATUS status;
95 - uint32_t unique_id;
115
116 //get the previous sent event of this alarm_id
117 //base the search on the last filtered event
99 - snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_SELECT_ALERT_BY_ID, uuid_str);
118 + snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_SELECT_ALERT_BY_ID, host->aclk_config->uuid_str);
119
120 int rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
121 if (rc != SQLITE_OK) {
@@ -104,6 +123,8 @@ static bool should_send_to_cloud(RRDHOST *host, ALARM_ENTRY *ae)
123 return true;
124 }
125
126 + bool send = false;
127 +
128 rc = sqlite3_bind_blob(res, 1, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);
129 if (unlikely(rc != SQLITE_OK)) {
130 error_report("Failed to bind host_id for checking should_send_to_cloud");
@@ -119,17 +140,18 @@ static bool should_send_to_cloud(RRDHOST *host, ALARM_ENTRY *ae)
140 rc = sqlite3_step_monitored(res);
141
142 if (likely(rc == SQLITE_ROW)) {
122 - status = (RRDCALC_STATUS)sqlite3_column_int(res, 0);
143 + uuid_t config_hash_id;
144 + RRDCALC_STATUS status = (RRDCALC_STATUS)sqlite3_column_int(res, 0);
145
146 if (sqlite3_column_type(res, 1) != SQLITE_NULL)
147 uuid_copy(config_hash_id, *((uuid_t *)sqlite3_column_blob(res, 1)));
148
127 - unique_id = (uint32_t)sqlite3_column_int64(res, 2);
149 + int64_t unique_id = sqlite3_column_int64(res, 2);
150
151 if (ae->new_status != (RRDCALC_STATUS)status || uuid_memcmp(&ae->config_hash_id, &config_hash_id))
152 send = true;
153 else
132 - update_filtered(ae, unique_id, uuid_str);
154 + update_filtered(ae, unique_id, host->aclk_config->uuid_str);
155 } else
156 send = true;
157
@@ -149,7 +171,6 @@ void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, bool skip_filter)
171 {
172 sqlite3_stmt *res_alert = NULL;
173 char sql[ACLK_SYNC_QUERY_SIZE];
152 - char uuid_str[UUID_STR_LEN];
174
175 if (!service_running(SERVICE_ACLK))
176 return;
@@ -163,8 +184,7 @@ void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, bool skip_filter)
184 if (is_event_from_alert_variable_config(ae->unique_id, &host->host_uuid))
185 return;
186
166 - uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
167 - snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_QUEUE_ALERT_TO_CLOUD, uuid_str);
187 + snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_QUEUE_ALERT_TO_CLOUD, host->aclk_config->uuid_str);
188
189 int rc = sqlite3_prepare_v2(db_meta, sql, -1, &res_alert, 0);
190 if (unlikely(rc != SQLITE_OK)) {
@@ -172,18 +192,18 @@ void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, bool skip_filter)
192 return;
193 }
194
175 - rc = sqlite3_bind_int(res_alert, 1, (int) ae->unique_id);
195 + rc = sqlite3_bind_int64(res_alert, 1, ae->unique_id);
196 if (unlikely(rc != SQLITE_OK))
177 - goto bind_fail;
197 + goto done;
198
199 rc = execute_insert(res_alert);
200 if (unlikely(rc == SQLITE_DONE)) {
201 ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
202 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
203 } else
184 - error_report("Failed to store alert event %u, rc = %d", ae->unique_id, rc);
204 + error_report("Failed to store alert event %"PRId64", rc = %d", ae->unique_id, rc);
205
186 -bind_fail:
206 +done:
207 if (unlikely(sqlite3_finalize(res_alert) != SQLITE_OK))
208 error_report("Failed to reset statement in store alert event, rc = %d", rc);
209 }
@@ -239,7 +259,7 @@ static inline char *sqlite3_text_strdupz_empty(sqlite3_stmt *res, int iCol) {
259 }
260
261
242 -void aclk_push_alert_event(struct aclk_sync_host_config *wc)
262 +void aclk_push_alert_event(struct aclk_sync_cfg_t *wc)
263 {
264 #ifndef ENABLE_ACLK
265 UNUSED(wc);
@@ -388,9 +408,13 @@ void aclk_push_alert_event(struct aclk_sync_host_config *wc)
408
409 if (first_sequence_id) {
410 buffer_flush(sql);
391 - buffer_sprintf(sql, "UPDATE aclk_alert_%s SET date_submitted=unixepoch() "
392 - "WHERE +date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",
393 - wc->uuid_str, first_sequence_id, last_sequence_id);
411 + buffer_sprintf(
412 + sql,
413 + "UPDATE aclk_alert_%s SET date_submitted=unixepoch() "
414 + "WHERE +date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",
415 + wc->uuid_str,
416 + first_sequence_id,
417 + last_sequence_id);
418
419 if (unlikely(db_execute(db_meta, buffer_tostring(sql))))
420 error_report("Failed to mark ACLK alert entries as submitted for host %s", rrdhost_hostname(wc->host));
@@ -430,7 +454,7 @@ void aclk_push_alert_events_for_all_hosts(void)
454
455 rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
456
433 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
457 + struct aclk_sync_cfg_t *wc = host->aclk_config;
458 if (likely(wc))
459 aclk_push_alert_event(wc);
460 }
@@ -439,59 +463,54 @@ void aclk_push_alert_events_for_all_hosts(void)
463
464 void sql_queue_existing_alerts_to_aclk(RRDHOST *host)
465 {
442 - char uuid_str[UUID_STR_LEN];
443 - uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
444 - BUFFER *sql = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
466 sqlite3_stmt *res = NULL;
467 int rc;
468
469 + struct aclk_sync_cfg_t *wc = host->aclk_config;
470 +
471 + BUFFER *sql = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
472 +
473 rw_spinlock_write_lock(&host->health_log.spinlock);
474
450 - buffer_sprintf(sql, "delete from aclk_alert_%s; ", uuid_str);
451 - if (unlikely(db_execute(db_meta, buffer_tostring(sql)))) {
452 - rw_spinlock_write_unlock(&host->health_log.spinlock);
453 - buffer_free(sql);
454 - return;
455 - }
475 + buffer_sprintf(sql, "DELETE FROM aclk_alert_%s", wc->uuid_str);
476 + if (unlikely(db_execute(db_meta, buffer_tostring(sql))))
477 + goto skip;
478
479 buffer_flush(sql);
480 +
481 buffer_sprintf(
482 sql,
483 "insert into aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) "
484 "select hld.unique_id alert_unique_id, unixepoch(), hld.unique_id alert_unique_id from health_log_detail hld, health_log hl "
485 "where hld.new_status <> 0 and hld.new_status <> -2 and hl.health_log_id = hld.health_log_id and hl.config_hash_id is not null "
486 "and hld.updated_by_id = 0 and hl.host_id = @host_id order by hld.unique_id asc on conflict (alert_unique_id) do nothing;",
464 - uuid_str);
487 + wc->uuid_str);
488
489 rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
490 if (rc != SQLITE_OK) {
491 error_report("Failed to prepare statement when trying to queue existing alerts.");
469 - rw_spinlock_write_unlock(&host->health_log.spinlock);
470 - buffer_free(sql);
471 - return;
492 + goto skip;
493 }
494
495 rc = sqlite3_bind_blob(res, 1, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);
496 if (unlikely(rc != SQLITE_OK)) {
497 error_report("Failed to bind host_id for when trying to queue existing alerts.");
477 - sqlite3_finalize(res);
478 - rw_spinlock_write_unlock(&host->health_log.spinlock);
479 - buffer_free(sql);
480 - return;
498 + goto done;
499 }
500
501 rc = execute_insert(res);
502 if (unlikely(rc != SQLITE_DONE))
503 error_report("Failed to queue existing alerts, rc = %d", rc);
486 -
504 + else
505 + rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
506 +done:
507 rc = sqlite3_finalize(res);
508 if (unlikely(rc != SQLITE_OK))
509 error_report("Failed to finalize statement to queue existing alerts, rc = %d", rc);
510
511 +skip:
512 rw_spinlock_write_unlock(&host->health_log.spinlock);
492 -
513 buffer_free(sql);
494 - rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
514 }
515
516 void aclk_send_alarm_configuration(char *config_hash)
@@ -499,16 +518,13 @@ void aclk_send_alarm_configuration(char *config_hash)
518 if (unlikely(!config_hash))
519 return;
520
502 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) localhost->aclk_sync_host_config;
521 + struct aclk_sync_cfg_t *wc = localhost->aclk_config;
522
523 if (unlikely(!wc))
524 return;
525
526 netdata_log_access(
508 - "ACLK REQ [%s (%s)]: Request to send alert config %s.",
509 - wc->node_id,
510 - wc->host ? rrdhost_hostname(wc->host) : "N/A",
511 - config_hash);
527 + "ACLK REQ [%s (%s)]: Request to send alert config %s.", wc->node_id, rrdhost_hostname(wc->host), config_hash);
528
529 aclk_push_alert_config(wc->node_id, config_hash);
530 }
@@ -521,25 +537,18 @@ void aclk_send_alarm_configuration(char *config_hash)
537
538 int aclk_push_alert_config_event(char *node_id __maybe_unused, char *config_hash __maybe_unused)
539 {
524 - int rc = 0;
540 + int rc;
541
542 #ifdef ENABLE_ACLK
543
544 CHECK_SQLITE_CONNECTION(db_meta);
545
546 sqlite3_stmt *res = NULL;
547 + struct aclk_sync_cfg_t *wc;
548
532 - struct aclk_sync_host_config *wc = NULL;
549 RRDHOST *host = find_host_by_node_id(node_id);
550
535 - if (unlikely(!host)) {
536 - freez(config_hash);
537 - freez(node_id);
538 - return 1;
539 - }
540 -
541 - wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
542 - if (unlikely(!wc)) {
551 + if (unlikely(!host || !(wc = host->aclk_config))) {
552 freez(config_hash);
553 freez(node_id);
554 return 1;
@@ -653,45 +662,48 @@ void aclk_start_alert_streaming(char *node_id, bool resets)
662 if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))
663 return;
664
656 - RRDHOST *host = find_host_by_node_id(node_id);
657 -
658 - if (unlikely(!host))
659 - return;
660 -
661 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
665 + struct aclk_sync_cfg_t *wc;
666
663 - if (unlikely(!wc))
667 + RRDHOST *host = find_host_by_node_id(node_id);
668 + if (unlikely(!host || !(wc = host->aclk_config)))
669 return;
670
671 if (unlikely(!host->health.health_enabled)) {
667 - netdata_log_access("ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, health is disabled.", node_id);
672 + netdata_log_access(
673 + "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, health is disabled.", node_id);
674 return;
675 }
676
677 if (resets) {
672 - netdata_log_access("ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED (RESET REQUESTED)", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
678 + netdata_log_access(
679 + "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED (RESET REQUESTED)",
680 + node_id,
681 + wc->host ? rrdhost_hostname(wc->host) : "N/A");
682 sql_queue_existing_alerts_to_aclk(host);
683 } else
675 - netdata_log_access("ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
684 + netdata_log_access(
685 + "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
686
687 wc->alert_updates = 1;
688 wc->alert_queue_removed = SEND_REMOVED_AFTER_HEALTH_LOOPS;
689 }
690
681 -#define SQL_QUEUE_REMOVE_ALERTS "INSERT INTO aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) " \
691 +#define SQL_QUEUE_REMOVE_ALERTS \
692 + "INSERT INTO aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) " \
693 "SELECT hld.unique_id alert_unique_id, UNIXEPOCH(), hld.unique_id alert_unique_id FROM health_log hl, health_log_detail hld " \
683 - "WHERE hl.host_id = @host_id AND hl.health_log_id = hld.health_log_id AND hld.new_status = -2 AND hld.updated_by_id = 0 " \
684 - "AND hld.unique_id NOT IN (SELECT alert_unique_id FROM aclk_alert_%s) " \
685 - "AND hl.config_hash_id NOT IN (select hash_id from alert_hash where warn is null and crit is null) " \
686 - "AND hl.name || hl.chart NOT IN (select name || chart from health_log where name = hl.name and chart = hl.chart and alarm_id > hl.alarm_id and host_id = hl.host_id) " \
694 + "WHERE hl.host_id = @host_id AND hl.health_log_id = hld.health_log_id AND hld.new_status = -2 AND hld.updated_by_id = 0 " \
695 + "AND hld.unique_id NOT IN (SELECT alert_unique_id FROM aclk_alert_%s) " \
696 + "AND hl.config_hash_id NOT IN (select hash_id from alert_hash where warn is null and crit is null) " \
697 + "AND hl.name || hl.chart NOT IN (select name || chart from health_log where name = hl.name and " \
698 + "chart = hl.chart and alarm_id > hl.alarm_id and host_id = hl.host_id) " \
699 "ORDER BY hld.unique_id ASC ON CONFLICT (alert_unique_id) DO NOTHING;"
700 void sql_process_queue_removed_alerts_to_aclk(char *node_id)
701 {
690 - struct aclk_sync_host_config *wc;
702 + struct aclk_sync_cfg_t *wc;
703 RRDHOST *host = find_host_by_node_id(node_id);
704 freez(node_id);
705
694 - if (unlikely(!host || !(wc = host->aclk_sync_host_config)))
706 + if (unlikely(!host || !(wc = host->aclk_config)))
707 return;
708
709 char sql[ACLK_SYNC_QUERY_SIZE * 2];
@@ -708,33 +720,25 @@ void sql_process_queue_removed_alerts_to_aclk(char *node_id)
720 rc = sqlite3_bind_blob(res, 1, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);
721 if (unlikely(rc != SQLITE_OK)) {
722 error_report("Failed to bind host_id for when trying to queue remvoed alerts.");
711 - sqlite3_finalize(res);
712 - return;
723 + goto skip;
724 }
725
726 rc = execute_insert(res);
716 - if (unlikely(rc != SQLITE_DONE)) {
717 - sqlite3_finalize(res);
718 - error_report("Failed to queue removed alerts, rc = %d", rc);
719 - return;
727 + if (likely(rc == SQLITE_DONE)) {
728 + netdata_log_access("ACLK STA [%s (%s)]: QUEUED REMOVED ALERTS", wc->node_id, rrdhost_hostname(wc->host));
729 + rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
730 + wc->alert_queue_removed = 0;
731 }
732
733 +skip:
734 rc = sqlite3_finalize(res);
735 if (unlikely(rc != SQLITE_OK))
736 error_report("Failed to finalize statement to queue removed alerts, rc = %d", rc);
725 -
726 - netdata_log_access("ACLK STA [%s (%s)]: QUEUED REMOVED ALERTS", wc->node_id, rrdhost_hostname(wc->host));
727 -
728 - rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
729 - wc->alert_queue_removed = 0;
737 }
738
739 void sql_queue_removed_alerts_to_aclk(RRDHOST *host)
740 {
734 - if (unlikely(!host->aclk_sync_host_config))
735 - return;
736 -
737 - if (!claimed() || !host->node_id)
741 + if (unlikely(!host->aclk_config || !claimed() || !host->node_id))
742 return;
743
744 char node_id[UUID_STR_LEN];
@@ -746,32 +750,28 @@ void sql_queue_removed_alerts_to_aclk(RRDHOST *host)
750 void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id __maybe_unused, char *snapshot_uuid)
751 {
752 uuid_t node_uuid;
753 +
754 if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))
755 return;
756
757 + struct aclk_sync_cfg_t *wc;
758 +
759 RRDHOST *host = find_host_by_node_id(node_id);
753 - if (unlikely(!host)) {
760 + if (unlikely(!host || !(wc = host->aclk_config))) {
761 netdata_log_access("ACLK STA [%s (N/A)]: ACLK node id does not exist", node_id);
762 return;
763 }
764
758 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
759 -
760 - if (unlikely(!wc)) {
761 - netdata_log_access("ACLK STA [%s (N/A)]: ACLK node id does not exist", node_id);
762 - return;
763 - }
764 -
765 netdata_log_access(
766 - "IN [%s (%s)]: Request to send alerts snapshot, snapshot_uuid %s",
767 - node_id,
768 - wc->host ? rrdhost_hostname(wc->host) : "N/A",
769 - snapshot_uuid);
766 + "IN [%s (%s)]: Request to send alerts snapshot, snapshot_uuid %s",
767 + node_id,
768 + wc->host ? rrdhost_hostname(wc->host) : "N/A",
769 + snapshot_uuid);
770 +
771 if (wc->alerts_snapshot_uuid && !strcmp(wc->alerts_snapshot_uuid,snapshot_uuid))
772 return;
772 - __sync_synchronize();
773 +
774 wc->alerts_snapshot_uuid = strdupz(snapshot_uuid);
774 - __sync_synchronize();
775
776 aclk_push_node_alert_snapshot(node_id);
777 }
@@ -788,9 +788,7 @@ void health_alarm_entry2proto_nolock(struct alarm_log_entry *alarm_log, ALARM_EN
788 alarm_log->chart = strdupz(ae_chart_id(ae));
789 alarm_log->name = strdupz(ae_name(ae));
790
791 - alarm_log->batch_id = 0;
792 - alarm_log->sequence_id = 0;
793 - alarm_log->when = (time_t)ae->when;
791 + alarm_log->when = ae->when;
792
793 alarm_log->config_hash = strdupz((char *)config_hash_id);
794
@@ -805,7 +803,7 @@ void health_alarm_entry2proto_nolock(struct alarm_log_entry *alarm_log, ALARM_EN
803 alarm_log->non_clear_duration = (time_t)ae->non_clear_duration;
804 alarm_log->status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)ae->new_status);
805 alarm_log->old_status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)ae->old_status);
808 - alarm_log->delay = (int)ae->delay;
806 + alarm_log->delay = ae->delay;
807 alarm_log->delay_up_to_timestamp = (time_t)ae->delay_up_to_timestamp;
808 alarm_log->last_repeat = (time_t)ae->last_repeat;
809
@@ -835,18 +833,18 @@ void health_alarm_entry2proto_nolock(struct alarm_log_entry *alarm_log, ALARM_EN
833 #endif
834
835 #ifdef ENABLE_ACLK
838 -static int have_recent_alarm(RRDHOST *host, uint32_t alarm_id, uint32_t mark)
836 +static bool have_recent_alarm(RRDHOST *host, int64_t alarm_id, int64_t mark)
837 {
838 ALARM_ENTRY *ae = host->health_log.alarms;
839
840 while (ae) {
841 if (ae->alarm_id == alarm_id && ae->unique_id >mark &&
842 (ae->new_status != RRDCALC_STATUS_WARNING && ae->new_status != RRDCALC_STATUS_CRITICAL))
845 - return 1;
843 + return true;
844 ae = ae->next;
845 }
846
849 - return 0;
847 + return false;
848 }
849 #endif
850
@@ -863,7 +861,7 @@ void aclk_push_alert_snapshot_event(char *node_id __maybe_unused)
861 }
862 freez(node_id);
863
866 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
864 + struct aclk_sync_cfg_t *wc = host->aclk_config;
865
866 // we perhaps we don't need this for snapshots
867 if (unlikely(!wc->alert_updates)) {
@@ -884,8 +882,6 @@ void aclk_push_alert_snapshot_event(char *node_id __maybe_unused)
882 netdata_log_access("ACLK REQ [%s (%s)]: Sending alerts snapshot, snapshot_uuid %s", wc->node_id, rrdhost_hostname(wc->host), wc->alerts_snapshot_uuid);
883
884 uint32_t cnt = 0;
887 - char uuid_str[UUID_STR_LEN];
888 - uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
885
886 rw_spinlock_read_lock(&host->health_log.spinlock);
887
@@ -908,7 +904,7 @@ void aclk_push_alert_snapshot_event(char *node_id __maybe_unused)
904 }
905
906 if (cnt) {
911 - uint32_t chunk = 1, chunks = 0;
907 + uint32_t chunk = 1, chunks;
908
909 chunks = (cnt / ALARM_EVENTS_PER_CHUNK) + (cnt % ALARM_EVENTS_PER_CHUNK != 0);
910 ae = host->health_log.alarms;
@@ -979,21 +975,38 @@ void aclk_push_alert_snapshot_event(char *node_id __maybe_unused)
975 #endif
976 }
977
982 -#define SQL_DELETE_ALERT_ENTRIES "DELETE FROM aclk_alert_%s WHERE date_created + %d < UNIXEPOCH();"
978 +#define SQL_DELETE_ALERT_ENTRIES "DELETE FROM aclk_alert_%s WHERE date_created < UNIXEPOCH() - @period"
979 +
980 void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
981 {
985 - char uuid_str[UUID_STR_LEN];
986 - uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
982 + struct aclk_sync_cfg_t *wc = host->aclk_config;
983 + if (unlikely(!wc))
984 + return;
985
986 char sql[ACLK_SYNC_QUERY_SIZE];
989 - snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_DELETE_ALERT_ENTRIES, uuid_str, MAX_REMOVED_PERIOD);
987 + snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_DELETE_ALERT_ENTRIES, wc->uuid_str);
988
991 - char *err_msg = NULL;
992 - int rc = sqlite3_exec_monitored(db_meta, sql, NULL, NULL, &err_msg);
989 + sqlite3_stmt *res = NULL;
990 + int rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
991 if (rc != SQLITE_OK) {
994 - error_report("Failed when trying to clean stale ACLK alert entries from aclk_alert_%s, error message \"%s\"", uuid_str, err_msg);
995 - sqlite3_free(err_msg);
992 + error_report("Failed to prepare statement for cleaning stale ACLK alert entries.");
993 + return;
994 }
995 +
996 + rc = sqlite3_bind_int64(res, 1, MAX_REMOVED_PERIOD);
997 + if (unlikely(rc != SQLITE_OK)) {
998 + error_report("Failed to bind MAX_REMOVED_PERIOD parameter.");
999 + goto skip;
1000 + }
1001 +
1002 + rc = sqlite3_step_monitored(res);
1003 + if (rc != SQLITE_DONE)
1004 + error_report("Failed to execute DELETE query for cleaning stale ACLK alert entries.");
1005 +
1006 +skip:
1007 + rc = sqlite3_finalize(res);
1008 + if (unlikely(rc != SQLITE_OK))
1009 + error_report("Failed to finalize statement for cleaning stale ACLK alert entries.");
1010 }
1011
1012 #define SQL_GET_MIN_MAX_ALERT_SEQ "SELECT MIN(sequence_id), MAX(sequence_id), " \
@@ -1001,29 +1014,31 @@ void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
1014 "FROM aclk_alert_%s WHERE date_submitted IS NULL;"
1015 int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status)
1016 {
1004 - int rc;
1005 - struct aclk_sync_host_config *wc = NULL;
1006 - wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
1017 +
1018 + struct aclk_sync_cfg_t *wc = host->aclk_config;
1019 if (!wc)
1020 return 1;
1021
1022 proto_alert_status->alert_updates = wc->alert_updates;
1023
1024 char sql[ACLK_SYNC_QUERY_SIZE];
1013 - sqlite3_stmt *res = NULL;
1025
1026 + sqlite3_stmt *res = NULL;
1027 snprintfz(sql, ACLK_SYNC_QUERY_SIZE - 1, SQL_GET_MIN_MAX_ALERT_SEQ, wc->uuid_str, wc->uuid_str);
1028
1017 - rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
1029 + int rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
1030 if (rc != SQLITE_OK) {
1031 error_report("Failed to prepare statement to get alert log status from the database.");
1032 return 1;
1033 }
1034
1035 while (sqlite3_step_monitored(res) == SQLITE_ROW) {
1024 - proto_alert_status->pending_min_sequence_id = sqlite3_column_bytes(res, 0) > 0 ? (uint64_t) sqlite3_column_int64(res, 0) : 0;
1025 - proto_alert_status->pending_max_sequence_id = sqlite3_column_bytes(res, 1) > 0 ? (uint64_t) sqlite3_column_int64(res, 1) : 0;
1026 - proto_alert_status->last_submitted_sequence_id = sqlite3_column_bytes(res, 2) > 0 ? (uint64_t) sqlite3_column_int64(res, 2) : 0;
1036 + proto_alert_status->pending_min_sequence_id =
1037 + sqlite3_column_bytes(res, 0) > 0 ? (uint64_t)sqlite3_column_int64(res, 0) : 0;
1038 + proto_alert_status->pending_max_sequence_id =
1039 + sqlite3_column_bytes(res, 1) > 0 ? (uint64_t)sqlite3_column_int64(res, 1) : 0;
1040 + proto_alert_status->last_submitted_sequence_id =
1041 + sqlite3_column_bytes(res, 2) > 0 ? (uint64_t)sqlite3_column_int64(res, 2) : 0;
1042 }
1043
1044 rc = sqlite3_finalize(res);
@@ -1038,21 +1053,15 @@ void aclk_send_alarm_checkpoint(char *node_id, char *claim_id __maybe_unused)
1053 if (unlikely(!node_id))
1054 return;
1055
1041 - struct aclk_sync_host_config *wc = NULL;
1056 + struct aclk_sync_cfg_t *wc;
1057 RRDHOST *host = find_host_by_node_id(node_id);
1058
1044 - if (unlikely(!host))
1045 - return;
1046 -
1047 - wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
1048 - if (unlikely(!wc)) {
1059 + if (unlikely(!host || !(wc = host->aclk_config)))
1060 netdata_log_access("ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT REQUEST RECEIVED FOR INVALID NODE", node_id);
1050 - return;
1061 + else {
1062 + netdata_log_access("ACLK REQ [%s (%s)]: ALERTS CHECKPOINT REQUEST RECEIVED", node_id, rrdhost_hostname(host));
1063 + wc->alert_checkpoint_req = SEND_CHECKPOINT_AFTER_HEALTH_LOOPS;
1064 }
1052 -
1053 - netdata_log_access("ACLK REQ [%s (%s)]: ALERTS CHECKPOINT REQUEST RECEIVED", node_id, rrdhost_hostname(host));
1054 -
1055 - wc->alert_checkpoint_req = SEND_CHECKPOINT_AFTER_HEALTH_LOOPS;
1065 }
1066
1067 typedef struct active_alerts {
@@ -1061,15 +1070,14 @@ typedef struct active_alerts {
1070 RRDCALC_STATUS status;
1071 } active_alerts_t;
1072
1064 -static inline int compare_active_alerts(const void * a, const void * b) {
1073 +static inline int compare_active_alerts(const void *a, const void *b)
1074 +{
1075 active_alerts_t *active_alerts_a = (active_alerts_t *)a;
1076 active_alerts_t *active_alerts_b = (active_alerts_t *)b;
1077
1068 - if( !(strcmp(active_alerts_a->name, active_alerts_b->name)) )
1069 - {
1070 - return strcmp(active_alerts_a->chart, active_alerts_b->chart);
1071 - }
1072 - else
1078 + if (!(strcmp(active_alerts_a->name, active_alerts_b->name))) {
1079 + return strcmp(active_alerts_a->chart, active_alerts_b->chart);
1080 + } else
1081 return strcmp(active_alerts_a->name, active_alerts_b->name);
1082 }
1083
@@ -1077,7 +1085,7 @@ static inline int compare_active_alerts(const void * a, const void * b) {
1085 void aclk_push_alarm_checkpoint(RRDHOST *host __maybe_unused)
1086 {
1087 #ifdef ENABLE_ACLK
1080 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
1088 + struct aclk_sync_cfg_t *wc = host->aclk_config;
1089 if (unlikely(!wc)) {
1090 netdata_log_access("ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT REQUEST RECEIVED FOR INVALID NODE", rrdhost_hostname(host));
1091 return;
@@ -1085,7 +1093,7 @@ void aclk_push_alarm_checkpoint(RRDHOST *host __maybe_unused)
1093
1094 if (rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS)) {
1095 //postpone checkpoint send
1088 - wc->alert_checkpoint_req+=3;
1096 + wc->alert_checkpoint_req += 3;
1097 netdata_log_access("ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT POSTPONED", rrdhost_hostname(host));
1098 return;
1099 }
@@ -1119,16 +1127,16 @@ void aclk_push_alarm_checkpoint(RRDHOST *host __maybe_unused)
1127
1128 BUFFER *alarms_to_hash;
1129 if (cnt) {
1122 - qsort (active_alerts, cnt, sizeof(active_alerts_t), compare_active_alerts);
1130 + qsort(active_alerts, cnt, sizeof(active_alerts_t), compare_active_alerts);
1131
1132 alarms_to_hash = buffer_create(len, NULL);
1125 - for (uint32_t i=0;i<cnt;i++) {
1133 + for (uint32_t i = 0; i < cnt; i++) {
1134 buffer_strcat(alarms_to_hash, active_alerts[i].name);
1135 buffer_strcat(alarms_to_hash, active_alerts[i].chart);
1136 if (active_alerts[i].status == RRDCALC_STATUS_WARNING)
1129 - buffer_strcat(alarms_to_hash, "W");
1137 + buffer_fast_strcat(alarms_to_hash, "W", 1);
1138 else if (active_alerts[i].status == RRDCALC_STATUS_CRITICAL)
1131 - buffer_strcat(alarms_to_hash, "C");
1139 + buffer_fast_strcat(alarms_to_hash, "C", 1);
1140 }
1141 } else {
1142 alarms_to_hash = buffer_create(1, NULL);
@@ -1150,9 +1158,9 @@ void aclk_push_alarm_checkpoint(RRDHOST *host __maybe_unused)
1158 aclk_send_provide_alarm_checkpoint(&alarm_checkpoint);
1159 freez(claim_id);
1160 netdata_log_access("ACLK RES [%s (%s)]: ALERTS CHECKPOINT SENT", wc->node_id, rrdhost_hostname(host));
1153 - } else {
1161 + } else
1162 netdata_log_access("ACLK RES [%s (%s)]: FAILED TO CREATE ALERTS CHECKPOINT HASH", wc->node_id, rrdhost_hostname(host));
1155 - }
1163 +
1164 wc->alert_checkpoint_req = 0;
1165 buffer_free(alarms_to_hash);
1166 #endif
database/sqlite/sqlite_aclk_alert.h
+1 -1
@@ -15,7 +15,7 @@ struct proto_alert_status {
15 uint64_t last_submitted_sequence_id;
16 };
17
18 -void aclk_push_alert_event(struct aclk_sync_host_config *wc);
18 +void aclk_push_alert_event(struct aclk_sync_cfg_t *wc);
19 void aclk_send_alarm_configuration (char *config_hash);
20 int aclk_push_alert_config_event(char *node_id, char *config_hash);
21 void aclk_start_alert_streaming(char *node_id, bool resets);
database/sqlite/sqlite_aclk_node.c
+3 -3
@@ -29,7 +29,7 @@ DICTIONARY *collectors_from_charts(RRDHOST *host, DICTIONARY *dict) {
29
30 static void build_node_collectors(RRDHOST *host)
31 {
32 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) host->aclk_sync_host_config;
32 + struct aclk_sync_cfg_t *wc = host->aclk_config;
33
34 struct update_node_collectors upd_node_collectors;
35 DICTIONARY *dict = dictionary_create(DICT_OPTION_SINGLE_THREADED);
@@ -50,7 +50,7 @@ static void build_node_info(RRDHOST *host)
50 {
51 struct update_node_info node_info;
52
53 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *) host->aclk_sync_host_config;
53 + struct aclk_sync_cfg_t *wc = host->aclk_config;
54
55 rrd_rdlock();
56 node_info.node_id = wc->node_id;
@@ -138,7 +138,7 @@ void aclk_check_node_info_and_collectors(void)
138 size_t replicating = 0;
139 dfe_start_reentrant(rrdhost_root_index, host)
140 {
141 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
141 + struct aclk_sync_cfg_t *wc = host->aclk_config;
142 if (unlikely(!wc))
143 continue;
144
database/sqlite/sqlite_functions.c
+1 -1
@@ -548,7 +548,7 @@ static inline void set_host_node_id(RRDHOST *host, uuid_t *node_id)
548 return;
549 }
550
551 - struct aclk_sync_host_config *wc = host->aclk_sync_host_config;
551 + struct aclk_sync_cfg_t *wc = host->aclk_config;
552
553 if (unlikely(!host->node_id)) {
554 uuid_t *t = mallocz(sizeof(*host->node_id));
health/health.c
+5 -6
@@ -383,7 +383,7 @@ static void health_reload_host(RRDHOST *host) {
383
384 #ifdef ENABLE_ACLK
385 if (netdata_cloud_enabled) {
386 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
386 + struct aclk_sync_cfg_t *wc = host->aclk_config;
387 if (likely(wc)) {
388 wc->alert_queue_removed = SEND_REMOVED_AFTER_HEALTH_LOOPS;
389 }
@@ -422,7 +422,7 @@ static inline int compare_active_alerts(const void * a, const void * b) {
422 active_alerts_t *active_alerts_a = (active_alerts_t *)a;
423 active_alerts_t *active_alerts_b = (active_alerts_t *)b;
424
425 - return ( active_alerts_b->last_status_change - active_alerts_a->last_status_change );
425 + return (int) ( active_alerts_b->last_status_change - active_alerts_a->last_status_change );
426 }
427
428 static inline void health_alarm_execute(RRDHOST *host, ALARM_ENTRY *ae) {
@@ -943,7 +943,7 @@ static int update_disabled_silenced(RRDHOST *host, RRDCALC *rc) {
943 static void sql_health_postpone_queue_removed(RRDHOST *host __maybe_unused) {
944 #ifdef ENABLE_ACLK
945 if (netdata_cloud_enabled) {
946 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
946 + struct aclk_sync_cfg_t *wc = host->aclk_config;
947 if (unlikely(!wc)) {
948 return;
949 }
@@ -1554,10 +1554,9 @@ void *health_main(void *ptr) {
1554 }
1555 #ifdef ENABLE_ACLK
1556 if (netdata_cloud_enabled) {
1557 - struct aclk_sync_host_config *wc = (struct aclk_sync_host_config *)host->aclk_sync_host_config;
1558 - if (unlikely(!wc)) {
1557 + struct aclk_sync_cfg_t *wc = host->aclk_config;
1558 + if (unlikely(!wc))
1559 continue;
1560 - }
1560
1561 if (wc->alert_queue_removed == 1) {
1562 sql_queue_removed_alerts_to_aclk(host);