netdata/database/sqlite/sqlite_aclk_chart.c

// 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 (%s)]: ACLK synchronization thread is not active.", node_id, host ? host->hostname : "N/A");
    }
    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 (%s)]: ACLK synchronization thread is not active.", wc->node_id, wc->host ? wc->host->hostname : "N/A");
            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
}