Pause alert pushes to the cloud (#12852)
* pause and unpause alert pushes to the cloud * move the check to when creating opcode * check for worker * remove previous checks for dbsync_workers. queue and clean aclk_alert tables even if no workers are up. Get wc then check before setting pause * remove sync_syncronize * remove sync_synchronize_2
Emmanuel Vasilakis committed
May 12, 2022 at 15:52 UTC
73bb8888f3bffcb36b99b9341f15d21b3de844d0
3 files changed
+15
-7
database/sqlite/sqlite_aclk.c
+1
-1
@@ -344,7 +344,7 @@ static void timer_cb(uv_timer_t* handle)
344
}
345
}
346
347
- if (wc->alert_updates) {
347
+ if (wc->alert_updates && !wc->pause_alert_updates) {
348
cmd.opcode = ACLK_DATABASE_PUSH_ALERT;
349
cmd.count = ACLK_MAX_ALERT_UPDATES;
350
aclk_database_enq_cmd_noblock(wc, &cmd);
database/sqlite/sqlite_aclk.h
+1
@@ -178,6 +178,7 @@ struct aclk_database_worker_config {
178
uint64_t alerts_batch_id; // batch id for alerts to use
179
uint64_t alerts_start_seq_id; // cloud has asked to start streaming from
180
uint64_t alert_sequence_id; // last alert sequence_id
181
+ int pause_alert_updates;
182
uint32_t chart_payload_count;
183
uint64_t alerts_snapshot_id; //will contain the snapshot_id value if snapshot was requested
184
uint64_t alerts_ack_sequence_id; //last sequence_id ack'ed from cloud via sendsnapshot message
database/sqlite/sqlite_aclk_alert.c
+13
-6
@@ -141,9 +141,6 @@ int sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, int skip_filter)
141
if (!claimed())
142
return 0;
143
144
- if (unlikely(!host->dbsync_worker))
145
- return 1;
146
-
144
if (ae->flags & HEALTH_ENTRY_FLAG_ACLK_QUEUED) {
145
return 0;
146
}
@@ -188,6 +185,10 @@ int sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae, int skip_filter)
185
}
186
187
ae->flags |= HEALTH_ENTRY_FLAG_ACLK_QUEUED;
188
+ struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *)host->dbsync_worker;
189
+ if (wc) {
190
+ wc->pause_alert_updates = 0;
191
+ }
192
193
bind_fail:
194
if (unlikely(sqlite3_finalize(res_alert) != SQLITE_OK))
@@ -396,6 +397,7 @@ void aclk_push_alert_event(struct aclk_database_worker_config *wc, struct aclk_d
397
wc->alerts_batch_id);
398
log_first_sequence_id = 0;
399
log_last_sequence_id = 0;
400
+ wc->pause_alert_updates = 1;
401
}
402
403
rc = sqlite3_finalize(res);
@@ -423,6 +425,11 @@ void sql_queue_existing_alerts_to_aclk(RRDHOST *host)
425
db_execute(buffer_tostring(sql));
426
427
buffer_free(sql);
428
+
429
+ struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *)host->dbsync_worker;
430
+ if (wc) {
431
+ wc->pause_alert_updates = 0;
432
+ }
433
}
434
435
void aclk_send_alarm_health_log(char *node_id)
@@ -734,6 +741,7 @@ void aclk_start_alert_streaming(char *node_id, uint64_t batch_id, uint64_t start
741
wc->alerts_batch_id = batch_id;
742
wc->alerts_start_seq_id = start_seq_id;
743
wc->alert_updates = 1;
744
+ wc->pause_alert_updates = 0;
745
__sync_synchronize();
746
}
747
else
@@ -766,6 +774,8 @@ void sql_process_queue_removed_alerts_to_aclk(struct aclk_database_worker_config
774
log_access("ACLK STA [%s (%s)]: Queued removed alerts.", wc->node_id, wc->host ? wc->host->hostname : "N/A");
775
776
buffer_free(sql);
777
+
778
+ wc->pause_alert_updates = 0;
779
#endif
780
return;
781
}
@@ -1047,9 +1057,6 @@ void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
1057
if (!claimed())
1058
return;
1059
1050
- if (unlikely(!host->dbsync_worker))
1051
- return;
1052
-
1060
char uuid_str[GUID_LEN + 1];
1061
uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
1062