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

CMakeLists.txt

@@ -748,6 +748,8 @@ set(RRD_PLUGIN_FILES
         database/storage_engine.h
         database/ram/rrddim_mem.c
         database/ram/rrddim_mem.h
+        database/sqlite/sqlite_metadata.c
+        database/sqlite/sqlite_metadata.h
         database/sqlite/sqlite_functions.c
         database/sqlite/sqlite_functions.h
 		database/sqlite/sqlite_context.c
@@ -779,16 +781,6 @@ set(RRD_PLUGIN_FILES
         database/engine/pagecache.h
         database/engine/rrdenglocking.c
         database/engine/rrdenglocking.h
-        database/engine/metadata_log/metadatalog.h
-        database/engine/metadata_log/metadatalogapi.c
-        database/engine/metadata_log/metadatalogapi.h
-        database/engine/metadata_log/logfile.h
-        database/engine/metadata_log/logfile.c
-        database/engine/metadata_log/metadatalogprotocol.h
-        database/engine/metadata_log/metalogpluginsd.c
-        database/engine/metadata_log/metalogpluginsd.h
-        database/engine/metadata_log/compaction.c
-        database/engine/metadata_log/compaction.h
         database/KolmogorovSmirnovDist.c
         database/KolmogorovSmirnovDist.h
         )

Makefile.am

@@ -456,7 +456,9 @@ RRD_PLUGIN_FILES = \
     database/sqlite/sqlite_db_migration.c \
     database/sqlite/sqlite_db_migration.h \
     database/sqlite/sqlite_aclk.c \
-    database/sqlite/sqlite_aclk.h \
+	database/sqlite/sqlite_aclk.h \
+    database/sqlite/sqlite_metadata.c \
+    database/sqlite/sqlite_metadata.h \
     database/sqlite/sqlite_health.c \
     database/sqlite/sqlite_health.h \
     database/sqlite/sqlite_aclk_node.c \
@@ -547,16 +549,6 @@ if ENABLE_DBENGINE
         database/engine/pagecache.h \
         database/engine/rrdenglocking.c \
         database/engine/rrdenglocking.h \
-        database/engine/metadata_log/metadatalog.h \
-        database/engine/metadata_log/metadatalogapi.c \
-        database/engine/metadata_log/metadatalogapi.h \
-        database/engine/metadata_log/logfile.h \
-        database/engine/metadata_log/logfile.c \
-        database/engine/metadata_log/metadatalogprotocol.h \
-        database/engine/metadata_log/metalogpluginsd.c \
-        database/engine/metadata_log/metalogpluginsd.h \
-        database/engine/metadata_log/compaction.c \
-        database/engine/metadata_log/compaction.h \
         $(NULL)
 endif
 

claim/claim.c

@@ -170,7 +170,7 @@ void load_claiming_state(void)
     }
 
     invalidate_node_instances(&localhost->host_uuid, claimed_id ? &uuid : NULL);
-    store_claim_id(&localhost->host_uuid, claimed_id ? &uuid : NULL);
+    metaqueue_store_claim_id(&localhost->host_uuid, claimed_id ? &uuid : NULL);
 
     rrdhost_aclk_state_unlock(localhost);
     if (!claimed_id) {

collectors/plugins.d/pluginsd_parser.c

@@ -80,7 +80,6 @@ PARSER_RC pluginsd_begin(char **words, void *user, PLUGINSD_ACTION  *plugins_act
         } else
             rrdset_next(st);
     }
-
     return PARSER_RC_OK;
 disable:
     ((PARSER_USER_OBJECT *)user)->enabled = 0;
@@ -306,14 +305,14 @@ PARSER_RC pluginsd_dimension(char **words, void *user, PLUGINSD_ACTION  *plugins
     if (likely(unhide_dimension)) {
         rrddim_option_clear(rd, RRDDIM_OPTION_HIDDEN);
         if (rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN)) {
-            (void)sql_set_dimension_option(&rd->metric_uuid, NULL);
             rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
+            metaqueue_dimension_update_flags(rd);
         }
     } else {
         rrddim_option_set(rd, RRDDIM_OPTION_HIDDEN);
         if (!rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN)) {
-            (void)sql_set_dimension_option(&rd->metric_uuid, "hidden");
             rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
+            metaqueue_dimension_update_flags(rd);
         }
     }
 
@@ -639,7 +638,6 @@ PARSER_RC pluginsd_variable(char **words, void *user, PLUGINSD_ACTION  *plugins_
             error("cannot find/create CHART VARIABLE '%s' on host '%s', chart '%s'", name, rrdhost_hostname(host), rrdset_id(st));
     }
 
-
     return PARSER_RC_OK;
 }
 
