Always queue alerts to aclk_alert (#11806)

* always queue to aclk_alert

* proper function name

5 files changed, 34 insertions, 3 deletions

diff --git a/database/sqlite/sqlite_aclk.c b/database/sqlite/sqlite_aclk.c
index 3d9c26d5ff..819b69db4d 100644
--- a/database/sqlite/sqlite_aclk.c
+++ b/database/sqlite/sqlite_aclk.c
@@ -30,7 +30,7 @@ const char *aclk_sync_config[] = {
 uv_mutex_t aclk_async_lock;
 struct aclk_database_worker_config  *aclk_thread_head = NULL;
 
-static inline int claimed()
+int claimed()
 {
     int rc;
     rrdhost_aclk_state_lock(localhost);
diff --git a/database/sqlite/sqlite_aclk.h b/database/sqlite/sqlite_aclk.h
index 9bf6482a83..d0499903b2 100644
--- a/database/sqlite/sqlite_aclk.h
+++ b/database/sqlite/sqlite_aclk.h
@@ -226,4 +226,5 @@ void sql_aclk_sync_init(void);
 void sql_check_aclk_table_list(struct aclk_database_worker_config *wc);
 void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
 void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd);
+int claimed();
 #endif //NETDATA_SQLITE_ACLK_H
diff --git a/database/sqlite/sqlite_aclk_alert.c b/database/sqlite/sqlite_aclk_alert.c
index 950eadbf60..a71d686fd3 100644
--- a/database/sqlite/sqlite_aclk_alert.c
+++ b/database/sqlite/sqlite_aclk_alert.c
@@ -16,18 +16,20 @@ int sql_queue_alarm_to_aclk(RRDHOST *host, ALARM_ENTRY *ae)
     //include also the valid statuses for this case
 #ifdef ENABLE_ACLK
 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
-    if (!aclk_use_new_cloud_arch) {
+    if (!aclk_use_new_cloud_arch && aclk_connected) {
 #endif
 
         if ((ae->new_status == RRDCALC_STATUS_WARNING || ae->new_status == RRDCALC_STATUS_CRITICAL) ||
             ((ae->old_status == RRDCALC_STATUS_WARNING || ae->old_status == RRDCALC_STATUS_CRITICAL))) {
             aclk_update_alarm(host, ae);
         }
-        return 0;
 #endif
 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
     }
 
+    if (!claimed())
+        return 0;
+
     if (ae->flags & HEALTH_ENTRY_FLAG_ACLK_QUEUED)
         return 0;
 
@@ -846,3 +848,28 @@ void aclk_push_alert_snapshot_event(struct aclk_database_worker_config *wc, stru
 #endif
     return;
 }
+
+void sql_aclk_alert_clean_dead_entries(RRDHOST *host)
+{
+#ifdef ENABLE_NEW_CLOUD_PROTOCOL
+    if (!claimed())
+        return;
+
+    if (unlikely(!host->dbsync_worker))
+        return;
+
+    char uuid_str[GUID_LEN + 1];
+    uuid_unparse_lower_fix(&host->host_uuid, uuid_str);
+
+    BUFFER *sql = buffer_create(1024);
+
+    buffer_sprintf(sql,"delete from aclk_alert_%s where alert_unique_id not in "
+                   " (select unique_id from health_log_%s); ", uuid_str, uuid_str);
+
+    db_execute(buffer_tostring(sql));
+
+    buffer_free(sql);
+#else
+    UNUSED(host);
+#endif
+}
diff --git a/database/sqlite/sqlite_health.c b/database/sqlite/sqlite_health.c
index fd1e7856ac..116cb4f3e0 100644
--- a/database/sqlite/sqlite_health.c
+++ b/database/sqlite/sqlite_health.c
@@ -392,6 +392,8 @@ void sql_health_alarm_log_cleanup(RRDHOST *host) {
         error_report("Failed to finalize the prepared statement to cleanup health log table");
 
     host->health_log_entries_written = rotate_every;
+
+    sql_aclk_alert_clean_dead_entries(host);
 }
 
 /* Health related SQL queries
diff --git a/database/sqlite/sqlite_health.h b/database/sqlite/sqlite_health.h
index 6396aca1de..3b94608975 100644
--- a/database/sqlite/sqlite_health.h
+++ b/database/sqlite/sqlite_health.h
@@ -13,4 +13,5 @@ extern void sql_health_alarm_log_insert(RRDHOST *host, ALARM_ENTRY *ae);
 extern void sql_health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae);
 extern void sql_health_alarm_log_cleanup(RRDHOST *host);
 extern int alert_hash_and_store_config(uuid_t hash_id, struct alert_config *cfg);
+extern void sql_aclk_alert_clean_dead_entries(RRDHOST *host);
 #endif //NETDATA_SQLITE_HEALTH_H