flb_fstore.c

/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2024 The Fluent Bit Authors
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_fstore.h>
#include <fluent-bit/flb_log.h>
#include <fluent-bit/flb_mem.h>
#include <fluent-bit/flb_sds.h>
#include <chunkio/chunkio.h>

static int log_cb(struct cio_ctx *ctx, int level, const char *file, int line,
                  char *str)
{
    if (level == CIO_LOG_ERROR) {
        flb_error("[fstore] %s", str);
    }
    else if (level == CIO_LOG_WARN) {
        flb_warn("[fstore] %s", str);
    }
    else if (level == CIO_LOG_INFO) {
        flb_info("[fstore] %s", str);
    }
    else if (level == CIO_LOG_DEBUG) {
        flb_debug("[fstore] %s", str);
    }

    return 0;
}

/*
 * this function sets metadata into a fstore_file structure, note that it makes
 * it own copy of the data to set a NULL byte at the end.
 */
static int meta_set(struct flb_fstore_file *fsf, void *meta, size_t size)
{

    char *p;

    p = flb_calloc(1, size + 1);
    if (!p) {
        flb_errno();
        flb_error("[fstore] could not cache metadata in file: %s:%s",
                  fsf->stream->name, fsf->chunk->name);
        return -1;
    }

    if (fsf->meta_buf) {
        flb_free(fsf->meta_buf);
    }
    fsf->meta_buf = p;
    memcpy(fsf->meta_buf, meta, size);
    fsf->meta_size = size;

    return 0;
}

/* Set a file metadata */
int flb_fstore_file_meta_set(struct flb_fstore *fs,
                             struct flb_fstore_file *fsf,
                             void *meta, size_t size)
{
    int ret;
    int set_down = FLB_FALSE;

    /* Check if the chunk is up */
    if (cio_chunk_is_up(fsf->chunk) == CIO_FALSE) {
        ret = cio_chunk_up_force(fsf->chunk);
        if (ret != CIO_OK) {
            flb_error("[fstore] error loading up file chunk");
            return -1;
        }
        set_down = FLB_TRUE;
    }

    ret = cio_meta_write(fsf->chunk, meta, size);
    if (ret == -1) {
        flb_error("[fstore] could not write metadata to file: %s:%s",
                  fsf->stream->name, fsf->chunk->name);

        if (set_down == FLB_TRUE) {
            cio_chunk_down(fsf->chunk);
        }

        return -1;
    }

    if (set_down == FLB_TRUE) {
        cio_chunk_down(fsf->chunk);
    }

    return meta_set(fsf, meta, size);
}

/* Re-read Chunk I/O metadata into fstore file */
int flb_fstore_file_meta_get(struct flb_fstore *fs,
                             struct flb_fstore_file *fsf)
{
    int ret;
    int set_down = FLB_FALSE;
    char *meta_buf = NULL;
    int meta_size = 0;

    /* Check if the chunk is up */
    if (cio_chunk_is_up(fsf->chunk) == CIO_FALSE) {
        ret = cio_chunk_up_force(fsf->chunk);
        if (ret != CIO_OK) {
            flb_error("[fstore] error loading up file chunk");
            return -1;
        }
        set_down = FLB_TRUE;
    }

    ret = cio_meta_read(fsf->chunk, &meta_buf, &meta_size);
    if (ret == -1) {
        flb_error("[fstore] error reading file chunk metadata");
        if (set_down == FLB_TRUE) {
            cio_chunk_down(fsf->chunk);
        }
    }

    ret = meta_set(fsf, meta_buf, meta_size);
    if (ret == -1) {
        flb_free(meta_buf);
        if (set_down == FLB_TRUE) {
            cio_chunk_down(fsf->chunk);
        }
        return -1;
    }

    if (set_down == FLB_TRUE) {
        cio_chunk_down(fsf->chunk);
    }
    return 0;
}

