Adding upstream version 1.38.0.upstream/1.38.0

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 154 insertions, 182 deletions

diff --git a/database/sqlite/sqlite_aclk.c b/database/sqlite/sqlite_aclk.c
index 7e3a9b2eb..3b0c40522 100644
--- a/database/sqlite/sqlite_aclk.c
+++ b/database/sqlite/sqlite_aclk.c
@@ -10,10 +10,140 @@ void sanity_check(void) {
     BUILD_BUG_ON(WORKER_UTILIZATION_MAX_JOB_TYPES < ACLK_MAX_ENUMERATIONS_DEFINED);
 }
 
-const char *aclk_sync_config[] = {
+static int sql_check_aclk_table(void *data, int argc, char **argv, char **column)
+{
+    struct aclk_database_worker_config *wc = data;
+    UNUSED(argc);
+    UNUSED(column);
 
-    NULL,
-};
+    debug(D_ACLK_SYNC,"Scheduling aclk sync table check for node %s", (char *) argv[0]);
+    struct aclk_database_cmd cmd;
+    memset(&cmd, 0, sizeof(cmd));
+    cmd.opcode = ACLK_DATABASE_DELETE_HOST;
+    cmd.data = strdupz((char *) argv[0]);
+    aclk_database_enq_cmd_noblock(wc, &cmd);
+    return 0;
+}
+
+#define SQL_SELECT_ACLK_ACTIVE_LIST "SELECT REPLACE(SUBSTR(name,19),'_','-') FROM sqlite_schema " \
+        "WHERE name LIKE 'aclk_chart_latest_%' AND type IN ('table');"
+
+static void sql_check_aclk_table_list(struct aclk_database_worker_config *wc)
+{
+    char *err_msg = NULL;
+    debug(D_ACLK_SYNC,"Cleaning tables for nodes that do not exist");
+    int rc = sqlite3_exec_monitored(db_meta, SQL_SELECT_ACLK_ACTIVE_LIST, sql_check_aclk_table, (void *) wc, &err_msg);
+    if (rc != SQLITE_OK) {
+        error_report("Query failed when trying to check for obsolete ACLK sync tables, %s", err_msg);
+        sqlite3_free(err_msg);
+    }
+}
+
+static void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
+{
+    UNUSED(cmd);
+
+    debug(D_ACLK, "Checking database for %s", wc->host_guid);
+
+    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE, &netdata_buffers_statistics.buffers_sqlite);
+
+    buffer_sprintf(sql,"DELETE FROM aclk_alert_%s WHERE date_submitted IS NOT NULL AND "
+                        "CAST(date_cloud_ack AS INT) < unixepoch()-%d;", wc->uuid_str, ACLK_DELETE_ACK_ALERTS_INTERNAL);
+    db_execute(buffer_tostring(sql));
+
+    buffer_free(sql);
+}
+
+
+#define SQL_SELECT_HOST_BY_UUID  "SELECT host_id FROM host WHERE host_id = @host_id;"
+
+static int is_host_available(uuid_t *host_id)
+{
+    sqlite3_stmt *res = NULL;
+    int rc;
+
+    if (unlikely(!db_meta)) {
+        if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
+            error_report("Database has not been initialized");
+        return 1;
+    }
+
+    rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_UUID, -1, &res, 0);
+    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, host_id, sizeof(*host_id), SQLITE_STATIC);
+    if (unlikely(rc != SQLITE_OK)) {
+        error_report("Failed to bind host_id parameter to select node instance information");
+        goto failed;
+    }
+    rc = sqlite3_step_monitored(res);
+
+failed:
+    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
+        error_report("Failed to finalize the prepared statement when checking host existence");
+
+    return (rc == SQLITE_ROW);
+}
+
+// OPCODE: ACLK_DATABASE_DELETE_HOST
+void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
+{
+    UNUSED(wc);
+    char uuid_str[GUID_LEN + 1];
+    char host_str[GUID_LEN + 1];
+
+    int rc;
+    uuid_t host_uuid;
+    char *host_guid = (char *)cmd.data;
+
+    if (unlikely(!host_guid))
+        return;
+
+    rc = uuid_parse(host_guid, host_uuid);
+    freez(host_guid);
+    if (rc)
+        return;
+
+    uuid_unparse_lower(host_uuid, host_str);
+    uuid_unparse_lower_fix(&host_uuid, uuid_str);
+
+    debug(D_ACLK_SYNC, "Checking if I should delete aclk tables for node %s", host_str);
+
+    if (is_host_available(&host_uuid)) {
+        debug(D_ACLK_SYNC, "Host %s exists, not deleting aclk sync tables", host_str);
+        return;
+    }
+
+    debug(D_ACLK_SYNC, "Host %s does NOT exist, can delete aclk sync tables", host_str);
+
+    sqlite3_stmt *res = NULL;
+    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE, &netdata_buffers_statistics.buffers_sqlite);
+
+    buffer_sprintf(sql,"SELECT 'drop '||type||' IF EXISTS '||name||';' FROM sqlite_schema " \
+                        "WHERE name LIKE 'aclk_%%_%s' AND type IN ('table', 'trigger', 'index');", uuid_str);
+
+    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
+    if (rc != SQLITE_OK) {
+        error_report("Failed to prepare statement to clean up aclk tables");
+        goto fail;
+    }
+    buffer_flush(sql);
+
+    while (sqlite3_step_monitored(res) == SQLITE_ROW)
+        buffer_strcat(sql, (char *) sqlite3_column_text(res, 0));
+
+    rc = sqlite3_finalize(res);
+    if (unlikely(rc != SQLITE_OK))
+        error_report("Failed to finalize statement to clean up aclk tables, rc = %d", rc);
+
+    db_execute(buffer_tostring(sql));
+
+fail:
+    buffer_free(sql);
+}
 
 uv_mutex_t aclk_async_lock;
 struct aclk_database_worker_config  *aclk_thread_head = NULL;
