diff options
Diffstat (limited to 'database/sqlite')
| -rw-r--r-- | database/sqlite/sqlite_aclk.c | 241 | ||||
| -rw-r--r-- | database/sqlite/sqlite_aclk.h | 49 | ||||
| -rw-r--r-- | database/sqlite/sqlite_aclk_chart.c | 1304 | ||||
| -rw-r--r-- | database/sqlite/sqlite_aclk_chart.h | 71 | ||||
| -rw-r--r-- | database/sqlite/sqlite_aclk_node.c | 6 |
5 files changed, 14 insertions, 1657 deletions
diff --git a/database/sqlite/sqlite_aclk.c b/database/sqlite/sqlite_aclk.c index 076ef6b3be..b913c14bcb 100644 --- a/database/sqlite/sqlite_aclk.c +++ b/database/sqlite/sqlite_aclk.c @@ -3,7 +3,6 @@ #include "sqlite_functions.h" #include "sqlite_aclk.h" -#include "sqlite_aclk_chart.h" #include "sqlite_aclk_node.h" #ifdef ENABLE_ACLK @@ -34,28 +33,6 @@ const char *aclk_sync_config[] = { uv_mutex_t aclk_async_lock; struct aclk_database_worker_config *aclk_thread_head = NULL; -int retention_running = 0; - -#ifdef ENABLE_ACLK -static void stop_retention_run() -{ - uv_mutex_lock(&aclk_async_lock); - retention_running = 0; - uv_mutex_unlock(&aclk_async_lock); -} - -static int request_retention_run() -{ - int rc = 0; - uv_mutex_lock(&aclk_async_lock); - if (unlikely(retention_running)) - rc = 1; - else - retention_running = 1; - uv_mutex_unlock(&aclk_async_lock); - return rc; -} -#endif int claimed() { @@ -378,15 +355,13 @@ void sql_aclk_sync_init(void) info("SQLite aclk sync initialization completed"); fatal_assert(0 == uv_mutex_init(&aclk_async_lock)); - if (likely(rrdcontext_enabled == CONFIG_BOOLEAN_YES)) { - 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;", - create_host_callback, NULL, &err_msg); - if (rc != SQLITE_OK) { - error_report("SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg); - sqlite3_free(err_msg); - } + 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;", + create_host_callback, NULL, &err_msg); + if (rc != SQLITE_OK) { + error_report("SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg); + sqlite3_free(err_msg); } rc = sqlite3_exec_monitored(db_meta, "SELECT ni.host_id, ni.node_id FROM host h, node_instance ni WHERE " @@ -429,30 +404,6 @@ static void timer_cb(uv_timer_t* handle) } if (aclk_connected) { - if (wc->rotation_after && wc->rotation_after < now) { - cmd.opcode = ACLK_DATABASE_UPD_RETENTION; - if (!aclk_database_enq_cmd_noblock(wc, &cmd)) - wc->rotation_after = now + ACLK_DATABASE_ROTATION_INTERVAL; - } - - if (wc->chart_updates && !wc->chart_pending && wc->chart_payload_count) { - cmd.opcode = ACLK_DATABASE_PUSH_CHART; - cmd.count = ACLK_MAX_CHART_BATCH; - cmd.param1 = ACLK_MAX_CHART_BATCH_COUNT; - if (!aclk_database_enq_cmd_noblock(wc, &cmd)) { - if (wc->retry_count) - info("Queued chart/dimension payload command %s, retry count = %u", wc->host_guid, wc->retry_count); - wc->chart_pending = 1; - wc->retry_count = 0; - } else { - wc->retry_count++; - if (wc->retry_count % 100 == 0) - error_report("Failed to queue chart/dimension payload command %s, retry count = %u", - wc->host_guid, - wc->retry_count); - } - } - if (wc->alert_updates && !wc->pause_alert_updates) { cmd.opcode = ACLK_DATABASE_PUSH_ALERT; cmd.count = ACLK_MAX_ALERT_UPDATES; @@ -462,52 +413,12 @@ static void timer_cb(uv_timer_t* handle) #endif } - -#ifdef ENABLE_ACLK -void after_send_retention(uv_work_t *req, int status) -{ - struct aclk_database_worker_config *wc = req->data; - (void)status; - stop_retention_run(); - wc->retention_running = 0; - - struct aclk_database_cmd cmd; - memset(&cmd, 0, sizeof(cmd)); - cmd.opcode = ACLK_DATABASE_DIM_DELETION; - if (aclk_database_enq_cmd_noblock(wc, &cmd)) - info("Failed to queue a dimension deletion message"); - - cmd.opcode = ACLK_DATABASE_NODE_INFO; - if (aclk_database_enq_cmd_noblock(wc, &cmd)) - info("Failed to queue a node update info message"); -} - - -static void send_retention(uv_work_t *req) -{ - struct aclk_database_worker_config *wc = req->data; - - if (unlikely(wc->is_shutting_down)) - return; - - aclk_update_retention(wc); -} -#endif - #define MAX_CMD_BATCH_SIZE (256) void aclk_database_worker(void *arg) { worker_register("ACLKSYNC"); worker_register_job_name(ACLK_DATABASE_NOOP, "noop"); - worker_register_job_name(ACLK_DATABASE_ADD_CHART, "chart add"); - worker_register_job_name(ACLK_DATABASE_ADD_DIMENSION, "dimension add"); - worker_register_job_name(ACLK_DATABASE_PUSH_CHART, "chart push"); - worker_register_job_name(ACLK_DATABASE_PUSH_CHART_CONFIG, "chart conf push"); - worker_register_job_name(ACLK_DATABASE_RESET_CHART, "chart reset"); - worker_register_job_name(ACLK_DATABASE_CHART_ACK, "chart ack"); - worker_register_job_name(ACLK_DATABASE_UPD_RETENTION, "retention check"); - worker_register_job_name(ACLK_DATABASE_DIM_DELETION, "dimension delete"); worker_register_job_name(ACLK_DATABASE_ORPHAN_HOST, "node orphan"); worker_register_job_name(ACLK_DATABASE_ALARM_HEALTH_LOG, "alert log"); worker_register_job_name(ACLK_DATABASE_CLEANUP, "cleanup"); @@ -568,17 +479,9 @@ void aclk_database_worker(void *arg) 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)); -#ifdef ENABLE_ACLK - uv_work_t retention_work; - sql_get_last_chart_sequence(wc); - wc->chart_payload_count = sql_get_pending_count(wc); - if (!wc->chart_payload_count) - info("%s: No pending charts and dimensions detected during startup", wc->host_guid); -#endif wc->startup_time = now_realtime_sec(); wc->cleanup_after = wc->startup_time + ACLK_DATABASE_CLEANUP_FIRST; - wc->rotation_after = wc->startup_time + ACLK_DATABASE_ROTATION_DELAY; debug(D_ACLK_SYNC,"Node %s reports pending message count = %u", wc->node_id, wc->chart_payload_count); @@ -613,6 +516,9 @@ void aclk_database_worker(void *arg) sql_maint_aclk_sync_database(wc, cmd); if (wc->host == localhost) sql_check_aclk_table_list(wc); + cmd.opcode = ACLK_DATABASE_NODE_INFO; + cmd.completion = NULL; + (void) aclk_database_enq_cmd_noblock(wc, &cmd); break; case ACLK_DATABASE_DELETE_HOST: @@ -620,33 +526,6 @@ void aclk_database_worker(void *arg) sql_delete_aclk_table_list(wc, cmd); break; -// CHART / DIMENSION OPERATIONS -#ifdef ENABLE_ACLK - case ACLK_DATABASE_ADD_CHART: - debug(D_ACLK_SYNC, "Adding chart event for %s", wc->host_guid); - aclk_add_chart_event(wc, cmd); - break; - case ACLK_DATABASE_ADD_DIMENSION: - debug(D_ACLK_SYNC, "Adding dimension event for %s", wc->host_guid); - aclk_add_dimension_event(wc, cmd); - break; - case ACLK_DATABASE_PUSH_CHART: - debug(D_ACLK_SYNC, "Pushing chart info to the cloud for node %s", wc->host_guid); - aclk_send_chart_event(wc, cmd); - break; - case ACLK_DATABASE_PUSH_CHART_CONFIG: - debug(D_ACLK_SYNC, "Pushing chart config info to the cloud for node %s", wc->host_guid); - aclk_send_chart_config(wc, cmd); - break; - case ACLK_DATABASE_CHART_ACK: - debug(D_ACLK_SYNC, "ACK chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1); - aclk_receive_chart_ack(wc, cmd); - break; - case ACLK_DATABASE_RESET_CHART: - debug(D_ACLK_SYNC, "RESET chart SEQ for %s to %"PRIu64, wc->uuid_str, (uint64_t) cmd.param1); - aclk_receive_chart_reset(wc, cmd); - break; -#endif // ALERTS case ACLK_DATABASE_PUSH_ALERT_CONFIG: debug(D_ACLK_SYNC,"Pushing chart config info to the cloud for %s", wc->host_guid); @@ -679,27 +558,6 @@ void aclk_database_worker(void *arg) sql_build_node_collectors(wc); break; #ifdef ENABLE_ACLK - case ACLK_DATABASE_DIM_DELETION: - debug(D_ACLK_SYNC,"Sending dimension deletion information %s", wc->uuid_str); - aclk_process_dimension_deletion(wc, cmd); - break; - case ACLK_DATABASE_UPD_RETENTION: - if (unlikely(wc->retention_running)) - break; - - if (unlikely(request_retention_run())) { - wc->rotation_after = now_realtime_sec() + ACLK_DATABASE_RETENTION_RETRY; - break; - } - - debug(D_ACLK_SYNC,"Sending retention info for %s", wc->uuid_str); - retention_work.data = wc; - wc->retention_running = 1; - if (unlikely(uv_queue_work(loop, &retention_work, send_retention, after_send_retention))) { - wc->retention_running = 0; - stop_retention_run(); - } - break; // NODE_INSTANCE DETECTION case ACLK_DATABASE_ORPHAN_HOST: @@ -809,30 +667,6 @@ void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id) BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE); - buffer_sprintf(sql, TABLE_ACLK_CHART, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql, TABLE_ACLK_CHART_PAYLOAD, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql, TABLE_ACLK_CHART_LATEST, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql, INDEX_ACLK_CHART, uuid_str, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql, INDEX_ACLK_CHART_LATEST, uuid_str, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql, TRIGGER_ACLK_CHART_PAYLOAD, uuid_str, uuid_str, uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - buffer_sprintf(sql, TABLE_ACLK_ALERT, uuid_str); db_execute(buffer_tostring(sql)); buffer_flush(sql); @@ -857,9 +691,9 @@ 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->chart_updates = 0; wc->alert_updates = 0; - wc->retry_count = 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)); @@ -879,31 +713,9 @@ void sql_maint_aclk_sync_database(struct aclk_database_worker_config *wc, struct BUFFER *sql = buffer_create(ACLK_SYNC_QUERY_SIZE); - buffer_sprintf(sql,"DELETE FROM aclk_chart_%s WHERE date_submitted IS NOT NULL AND " - "CAST(date_updated AS INT) < unixepoch()-%d;", wc->uuid_str, ACLK_DELETE_ACK_INTERNAL); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql,"DELETE FROM aclk_chart_payload_%s WHERE unique_id NOT IN " - "(SELECT unique_id FROM aclk_chart_%s) AND unique_id NOT IN (SELECT unique_id FROM aclk_chart_latest_%s);", - wc->uuid_str, wc->uuid_str, wc->uuid_str); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - 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_flush(sql); - - buffer_sprintf(sql,"UPDATE aclk_chart_%s SET status = NULL, date_submitted=unixepoch() WHERE " - "date_submitted IS NULL AND CAST(date_created AS INT) < unixepoch()-%d;", wc->uuid_str, ACLK_AUTO_MARK_SUBMIT_INTERVAL); - db_execute(buffer_tostring(sql)); - buffer_flush(sql); - - buffer_sprintf(sql,"UPDATE aclk_chart_%s SET date_updated = unixepoch() WHERE date_updated IS NULL" - " AND date_submitted IS NOT NULL AND CAST(date_submitted AS INT) < unixepoch()-%d;", - wc->uuid_str, ACLK_AUTO_MARK_UPDATED_INTERVAL); - db_execute(buffer_tostring(sql)); buffer_free(sql); return; @@ -1032,32 +844,3 @@ void sql_check_aclk_table_list(struct aclk_database_worker_config *wc) return; } - -void aclk_data_rotated(void) -{ -#ifdef ENABLE_ACLK - - if (!aclk_connected) - return; - - time_t next_rotation_time = now_realtime_sec()+ACLK_DATABASE_ROTATION_DELAY; - rrd_rdlock(); - RRDHOST *this_host; - rrdhost_foreach_read(this_host) { - struct aclk_database_worker_config *wc = this_host->dbsync_worker; - if (wc) - wc->rotation_after = next_rotation_time; - } - rrd_unlock(); - - struct aclk_database_worker_config *tmp = aclk_thread_head; - - uv_mutex_lock(&aclk_async_lock); - while (tmp) { - tmp->rotation_after = next_rotation_time; - tmp = tmp->next; - } - uv_mutex_unlock(&aclk_async_lock); -#endif - return; -} diff --git a/database/sqlite/sqlite_aclk.h b/database/sqlite/sqlite_aclk.h index 3560f0d3d6..04e58e924e 100644 --- a/database/sqlite/sqlite_aclk.h +++ b/database/sqlite/sqlite_aclk.h @@ -5,8 +5,6 @@ #include "sqlite3.h" -// TODO: To be added -#include "../../aclk/schema-wrappers/chart_stream.h" #ifndef ACLK_MAX_CHART_BATCH #define ACLK_MAX_CHART_BATCH (200) @@ -16,14 +14,8 @@ #endif #define ACLK_MAX_ALERT_UPDATES (5) #define ACLK_DATABASE_CLEANUP_FIRST (60) -#define ACLK_DATABASE_ROTATION_DELAY (180) -#define ACLK_DATABASE_RETENTION_RETRY (60) #define ACLK_DATABASE_CLEANUP_INTERVAL (3600) -#define ACLK_DATABASE_ROTATION_INTERVAL (3600) -#define ACLK_DELETE_ACK_INTERNAL (600) #define ACLK_DELETE_ACK_ALERTS_INTERNAL (86400) -#define ACLK_AUTO_MARK_SUBMIT_INTERVAL (3600) -#define ACLK_AUTO_MARK_UPDATED_INTERVAL (1800) #define ACLK_SYNC_QUERY_SIZE 512 struct aclk_completion { @@ -87,44 +79,14 @@ static inline char *get_str_from_uuid(uuid_t *uuid) return strdupz(uuid_str); } -#define TABLE_ACLK_CHART "CREATE TABLE IF NOT EXISTS aclk_chart_%s (sequence_id INTEGER PRIMARY KEY, " \ - "date_created, date_updated, date_submitted, status, uuid, type, unique_id, " \ - "update_count default 1, unique(uuid, status));" - -#define TABLE_ACLK_CHART_PAYLOAD "CREATE TABLE IF NOT EXISTS aclk_chart_payload_%s (unique_id BLOB PRIMARY KEY, " \ - "uuid, claim_id, type, date_created, payload);" - -#define TABLE_ACLK_CHART_LATEST "CREATE TABLE IF NOT EXISTS aclk_chart_latest_%s (uuid BLOB PRIMARY KEY, " \ - "unique_id, date_submitted);" - -#define TRIGGER_ACLK_CHART_PAYLOAD "CREATE TRIGGER IF NOT EXISTS aclk_tr_chart_payload_%s " \ - "after insert on aclk_chart_payload_%s " \ - "begin insert into aclk_chart_%s (uuid, unique_id, type, status, date_created) values " \ - " (new.uuid, new.unique_id, new.type, 'pending', unixepoch()) on conflict(uuid, status) " \ - " do update set unique_id = new.unique_id, update_count = update_count + 1; " \ - "end;" - #define TABLE_ACLK_ALERT "CREATE TABLE IF NOT EXISTS aclk_alert_%s (sequence_id INTEGER PRIMARY KEY, " \ "alert_unique_id, date_created, date_submitted, date_cloud_ack, " \ "unique(alert_unique_id));" -#define INDEX_ACLK_CHART "CREATE INDEX IF NOT EXISTS aclk_chart_index_%s ON aclk_chart_%s (unique_id);" - -#define INDEX_ACLK_CHART_LATEST "CREATE INDEX IF NOT EXISTS aclk_chart_latest_index_%s ON aclk_chart_latest_%s (unique_id);" - #define INDEX_ACLK_ALERT "CREATE INDEX IF NOT EXISTS aclk_alert_index_%s ON aclk_alert_%s (alert_unique_id);" - enum aclk_database_opcode { ACLK_DATABASE_NOOP = 0, - ACLK_DATABASE_ADD_CHART, - ACLK_DATABASE_ADD_DIMENSION, - ACLK_DATABASE_PUSH_CHART, - ACLK_DATABASE_PUSH_CHART_CONFIG, - ACLK_DATABASE_RESET_CHART, - ACLK_DATABASE_CHART_ACK, - ACLK_DATABASE_UPD_RETENTION, - ACLK_DATABASE_DIM_DELETION, ACLK_DATABASE_ORPHAN_HOST, ACLK_DATABASE_ALARM_HEALTH_LOG, ACLK_DATABASE_CLEANUP, @@ -172,12 +134,8 @@ struct aclk_database_worker_config { char node_id[GUID_LEN + 1]; char host_guid[GUID_LEN + 1]; char *hostname; // hostname to avoid constant lookups - uint64_t chart_sequence_id; // last chart_sequence_id - time_t chart_timestamp; // last chart timestamp time_t cleanup_after; // Start a cleanup after this timestamp time_t startup_time; // When the sync thread started - time_t rotation_after; - uint64_t batch_id; // batch id to use uint64_t alerts_batch_id; // batch id for alerts to use uint64_t alerts_start_seq_id; // cloud has asked to start streaming from uint64_t alert_sequence_id; // last alert sequence_id @@ -193,15 +151,9 @@ struct aclk_database_worker_config { uv_cond_t cmd_cond; volatile unsigned queue_size; struct aclk_database_cmdqueue cmd_queue; - uint32_t retry_count; - int chart_updates; int alert_updates; - time_t batch_created; int node_info_send; time_t node_collectors_send; - int chart_pending; - int chart_reset_count; - int retention_running; volatile unsigned is_shutting_down; volatile unsigned is_orphan; struct aclk_database_worker_config *next; @@ -236,7 +188,6 @@ extern int aclk_database_enq_cmd_noblock(struct aclk_database_worker_config *wc, extern void aclk_database_enq_cmd(struct aclk_database_worker_config *wc, struct aclk_database_cmd *cmd); extern void sql_create_aclk_table(RRDHOST *host, uuid_t *host_uuid, uuid_t *node_id); int aclk_worker_enq_cmd(char *node_id, struct aclk_database_cmd *cmd); -void aclk_data_rotated(void); void sql_aclk_sync_init(void); void sql_check_aclk_table_list(struct aclk_database_worker_config *wc); void sql_delete_aclk_table_list(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd); diff --git a/database/sqlite/sqlite_aclk_chart.c b/database/sqlite/sqlite_aclk_chart.c deleted file mode 100644 index d373dd7637..0000000000 --- a/database/sqlite/sqlite_aclk_chart.c +++ /dev/null @@ -1,1304 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "sqlite_functions.h" -#include "sqlite_aclk_chart.h" - -#ifdef ENABLE_ACLK -#include "../../aclk/aclk_charts_api.h" -#include "../../aclk/aclk.h" - -static inline int -sql_queue_chart_payload(struct aclk_database_worker_config *wc, void *data, enum aclk_database_opcode opcode) -{ - int rc; - if (unlikely(!wc)) - return 1; - - struct aclk_database_cmd cmd; - memset(&cmd, 0, sizeof(cmd)); - cmd.opcode = opcode; - cmd.data = data; - rc = aclk_database_enq_cmd_noblock(wc, &cmd); - return rc; -} - -static time_t payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size) -{ - static __thread sqlite3_stmt *res = NULL; - int rc; - time_t send_status = 0; - - if (unlikely(!res)) { - char sql[ACLK_SYNC_QUERY_SIZE]; - snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "SELECT acl.date_submitted 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, sql, &res); - if (rc != SQLITE_OK) { - error_report("Failed to prepare statement to check payload data on %s", sql); - return 0; - } - } - - rc = sqlite3_bind_blob(res, 1, uuid, sizeof(*uuid), SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = sqlite3_bind_blob(res, 2, payload, payload_size, SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - while (sqlite3_step_monitored(res) == SQLITE_ROW) { - send_status = (time_t) sqlite3_column_int64(res, 0); - } - -bind_fail: - if (unlikely(sqlite3_reset(res) != SQLITE_OK)) - error_report("Failed to reset statement in check payload, rc = %d", rc); - return send_status; -} - -static int aclk_add_chart_payload( - struct aclk_database_worker_config *wc, - uuid_t *uuid, - char *claim_id, - ACLK_PAYLOAD_TYPE payload_type, - void *payload, - size_t payload_size, - time_t *send_status, - int check_sent) -{ - static __thread sqlite3_stmt *res_chart = NULL; - int rc; - time_t date_submitted; - - if (unlikely(!payload)) - return 0; - - if (check_sent) { - date_submitted = payload_sent(wc->uuid_str, uuid, payload, payload_size); - if (send_status) - *send_status = date_submitted; - if (date_submitted) - return 0; - } - - if (unlikely(!res_chart)) { - 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, unixepoch(), @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; - } - } - - uuid_t unique_uuid; - uuid_generate(unique_uuid); - - uuid_t claim_uuid; - if (uuid_parse(claim_id, claim_uuid)) - return 1; - - rc = sqlite3_bind_blob(res_chart, 1, &unique_uuid, sizeof(unique_uuid), SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = sqlite3_bind_blob(res_chart, 2, uuid, sizeof(*uuid), SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = sqlite3_bind_blob(res_chart, 3, &claim_uuid, sizeof(claim_uuid), SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = sqlite3_bind_int(res_chart, 4, payload_type); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = sqlite3_bind_blob(res_chart, 5, payload, payload_size, SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - rc = execute_insert(res_chart); - if (unlikely(rc != SQLITE_DONE)) - error_report("Failed store chart payload event, rc = %d", rc); - else { - wc->chart_payload_count++; - time_t now = now_realtime_sec(); - if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY) - wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY; - } - -bind_fail: - if (unlikely(sqlite3_reset(res_chart) != SQLITE_OK)) - error_report("Failed to reset statement in store chart payload, rc = %d", rc); - return (rc != SQLITE_DONE); -} - -int aclk_add_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) -{ - int rc = 0; - CHECK_SQLITE_CONNECTION(db_meta); - - char *claim_id = get_agent_claimid(); - - RRDSET *st = cmd.data; - - if (likely(claim_id)) { - struct chart_instance_updated chart_payload; - memset(&chart_payload, 0, sizeof(chart_payload)); - chart_payload.config_hash = get_str_from_uuid(&st->hash_uuid); - chart_payload.update_every = st->update_every; - chart_payload.memory_mode = st->rrd_memory_mode; - chart_payload.name = (char *)rrdset_name(st); - chart_payload.node_id = wc->node_id; - chart_payload.claim_id = claim_id; - chart_payload.id = strdupz(rrdset_id(st)); - - chart_payload.chart_labels = rrdlabels_create(); - rrdlabels_copy(chart_payload.chart_labels, st->rrdlabels); - - size_t size; - char *payload = generate_chart_instance_updated(&size, &chart_payload); - if (likely(payload)) - rc = aclk_add_chart_payload(wc, &st->chart_uuid, claim_id, ACLK_PAYLOAD_CHART, (void *) payload, size, NULL, 1); - freez(payload); - chart_instance_updated_destroy(&chart_payload); - } - return rc; -} - -static inline int aclk_upd_dimension_event(struct aclk_database_worker_config *wc, char *claim_id, uuid_t *dim_uuid, - const char *dim_id, const char *dim_name, const char *chart_type_id, time_t first_time, time_t last_time, - time_t *send_status) -{ - int rc = 0; - size_t size; - - if (unlikely(!dim_uuid || !dim_id || !dim_name || !chart_type_id)) - return 0; - - struct chart_dimension_updated dim_payload; - memset(&dim_payload, 0, sizeof(dim_payload)); - -#ifdef NETDATA_INTERNAL_CHECKS - if (!first_time) - info("Host %s (node %s) deleting dimension id=[%s] name=[%s] chart=[%s]", - wc->host_guid, wc->node_id, dim_id, dim_name, chart_type_id); - if (last_time) - info("Host %s (node %s) stopped collecting dimension id=[%s] name=[%s] chart=[%s] %ld seconds ago at %ld", - wc->host_guid, wc->node_id, dim_id, dim_name, chart_type_id, now_realtime_sec() - last_time, last_time); -#endif - - dim_payload.node_id = wc->node_id; - dim_payload.claim_id = claim_id; - dim_payload.name = dim_name; - dim_payload.id = dim_id; - dim_payload.chart_id = chart_type_id; - dim_payload.created_at.tv_sec = first_time; - dim_payload.last_timestamp.tv_sec = last_time; - char *payload = generate_chart_dimension_updated(&size, &dim_payload); - if (likely(payload)) - rc = aclk_add_chart_payload(wc, dim_uuid, claim_id, ACLK_PAYLOAD_DIMENSION, (void *)payload, size, send_status, 1); - freez(payload); - return rc; -} - -void aclk_process_dimension_deletion(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) -{ - int rc = 0; - sqlite3_stmt *res = NULL; - - if (!aclk_connected) - return; - - if (unlikely(!db_meta)) - return; - - uuid_t host_id; - if (uuid_parse(wc->host_guid, host_id)) - return; - - char *claim_id = get_agent_claimid(); - if (!claim_id) - return; - - rc = sqlite3_prepare_v2( - db_meta, - "DELETE FROM dimension_delete where host_id = @host_id " - "RETURNING dimension_id, dimension_name, chart_type_id, dim_id LIMIT 10;", - -1, - &res, - 0); - - if (rc != SQLITE_OK) { - error_report("Failed to prepare statement when trying to delete dimension deletes"); - freez(claim_id); - return; - } - - rc = sqlite3_bind_blob(res, 1, &host_id, sizeof(host_id), SQLITE_STATIC); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - unsigned count = 0; - while (sqlite3_step_monitored(res) == SQLITE_ROW) { - (void) aclk_upd_dimension_event( - wc, - claim_id, - (uuid_t *)sqlite3_column_text(res, 3), - (const char *)sqlite3_column_text(res, 0), - (const char *)sqlite3_column_text(res, 1), - (const char *)sqlite3_column_text(res, 2), - 0, - 0, - NULL); - count++; - } - - if (count) { - memset(&cmd, 0, sizeof(cmd)); - cmd.opcode = ACLK_DATABASE_DIM_DELETION; - if (aclk_database_enq_cmd_noblock(wc, &cmd)) - info("Failed to queue a dimension deletion message"); - } - -bind_fail: - rc = sqlite3_finalize(res); - if (unlikely(rc != SQLITE_OK)) - error_report("Failed to finalize statement when adding dimension deletion events, rc = %d", rc); - freez(claim_id); - return; -} - -int aclk_add_dimension_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) -{ - int rc = 1; - CHECK_SQLITE_CONNECTION(db_meta); - - struct aclk_chart_dimension_data *aclk_cd_data = cmd.data; - - char *claim_id = get_agent_claimid(); - if (!claim_id) - goto cleanup; - - rc = aclk_add_chart_payload(wc, &aclk_cd_data->uuid, claim_id, ACLK_PAYLOAD_DIMENSION, - (void *) aclk_cd_data->payload, aclk_cd_data->payload_size, NULL, aclk_cd_data->check_payload); - - freez(claim_id); -cleanup: - freez(aclk_cd_data->payload); - freez(aclk_cd_data); - return rc; -} - -void aclk_send_chart_event(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) -{ - int rc; - - wc->chart_pending = 0; - if (unlikely(!wc->chart_updates)) { - log_access( - "ACLK STA [%s (%s)]: Ignoring chart push event, updates have been turned off for this node.", - wc->node_id, - wc->host ? rrdhost_hostname(wc->host) : "N/A"); - return; - } - - char *claim_id = get_agent_claimid(); - if (unlikely(!claim_id)) - return; - - uuid_t claim_uuid; - if (uuid_parse(claim_id, claim_uuid)) - return; - - int limit = cmd.count > 0 ? cmd.count : 1; - - uint64_t first_sequence; - uint64_t last_sequence; - time_t last_timestamp = 0; - - char sql[ACLK_SYNC_QUERY_SIZE]; - static __thread sqlite3_stmt *res = NULL; - - 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); - if (unlikely(rc != SQLITE_OK)) - goto bind_fail; - - char **payload_list = callocz(limit + 1, sizeof(char *)); - size_t *payload_list_size = callocz(limit + 1, sizeof(size_t)); - size_t *payload_list_max_size = callocz(limit + 1, sizeof(size_t)); - struct aclk_message_position *position_list = callocz(limit + 1, sizeof(*position_list)); - int *is_dim = callocz(limit + 1, sizeof(*is_dim)); - - int loop = cmd.param1; - - uint64_t start_sequence_id = wc->chart_sequence_id; - - while (loop > 0) { - uint64_t previous_sequence_id = wc->chart_sequence_id; - int count = 0; - first_sequence = 0; - last_sequence = 0; - while (count < limit && sqlite3_step_monitored(res) == SQLITE_ROW) { - size_t payload_size = sqlite3_column_bytes(res, 1); - if (payload_list_max_size[count] < payload_size) { - freez(payload_list[count]); - payload_list_max_size[count] = payload_size; - payload_list[count] = mallocz(payload_size); - } - payload_list_size[count] = payload_size; - memcpy(payload_list[count], sqlite3_column_blob(res, 1), payload_size); - position_list[count].sequence_id = (uint64_t)sqlite3_column_int64(res, 0); - position_list[count].previous_sequence_id = previous_sequence_id; - position_list[count].seq_id_creation_time.tv_sec = sqlite3_column_int64(res, 2); - position_list[count].seq_id_creation_time.tv_usec = 0; - if (!first_sequence) - first_sequence = position_list[count].sequence_id; - last_sequence = position_list[count].sequence_id; - last_timestamp = position_list[count].seq_id_creation_time.tv_sec; - previous_sequence_id = last_sequence; - is_dim[count] = sqlite3_column_int(res, 3) > 0; - count++; - if (wc->chart_payload_count) - wc->chart_payload_count--; - } - freez(payload_list[count]); - payload_list_max_size[count] = 0; - payload_list[count] = NULL; - - rc = sqlite3_reset(res); - if (unlikely(rc != SQLITE_OK)) - error_report("Failed to reset statement when pushing chart events, rc = %d", rc); - - if (likely(first_sequence)) { - - db_lock(); - snprintfz(sql,ACLK_SYNC_QUERY_SIZE-1, "UPDATE aclk_chart_%s SET status = NULL, date_submitted=unixepoch() " - "WHERE date_submitted IS NULL AND sequence_id BETWEEN %" PRIu64 " AND %" PRIu64 ";", - wc->uuid_str, first_sequence, last_sequence); - 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(sql); - db_unlock(); - - aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id); - log_access( - "ACLK RES [%s (%s)]: CHARTS SENT from %" PRIu64 " to %" PRIu64 " batch=%" PRIu64, - wc->node_id, - wc->hostname ? wc->hostname : "N/A", - first_sequence, - last_sequence, - wc->batch_id); - wc->chart_sequence_id = last_sequence; - wc->chart_timestamp = last_timestamp; - } else - break; - --loop; - } - - if (start_sequence_id != wc->chart_sequence_id) { - time_t now = now_realtime_sec(); - if (wc->rotation_after > now && wc->rotation_after < now + ACLK_DATABASE_ROTATION_DELAY) - wc->rotation_after = now + ACLK_DATABASE_ROTATION_DELAY; - } else { - wc->chart_payload_count = sql_get_pending_count(wc); - if (!wc->chart_payload_count) - log_access( - "ACLK STA [%s (%s)]: Sync of charts and dimensions done in %ld seconds.", - wc->node_id, - wc->hostname |