path: root/src/database/engine/rrdengine.h

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

#ifndef NETDATA_RRDENGINE_H
#define NETDATA_RRDENGINE_H

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif
#include <fcntl.h>
#include <lz4.h>
#include <Judy.h>
#include <openssl/sha.h>
#include <openssl/evp.h>
#include "daemon/common.h"
#include "../rrd.h"
#include "rrddiskprotocol.h"
#include "rrdenginelib.h"
#include "datafile.h"
#include "journalfile.h"
#include "rrdengineapi.h"
#include "pagecache.h"
#include "metric.h"
#include "cache.h"
#include "pdc.h"
#include "page.h"

extern unsigned rrdeng_pages_per_extent;

/* Forward declarations */
struct rrdengine_instance;
struct rrdeng_cmd;

#define MAX_PAGES_PER_EXTENT (64) /* TODO: can go higher only when journal supports bigger than 4KiB transactions */

#define RRDENG_FILE_NUMBER_SCAN_TMPL "%1u-%10u"
#define RRDENG_FILE_NUMBER_PRINT_TMPL "%1.1u-%10.10u"

typedef enum __attribute__ ((__packed__)) {
    // final status for all pages
    // if a page does not have one of these, it is considered unroutable
    PDC_PAGE_READY     = (1 << 0),                  // ready to be processed (pd->page is not null)
    PDC_PAGE_FAILED    = (1 << 1),                  // failed to be loaded (pd->page is null)
    PDC_PAGE_SKIP      = (1 << 2),                  // don't use this page, it is not good for us
    PDC_PAGE_INVALID   = (1 << 3),                  // don't use this page, it is invalid
    PDC_PAGE_EMPTY     = (1 << 4),                  // the page is empty, does not have any data

    // other statuses for tracking issues
    PDC_PAGE_PREPROCESSED              = (1 << 5),  // used during preprocessing
    PDC_PAGE_PROCESSED                 = (1 << 6),  // processed by the query caller
    PDC_PAGE_RELEASED                  = (1 << 7),  // already released

    // data found in cache (preloaded) or on disk?
    PDC_PAGE_PRELOADED                 = (1 << 8),  // data found in memory
    PDC_PAGE_DISK_PENDING              = (1 << 9),  // data need to be loaded from disk

    // worker related statuses
    PDC_PAGE_FAILED_INVALID_EXTENT     = (1 << 10),
    PDC_PAGE_FAILED_NOT_IN_EXTENT      = (1 << 11),
    PDC_PAGE_FAILED_TO_MAP_EXTENT      = (1 << 12),
    PDC_PAGE_FAILED_TO_ACQUIRE_DATAFILE= (1 << 13),

    PDC_PAGE_EXTENT_FROM_CACHE         = (1 << 14),
    PDC_PAGE_EXTENT_FROM_DISK          = (1 << 15),

    PDC_PAGE_CANCELLED                 = (1 << 16), // the query thread had left when we try to load the page

    PDC_PAGE_SOURCE_MAIN_CACHE         = (1 << 17),
    PDC_PAGE_SOURCE_OPEN_CACHE         = (1 << 18),
    PDC_PAGE_SOURCE_JOURNAL_V2         = (1 << 19),
    PDC_PAGE_PRELOADED_PASS4           = (1 << 20),

    // datafile acquired
    PDC_PAGE_DATAFILE_ACQUIRED         = (1 << 30),
} PDC_PAGE_STATUS;

#define PDC_PAGE_QUERY_GLOBAL_SKIP_LIST (PDC_PAGE_FAILED | PDC_PAGE_SKIP | PDC_PAGE_INVALID | PDC_PAGE_RELEASED)

typedef struct page_details_control {
    struct rrdengine_instance *ctx;
    struct metric *metric;

    struct completion prep_completion;
    struct completion page_completion;   // sync between the query thread and the workers

    Pvoid_t page_list_JudyL;        // the list of page details
    unsigned completed_jobs;        // the number of jobs completed last time the query thread checked
    bool workers_should_stop;       // true when the query thread left and the workers should stop
    bool prep_done;

    PDC_PAGE_STATUS common_status;
    size_t pages_to_load_from_disk;

    SPINLOCK refcount_spinlock;     // spinlock to protect refcount
    int32_t refcount;               // the number of workers currently working on this request + 1 for the query thread
    size_t executed_with_gaps;

    time_t start_time_s;
    time_t end_time_s;
    STORAGE_PRIORITY priority;

    time_t optimal_end_time_s;
} PDC;

PDC *pdc_get(void);

