Improve the ACLK sync process for the new cloud architecture (#11744)

* Move retention code to the charts

* Log information about node registration and updates

* Prevent deadlock if aclk_database_enq_cmd locks for a node

* Improve message (indicate that it comes from alerts). This will be improved in a followup PR

* Disable parts that can't be used if the new cloud env is not available

* Set dimension FLAG if message has been queued

* Queue messages using the correct protocol enabled

* Cleanup unused functions
Rename functions that queue charts and dimensions
Improve the generic chart payload add function
Add a counter for pending charts/dimension payloads to avoid polling the db
Delay the retention update message until we are done with the updates
Fix full resync command to handle sequence_id = 0 correctly
Disable functions not needed when the new cloud env functionality is not compiled

* Add chart_payload count and retry count
Output information or error message if we fail to queue chart/dimension PUSH commands
Only try to queue commands if we have chart_payload_count>0
Remove the event loop shutdown opcode handle

* Improve detection of shutdown (check netdata_exit)

* Adjusting info messages

1 files changed, 0 insertions, 159 deletions

diff --git a/database/sqlite/sqlite_aclk_node.c b/database/sqlite/sqlite_aclk_node.c
index 4943b458da..3fb90d666c 100644
--- a/database/sqlite/sqlite_aclk_node.c
+++ b/database/sqlite/sqlite_aclk_node.c
@@ -5,7 +5,6 @@
 
 #ifdef ENABLE_NEW_CLOUD_PROTOCOL
 #include "../../aclk/aclk_charts_api.h"
-#include "../../aclk/aclk.h"
 #endif
 
 void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
@@ -63,161 +62,3 @@ void sql_build_node_info(struct aclk_database_worker_config *wc, struct aclk_dat
 
     return;
 }
-#define SQL_SELECT_HOST_MEMORY_MODE "select memory_mode from chart where host_id = @host_id limit 1;"
-static RRD_MEMORY_MODE sql_get_host_memory_mode(uuid_t *host_id)
-{
-    int rc;
-
-    RRD_MEMORY_MODE memory_mode = RRD_MEMORY_MODE_RAM;
-    sqlite3_stmt *res = NULL;
-
-    rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_MEMORY_MODE, -1, &res, 0);
-
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to prepare statement to read host memory mode");
-        return memory_mode;
-    }
-
-    rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to bind host parameter to fetch host memory mode");
-        goto failed;
-    }
-
-    while (sqlite3_step(res) == SQLITE_ROW) {
-        memory_mode = (RRD_MEMORY_MODE) sqlite3_column_int(res, 0);
-    }
-
-failed:
-    rc = sqlite3_finalize(res);
-    if (unlikely(rc != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when reading host memory mode");
-    return memory_mode;
-}
-
-#define SELECT_HOST_DIMENSION_LIST  "SELECT d.dim_id, c.update_every, c.type||'.'||c.id FROM chart c, dimension d, host h " \
-        "WHERE d.chart_id = c.chart_id AND c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
-
-#define SELECT_HOST_CHART_LIST  "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \
-        "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;"
-
-void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
-{
-    UNUSED(cmd);
-#ifdef ENABLE_NEW_CLOUD_PROTOCOL
-    int rc;
-
-    if (!aclk_use_new_cloud_arch || !aclk_connected)
-        return;
-
-    char *claim_id = is_agent_claimed();
-    if (unlikely(!claim_id))
-        return;
-
-    sqlite3_stmt *res = NULL;
-    RRD_MEMORY_MODE memory_mode;
-
-    uuid_t host_uuid;
-    rc = uuid_parse(wc->host_guid, host_uuid);
-    if (unlikely(rc)) {
-        freez(claim_id);
-        return;
-    }
-
-    if (wc->host)
-        memory_mode = wc->host->rrd_memory_mode;
-    else
-        memory_mode = sql_get_host_memory_mode(&host_uuid);
-
-    if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
-        rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0);
-    else
-        rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0);
-
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to prepare statement to fetch host dimensions");
-        freez(claim_id);
-        return;
-    }
-
-    rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to bind host parameter to fetch host dimensions");
-        goto failed;
-    }
-
-    time_t  start_time = LONG_MAX;
-    time_t  first_entry_t;
-    uint32_t update_every = 0;
-
-    struct retention_updated rotate_data;
-
-    memset(&rotate_data, 0, sizeof(rotate_data));
-
-    int max_intervals = 32;
-
-    rotate_data.interval_duration_count = 0;
-    rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations));
-
-    now_realtime_timeval(&rotate_data.rotation_timestamp);
-    rotate_data.memory_mode = memory_mode;
-    rotate_data.claim_id = claim_id;
-    rotate_data.node_id = strdupz(wc->node_id);
-
-    while (sqlite3_step(res) == SQLITE_ROW) {
-        if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) {
-            if (update_every) {
-                debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
-                rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
-                rotate_data.interval_duration_count++;
-            }
-            update_every = (uint32_t) sqlite3_column_int(res, 1);
-            rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every;
-            start_time = LONG_MAX;
-        }
-#ifdef ENABLE_DBENGINE
-        time_t  last_entry_t;
-        if (memory_mode == RRD_MEMORY_MODE_DBENGINE)
-            rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t);
-        else
-#endif
-        {
-            if (wc->host) {
-                RRDSET *st = NULL;
-                rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1;
-                if (!rc)
-                    first_entry_t = rrdset_first_entry_t(st);
-            }
-            else {
-                 rc = 0;
-                 first_entry_t = rotate_data.rotation_timestamp.tv_sec;
-            }
-        }
-
-        if (likely(!rc && first_entry_t))
-            start_time = MIN(start_time, first_entry_t);
-    }
-    if (update_every) {
-        debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time);
-        rotate_data.interval_durations[rotate_data.interval_duration_count].retention = rotate_data.rotation_timestamp.tv_sec - start_time;
-        rotate_data.interval_duration_count++;
-    }
-
-    for (int i = 0; i < rotate_data.interval_duration_count; ++i) {
-        debug(D_ACLK_SYNC,"%d --> Update %s for %u  Retention = %u", i, wc->host_guid,
-             rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention);
-    };
-    aclk_retention_updated(&rotate_data);
-    freez(rotate_data.node_id);
-    freez(rotate_data.interval_durations);
-
-failed:
-    freez(claim_id);
-    rc = sqlite3_finalize(res);
-    if (unlikely(rc != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when reading host dimensions");
-#else
-    UNUSED(wc);
-#endif
-    return;
-}
\ No newline at end of file