@@ -38,7 +168,6 @@ void aclk_add_worker_thread(struct aclk_database_worker_config *wc)
         aclk_thread_head = wc;
     }
     uv_mutex_unlock(&aclk_async_lock);
-    return;
 }
 
 void aclk_del_worker_thread(struct aclk_database_worker_config *wc)
@@ -53,7 +182,6 @@ void aclk_del_worker_thread(struct aclk_database_worker_config *wc)
     if (*tmp)
         *tmp = wc->next;
     uv_mutex_unlock(&aclk_async_lock);
-    return;
 }
 
 int aclk_worker_thread_exists(char *guid)
@@ -199,7 +327,6 @@ void aclk_sync_exit_all()
     uv_mutex_unlock(&aclk_async_lock);
 }
 
-#ifdef ENABLE_ACLK
 enum {
     IDX_HOST_ID,
     IDX_HOSTNAME,
@@ -228,6 +355,8 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
     uuid_unparse_lower(*(uuid_t *)argv[IDX_HOST_ID], guid);
 
     struct rrdhost_system_info *system_info = callocz(1, sizeof(struct rrdhost_system_info));
+    __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(struct rrdhost_system_info), __ATOMIC_RELAXED);
+
     system_info->hops = str2i((const char *) argv[IDX_HOPS]);
 
     sql_build_host_system_info((uuid_t *)argv[IDX_HOST_ID], system_info);
@@ -268,9 +397,9 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
 #endif
     return 0;
 }
-#endif
 
-int aclk_start_sync_thread(void *data, int argc, char **argv, char **column)
+#ifdef ENABLE_ACLK
+static int aclk_start_sync_thread(void *data, int argc, char **argv, char **column)
 {
     char uuid_str[GUID_LEN + 1];
     UNUSED(data);
@@ -286,10 +415,9 @@ int aclk_start_sync_thread(void *data, int argc, char **argv, char **column)
     sql_create_aclk_table(host, (uuid_t *) argv[0], (uuid_t *) argv[1]);
     return 0;
 }
