netdata/database/contexts/worker.c

// SPDX-License-Identifier: GPL-3.0-or-later

#include "internal.h"

static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc);

static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused);

static void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc);

static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc);
static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs);

static void rrdcontext_garbage_collect_single_host(RRDHOST *host, bool worker_jobs);
static void rrdcontext_garbage_collect_for_all_hosts(void);

extern usec_t rrdcontext_next_db_rotation_ut;

// ----------------------------------------------------------------------------
// load from SQL

static void rrdinstance_load_clabel(SQL_CLABEL_DATA *sld, void *data) {
    RRDINSTANCE *ri = data;
    rrdlabels_add(ri->rrdlabels, sld->label_key, sld->label_value, sld->label_source);
}

static void rrdinstance_load_dimension(SQL_DIMENSION_DATA *sd, void *data) {
    RRDINSTANCE *ri = data;

    RRDMETRIC trm = {
            .id = string_strdupz(sd->id),
            .name = string_strdupz(sd->name),
            .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomic
    };
    if(sd->hidden) trm.flags |= RRD_FLAG_HIDDEN;

    uuid_copy(trm.uuid, sd->dim_id);

    dictionary_set(ri->rrdmetrics, string2str(trm.id), &trm, sizeof(trm));
}

static void rrdinstance_load_chart_callback(SQL_CHART_DATA *sc, void *data) {
    RRDHOST *host = data;

    RRDCONTEXT tc = {
            .id = string_strdupz(sc->context),
            .title = string_strdupz(sc->title),
            .units = string_strdupz(sc->units),
            .family = string_strdupz(sc->family),
            .priority = sc->priority,
            .chart_type = sc->chart_type,
            .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics
            .rrdhost = host,
    };

    RRDCONTEXT_ACQUIRED *rca = (RRDCONTEXT_ACQUIRED *)dictionary_set_and_acquire_item(host->rrdctx.contexts, string2str(tc.id), &tc, sizeof(tc));
    RRDCONTEXT *rc = rrdcontext_acquired_value(rca);

    RRDINSTANCE tri = {
            .id = string_strdupz(sc->id),
            .name = string_strdupz(sc->name),
            .title = string_strdupz(sc->title),
            .units = string_strdupz(sc->units),
            .family = string_strdupz(sc->family),
            .chart_type = sc->chart_type,
            .priority = sc->priority,
            .update_every_s = sc->update_every,
            .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics
    };
    uuid_copy(tri.uuid, sc->chart_id);

    RRDINSTANCE_ACQUIRED *ria = (RRDINSTANCE_ACQUIRED *)dictionary_set_and_acquire_item(rc->rrdinstances, sc->id, &tri, sizeof(tri));
    RRDINSTANCE *ri = rrdinstance_acquired_value(ria);

    ctx_get_dimension_list(&ri->uuid, rrdinstance_load_dimension, ri);
    ctx_get_label_list(&ri->uuid, rrdinstance_load_clabel, ri);
    rrdinstance_trigger_updates(ri, __FUNCTION__ );
    rrdinstance_release(ria);
    rrdcontext_release(rca);
}

static void rrdcontext_load_context_callback(VERSIONED_CONTEXT_DATA *ctx_data, void *data) {
    RRDHOST *host = data;
    (void)host;

    RRDCONTEXT trc = {
            .id = string_strdupz(ctx_data->id),
            .flags = RRD_FLAG_ARCHIVED | RRD_FLAG_UPDATE_REASON_LOAD_SQL, // no need for atomics

            // no need to set more data here
            // we only need the hub data

            .hub = *ctx_data,
    };
    dictionary_set(host->rrdctx.contexts, string2str(trc.id), &trc, sizeof(trc));
}

void rrdhost_load_rrdcontext_data(RRDHOST *host) {
    if(host->rrdctx.contexts) return;

    rrdhost_create_rrdcontexts(host);
    ctx_get_context_list(&host->host_uuid, rrdcontext_load_context_callback, host);
    ctx_get_chart_list(&host->host_uuid, rrdinstance_load_chart_callback, host);

    RRDCONTEXT *rc;
    dfe_start_read(host->rrdctx.contexts, rc) {
                rrdcontext_trigger_updates(rc, __FUNCTION__ );
            }
    dfe_done(rc);

    rrdcontext_garbage_collect_single_host(host, false);
}

// ----------------------------------------------------------------------------
// version hash calculation

uint64_t rrdcontext_version_hash_with_callback(
        RRDHOST *host,
        void (*callback)(RRDCONTEXT *, bool, void *),
        bool snapshot,
        void *bundle) {

    if(unlikely(!host || !host->rrdctx.contexts)) return 0;

    RRDCONTEXT *rc;
    uint64_t hash = 0;

    // loop through all contexts of the host
    dfe_start_read(host->rrdctx.contexts, rc) {

                rrdcontext_lock(rc);

                if(unlikely(rrd_flag_check(rc, RRD_FLAG_HIDDEN))) {
                    rrdcontext_unlock(rc);
                    continue;
                }

                if(unlikely(callback))
                    callback(rc, snapshot, bundle);

                // skip any deleted contexts
                if(unlikely(rrd_flag_is_deleted(rc))) {
                    rrdcontext_unlock(rc);
                    continue;
                }

                // we use rc->hub.* which has the latest
                // metadata we have sent to the hub

                // if a context is currently queued, rc->hub.* does NOT
                // reflect the queued changes. rc->hub.* is updated with
                // their metadata, after messages are dispatched to hub.

                // when the context is being collected,
                // rc->hub.last_time_t is already zero

                hash += rc->hub.version + rc->hub.last_time_s - rc->hub.first_time_s;

                rrdcontext_unlock(rc);

            }
    dfe_done(rc);

    return hash;
}

