Queue removed alerts to cloud for new architecture (#11704)
* rebased * add error message * make function void * fix return
Emmanuel Vasilakis committed
Oct 25, 2021 at 16:39 UTC
eefa40cb545cca664afb552e6d3299f6b87c00e7
5 files changed
+44
-15
database/sqlite/sqlite_aclk.c
+4
@@ -416,6 +416,10 @@ void aclk_database_worker(void *arg)
416
debug(D_ACLK_SYNC, "Pushing alert snapshot to the cloud for node %s", wc->host_guid);
417
aclk_push_alert_snapshot_event(wc, cmd);
418
break;
419
+ case ACLK_DATABASE_QUEUE_REMOVED_ALERTS:
420
+ debug(D_ACLK_SYNC, "Queueing removed alerts for node %s", wc->host_guid);
421
+ sql_process_queue_removed_alerts_to_aclk(wc, cmd);
422
+ break;
423
424
// NODE OPERATIONS
425
case ACLK_DATABASE_NODE_INFO:
database/sqlite/sqlite_aclk.h
+1
@@ -125,6 +125,7 @@ enum aclk_database_opcode {
125
ACLK_DATABASE_PUSH_ALERT,
126
ACLK_DATABASE_PUSH_ALERT_CONFIG,
127
ACLK_DATABASE_PUSH_ALERT_SNAPSHOT,
128
+ ACLK_DATABASE_QUEUE_REMOVED_ALERTS,
129
ACLK_DATABASE_PUSH_CHART,
130
ACLK_DATABASE_PUSH_CHART_CONFIG,
131
ACLK_DATABASE_RESET_CHART,
database/sqlite/sqlite_aclk_alert.c
+24
-14
@@ -558,20 +558,12 @@ void aclk_start_alert_streaming(char *node_id, uint64_t batch_id, uint64_t start
558
return;
559
}
560
561
-int sql_queue_removed_alerts_to_aclk(RRDHOST *host)
561
+void sql_process_queue_removed_alerts_to_aclk(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
562
{
563
-#ifdef ENABLE_NEW_CLOUD_PROTOCOL
564
- if (!aclk_use_new_cloud_arch) {
565
- return 0;
566
- }
567
-
568
- CHECK_SQLITE_CONNECTION(db_meta);
569
-
570
- struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *) host->dbsync_worker;
571
- if (unlikely(!wc)) {
572
- return 1;
573
- }
574
-
563
+ UNUSED(cmd);
564
+#ifndef ENABLE_NEW_CLOUD_PROTOCOL
565
+ UNUSED(wc);
566
+#else
567
BUFFER *sql = buffer_create(1024);
568
569
buffer_sprintf(sql,"insert into aclk_alert_%s (alert_unique_id, date_created) " \
@@ -583,10 +575,28 @@ int sql_queue_removed_alerts_to_aclk(RRDHOST *host)
575
db_execute(buffer_tostring(sql));
576
577
buffer_free(sql);
578
+#endif
579
+ return;
580
+}
581
+
582
+void sql_queue_removed_alerts_to_aclk(RRDHOST *host)
583
+{
584
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
585
+ if (unlikely(!host->dbsync_worker)) {
586
+ error("ACLK synchronization thread is not active for host %s", host->hostname);
587
+ return;
588
+ }
589
+
590
+ struct aclk_database_cmd cmd;
591
+ memset(&cmd, 0, sizeof(cmd));
592
+ cmd.opcode = ACLK_DATABASE_QUEUE_REMOVED_ALERTS;
593
+ cmd.data = NULL;
594
+ cmd.data_param = NULL;
595
+ cmd.completion = NULL;
596
+ aclk_database_enq_cmd((struct aclk_database_worker_config *) host->dbsync_worker, &cmd);
597
#else
598
UNUSED(host);
599
#endif
589
- return 0;
600
}
601
602
void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id, uint64_t snapshot_id, uint64_t sequence_id)
database/sqlite/sqlite_aclk_alert.h
+2
-1
@@ -12,7 +12,8 @@ void aclk_push_alarm_health_log(struct aclk_database_worker_config *wc, struct a
12
void aclk_send_alarm_configuration (char *config_hash);
13
int aclk_push_alert_config_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
14
void aclk_start_alert_streaming(char *node_id, uint64_t batch_id, uint64_t start_seq_id);
15
-int sql_queue_removed_alerts_to_aclk(RRDHOST *host);
15
+void sql_queue_removed_alerts_to_aclk(RRDHOST *host);
16
+void sql_process_queue_removed_alerts_to_aclk(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
17
void aclk_push_alert_snapshot_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
18
void aclk_process_send_alarm_snapshot(char *node_id, char *claim_id, uint64_t snapshot_id, uint64_t sequence_id);
19
health/health.c
+13
@@ -1038,6 +1038,14 @@ void *health_main(void *ptr) {
1038
rrdhost_unlock(host);
1039
}
1040
1041
+#ifdef ENABLE_ACLK
1042
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1043
+ if (netdata_cloud_setting && unlikely(aclk_alert_reloaded)) {
1044
+ sql_queue_removed_alerts_to_aclk(host);
1045
+ }
1046
+#endif
1047
+#endif
1048
+
1049
if (unlikely(netdata_exit))
1050
break;
1051
@@ -1062,6 +1070,11 @@ void *health_main(void *ptr) {
1070
health_alarm_wait_for_execution(ae);
1071
}
1072
1073
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
1074
+ if (netdata_cloud_setting && unlikely(aclk_alert_reloaded))
1075
+ aclk_alert_reloaded = 0;
1076
+#endif
1077
+
1078
rrd_unlock();
1079
1080
if(unlikely(netdata_exit))