-
+#endif
 void sql_aclk_sync_init(void)
 {
-#ifdef ENABLE_ACLK
     char *err_msg = NULL;
     int rc;
 
@@ -301,21 +429,7 @@ void sql_aclk_sync_init(void)
         return;
     }
 
-    info("SQLite aclk sync initialization");
-
-    for (int i = 0; aclk_sync_config[i]; i++) {
-        debug(D_ACLK_SYNC, "Executing %s", aclk_sync_config[i]);
-        rc = sqlite3_exec_monitored(db_meta, aclk_sync_config[i], 0, 0, &err_msg);
-        if (rc != SQLITE_OK) {
-            error_report("SQLite error aclk sync initialization setup, rc = %d (%s)", rc, err_msg);
-            error_report("SQLite failed statement %s", aclk_sync_config[i]);
-            sqlite3_free(err_msg);
-            return;
-        }
-    }
-    info("SQLite aclk sync initialization completed");
-    fatal_assert(0 == uv_mutex_init(&aclk_async_lock));
-
+    info("Creating archived hosts");
     rc = sqlite3_exec_monitored(db_meta, "SELECT host_id, hostname, registry_hostname, update_every, os, "
        "timezone, tags, hops, memory_mode, abbrev_timezone, utc_offset, program_name, "
        "program_version, entries, health_enabled FROM host WHERE hops >0;",
@@ -325,14 +439,16 @@ void sql_aclk_sync_init(void)
         sqlite3_free(err_msg);
     }
 
+#ifdef ENABLE_ACLK
+    fatal_assert(0 == uv_mutex_init(&aclk_async_lock));
     rc = sqlite3_exec_monitored(db_meta, "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni WHERE "
         "h.host_id = ni.host_id AND ni.node_id IS NOT NULL;", aclk_start_sync_thread, NULL, &err_msg);
     if (rc != SQLITE_OK) {
         error_report("SQLite error when starting ACLK sync threads, rc = %d (%s)", rc, err_msg);
         sqlite3_free(err_msg);
     }
+    info("ACLK sync initialization completed");
 #endif
-    return;
 }
 
 static void async_cb(uv_async_t *handle)
@@ -374,10 +490,9 @@ static void timer_cb(uv_timer_t* handle)
 #endif
 }
 
-#define MAX_CMD_BATCH_SIZE (256)
-
-void aclk_database_worker(void *arg)
+static void aclk_database_worker(void *arg)
 {
+    service_register(SERVICE_THREAD_TYPE_EVENT_LOOP, NULL, NULL, NULL, true);
     worker_register("ACLKSYNC");
     worker_register_job_name(ACLK_DATABASE_NOOP,                 "noop");
     worker_register_job_name(ACLK_DATABASE_ORPHAN_HOST,          "node orphan");
@@ -398,15 +513,12 @@ void aclk_database_worker(void *arg)
     enum aclk_database_opcode opcode;
     uv_timer_t timer_req;
     struct aclk_database_cmd cmd;
-    unsigned cmd_batch_size;
-
-    //aclk_database_init_cmd_queue(wc);
 
     char threadname[NETDATA_THREAD_NAME_MAX+1];
     if (wc->host)
-        snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", rrdhost_hostname(wc->host));
+        snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "ACLK[%s]", rrdhost_hostname(wc->host));
     else {
-        snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", wc->uuid_str);
+        snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "ACLK[%s]", wc->uuid_str);
         threadname[11] = '\0';
     }
     uv_thread_set_name_np(wc->thread, threadname);
@@ -449,17 +561,13 @@ void aclk_database_worker(void *arg)
         uv_run(loop, UV_RUN_DEFAULT);
 
         /* wait for commands */