/* Create a new file */
struct flb_fstore_file *flb_fstore_file_create(struct flb_fstore *fs,
                                               struct flb_fstore_stream *fs_stream,
                                               char *name, size_t size)
{
    int err;
    struct cio_chunk *chunk;
    struct flb_fstore_file *fsf;

    fsf = flb_calloc(1, sizeof(struct flb_fstore_file));
    if (!fsf) {
        flb_errno();
        return NULL;
    }
    fsf->stream = fs_stream->stream;

    fsf->name = flb_sds_create(name);
    if (!fsf->name) {
        flb_error("[fstore] could not create file: %s:%s",
                  fsf->stream->name, name);
        flb_free(fsf);
        return NULL;
    }

    chunk = cio_chunk_open(fs->cio, fs_stream->stream, name,
                           CIO_OPEN, size, &err);
    if (!chunk) {
        flb_error("[fstore] could not create file: %s:%s",
                  fsf->stream->name, name);
        flb_sds_destroy(fsf->name);
        flb_free(fsf);
        return NULL;
    }

    fsf->chunk = chunk;
    mk_list_add(&fsf->_head, &fs_stream->files);

    return fsf;
}

/* Lookup file on stream by using it name */
struct flb_fstore_file *flb_fstore_file_get(struct flb_fstore *fs,
                                            struct flb_fstore_stream *fs_stream,
                                            char *name, size_t size)
{
    struct mk_list *head;
    struct flb_fstore_file *fsf;

    mk_list_foreach(head, &fs_stream->files) {
        fsf = mk_list_entry(head, struct flb_fstore_file, _head);
        if (flb_sds_len(fsf->name) != size) {
            continue;
        }

        if (strncmp(fsf->name, name, size) == 0) {
            return fsf;
        }
    }

    return NULL;
}

/*
 * Set a file to inactive mode. Inactive means just to remove the reference
 * from the list.
 */
int flb_fstore_file_inactive(struct flb_fstore *fs,
                             struct flb_fstore_file *fsf)
{
    /* close the Chunk I/O reference, but don't delete the real file */
    if (fsf->chunk) {
        cio_chunk_close(fsf->chunk, CIO_FALSE);
    }

    /* release */
    mk_list_del(&fsf->_head);
    flb_sds_destroy(fsf->name);
    if (fsf->meta_buf) {
        flb_free(fsf->meta_buf);
    }
    flb_free(fsf);

    return 0;
}

/* Delete a file (permantent deletion) */
int flb_fstore_file_delete(struct flb_fstore *fs,
                           struct flb_fstore_file *fsf)
{
    /* close the Chunk I/O reference, but don't delete it the real file */
    cio_chunk_close(fsf->chunk, CIO_TRUE);

    /* release */
    mk_list_del(&fsf->_head);
    if (fsf->meta_buf) {
        flb_free(fsf->meta_buf);
    }
    flb_sds_destroy(fsf->name);
    flb_free(fsf);

    return 0;
}

/*
 * Set an output buffer that contains a copy of the file. Note that this buffer
 * needs to be freed by the caller (heap memory).
 */
int flb_fstore_file_content_copy(struct flb_fstore *fs,
                                 struct flb_fstore_file *fsf,
                                 void **out_buf, size_t *out_size)
{
    int ret;

    ret = cio_chunk_get_content_copy(fsf->chunk, out_buf, out_size);
    if (ret == CIO_OK) {
        return 0;
    }

    return -1;
}

