path: root/database/engine/metadata_log/compaction.c

// SPDX-License-Identifier: GPL-3.0-or-later
#define NETDATA_RRD_INTERNALS

#include "metadatalog.h"

void after_compact_old_records(struct metalog_worker_config* wc)
{
    struct metalog_instance *ctx = wc->ctx;
    int error;

    mlf_flush_records_buffer(wc, &ctx->compaction_state.records_log, &ctx->compaction_state.new_metadata_logfiles);
    uv_run(wc->loop, UV_RUN_DEFAULT);

    error = uv_thread_join(wc->now_compacting_files);
    if (error) {
        error("uv_thread_join(): %s", uv_strerror(error));
    }
    freez(wc->now_compacting_files);
    /* unfreeze command processing */
    wc->now_compacting_files = NULL;

    wc->cleanup_thread_compacting_files = 0;

    /* interrupt event loop */
    uv_stop(wc->loop);

    info("Finished metadata log compaction (id:%"PRIu32").", ctx->current_compaction_id);
}

static void metalog_flush_compaction_records(struct metalog_instance *ctx)
{
    struct metalog_cmd cmd;
    struct completion compaction_completion;

    init_completion(&compaction_completion);

    cmd.opcode = METALOG_COMPACTION_FLUSH;
    cmd.record_io_descr.completion = &compaction_completion;
    metalog_enq_cmd(&ctx->worker_config, &cmd);

    wait_for_completion(&compaction_completion);
    destroy_completion(&compaction_completion);
}

/* The caller must have called metalog_flush_compaction_records() before to synchronize and quiesce the event loop. */
static void compaction_test_quota(struct metalog_worker_config *wc)
{
    struct metalog_instance *ctx = wc->ctx;
    struct logfile_compaction_state *compaction_state;
    struct metadata_logfile *oldmetalogfile, *newmetalogfile;
    unsigned current_size;
    int ret;

    compaction_state = &ctx->compaction_state;
    newmetalogfile = compaction_state->new_metadata_logfiles.last;

    oldmetalogfile = ctx->metadata_logfiles.first;

    current_size = newmetalogfile->pos;
    if (unlikely(current_size >= MAX_METALOGFILE_SIZE && newmetalogfile->starting_fileno < oldmetalogfile->fileno)) {
        /* It's safe to finalize the compacted metadata log file and create a new one since it has already replaced
         * an older one. */

        /* Finalize as the immediately previous file than the currently compacted one. */
        ret = rename_metadata_logfile(newmetalogfile, 0, newmetalogfile->fileno - 1);
        if (ret < 0)
            return;

        ret = add_new_metadata_logfile(ctx, &compaction_state->new_metadata_logfiles,
                                       ctx->metadata_logfiles.first->fileno, ctx->metadata_logfiles.first->fileno);

        if (likely(!ret)) {
            compaction_state->fileno = ctx->metadata_logfiles.first->fileno;
        }
    }
}