-        cmd_batch_size = 0;
         do {
-            if (unlikely(cmd_batch_size >= MAX_CMD_BATCH_SIZE))
-                break;
             cmd = aclk_database_deq_cmd(wc);
 
             if (netdata_exit)
                 break;
 
             opcode = cmd.opcode;
-            ++cmd_batch_size;
 
             if(likely(opcode != ACLK_DATABASE_NOOP))
                 worker_is_busy(opcode);
@@ -535,7 +643,7 @@ void aclk_database_worker(void *arg)
                             wc->host = rrdhost_find_by_guid(wc->host_guid);
                             if (wc->host) {
                                 info("HOST %s (%s) detected as active", rrdhost_hostname(wc->host), wc->host_guid);
-                                snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "AS_%s", rrdhost_hostname(wc->host));
+                                snprintfz(threadname, NETDATA_THREAD_NAME_MAX, "ACLK[%s]", rrdhost_hostname(wc->host));
                                 uv_thread_set_name_np(wc->thread, threadname);
                                 wc->host->dbsync_worker = wc;
                                 if (unlikely(!wc->hostname))
@@ -584,10 +692,8 @@ void aclk_database_worker(void *arg)
     info("Shutting down ACLK sync event loop complete for host %s", wc->host_guid);
     /* TODO: don't let the API block by waiting to enqueue commands */
     uv_cond_destroy(&wc->cmd_cond);
-/*  uv_mutex_destroy(&wc->cmd_mutex); */
-    //fatal_assert(0 == uv_loop_close(loop));
-    int rc;
 
+    int rc;
     do {
         rc = uv_loop_close(loop);
     } while (rc != UV_EBUSY);
@@ -628,7 +734,7 @@ void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
 
     uuid_unparse_lower(*host_uuid, host_guid);
 
-    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
+    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE, &netdata_buffers_statistics.buffers_sqlite);
 
     buffer_sprintf(sql, TABLE_ACLK_ALERT, uuid_str);
     db_execute(buffer_tostring(sql));
@@ -648,6 +754,10 @@ void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
     if (likely(host)) {
         host->dbsync_worker = (void *)wc;
         wc->hostname = strdupz(rrdhost_hostname(host));
+        if (node_id && !host->node_id) {
+            host->node_id = mallocz(sizeof(*host->node_id));
+            uuid_copy(*host->node_id, *node_id);
+        }
     }
     else
         wc->hostname = get_hostname_by_node_id(wc->node_id);
@@ -663,142 +773,4 @@ void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id)
     UNUSED(host_uuid);
     UNUSED(node_id);
 #endif
