Improve agent to cloud synchronization performance (#12348)

* Switch to prepare statement when storing active charts / dimensions

* Switch to prepare statement when storing chart labels

* Switch to prepare statement when doing a node id lookup

* Switch to prepare statement when loading the node id for a host

* Improve performance by avoiding db query

* Use prepare statement when counting pending chart messages to send to the cloud

* Delay locking while preparing commands

* No need to use buffer, avoid memory allocation overhead

* Switch to prepare statement when loading pending chart updates to send to the cloud

3 files changed, 93 insertions, 95 deletions

diff --git a/aclk/aclk_query.c b/aclk/aclk_query.c
index 4bfa03cd65..b30e6be14b 100644
--- a/aclk/aclk_query.c
+++ b/aclk/aclk_query.c
@@ -58,6 +58,13 @@ static RRDHOST *node_id_2_rrdhost(const char *node_id)
 {
     int res;
     uuid_t node_id_bin, host_id_bin;
+
+    rrd_rdlock();
+    RRDHOST *host = find_host_by_node_id((char *) node_id);
+    rrd_unlock();
+    if (host)
+        return host;
+
     char host_id[UUID_STR_LEN];
     if (uuid_parse(node_id, node_id_bin)) {
         error("Couldn't parse UUID %s", node_id);
diff --git a/database/sqlite/sqlite_aclk_chart.c b/database/sqlite/sqlite_aclk_chart.c
index f6f3ebeb32..e57a26bc04 100644
--- a/database/sqlite/sqlite_aclk_chart.c
+++ b/database/sqlite/sqlite_aclk_chart.c
@@ -29,12 +29,11 @@ static int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payl
     int send_status = 0;
 
     if (unlikely(!res)) {
-        BUFFER *sql = buffer_create(1024);
-        buffer_sprintf(sql,"SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
+        char sql[ACLK_SYNC_QUERY_SIZE];
+        snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT 1 FROM aclk_chart_latest_%s acl, aclk_chart_payload_%s acp "
                             "WHERE acl.unique_id = acp.unique_id AND acl.uuid = @uuid AND acp.payload = @payload;",
                        uuid_str, uuid_str);
-        rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res);
-        buffer_free(sql);
+        rc = prepare_statement(db_meta, sql, &res);
         if (rc != SQLITE_OK) {
             error_report("Failed to prepare statement to check payload data");
             return 0;
@@ -70,14 +69,11 @@ static int aclk_add_chart_payload(struct aclk_database_worker_config *wc, uuid_t
         return 0;
 
     if (unlikely(!res_chart)) {
-        BUFFER *sql = buffer_create(1024);
-
-        buffer_sprintf(sql,"INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
-                "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);
-
-        rc = prepare_statement(db_meta, (char *) buffer_tostring(sql), &res_chart);
-        buffer_free(sql);
-
+        char sql[ACLK_SYNC_QUERY_SIZE];
+        snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,
+            "INSERT INTO aclk_chart_payload_%s (unique_id, uuid, claim_id, date_created, type, payload) " \
+            "VALUES (@unique_id, @uuid, @claim_id, strftime('%%s','now'), @type, @payload);", wc->uuid_str);
+        rc = prepare_statement(db_meta, sql, &res_chart);
         if (rc != SQLITE_OK) {
             error_report("Failed to prepare statement to store chart payload data");
             return 1;
@@ -320,21 +316,20 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
     uint64_t last_sequence;
     time_t last_timestamp = 0;
 
-    BUFFER *sql = buffer_create(1024);
-
-    sqlite3_stmt *res = NULL;
-
-    buffer_sprintf(sql, "SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid  " \
-    "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \
-    "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \
-    "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);
+    char sql[ACLK_SYNC_QUERY_SIZE];
+    static __thread sqlite3_stmt *res = NULL;
 
-    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
-    if (rc != SQLITE_OK) {
-        error_report("Failed to prepare statement when trying to send a chart update via ACLK");
-        buffer_free(sql);
-        freez(claim_id);
-        return;
+    if (unlikely(!res)) {
+        snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"SELECT ac.sequence_id, acp.payload, ac.date_created, ac.type, ac.uuid  " \
+             "FROM aclk_chart_%s ac, aclk_chart_payload_%s acp " \
+             "WHERE ac.date_submitted IS NULL AND ac.unique_id = acp.unique_id AND ac.update_count > 0 " \
+             "AND acp.claim_id = @claim_id ORDER BY ac.sequence_id ASC LIMIT %d;", wc->uuid_str, wc->uuid_str, limit);
+        rc = prepare_statement(db_meta, sql, &res);
+        if (rc != SQLITE_OK) {
+            error_report("Failed to prepare statement when trying to send a chart update via ACLK");
+            freez(claim_id);
+            return;
+        }
     }
 
     rc = sqlite3_bind_blob(res, 1, claim_uuid , sizeof(claim_uuid), SQLITE_STATIC);
@@ -388,21 +383,18 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
             error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
 
         if (likely(first_sequence)) {
-            buffer_flush(sql);
 
             db_lock();
-            buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "
+            snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=strftime('%%s','now') "
                                 "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";",
                                 wc->uuid_str, first_sequence, last_sequence);
-            db_execute(buffer_tostring(sql));
-
-            buffer_flush(sql);
-            buffer_sprintf(sql, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
+            db_execute(sql);
+            snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "INSERT OR REPLACE INTO aclk_chart_latest_%s (uuid, unique_id, date_submitted) "
                                 " SELECT uuid, unique_id, date_submitted FROM aclk_chart_%s s "
                                 " WHERE date_submitted IS NOT NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64
                                 " ;",
                                 wc->uuid_str, wc->uuid_str, first_sequence, last_sequence);
-            db_execute(buffer_tostring(sql));
+            db_execute(sql);
             db_unlock();
 
             aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id);
@@ -436,11 +428,10 @@ void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_d
     freez(is_dim);
 
 bind_fail:
-    rc = sqlite3_finalize(res);
+    rc = sqlite3_reset(res);
     if (unlikely(rc != SQLITE_OK))
-        error_report("Failed to finalize statement when pushing chart events, rc = %d", rc);
+        error_report("Failed to reset statement when pushing chart events, rc = %d", rc);
 
-    buffer_free(sql);
     freez(claim_id);
     return;
 }
@@ -521,15 +512,15 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_
 
     log_access("IN [%s (%s)]: Received ack chart sequence id %"PRIu64, wc->node_id, wc->host ? wc->host->hostname : "N/A", cmd.param1);
 
-    BUFFER *sql = buffer_create(1024);
+    char sql[ACLK_SYNC_QUERY_SIZE];
 
-    buffer_sprintf(sql, "UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "
-                        "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);
+    snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1,"UPDATE aclk_chart_%s SET date_updated=strftime('%%s','now') WHERE sequence_id <= @sequence_id "
+            "AND date_submitted IS NOT NULL AND date_updated IS NULL;", wc->uuid_str);
 
-    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
+    rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
     if (rc != SQLITE_OK) {
-        error_report("Failed to prepare statement count sequence ids in the database");
-        goto prepare_fail;
+        error_report("Failed to prepare statement to ack chart sequence ids");
+        return;
     }
 
     rc = sqlite3_bind_int64(res, 1, (uint64_t) cmd.param1);
@@ -540,13 +531,10 @@ void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_
     if (rc != SQLITE_DONE)
         error_report("Failed to ACK sequence id, rc = %d", rc);
 
-    bind_fail:
-        if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
-            error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);
-
-        prepare_fail:
-        buffer_free(sql);
-        return;
+bind_fail:
+    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
+        error_report("Failed to finalize statement to ACK older sequence ids, rc = %d", rc);
+    return;
 }
 
 void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
@@ -556,12 +544,11 @@ void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct acl
                    wc->uuid_str, cmd.param1);
     db_execute(buffer_tostring(sql));
     if (cmd.param1 == 1) {
-        db_lock();
         buffer_flush(sql);
         log_access("IN [%s (%s)]: Received chart full resync.", wc->node_id, wc->host ? wc->host->hostname : "N/A");
         buffer_sprintf(sql, "DELETE FROM aclk_chart_payload_%s; DELETE FROM aclk_chart_%s; " \
                             "DELETE FROM aclk_chart_latest_%s;", wc->uuid_str, wc->uuid_str, wc->uuid_str);
-
+        db_lock();
         db_execute("BEGIN TRANSACTION;");
         db_execute(buffer_tostring(sql));
         db_execute("COMMIT TRANSACTION;");
@@ -909,43 +896,43 @@ failed:
 
 uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc)
 {
-    BUFFER *sql = buffer_create(1024);
-    sqlite3_stmt *res = NULL;
+    char sql[ACLK_SYNC_QUERY_SIZE];
+    static __thread sqlite3_stmt *res = NULL;
 
-    buffer_sprintf(sql,"SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);
+    snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT count(1) FROM aclk_chart_%s ac WHERE ac.date_submitted IS NULL;", wc->uuid_str);
 
     int rc;
     uint32_t chart_payload_count = 0;
-    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
-    if (rc != SQLITE_OK) {
-        error_report("Failed to prepare statement to count pending messages");
-        goto fail;
+    if (unlikely(!res)) {
+        rc = prepare_statement(db_meta, sql, &res);
+        if (rc != SQLITE_OK) {
+            error_report("Failed to prepare statement to count pending messages");
+            return 0;
+        }
     }
     while (sqlite3_step(res) == SQLITE_ROW)
         chart_payload_count = (uint32_t) sqlite3_column_int(res, 0);
 
-    rc = sqlite3_finalize(res);
+    rc = sqlite3_reset(res);
     if (unlikely(rc != SQLITE_OK))
         error_report("Failed to reset statement when fetching pending messages, rc = %d", rc);
 
-fail:
-    buffer_free(sql);
     return chart_payload_count;
 }
 
 void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)
 {
-    BUFFER *sql = buffer_create(1024);
+    char sql[ACLK_SYNC_QUERY_SIZE];
 
-    buffer_sprintf(sql,"SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \
-                        "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);
+    snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT ac.sequence_id, ac.date_created FROM aclk_chart_%s ac " \
+        "WHERE ac.date_submitted IS NOT NULL ORDER BY ac.sequence_id DESC LIMIT 1;", wc->uuid_str);
 
     int rc;
     sqlite3_stmt *res = NULL;
-    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
+    rc = sqlite3_prepare_v2(db_meta, sql, -1, &res, 0);
     if (rc != SQLITE_OK) {
         error_report("Failed to prepare statement to find last chart sequence id");
-        goto fail;
+        return;
     }
 
     wc->chart_sequence_id = 0;
@@ -961,8 +948,6 @@ void sql_get_last_chart_sequence(struct aclk_database_worker_config *wc)
     if (unlikely(rc != SQLITE_OK))
         error_report("Failed to reset statement when fetching chart sequence info, rc = %d", rc);
 
-fail:
-    buffer_free(sql);
     return;
 }
 
diff --git a/database/sqlite/sqlite_functions.c b/database/sqlite/sqlite_functions.c
index bc1da9fb53..4c811624a1 100644
--- a/database/sqlite/sqlite_functions.c
+++ b/database/sqlite/sqlite_functions.c
@@ -121,7 +121,7 @@ static int store_active_uuid_object(sqlite3_stmt **res, char *statement, uuid_t
 
     // Check if we should need to prepare the statement
     if (!*res) {
-        rc = sqlite3_prepare_v2(db_meta, statement, -1, res, 0);
+        rc = prepare_statement(db_meta, statement, res);
         if (unlikely(rc != SQLITE_OK)) {
             error_report("Failed to prepare statement to store active object, rc = %d", rc);
             return rc;
@@ -142,7 +142,7 @@ static int store_active_uuid_object(sqlite3_stmt **res, char *statement, uuid_t
  */
 void store_active_chart(uuid_t *chart_uuid)
 {
-    sqlite3_stmt *res = NULL;
+    static __thread sqlite3_stmt *res = NULL;
     int rc;
 
     if (unlikely(!db_meta)) {
@@ -158,7 +158,7 @@ void store_active_chart(uuid_t *chart_uuid)
     if (rc != SQLITE_DONE)
         error_report("Failed to store active chart, rc = %d", rc);
 
-    rc = sqlite3_finalize(res);
+    rc = sqlite3_reset(res);
     if (unlikely(rc != SQLITE_OK))
         error_report("Failed to finalize statement in store active chart, rc = %d", rc);
     return;
@@ -170,7 +170,7 @@ void store_active_chart(uuid_t *chart_uuid)
  */
 void store_active_dimension(uuid_t *dimension_uuid)
 {
-    sqlite3_stmt *res = NULL;
+    static __thread sqlite3_stmt *res = NULL;
     int rc;
 
     if (unlikely(!db_meta)) {
@@ -186,7 +186,7 @@ void store_active_dimension(uuid_t *dimension_uuid)
     if (rc != SQLITE_DONE)
         error_report("Failed to store active dimension, rc = %d", rc);
 
-    rc = sqlite3_finalize(res);
+    rc = sqlite3_reset(res);
     if (unlikely(rc != SQLITE_OK))
         error_report("Failed to finalize statement in store active dimension, rc = %d", rc);
     return;
@@ -1335,7 +1335,7 @@ void add_migrated_file(char *path, uint64_t file_size)
 
 void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, char *value)
 {
-    sqlite3_stmt *res = NULL;
+    static __thread sqlite3_stmt *res = NULL;
     int rc;
 
     if (unlikely(!db_meta)) {
@@ -1344,10 +1344,12 @@ void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, cha
         return;
     }
 
-    rc = sqlite3_prepare_v2(db_meta, SQL_INS_CHART_LABEL, -1, &res, 0);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to prepare statement store chart labels");
-        return;
+    if (unlikely(!res)) {
+        rc = prepare_statement(db_meta, SQL_INS_CHART_LABEL, &res);
+        if (unlikely(rc != SQLITE_OK)) {
+            error_report("Failed to prepare statement store chart labels");
+            return;
+        }
     }
 
     rc = sqlite3_bind_blob(res, 1, chart_uuid, sizeof(*chart_uuid), SQLITE_STATIC);
@@ -1379,8 +1381,8 @@ void sql_store_chart_label(uuid_t *chart_uuid, int source_type, char *label, cha
         error_report("Failed to store chart label entry, rc = %d", rc);
 
 failed:
-    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when storing chart label information");
+    if (unlikely(sqlite3_reset(res) != SQLITE_OK))
+        error_report("Failed to reset the prepared statement when storing chart label information");
 
     return;
 }
@@ -1881,7 +1883,7 @@ failed:
 
 int get_host_id(uuid_t *node_id, uuid_t *host_id)
 {
-    sqlite3_stmt *res = NULL;
+    static __thread sqlite3_stmt *res = NULL;
     int rc;
 
     if (unlikely(!db_meta)) {
@@ -1890,10 +1892,12 @@ int get_host_id(uuid_t *node_id, uuid_t *host_id)
         return 1;
     }
 
-    rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_NODE_ID, -1, &res, 0);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to prepare statement to select node instance information for a node");
-        return 1;
+    if (unlikely(!res)) {
+        rc = prepare_statement(db_meta, SQL_SELECT_HOST_BY_NODE_ID, &res);
+        if (unlikely(rc != SQLITE_OK)) {
+            error_report("Failed to prepare statement to select node instance information for a node");
+            return 1;
+        }
     }
 
     rc = sqlite3_bind_blob(res, 1, node_id, sizeof(*node_id), SQLITE_STATIC);
@@ -1907,8 +1911,8 @@ int get_host_id(uuid_t *node_id, uuid_t *host_id)
         uuid_copy(*host_id, *((uuid_t *) sqlite3_column_blob(res, 0)));
 
 failed:
-    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when selecting node instance information");
+    if (unlikely(sqlite3_reset(res) != SQLITE_OK))
+        error_report("Failed to reset the prepared statement when selecting node instance information");
 
     return (rc == SQLITE_ROW) ? 0 : -1;
 }
@@ -2060,7 +2064,7 @@ failed:
 
 void sql_load_node_id(RRDHOST *host)
 {
-    sqlite3_stmt *res = NULL;
+    static __thread sqlite3_stmt *res = NULL;
     int rc;
 
     if (unlikely(!db_meta)) {
@@ -2069,11 +2073,13 @@ void sql_load_node_id(RRDHOST *host)
         return;
     }
 
-    rc = sqlite3_prepare_v2(db_meta, SQL_GET_HOST_NODE_ID, -1, &res, 0);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to prepare statement to fetch node id");
-        return;
-    };
+    if (unlikely(!res)) {
+        rc = prepare_statement(db_meta, SQL_GET_HOST_NODE_ID, &res);
+        if (unlikely(rc != SQLITE_OK)) {
+            error_report("Failed to prepare statement to fetch node id");
+            return;
+        };
+    }
 
     rc = sqlite3_bind_blob(res, 1, &host->host_uuid, sizeof(host->host_uuid), SQLITE_STATIC);
     if (unlikely(rc != SQLITE_OK)) {
@@ -2090,8 +2096,8 @@ void sql_load_node_id(RRDHOST *host)
     }
 
 failed:
-    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when loading node instance information");
+    if (unlikely(sqlite3_reset(res) != SQLITE_OK))
+        error_report("Failed to reset the prepared statement when loading node instance information");
 
     return;
 };