Add a thread to asynchronously process metadata updates (#13783)

* Remove old metalog text fle processing

* Add metadata event loop

* Move functions from sqlite_functions.c to sqlite_metadata.c
Queue updates to the metadata event loop
Migration to remove unused tables
Cleanup unused functions

* Queue chart labels to metadata

* Store chart labels to metadata

* During shutdown, run full speed

* Add shutdown prepare
Handle SHUTDOWN in the cmd queue function
Add worker thread to handle host/chart/dimension metadata doing dictionary traversals

* Remove unused RRDIM_FLAG_ACLK
Add flags to trigger host/chart/dimension metadata processing

* Incremental processing of chart metadata writes

* Store host labels

* Remove redundant return statements

* Change unit tests / cleanup

* Fix rescheduling

* Schedule chart labels update by setting the RRDSET_FLAG_METADATA_UPDATE flag

* Queue commands to update metadata for dimension and host labels

* Make sure we do a final scan to store metadata during shutdown (if needed)

* Remove unused structures
Adjust queue size since we do batch processing of updates without queueing individual messages
Remove pragma mmap for now
Fix memory leak during sqlite unittest (minor)

* Dont update if we are in archive mode

* Cleanup

* Build entire message payload and store

* Initialize worker completion properly

* Properly skip host check for pending metadata updates

* Report bind param failures
Add worker request inside the data payload
Initialize variables to silence warnings
Rebase on master

* Report the chart id (not the dimension) and the dimension id when storing a dimension

* Compilation warnings in 32bit

* Add DEFINE for the queries

* Remove commented out code

* * Remove items parameter from unitest
* Remove commented out code
* sqlite_metadata.h contains only public items
* Use sleep_usec instead of usleep
* Rename metadata_database_init_cmd_queue to metadata_init_cmd_queue
* Rename metadata_database_enq_cmd_noblock to metadata_enq_cmd_noblock

1 files changed, 0 insertions, 42 deletions

diff --git a/database/sqlite/sqlite_aclk.c b/database/sqlite/sqlite_aclk.c
index 03edd6bc8a..18e997bc71 100644
--- a/database/sqlite/sqlite_aclk.c
+++ b/database/sqlite/sqlite_aclk.c
@@ -5,28 +5,12 @@
 
 #include "sqlite_aclk_node.h"
 
-#ifdef ENABLE_ACLK
-#include "../../aclk/aclk.h"
-#endif
-
 void sanity_check(void) {
     // make sure the compiler will stop on misconfigurations
     BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
 }
 
 const char *aclk_sync_config[] = {
-    "CREATE TABLE IF NOT EXISTS dimension_delete (dimension_id blob, dimension_name text, chart_type_id text, "
-    "dim_id blob, chart_id blob, host_id blob, date_created);",
-
-    "CREATE INDEX IF NOT EXISTS ind_h1 ON dimension_delete (host_id);",
-
-    "CREATE TRIGGER IF NOT EXISTS tr_dim_del AFTER DELETE ON dimension BEGIN INSERT INTO dimension_delete "
-    "(dimension_id, dimension_name, chart_type_id, dim_id, chart_id, host_id, date_created)"
-    " select old.id, old.name, c.type||\".\"||c.id, old.dim_id, old.chart_id, c.host_id, unixepoch() FROM"
-    " chart c WHERE c.chart_id = old.chart_id; END;",
-
-    "DELETE FROM dimension_delete WHERE host_id NOT IN"
-    " (SELECT host_id FROM host) OR unixepoch() - date_created > 604800;",
 
     NULL,
 };
@@ -174,25 +158,6 @@ struct aclk_database_cmd aclk_database_deq_cmd(struct aclk_database_worker_confi
     return ret;
 }
 
-int aclk_worker_enq_cmd(char *node_id, struct aclk_database_cmd *cmd)
-{
-    if (unlikely(!node_id || !cmd))
-        return 0;
-
-    uv_mutex_lock(&aclk_async_lock);
-    struct aclk_database_worker_config *wc = aclk_thread_head;
-
-    while (wc) {
-        if (!strcmp(wc->node_id, node_id))
-            break;
-        wc = wc->next;
-    }
-    uv_mutex_unlock(&aclk_async_lock);
-    if (wc)
-        aclk_database_enq_cmd(wc, cmd);
-    return (wc == NULL);
-}
-
 struct aclk_database_worker_config *find_inactive_wc_by_node_id(char *node_id)
 {
     if (unlikely(!node_id))
@@ -466,9 +431,7 @@ void aclk_database_worker(void *arg)
     timer_req.data = wc;
     fatal_assert(0 == uv_timer_start(&timer_req, timer_cb, TIMER_PERIOD_MS, TIMER_PERIOD_MS));
 
-//    wc->retry_count = 0;
     wc->node_info_send = 1;
-//    aclk_add_worker_thread(wc);
     info("Starting ACLK sync thread for host %s -- scratch area %lu bytes", wc->host_guid, (unsigned long int) sizeof(*wc));
 
     memset(&cmd, 0, sizeof(cmd));
@@ -688,9 +651,7 @@ void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
     wc->host = host;
     strcpy(wc->uuid_str, uuid_str);
     strcpy(wc->host_guid, host_guid);
-//    wc->chart_updates = 0;
     wc->alert_updates = 0;
-//    wc->retry_count = 0;
     aclk_database_init_cmd_queue(wc);
     aclk_add_worker_thread(wc);
     fatal_assert(0 == uv_thread_create(&(wc->thread), aclk_database_worker, wc));
@@ -836,8 +797,5 @@ void sql_check_aclk_table_list(struct aclk_database_worker_config *wc)
         error_report("Query failed when trying to check for obsolete ACLK sync tables, %s", err_msg);
         sqlite3_free(err_msg);
     }
-    db_execute("DELETE FROM dimension_delete WHERE host_id NOT IN (SELECT host_id FROM host) "
-               " OR unixepoch() - date_created > 604800;");
-
     return;
 }