// ----------------------------------------------------------------------------
// retention recalculation

static void rrdhost_update_cached_retention(RRDHOST *host, time_t first_time_s, time_t last_time_s, bool global) {
    if(unlikely(!host))
        return;

    spinlock_lock(&host->retention.spinlock);

    if(global) {
        host->retention.first_time_s = first_time_s;
        host->retention.last_time_s = last_time_s;
    }
    else {
        if(!host->retention.first_time_s || first_time_s < host->retention.first_time_s)
            host->retention.first_time_s = first_time_s;

        if(!host->retention.last_time_s || last_time_s > host->retention.last_time_s)
            host->retention.last_time_s = last_time_s;
    }

    spinlock_unlock(&host->retention.spinlock);
}

void rrdcontext_recalculate_context_retention(RRDCONTEXT *rc, RRD_FLAGS reason, bool worker_jobs) {
    rrdcontext_post_process_updates(rc, true, reason, worker_jobs);
}

void rrdcontext_recalculate_host_retention(RRDHOST *host, RRD_FLAGS reason, bool worker_jobs) {
    if(unlikely(!host || !host->rrdctx.contexts)) return;

    time_t first_time_s = 0;
    time_t last_time_s = 0;

    RRDCONTEXT *rc;
    dfe_start_read(host->rrdctx.contexts, rc) {
        rrdcontext_recalculate_context_retention(rc, reason, worker_jobs);

        if(!first_time_s || rc->first_time_s < first_time_s)
            first_time_s = rc->first_time_s;

        if(!last_time_s || rc->last_time_s > last_time_s)
            last_time_s = rc->last_time_s;
    }
    dfe_done(rc);

    rrdhost_update_cached_retention(host, first_time_s, last_time_s, true);
}

static void rrdcontext_recalculate_retention_all_hosts(void) {
    rrdcontext_next_db_rotation_ut = 0;
    RRDHOST *host;
    dfe_start_reentrant(rrdhost_root_index, host) {
        worker_is_busy(WORKER_JOB_RETENTION);
        rrdcontext_recalculate_host_retention(host, RRD_FLAG_UPDATE_REASON_DB_ROTATION, true);
    }
    dfe_done(host);
}

// ----------------------------------------------------------------------------
// garbage collector

