diff options
Diffstat (limited to 'database/sqlite/sqlite_aclk_chart.c')
| -rw-r--r-- | database/sqlite/sqlite_aclk_chart.c | 993 |
1 files changed, 993 insertions, 0 deletions
diff --git a/database/sqlite/sqlite_aclk_chart.c b/database/sqlite/sqlite_aclk_chart.c new file mode 100644 index 000000000..4b887abaa --- /dev/null +++ b/database/sqlite/sqlite_aclk_chart.c @@ -0,0 +1,993 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +#include "sqlite_functions.h" +#include "sqlite_aclk_chart.h" + +#ifdef ENABLE_NEW_CLOUD_PROTOCOL +#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 int payload_sent(char *uuid_str, uuid_t *uuid, void *payload, size_t payload_size) +{ + static __thread sqlite3_stmt *res = NULL; + int rc; + 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 " + "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); + if (rc != SQLITE_OK) { + error_report("Failed to prepare statement to check payload data"); + 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(res) == SQLITE_ROW) { + send_status = sqlite3_column_int(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) +{ + static __thread sqlite3_stmt *res_chart = NULL; + int rc; + + rc = payload_sent(wc->uuid_str, uuid, payload, payload_size); + if (rc == 1) + 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); + + 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 = is_agent_claimed(); + + 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->state->hash_id); + chart_payload.update_every = st->update_every; + chart_payload.memory_mode = st->rrd_memory_mode; + chart_payload.name = (char *)st->name; + chart_payload.node_id = wc->node_id; + chart_payload.claim_id = claim_id; + chart_payload.id = strdupz(st->id); + + struct label_index *labels = &st->state->labels; + netdata_rwlock_wrlock(&labels->labels_rwlock); + struct label *label_list = labels->head; + struct label *chart_label = NULL; + while (label_list) { + chart_label = add_label_to_list(chart_label, label_list->key, label_list->value, label_list->label_source); + label_list = label_list->next; + } + netdata_rwlock_unlock(&labels->labels_rwlock); + chart_payload.label_head = chart_label; + + 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); + 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) +{ + 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); +#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); + 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_use_new_cloud_arch || !aclk_connected) + return; + + if (unlikely(!db_meta)) + return; + + uuid_t host_id; + if (uuid_parse(wc->host_guid, host_id)) + return; + + char *claim_id = is_agent_claimed(); + 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(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); + 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 = 0; + CHECK_SQLITE_CONNECTION(db_meta); + + char *claim_id = is_agent_claimed(); + + RRDDIM *rd = cmd.data; + + if (likely(claim_id)) { + time_t now = now_realtime_sec(); + + time_t first_t = rd->state->query_ops.oldest_time(rd); + time_t last_t = rd->state->query_ops.latest_time(rd); + + int live = ((now - last_t) < (RRDSET_MINIMUM_LIVE_COUNT * rd->update_every)); + + rc = aclk_upd_dimension_event( + wc, + claim_id, + &rd->state->metric_uuid, + rd->id, + rd->name, + rd->rrdset->id, + first_t, + live ? 0 : last_t); + + freez(claim_id); + } + rrddim_flag_clear(rd, RRDDIM_FLAG_ACLK); + 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("AC [%s (%s)]: Ignoring chart push event, updates have been turned off for this node.", wc->node_id, wc->host ? wc->host->hostname : "N/A"); + return; + } + + char *claim_id = is_agent_claimed(); + 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; + + 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); + + 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; + } + + 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(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)) { + buffer_flush(sql); + + db_lock(); + buffer_sprintf(sql, "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) " + " 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_unlock(); + + aclk_chart_inst_and_dim_update(payload_list, payload_list_size, is_dim, position_list, wc->batch_id); + log_access("OG [%s (%s)]: Sending charts and dimensions update, batch_id %"PRIu64", first sequence %"PRIu64", last sequence %"PRIu64, wc->node_id, wc->host ? wc->host->hostname : "N/A", wc->batch_id, first_sequence, last_sequence); + 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("AC [%s (%s)]: Sync of charts and dimensions done in %ld seconds.", wc->node_id, wc->host ? wc->host->hostname : "N/A", now_realtime_sec() - wc->startup_time); + } + + for (int i = 0; i <= limit; ++i) + freez(payload_list[i]); + + freez(payload_list); + freez(payload_list_size); + freez(payload_list_max_size); + freez(position_list); + freez(is_dim); + +bind_fail: + rc = sqlite3_finalize(res); + if (unlikely(rc != SQLITE_OK)) + error_report("Failed to finalize statement when pushing chart events, rc = %d", rc); + + buffer_free(sql); + freez(claim_id); + return; +} + +// Push one chart config to the cloud +int aclk_send_chart_config(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) +{ + UNUSED(wc); + + CHECK_SQLITE_CONNECTION(db_meta); + + sqlite3_stmt *res = NULL; + int rc = 0; + + char *hash_id = (char *) cmd.data_param; + + uuid_t hash_uuid; + rc = uuid_parse(hash_id, hash_uuid); + + if (unlikely(rc)) { + freez((char *) cmd.data_param); + return 1; + } + + BUFFER *sql = buffer_create(1024); + buffer_sprintf(sql, "SELECT type, family, context, title, priority, plugin, module, unit, chart_type " \ + "FROM chart_hash WHERE hash_id = @hash_id;"); + + 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 fetch a chart hash configuration"); + goto fail; + } + + rc = sqlite3_bind_blob(res, 1, &hash_uuid , sizeof(hash_uuid), SQLITE_STATIC); + if (unlikely(rc != SQLITE_OK)) + goto bind_fail; + + struct chart_config_updated chart_config; + chart_config.config_hash = NULL; + + while (sqlite3_step(res) == SQLITE_ROW) { + chart_config.type = strdupz((char *)sqlite3_column_text(res, 0)); + chart_config.family = strdupz((char *)sqlite3_column_text(res, 1)); + chart_config.context = strdupz((char *)sqlite3_column_text(res, 2)); + chart_config.title = strdupz((char *)sqlite3_column_text(res, 3)); + chart_config.priority = sqlite3_column_int64(res, 4); + chart_config.plugin = strdupz((char *)sqlite3_column_text(res, 5)); + chart_config.module = sqlite3_column_bytes(res, 6) > 0 ? strdupz((char *)sqlite3_column_text(res, 6)) : NULL; + chart_config.chart_type = (RRDSET_TYPE) sqlite3_column_int(res,8); + chart_config.units = strdupz((char *)sqlite3_column_text(res, 7)); + chart_config.config_hash = strdupz(hash_id); + } + + if (likely(chart_config.config_hash)) { + log_access("OG [%s (%s)]: Sending chart config for %s.", wc->node_id, wc->host ? wc->host->hostname : "N/A", hash_id); + aclk_chart_config_updated(&chart_config, 1); + destroy_chart_config_updated(&chart_config); + } + else + log_access("AC [%s (%s)]: Chart config for %s not found.", wc->node_id, wc->host ? wc->host->hostname : "N/A", hash_id); + + bind_fail: + rc = sqlite3_finalize(res); + if (unlikely(rc != SQLITE_OK)) + error_report("Failed to reset statement when pushing chart config hash, rc = %d", rc); + fail: + freez((char *) cmd.data_param); + buffer_free(sql); + return rc; +} + + +void aclk_receive_chart_ack(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) +{ + int rc; + sqlite3_stmt *res = NULL; + + 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); + + 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); + + rc = sqlite3_prepare_v2(db_meta, buffer_tostring(sql), -1, &res, 0); + if (rc != SQLITE_OK) { + error_report("Failed to prepare statement count sequence ids in the database"); + goto prepare_fail; + } + + rc = sqlite3_bind_int64(res, 1, (uint64_t) cmd.param1); + if (unlikely(rc != SQLITE_OK)) + goto bind_fail; + + rc = execute_insert(res); + 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; +} + +void aclk_receive_chart_reset(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) +{ + BUFFER *sql = buffer_create(1024); + buffer_sprintf(sql, "UPDATE aclk_chart_%s SET status = NULL, date_submitted = NULL WHERE sequence_id >= %"PRIu64";", + 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_execute("BEGIN TRANSACTION;"); + db_execute(buffer_tostring(sql)); + db_execute("COMMIT TRANSACTION;"); + + db_unlock(); + wc->chart_sequence_id = 0; + wc->chart_timestamp = 0; + wc->chart_payload_count = 0; + + RRDHOST *host = wc->host; + if (likely(host)) { + rrdhost_rdlock(host); + RRDSET *st; + rrdset_foreach_read(st, host) + { + rrdset_rdlock(st); + rrdset_flag_clear(st, RRDSET_FLAG_ACLK); + RRDDIM *rd; + rrddim_foreach_read(rd, st) + { + rd->state->aclk_live_status = (rd->state->aclk_live_status == 0); + } + rrdset_unlock(st); + } + rrdhost_unlock(host); + } + else + error_report("ACLK synchronization thread for %s is not linked to HOST", wc->host_guid); + } + else { + log_access("AC [%s (%s)]: Restarting chart sync from sequence %"PRIu64, wc->node_id, wc->host ? wc->host->hostname : "N/A", cmd.param1); + wc->chart_payload_count = sql_get_pending_count(wc); + sql_get_last_chart_sequence(wc); + } + buffer_free(sql); + wc->chart_updates = 1; + return; +} + + +// +// Functions called directly from ACLK threads and will queue commands +// +void aclk_get_chart_config(char **hash_id) +{ + struct aclk_database_worker_config *wc = (struct aclk_database_worker_config *)localhost->dbsync_worker; + + if (unlikely(!wc || !hash_id)) + return; + + struct aclk_database_cmd cmd; + memset(&cmd, 0, sizeof(cmd)); + cmd.opcode = ACLK_DATABASE_PUSH_CHART_CONFIG; + for (int i = 0; hash_id[i]; ++i) { + // TODO: Verify that we have a valid hash_id + log_access("IN [%s (%s)]: Request %d for chart config with hash %s received.", wc->node_id, wc->host ? wc->host->hostname : "N/A", i, hash_id[i]); + cmd.data_param = (void *)strdupz(hash_id[i]); + aclk_database_enq_cmd(wc, &cmd); + } + return; +} + +// Send a command to a node_id +// Need to discover the thread that will handle the request +// if thread not in active hosts, then try to find in the queue +static void aclk_submit_param_command(char *node_id, enum aclk_database_opcode aclk_command, uint64_t param) +{ + if (unlikely(!node_id)) + return; + + struct aclk_database_worker_config *wc = NULL; + struct aclk_database_cmd cmd; + memset(&cmd, 0, sizeof(cmd)); + cmd.opcode = aclk_command; + cmd.param1 = param; + + rrd_wrlock(); + RRDHOST *host = find_host_by_node_id(node_id); + if (likely(host)) + wc = (struct aclk_database_worker_config *)host->dbsync_worker; + rrd_unlock(); + if (wc) + aclk_database_enq_cmd(wc, &cmd); + else { + if (aclk_worker_enq_cmd(node_id, &cmd)) + log_access("AC [%s (N/A)]: ACLK synchronization thread is not active.", node_id); + } + return; +} + +void aclk_ack_chart_sequence_id(char *node_id, uint64_t last_sequence_id) +{ + if (unlikely(!node_id)) + return; + + log_access("AC [%s (N/A)]: Node reports last sequence id received %"PRIu64, node_id, last_sequence_id); + aclk_submit_param_command(node_id, ACLK_DATABASE_CHART_ACK, last_sequence_id); + return; +} + +// Start streaming charts / dimensions for node_id +void aclk_start_streaming(char *node_id, uint64_t sequence_id, time_t created_at, uint64_t batch_id) +{ + UNUSED(created_at); + if (unlikely(!node_id)) + return; + + log_access("IN [%s (N/A)]: Start streaming charts from sequence %"PRIu64" t=%ld, batch=%"PRIu64, node_id, + sequence_id, created_at, batch_id); + + uuid_t node_uuid; + if (uuid_parse(node_id, node_uuid)) + return; + + struct aclk_database_worker_config *wc = NULL; + rrd_wrlock(); + RRDHOST *host = localhost; + while(host) { + if (host->node_id && !(uuid_compare(*host->node_id, node_uuid))) { + rrd_unlock(); + wc = (struct aclk_database_worker_config *)host->dbsync_worker; + if (likely(wc)) { + wc->chart_reset_count++; + __sync_synchronize(); + wc->chart_updates = 0; + wc->batch_id = batch_id; + __sync_synchronize(); + wc->batch_created = now_realtime_sec(); + if (sequence_id > wc->chart_sequence_id || wc->chart_reset_count > 10) { + log_access("AC [%s (%s)]: Requesting full resync from the cloud " + "(reset=%d, remote_seq=%"PRIu64", local_seq=%"PRIu64")" + , wc->node_id, wc->host ? wc->host->hostname : "N/A", wc->chart_reset_count, sequence_id, wc->chart_sequence_id); + chart_reset_t chart_reset; + chart_reset.claim_id = is_agent_claimed(); + if (chart_reset.claim_id) { + chart_reset.node_id = node_id; + chart_reset.reason = SEQ_ID_NOT_EXISTS; + aclk_chart_reset(chart_reset); + freez(chart_reset.claim_id); + wc->chart_reset_count = -1; + } + return; + } else { + struct aclk_database_cmd cmd; + memset(&cmd, 0, sizeof(cmd)); + // TODO: handle timestamp + if (sequence_id < wc->chart_sequence_id || !sequence_id) { // || created_at != wc->chart_timestamp) { + log_access("AC [%s (%s)]: Reset streaming charts from sequence %"PRIu64 \ + " t=%ld (reset count=%d)", wc->node_id, wc->host ? wc->host->hostname : "N/A", wc->chart_sequence_id, + wc->chart_timestamp, wc->chart_reset_count); + cmd.opcode = ACLK_DATABASE_RESET_CHART; + cmd.param1 = sequence_id + 1; + cmd.completion = NULL; + aclk_database_enq_cmd(wc, &cmd); + } + else { + log_access("AC [%s (%s)]: Start streaming charts enabled -- last streamed sequence %"PRIu64 \ + " t=%ld (reset count=%d)", wc->node_id, wc->host ? wc->host->hostname : "N/A", wc->chart_sequence_id, + wc->chart_timestamp, wc->chart_reset_count); + wc->chart_reset_count = 0; + wc->chart_updates = 1; + } + } + } + else + log_access("AC [%s (N/A)]: ACLK synchronization thread is not active.", node_id); + return; + } + host = host->next; + } + rrd_unlock(); + return; +} + +#define SQL_SELECT_HOST_MEMORY_MODE "SELECT memory_mode FROM chart WHERE host_id = @host_id LIMIT 1;" + +static RRD_MEMORY_MODE sql_get_host_memory_mode(uuid_t *host_id) +{ + int rc; + + RRD_MEMORY_MODE memory_mode = RRD_MEMORY_MODE_RAM; + sqlite3_stmt *res = NULL; + + rc = sqlite3_prepare_v2(db_meta, SQL_SELECT_HOST_MEMORY_MODE, -1, &res, 0); + + if (unlikely(rc != SQLITE_OK)) { + error_report("Failed to prepare statement to read host memory mode"); + return memory_mode; + } + + rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC); + if (unlikely(rc != SQLITE_OK)) { + error_report("Failed to bind host parameter to fetch host memory mode"); + goto failed; + } + + while (sqlite3_step(res) == SQLITE_ROW) { + memory_mode = (RRD_MEMORY_MODE) sqlite3_column_int(res, 0); + } + +failed: + rc = sqlite3_finalize(res); + if (unlikely(rc != SQLITE_OK)) + error_report("Failed to finalize the prepared statement when reading host memory mode"); + return memory_mode; +} + +#define SELECT_HOST_DIMENSION_LIST "SELECT d.dim_id, c.update_every, c.type||'.'||c.id, d.id, d.name FROM chart c, dimension d " \ + "WHERE d.chart_id = c.chart_id AND c.host_id = @host_id ORDER BY c.update_every ASC;" + +#define SELECT_HOST_CHART_LIST "SELECT distinct h.host_id, c.update_every, c.type||'.'||c.id FROM chart c, host h " \ + "WHERE c.host_id = h.host_id AND c.host_id = @host_id ORDER BY c.update_every ASC;" + +void aclk_update_retention(struct aclk_database_worker_config *wc, struct aclk_database_cmd cmd) +{ + UNUSED(cmd); + int rc; + + if (!aclk_use_new_cloud_arch || !aclk_connected) + return; + + char *claim_id = is_agent_claimed(); + if (unlikely(!claim_id)) + return; + + sqlite3_stmt *res = NULL; + RRD_MEMORY_MODE memory_mode; + + uuid_t host_uuid; + rc = uuid_parse(wc->host_guid, host_uuid); + if (unlikely(rc)) { + freez(claim_id); + return; + } + + if (wc->host) + memory_mode = wc->host->rrd_memory_mode; + else + memory_mode = sql_get_host_memory_mode(&host_uuid); + + if (memory_mode == RRD_MEMORY_MODE_DBENGINE) + rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_DIMENSION_LIST, -1, &res, 0); + else + rc = sqlite3_prepare_v2(db_meta, SELECT_HOST_CHART_LIST, -1, &res, 0); + + if (unlikely(rc != SQLITE_OK)) { + error_report("Failed to prepare statement to fetch host dimensions"); + freez(claim_id); + return; + } + + rc = sqlite3_bind_blob(res, 1, &host_uuid, sizeof(host_uuid), SQLITE_STATIC); + if (unlikely(rc != SQLITE_OK)) { + error_report("Failed to bind host parameter to fetch host dimensions"); + goto failed; + } + + time_t start_time = LONG_MAX; + time_t first_entry_t; + time_t last_entry_t; + uint32_t update_every = 0; + + struct retention_updated rotate_data; + + memset(&rotate_data, 0, sizeof(rotate_data)); + + int max_intervals = 32; + + rotate_data.interval_duration_count = 0; + rotate_data.interval_durations = callocz(max_intervals, sizeof(*rotate_data.interval_durations)); + + now_realtime_timeval(&rotate_data.rotation_timestamp); + rotate_data.memory_mode = memory_mode; + rotate_data.claim_id = claim_id; + rotate_data.node_id = strdupz(wc->node_id); + + // time_t now = now_realtime_sec(); + while (sqlite3_step(res) == SQLITE_ROW) { + if (!update_every || update_every != (uint32_t) sqlite3_column_int(res, 1)) { + if (update_every) { + debug(D_ACLK_SYNC,"Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time); + if (start_time == LONG_MAX) + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = 0; + else + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = + rotate_data.rotation_timestamp.tv_sec - start_time; + rotate_data.interval_duration_count++; + } + update_every = (uint32_t) sqlite3_column_int(res, 1); + rotate_data.interval_durations[rotate_data.interval_duration_count].update_every = update_every; + start_time = LONG_MAX; + } +#ifdef ENABLE_DBENGINE + if (memory_mode == RRD_MEMORY_MODE_DBENGINE) + rc = rrdeng_metric_latest_time_by_uuid((uuid_t *)sqlite3_column_blob(res, 0), &first_entry_t, &last_entry_t); + else +#endif + { + if (wc->host) { + RRDSET *st = NULL; + rc = (st = rrdset_find(wc->host, (const char *)sqlite3_column_text(res, 2))) ? 0 : 1; + if (!rc) { + first_entry_t = rrdset_first_entry_t(st); + last_entry_t = rrdset_last_entry_t(st); + } + } + else { + rc = 0; + first_entry_t = rotate_data.rotation_timestamp.tv_sec; + } + } + + if (likely(!rc && first_entry_t)) + start_time = MIN(start_time, first_entry_t); + } + if (update_every) { + debug(D_ACLK_SYNC, "Update %s for %u oldest time = %ld", wc->host_guid, update_every, start_time); + if (start_time == LONG_MAX) + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = 0; + else + rotate_data.interval_durations[rotate_data.interval_duration_count].retention = + rotate_data.rotation_timestamp.tv_sec - start_time; + rotate_data.interval_duration_count++; + } + +#ifdef NETDATA_INTERNAL_CHECKS + for (int i = 0; i < rotate_data.interval_duration_count; ++i) + info("Update for host %s (node %s) for %u Retention = %u", wc->host_guid, wc->node_id, + rotate_data.interval_durations[i].update_every, rotate_data.interval_durations[i].retention); +#endif + aclk_retention_updated(&rotate_data); + freez(rotate_data.node_id); + freez(rotate_data.interval_durations); + +failed: + freez(claim_id); + rc = sqlite3_finalize(res); + if (unlikely(rc != SQLITE_OK)) + error_report("Failed to finalize the prepared statement when reading host dimensions"); + return; +} + + +uint32_t sql_get_pending_count(struct aclk_database_worker_config *wc) +{ + BUFFER *sql = buffer_create(1024); + sqlite3_stmt *res = NULL; + + buffer_sprintf(sql,"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; + } + while (sqlite3_step(res) == SQLITE_ROW) + chart_payload_count = (uint32_t) sqlite3_column_int(res, 0); + + rc = sqlite3_finalize(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); + + 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); + + int rc; + 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 to find last chart sequence id"); + goto fail; + } + + wc->chart_sequence_id = 0; + wc->chart_timestamp = 0; + while (sqlite3_step(res) == SQLITE_ROW) { + wc->chart_sequence_id = (uint64_t) sqlite3_column_int64(res, 0); + wc->chart_timestamp = (time_t) sqlite3_column_int64(res, 1); + } + + debug(D_ACLK_SYNC,"Node %s reports last sequence_id=%"PRIu64, wc->node_id, wc->chart_sequence_id); + + rc = sqlite3_finalize(res); + if (unlikely(rc != SQLITE_OK)) + error_report("Failed to reset statement when fetching chart sequence info, rc = %d", rc); + +fail: + buffer_free(sql); + return; +} + +int queue_dimension_to_aclk(RRDDIM *rd) +{ + int rc = sql_queue_chart_payload((struct aclk_database_worker_config *) rd->rrdset->rrdhost->dbsync_worker, + rd, ACLK_DATABASE_ADD_DIMENSION); + return rc; +} + +#endif //ENABLE_NEW_CLOUD_PROTOCOL + +// ST is read locked +int queue_chart_to_aclk(RRDSET *st) +{ +#ifndef ENABLE_NEW_CLOUD_PROTOCOL +#ifdef ENABLE_ACLK + aclk_update_chart(st->rrdhost, st->id, 1); +#endif + return 0; +#else + if (!aclk_use_new_cloud_arch && aclk_connected) { + aclk_update_chart(st->rrdhost, st->id, 1); + return 0; + } + return sql_queue_chart_payload((struct aclk_database_worker_config *) st->rrdhost->dbsync_worker, + st, ACLK_DATABASE_ADD_CHART); +#endif +} + |