static void compact_record_by_uuid(struct metalog_instance *ctx, uuid_t *uuid)
{
    GUID_TYPE ret;
    RRDHOST *host = ctx->rrdeng_ctx->host;
    RRDSET *st;
    RRDDIM *rd;
    BUFFER *buffer;

    ret = find_object_by_guid(uuid, NULL, 0);
    switch (ret) {
        case GUID_TYPE_CHAR:
            assert(0);
            break;
        case GUID_TYPE_CHART:
            st = metalog_get_chart_from_uuid(ctx, uuid);
            if (st) {
                if (ctx->current_compaction_id > st->compaction_id) {
                    st->compaction_id = ctx->current_compaction_id;
                    buffer = metalog_update_chart_buffer(st, ctx->current_compaction_id);
                    metalog_commit_record(ctx, buffer, METALOG_COMMIT_CREATION_RECORD, uuid, 1);
                } else {
                    debug(D_METADATALOG, "Chart has already been compacted, ignoring record.");
                }
            } else {
                debug(D_METADATALOG, "Ignoring nonexistent chart metadata record.");
            }
            break;
        case GUID_TYPE_DIMENSION:
            rd = metalog_get_dimension_from_uuid(ctx, uuid);
            if (rd) {
                if (ctx->current_compaction_id > rd->state->compaction_id) {
                    rd->state->compaction_id = ctx->current_compaction_id;
                    buffer = metalog_update_dimension_buffer(rd);
                    metalog_commit_record(ctx, buffer, METALOG_COMMIT_CREATION_RECORD, uuid, 1);
                } else {
                    debug(D_METADATALOG, "Dimension has already been compacted, ignoring record.");
                }
            } else {
                debug(D_METADATALOG, "Ignoring nonexistent dimension metadata record.");
            }
            break;
        case GUID_TYPE_HOST:
            if (ctx->current_compaction_id > host->compaction_id) {
                host->compaction_id = ctx->current_compaction_id;
                buffer = metalog_update_host_buffer(host);
                metalog_commit_record(ctx, buffer, METALOG_COMMIT_CREATION_RECORD, uuid, 1);
            } else {
                debug(D_METADATALOG, "Host has already been compacted, ignoring record.");
            }
            break;
        case GUID_TYPE_NOTFOUND:
            debug(D_METADATALOG, "Ignoring nonexistent metadata record.");
            break;
        default:
            assert(0);
            break;
    }
}

/* Returns 0 on success. */
static int compact_metadata_logfile_records(struct metalog_instance *ctx, struct metadata_logfile *metalogfile)
{
    struct metalog_worker_config* wc = &ctx->worker_config;
    struct logfile_compaction_state *compaction_state;
    struct metalog_record *record;
    struct metalog_record_block *record_block, *prev_record_block;
    int ret;
    unsigned iterated_records;
#define METADATA_LOG_RECORD_BATCH 128 /* Flush I/O and check sizes whenever this many records have been iterated */

    info("Compacting metadata log file \"%s/"METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION"\".",
         ctx->rrdeng_ctx->dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno);

    compaction_state = &ctx->compaction_state;
    record_block = prev_record_block = NULL;
    iterated_records = 0;
    for (record = mlf_record_get_first(metalogfile) ; record != NULL ; record = mlf_record_get_next(metalogfile)) {
        if ((record_block = metalogfile->records.iterator.current) != prev_record_block) {
            if (prev_record_block) { /* Deallocate iterated record blocks */
                rrd_atomic_fetch_add(&ctx->records_nr, -prev_record_block->records_nr);
                freez(prev_record_block);
            }
            prev_record_block = record_block;
        }
        compact_record_by_uuid(ctx, &record->uuid);
        if (0 == ++iterated_records % METADATA_LOG_RECORD_BATCH) {
            metalog_flush_compaction_records(ctx);
            if (compaction_state->throttle) {
                (void)sleep_usec(10000); /* 10 msec throttle compaction */
            }
            compaction_test_quota(wc);
        }
    }
    if (prev_record_block) { /* Deallocate iterated record blocks */
        rrd_atomic_fetch_add(&ctx->records_nr, -prev_record_block->records_nr);
        freez(prev_record_block);
    }

    info("Compacted metadata log file \"%s/"METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION"\".",
         ctx->rrdeng_ctx->dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno);

    metadata_logfile_list_delete(&ctx->metadata_logfiles, metalogfile);
    ret = destroy_metadata_logfile(metalogfile);
    if (!ret) {
        info("Deleted file \"%s/"METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION"\".",
             ctx->rrdeng_ctx->dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno);
        rrd_atomic_fetch_add(&ctx->disk_space, -metalogfile->pos);
    } else {
        error("Failed to delete file \"%s/"METALOG_PREFIX METALOG_FILE_NUMBER_PRINT_TMPL METALOG_EXTENSION"\".",
             ctx->rrdeng_ctx->dbfiles_path, metalogfile->starting_fileno, metalogfile->fileno);
    }
    freez(metalogfile);

    return ret;
}

static void compact_old_records(void *arg)
{
    struct metalog_instance