-    return;
-}
-
-void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
-{
-    UNUSED(cmd);
-
-    debug(D_ACLK, "Checking database for %s", wc->host_guid);
-
-    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
-
-    buffer_sprintf(sql,"DELETE FROM aclk_alert_%s WHERE date_submitted IS NOT NULL AND "
-        "CAST(date_cloud_ack AS INT) < unixepoch()-%d;", wc->uuid_str, ACLK_DELETE_ACK_ALERTS_INTERNAL);
-    db_execute(buffer_tostring(sql));
-
-    buffer_free(sql);
-    return;
-}
-
-#define SQL_SELECT_HOST_BY_UUID  "SELECT host_id FROM host WHERE host_id = @host_id;"
-
-static int is_host_available(uuid_t *host_id)
-{
-    sqlite3_stmt *res = NULL;
-    int rc;
-
-    if (unlikely(!db_meta)) {
-        if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
-            error_report("Database has not been initialized");
-        return 1;
-    }
-
-    rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_BY_UUID, -1, &res, 0);
-    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, host_id, sizeof(*host_id), SQLITE_STATIC);
-    if (unlikely(rc != SQLITE_OK)) {
-        error_report("Failed to bind host_id parameter to select node instance information");
-        goto failed;
-    }
-    rc = sqlite3_step_monitored(res);
-
-    failed:
-    if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
-        error_report("Failed to finalize the prepared statement when checking host existence");
-
-    return (rc == SQLITE_ROW);
-}
-
-// OPCODE: ACLK_DATABASE_DELETE_HOST
-void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd)
-{
-    UNUSED(wc);
-    char uuid_str[GUID_LEN + 1];
-    char host_str[GUID_LEN + 1];
-
-    int rc;
-    uuid_t host_uuid;
-    char *host_guid = (char *)cmd.data;
-
-    if (unlikely(!host_guid))
-        return;
-
-    rc = uuid_parse(host_guid, host_uuid);
-    freez(host_guid);
-    if (rc)
-        return;
-
-    uuid_unparse_lower(host_uuid, host_str);
-    uuid_unparse_lower_fix(&host_uuid, uuid_str);
-
-    debug(D_ACLK_SYNC, "Checking if I should delete aclk tables for node %s", host_str);
-
-    if (is_host_available(&host_uuid)) {
-        debug(D_ACLK_SYNC, "Host %s exists, not deleting aclk sync tables", host_str);
-        return;
-    }
-
-    debug(D_ACLK_SYNC, "Host %s does NOT exist, can delete aclk sync tables", host_str);
-
-    sqlite3_stmt *res = NULL;
-    BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE);
-
-    buffer_sprintf(sql,"SELECT 'drop '||type||' IF EXISTS '||name||';' FROM sqlite_schema " \
-        "WHERE name LIKE 'aclk_%%_%s' AND type IN ('table', 'trigger', 'index');", uuid_str);
-
-    rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0);
-    if (rc != SQLITE_OK) {
-        error_report("Failed to prepare statement to clean up aclk tables");
-        goto fail;
-    }
-    buffer_flush(sql);
-
-    while (sqlite3_step_monitored(res) == SQLITE_ROW)
-        buffer_strcat(sql, (char *) sqlite3_column_text(res, 0));
-
-    rc = sqlite3_finalize(res);
-    if (unlikely(rc != SQLITE_OK))
-        error_report("Failed to finalize statement to clean up aclk tables, rc = %d", rc);
-
-    db_execute(buffer_tostring(sql));
-
-fail:
-    buffer_free(sql);
-    return;
-}
-
-static int sql_check_aclk_table(void *data, int argc, char **argv, char **column)
-{
-    struct aclk_database_worker_config *wc = data;
-    UNUSED(argc);
-    UNUSED(column);
-
-    debug(D_ACLK_SYNC,"Scheduling aclk sync table check for node %s", (char *) argv[0]);
-    struct aclk_database_cmd cmd;
-    memset(&cmd, 0, sizeof(cmd));
-    cmd.opcode = ACLK_DATABASE_DELETE_HOST;
-    cmd.data = strdupz((char *) argv[0]);
-    aclk_database_enq_cmd_noblock(wc, &cmd);
-    return 0;
-}
-
-#define SQL_SELECT_ACLK_ACTIVE_LIST "SELECT REPLACE(SUBSTR(name,19),'_','-') FROM sqlite_schema " \
-        "WHERE name LIKE 'aclk_chart_latest_%' AND type IN ('table');"
-
-void sql_check_aclk_table_list(struct aclk_database_worker_config *wc)
-{
-    char *err_msg = NULL;
-    debug(D_ACLK_SYNC,"Cleaning tables for nodes that do not exist");
-    int rc = sqlite3_exec_monitored(db_meta, SQL_SELECT_ACLK_ACTIVE_LIST, sql_check_aclk_table, (void *) wc, &err_msg);
-    if (rc != SQLITE_OK) {
-        error_report("Query failed when trying to check for obsolete ACLK sync tables, %s", err_msg);
-        sqlite3_free(err_msg);
-    }
-    return;
-}
+}
\ No newline at end of file