/* Append data to an existing file */
int flb_fstore_file_append(struct flb_fstore_file *fsf, void *data, size_t size)
{
    int ret;
    int set_down = FLB_FALSE;

    /* Check if the chunk is up */
    if (cio_chunk_is_up(fsf->chunk) == CIO_FALSE) {
        ret = cio_chunk_up_force(fsf->chunk);
        if (ret != CIO_OK) {
            flb_error("[fstore] error loading up file chunk");
            return -1;
        }
        set_down = FLB_TRUE;
    }

    ret = cio_chunk_write(fsf->chunk, data, size);
    if (ret != CIO_OK) {
        flb_error("[fstore] could not write data to file %s", fsf->name);

        if (set_down == FLB_TRUE) {
            cio_chunk_down(fsf->chunk);
        }

        return -1;
    }

    if (set_down == FLB_TRUE) {
        cio_chunk_down(fsf->chunk);
    }

    return 0;
}

/*
 * Create a new stream, if it already exists, it returns the stream
 * reference.
 */
struct flb_fstore_stream *flb_fstore_stream_create(struct flb_fstore *fs,
                                                   char *stream_name)
{
    flb_sds_t path = NULL;
    struct mk_list *head;
    struct cio_ctx *ctx = NULL;
    struct cio_stream *stream = NULL;
    struct flb_fstore_stream *fs_stream = NULL;

    ctx = fs->cio;

    /* Check if the stream already exists in Chunk I/O */
    mk_list_foreach(head, &ctx->streams) {
        stream = mk_list_entry(head, struct cio_stream, _head);
        if (strcmp(stream->name, stream_name) == 0) {
            break;
        }
        stream = NULL;
    }

    /* If the stream exists, check if we have a fstore_stream reference */
    if (stream) {
        mk_list_foreach(head, &fs->streams) {
            fs_stream = mk_list_entry(head, struct flb_fstore_stream, _head);
            if (fs_stream->stream == stream) {
                break;
            }
            fs_stream = NULL;
        }

        /* The stream was found, just return the reference */
        if (fs_stream) {
            return fs_stream;
        }
    }

    if (!stream) {
        /* create file-system based stream */
        stream = cio_stream_create(fs->cio, stream_name, fs->store_type);
        if (!stream) {
            flb_error("[fstore] cannot create stream %s", stream_name);
            return NULL;
        }
    }

    fs_stream = flb_calloc(1, sizeof(struct flb_fstore_stream));
    if (!fs_stream) {
        flb_errno();
        cio_stream_destroy(stream);
        return NULL;
    }
    fs_stream->stream = stream;

    path = flb_sds_create_size(256);
    if (!path) {
        cio_stream_destroy(stream);
        flb_free(fs_stream);
        return NULL;
    }
    path = flb_sds_printf(&path, "%s/%s", fs->root_path, stream->name);
    fs_stream->path = path;
    fs_stream->name = stream->name;

    mk_list_init(&fs_stream->files);
    mk_list_add(&fs_stream->_head, &fs->streams);

    return fs_stream;
}

void flb_fstore_stream_destroy(struct flb_fstore_stream *stream, int delete)
{
    if (delete == FLB_TRUE) {
        cio_stream_delete(stream->stream);
    }

    /*
     * FYI: in this function we just release the fstore_stream context, the
     * underlaying cio_stream is closed when the main Chunk I/O is destroyed.
     */
    mk_list_del(&stream->_head);
    flb_sds_destroy(stream->path);
    flb_free(stream);
}

static int map_chunks(struct flb_fstore *ctx, struct flb_fstore_stream *fs_stream,
                      struct cio_stream *stream)
{
    struct mk_list *head;
    struct cio_chunk *chunk;
    struct flb_fstore_file *fsf;

    mk_list_foreach(head, &stream->chunks) {
        chunk = mk_list_entry(head, struct cio_chunk, _head);

        fsf = flb_calloc(1, sizeof(struct flb_fstore_file));
        if (!fsf) {
            flb_errno();
            return -1;
        }
        fsf->name = flb_sds_create(chunk->name);
        if (!fsf->name) {
            flb_free(fsf);
            flb_error("[fstore] could not create file: %s:%s",
                      stream->name, chunk->name);
            return -1;
        }

        fsf->chunk = chunk;

        /* load metadata */
        flb_fstore_file_meta_get(ctx, fsf);
        mk_list_add(&fsf->_head, &fs_stream->files);
    }