bool rrdmetric_update_retention(RRDMETRIC *rm) {
    time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;

    if(rm->rrddim) {
        min_first_time_t = rrddim_first_entry_s(rm->rrddim);
        max_last_time_t = rrddim_last_entry_s(rm->rrddim);
    }
    else {
        RRDHOST *rrdhost = rm->ri->rc->rrdhost;
        for (size_t tier = 0; tier < storage_tiers; tier++) {
            STORAGE_ENGINE *eng = rrdhost->db[tier].eng;

            time_t first_time_t, last_time_t;
            if (eng->api.metric_retention_by_uuid(rrdhost->db[tier].instance, &rm->uuid, &first_time_t, &last_time_t)) {
                if (first_time_t < min_first_time_t)
                    min_first_time_t = first_time_t;

                if (last_time_t > max_last_time_t)
                    max_last_time_t = last_time_t;
            }
        }
    }

    if((min_first_time_t == LONG_MAX || min_first_time_t == 0) && max_last_time_t == 0)
        return false;

    if(min_first_time_t == LONG_MAX)
        min_first_time_t = 0;

    if(min_first_time_t > max_last_time_t) {
        internal_error(true, "RRDMETRIC: retention of '%s' is flipped, first_time_t = %ld, last_time_t = %ld", string2str(rm->id), min_first_time_t, max_last_time_t);
        time_t tmp = min_first_time_t;
        min_first_time_t = max_last_time_t;
        max_last_time_t = tmp;
    }

    // check if retention changed

    if (min_first_time_t != rm->first_time_s) {
        rm->first_time_s = min_first_time_t;
        rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
    }

    if (max_last_time_t != rm->last_time_s) {
        rm->last_time_s = max_last_time_t;
        rrd_flag_set_updated(rm, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
    }

    if(unlikely(!rm->first_time_s && !rm->last_time_s))
        rrd_flag_set_deleted(rm, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);

    rrd_flag_set(rm, RRD_FLAG_LIVE_RETENTION);

    return true;
}

static inline bool rrdmetric_should_be_deleted(RRDMETRIC *rm) {
    if(likely(!rrd_flag_check(rm, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
        return false;

    if(likely(rrd_flag_check(rm, RRD_FLAGS_PREVENTING_DELETIONS)))
        return false;

    if(likely(rm->rrddim))
        return false;

    rrdmetric_update_retention(rm);
    if(rm->first_time_s || rm->last_time_s)
        return false;

    return true;
}

static inline bool rrdinstance_should_be_deleted(RRDINSTANCE *ri) {
    if(likely(!rrd_flag_check(ri, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
        return false;

    if(likely(rrd_flag_check(ri, RRD_FLAGS_PREVENTING_DELETIONS)))
        return false;

    if(likely(ri->rrdset))
        return false;

    if(unlikely(dictionary_referenced_items(ri->rrdmetrics) != 0))
        return false;

    if(unlikely(dictionary_entries(ri->rrdmetrics) != 0))
        return false;

    if(ri->first_time_s || ri->last_time_s)
        return false;

    return true;
}

static inline bool rrdcontext_should_be_deleted(RRDCONTEXT *rc) {
    if(likely(!rrd_flag_check(rc, RRD_FLAGS_REQUIRED_FOR_DELETIONS)))
        return false;

    if(likely(rrd_flag_check(rc, RRD_FLAGS_PREVENTING_DELETIONS)))
        return false;

    if(unlikely(dictionary_referenced_items(rc->rrdinstances) != 0))
        return false;

    if(unlikely(dictionary_entries(rc->rrdinstances) != 0))
        return false;

    if(unlikely(rc->first_time_s || rc->last_time_s))
        return false;

    return true;
}

void rrdcontext_delete_from_sql_unsafe(RRDCONTEXT *rc) {
    // we need to refresh the string pointers in rc->hub
    // in case the context changed values
    rc->hub.id = string2str(rc->id);
    rc->hub.title = string2str(rc->title);
    rc->hub.units = string2str(rc->units);
    rc->hub.family = string2str(rc->family);

    // delete it from SQL
    if(ctx_delete_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
        netdata_log_error("RRDCONTEXT: failed to delete context '%s' version %"PRIu64" from SQL.",
                          rc->hub.id, rc->hub.version);
}

static void rrdcontext_garbage_collect_single_host(RRDHOST *host, bool worker_jobs) {

    internal_error(true, "RRDCONTEXT: garbage collecting context structures of host '%s'", rrdhost_hostname(host));

    RRDCONTEXT *rc;
    dfe_start_reentrant(host->rrdctx.contexts, rc) {
                if(unlikely(worker_jobs && !service_running(SERVICE_CONTEXT))) break;

                if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP);

                rrdcontext_lock(rc);

                RRDINSTANCE *ri;
                dfe_start_reentrant(rc->rrdinstances, ri) {
                            if(unlikely(worker_jobs && !service_running(SERVICE_CONTEXT))) break;

                            RRDMETRIC *rm;
                            dfe_start_write(ri->rrdmetrics, rm) {
                                        if(rrdmetric_should_be_deleted(rm)) {
                                            if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
                                            if(!dictionary_del(ri->rrdmetrics, string2str(rm->id)))
                                                netdata_log_error("RRDCONTEXT: metric '%s' of instance '%s' of context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
                                                                  string2str(rm->id),
                                                                  string2str(ri->id),
                                                                  string2str(rc->id),
                                                                  rrdhost_hostname(host));
                                            else
                                                internal_error(
                                                        true,
                                                        "RRDCONTEXT: metric '%s' of instance '%s' of context '%s' of host '%s', deleted from rrdmetrics dictionary.",
                                                        string2str(rm->id),
                                                        string2str(ri->id),
                                                        string2str(rc->id),
                                                        rrdhost_hostname(host));
                                        }
                                    }
                            dfe_done(rm);

                            if(rrdinstance_should_be_deleted(ri)) {
                                if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
                                if(!dictionary_del(rc->rrdinstances, string2str(ri->id)))
                                    netdata_log_error("RRDCONTEXT: instance '%s' of context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
                                                      string2str(ri->id),
                                                      string2str(rc->id),
                                                      rrdhost_hostname(host));
                                else
                                    internal_error(
                                            true,
                                            "RRDCONTEXT: instance '%s' of context '%s' of host '%s', deleted from rrdmetrics dictionary.",
                                            string2str(ri->id),
                                            string2str(rc->id),
                                            rrdhost_hostname(host));
                            }
                        }
                dfe_done(ri);

                if(unlikely(rrdcontext_should_be_deleted(rc))) {
                    if(worker_jobs) worker_is_busy(WORKER_JOB_CLEANUP_DELETE);
                    rrdcontext_dequeue_from_post_processing(rc);
                    rrdcontext_delete_from_sql_unsafe(rc);

                    if(!dictionary_del(host->rrdctx.contexts, string2str(rc->id)))
                        netdata_log_error("RRDCONTEXT: context '%s' of host '%s', failed to be deleted from rrdmetrics dictionary.",
                              string2str(rc->id),
                              rrdhost_hostname(host));
                    else
                        internal_error(
                                true,
                                "RRDCONTEXT: context '%s' of host '%s', deleted from rrdmetrics dictionary.",
                                string2str(rc->id),
                                rrdhost_hostname(host));
                }

                // the item is referenced in the dictionary
                // so, it is still here to unlock, even if we have deleted it
                rrdcontext_unlock(rc);
            }
    dfe_done(rc);
}

static void rrdcontext_garbage_collect_for_all_hosts(void) {
    RRDHOST *host;
    dfe_start_reentrant(rrdhost_root_index, host) {
        rrdcontext_garbage_collect_single_host(host, true);
    }
    dfe_done(host);
}

// ----------------------------------------------------------------------------
// post processing

static void rrdmetric_process_updates(RRDMETRIC *rm, bool force, RRD_FLAGS reason, bool worker_jobs) {
    if(reason != RRD_FLAG_NONE)
        rrd_flag_set_updated(rm, reason);

    if(!force && !rrd_flag_is_updated(rm) && rrd_flag_check(rm, RRD_FLAG_LIVE_RETENTION) && !rrd_flag_check(rm, RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION))
        return;

    if(worker_jobs)
        worker_is_busy(WORKER_JOB_PP_METRIC);

    if(reason & RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD) {
        rrd_flag_set_archived(rm);
        rrd_flag_set(rm, RRD_FLAG_UPDATE_REASON_DISCONNECTED_CHILD);
    }
    if(rrd_flag_is_deleted(rm) && (reason & RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION))
        rrd_flag_set_archived(rm);

    rrdmetric_update_retention(rm);

    rrd_flag_unset_updated(rm);
}

static void rrdinstance_post_process_updates(RRDINSTANCE *ri, bool force, RRD_FLAGS reason, bool worker_jobs) {
    if(reason != RRD_FLAG_NONE)
        rrd_flag_set_updated(ri, reason);

    if(!force && !rrd_flag_is_updated(ri) && rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION))
        return;

    if(worker_jobs)
        worker_is_busy(WORKER_JOB_PP_INSTANCE);

    time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
    size_t metrics_active = 0, metrics_deleted = 0;
    bool live_retention = true, currently_collected = false;
    if(dictionary_entries(ri->rrdmetrics) > 0) {
        RRDMETRIC *rm;
        dfe_start_read((DICTIONARY *)ri->rrdmetrics, rm) {
                    if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                    RRD_FLAGS reason_to_pass = reason;
                    if(rrd_flag_check(ri, RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION))
                        reason_to_pass |= RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION;

                    rrdmetric_process_updates(rm, force, reason_to_pass, worker_jobs);

                    if(unlikely(!rrd_flag_check(rm, RRD_FLAG_LIVE_RETENTION)))
                        live_retention = false;

                    if (unlikely((rrdmetric_should_be_deleted(rm)))) {
                        metrics_deleted++;
                        continue;
                    }

                    if(!currently_collected && rrd_flag_check(rm, RRD_FLAG_COLLECTED) && rm->first_time_s)
                        currently_collected = true;

                    metrics_active++;

                    if (rm->first_time_s && rm->first_time_s < min_first_time_t)
                        min_first_time_t = rm->first_time_s;

                    if (rm->last_time_s && rm->last_time_s > max_last_time_t)
                        max_last_time_t = rm->last_time_s;
                }
        dfe_done(rm);
    }

    if(unlikely(live_retention && !rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
        rrd_flag_set(ri, RRD_FLAG_LIVE_RETENTION);
    else if(unlikely(!live_retention && rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
        rrd_flag_clear(ri, RRD_FLAG_LIVE_RETENTION);

    if(unlikely(!metrics_active)) {
        // no metrics available

        if(ri->first_time_s) {
            ri->first_time_s = 0;
            rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
        }

        if(ri->last_time_s) {
            ri->last_time_s = 0;
            rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
        }

        rrd_flag_set_deleted(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
    }
    else {
        // we have active metrics...

        if (unlikely(min_first_time_t == LONG_MAX))
            min_first_time_t = 0;

        if (unlikely(min_first_time_t == 0 || max_last_time_t == 0)) {
            if(ri->first_time_s) {
                ri->first_time_s = 0;
                rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
            }

            if(ri->last_time_s) {
                ri->last_time_s = 0;
                rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
            }

            if(likely(live_retention))
                rrd_flag_set_deleted(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
        }
        else {
            rrd_flag_clear(ri, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);

            if (unlikely(ri->first_time_s != min_first_time_t)) {
                ri->first_time_s = min_first_time_t;
                rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
            }

            if (unlikely(ri->last_time_s != max_last_time_t)) {
                ri->last_time_s = max_last_time_t;
                rrd_flag_set_updated(ri, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
            }

            if(likely(currently_collected))
                rrd_flag_set_collected(ri);
            else
                rrd_flag_set_archived(ri);
        }
    }

    rrd_flag_unset_updated(ri);
}

static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAGS reason, bool worker_jobs) {
    if(reason != RRD_FLAG_NONE)
        rrd_flag_set_updated(rc, reason);

    if(worker_jobs)
        worker_is_busy(WORKER_JOB_PP_CONTEXT);

    size_t min_priority_collected = LONG_MAX;
    size_t min_priority_not_collected = LONG_MAX;
    size_t min_priority = LONG_MAX;
    time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
    size_t instances_active = 0, instances_deleted = 0, metrics = 0;
    bool live_retention = true, currently_collected = false, hidden = true;
    if(dictionary_entries(rc->rrdinstances) > 0) {
        RRDINSTANCE *ri;
        dfe_start_reentrant(rc->rrdinstances, ri) {
                    if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                    RRD_FLAGS reason_to_pass = reason;
                    if(rrd_flag_check(rc, RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION))
                        reason_to_pass |= RRD_FLAG_UPDATE_REASON_UPDATE_RETENTION;

                    rrdinstance_post_process_updates(ri, force, reason_to_pass, worker_jobs);

                    if(unlikely(hidden && !rrd_flag_check(ri, RRD_FLAG_HIDDEN)))
                        hidden = false;

                    if(unlikely(live_retention && !rrd_flag_check(ri, RRD_FLAG_LIVE_RETENTION)))
                        live_retention = false;

                    if (unlikely(rrdinstance_should_be_deleted(ri))) {
                        instances_deleted++;
                        continue;
                    }

                    bool ri_collected = rrd_flag_is_collected(ri);

                    if(ri_collected && !rrd_flag_check(ri, RRD_FLAG_MERGED_COLLECTED_RI_TO_RC)) {
                        rrdcontext_update_from_collected_rrdinstance(ri);
                        rrd_flag_set(ri, RRD_FLAG_MERGED_COLLECTED_RI_TO_RC);
                    }

                    if(unlikely(!currently_collected && rrd_flag_is_collected(ri) && ri->first_time_s))
                        currently_collected = true;

                    internal_error(rc->units != ri->units,
                                   "RRDCONTEXT: '%s' rrdinstance '%s' has different units, context '%s', instance '%s'",
                                   string2str(rc->id), string2str(ri->id),
                                   string2str(rc->units), string2str(ri->units));

                    instances_active++;
                    metrics += dictionary_entries(ri->rrdmetrics);

                    if (ri->priority >= RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY) {
                        if(rrd_flag_check(ri, RRD_FLAG_COLLECTED)) {
                            if(ri->priority < min_priority_collected)
                                min_priority_collected = ri->priority;
                        }
                        else {
                            if(ri->priority < min_priority_not_collected)
                                min_priority_not_collected = ri->priority;
                        }
                    }

                    if (ri->first_time_s && ri->first_time_s < min_first_time_t)
                        min_first_time_t = ri->first_time_s;

                    if (ri->last_time_s && ri->last_time_s > max_last_time_t)
                        max_last_time_t = ri->last_time_s;
                }
        dfe_done(ri);

        rc->stats.metrics = metrics;

        if(min_priority_collected != LONG_MAX)
            // use the collected priority
            min_priority = min_priority_collected;
        else
            // use the non-collected priority
            min_priority = min_priority_not_collected;
    }

    {
        bool previous_hidden = rrd_flag_check(rc, RRD_FLAG_HIDDEN);
        if (hidden != previous_hidden) {
            if (hidden && !rrd_flag_check(rc, RRD_FLAG_HIDDEN))
                rrd_flag_set(rc, RRD_FLAG_HIDDEN);
            else if (!hidden && rrd_flag_check(rc, RRD_FLAG_HIDDEN))
                rrd_flag_clear(rc, RRD_FLAG_HIDDEN);
        }

        bool previous_live_retention = rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION);
        if (live_retention != previous_live_retention) {
            if (live_retention && !rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION))
                rrd_flag_set(rc, RRD_FLAG_LIVE_RETENTION);
            else if (!live_retention && rrd_flag_check(rc, RRD_FLAG_LIVE_RETENTION))
                rrd_flag_clear(rc, RRD_FLAG_LIVE_RETENTION);
        }
    }

    rrdcontext_lock(rc);
    rc->pp.executions++;

    if(unlikely(!instances_active)) {
        // we had some instances, but they are gone now...

        if(rc->first_time_s) {
            rc->first_time_s = 0;
            rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
        }

        if(rc->last_time_s) {
            rc->last_time_s = 0;
            rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
        }

        rrd_flag_set_deleted(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
    }
    else {
        // we have some active instances...

        if (unlikely(min_first_time_t == LONG_MAX))
            min_first_time_t = 0;

        if (unlikely(min_first_time_t == 0 && max_last_time_t == 0)) {
            if(rc->first_time_s) {
                rc->first_time_s = 0;
                rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
            }

            if(rc->last_time_s) {
                rc->last_time_s = 0;
                rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
            }

            rrd_flag_set_deleted(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);
        }
        else {
            rrd_flag_clear(rc, RRD_FLAG_UPDATE_REASON_ZERO_RETENTION);

            if (unlikely(rc->first_time_s != min_first_time_t)) {
                rc->first_time_s = min_first_time_t;
                rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_FIRST_TIME_T);
            }

            if (rc->last_time_s != max_last_time_t) {
                rc->last_time_s = max_last_time_t;
                rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_LAST_TIME_T);
            }

            if(likely(currently_collected))
                rrd_flag_set_collected(rc);
            else
                rrd_flag_set_archived(rc);
        }

        if (min_priority != LONG_MAX && rc->priority != min_priority) {
            rc->priority = min_priority;
            rrd_flag_set_updated(rc, RRD_FLAG_UPDATE_REASON_CHANGED_METADATA);
        }
    }

    if(unlikely(rrd_flag_is_updated(rc) && rc->rrdhost->rrdctx.hub_queue)) {
        if(check_if_cloud_version_changed_unsafe(rc, false)) {
            rc->version = rrdcontext_get_next_version(rc);
            dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.hub_queue,
                           string2str(rc->id), rc, sizeof(*rc));
        }
    }

    rrd_flag_unset_updated(rc);
    rrdcontext_unlock(rc);
}

void rrdcontext_queue_for_post_processing(RRDCONTEXT *rc, const char *function __maybe_unused, RRD_FLAGS flags __maybe_unused) {
    if(unlikely(!rc->rrdhost->rrdctx.pp_queue)) return;

    if(!rrd_flag_check(rc, RRD_FLAG_QUEUED_FOR_PP)) {
        dictionary_set((DICTIONARY *)rc->rrdhost->rrdctx.pp_queue,
                       string2str(rc->id),
                       rc,
                       sizeof(*rc));

#if(defined(NETDATA_INTERNAL_CHECKS) && defined(LOG_POST_PROCESSING_QUEUE_INSERTIONS))
        {
            BUFFER *wb_flags = buffer_create(1000);
            rrd_flags_to_buffer(flags, wb_flags);

            BUFFER *wb_reasons = buffer_create(1000);
            rrd_reasons_to_buffer(flags, wb_reasons);

            internal_error(true, "RRDCONTEXT: '%s' update triggered by function %s(), due to flags: %s, reasons: %s",
                           string2str(rc->id), function,
                           buffer_tostring(wb_flags),
                           buffer_tostring(wb_reasons));

            buffer_free(wb_reasons);
            buffer_free(wb_flags);
        }
#endif
    }
}

static void rrdcontext_dequeue_from_post_processing(RRDCONTEXT *rc) {
    if(unlikely(!rc->rrdhost->rrdctx.pp_queue)) return;
    dictionary_del(rc->rrdhost->rrdctx.pp_queue, string2str(rc->id));
}

static void rrdcontext_post_process_queued_contexts(RRDHOST *host) {
    if(unlikely(!host->rrdctx.pp_queue)) return;

    RRDCONTEXT *rc;
    dfe_start_reentrant(host->rrdctx.pp_queue, rc) {
                if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                rrdcontext_dequeue_from_post_processing(rc);
                rrdcontext_post_process_updates(rc, false, RRD_FLAG_NONE, true);
            }
    dfe_done(rc);
}

// ----------------------------------------------------------------------------
// dispatching contexts to cloud

static uint64_t rrdcontext_get_next_version(RRDCONTEXT *rc) {
    time_t now = now_realtime_sec();
    uint64_t version = MAX(rc->version, rc->hub.version);
    version = MAX((uint64_t)now, version);
    version++;
    return version;
}

void rrdcontext_message_send_unsafe(RRDCONTEXT *rc, bool snapshot __maybe_unused, void *bundle __maybe_unused) {

    // save it, so that we know the last version we sent to hub
    rc->version = rc->hub.version = rrdcontext_get_next_version(rc);
    rc->hub.id = string2str(rc->id);
    rc->hub.title = string2str(rc->title);
    rc->hub.units = string2str(rc->units);
    rc->hub.family = string2str(rc->family);
    rc->hub.chart_type = rrdset_type_name(rc->chart_type);
    rc->hub.priority = rc->priority;
    rc->hub.first_time_s = rc->first_time_s;
    rc->hub.last_time_s = rrd_flag_is_collected(rc) ? 0 : rc->last_time_s;
    rc->hub.deleted = rrd_flag_is_deleted(rc) ? true : false;

#ifdef ENABLE_ACLK
    struct context_updated message = {
            .id = rc->hub.id,
            .version = rc->hub.version,
            .title = rc->hub.title,
            .units = rc->hub.units,
            .family = rc->hub.family,
            .chart_type = rc->hub.chart_type,
            .priority = rc->hub.priority,
            .first_entry = rc->hub.first_time_s,
            .last_entry = rc->hub.last_time_s,
            .deleted = rc->hub.deleted,
    };

    if(likely(!rrd_flag_check(rc, RRD_FLAG_HIDDEN))) {
        if (snapshot) {
            if (!rc->hub.deleted)
                contexts_snapshot_add_ctx_update(bundle, &message);
        }
        else
            contexts_updated_add_ctx_update(bundle, &message);
    }
#endif

    // store it to SQL

    if(rrd_flag_is_deleted(rc))
        rrdcontext_delete_from_sql_unsafe(rc);

    else if (ctx_store_context(&rc->rrdhost->host_uuid, &rc->hub) != 0)
        netdata_log_error("RRDCONTEXT: failed to save context '%s' version %"PRIu64" to SQL.", rc->hub.id, rc->hub.version);
}

static bool check_if_cloud_version_changed_unsafe(RRDCONTEXT *rc, bool sending __maybe_unused) {
    bool id_changed = false,
            title_changed = false,
            units_changed = false,
            family_changed = false,
            chart_type_changed = false,
            priority_changed = false,
            first_time_changed = false,
            last_time_changed = false,
            deleted_changed = false;

    RRD_FLAGS flags = rrd_flags_get(rc);

    if(unlikely(string2str(rc->id) != rc->hub.id))
        id_changed = true;

    if(unlikely(string2str(rc->title) != rc->hub.title))
        title_changed = true;

    if(unlikely(string2str(rc->units) != rc->hub.units))
        units_changed = true;

    if(unlikely(string2str(rc->family) != rc->hub.family))
        family_changed = true;

    if(unlikely(rrdset_type_name(rc->chart_type) != rc->hub.chart_type))
        chart_type_changed = true;

    if(unlikely(rc->priority != rc->hub.priority))
        priority_changed = true;

    if(unlikely((uint64_t)rc->first_time_s != rc->hub.first_time_s))
        first_time_changed = true;

    if(unlikely((uint64_t)((flags & RRD_FLAG_COLLECTED) ? 0 : rc->last_time_s) != rc->hub.last_time_s))
        last_time_changed = true;

    if(unlikely(((flags & RRD_FLAG_DELETED) ? true : false) != rc->hub.deleted))
        deleted_changed = true;

    if(unlikely(id_changed || title_changed || units_changed || family_changed || chart_type_changed || priority_changed || first_time_changed || last_time_changed || deleted_changed)) {

        internal_error(LOG_TRANSITIONS,
                       "RRDCONTEXT: %s NEW VERSION '%s'%s of host '%s', version %"PRIu64", title '%s'%s, units '%s'%s, family '%s'%s, chart type '%s'%s, priority %u%s, first_time_t %ld%s, last_time_t %ld%s, deleted '%s'%s, (queued for %llu ms, expected %llu ms)",
                       sending?"SENDING":"QUEUE",
                       string2str(rc->id), id_changed ? " (CHANGED)" : "",
                       rrdhost_hostname(rc->rrdhost),
                       rc->version,
                       string2str(rc->title), title_changed ? " (CHANGED)" : "",
                       string2str(rc->units), units_changed ? " (CHANGED)" : "",
                       string2str(rc->family), family_changed ? " (CHANGED)" : "",
                       rrdset_type_name(rc->chart_type), chart_type_changed ? " (CHANGED)" : "",
                       rc->priority, priority_changed ? " (CHANGED)" : "",
                       rc->first_time_s, first_time_changed ? " (CHANGED)" : "",
                       (flags & RRD_FLAG_COLLECTED) ? 0 : rc->last_time_s, last_time_changed ? " (CHANGED)" : "",
                       (flags & RRD_FLAG_DELETED) ? "true" : "false", deleted_changed ? " (CHANGED)" : "",
                       sending ? (now_realtime_usec() - rc->queue.queued_ut) / USEC_PER_MS : 0,
                       sending ? (rc->queue.scheduled_dispatch_ut - rc->queue.queued_ut) / USEC_PER_MS : 0
        );

        rrdhost_update_cached_retention(rc->rrdhost, rc->first_time_s, rc->last_time_s, false);

        return true;
    }

    if(!(flags & RRD_FLAG_COLLECTED))
        rrdhost_update_cached_retention(rc->rrdhost, rc->first_time_s, rc->last_time_s, false);

    return false;
}

static inline usec_t rrdcontext_calculate_queued_dispatch_time_ut(RRDCONTEXT *rc, usec_t now_ut) {

    if(likely(rc->queue.delay_calc_ut >= rc->queue.queued_ut))
        return rc->queue.scheduled_dispatch_ut;

    RRD_FLAGS flags = rc->queue.queued_flags;

    usec_t delay = LONG_MAX;
    int i;
    struct rrdcontext_reason *reason;
    for(i = 0, reason = &rrdcontext_reasons[i]; reason->name ; reason = &rrdcontext_reasons[++i]) {
        if(unlikely(flags & reason->flag)) {
            if(reason->delay_ut < delay)
                delay = reason->delay_ut;
        }
    }

    if(unlikely(delay == LONG_MAX)) {
        internal_error(true, "RRDCONTEXT: '%s', cannot find minimum delay of flags %x", string2str(rc->id), (unsigned int)flags);
        delay = 60 * USEC_PER_SEC;
    }

    rc->queue.delay_calc_ut = now_ut;
    usec_t dispatch_ut = rc->queue.scheduled_dispatch_ut = rc->queue.queued_ut + delay;
    return dispatch_ut;
}

static void rrdcontext_dequeue_from_hub_queue(RRDCONTEXT *rc) {
    dictionary_del(rc->rrdhost->rrdctx.hub_queue, string2str(rc->id));
}

static void rrdcontext_dispatch_queued_contexts_to_hub(RRDHOST *host, usec_t now_ut) {

    // check if we have received a streaming command for this host
    if(!rrdhost_flag_check(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS) || !aclk_connected || !host->rrdctx.hub_queue)
        return;

    // check if there are queued items to send
    if(!dictionary_entries(host->rrdctx.hub_queue))
        return;

    if(!host->node_id)
        return;

    size_t messages_added = 0;
    contexts_updated_t bundle = NULL;

    RRDCONTEXT *rc;
    dfe_start_reentrant(host->rrdctx.hub_queue, rc) {
                if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                if(unlikely(messages_added >= MESSAGES_PER_BUNDLE_TO_SEND_TO_HUB_PER_HOST))
                    break;

                worker_is_busy(WORKER_JOB_QUEUED);
                usec_t dispatch_ut = rrdcontext_calculate_queued_dispatch_time_ut(rc, now_ut);
                char *claim_id = get_agent_claimid();

                if(unlikely(now_ut >= dispatch_ut) && claim_id) {
                    worker_is_busy(WORKER_JOB_CHECK);

                    rrdcontext_lock(rc);

                    if(check_if_cloud_version_changed_unsafe(rc, true)) {
                        worker_is_busy(WORKER_JOB_SEND);

#ifdef ENABLE_ACLK
                        if(!bundle) {
                            // prepare the bundle to send the messages
                            char uuid[UUID_STR_LEN];
                            uuid_unparse_lower(*host->node_id, uuid);

                            bundle = contexts_updated_new(claim_id, uuid, 0, now_ut);
                        }
#endif
                        // update the hub data of the context, give a new version, pack the message
                        // and save an update to SQL
                        rrdcontext_message_send_unsafe(rc, false, bundle);
                        messages_added++;

                        rc->queue.dispatches++;
                        rc->queue.dequeued_ut = now_ut;
                    }
                    else
                        rc->version = rc->hub.version;

                    // remove it from the queue
                    worker_is_busy(WORKER_JOB_DEQUEUE);
                    rrdcontext_dequeue_from_hub_queue(rc);

                    if(unlikely(rrdcontext_should_be_deleted(rc))) {
                        // this is a deleted context - delete it forever...

                        worker_is_busy(WORKER_JOB_CLEANUP_DELETE);

                        rrdcontext_dequeue_from_post_processing(rc);
                        rrdcontext_delete_from_sql_unsafe(rc);

                        STRING *id = string_dup(rc->id);
                        rrdcontext_unlock(rc);

                        // delete it from the master dictionary
                        if(!dictionary_del(host->rrdctx.contexts, string2str(rc->id)))
                            netdata_log_error("RRDCONTEXT: '%s' of host '%s' failed to be deleted from rrdcontext dictionary.",
                                              string2str(id), rrdhost_hostname(host));

                        string_freez(id);
                    }
                    else
                        rrdcontext_unlock(rc);
                }
                freez(claim_id);
            }
    dfe_done(rc);

#ifdef ENABLE_ACLK
    if(service_running(SERVICE_CONTEXT) && bundle) {
        // we have a bundle to send messages

        // update the version hash
        contexts_updated_update_version_hash(bundle, rrdcontext_version_hash(host));

        // send it
        aclk_send_contexts_updated(bundle);
    }
    else if(bundle)
        contexts_updated_delete(bundle);
#endif

}

// ----------------------------------------------------------------------------
// worker thread

static void rrdcontext_main_cleanup(void *ptr) {
    struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
    static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

    // custom code
    worker_unregister();

    static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
}

void *rrdcontext_main(void *ptr) {
    netdata_thread_cleanup_push(rrdcontext_main_cleanup, ptr);

            worker_register("RRDCONTEXT");
            worker_register_job_name(WORKER_JOB_HOSTS, "hosts");
            worker_register_job_name(WORKER_JOB_CHECK, "dedup checks");
            worker_register_job_name(WORKER_JOB_SEND, "sent contexts");
            worker_register_job_name(WORKER_JOB_DEQUEUE, "deduplicated contexts");
            worker_register_job_name(WORKER_JOB_RETENTION, "metrics retention");
            worker_register_job_name(WORKER_JOB_QUEUED, "queued contexts");
            worker_register_job_name(WORKER_JOB_CLEANUP, "cleanups");
            worker_register_job_name(WORKER_JOB_CLEANUP_DELETE, "deletes");
            worker_register_job_name(WORKER_JOB_PP_METRIC, "check metrics");
            worker_register_job_name(WORKER_JOB_PP_INSTANCE, "check instances");
            worker_register_job_name(WORKER_JOB_PP_CONTEXT, "check contexts");

            worker_register_job_custom_metric(WORKER_JOB_HUB_QUEUE_SIZE, "hub queue size", "contexts", WORKER_METRIC_ABSOLUTE);
            worker_register_job_custom_metric(WORKER_JOB_PP_QUEUE_SIZE, "post processing queue size", "contexts", WORKER_METRIC_ABSOLUTE);

            heartbeat_t hb;
            heartbeat_init(&hb);
            usec_t step = RRDCONTEXT_WORKER_THREAD_HEARTBEAT_USEC;

            while (service_running(SERVICE_CONTEXT)) {
                worker_is_idle();
                heartbeat_next(&hb, step);

                if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                usec_t now_ut = now_realtime_usec();

                if(rrdcontext_next_db_rotation_ut && now_ut > rrdcontext_next_db_rotation_ut) {
                    rrdcontext_recalculate_retention_all_hosts();
                    rrdcontext_garbage_collect_for_all_hosts();
                    rrdcontext_next_db_rotation_ut = 0;
                }

                size_t hub_queued_contexts_for_all_hosts = 0;
                size_t pp_queued_contexts_for_all_hosts = 0;

                RRDHOST *host;
                dfe_start_reentrant(rrdhost_root_index, host) {
                    if(unlikely(!service_running(SERVICE_CONTEXT))) break;

                    worker_is_busy(WORKER_JOB_HOSTS);

                    if(host->rrdctx.pp_queue) {
                        pp_queued_contexts_for_all_hosts += dictionary_entries(host->rrdctx.pp_queue);
                        rrdcontext_post_process_queued_contexts(host);
                        dictionary_garbage_collect(host->rrdctx.pp_queue);
                    }

                    if(host->rrdctx.hub_queue) {
                        hub_queued_contexts_for_all_hosts += dictionary_entries(host->rrdctx.hub_queue);
                        rrdcontext_dispatch_queued_contexts_to_hub(host, now_ut);
                        dictionary_garbage_collect(host->rrdctx.hub_queue);
                    }

                    if (host->rrdctx.contexts)
                        dictionary_garbage_collect(host->rrdctx.contexts);

                    // calculate the number of metrics and instances in the host
                    RRDCONTEXT *rc;
                    uint32_t metrics = 0, instances = 0;
                    dfe_start_read(host->rrdctx.contexts, rc) {
                        metrics += rc->stats.metrics;
                        instances += dictionary_entries(rc->rrdinstances);
                    }
                    dfe_done(rc);
                    host->rrdctx.metrics = metrics;
                    host->rrdctx.instances = instances;
                }
                dfe_done(host);

                worker_set_metric(WORKER_JOB_HUB_QUEUE_SIZE, (NETDATA_DOUBLE)hub_queued_contexts_for_all_hosts);
                worker_set_metric(WORKER_JOB_PP_QUEUE_SIZE, (NETDATA_DOUBLE)pp_queued_contexts_for_all_hosts);
            }

    netdata_thread_cleanup_pop(1);
    return NULL;
}