5
6
#ifdef ENABLE_ACLK
7
#include "../../aclk/aclk_alarm_api.h"
8
-#endif
8
9
#define SQLITE3_COLUMN_STRDUPZ_OR_NULL(res, param) \
10
({ \
12
sqlite3_column_bytes((res), (_param)) ? strdupz((char *)sqlite3_column_text((res), (_param))) : NULL; \
13
})
14
16
-#define SQL_UPDATE_FILTERED_ALERT \
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, int64_t unique_id, char *uuid_str)
21
-{
22
- sqlite3_stmt *res = NULL;
23
-
24
- char sql[ACLK_SYNC_QUERY_SIZE];
25
- snprintfz(sql, sizeof(sql) - 1, SQL_UPDATE_FILTERED_ALERT, uuid_str);
26
-
27
- if (!PREPARE_STATEMENT(db_meta, sql, &res))
28
- return;
29
-
30
- int param = 0;
31
- SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, ae->unique_id));
32
- SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));
33
-
34
- param = 0;
35
- if (likely(sqlite3_step_monitored(res) == SQLITE_DONE))
36
- ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
37
-
38
-done:
39
- REPORT_BIND_FAIL(res, param);
40
- SQLITE_FINALIZE(res);
41
-}
42
-
15
#define SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID \
16
"SELECT hld.unique_id FROM health_log hl, alert_hash ah, health_log_detail hld " \
17
"WHERE hld.unique_id = @unique_id AND hl.config_hash_id = ah.hash_id AND hld.health_log_id = hl.health_log_id " \
19
20
static inline bool is_event_from_alert_variable_config(int64_t unique_id, nd_uuid_t *host_id)
21
{
50
- sqlite3_stmt *res = NULL;
22
+ static __thread sqlite3_stmt *res = NULL;
23
52
- if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &res))
24
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_VARIABLE_ALERT_BY_UNIQUE_ID, &res))
25
return false;
26
27
bool ret = false;
35
36
done:
37
REPORT_BIND_FAIL(res, param);
66
- SQLITE_FINALIZE(res);
38
+ SQLITE_RESET(res);
39
return ret;
40
}
41
42
#define MAX_REMOVED_PERIOD 604800 //a week
43
72
-//decide if some events should be sent or not
73
-#define SQL_SELECT_ALERT_BY_ID \
74
- "SELECT hld.new_status, hl.config_hash_id, hld.unique_id FROM health_log hl, aclk_alert_%s aa, health_log_detail hld " \
75
- "WHERE hl.host_id = @host_id AND hld.unique_id = aa.filtered_alert_unique_id " \
76
- "AND hld.alarm_id = @alarm_id AND hl.health_log_id = hld.health_log_id " \
77
- "ORDER BY hld.rowid DESC LIMIT 1"
44
+#define SQL_UPDATE_ALERT_VERSION_TRANSITION \
45
+ "UPDATE alert_version SET unique_id = @unique_id WHERE health_log_id = @health_log_id"
46
79
-static bool should_send_to_cloud(RRDHOST *host, ALARM_ENTRY *ae)
47
+static void update_alert_version_transition(int64_t health_log_id, int64_t unique_id)
48
{
81
- sqlite3_stmt *res = NULL;
49
+ static __thread sqlite3_stmt *res = NULL;
50
83
- if (ae->new_status == RRDCALC_STATUS_UNINITIALIZED ||
84
- (ae->new_status == RRDCALC_STATUS_REMOVED &&
85
- !(ae->old_status == RRDCALC_STATUS_WARNING || ae->old_status == RRDCALC_STATUS_CRITICAL)))
86
- return 0;
51
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION_TRANSITION, &res))
52
+ return;
53
88
- if (unlikely(uuid_is_null(ae->config_hash_id) || !host->aclk_config))
89
- return 0;
54
+ int param = 0;
55
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));
56
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));
57
+
58
+ param = 0;
59
+ int rc = sqlite3_step_monitored(res);
60
+ if (rc != SQLITE_DONE)
61
+ error_report("Failed to update alert_version to latest transition");
62
+
63
+done:
64
+ REPORT_BIND_FAIL(res, param);
65
+ SQLITE_RESET(res);
66
+}
67
91
- char sql[ACLK_SYNC_QUERY_SIZE];
68
+//decide if some events should be sent or not
69
+
70
+#define SQL_SELECT_LAST_ALERT_STATUS "SELECT status FROM alert_version WHERE health_log_id = @health_log_id "
71
93
- //get the previous sent event of this alarm_id
94
- //base the search on the last filtered event
95
- snprintfz(sql, sizeof(sql) - 1, SQL_SELECT_ALERT_BY_ID, host->aclk_config->uuid_str);
72
+static bool cloud_status_matches(int64_t health_log_id, RRDCALC_STATUS status)
73
+{
74
+ static __thread sqlite3_stmt *res = NULL;
75
97
- if (!PREPARE_STATEMENT(db_meta, sql, &res))
76
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_SELECT_LAST_ALERT_STATUS, &res))
77
return true;
78
79
bool send = false;
80
81
int param = 0;
103
- SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
104
- SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, (int) ae->alarm_id));
82
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, health_log_id));
83
84
param = 0;
85
int rc = sqlite3_step_monitored(res);
86
if (likely(rc == SQLITE_ROW)) {
109
- nd_uuid_t config_hash_id;
110
- RRDCALC_STATUS status = (RRDCALC_STATUS)sqlite3_column_int(res, 0);
111
-
112
- if (sqlite3_column_type(res, 1) != SQLITE_NULL)
113
- uuid_copy(config_hash_id, *((nd_uuid_t *)sqlite3_column_blob(res, 1)));
114
-
115
- int64_t unique_id = sqlite3_column_int64(res, 2);
116
-
117
- if (ae->new_status != (RRDCALC_STATUS)status || !uuid_eq(ae->config_hash_id, config_hash_id))
118
- send = true;
119
- else
120
- update_filtered(ae, unique_id, host->aclk_config->uuid_str);
121
- } else
122
- send = true;
87
+ RRDCALC_STATUS current_status = (RRDCALC_STATUS)sqlite3_column_int(res, 0);
88
+ send = (current_status == status);
89
+ }
90
91
done:
92
REPORT_BIND_FAIL(res, param);
126
- SQLITE_FINALIZE(res);
93
+ SQLITE_RESET(res);
94
return send;
95
}
96
97
#define SQL_QUEUE_ALERT_TO_CLOUD \
131
- "INSERT INTO aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) " \
132
- "VALUES (@alert_unique_id, UNIXEPOCH(), @alert_unique_id) ON CONFLICT (alert_unique_id) DO NOTHING"
133
-
134
-void sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, bool skip_filter)
98
+ "INSERT INTO aclk_queue (host_id, health_log_id, unique_id, date_created)" \
99
+ " VALUES (@host_id, @health_log_id, @unique_id, UNIXEPOCH())" \
100
+ " ON CONFLICT(host_id, health_log_id) DO UPDATE SET unique_id=excluded.unique_id, " \
101
+ " date_created=excluded.date_created"
102
+
103
+//
104
+// Attempt to insert an alert to the submit queue to reach the cloud
105
+//
106
+// The alert will NOT be added in the submit queue if
107
+// - Cloud is already aware of the alert status
108
+// - The transition refers to a variable
109
+//
110
+static int insert_alert_to_submit_queue(RRDHOST *host, int64_t health_log_id, uint32_t unique_id, RRDCALC_STATUS status)
111
{
136
- sqlite3_stmt *res = NULL;
137
- char sql[ACLK_SYNC_QUERY_SIZE];
112
+ static __thread sqlite3_stmt *res = NULL;
113
139
- if (!service_running(SERVICE_ACLK))
140
- return;
114
+ if (cloud_status_matches(health_log_id, status)) {
115
+ update_alert_version_transition(health_log_id, unique_id);
116
+ return 1;
117
+ }
118
142
- if (!claimed() || ae->flags & HEALTH_ENTRY_FLAG_ACLK_QUEUED)
143
- return;
119
+ if (is_event_from_alert_variable_config(unique_id, &host->host_uuid))
120
+ return 2;
121
145
- if (false == skip_filter && !should_send_to_cloud(host, ae))
146
- return;
122
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_QUEUE_ALERT_TO_CLOUD, &res))
123
+ return -1;
124
148
- if (is_event_from_alert_variable_config(ae->unique_id, &host->host_uuid))
149
- return;
125
+ int param = 0;
126
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
127
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));
128
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, (int64_t) unique_id));
129
151
- snprintfz(sql, sizeof(sql) - 1, SQL_QUEUE_ALERT_TO_CLOUD, host->aclk_config->uuid_str);
130
+ param = 0;
131
+ int rc = execute_insert(res);
132
+ if (unlikely(rc != SQLITE_DONE))
133
+ error_report("Failed to insert alert in the submit queue %"PRIu32", rc = %d", unique_id, rc);
134
153
- if (!PREPARE_STATEMENT(db_meta, sql, &res))
154
- return;
135
+done:
136
+ REPORT_BIND_FAIL(res, param);
137
+ SQLITE_RESET(res);
138
+ return 0;
139
+}
140
+
141
+#define SQL_DELETE_QUEUE_ALERT_TO_CLOUD \
142
+ "DELETE FROM aclk_queue WHERE host_id = @host_id AND sequence_id BETWEEN @seq1 AND @seq2"
143
+
144
+//
145
+// Delete a range of alerts from the submit queue (after being sent to the the cloud)
146
+//
147
+static int delete_alert_from_submit_queue(RRDHOST *host, int64_t first_seq_id, int64_t last_seq_id)
148
+{
149
+ static __thread sqlite3_stmt *res = NULL;
150
+
151
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_DELETE_QUEUE_ALERT_TO_CLOUD, &res))
152
+ return -1;
153
154
int param = 0;
157
- SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, ae->unique_id));
155
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
156
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, first_seq_id));
157
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, last_seq_id));
158
159
param = 0;
160
- int rc = execute_insert(res);
161
- if (unlikely(rc == SQLITE_DONE)) {
162
- ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
163
- rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
164
- } else
165
- error_report("Failed to store alert event %"PRIu32", rc = %d", ae->unique_id, rc);
160
+ int rc = sqlite3_step_monitored(res);
161
+ if (rc != SQLITE_DONE)
162
+ error_report("Failed to delete submitted to ACLK");
163
164
done:
165
REPORT_BIND_FAIL(res, param);
169
- SQLITE_FINALIZE(res);
166
+ SQLITE_RESET(res);
167
+ return 0;
168
}
169
170
int rrdcalc_status_to_proto_enum(RRDCALC_STATUS status)
171
{
174
-#ifdef ENABLE_ACLK
172
+
173
switch(status) {
174
case RRDCALC_STATUS_REMOVED:
175
return ALARM_STATUS_REMOVED;
189
default:
190
return ALARM_STATUS_UNKNOWN;
191
}
194
-#else
195
- UNUSED(status);
196
- return 1;
197
-#endif
192
}
193
194
static inline char *sqlite3_uuid_unparse_strdupz(sqlite3_stmt *res, int iCol) {
213
return strdupz(ret);
214
}
215
216
+#define SQL_UPDATE_ALERT_VERSION \
217
+ "INSERT INTO alert_version (health_log_id, unique_id, status, version, date_submitted)" \
218
+ " VALUES (@health_log_id, @unique_id, @status, @version, UNIXEPOCH())" \
219
+ " ON CONFLICT(health_log_id) DO UPDATE SET status = excluded.status, version = excluded.version, " \
220
+ " unique_id=excluded.unique_id, date_submitted=excluded.date_submitted"
221
+
222
+//
223
+// Store a new alert transition along with the version after sending to the cloud
224
+// - Update an existing alert with the updated version, status, transition and date submitted
225
+//
226
+static void sql_update_alert_version(int64_t health_log_id, int64_t unique_id, RRDCALC_STATUS status, uint64_t version)
227
+{
228
+ static __thread sqlite3_stmt *res = NULL;
229
+
230
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_UPDATE_ALERT_VERSION, &res))
231
+ return;
232
223
-static void aclk_push_alert_event(struct aclk_sync_cfg_t *wc __maybe_unused)
233
+ int param = 0;
234
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, health_log_id));
235
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, unique_id));
236
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int(res, ++param, status));
237
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, version));
238
+
239
+ param = 0;
240
+ int rc = sqlite3_step_monitored(res);
241
+ if (rc != SQLITE_DONE)
242
+ error_report("Failed to execute sql_update_alert_version");
243
+
244
+done:
245
+ REPORT_BIND_FAIL(res, param);
246
+ SQLITE_RESET(res);
247
+}
248
+
249
+#define SQL_SELECT_ALERT_TO_DUMMY \
250
+ "SELECT aq.sequence_id, hld.unique_id, hld.when_key, hld.new_status, hld.health_log_id" \
251
+ " FROM health_log hl, aclk_queue aq, alert_hash ah, health_log_detail hld" \
252
+ " WHERE hld.unique_id = aq.unique_id AND hl.config_hash_id = ah.hash_id" \
253
+ " AND hl.host_id = @host_id AND aq.host_id = hl.host_id AND hl.health_log_id = hld.health_log_id" \
254
+ " ORDER BY aq.sequence_id ASC"
255
+
256
+//
257
+// Check all queued alerts for a host and commit them as if they have been send to the cloud
258
+// this will produce new versions as needed. We need this because we are about to send a
259
+// a snapshot so we can include the latest transition.
260
+//
261
+static void commit_alert_events(RRDHOST *host)
262
{
225
-#ifdef ENABLE_ACLK
226
- int rc;
263
+ sqlite3_stmt *res = NULL;
264
228
- if (unlikely(!wc->alert_updates)) {
229
- nd_log(NDLS_ACCESS, NDLP_NOTICE,
230
- "ACLK STA [%s (%s)]: Ignoring alert push event, updates have been turned off for this node.",
231
- wc->node_id,
232
- wc->host ? rrdhost_hostname(wc->host) : "N/A");
265
+ if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_DUMMY, &res))
266
return;
267
+
268
+ int param = 0;
269
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
270
+
271
+ int64_t first_sequence_id = 0;
272
+ int64_t last_sequence_id = 0;
273
+
274
+ param = 0;
275
+ while (sqlite3_step_monitored(res) == SQLITE_ROW) {
276
+
277
+ last_sequence_id = sqlite3_column_int64(res, 0);
278
+ if (first_sequence_id == 0)
279
+ first_sequence_id = last_sequence_id;
280
+
281
+ int64_t unique_id = sqlite3_column_int(res, 1);
282
+ int64_t version = sqlite3_column_int64(res, 2);
283
+ RRDCALC_STATUS status = (RRDCALC_STATUS)sqlite3_column_int(res, 3);
284
+ int64_t health_log_id = sqlite3_column_int64(res, 4);
285
+
286
+ sql_update_alert_version(health_log_id, unique_id, status, version);
287
}
288
289
+ if (first_sequence_id)
290
+ delete_alert_from_submit_queue(host, first_sequence_id, last_sequence_id);
291
+
292
+done:
293
+ REPORT_BIND_FAIL(res, param);
294
+ SQLITE_FINALIZE(res);
295
+}
296
+
297
+typedef enum {
298
+ SEQUENCE_ID,
299
+ UNIQUE_ID,
300
+ ALARM_ID,
301
+ CONFIG_HASH_ID,
302
+ UPDATED_BY_ID,
303
+ WHEN_KEY,
304
+ DURATION,
305
+ NON_CLEAR_DURATION,
306
+ FLAGS,
307
+ EXEC_RUN_TIMESTAMP,
308
+ DELAY_UP_TO_TIMESTAMP,
309
+ NAME,
310
+ CHART,
311
+ EXEC,
312
+ RECIPIENT,
313
+ SOURCE,
314
+ UNITS,
315
+ INFO,
316
+ EXEC_CODE,
317
+ NEW_STATUS,
318
+ OLD_STATUS,
319
+ DELAY,
320
+ NEW_VALUE,
321
+ OLD_VALUE,
322
+ LAST_REPEAT,
323
+ CHART_CONTEXT,
324
+ TRANSITION_ID,
325
+ ALARM_EVENT_ID,
326
+ CHART_NAME,
327
+ SUMMARY,
328
+ HEALTH_LOG_ID,
329
+ VERSION
330
+} HealthLogDetails;
331
+
332
+void health_alarm_log_populate(
333
+ struct alarm_log_entry *alarm_log,
334
+ sqlite3_stmt *res,
335
+ RRDHOST *host,
336
+ RRDCALC_STATUS *status)
337
+{
338
+ char old_value_string[100 + 1];
339
+ char new_value_string[100 + 1];
340
+
341
+ RRDCALC_STATUS current_status = (RRDCALC_STATUS)sqlite3_column_int(res, NEW_STATUS);
342
+ if (status)
343
+ *status = current_status;
344
+
345
+ char *source = (char *) sqlite3_column_text(res, SOURCE);
346
+ alarm_log->command = source ? health_edit_command_from_source(source) : strdupz("UNKNOWN=0=UNKNOWN");
347
+
348
+ alarm_log->chart = strdupz((char *) sqlite3_column_text(res, CHART));
349
+ alarm_log->name = strdupz((char *) sqlite3_column_text(res, NAME));
350
+
351
+ alarm_log->when = sqlite3_column_int64(res, WHEN_KEY);
352
+
353
+ alarm_log->config_hash = sqlite3_uuid_unparse_strdupz(res, CONFIG_HASH_ID);
354
+
355
+ alarm_log->utc_offset = host->utc_offset;
356
+ alarm_log->timezone = strdupz(rrdhost_abbrev_timezone(host));
357
+ alarm_log->exec_path = sqlite3_column_bytes(res, EXEC) ?
358
+ strdupz((char *)sqlite3_column_text(res, EXEC)) :
359
+ strdupz((char *)string2str(host->health.health_default_exec));
360
+
361
+ alarm_log->conf_source = source ? strdupz(source) : strdupz("");
362
+
363
+ time_t duration = sqlite3_column_int64(res, DURATION);
364
+ alarm_log->duration = (duration > 0) ? duration : 0;
365
+
366
+ alarm_log->non_clear_duration = sqlite3_column_int64(res, NON_CLEAR_DURATION);
367
+
368
+ alarm_log->status = rrdcalc_status_to_proto_enum(current_status);
369
+ alarm_log->old_status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)sqlite3_column_int64(res, OLD_STATUS));
370
+ alarm_log->delay = sqlite3_column_int64(res, DELAY);
371
+ alarm_log->delay_up_to_timestamp = sqlite3_column_int64(res, DELAY_UP_TO_TIMESTAMP);
372
+ alarm_log->last_repeat = sqlite3_column_int64(res, LAST_REPEAT);
373
+
374
+ uint64_t flags = sqlite3_column_int64(res, FLAGS);
375
+ char *recipient = (char *) sqlite3_column_text(res, RECIPIENT);
376
+ alarm_log->silenced =
377
+ ((flags & HEALTH_ENTRY_FLAG_SILENCED) || (recipient && !strncmp(recipient, "silent", 6))) ? 1 : 0;
378
+
379
+ double value = sqlite3_column_double(res, NEW_VALUE);
380
+ double old_value = sqlite3_column_double(res, OLD_VALUE);
381
+
382
+ alarm_log->value_string =
383
+ sqlite3_column_type(res, NEW_VALUE) == SQLITE_NULL ?
384
+ strdupz((char *)"-") :
385
+ strdupz((char *)format_value_and_unit(
386
+ new_value_string, 100, value, (char *)sqlite3_column_text(res, UNITS), -1));
387
+
388
+ alarm_log->old_value_string =
389
+ sqlite3_column_type(res, OLD_VALUE) == SQLITE_NULL ?
390
+ strdupz((char *)"-") :
391
+ strdupz((char *)format_value_and_unit(
392
+ old_value_string, 100, old_value, (char *)sqlite3_column_text(res, UNITS), -1));
393
+
394
+ alarm_log->value = (!isnan(value)) ? (NETDATA_DOUBLE)value : 0;
395
+ alarm_log->old_value = (!isnan(old_value)) ? (NETDATA_DOUBLE)old_value : 0;
396
+
397
+ alarm_log->updated = (flags & HEALTH_ENTRY_FLAG_UPDATED) ? 1 : 0;
398
+ alarm_log->rendered_info = sqlite3_text_strdupz_empty(res, INFO);
399
+ alarm_log->chart_context = sqlite3_text_strdupz_empty(res, CHART_CONTEXT);
400
+ alarm_log->chart_name = sqlite3_text_strdupz_empty(res, CHART_NAME);
401
+
402
+ alarm_log->transition_id = sqlite3_uuid_unparse_strdupz(res, TRANSITION_ID);
403
+ alarm_log->event_id = sqlite3_column_int64(res, ALARM_EVENT_ID);
404
+ alarm_log->version = sqlite3_column_int64(res, VERSION);
405
+
406
+ alarm_log->summary = sqlite3_text_strdupz_empty(res, SUMMARY);
407
+
408
+ alarm_log->health_log_id = sqlite3_column_int64(res, HEALTH_LOG_ID);
409
+ alarm_log->unique_id = sqlite3_column_int64(res, UNIQUE_ID);
410
+ alarm_log->alarm_id = sqlite3_column_int64(res, ALARM_ID);
411
+ alarm_log->sequence_id = sqlite3_column_int64(res, SEQUENCE_ID);
412
+}
413
+
414
+#define SQL_SELECT_ALERT_TO_PUSH \
415
+ "SELECT aq.sequence_id, hld.unique_id, hld.alarm_id, hl.config_hash_id, hld.updated_by_id, hld.when_key," \
416
+ " hld.duration, hld.non_clear_duration, hld.flags, hld.exec_run_timestamp, hld.delay_up_to_timestamp, hl.name," \
417
+ " hl.chart, hl.exec, hl.recipient, ah.source, hl.units, hld.info, hld.exec_code, hld.new_status," \
418
+ " hld.old_status, hld.delay, hld.new_value, hld.old_value, hld.last_repeat, hl.chart_context, hld.transition_id," \
419
+ " hld.alarm_event_id, hl.chart_name, hld.summary, hld.health_log_id, hld.when_key" \
420
+ " FROM health_log hl, aclk_queue aq, alert_hash ah, health_log_detail hld" \
421
+ " WHERE hld.unique_id = aq.unique_id AND hl.config_hash_id = ah.hash_id" \
422
+ " AND hl.host_id = @host_id AND aq.host_id = hl.host_id AND hl.health_log_id = hld.health_log_id" \
423
+ " ORDER BY aq.sequence_id ASC LIMIT "ACLK_MAX_ALERT_UPDATES
424
+
425
+static void aclk_push_alert_event(RRDHOST *host __maybe_unused)
426
+{
427
+
428
char *claim_id = get_agent_claimid();
237
- if (unlikely(!claim_id))
429
+ if (!claim_id || !host->node_id)
430
return;
431
240
- if (unlikely(!wc->host)) {
432
+ sqlite3_stmt *res = NULL;
433
+
434
+ if (!PREPARE_STATEMENT(db_meta, SQL_SELECT_ALERT_TO_PUSH, &res)) {
435
freez(claim_id);
436
return;
437
}
438
245
- BUFFER *sql = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
246
-
247
- sqlite3_stmt *res = NULL;
439
+ int param = 0;
440
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
441
249
- buffer_sprintf(
250
- sql,
251
- "SELECT aa.sequence_id, hld.unique_id, hld.alarm_id, hl.config_hash_id, hld.updated_by_id, hld.when_key, "
252
- " hld.duration, hld.non_clear_duration, hld.flags, hld.exec_run_timestamp, hld.delay_up_to_timestamp, hl.name, "
253
- " hl.chart, hl.exec, hl.recipient, ha.source, hl.units, hld.info, hld.exec_code, hld.new_status, "
254
- " hld.old_status, hld.delay, hld.new_value, hld.old_value, hld.last_repeat, hl.chart_context, hld.transition_id, "
255
- " hld.alarm_event_id, hl.chart_name, hld.summary "
256
- " FROM health_log hl, aclk_alert_%s aa, alert_hash ha, health_log_detail hld "
257
- " WHERE hld.unique_id = aa.alert_unique_id AND hl.config_hash_id = ha.hash_id AND aa.date_submitted IS NULL "
258
- " AND hl.host_id = @host_id AND hl.health_log_id = hld.health_log_id "
259
- " ORDER BY aa.sequence_id ASC LIMIT "ACLK_MAX_ALERT_UPDATES,
260
- wc->uuid_str);
261
-
262
- if (!PREPARE_STATEMENT(db_meta, buffer_tostring(sql), &res)) {
263
-
264
- BUFFER *sql_fix = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
265
- buffer_sprintf(sql_fix, TABLE_ACLK_ALERT, wc->uuid_str);
266
-
267
- rc = db_execute(db_meta, buffer_tostring(sql_fix));
268
- if (unlikely(rc))
269
- error_report("Failed to create ACLK alert table for host %s", rrdhost_hostname(wc->host));
270
- buffer_free(sql_fix);
271
-
272
- // Try again
273
- if (!PREPARE_STATEMENT(db_meta, buffer_tostring(sql), &res)) {
274
- buffer_free(sql);
275
- freez(claim_id);
276
- return;
277
- }
278
- }
442
+ char node_id_str[UUID_STR_LEN];
443
+ uuid_unparse_lower(*host->node_id, node_id_str);
444
280
- rc = sqlite3_bind_blob(res, 1, &wc->host->host_uuid, sizeof(wc->host->host_uuid), SQLITE_STATIC);
281
- if (unlikely(rc != SQLITE_OK)) {
282
- error_report("Failed to bind host_id for pushing alert event.");
283
- goto done;
284
- }
445
+ struct alarm_log_entry alarm_log;
446
+ alarm_log.node_id = node_id_str;
447
+ alarm_log.claim_id = claim_id;
448
286
- uint64_t first_sequence_id = 0;
287
- uint64_t last_sequence_id = 0;
449
+ int64_t first_id = 0;
450
+ int64_t last_id = 0;
451
452
+ param = 0;
453
+ RRDCALC_STATUS status;
454
+ struct aclk_sync_cfg_t *wc = host->aclk_config;
455
while (sqlite3_step_monitored(res) == SQLITE_ROW) {
290
- struct alarm_log_entry alarm_log;
291
- char old_value_string[100 + 1];
292
- char new_value_string[100 + 1];
293
-
294
- alarm_log.node_id = wc->node_id;
295
- alarm_log.claim_id = claim_id;
296
- alarm_log.chart = strdupz((char *)sqlite3_column_text(res, 12));
297
- alarm_log.name = strdupz((char *)sqlite3_column_text(res, 11));
298
- alarm_log.when = (time_t) sqlite3_column_int64(res, 5);
299
- alarm_log.config_hash = sqlite3_uuid_unparse_strdupz(res, 3);
300
- alarm_log.utc_offset = wc->host->utc_offset;
301
- alarm_log.timezone = strdupz(rrdhost_abbrev_timezone(wc->host));
302
- alarm_log.exec_path = sqlite3_column_bytes(res, 13) > 0 ? strdupz((char *)sqlite3_column_text(res, 13)) :
303
- strdupz((char *)string2str(wc->host->health.health_default_exec));
304
- alarm_log.conf_source = sqlite3_column_bytes(res, 15) > 0 ? strdupz((char *)sqlite3_column_text(res, 15)) : strdupz("");
305
-
306
- char *edit_command = sqlite3_column_bytes(res, 15) > 0 ?
307
- health_edit_command_from_source((char *)sqlite3_column_text(res, 15)) :
308
- strdupz("UNKNOWN=0=UNKNOWN");
309
- alarm_log.command = strdupz(edit_command);
310
-
311
- time_t duration = (time_t) sqlite3_column_int64(res, 6);
312
- alarm_log.duration = (duration > 0) ? duration : 0;
313
- alarm_log.non_clear_duration = (time_t) sqlite3_column_int64(res, 7);
314
- alarm_log.status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS) sqlite3_column_int(res, 19));
315
- alarm_log.old_status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS) sqlite3_column_int(res, 20));
316
- alarm_log.delay = (int) sqlite3_column_int(res, 21);
317
- alarm_log.delay_up_to_timestamp = (time_t) sqlite3_column_int64(res, 10);
318
- alarm_log.last_repeat = (time_t) sqlite3_column_int64(res, 24);
319
- alarm_log.silenced = ((sqlite3_column_int64(res, 8) & HEALTH_ENTRY_FLAG_SILENCED) ||
320
- (sqlite3_column_type(res, 14) != SQLITE_NULL &&
321
- !strncmp((char *)sqlite3_column_text(res, 14), "silent", 6))) ?
322
- 1 :
323
- 0;
324
- alarm_log.value_string =
325
- sqlite3_column_type(res, 22) == SQLITE_NULL ?
326
- strdupz((char *)"-") :
327
- strdupz((char *)format_value_and_unit(
328
- new_value_string, 100, sqlite3_column_double(res, 22), (char *)sqlite3_column_text(res, 16), -1));
329
- alarm_log.old_value_string =
330
- sqlite3_column_type(res, 23) == SQLITE_NULL ?
331
- strdupz((char *)"-") :
332
- strdupz((char *)format_value_and_unit(
333
- old_value_string, 100, sqlite3_column_double(res, 23), (char *)sqlite3_column_text(res, 16), -1));
334
- alarm_log.value = (NETDATA_DOUBLE) sqlite3_column_double(res, 22);
335
- alarm_log.old_value = (NETDATA_DOUBLE) sqlite3_column_double(res, 23);
336
- alarm_log.updated = (sqlite3_column_int64(res, 8) & HEALTH_ENTRY_FLAG_UPDATED) ? 1 : 0;
337
- alarm_log.rendered_info = sqlite3_text_strdupz_empty(res, 17);
338
- alarm_log.chart_context = sqlite3_text_strdupz_empty(res, 25);
339
- alarm_log.transition_id = sqlite3_uuid_unparse_strdupz(res, 26);
340
- alarm_log.event_id = (time_t) sqlite3_column_int64(res, 27);
341
- alarm_log.chart_name = sqlite3_text_strdupz_empty(res, 28);
342
- alarm_log.summary = sqlite3_text_strdupz_empty(res, 29);
343
-
456
+ health_alarm_log_populate(&alarm_log, res, host, &status);
457
aclk_send_alarm_log_entry(&alarm_log);
458
+ wc->alert_count++;
459
346
- if (first_sequence_id == 0)
347
- first_sequence_id = (uint64_t) sqlite3_column_int64(res, 0);
348
-
349
- if (wc->alerts_log_first_sequence_id == 0)
350
- wc->alerts_log_first_sequence_id = (uint64_t) sqlite3_column_int64(res, 0);
460
+ last_id = alarm_log.sequence_id;
461
+ if (first_id == 0)
462
+ first_id = last_id;
463
352
- last_sequence_id = (uint64_t) sqlite3_column_int64(res, 0);
353
- wc->alerts_log_last_sequence_id = (uint64_t) sqlite3_column_int64(res, 0);
464
+ sql_update_alert_version(alarm_log.health_log_id, alarm_log.unique_id, status, alarm_log.version);
465
466
destroy_alarm_log_entry(&alarm_log);
356
- freez(edit_command);
467
}
468
359
- if (first_sequence_id) {
360
- buffer_flush(sql);
361
- buffer_sprintf(
362
- sql,
363
- "UPDATE aclk_alert_%s SET date_submitted=unixepoch() "
364
- "WHERE +date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64,
365
- wc->uuid_str,
366
- first_sequence_id,
367
- last_sequence_id);
368
-
369
- if (unlikely(db_execute(db_meta, buffer_tostring(sql))))
370
- error_report("Failed to mark ACLK alert entries as submitted for host %s", rrdhost_hostname(wc->host));
371
-
469
+ if (first_id) {
470
+ nd_log(
471
+ NDLS_ACCESS,
472
+ NDLP_DEBUG,
473
+ "ACLK RES [%s (%s)]: ALERTS SENT from %ld - %ld",
474
+ node_id_str,
475
+ rrdhost_hostname(host),
476
+ first_id,
477
+ last_id);
478
+
479
+ delete_alert_from_submit_queue(host, first_id, last_id);
480
// Mark to do one more check
373
- rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
374
-
375
- } else {
376
- if (wc->alerts_log_first_sequence_id)
377
- nd_log(NDLS_ACCESS, NDLP_DEBUG,
378
- "ACLK RES [%s (%s)]: ALERTS SENT from %" PRIu64 " to %" PRIu64 "",
379
- wc->node_id,
380
- wc->host ? rrdhost_hostname(wc->host) : "N/A",
381
- wc->alerts_log_first_sequence_id,
382
- wc->alerts_log_last_sequence_id);
383
- wc->alerts_log_first_sequence_id = 0;
384
- wc->alerts_log_last_sequence_id = 0;
481
+ rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
482
}
483
484
done:
485
+ REPORT_BIND_FAIL(res, param);
486
SQLITE_FINALIZE(res);
487
488
freez(claim_id);
391
- buffer_free(sql);
392
-#endif
489
}
490
395
-void aclk_push_alert_events_for_all_hosts(void)
491
+#define SQL_DELETE_PROCESSED_ROWS \
492
+ "DELETE FROM alert_queue WHERE host_id = @host_id AND rowid between @row1 AND @row2"
493
+
494
+static void delete_alert_from_pending_queue(RRDHOST *host, int64_t row1, int64_t row2)
495
{
397
- RRDHOST *host;
496
+ static __thread sqlite3_stmt *res = NULL;
497
399
- dfe_start_reentrant(rrdhost_root_index, host) {
400
- if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) ||
401
- !rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS))
402
- continue;
498
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_DELETE_PROCESSED_ROWS, &res))
499
+ return;
500
404
- rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
501
+ int param = 0;
502
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
503
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, row1));
504
+ SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, row2));
505
406
- struct aclk_sync_cfg_t *wc = host->aclk_config;
407
- if (likely(wc))
408
- aclk_push_alert_event(wc);
409
- }
410
- dfe_done(host);
506
+ param = 0;
507
+ int rc = sqlite3_step_monitored(res);
508
+ if (rc != SQLITE_DONE)
509
+ error_report("Failed to delete processed rows, rc = %d", rc);
510
+
511
+done:
512
+ REPORT_BIND_FAIL(res, param);
513
+ SQLITE_RESET(res);
514
}
515
413
-void sql_queue_existing_alerts_to_aclk(RRDHOST *host)
516
+#define SQL_REBUILD_HOST_ALERT_VERSION_TABLE \
517
+ "INSERT INTO alert_version (health_log_id, unique_id, status, version, date_submitted) " \
518
+ " SELECT hl.health_log_id, hld.unique_id, hld.new_status, hld.when_key, UNIXEPOCH() " \
519
+ " FROM health_log hl, health_log_detail hld WHERE " \
520
+ " hl.host_id = @host_id AND hld.health_log_id = hl.health_log_id AND hld.transition_id = hl.last_transition_id"
521
+
522
+#define SQL_DELETE_HOST_ALERT_VERSION_TABLE \
523
+ "DELETE FROM alert_version WHERE health_log_id IN (SELECT health_log_id FROM health_log WHERE host_id = @host_id)"
524
+
525
+void rebuild_host_alert_version_table(RRDHOST *host)
526
{
527
sqlite3_stmt *res = NULL;
416
- int rc;
528
418
- struct aclk_sync_cfg_t *wc = host->aclk_config;
529
+ if (!PREPARE_STATEMENT(db_meta, SQL_DELETE_HOST_ALERT_VERSION_TABLE, &res))
530
+ return;
531
420
- BUFFER *sql = buffer_create(1024, &netdata_buffers_statistics.buffers_sqlite);
532
+ int param = 0;
533
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
534
422
- rw_spinlock_write_lock(&host->health_log.spinlock);
535
+ param = 0;
536
+ int rc = execute_insert(res);
537
+ if (rc != SQLITE_DONE) {
538
+ netdata_log_error("Failed to delete the host alert version table");
539
+ goto done;
540
+ }
541
+
542
+ SQLITE_FINALIZE(res);
543
+ if (!PREPARE_STATEMENT(db_meta, SQL_REBUILD_HOST_ALERT_VERSION_TABLE, &res))
544
+ return;
545
424
- buffer_sprintf(sql, "DELETE FROM aclk_alert_%s", wc->uuid_str);
425
- if (unlikely(db_execute(db_meta, buffer_tostring(sql))))
426
- goto skip;
546
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
547
+
548
+ param = 0;
549
+ rc = execute_insert(res);
550
+ if (rc != SQLITE_DONE)
551
+ netdata_log_error("Failed to rebuild the host alert version table");
552
+
553
+done:
554
+ REPORT_BIND_FAIL(res, param);
555
+ SQLITE_FINALIZE(res);
556
+}
557
428
- buffer_flush(sql);
558
+#define SQL_PROCESS_ALERT_PENDING_QUEUE \
559
+ "SELECT health_log_id, unique_id, status, rowid" \
560
+ " FROM alert_queue WHERE host_id = @host_id AND date_scheduled <= UNIXEPOCH() ORDER BY rowid ASC"
561
430
- buffer_sprintf(
431
- sql,
432
- "INSERT INTO aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) "
433
- "SELECT hld.unique_id alert_unique_id, unixepoch(), hld.unique_id alert_unique_id FROM health_log_detail hld, health_log hl "
434
- "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 "
435
- "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",
436
- wc->uuid_str);
562
+bool process_alert_pending_queue(RRDHOST *host)
563
+{
564
+ static __thread sqlite3_stmt *res = NULL;
565
438
- if (!PREPARE_STATEMENT(db_meta, buffer_tostring(sql), &res))
439
- goto skip;
566
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_PROCESS_ALERT_PENDING_QUEUE, &res))
567
+ return false;
568
569
int param = 0;
570
+ int added =0, count = 0;
571
SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
572
573
param = 0;
445
- rc = execute_insert(res);
446
- if (unlikely(rc != SQLITE_DONE))
447
- error_report("Failed to queue existing alerts, rc = %d", rc);
448
- else
449
- rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
574
+ int64_t start_row = 0;
575
+ int64_t end_row = 0;
576
+ while (sqlite3_step_monitored(res) == SQLITE_ROW) {
577
+
578
+ int64_t health_log_id = sqlite3_column_int64(res, 0);
579
+ uint32_t unique_id = sqlite3_column_int64(res, 1);
580
+ RRDCALC_STATUS new_status = sqlite3_column_int(res, 2);
581
+ int64_t row = sqlite3_column_int64(res, 3);
582
+
583
+ if (host->aclk_config) {
584
+ int ret = insert_alert_to_submit_queue(host, health_log_id, unique_id, new_status);
585
+ if (ret == 0)
586
+ added++;
587
+ }
588
+
589
+ if (!start_row)
590
+ start_row = row;
591
+ end_row = row;
592
593
+ count++;
594
+ }
595
+
596
+ if (start_row)
597
+ delete_alert_from_pending_queue(host, start_row, end_row);
598
+
599
+ if(count)
600
+ nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Processed %d entries, queued %d", rrdhost_hostname(host), count, added);
601
done:
602
REPORT_BIND_FAIL(res, param);
453
- SQLITE_FINALIZE(res);
603
+ SQLITE_RESET(res);
604
+ return added > 0;
605
+}
606
+
607
+void aclk_push_alert_events_for_all_hosts(void)
608
+{
609
+ RRDHOST *host;
610
+
611
+ // Checking if we shutting down
612
+ if (!service_running(SERVICE_ACLK))
613
+ return;
614
+
615
+ dfe_start_reentrant(rrdhost_root_index, host) {
616
+ if (!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS) ||
617
+ rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
618
+ continue;
619
455
-skip:
456
- rw_spinlock_write_unlock(&host->health_log.spinlock);
457
- buffer_free(sql);
620
+ rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
621
+
622
+ struct aclk_sync_cfg_t *wc = host->aclk_config;
623
+ if (!wc || false == wc->stream_alerts || rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {
624
+ (void)process_alert_pending_queue(host);
625
+ commit_alert_events(host);
626
+ continue;
627
+ }
628
+
629
+ if (wc->send_snapshot) {
630
+ rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
631
+ if (wc->send_snapshot == 1)
632
+ continue;
633
+ (void)process_alert_pending_queue(host);
634
+ commit_alert_events(host);
635
+ rebuild_host_alert_version_table(host);
636
+ send_alert_snapshot_to_cloud(host);
637
+ wc->snapshot_count++;
638
+ wc->send_snapshot = 0;
639
+ }
640
+ else
641
+ aclk_push_alert_event(host);
642
+ }
643
+ dfe_done(host);
644
}
645
460
-void aclk_send_alarm_configuration(char *config_hash)
646
+void aclk_send_alert_configuration(char *config_hash)
647
{
648
if (unlikely(!config_hash))
649
return;
670
671
void aclk_push_alert_config_event(char *node_id __maybe_unused, char *config_hash __maybe_unused)
672
{
487
-#ifdef ENABLE_ACLK
673
sqlite3_stmt *res = NULL;
674
struct aclk_sync_cfg_t *wc;
675
771
SQLITE_FINALIZE(res);
772
freez(config_hash);
773
freez(node_id);
589
-#endif
774
}
775
776
+#define SQL_ALERT_VERSION_CALC \
777
+ "SELECT SUM(version) FROM health_log hl, alert_version av" \
778
+ " WHERE hl.host_id = @host_uuid AND hl.health_log_id = av.health_log_id AND av.status <> -2"
779
593
-// Start streaming alerts
594
-void aclk_start_alert_streaming(char *node_id, bool resets)
780
+static uint64_t calculate_node_alert_version(RRDHOST *host)
781
{
596
- nd_uuid_t node_uuid;
597
-
598
- if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))
599
- return;
600
-
601
- struct aclk_sync_cfg_t *wc;
602
-
603
- RRDHOST *host = find_host_by_node_id(node_id);
604
- if (unlikely(!host || !(wc = host->aclk_config)))
605
- return;
606
-
607
- if (unlikely(!host->health.health_enabled)) {
608
- nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, health is disabled.", node_id);
609
- return;
610
- }
611
-
612
- if (resets) {
613
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED (RESET REQUESTED)", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
614
- sql_queue_existing_alerts_to_aclk(host);
615
- } else
616
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
617
-
618
- wc->alert_updates = 1;
619
- wc->alert_queue_removed = SEND_REMOVED_AFTER_HEALTH_LOOPS;
620
-}
621
-
622
-#define SQL_QUEUE_REMOVE_ALERTS \
623
- "INSERT INTO aclk_alert_%s (alert_unique_id, date_created, filtered_alert_unique_id) " \
624
- "SELECT hld.unique_id alert_unique_id, UNIXEPOCH(), hld.unique_id alert_unique_id FROM health_log hl, health_log_detail hld " \
625
- "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 " \
626
- "AND hld.unique_id NOT IN (SELECT alert_unique_id FROM aclk_alert_%s) " \
627
- "AND hl.config_hash_id NOT IN (SELECT hash_id FROM alert_hash WHERE warn IS NULL AND crit IS NULL) " \
628
- "AND hl.name || hl.chart NOT IN (select name || chart FROM health_log WHERE name = hl.name AND " \
629
- "chart = hl.chart AND alarm_id > hl.alarm_id AND host_id = hl.host_id) " \
630
- "ORDER BY hld.unique_id ASC ON CONFLICT (alert_unique_id) DO NOTHING"
631
-
632
-void sql_process_queue_removed_alerts_to_aclk(char *node_id)
633
-{
634
- struct aclk_sync_cfg_t *wc;
635
- RRDHOST *host = find_host_by_node_id(node_id);
636
- freez(node_id);
637
-
638
- if (unlikely(!host || !(wc = host->aclk_config)))
639
- return;
640
-
641
- sqlite3_stmt *res = NULL;
782
+ static __thread sqlite3_stmt *res = NULL;
783
643
- CLEAN_BUFFER *wb = buffer_create(1024, NULL); // Note buffer auto free on function return
644
- buffer_sprintf(wb, SQL_QUEUE_REMOVE_ALERTS, wc->uuid_str, wc->uuid_str);
645
-
646
- if (!PREPARE_STATEMENT(db_meta, buffer_tostring(wb), &res))
647
- return;
784
+ if (!PREPARE_COMPILED_STATEMENT(db_meta, SQL_ALERT_VERSION_CALC, &res))
785
+ return 0;
786
787
+ uint64_t version = 0;
788
int param = 0;
789
SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
790
791
param = 0;
653
- int rc = execute_insert(res);
654
- if (likely(rc == SQLITE_DONE)) {
655
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK STA [%s (%s)]: QUEUED REMOVED ALERTS", wc->node_id, rrdhost_hostname(wc->host));
656
- rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
657
- wc->alert_queue_removed = 0;
792
+ while (sqlite3_step_monitored(res) == SQLITE_ROW) {
793
+ version = (uint64_t)sqlite3_column_int64(res, 0);
794
}
795
796
done:
797
REPORT_BIND_FAIL(res, param);
662
- SQLITE_FINALIZE(res);
798
+ SQLITE_RESET(res);
799
+ return version;
800
}
801
665
-void sql_queue_removed_alerts_to_aclk(RRDHOST *host)
802
+static void schedule_alert_snapshot_if_needed(struct aclk_sync_cfg_t *wc, uint64_t cloud_version)
803
{
667
- if (unlikely(!host->aclk_config || !claimed() || !host->node_id))
668
- return;
669
-
670
- char node_id[UUID_STR_LEN];
671
- uuid_unparse_lower(*host->node_id, node_id);
804
+ uint64_t local_version = calculate_node_alert_version(wc->host);
805
+ if (local_version != cloud_version) {
806
+ nd_log(
807
+ NDLS_ACCESS,
808
+ NDLP_NOTICE,
809
+ "Scheduling alert snapshot for host \"%s\", node \"%s\" (version: cloud %zu, local %zu)",
810
+ rrdhost_hostname(wc->host),
811
+ wc->node_id,
812
+ cloud_version,
813
+ local_version);
814
673
- aclk_push_node_removed_alerts(node_id);
815
+ wc->send_snapshot = 1;
816
+ rrdhost_flag_set(wc->host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
817
+ }
818
+ else
819
+ nd_log(
820
+ NDLS_ACCESS,
821
+ NDLP_DEBUG,
822
+ "Alert check on \"%s\", node \"%s\" (version: cloud %zu, local %zu)",
823
+ rrdhost_hostname(wc->host),
824
+ wc->node_id,
825
+ cloud_version,
826
+ local_version);
827
+ wc->checkpoint_count++;
828
}
829
830
void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id __maybe_unused, char *snapshot_uuid)
853
854
wc->alerts_snapshot_uuid = strdupz(snapshot_uuid);
855
702
- aclk_push_node_alert_snapshot(node_id);
856
+ wc->send_snapshot = 1;
857
+ rrdhost_flag_set(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS);
858
}
859
705
-#ifdef ENABLE_ACLK
706
-void health_alarm_entry2proto_nolock(struct alarm_log_entry *alarm_log, ALARM_ENTRY *ae, RRDHOST *host)
707
-{
708
- char *edit_command = ae->source ? health_edit_command_from_source(ae_source(ae)) : strdupz("UNKNOWN=0=UNKNOWN");
709
- char config_hash_id[UUID_STR_LEN];
710
- uuid_unparse_lower(ae->config_hash_id, config_hash_id);
711
- char transition_id[UUID_STR_LEN];
712
- uuid_unparse_lower(ae->transition_id, transition_id);
713
-
714
- alarm_log->chart = strdupz(ae_chart_id(ae));
715
- alarm_log->name = strdupz(ae_name(ae));
716
-
717
- alarm_log->when = ae->when;
860
+#define SQL_COUNT_SNAPSHOT_ENTRIES \
861
+ "SELECT COUNT(1) FROM alert_version av, health_log hl " \
862
+ "WHERE hl.host_id = @host_id AND hl.health_log_id = av.health_log_id"
863
719
- alarm_log->config_hash = strdupz((char *)config_hash_id);
720
-
721
- alarm_log->utc_offset = host->utc_offset;
722
- alarm_log->timezone = strdupz(rrdhost_abbrev_timezone(host));
723
- alarm_log->exec_path = ae->exec ? strdupz(ae_exec(ae)) : strdupz((char *)string2str(host->health.health_default_exec));
724
- alarm_log->conf_source = ae->source ? strdupz(ae_source(ae)) : strdupz((char *)"");
725
-
726
- alarm_log->command = strdupz((char *)edit_command);
727
-
728
- alarm_log->duration = (time_t)ae->duration;
729
- alarm_log->non_clear_duration = (time_t)ae->non_clear_duration;
730
- alarm_log->status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)ae->new_status);
731
- alarm_log->old_status = rrdcalc_status_to_proto_enum((RRDCALC_STATUS)ae->old_status);
732
- alarm_log->delay = ae->delay;
733
- alarm_log->delay_up_to_timestamp = (time_t)ae->delay_up_to_timestamp;
734
- alarm_log->last_repeat = (time_t)ae->last_repeat;
735
-
736
- alarm_log->silenced =
737
- ((ae->flags & HEALTH_ENTRY_FLAG_SILENCED) || (ae->recipient && !strncmp(ae_recipient(ae), "silent", 6))) ?
738
- 1 :
739
- 0;
864
+static int calculate_alert_snapshot_entries(nd_uuid_t *host_uuid)
865
+{
866
+ int count = 0;
867
741
- alarm_log->value_string = strdupz(ae_new_value_string(ae));
742
- alarm_log->old_value_string = strdupz(ae_old_value_string(ae));
868
+ sqlite3_stmt *res = NULL;
869
744
- alarm_log->value = (!isnan(ae->new_value)) ? (NETDATA_DOUBLE)ae->new_value : 0;
745
- alarm_log->old_value = (!isnan(ae->old_value)) ? (NETDATA_DOUBLE)ae->old_value : 0;
870
+ if (!PREPARE_STATEMENT(db_meta, SQL_COUNT_SNAPSHOT_ENTRIES, &res))
871
+ return 0;
872
747
- alarm_log->updated = (ae->flags & HEALTH_ENTRY_FLAG_UPDATED) ? 1 : 0;
748
- alarm_log->rendered_info = strdupz(ae_info(ae));
749
- alarm_log->chart_context = strdupz(ae_chart_context(ae));
750
- alarm_log->chart_name = strdupz(ae_chart_name(ae));
873
+ int param = 0;
874
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, host_uuid, sizeof(*host_uuid), SQLITE_STATIC));
875
752
- alarm_log->transition_id = strdupz((char *)transition_id);
753
- alarm_log->event_id = (uint64_t) ae->alarm_event_id;
876
+ param = 0;
877
+ int rc = sqlite3_step_monitored(res);
878
+ if (rc == SQLITE_ROW)
879
+ count = sqlite3_column_int(res, 0);
880
+ else
881
+ error_report("Failed to select snapshot count");
882
755
- alarm_log->summary = strdupz(ae_summary(ae));
883
+done:
884
+ REPORT_BIND_FAIL(res, param);
885
+ SQLITE_FINALIZE(res);
886
757
- freez(edit_command);
887
+ return count;
888
}
759
-#endif
760
-
761
-#ifdef ENABLE_ACLK
762
-static bool have_recent_alarm_unsafe(RRDHOST *host, int64_t alarm_id, int64_t mark)
763
-{
764
- ALARM_ENTRY *ae = host->health_log.alarms;
765
-
766
- while (ae) {
767
- if (ae->alarm_id == alarm_id && ae->unique_id >mark &&
768
- (ae->new_status != RRDCALC_STATUS_WARNING && ae->new_status != RRDCALC_STATUS_CRITICAL))
769
- return true;
770
- ae = ae->next;
771
- }
889
773
- return false;
774
-}
775
-#endif
890
+#define SQL_GET_SNAPSHOT_ENTRIES \
891
+ " SELECT 0, hld.unique_id, hld.alarm_id, hl.config_hash_id, hld.updated_by_id, hld.when_key, " \
892
+ " hld.duration, hld.non_clear_duration, hld.flags, hld.exec_run_timestamp, hld.delay_up_to_timestamp, hl.name, " \
893
+ " hl.chart, hl.exec, hl.recipient, ah.source, hl.units, hld.info, hld.exec_code, hld.new_status, " \
894
+ " hld.old_status, hld.delay, hld.new_value, hld.old_value, hld.last_repeat, hl.chart_context, hld.transition_id, " \
895
+ " hld.alarm_event_id, hl.chart_name, hld.summary, hld.health_log_id, av.version " \
896
+ " FROM health_log hl, alert_hash ah, health_log_detail hld, alert_version av " \
897
+ " WHERE hl.config_hash_id = ah.hash_id" \
898
+ " AND hl.host_id = @host_id AND hl.health_log_id = hld.health_log_id " \
899
+ " AND hld.health_log_id = av.health_log_id AND av.unique_id = hld.unique_id"
900
901
#define ALARM_EVENTS_PER_CHUNK 1000
778
-void aclk_push_alert_snapshot_event(char *node_id __maybe_unused)
902
+void send_alert_snapshot_to_cloud(RRDHOST *host __maybe_unused)
903
{
780
-#ifdef ENABLE_ACLK
781
- RRDHOST *host = find_host_by_node_id(node_id);
904
+ struct aclk_sync_cfg_t *wc = host->aclk_config;
905
906
if (unlikely(!host)) {
784
- nd_log(NDLS_ACCESS, NDLP_WARNING, "AC [%s (N/A)]: Node id not found", node_id);
785
- freez(node_id);
907
+ nd_log(NDLS_ACCESS, NDLP_WARNING, "AC [%s (N/A)]: Node id not found", wc->node_id);
908
return;
909
}
788
- freez(node_id);
789
-
790
- struct aclk_sync_cfg_t *wc = host->aclk_config;
910
792
- // we perhaps we don't need this for snapshots
793
- if (unlikely(!wc->alert_updates)) {
794
- nd_log(NDLS_ACCESS, NDLP_NOTICE,
795
- "ACLK STA [%s (%s)]: Ignoring alert snapshot event, updates have been turned off for this node.",
796
- wc->node_id,
797
- wc->host ? rrdhost_hostname(wc->host) : "N/A");
911
+ char *claim_id = get_agent_claimid();
912
+ if (unlikely(!claim_id))
913
return;
799
- }
914
801
- if (unlikely(!wc->alerts_snapshot_uuid))
915
+ // Check database for this node to see how many alerts we will need to put in the snapshot
916
+ int cnt = calculate_alert_snapshot_entries(&host->host_uuid);
917
+ if (!cnt) {
918
+ freez(claim_id);
919
return;
920
+ }
921
804
- char *claim_id = get_agent_claimid();
805
- if (unlikely(!claim_id))
922
+ sqlite3_stmt *res = NULL;
923
+ if (!PREPARE_STATEMENT(db_meta, SQL_GET_SNAPSHOT_ENTRIES, &res))
924
return;
925
808
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: Sending alerts snapshot, snapshot_uuid %s", wc->node_id, rrdhost_hostname(wc->host), wc->alerts_snapshot_uuid);
926
+ int param = 0;
927
+ SQLITE_BIND_FAIL(done, sqlite3_bind_blob(res, ++param, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC));
928
810
- uint32_t cnt = 0;
929
+ nd_uuid_t local_snapshot_uuid;
930
+ char snapshot_uuid_str[UUID_STR_LEN];
931
+ uuid_generate_random(local_snapshot_uuid);
932
+ uuid_unparse_lower(local_snapshot_uuid, snapshot_uuid_str);
933
+ char *snapshot_uuid = &snapshot_uuid_str[0];
934
812
- rw_spinlock_read_lock(&host->health_log.spinlock);
935
+ nd_log(NDLS_ACCESS, NDLP_DEBUG,
936
+ "ACLK REQ [%s (%s)]: Sending %d alerts snapshot, snapshot_uuid %s", wc->node_id, rrdhost_hostname(host),
937
+ cnt, snapshot_uuid);
938
814
- ALARM_ENTRY *ae = host->health_log.alarms;
939
+ uint32_t chunks;
940
+ chunks = (cnt / ALARM_EVENTS_PER_CHUNK) + (cnt % ALARM_EVENTS_PER_CHUNK != 0);
941
816
- for (; ae; ae = ae->next) {
817
- if (likely(ae->updated_by_id))
818
- continue;
942
+ alarm_snapshot_proto_ptr_t snapshot_proto = NULL;
943
+ struct alarm_snapshot alarm_snap;
944
+ struct alarm_log_entry alarm_log;
945
820
- if (unlikely(ae->new_status == RRDCALC_STATUS_UNINITIALIZED))
821
- continue;
946
+ alarm_snap.node_id = wc->node_id;
947
+ alarm_snap.claim_id = claim_id;
948
+ alarm_snap.snapshot_uuid = snapshot_uuid;
949
+ alarm_snap.chunks = chunks;
950
+ alarm_snap.chunk = 1;
951
823
- if (have_recent_alarm_unsafe(host, ae->alarm_id, ae->unique_id))
824
- continue;
952
+ alarm_log.node_id = wc->node_id;
953
+ alarm_log.claim_id = claim_id;
954
826
- if (is_event_from_alert_variable_config(ae->unique_id, &host->host_uuid))
827
- continue;
955
956
+ cnt = 0;
957
+ param = 0;
958
+ uint64_t version = 0;
959
+ while (sqlite3_step_monitored(res) == SQLITE_ROW) {
960
cnt++;
830
- }
831
-
832
- if (cnt) {
833
- uint32_t chunks;
834
-
835
- chunks = (cnt / ALARM_EVENTS_PER_CHUNK) + (cnt % ALARM_EVENTS_PER_CHUNK != 0);
836
- ae = host->health_log.alarms;
961
838
- cnt = 0;
839
- struct alarm_snapshot alarm_snap;
840
- alarm_snap.node_id = wc->node_id;
841
- alarm_snap.claim_id = claim_id;
842
- alarm_snap.snapshot_uuid = wc->alerts_snapshot_uuid;
843
- alarm_snap.chunks = chunks;
844
- alarm_snap.chunk = 1;
962
+ if (!snapshot_proto)
963
+ snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
964
846
- alarm_snapshot_proto_ptr_t snapshot_proto = NULL;
965
+ health_alarm_log_populate(&alarm_log, res, host, NULL);
966
848
- for (; ae; ae = ae->next) {
849
- if (likely(ae->updated_by_id) || unlikely(ae->new_status == RRDCALC_STATUS_UNINITIALIZED))
850
- continue;
851
-
852
- if (have_recent_alarm_unsafe(host, ae->alarm_id, ae->unique_id))
853
- continue;
854
-
855
- if (is_event_from_alert_variable_config(ae->unique_id, &host->host_uuid))
856
- continue;
857
-
858
- cnt++;
859
-
860
- struct alarm_log_entry alarm_log;
861
- alarm_log.node_id = wc->node_id;
862
- alarm_log.claim_id = claim_id;
863
-
864
- if (!snapshot_proto)
865
- snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
967
+ //log_alarm_log(&alarm_log);
968
867
- health_alarm_entry2proto_nolock(&alarm_log, ae, host);
868
- add_alarm_log_entry2snapshot(snapshot_proto, &alarm_log);
969
+ add_alarm_log_entry2snapshot(snapshot_proto, &alarm_log);
970
+ version += alarm_log.version;
971
870
- if (cnt == ALARM_EVENTS_PER_CHUNK) {
972
+ if (cnt == ALARM_EVENTS_PER_CHUNK) {
973
+ if (aclk_connected)
974
aclk_send_alarm_snapshot(snapshot_proto);
872
- cnt = 0;
873
- if (alarm_snap.chunk < chunks) {
874
- alarm_snap.chunk++;
875
- snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
876
- }
975
+ cnt = 0;
976
+ if (alarm_snap.chunk < chunks) {
977
+ alarm_snap.chunk++;
978
+ snapshot_proto = generate_alarm_snapshot_proto(&alarm_snap);
979
}
878
- destroy_alarm_log_entry(&alarm_log);
980
}
880
- if (cnt)
881
- aclk_send_alarm_snapshot(snapshot_proto);
981
+ destroy_alarm_log_entry(&alarm_log);
982
}
983
+ if (cnt)
984
+ aclk_send_alarm_snapshot(snapshot_proto);
985
884
- rw_spinlock_read_unlock(&host->health_log.spinlock);
885
- wc->alerts_snapshot_uuid = NULL;
886
-
887
- freez(claim_id);
888
-#endif
889
-}
890
-
891
-#define SQL_DELETE_ALERT_ENTRIES "DELETE FROM aclk_alert_%s WHERE date_created < UNIXEPOCH() - @period"
892
-
893
-void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
894
-{
895
- struct aclk_sync_cfg_t *wc = host->aclk_config;
896
- if (unlikely(!wc))
897
- return;
898
-
899
- char sql[ACLK_SYNC_QUERY_SIZE];
900
- snprintfz(sql, sizeof(sql) - 1, SQL_DELETE_ALERT_ENTRIES, wc->uuid_str);
901
-
902
- sqlite3_stmt *res = NULL;
903
-
904
- if (!PREPARE_STATEMENT(db_meta, sql, &res))
905
- return;
906
-
907
- int param = 0;
908
- SQLITE_BIND_FAIL(done, sqlite3_bind_int64(res, ++param, MAX_REMOVED_PERIOD));
909
-
910
- param = 0;
911
- int rc = sqlite3_step_monitored(res);
912
- if (rc != SQLITE_DONE)
913
- error_report("Failed to execute DELETE query for cleaning stale ACLK alert entries.");
986
+ nd_log(
987
+ NDLS_ACCESS,
988
+ NDLP_DEBUG,
989
+ "ACLK REQ [%s (%s)]: Sent! %d alerts snapshot, snapshot_uuid %s (version = %zu)",
990
+ wc->node_id,
991
+ rrdhost_hostname(host),
992
+ cnt,
993
+ snapshot_uuid,
994
+ version);
995
996
done:
997
REPORT_BIND_FAIL(res, param);
998
SQLITE_FINALIZE(res);
918
-}
919
-
920
-#define SQL_GET_MIN_MAX_ALERT_SEQ "SELECT MIN(sequence_id), MAX(sequence_id), " \
921
- "(SELECT MAX(sequence_id) FROM aclk_alert_%s WHERE date_submitted IS NOT NULL) " \
922
- "FROM aclk_alert_%s WHERE date_submitted IS NULL"
923
-int get_proto_alert_status(RRDHOST *host, struct proto_alert_status *proto_alert_status)
924
-{
925
-
926
- struct aclk_sync_cfg_t *wc = host->aclk_config;
927
- if (!wc)
928
- return 1;
929
-
930
- proto_alert_status->alert_updates = wc->alert_updates;
931
-
932
- char sql[ACLK_SYNC_QUERY_SIZE];
933
- snprintfz(sql, sizeof(sql) - 1, SQL_GET_MIN_MAX_ALERT_SEQ, wc->uuid_str, wc->uuid_str);
934
-
935
- sqlite3_stmt *res = NULL;
936
- if (!PREPARE_STATEMENT(db_meta, sql, &res))
937
- return 1;
938
-
939
- while (sqlite3_step_monitored(res) == SQLITE_ROW) {
940
- proto_alert_status->pending_min_sequence_id =
941
- sqlite3_column_bytes(res, 0) > 0 ? (uint64_t)sqlite3_column_int64(res, 0) : 0;
942
- proto_alert_status->pending_max_sequence_id =
943
- sqlite3_column_bytes(res, 1) > 0 ? (uint64_t)sqlite3_column_int64(res, 1) : 0;
944
- proto_alert_status->last_submitted_sequence_id =
945
- sqlite3_column_bytes(res, 2) > 0 ? (uint64_t)sqlite3_column_int64(res, 2) : 0;
946
- }
947
-
948
- SQLITE_FINALIZE(res);
999
950
- return 0;
1000
+ freez(claim_id);
1001
}
1002
953
-void aclk_send_alarm_checkpoint(char *node_id, char *claim_id __maybe_unused)
1003
+// Start streaming alerts
1004
+void aclk_start_alert_streaming(char *node_id, uint64_t cloud_version)
1005
{
955
- if (unlikely(!node_id))
1006
+ nd_uuid_t node_uuid;
1007
+
1008
+ if (unlikely(!node_id || uuid_parse(node_id, node_uuid)))
1009
return;
1010
1011
struct aclk_sync_cfg_t *wc;
1012
RRDHOST *host = find_host_by_node_id(node_id);
1013
961
- if (unlikely(!host || !(wc = host->aclk_config)))
962
- nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT REQUEST RECEIVED FOR INVALID NODE", node_id);
963
- else {
964
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: ALERTS CHECKPOINT REQUEST RECEIVED", node_id, rrdhost_hostname(host));
965
- wc->alert_checkpoint_req = SEND_CHECKPOINT_AFTER_HEALTH_LOOPS;
966
- }
967
-}
968
-
969
-typedef struct active_alerts {
970
- char *name;
971
- char *chart;
972
- RRDCALC_STATUS status;
973
-} active_alerts_t;
974
-
975
-static inline int compare_active_alerts(const void *a, const void *b)
976
-{
977
- active_alerts_t *active_alerts_a = (active_alerts_t *)a;
978
- active_alerts_t *active_alerts_b = (active_alerts_t *)b;
979
-
980
- if (!(strcmp(active_alerts_a->name, active_alerts_b->name))) {
981
- return strcmp(active_alerts_a->chart, active_alerts_b->chart);
982
- } else
983
- return strcmp(active_alerts_a->name, active_alerts_b->name);
984
-}
985
-
986
-#define BATCH_ALLOCATED 10
987
-void aclk_push_alarm_checkpoint(RRDHOST *host __maybe_unused)
988
-{
989
-#ifdef ENABLE_ACLK
990
- struct aclk_sync_cfg_t *wc = host->aclk_config;
991
- if (unlikely(!wc)) {
992
- nd_log(NDLS_ACCESS, NDLP_WARNING, "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT REQUEST RECEIVED FOR INVALID NODE", rrdhost_hostname(host));
1014
+ if (unlikely(!host || !(wc = host->aclk_config))) {
1015
+ nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, invalid node.", node_id);
1016
return;
1017
}
1018
996
- if (rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_ALERTS)) {
997
- //postpone checkpoint send
998
- wc->alert_checkpoint_req += 3;
999
- nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT POSTPONED", rrdhost_hostname(host));
1019
+ if (unlikely(!host->health.health_enabled)) {
1020
+ nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK STA [%s (N/A)]: Ignoring request to stream alert state changes, health is disabled.", node_id);
1021
return;
1022
}
1023
1003
- RRDCALC *rc;
1004
- uint32_t cnt = 0;
1005
- size_t len = 0;
1006
-
1007
- active_alerts_t *active_alerts = callocz(BATCH_ALLOCATED, sizeof(active_alerts_t));
1008
- foreach_rrdcalc_in_rrdhost_read(host, rc) {
1009
- if(unlikely(!rc->rrdset || !rc->rrdset->last_collected_time.tv_sec))
1010
- continue;
1024
+ nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK REQ [%s (%s)]: STREAM ALERTS ENABLED", node_id, wc->host ? rrdhost_hostname(wc->host) : "N/A");
1025
+ schedule_alert_snapshot_if_needed(wc, cloud_version);
1026
+ wc->stream_alerts = true;
1027
+}
1028
1012
- if (rc->status == RRDCALC_STATUS_WARNING ||
1013
- rc->status == RRDCALC_STATUS_CRITICAL) {
1029
+// Do checkpoint alert version check
1030
+void aclk_alert_version_check(char *node_id, char *claim_id, uint64_t cloud_version)
1031
+{
1032
+ nd_uuid_t node_uuid;
1033
1015
- if (cnt && !(cnt % BATCH_ALLOCATED)) {
1016
- active_alerts = reallocz(active_alerts, (BATCH_ALLOCATED * ((cnt / BATCH_ALLOCATED) + 1)) * sizeof(active_alerts_t));
1017
- }
1034
+ if (unlikely(!node_id || !claim_id || !claimed() || uuid_parse(node_id, node_uuid)))
1035
+ return;
1036
1019
- active_alerts[cnt].name = (char *)rrdcalc_name(rc);
1020
- len += string_strlen(rc->config.name);
1021
- active_alerts[cnt].chart = (char *)rrdcalc_chart_name(rc);
1022
- len += string_strlen(rc->chart);
1023
- active_alerts[cnt].status = rc->status;
1024
- len++;
1025
- cnt++;
1026
- }
1027
- }
1028
- foreach_rrdcalc_in_rrdhost_done(rc);
1029
-
1030
- BUFFER *alarms_to_hash;
1031
- if (cnt) {
1032
- qsort(active_alerts, cnt, sizeof(active_alerts_t), compare_active_alerts);
1033
-
1034
- alarms_to_hash = buffer_create(len, NULL);
1035
- for (uint32_t i = 0; i < cnt; i++) {
1036
- buffer_strcat(alarms_to_hash, active_alerts[i].name);
1037
- buffer_strcat(alarms_to_hash, active_alerts[i].chart);
1038
- if (active_alerts[i].status == RRDCALC_STATUS_WARNING)
1039
- buffer_fast_strcat(alarms_to_hash, "W", 1);
1040
- else if (active_alerts[i].status == RRDCALC_STATUS_CRITICAL)
1041
- buffer_fast_strcat(alarms_to_hash, "C", 1);
1042
- }
1043
- } else {
1044
- alarms_to_hash = buffer_create(1, NULL);
1045
- buffer_strcat(alarms_to_hash, "");
1046
- len = 0;
1037
+ char *agent_claim_id = get_agent_claimid();
1038
+ if (claim_id && agent_claim_id && strcmp(agent_claim_id, claim_id) != 0) {
1039
+ nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT VALIDATION REQUEST RECEIVED WITH INVALID CLAIM ID", node_id);
1040
+ goto done;
1041
}
1048
- freez(active_alerts);
1042
1050
- char hash[SHA256_DIGEST_LENGTH + 1];
1051
- if (hash256_string((const unsigned char *)buffer_tostring(alarms_to_hash), len, hash)) {
1052
- hash[SHA256_DIGEST_LENGTH] = 0;
1043
+ struct aclk_sync_cfg_t *wc;
1044
+ RRDHOST *host = find_host_by_node_id(node_id);
1045
1054
- struct alarm_checkpoint alarm_checkpoint;
1055
- char *claim_id = get_agent_claimid();
1056
- alarm_checkpoint.claim_id = claim_id;
1057
- alarm_checkpoint.node_id = wc->node_id;
1058
- alarm_checkpoint.checksum = (char *)hash;
1046
+ if ((!host || !(wc = host->aclk_config)))
1047
+ nd_log(NDLS_ACCESS, NDLP_NOTICE, "ACLK REQ [%s (N/A)]: ALERTS CHECKPOINT VALIDATION REQUEST RECEIVED FOR INVALID NODE", node_id);
1048
+ else
1049
+ schedule_alert_snapshot_if_needed(wc, cloud_version);
1050
1060
- aclk_send_provide_alarm_checkpoint(&alarm_checkpoint);
1061
- freez(claim_id);
1062
- nd_log(NDLS_ACCESS, NDLP_DEBUG, "ACLK RES [%s (%s)]: ALERTS CHECKPOINT SENT", wc->node_id, rrdhost_hostname(host));
1063
- } else
1064
- nd_log(NDLS_ACCESS, NDLP_ERR, "ACLK RES [%s (%s)]: FAILED TO CREATE ALERTS CHECKPOINT HASH", wc->node_id, rrdhost_hostname(host));
1051
+done:
1052
+ freez(agent_claim_id);
1053
+}
1054
1066
- wc->alert_checkpoint_req = 0;
1067
- buffer_free(alarms_to_hash);
1055
#endif
1069
-}