    return 0;
}

static int load_references(struct flb_fstore *fs)
{
    int ret;
    struct mk_list *head;
    struct cio_stream *stream;
    struct flb_fstore_stream *fs_stream;

    mk_list_foreach(head, &fs->cio->streams) {
        stream = mk_list_entry(head, struct cio_stream, _head);
        fs_stream = flb_fstore_stream_create(fs, stream->name);
        if (!fs_stream) {
            flb_error("[fstore] error loading stream reference: %s",
                      stream->name);
            return -1;
        }

        /* Map chunks */
        ret = map_chunks(fs, fs_stream, stream);
        if (ret == -1) {
            return -1;
        }
    }

    return 0;
}

struct flb_fstore *flb_fstore_create(char *path, int store_type)
{
    int ret;
    int flags;
    struct cio_ctx *cio;
    struct flb_fstore *fs;
    struct cio_options opts = {0};
    flags = CIO_OPEN;

    /* Create Chunk I/O context */
    cio_options_init(&opts);

    opts.root_path = path;
    opts.log_cb = log_cb;
    opts.flags = flags;
    opts.log_level = CIO_LOG_INFO;

    cio = cio_create(&opts);
    if (!cio) {
        flb_error("[fstore] error initializing on path '%s'", path);
        return NULL;
    }

    /* Load content from the file system if any */
    ret = cio_load(cio, NULL);
    if (ret == -1) {
        flb_error("[fstore] error scanning root path content: %s", path);
        cio_destroy(cio);
        return NULL;
    }

    fs = flb_calloc(1, sizeof(struct flb_fstore));
    if (!fs) {
        flb_errno();
        cio_destroy(cio);
        return NULL;
    }
    fs->cio = cio;
    fs->root_path = cio->options.root_path;
    fs->store_type = store_type;
    mk_list_init(&fs->streams);

    /* Map Chunk I/O streams and chunks into fstore context */
    load_references(fs);

    return fs;
}

int flb_fstore_destroy(struct flb_fstore *fs)
{
    int files = 0;
    int delete;
    struct mk_list *head;
    struct mk_list *f_head;
    struct mk_list *tmp;
    struct mk_list *f_tmp;
    struct flb_fstore_stream *fs_stream;
    struct flb_fstore_file *fsf;

    mk_list_foreach_safe(head, tmp, &fs->streams) {
        fs_stream = mk_list_entry(head, struct flb_fstore_stream, _head);

        /* delete file references */
        files = 0;
        mk_list_foreach_safe(f_head, f_tmp, &fs_stream->files) {
            fsf = mk_list_entry(f_head, struct flb_fstore_file, _head);
            flb_fstore_file_inactive(fs, fsf);
            files++;
        }

        if (files == 0) {
            delete = FLB_TRUE;
        }
        else {
            delete = FLB_FALSE;
        }

        flb_fstore_stream_destroy(fs_stream, delete);
    }

    if (fs->cio) {
        cio_destroy(fs->cio);
    }
    flb_free(fs);
    return 0;
}

void flb_fstore_dump(struct flb_fstore *fs)
{
    struct mk_list *head;
    struct mk_list *f_head;
    struct flb_fstore_stream *fs_stream;
    struct flb_fstore_file *fsf;

    printf("===== FSTORE DUMP =====\n");
    mk_list_foreach(head, &fs->streams) {
        fs_stream = mk_list_entry(head, struct flb_fstore_stream, _head);
        printf("- stream: %s\n", fs_stream->name);
        mk_list_foreach(f_head, &fs_stream->files) {
            fsf = mk_list_entry(f_head, struct flb_fstore_file, _head);
            printf("          %s/%s\n", fsf->stream->name, fsf->name);
        }
    }
    printf("\n");
}