@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 1,201 lines 44.4 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "sqlite_functions.h"
4 #include "sqlite_aclk_alert.h"
5
6 #include "../../aclk/aclk_alarm_api.h"
7
8 extern __thread bool is_health_thread;
9
10 #define SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param) \
11 ({ \
12 int _param = (param); \
13 sqlite3_column_bytes((res), (_param)) ? strdupz((char *)sqlite3_column_text((res), (_param))) : NULL; \
14 })
15
16 #define SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID \
17 "SELECT hld.unique_id FROM health_log hl, alert_hash ah, health_log_detail hld " \
18 "WHERE hld.unique_id = @unique_id AND hl.config_hash_id = ah.hash_id AND hld.health_log_id = hl.health_log_id " \
19 "AND hl.host_id = @host_id AND ah.warn IS NULL AND ah.crit IS NULL"
20
21 static inline bool is_event_from_alert_variable_config(int64_t unique_id, nd_uuid_t *host_id)
22 {
23 static __thread sqlite3_stmt *compiled_res = NULL;
24 sqlite3_stmt *res = NULL;
25
26 if (is_health_thread) {
27 if (!compiled_res) {
28 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &compiled_res))
29 return false;
30 }
31 res = compiled_res;
32 } else {
33 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &res))
34 return false;
35 }
36
37 bool ret = false;
38
39 int param = 0;
40 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));
41 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_id, sizeof(*host_id), SQLITE_STATIC));
42
43 param = 0;
44 ret = (sqlite3_step_monitored(res) == SQLITE_ROW);
45
46 done:
47 REPORT_BIND_FAIL(res, param);
48 if (is_health_thread)
49 SQLITE_RESET(res);
50 else
51 SQLITE_FINALIZE(res);
52 return ret;
53 }
54
55 #define SQL_UPDATE_ALERT_VERSION_TRANSITION \
56 "UPDATE alert_version SET unique_id = @unique_id WHERE health_log_id = @health_log_id"
57
58 static void update_alert_version_transition(int64_t health_log_id, int64_t unique_id)
59 {
60 static __thread sqlite3_stmt *compiled_res = NULL;
61 sqlite3_stmt *res = NULL;
62
63 if (is_health_thread) {
64 if (!compiled_res) {
65 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &compiled_res))
66 return;
67 }
68 res = compiled_res;
69 } else {
70 if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &res))
71 return;
72 }
73
74 int param = 0;
75 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));
76 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));
77
78 param = 0;
79 int rc = sqlite3_step_monitored(res);
80 if (rc != SQLITE_DONE)
81 error_report("Failed to update alert_version to latest transition");
82
83 done:
84 REPORT_BIND_FAIL(res, param);
85 if (is_health_thread)
86 SQLITE_RESET(res);
87 else
88 SQLITE_FINALIZE(res);
89 }
90
91 //decide if some events should be sent or not
92
93 #define SQL_SELECT_LAST_ALERT_STATUS "SELECT status FROM alert_version WHERE health_log_id = @health_log_id "
94
95 static bool cloud_status_matches(int64_t health_log_id, RRDCALC_STATUS status)
96 {
97 static __thread sqlite3_stmt *compiled_res = NULL;
98 sqlite3_stmt *res = NULL;
99
100 if (is_health_thread) {
101 if (!compiled_res) {
102 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &compiled_res))
103 return true;
104 }
105 res = compiled_res;
106 } else {
107 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &res))
108 return true;
109 }
110
111 bool send = false;
112
113 int param = 0;
114 SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, health_log_id));
115
116 param = 0;
117 int rc = sqlite3_step_monitored(res);
118 if (likely(rc == SQLITE_ROW)) {
119 RRDCALC_STATUS current_status = (RRDCALC_STATUS)sqlite3_column_int(res, 0);
120 send = (current_status == status);
121 }
122
123 done:
124 REPORT_BIND_FAIL(res, param);
125 if (is_health_thread)
126 SQLITE_RESET(res);
127 else
128 SQLITE_FINALIZE(res);
129 return send;
130 }
131
132 #define SQL_QUEUE_ALERT_TO_CLOUD \
133 "INSERT INTO aclk_queue (host_id, health_log_id, unique_id, date_created)" \
134 " VALUES (@host_id, @health_log_id, @unique_id, UNIXEPOCH())" \
135 " ON CONFLICT(host_id, health_log_id) DO UPDATE SET unique_id=excluded.unique_id, " \
136 " date_created=excluded.date_created"
137
138 //
139 // Attempt to insert an alert to the submit queue to reach the cloud
140 //
141 // The alert will NOT be added in the submit queue if
142 // - Cloud is already aware of the alert status
143 // - The transition refers to a variable
144 //
145 static int insert_alert_to_submit_queue(RRDHOST *host, int64_t health_log_id, uint32_t unique_id, RRDCALC_STATUS status)
146 {
147 static __thread sqlite3_stmt *compiled_res = NULL;
148 sqlite3_stmt *res = NULL;
149
150 if (cloud_status_matches(health_log_id, status)) {
151 update_alert_version_transition(health_log_id, unique_id);
152 return 1;
153 }
154
155 if (is_event_from_alert_variable_config(unique_id, &host->host_id.uuid))
156 return 2;
157
158 if (is_health_thread) {
159 if (!compiled_res) {
160 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &compiled_res))
161 return -1;
162 }
163 res = compiled_res;
164 } else {
165 if (!PREPARE_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &res))
166 return -1;
167 }
168
169 int param = 0;
170 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
171 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));
172 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, (int64_t) unique_id));
173
174 param = 0;
175 int rc = sqlite3_step_monitored(res);
176 if (unlikely(rc != SQLITE_DONE))
177 error_report("Failed to insert alert in the submit queue %"PRIu32", rc = %d", unique_id, rc);
178
179 done:
180 REPORT_BIND_FAIL(res, param);
181 if (is_health_thread)
182 SQLITE_RESET(res);
183 else
184 SQLITE_FINALIZE(res);
185 return 0;
186 }
187
188 #define SQL_DELETE_QUEUE_ALERT_TO_CLOUD \
189 "DELETE FROM aclk_queue WHERE host_id = @host_id AND sequence_id BETWEEN @seq1 AND @seq2"
190
191 //
192 // Delete a range of alerts from the submit queue (after being sent to the the cloud)
193 //
194 static int delete_alert_from_submit_queue(RRDHOST *host, int64_t first_seq_id, int64_t last_seq_id)
195 {
196 sqlite3_stmt *res = NULL;
197
198 if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_QUEUE_ALERT_TO_CLOUD, &res))
199 return -1;
200
201 int param = 0;
202 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
203 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, first_seq_id));
204 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, last_seq_id));
205
206 param = 0;
207 int rc = sqlite3_step_monitored(res);
208 if (rc != SQLITE_DONE)
209 error_report("Failed to delete submitted to ACLK");
210
211 done:
212 REPORT_BIND_FAIL(res, param);
213 SQLITE_FINALIZE(res);
214 return 0;
215 }
216
217 int rrdcalc_status_to_proto_enum(RRDCALC_STATUS status)
218 {
219
220 switch(status) {
221 case RRDCALC_STATUS_REMOVED:
222 return ALARM_STATUS_REMOVED;
223
224 case RRDCALC_STATUS_UNDEFINED:
225 return ALARM_STATUS_NOT_A_NUMBER;
226
227 case RRDCALC_STATUS_CLEAR:
228 return ALARM_STATUS_CLEAR;
229
230 case RRDCALC_STATUS_WARNING:
231 return ALARM_STATUS_WARNING;
232
233 case RRDCALC_STATUS_CRITICAL:
234 return ALARM_STATUS_CRITICAL;
235
236 default:
237 return ALARM_STATUS_UNKNOWN;
238 }
239 }
240
241 static inline char *sqlite3_uuid_unparse_strdupz(sqlite3_stmt *res, int iCol) {
242 char uuid_str[UUID_STR_LEN];
243
244 if(sqlite3_column_type(res, iCol) == SQLITE_NULL)
245 uuid_str[0] = '\0';
246 else if (!sqlite3_column_uuid_unparse_lower(res, iCol, uuid_str)) {
247 error_report("ACLK ALERT: Got invalid UUID blob at column %d. Returning empty string.", iCol);
248 uuid_str[0] = '\0';
249 }
250
251 return strdupz(uuid_str);
252 }
253
254 static inline char *sqlite3_text_strdupz_empty(sqlite3_stmt *res, int iCol) {
255 char *ret;
256
257 if(sqlite3_column_type(res, iCol) == SQLITE_NULL)
258 ret = "";
259 else
260 ret = (char *)sqlite3_column_text(res, iCol);
261
262 return strdupz(ret);
263 }
264
265 #define SQL_UPDATE_ALERT_VERSION \
266 "INSERT INTO alert_version (health_log_id, unique_id, status, version, date_submitted)" \
267 " VALUES (@health_log_id, @unique_id, @status, @version, UNIXEPOCH())" \
268 " ON CONFLICT(health_log_id) DO UPDATE SET status = excluded.status, version = excluded.version, " \
269 " unique_id=excluded.unique_id, date_submitted=excluded.date_submitted"
270
271 //
272 // Store a new alert transition along with the version after sending to the cloud
273 // - Update an existing alert with the updated version, status, transition and date submitted
274 //
275 static void sql_update_alert_version(
276 int64_t health_log_id,
277 int64_t unique_id,
278 RRDCALC_STATUS status,
279 uint64_t version,
280 sqlite3_stmt **res)
281 {
282 if (!*res) {
283 if (!PREPARE_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION, res))
284 return;
285 }
286
287 int param = 0;
288 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, health_log_id));
289 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, unique_id));
290 SQLITE_BIND_FAIL(done, sqlite3_bind_int(*res, ++param, status));
291 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(*res, ++param, version));
292
293 param = 0;
294 int rc = sqlite3_step_monitored(*res);
295 if (rc != SQLITE_DONE)
296 error_report("Failed to execute sql_update_alert_version");
297
298 done:
299 REPORT_BIND_FAIL(*res, param);
300 SQLITE_RESET(*res);
301 }
302
303 #define SQL_SELECT_ALERT_TO_DUMMY \
304 "SELECT aq.sequence_id, hld.unique_id, hld.when_key, hld.new_status, hld.health_log_id" \
305 " FROM health_log hl, aclk_queue aq, alert_hash ah, health_log_detail hld" \
306 " WHERE hld.unique_id = aq.unique_id AND hl.config_hash_id = ah.hash_id" \
307 " AND hl.host_id = @host_id AND aq.host_id = hl.host_id AND hl.health_log_id = hld.health_log_id" \
308 " ORDER BY aq.sequence_id ASC"
309
310 //
311 // Check all queued alerts for a host and commit them as if they have been send to the cloud
312 // this will produce new versions as needed. We need this because we are about to send a
313 // a snapshot so we can include the latest transition.
314 //
315 static void commit_alert_events(RRDHOST *host)
316 {
317 sqlite3_stmt *res = NULL;
318
319 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_DUMMY, &res))
320 return;
321
322 int param = 0;
323 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
324
325 int64_t first_sequence_id = 0;
326 int64_t last_sequence_id = 0;
327
328 sqlite3_stmt *res_version = NULL;
329 param = 0;
330 while (sqlite3_step_monitored(res) == SQLITE_ROW) {
331
332 last_sequence_id = sqlite3_column_int64(res, 0);
333 if (first_sequence_id == 0)
334 first_sequence_id = last_sequence_id;
335
336 int64_t unique_id = sqlite3_column_int(res, 1);
337 int64_t version = sqlite3_column_int64(res, 2);
338 RRDCALC_STATUS status = (RRDCALC_STATUS)sqlite3_column_int(res, 3);
339 int64_t health_log_id = sqlite3_column_int64(res, 4);
340
341 // Prepare the statement on the first time (res_version) then reuse it
342 // finalize when we are done
343 sql_update_alert_version(health_log_id, unique_id, status, version, &res_version);
344 }
345
346 if (first_sequence_id)
347 delete_alert_from_submit_queue(host, first_sequence_id, last_sequence_id);
348
349 done:
350 REPORT_BIND_FAIL(res, param);
351 SQLITE_FINALIZE(res);
352 }
353
354 typedef enum {
355 SEQUENCE_ID,
356 UNIQUE_ID,
357 ALARM_ID,
358 CONFIG_HASH_ID,
359 UPDATED_BY_ID,
360 WHEN_KEY,
361 DURATION,
362 NON_CLEAR_DURATION,
363 FLAGS,
364 EXEC_RUN_TIMESTAMP,
365 DELAY_UP_TO_TIMESTAMP,
366 NAME,
367 CHART,
368 EXEC,
369 RECIPIENT,
370 SOURCE,
371 UNITS,
372 INFO,
373 EXEC_CODE,
374 NEW_STATUS,
375 OLD_STATUS,
376 DELAY,
377 NEW_VALUE,
378 OLD_VALUE,
379 LAST_REPEAT,
380 CHART_CONTEXT,
381 TRANSITION_ID,
382 ALARM_EVENT_ID,
383 CHART_NAME,
384 SUMMARY,
385 HEALTH_LOG_ID,
386 VERSION
387 } HealthLogDetails;
388
389 void health_alarm_log_populate(
390 struct alarm_log_entry *alarm_log,
391 sqlite3_stmt *res,
392 RRDHOST *host,
393 RRDCALC_STATUS *status)
394 {
395 char old_value_string[100 + 1];
396 char new_value_string[100 + 1];
397
398 RRDCALC_STATUS current_status = (RRDCALC_STATUS)sqlite3_column_int(res, NEW_STATUS);
399 if (status)
400 *status = current_status;
401
402 char *source = (char *) sqlite3_column_text(res, SOURCE);
403 alarm_log->command = source ? health_edit_command_from_source(source) : strdupz("UNKNOWN=0=UNKNOWN");
404
405 alarm_log->chart = strdupz((char *) sqlite3_column_text(res, CHART));
406 alarm_log->name = strdupz((char *) sqlite3_column_text(res, NAME));
407
408 alarm_log->when = sqlite3_column_int64(res, WHEN_KEY);
409
410 alarm_log->config_hash = sqlite3_uuid_unparse_strdupz(res, CONFIG_HASH_ID);
411
412 {
413 RRDHOST_TZ host_tz = rrdhost_tz_get(host);
414 alarm_log->utc_offset = host_tz.utc_offset;
415 // Transfer ownership of the strdup'd copy instead of duplicating again
416 alarm_log->timezone = host_tz.abbrev_timezone;
417 host_tz.abbrev_timezone = NULL;
418 rrdhost_tz_free(&host_tz);
419 }
420 alarm_log->exec_path = sqlite3_column_bytes(res, EXEC) ?
421 strdupz((char *)sqlite3_column_text(res, EXEC)) :
422 strdupz((char *)string2str(host->health.default_exec));
423
424 alarm_log->conf_source = source ? strdupz(source) : strdupz("");
425
426 time_t duration = sqlite3_column_int64(res, DURATION);
427 alarm_log->duration = (duration > 0) ? duration : 0;
428
429 int64_t non_clear_duration = sqlite3_column_int64(res, NON_CLEAR_DURATION);
430 alarm_log->non_clear_duration = (non_clear_duration <= 0) ? 0 :
431 (non_clear_duration > UINT32_MAX) ? UINT32_MAX : (uint32_t)non_clear_duration;
432
433 alarm_log->status = rrdcalc_status_to_proto_enum(current_status);
434 alarm_log->old_status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)sqlite3_column_int64(res, OLD_STATUS));
435 alarm_log->delay = sqlite3_column_int64(res, DELAY);
436 alarm_log->delay_up_to_timestamp = sqlite3_column_int64(res, DELAY_UP_TO_TIMESTAMP);
437 alarm_log->last_repeat = sqlite3_column_int64(res, LAST_REPEAT);
438
439 uint64_t flags = sqlite3_column_int64(res, FLAGS);
440 char *recipient = (char *) sqlite3_column_text(res, RECIPIENT);
441 alarm_log->silenced =
442 ((flags & HEALTH_ENTRY_FLAG_SILENCED) || (recipient && !strncmp(recipient, "silent", 6))) ? 1 : 0;
443
444 double value = sqlite3_column_double(res, NEW_VALUE);
445 double old_value = sqlite3_column_double(res, OLD_VALUE);
446
447 alarm_log->value_string =
448 sqlite3_column_type(res, NEW_VALUE) == SQLITE_NULL ?
449 strdupz((char *)"-") :
450 strdupz((char *)format_value_and_unit(
451 new_value_string, 100, value, (char *)sqlite3_column_text(res, UNITS), -1));
452
453 alarm_log->old_value_string =
454 sqlite3_column_type(res, OLD_VALUE) == SQLITE_NULL ?
455 strdupz((char *)"-") :
456 strdupz((char *)format_value_and_unit(
457 old_value_string, 100, old_value, (char *)sqlite3_column_text(res, UNITS), -1));
458
459 alarm_log->value = (!isnan(value)) ? (NETDATA_DOUBLE)value : 0;
460 alarm_log->old_value = (!isnan(old_value)) ? (NETDATA_DOUBLE)old_value : 0;
461
462 alarm_log->updated = (flags & HEALTH_ENTRY_FLAG_UPDATED) ? 1 : 0;
463 alarm_log->rendered_info = sqlite3_text_strdupz_empty(res, INFO);
464 alarm_log->chart_context = sqlite3_text_strdupz_empty(res, CHART_CONTEXT);
465 alarm_log->chart_name = sqlite3_text_strdupz_empty(res, CHART_NAME);
466
467 alarm_log->transition_id = sqlite3_uuid_unparse_strdupz(res, TRANSITION_ID);
468 alarm_log->event_id = sqlite3_column_int64(res, ALARM_EVENT_ID);
469 alarm_log->version = sqlite3_column_int64(res, VERSION);
470
471 alarm_log->summary = sqlite3_text_strdupz_empty(res, SUMMARY);
472
473 alarm_log->health_log_id = sqlite3_column_int64(res, HEALTH_LOG_ID);
474 alarm_log->unique_id = sqlite3_column_int64(res, UNIQUE_ID);
475 alarm_log->alarm_id = sqlite3_column_int64(res, ALARM_ID);
476 alarm_log->sequence_id = sqlite3_column_int64(res, SEQUENCE_ID);
477 }
478
479 #define SQL_SELECT_ALERT_TO_PUSH \
480 "SELECT aq.sequence_id, hld.unique_id, hld.alarm_id, hl.config_hash_id, hld.updated_by_id, hld.when_key," \
481 " hld.duration, hld.non_clear_duration, hld.flags, hld.exec_run_timestamp, hld.delay_up_to_timestamp, hl.name," \
482 " hl.chart, hl.exec, hl.recipient, ah.source, hl.units, hld.info, hld.exec_code, hld.new_status," \
483 " hld.old_status, hld.delay, hld.new_value, hld.old_value, hld.last_repeat, hl.chart_context, hld.transition_id," \
484 " hld.alarm_event_id, hl.chart_name, hld.summary, hld.health_log_id, hld.when_key" \
485 " FROM health_log hl, aclk_queue aq, alert_hash ah, health_log_detail hld" \
486 " WHERE hld.unique_id = aq.unique_id AND hl.config_hash_id = ah.hash_id" \
487 " AND hl.host_id = @host_id AND aq.host_id = hl.host_id AND hl.health_log_id = hld.health_log_id" \
488 " ORDER BY aq.sequence_id ASC LIMIT "ACLK_MAX_ALERT_UPDATES
489
490 static void aclk_push_alert_event(RRDHOST *host, sqlite3_stmt **res, sqlite3_stmt **res_version)
491 {
492 CLAIM_ID claim_id = claim_id_get();
493
494 if (!claim_id_is_set(claim_id) || UUIDiszero(host->node_id))
495 return;
496
497 if (!*res) {
498 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_PUSH, res))
499 return;
500 }
501
502 int param = 0;
503 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(*res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
504
505 char node_id_str[UUID_STR_LEN];
506 uuid_unparse_lower(host->node_id.uuid, node_id_str);
507
508 struct alarm_log_entry alarm_log;
509 alarm_log.node_id = node_id_str;
510 alarm_log.claim_id = claim_id.str;
511
512 int64_t first_id = 0;
513 int64_t last_id = 0;
514
515 param = 0;
516 RRDCALC_STATUS status;
517 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
518 while (sqlite3_step_monitored(*res) == SQLITE_ROW) {
519 health_alarm_log_populate(&alarm_log, *res, host, &status);
520 aclk_send_alarm_log_entry(&alarm_log);
521
522 nd_uuid_t hash_id;
523 if (alarm_log.config_hash && !uuid_parse(alarm_log.config_hash, hash_id))
524 alert_hash_mark_sent(&hash_id);
525
526 aclk_host_config->alert_count++;
527
528 last_id = alarm_log.sequence_id;
529 if (first_id == 0)
530 first_id = last_id;
531
532 // The statement to set the version will be compiled once and reset when done
533 // out caller will finalize the statement to release resources
534 sql_update_alert_version(alarm_log.health_log_id, alarm_log.unique_id, status, alarm_log.version, res_version);
535
536 destroy_alarm_log_entry(&alarm_log);
537 }
538
539 if (first_id) {
540 nd_log(
541 NDLS_ACCESS,
542 NDLP_DEBUG,
543 "ACLK RES [%s (%s)]: ALERTS SENT from %lld - %lld",
544 node_id_str,
545 rrdhost_hostname(host),
546 (long long)first_id,
547 (long long)last_id);
548
549 delete_alert_from_submit_queue(host, first_id, last_id);
550 // Mark to do one more check
551 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
552 }
553
554 done:
555 REPORT_BIND_FAIL(*res, param);
556 SQLITE_RESET(*res);
557 }
558
559 #define SQL_DELETE_PROCESSED_ROWS "DELETE FROM alert_queue WHERE host_id = @host_id AND rowid = @row"
560
561 static void delete_alert_from_pending_queue(RRDHOST *host, int64_t row)
562 {
563 static __thread sqlite3_stmt *compiled_res = NULL;
564 sqlite3_stmt *res = NULL;
565
566 if (is_health_thread) {
567 if (!compiled_res) {
568 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &compiled_res))
569 return;
570 }
571 res = compiled_res;
572 } else {
573 if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &res))
574 return;
575 }
576
577 int param = 0;
578 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
579 SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, row));
580
581 param = 0;
582 int rc = sqlite3_step_monitored(res);
583 if (rc != SQLITE_DONE)
584 error_report("Failed to delete processed rows, rc = %d", rc);
585
586 done:
587 REPORT_BIND_FAIL(res, param);
588 if (is_health_thread)
589 SQLITE_RESET(res);
590 else
591 SQLITE_FINALIZE(res);
592 }
593
594 #define SQL_REBUILD_HOST_ALERT_VERSION_TABLE \
595 "INSERT OR IGNORE INTO alert_version (health_log_id, unique_id, status, version, date_submitted) " \
596 " SELECT hl.health_log_id, hld.unique_id, hld.new_status, hld.when_key, UNIXEPOCH() " \
597 " FROM health_log hl, health_log_detail hld WHERE " \
598 " hl.host_id = @host_id AND hld.health_log_id = hl.health_log_id AND hld.transition_id = hl.last_transition_id"
599
600 #define SQL_DELETE_HOST_ALERT_VERSION_TABLE \
601 "DELETE FROM alert_version WHERE health_log_id IN (SELECT health_log_id FROM health_log WHERE host_id = @host_id)"
602
603 void rebuild_host_alert_version_table(RRDHOST *host)
604 {
605 sqlite3_stmt *res = NULL;
606
607 if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_HOST_ALERT_VERSION_TABLE, &res))
608 return;
609
610 int param = 0;
611 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
612
613 param = 0;
614 int rc = execute_insert(res);
615 if (rc != SQLITE_DONE) {
616 netdata_log_error("Failed to delete the host alert version table");
617 goto done;
618 }
619
620 SQLITE_FINALIZE(res);
621 if (!PREPARE_STATEMENT(db_meta, SQL_REBUILD_HOST_ALERT_VERSION_TABLE, &res))
622 return;
623
624 param = 0;
625 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
626
627 param = 0;
628 rc = execute_insert(res);
629 if (rc != SQLITE_DONE)
630 netdata_log_error("Failed to rebuild the host alert version table");
631
632 done:
633 REPORT_BIND_FAIL(res, param);
634 SQLITE_FINALIZE(res);
635 }
636
637 #define SQL_PROCESS_ALERT_PENDING_QUEUE \
638 "SELECT health_log_id, unique_id, status, rowid" \
639 " FROM alert_queue WHERE host_id = @host_id AND date_scheduled <= UNIXEPOCH() ORDER BY rowid ASC"
640
641 bool process_alert_pending_queue(RRDHOST *host)
642 {
643 if (!REQUIRE_HEALTH_DB_OPEN())
644 return false;
645
646 static __thread sqlite3_stmt *compiled_res = NULL;
647 sqlite3_stmt *res = NULL;
648
649 if (is_health_thread) {
650 if (!compiled_res) {
651 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &compiled_res))
652 return false;
653 }
654 res = compiled_res;
655 } else {
656 if (!PREPARE_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &res))
657 return false;
658 }
659
660 int param = 0;
661 int added =0, count = 0;
662 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
663
664 param = 0;
665 while (sqlite3_step_monitored(res) == SQLITE_ROW) {
666
667 int64_t health_log_id = sqlite3_column_int64(res, 0);
668 uint32_t unique_id = sqlite3_column_int64(res, 1);
669 RRDCALC_STATUS new_status = sqlite3_column_int(res, 2);
670 int64_t row = sqlite3_column_int64(res, 3);
671
672 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
673 if (aclk_host_config) {
674 int ret = insert_alert_to_submit_queue(host, health_log_id, unique_id, new_status);
675 if (ret == 0)
676 added++;
677 }
678
679 delete_alert_from_pending_queue(host, row);
680
681 count++;
682 }
683
684 if(count)
685 nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Processed %d entries, queued %d", rrdhost_hostname(host), count, added);
686 done:
687 REPORT_BIND_FAIL(res, param);
688 if (is_health_thread)
689 SQLITE_RESET(res);
690 else
691 SQLITE_FINALIZE(res);
692 return added > 0;
693 }
694
695 void aclk_push_alert_events_for_all_hosts(void)
696 {
697 RRDHOST *host;
698
699 sqlite3_stmt *res = NULL; // used to scan pending alerts to send
700 sqlite3_stmt *res_version = NULL; // used to update the alert version
701 dfe_start_reentrant(rrdhost_root_index, host) {
702 if (!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS) ||
703 rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
704 continue;
705
706 rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
707
708 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
709 if (!aclk_host_config || false == aclk_host_config->stream_alerts || rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {
710 (void)process_alert_pending_queue(host);
711 commit_alert_events(host);
712 continue;
713 }
714
715 if (aclk_host_config->send_snapshot) {
716 rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
717 if (aclk_host_config->send_snapshot == 1)
718 continue;
719 (void)process_alert_pending_queue(host);
720 commit_alert_events(host);
721 rebuild_host_alert_version_table(host);
722 send_alert_snapshot_to_cloud(host);
723 aclk_host_config->snapshot_count++;
724 aclk_host_config->send_snapshot = 0;
725 }
726 else
727 aclk_push_alert_event(host, &res, &res_version);
728 }
729 dfe_done(host);
730 SQLITE_FINALIZE(res);
731 SQLITE_FINALIZE(res_version);
732 }
733
734 #define SQL_SELECT_ALERT_HASH_CLOUD "SELECT 1 FROM alert_hash_cloud WHERE hash_id = @hash_id"
735 #define SQL_INSERT_ALERT_HASH_CLOUD "INSERT OR IGNORE INTO alert_hash_cloud (hash_id) VALUES (@hash_id)"
736
737 void alert_hash_mark_sent(nd_uuid_t *hash_id)
738 {
739 if (!hash_id)
740 return;
741
742 static __thread sqlite3_stmt *compiled_res = NULL;
743 sqlite3_stmt *res = NULL;
744
745 if (is_health_thread) {
746 if (!compiled_res) {
747 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_INSERT_ALERT_HASH_CLOUD, &compiled_res))
748 return;
749 }
750 res = compiled_res;
751 } else {
752 if (!PREPARE_STATEMENT(db_meta, SQL_INSERT_ALERT_HASH_CLOUD, &res))
753 return;
754 }
755
756 int param = 0;
757 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, hash_id, sizeof(*hash_id), SQLITE_STATIC));
758
759 param = 0;
760 (void)sqlite3_step_monitored(res);
761
762 done:
763 REPORT_BIND_FAIL(res, param);
764 if (is_health_thread)
765 SQLITE_RESET(res);
766 else
767 SQLITE_FINALIZE(res);
768 }
769
770 bool alert_hash_has_transitioned(nd_uuid_t *hash_id)
771 {
772 if (!hash_id)
773 return false;
774
775 static __thread sqlite3_stmt *compiled_res = NULL;
776 sqlite3_stmt *res = NULL;
777
778 if (is_health_thread) {
779 if (!compiled_res) {
780 if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_ALERT_HASH_CLOUD, &compiled_res))
781 return false;
782 }
783 res = compiled_res;
784 } else {
785 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_HASH_CLOUD, &res))
786 return false;
787 }
788
789 int param = 0;
790 bool found = true;
791 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, hash_id, sizeof(*hash_id), SQLITE_STATIC));
792
793 param = 0;
794 found = (sqlite3_step_monitored(res) == SQLITE_ROW);
795
796 done:
797 REPORT_BIND_FAIL(res, param);
798 if (is_health_thread)
799 SQLITE_RESET(res);
800 else
801 SQLITE_FINALIZE(res);
802 return found;
803 }
804
805 void aclk_send_alert_configuration(char *config_hash)
806 {
807 if (unlikely(!config_hash))
808 return;
809
810 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&localhost->aclk_host_config, __ATOMIC_ACQUIRE);
811
812 if (unlikely(!aclk_host_config))
813 return;
814
815 nd_log(NDLS_ACCESS, NDLP_DEBUG,
816 "ACLK REQ [%s (%s)]: Request to send alert config %s.",
817 aclk_host_config->node_id,
818 aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A",
819 config_hash);
820
821 aclk_push_alert_config(aclk_host_config->node_id, config_hash);
822 }
823
824 #define SQL_SELECT_ALERT_CONFIG \
825 "SELECT alarm, template, on_key, class, type, component, os, hosts, plugin," \
826 "module, charts, lookup, every, units, green, red, calc, warn, crit, to_key, exec, delay, repeat, info," \
827 "options, host_labels, p_db_lookup_dimensions, p_db_lookup_method, p_db_lookup_options, p_db_lookup_after," \
828 "p_db_lookup_before, p_update_every, chart_labels, summary FROM alert_hash WHERE hash_id = @hash_id"
829
830 void aclk_push_alert_config_event(char *node_id __maybe_unused, char *config_hash __maybe_unused)
831 {
832 sqlite3_stmt *res = NULL;
833 struct aclk_sync_cfg_t *aclk_host_config;
834
835 RRDHOST *host = rrdhost_find_by_node_id(node_id);
836
837 if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE))) {
838 freez(config_hash);
839 freez(node_id);
840 return;
841 }
842
843 nd_uuid_t hash_uuid;
844 if (uuid_parse(config_hash, hash_uuid)) {
845 freez(config_hash);
846 freez(node_id);
847 return;
848 }
849
850 if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_CONFIG, &res)) {
851 freez(config_hash);
852 freez(node_id);
853 return;
854 }
855
856 int param = 0;
857 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &hash_uuid , sizeof(hash_uuid), SQLITE_STATIC));
858
859 struct aclk_alarm_configuration alarm_config;
860 struct provide_alarm_configuration p_alarm_config;
861 p_alarm_config.cfg_hash = NULL;
862
863 param = 0;
864 if (sqlite3_step_monitored(res) == SQLITE_ROW) {
865 alarm_config.alarm = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
866 alarm_config.tmpl = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
867 alarm_config.on_chart = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
868 alarm_config.classification = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
869 alarm_config.type = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
870 alarm_config.component = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
871 alarm_config.os = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
872 alarm_config.hosts = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
873 alarm_config.plugin = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
874 alarm_config.module = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
875 alarm_config.charts = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
876 alarm_config.lookup = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
877 alarm_config.every = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
878 alarm_config.units = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
879 alarm_config.green = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
880 alarm_config.red = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
881 alarm_config.calculation_expr = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
882 alarm_config.warning_expr = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
883 alarm_config.critical_expr = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
884 alarm_config.recipient = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
885 alarm_config.exec = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
886 alarm_config.delay = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
887 alarm_config.repeat = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
888 alarm_config.info = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
889 alarm_config.options = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++);
890 alarm_config.host_labels = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++); // Current param 25
891
892 alarm_config.p_db_lookup_dimensions = NULL;
893 alarm_config.p_db_lookup_method = NULL;
894 alarm_config.p_db_lookup_options = NULL;
895 alarm_config.p_db_lookup_after = 0;
896 alarm_config.p_db_lookup_before = 0;
897
898 if (sqlite3_column_bytes(res, 29) > 0) {
899
900 alarm_config.p_db_lookup_dimensions = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++); // Current param 26
901 alarm_config.p_db_lookup_method = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param++); // Current param 27
902 if (param != 28)
903 netdata_log_error("aclk_push_alert_config_event: Unexpected param number %d", param);
904
905 BUFFER *tmp_buf = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
906 rrdr_options_to_buffer(tmp_buf, sqlite3_column_int(res, 28));
907 alarm_config.p_db_lookup_options = strdupz((char *)buffer_tostring(tmp_buf));
908 buffer_free(tmp_buf);
909
910 alarm_config.p_db_lookup_after = sqlite3_column_int(res, 29);
911 alarm_config.p_db_lookup_before = sqlite3_column_int(res, 30);
912 }
913
914 alarm_config.p_update_every = sqlite3_column_int(res, 31);
915
916 alarm_config.chart_labels = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, 32);
917 alarm_config.summary = SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, 33);
918
919 p_alarm_config.cfg_hash = strdupz((char *) config_hash);
920 p_alarm_config.cfg = alarm_config;
921 }
922
923 param = 0;
924
925 if (likely(p_alarm_config.cfg_hash)) {
926 nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK RES [%s (%s)]: Sent alert config %s.",
927 aclk_host_config->node_id,
928 aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A", config_hash);
929 aclk_send_provide_alarm_cfg(&p_alarm_config);
930 alert_hash_mark_sent(&hash_uuid);
931 freez(p_alarm_config.cfg_hash);
932 destroy_aclk_alarm_configuration(&alarm_config);
933 }
934 else
935 nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK STA [%s (%s)]: Alert config for %s not found.",
936 aclk_host_config->node_id,
937 aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A", config_hash);
938
939 done:
940 REPORT_BIND_FAIL(res, param);
941 SQLITE_FINALIZE(res);
942 freez(config_hash);
943 freez(node_id);
944 }
945
946 #define SQL_ALERT_VERSION_CALC \
947 "SELECT SUM(version) FROM health_log hl, alert_version av" \
948 " WHERE hl.host_id = @host_uuid AND hl.health_log_id = av.health_log_id AND av.status <> -2"
949
950 uint64_t calculate_node_alert_version(RRDHOST *host)
951 {
952 sqlite3_stmt *res = NULL;
953
954 if (!PREPARE_STATEMENT(db_meta, SQL_ALERT_VERSION_CALC, &res))
955 return 0;
956
957 uint64_t version = 0;
958 int param = 0;
959 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
960
961 param = 0;
962 while (sqlite3_step_monitored(res) == SQLITE_ROW) {
963 version = (uint64_t)sqlite3_column_int64(res, 0);
964 }
965
966 done:
967 REPORT_BIND_FAIL(res, param);
968 SQLITE_FINALIZE(res);
969 return version;
970 }
971
972 static void schedule_alert_snapshot_if_needed(struct aclk_sync_cfg_t *aclk_host_config, uint64_t cloud_version)
973 {
974 if (cloud_version == 1) {
975 nd_log(
976 NDLS_ACCESS,
977 NDLP_NOTICE,
978 "Cloud requested to skip alert version verification for host \"%s\", node \"%s\"",
979 rrdhost_hostname(aclk_host_config->host),
980 aclk_host_config->node_id);
981 return;
982 }
983
984 uint64_t local_version = calculate_node_alert_version(aclk_host_config->host);
985 if (local_version != cloud_version) {
986 nd_log(
987 NDLS_ACCESS,
988 NDLP_NOTICE,
989 "Scheduling alert snapshot for host \"%s\", node \"%s\" (version: cloud %llu, local %llu)",
990 rrdhost_hostname(aclk_host_config->host),
991 aclk_host_config->node_id,
992 (long long unsigned)cloud_version,
993 (long long unsigned)local_version);
994
995 aclk_host_config->send_snapshot = 1;
996 rrdhost_flag_set(aclk_host_config->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
997 }
998 else
999 nd_log(
1000 NDLS_ACCESS,
1001 NDLP_DEBUG,
1002 "Alert check on \"%s\", node \"%s\" (version: cloud %llu, local %llu)",
1003 rrdhost_hostname(aclk_host_config->host),
1004 aclk_host_config->node_id,
1005 (unsigned long long)cloud_version,
1006 (unsigned long long)local_version);
1007 aclk_host_config->checkpoint_count++;
1008 }
1009
1010 #define SQL_COUNT_SNAPSHOT_ENTRIES \
1011 "SELECT COUNT(1) FROM alert_version av, health_log hl " \
1012 "WHERE hl.host_id = @host_id AND hl.health_log_id = av.health_log_id AND av.status <> -2"
1013
1014 static int calculate_alert_snapshot_entries(nd_uuid_t *host_uuid)
1015 {
1016 int count = 0;
1017
1018 sqlite3_stmt *res = NULL;
1019
1020 if (!PREPARE_STATEMENT(db_meta, SQL_COUNT_SNAPSHOT_ENTRIES, &res))
1021 return 0;
1022
1023 int param = 0;
1024 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_uuid, sizeof(*host_uuid), SQLITE_STATIC));
1025
1026 param = 0;
1027 int rc = sqlite3_step_monitored(res);
1028 if (rc == SQLITE_ROW)
1029 count = sqlite3_column_int(res, 0);
1030 else
1031 error_report("Failed to select snapshot count");
1032
1033 done:
1034 REPORT_BIND_FAIL(res, param);
1035 SQLITE_FINALIZE(res);
1036
1037 return count;
1038 }
1039
1040 #define SQL_GET_SNAPSHOT_ENTRIES \
1041 " SELECT 0, hld.unique_id, hld.alarm_id, hl.config_hash_id, hld.updated_by_id, hld.when_key, " \
1042 " hld.duration, hld.non_clear_duration, hld.flags, hld.exec_run_timestamp, hld.delay_up_to_timestamp, hl.name, " \
1043 " hl.chart, hl.exec, hl.recipient, ah.source, hl.units, hld.info, hld.exec_code, hld.new_status, " \
1044 " hld.old_status, hld.delay, hld.new_value, hld.old_value, hld.last_repeat, hl.chart_context, hld.transition_id, " \
1045 " hld.alarm_event_id, hl.chart_name, hld.summary, hld.health_log_id, av.version " \
1046 " FROM health_log hl, alert_hash ah, health_log_detail hld, alert_version av " \
1047 " WHERE hl.config_hash_id = ah.hash_id" \
1048 " AND hl.host_id = @host_id AND hl.health_log_id = hld.health_log_id " \
1049 " AND hld.health_log_id = av.health_log_id AND av.unique_id = hld.unique_id AND av.status <> -2"
1050
1051 #define ALARM_EVENTS_PER_CHUNK 1000
1052 void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)
1053 {
1054 struct aclk_sync_cfg_t *aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE);
1055
1056 if (unlikely(!host)) {
1057 nd_log(NDLS_ACCESS, NDLP_WARNING, "AC [%s (N/A)]: Node id not found", aclk_host_config->node_id);
1058 return;
1059 }
1060
1061 CLAIM_ID claim_id = claim_id_get();
1062 if (unlikely(!claim_id_is_set(claim_id)))
1063 return;
1064
1065 // Check the database for this node to see how many alerts we will need to put in the snapshot
1066 int cnt = calculate_alert_snapshot_entries(&host->host_id.uuid);
1067 if (!cnt)
1068 return;
1069
1070 sqlite3_stmt *res = NULL;
1071 if (!PREPARE_STATEMENT(db_meta, SQL_GET_SNAPSHOT_ENTRIES, &res))
1072 return;
1073
1074 int param = 0;
1075 SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_id.uuid, sizeof(host->host_id.uuid), SQLITE_STATIC));
1076
1077 nd_uuid_t local_snapshot_uuid;
1078 char snapshot_uuid_str[UUID_STR_LEN];
1079 uuid_generate_random(local_snapshot_uuid);
1080 uuid_unparse_lower(local_snapshot_uuid, snapshot_uuid_str);
1081 char *snapshot_uuid = &snapshot_uuid_str[0];
1082
1083 nd_log(NDLS_ACCESS, NDLP_DEBUG,
1084 "ACLK REQ [%s (%s)]: Sending %d alerts snapshot, snapshot_uuid %s",
1085 aclk_host_config->node_id, rrdhost_hostname(host),
1086 cnt, snapshot_uuid);
1087
1088 uint32_t chunks;
1089 chunks = (cnt / ALARM_EVENTS_PER_CHUNK) + (cnt % ALARM_EVENTS_PER_CHUNK != 0);
1090
1091 alarm_snapshot_proto_ptr_t snapshot_proto = NULL;
1092 struct alarm_snapshot alarm_snap;
1093 struct alarm_log_entry alarm_log;
1094
1095 alarm_snap.node_id = aclk_host_config->node_id;
1096 alarm_snap.claim_id = claim_id.str;
1097 alarm_snap.snapshot_uuid = snapshot_uuid;
1098 alarm_snap.chunks = chunks;
1099 alarm_snap.chunk = 1;
1100
1101 alarm_log.node_id = aclk_host_config->node_id;
1102 alarm_log.claim_id = claim_id.str;
1103
1104 cnt = 0;
1105 param = 0;
1106 uint64_t version = 0;
1107 int total_count = 0;
1108 while (sqlite3_step_monitored(res) == SQLITE_ROW) {
1109 cnt++;
1110 total_count++;
1111
1112 if (!snapshot_proto)
1113 snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
1114
1115 health_alarm_log_populate(&alarm_log, res, host, NULL);
1116
1117 add_alarm_log_entry2snapshot(snapshot_proto, &alarm_log);
1118 version += alarm_log.version;
1119
1120 if (cnt == ALARM_EVENTS_PER_CHUNK) {
1121 if (aclk_online_for_alerts())
1122 aclk_send_alarm_snapshot(snapshot_proto);
1123 cnt = 0;
1124 if (alarm_snap.chunk < chunks) {
1125 alarm_snap.chunk++;
1126 snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
1127 }
1128 }
1129 destroy_alarm_log_entry(&alarm_log);
1130 }
1131 if (cnt)
1132 aclk_send_alarm_snapshot(snapshot_proto);
1133
1134 nd_log(
1135 NDLS_ACCESS,
1136 NDLP_DEBUG,
1137 "ACLK REQ [%s (%s)]: Created snapshot %s with %d alerts (version = %llu)",
1138 aclk_host_config->node_id,
1139 rrdhost_hostname(host),
1140 snapshot_uuid,
1141 total_count,
1142 (long long unsigned)version);
1143
1144 done:
1145 REPORT_BIND_FAIL(res, param);
1146 SQLITE_FINALIZE(res);
1147 }
1148
1149 // Start streaming alerts
1150 void aclk_start_alert_streaming(char *node_id, uint64_t cloud_version)
1151 {
1152 nd_uuid_t node_uuid;
1153
1154 if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))
1155 return;
1156
1157 struct aclk_sync_cfg_t *aclk_host_config;
1158 RRDHOST *host = rrdhost_find_by_node_id(node_id);
1159
1160 if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE))) {
1161 nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, invalid node.", node_id);
1162 return;
1163 }
1164
1165 if (unlikely(!host->health.enabled)) {
1166 nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, health is disabled.", node_id);
1167 return;
1168 }
1169
1170 nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id,
1171 aclk_host_config->host ? rrdhost_hostname(aclk_host_config->host) : "N/A");
1172 schedule_alert_snapshot_if_needed(aclk_host_config, cloud_version);
1173 aclk_host_config->stream_alerts = true;
1174 }
1175
1176 // Do checkpoint alert version check
1177 void aclk_alert_version_check(char *node_id, char *claim_id, uint64_t cloud_version)
1178 {
1179 nd_uuid_t node_uuid;
1180
1181 if (unlikely(!node_id || !claim_id || !is_agent_claimed() || uuid_parse(node_id, node_uuid)))
1182 return;
1183
1184 CLAIM_ID agent_claim_id = claim_id_get();
1185 if (claim_id && claim_id_is_set(agent_claim_id) && strcmp(agent_claim_id.str, claim_id) != 0) {
1186 nd_log(NDLS_ACCESS, NDLP_NOTICE,
1187 "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT VALIDATION REQUEST RECEIVED WITH INVALID CLAIM ID",
1188 node_id);
1189 return;
1190 }
1191
1192 struct aclk_sync_cfg_t *aclk_host_config;
1193 RRDHOST *host = rrdhost_find_by_node_id(node_id);
1194
1195 if (!host || !(aclk_host_config = __atomic_load_n(&host->aclk_host_config, __ATOMIC_ACQUIRE)))
1196 nd_log(NDLS_ACCESS, NDLP_NOTICE,
1197 "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT VALIDATION REQUEST RECEIVED FOR INVALID NODE",
1198 node_id);
1199 else
1200 schedule_alert_snapshot_if_needed(aclk_host_config, cloud_version);
1201 }