struct page_details {
    struct {
        struct rrdengine_datafile *ptr;
        uv_file file;
        unsigned fileno;

        struct {
            uint64_t pos;
            uint32_t bytes;
        } extent;
    } datafile;

    struct pgc_page *page;
    Word_t metric_id;
    time_t first_time_s;
    time_t last_time_s;
    uint32_t update_every_s;
    PDC_PAGE_STATUS status;

    struct {
        struct page_details *prev;
        struct page_details *next;
    } load;
};

struct page_details *page_details_get(void);

#define pdc_page_status_check(pd, flag) (__atomic_load_n(&((pd)->status), __ATOMIC_ACQUIRE) & (flag))
#define pdc_page_status_set(pd, flag)   __atomic_or_fetch(&((pd)->status), flag, __ATOMIC_RELEASE)
#define pdc_page_status_clear(pd, flag) __atomic_and_fetch(&((od)->status), ~(flag), __ATOMIC_RELEASE)

struct jv2_extents_info {
    size_t index;
    uint64_t pos;
    unsigned bytes;
    size_t number_of_pages;
};

struct jv2_metrics_info {
    uuid_t *uuid;
    uint32_t page_list_header;
    time_t first_time_s;
    time_t last_time_s;
    size_t number_of_pages;
    Pvoid_t JudyL_pages_by_start_time;
};

struct jv2_page_info {
    time_t start_time_s;
    time_t end_time_s;
    uint32_t update_every_s;
    uint32_t extent_index;
    size_t page_length;
    void *custom_data;

    // private
    struct pgc_page *page;
};

typedef enum __attribute__ ((__packed__)) {
    RRDENG_1ST_METRIC_WRITER    = (1 << 0),
} RRDENG_COLLECT_HANDLE_OPTIONS;

typedef enum __attribute__ ((__packed__)) {
    RRDENG_PAGE_PAST_COLLECTION       = (1 << 0),
    RRDENG_PAGE_REPEATED_COLLECTION   = (1 << 1),
    RRDENG_PAGE_BIG_GAP               = (1 << 2),
    RRDENG_PAGE_GAP                   = (1 << 3),
    RRDENG_PAGE_FUTURE_POINT          = (1 << 4),
    RRDENG_PAGE_CREATED_IN_FUTURE     = (1 << 5),
    RRDENG_PAGE_COMPLETED_IN_FUTURE   = (1 << 6),
    RRDENG_PAGE_UNALIGNED             = (1 << 7),
    RRDENG_PAGE_CONFLICT              = (1 << 8),
    RRDENG_PAGE_FULL                  = (1 << 9),
    RRDENG_PAGE_COLLECT_FINALIZE      = (1 << 10),
    RRDENG_PAGE_UPDATE_EVERY_CHANGE   = (1 << 11),
    RRDENG_PAGE_STEP_TOO_SMALL        = (1 << 12),
    RRDENG_PAGE_STEP_UNALIGNED        = (1 << 13),
} RRDENG_COLLECT_PAGE_FLAGS;

struct rrdeng_collect_handle {
    struct storage_collect_handle common; // has to be first item

    RRDENG_COLLECT_PAGE_FLAGS page_flags;
    RRDENG_COLLECT_HANDLE_OPTIONS options;
    uint8_t type;

    struct rrdengine_instance *ctx;
    struct metric *metric;
    struct pgc_page *pgc_page;
    struct pgd *page_data;
    size_t page_data_size;
    struct pg_alignment *alignment;
    uint32_t page_entries_max;
    uint32_t page_position;                   // keep track of the current page size, to make sure we don't exceed it
    usec_t page_start_time_ut;
    usec_t page_end_time_ut;
    usec_t update_every_ut;
};

struct rrdeng_query_handle {
    struct metric *metric;
    struct pgc_page *page;
    struct rrdengine_instance *ctx;
    struct pgd_cursor pgdc;
    struct page_details_control *pdc;

    // the request
    time_t start_time_s;
    time_t end_time_s;
    STORAGE_PRIORITY priority;

    // internal data
    time_t now_s;
    uint32_t dt_s;

    unsigned position;
    unsigned entries;

#ifdef NETDATA_INTERNAL_CHECKS
    usec_t started_time_s;
    pid_t query_pid;
    struct rrdeng_query_handle *prev, *next;
#endif
};

struct rrdeng_query_handle *rrdeng_query_handle_get(void);
void rrdeng_query_handle_release(struct rrdeng_query_handle *handle);

enum rrdeng_opcode {
    /* can be used to return empty status or flush the command queue */
    RRDENG_OPCODE_NOOP = 0,