@@ -713,7 +711,7 @@ PARSER_RC pluginsd_overwrite(char **words, void *user, PLUGINSD_ACTION  *plugins
         host->rrdlabels = rrdlabels_create();
 
     rrdlabels_migrate_to_these(host->rrdlabels, (DICTIONARY *) (((PARSER_USER_OBJECT *)user)->new_host_labels));
-    sql_store_host_labels(host);
+    metaqueue_store_host_labels(host->machine_guid);
 
     rrdlabels_destroy(((PARSER_USER_OBJECT *)user)->new_host_labels);
     ((PARSER_USER_OBJECT *)user)->new_host_labels = NULL;
@@ -743,6 +741,11 @@ PARSER_RC pluginsd_clabel_commit(char **words, void *user, PLUGINSD_ACTION  *plu
     UNUSED(words);
 
     RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
+    RRDSET *st = ((PARSER_USER_OBJECT *)user)->st;
+
+    if (unlikely(!st))
+        return PARSER_RC_OK;
+
     debug(D_PLUGINSD, "requested to commit chart labels");
 
     if(!((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily) {
@@ -752,6 +755,9 @@ PARSER_RC pluginsd_clabel_commit(char **words, void *user, PLUGINSD_ACTION  *plu
 
     rrdlabels_remove_all_unmarked(((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily);
 
+    rrdset_flag_set(st, RRDSET_FLAG_METADATA_UPDATE);
+    rrdhost_flag_set(st->rrdhost, RRDHOST_FLAG_METADATA_UPDATE);
+
     ((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily = NULL;
     return PARSER_RC_OK;
 }

configure.ac

@@ -1713,7 +1713,6 @@ AC_CONFIG_FILES([
     daemon/Makefile
     database/Makefile
     database/engine/Makefile
-    database/engine/metadata_log/Makefile
     database/ram/Makefile
     database/sqlite/Makefile
     diagrams/Makefile

daemon/global_statistics.c

@@ -1832,6 +1832,7 @@ static struct worker_utilization all_workers_utilization[] = {
     { .name = "WEB",         .family = "workers web server",              .priority = 1000000 },
     { .name = "ACLKQUERY",   .family = "workers aclk query",              .priority = 1000000 },
     { .name = "ACLKSYNC",    .family = "workers aclk host sync",          .priority = 1000000 },
+    { .name = "METASYNC",    .family = "workers metadata sync",           .priority = 1000000 },
     { .name = "PLUGINSD",    .family = "workers plugins.d",               .priority = 1000000 },
     { .name = "STATSD",      .family = "workers plugin statsd",           .priority = 1000000 },
     { .name = "STATSDFLUSH", .family = "workers plugin statsd flush",     .priority = 1000000 },

daemon/main.c

@@ -60,7 +60,9 @@ void netdata_cleanup_and_exit(int ret) {
                 rrdeng_prepare_exit(multidb_ctx[tier]);
         }
 #endif
+        metadata_sync_shutdown_prepare();
         rrdhost_free_all();
+        metadata_sync_shutdown();
 #ifdef ENABLE_DBENGINE
         if(dbengine_enabled) {
             for (int tier = 0; tier < storage_tiers; tier++)
@@ -1039,6 +1041,9 @@ int main(int argc, char **argv) {
                         else if(strcmp(optarg, "rrdlabelstest") == 0) {
                             return rrdlabels_unittest();
                         }
+                        else if(strcmp(optarg, "metatest") == 0) {
+                            return metadata_unittest();
+                        }
                         else if(strncmp(optarg, createdataset_string, strlen(createdataset_string)) == 0) {
                             optarg += strlen(createdataset_string);
                             unsigned history_seconds = strtoul(optarg, NULL, 0);

daemon/service.c

@@ -60,7 +60,7 @@ static void svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
 
         if (tiers_available == tiers_said_yes && tiers_said_yes) {
             /* This metric has no data and no references */
-            delete_dimension_uuid(&rd->metric_uuid);
+            metaqueue_delete_dimension_uuid(&rd->metric_uuid);
         }
         else {
             /* Do not delete this dimension */

daemon/unit_test.c

@@ -1564,8 +1564,10 @@ int test_sqlite(void) {
 
     buffer_free(sql);
     fprintf(stderr,"SQLite is OK\n");
+    rc = sqlite3_close_v2(db_meta);
     return 0;
 error:
+    rc = sqlite3_close_v2(db_meta);
     fprintf(stderr,"SQLite statement failed: %s\n", buffer_tostring(sql));
     buffer_free(sql);
     fprintf(stderr,"SQLite tests failed\n");

database/engine/Makefile.am

@@ -4,7 +4,6 @@ AUTOMAKE_OPTIONS = subdir-objects
 MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
 
 SUBDIRS = \
-    metadata_log \
     $(NULL)
 
 dist_noinst_DATA = \