netdata/logsmanagement/functions.c

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

/** @file functions.c
 *  
 *  @brief This is the file containing the implementation of the 
 *  logs management functions API.
 */

#include "functions.h"
#include "helper.h"
#include "query.h"

#define LOGS_MANAG_MAX_PARAMS       100
#define LOGS_MANAGEMENT_DEFAULT_QUERY_DURATION_IN_SEC  3600
#define LOGS_MANAGEMENT_DEFAULT_ITEMS_PER_QUERY 200

#define LOGS_MANAG_FUNC_PARAM_HELP              "help"
#define LOGS_MANAG_FUNC_PARAM_ANCHOR            "anchor"
#define LOGS_MANAG_FUNC_PARAM_LAST              "last"
#define LOGS_MANAG_FUNC_PARAM_QUERY             "query"
#define LOGS_MANAG_FUNC_PARAM_FACETS            "facets"
#define LOGS_MANAG_FUNC_PARAM_HISTOGRAM         "histogram"
#define LOGS_MANAG_FUNC_PARAM_DIRECTION         "direction"
#define LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE "if_modified_since"
#define LOGS_MANAG_FUNC_PARAM_DATA_ONLY         "data_only"
#define LOGS_MANAG_FUNC_PARAM_SOURCE            "source"
#define LOGS_MANAG_FUNC_PARAM_INFO              "info"
#define LOGS_MANAG_FUNC_PARAM_ID                "id"
#define LOGS_MANAG_FUNC_PARAM_PROGRESS          "progress"
#define LOGS_MANAG_FUNC_PARAM_SLICE             "slice"
#define LOGS_MANAG_FUNC_PARAM_DELTA             "delta"
#define LOGS_MANAG_FUNC_PARAM_TAIL              "tail"

#define LOGS_MANAG_DEFAULT_DIRECTION            FACETS_ANCHOR_DIRECTION_BACKWARD

#define FACET_MAX_VALUE_LENGTH      8192

#define FUNCTION_LOGSMANAGEMENT_HELP_LONG \
    LOGS_MANAGEMENT_PLUGIN_STR " / " LOGS_MANAG_FUNC_NAME"\n" \
    "\n" \
    FUNCTION_LOGSMANAGEMENT_HELP_SHORT"\n" \
    "\n" \
    "The following parameters are supported::\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_HELP"\n" \
    "      Shows this help message\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_INFO"\n" \
    "      Request initial configuration information about the plugin.\n" \
    "      The key entity returned is the required_params array, which includes\n" \
    "      all the available "LOGS_MANAG_FUNC_NAME" sources.\n" \
    "      When `"LOGS_MANAG_FUNC_PARAM_INFO"` is requested, all other parameters are ignored.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_DATA_ONLY":true or "LOGS_MANAG_FUNC_PARAM_DATA_ONLY":false\n" \
    "      Quickly respond with data requested, without generating a\n" \
    "      `histogram`, `facets` counters and `items`.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_SOURCE":SOURCE\n" \
    "      Query only the specified "LOGS_MANAG_FUNC_NAME" sources.\n" \
    "      Do an `"LOGS_MANAG_FUNC_PARAM_INFO"` query to find the sources.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_BEFORE":TIMESTAMP_IN_SECONDS\n" \
    "      Absolute or relative (to now) timestamp in seconds, to start the query.\n" \
    "      The query is always executed from the most recent to the oldest log entry.\n" \
    "      If not given the default is: now.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_AFTER":TIMESTAMP_IN_SECONDS\n" \
    "      Absolute or relative (to `before`) timestamp in seconds, to end the query.\n" \
    "      If not given, the default is "LOGS_MANAG_STR(-LOGS_MANAGEMENT_DEFAULT_QUERY_DURATION_IN_SEC)".\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_LAST":ITEMS\n" \
    "      The number of items to return.\n" \
    "      The default is "LOGS_MANAG_STR(LOGS_MANAGEMENT_DEFAULT_ITEMS_PER_QUERY)".\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_ANCHOR":TIMESTAMP_IN_MICROSECONDS\n" \
    "      Return items relative to this timestamp.\n" \
    "      The exact items to be returned depend on the query `"LOGS_MANAG_FUNC_PARAM_DIRECTION"`.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_DIRECTION":forward or "LOGS_MANAG_FUNC_PARAM_DIRECTION":backward\n" \
    "      When set to `backward` (default) the items returned are the newest before the\n" \
    "      `"LOGS_MANAG_FUNC_PARAM_ANCHOR"`, (or `"LOGS_MANAG_FUNC_PARAM_BEFORE"` if `"LOGS_MANAG_FUNC_PARAM_ANCHOR"` is not set)\n" \
    "      When set to `forward` the items returned are the oldest after the\n" \
    "      `"LOGS_MANAG_FUNC_PARAM_ANCHOR"`, (or `"LOGS_MANAG_FUNC_PARAM_AFTER"` if `"LOGS_MANAG_FUNC_PARAM_ANCHOR"` is not set)\n" \
    "      The default is: backward\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_QUERY":SIMPLE_PATTERN\n" \
    "      Do a full text search to find the log entries matching the pattern given.\n" \
    "      The plugin is searching for matches on all fields of the database.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE":TIMESTAMP_IN_MICROSECONDS\n" \
    "      Each successful response, includes a `last_modified` field.\n" \
    "      By providing the timestamp to the `"LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE"` parameter,\n" \
    "      the plugin will return 200 with a successful response, or 304 if the source has not\n" \
    "      been modified since that timestamp.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_HISTOGRAM":facet_id\n" \
    "      Use the given `facet_id` for the histogram.\n" \
    "      This parameter is ignored in `"LOGS_MANAG_FUNC_PARAM_DATA_ONLY"` mode.\n" \
    "\n" \
    "   "LOGS_MANAG_FUNC_PARAM_FACETS":facet_id1,facet_id2,facet_id3,...\n" \
    "      Add the given facets to the list of fields for which analysis is required.\n" \
    "      The plugin will offer both a histogram and facet value counters for its values.\n" \
    "      This parameter is ignored in `"LOGS_MANAG_FUNC_PARAM_DATA_ONLY"` mode.\n" \
    "\n" \
    "   facet_id:value_id1,value_id2,value_id3,...\n" \
    "      Apply filters to the query, based on the facet IDs returned.\n" \
    "      Each `facet_id` can be given once, but multiple `facet_ids` can be given.\n" \
    "\n"


extern netdata_mutex_t stdout_mut;

static DICTIONARY *function_query_status_dict = NULL;

static DICTIONARY *used_hashes_registry = NULL;

typedef struct function_query_status {
    bool *cancelled; // a pointer to the cancelling boolean
    usec_t stop_monotonic_ut;

    usec_t started_monotonic_ut;

    // request
    STRING *source;
    usec_t after_ut;
    usec_t before_ut;

    struct {
        usec_t start_ut;
        usec_t stop_ut;
    } anchor;

    FACETS_ANCHOR_DIRECTION direction;
    size_t entries;
    usec_t if_modified_since;
    bool delta;
    bool tail;
    bool data_only;
    bool slice;
    size_t filters;
    usec_t last_modified;
    const char *query;
    const char *histogram;

    // per file progress info
    size_t cached_count;

    // progress statistics
    usec_t matches_setup_ut;
    size_t rows_useful;
    size_t rows_read;
    size_t bytes_read;
    size_t files_matched;
    size_t file_working;
} FUNCTION_QUERY_STATUS;


#define LOGS_MANAG_KEYS_INCLUDED_IN_FACETS      \
    "log_source"                                \
    "|log_type"                                 \
    "|filename"                                 \
    "|basename"                                 \
    "|chartname"                                \
    "|message"                                  \
    ""

static void logsmanagement_function_facets(const char *transaction, char *function, int timeout, bool *cancelled){

    struct rusage start, end;
    getrusage(RUSAGE_THREAD, &start);

    const logs_qry_res_err_t *ret = &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_SERVER_ERR];

    BUFFER *wb = buffer_create(0, NULL);
    buffer_flush(wb);
    buffer_json_initialize(wb, "\"", "\"", 0, true, BUFFER_JSON_OPTIONS_MINIFY);

    usec_t now_monotonic_ut = now_monotonic_usec();
    FUNCTION_QUERY_STATUS tmp_fqs = {
            .cancelled = cancelled,
            .started_monotonic_ut = now_monotonic_ut,
            .stop_monotonic_ut = now_monotonic_ut + (timeout * USEC_PER_SEC),
    };
    FUNCTION_QUERY_STATUS *fqs = NULL;
    const DICTIONARY_ITEM *fqs_item = NULL;

    FACETS *facets = facets_create(50, FACETS_OPTION_ALL_KEYS_FTS,
                                   NULL,
                                   LOGS_MANAG_KEYS_INCLUDED_IN_FACETS,
                                   NULL);
    
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_INFO);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_SOURCE);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_AFTER);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_BEFORE);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_ANCHOR);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_DIRECTION);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_LAST);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_QUERY);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_FACETS);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_HISTOGRAM);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_DATA_ONLY);
    // facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_ID);
    // facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_PROGRESS);
    facets_accepted_param(facets, LOGS_MANAG_FUNC_PARAM_DELTA);
    // facets_accepted_param(facets, JOURNAL_PARAMETER_TAIL);

// #ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
//     facets_accepted_param(facets, JOURNAL_PARAMETER_SLICE);
// #endif // HAVE_SD_JOURNAL_RESTART_FIELDS

    // register the fields in the order you want them on the dashboard

    facets_register_key_name(facets, "log_source",  FACET_KEY_OPTION_FACET | 
                                                    FACET_KEY_OPTION_FTS);

    facets_register_key_name(facets, "log_type",    FACET_KEY_OPTION_FACET | 
                                                    FACET_KEY_OPTION_FTS);

    facets_register_key_name(facets, "filename",    FACET_KEY_OPTION_FACET | 
                                                    FACET_KEY_OPTION_FTS);

    facets_register_key_name(facets, "basename",    FACET_KEY_OPTION_FACET | 
                                                    FACET_KEY_OPTION_FTS);

    facets_register_key_name(facets, "chartname",   FACET_KEY_OPTION_VISIBLE | 
                                                    FACET_KEY_OPTION_FACET | 
                                                    FACET_KEY_OPTION_FTS);

    facets_register_key_name(facets, "message",     FACET_KEY_OPTION_NEVER_FACET | 
                                                    FACET_KEY_OPTION_MAIN_TEXT | 
                                                    FACET_KEY_OPTION_VISIBLE |
                                                    FACET_KEY_OPTION_FTS);

    bool    info        = false, 
            data_only   = false, 
            progress    = false, 
            /* slice       = true, */
            delta       = false, 
            tail        = false;
    time_t after_s = 0, before_s = 0;
    usec_t anchor = 0;
    usec_t if_modified_since = 0;
    size_t last = 0;
    FACETS_ANCHOR_DIRECTION direction = LOGS_MANAG_DEFAULT_DIRECTION;
    const char *query = NULL;
    const char *chart = NULL;
    const char *source = NULL;
    const char *progress_id = NULL;
    // size_t filters = 0;  

    buffer_json_member_add_object(wb, "_request");

    logs_query_params_t query_params = {0};
    unsigned long req_quota = 0;

    // unsigned int fn_off = 0, cn_off = 0;

    char *words[LOGS_MANAG_MAX_PARAMS] = { NULL };
    size_t num_words = quoted_strings_splitter_pluginsd(function, words, LOGS_MANAG_MAX_PARAMS);
    for(int i = 1; i < LOGS_MANAG_MAX_PARAMS ; i++) {
        char *keyword = get_word(words, num_words, i);
        if(!keyword) break;

        
        if(!strcmp(keyword, LOGS_MANAG_FUNC_PARAM_HELP)){
            BUFFER *wb = buffer_create(0, NULL);
            buffer_sprintf(wb, FUNCTION_LOGSMANAGEMENT_HELP_LONG);
            netdata_mutex_lock(&stdout_mut);
            pluginsd_function_result_to_stdout(transaction, HTTP_RESP_OK, "text/plain", now_realtime_sec() + 3600, wb);
            netdata_mutex_unlock(&stdout_mut);
            buffer_free(wb);
            goto cleanup;
        }
        else if(!strcmp(keyword, LOGS_MANAG_FUNC_PARAM_INFO)){
            info = true;
        }
        else if(!strcmp(keyword, LOGS_MANAG_FUNC_PARAM_PROGRESS)){
            progress = true;
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_DELTA ":", sizeof(LOGS_MANAG_FUNC_PARAM_DELTA ":") - 1) == 0) {
            char *v = &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_DELTA ":") - 1];

            if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
                delta = false;
            else
                delta = true;
        }
        // else if(strncmp(keyword, JOURNAL_PARAMETER_TAIL ":", sizeof(JOURNAL_PARAMETER_TAIL ":") - 1) == 0) {
        //     char *v = &keyword[sizeof(JOURNAL_PARAMETER_TAIL ":") - 1];

        //     if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
        //         tail = false;
        //     else
        //         tail = true;
        // }
        else if(!strncmp(   keyword, 
                            LOGS_MANAG_FUNC_PARAM_DATA_ONLY ":", 
                            sizeof(LOGS_MANAG_FUNC_PARAM_DATA_ONLY ":") - 1)) {

            char *v = &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_DATA_ONLY ":") - 1];

            if(!strcmp(v, "false") || !strcmp(v, "no") || !strcmp(v, "0"))
                data_only = false;
            else
                data_only = true;
        }
        // else if(strncmp(keyword, JOURNAL_PARAMETER_SLICE ":", sizeof(JOURNAL_PARAMETER_SLICE ":") - 1) == 0) {
        //     char *v = &keyword[sizeof(JOURNAL_PARAMETER_SLICE ":") - 1];

        //     if(strcmp(v, "false") == 0 || strcmp(v, "no") == 0 || strcmp(v, "0") == 0)
        //         slice = false;
        //     else
        //         slice = true;
        // }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_ID ":", sizeof(LOGS_MANAG_FUNC_PARAM_ID ":") - 1) == 0) {
            char *id = &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_ID ":") - 1];

            if(*id)
                progress_id = id;
        }

        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_SOURCE ":", sizeof(LOGS_MANAG_FUNC_PARAM_SOURCE ":") - 1) == 0) {
            source = !strcmp("all", &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_SOURCE ":") - 1]) ? 
                NULL : &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_SOURCE ":") - 1];
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_AFTER ":", sizeof(LOGS_MANAG_FUNC_PARAM_AFTER ":") - 1) == 0) {
            after_s = str2l(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_AFTER ":") - 1]);
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_BEFORE ":", sizeof(LOGS_MANAG_FUNC_PARAM_BEFORE ":") - 1) == 0) {
            before_s = str2l(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_BEFORE ":") - 1]);
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE ":", sizeof(LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE ":") - 1) == 0) {
            if_modified_since = str2ull(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE ":") - 1], NULL);
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_ANCHOR ":", sizeof(LOGS_MANAG_FUNC_PARAM_ANCHOR ":") - 1) == 0) {
            anchor = str2ull(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_ANCHOR ":") - 1], NULL);
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_DIRECTION ":", sizeof(LOGS_MANAG_FUNC_PARAM_DIRECTION ":") - 1) == 0) {
            direction = !strcasecmp(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_DIRECTION ":") - 1], "forward") ? 
                FACETS_ANCHOR_DIRECTION_FORWARD : FACETS_ANCHOR_DIRECTION_BACKWARD;
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_LAST ":", sizeof(LOGS_MANAG_FUNC_PARAM_LAST ":") - 1) == 0) {
            last = str2ul(&keyword[sizeof(LOGS_MANAG_FUNC_PARAM_LAST ":") - 1]);
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_QUERY ":", sizeof(LOGS_MANAG_FUNC_PARAM_QUERY ":") - 1) == 0) {
            query= &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_QUERY ":") - 1];
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_HISTOGRAM ":", sizeof(LOGS_MANAG_FUNC_PARAM_HISTOGRAM ":") - 1) == 0) {
            chart = &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_HISTOGRAM ":") - 1];
        }
        else if(strncmp(keyword, LOGS_MANAG_FUNC_PARAM_FACETS ":", sizeof(LOGS_MANAG_FUNC_PARAM_FACETS ":") - 1) == 0) {
            char *value = &keyword[sizeof(LOGS_MANAG_FUNC_PARAM_FACETS ":") - 1];
            if(*value) {
                buffer_json_member_add_array(wb, LOGS_MANAG_FUNC_PARAM_FACETS);

                while(value) {
                    char *sep = strchr(value, ',');
                    if(sep)
                        *sep++ = '\0';

                    facets_register_facet_id(facets, value, FACET_KEY_OPTION_FACET|FACET_KEY_OPTION_FTS|FACET_KEY_OPTION_REORDER);
                    buffer_json_add_array_item_string(wb, value);

                    value = sep;
                }

                buffer_json_array_close(wb); // LOGS_MANAG_FUNC_PARAM_FACETS
            }
        }
        else {
            char *value = strchr(keyword, ':');
            if(value) {
                *value++ = '\0';

                buffer_json_member_add_array(wb, keyword);

                while(value) {
                    char *sep = strchr(value, ',');
                    if(sep)
                        *sep++ = '\0';

                    facets_register_facet_id_filter(facets, keyword, value, FACET_KEY_OPTION_FACET|FACET_KEY_OPTION_FTS|FACET_KEY_OPTION_REORDER);
                    buffer_json_add_array_item_string(wb, value);
                    // filters++;

                    value = sep;
                }

                buffer_json_array_close(wb); // keyword
            }
        }
    }

    // ------------------------------------------------------------------------
    // put this request into the progress db

    if(progress_id && *progress_id) {
        fqs_item = dictionary_set_and_acquire_item(function_query_status_dict, progress_id, &tmp_fqs, sizeof(tmp_fqs));
        fqs = dictionary_acquired_item_value(fqs_item);
    }
    else {
        // no progress id given, proceed without registering our progress in the dictionary
        fqs = &tmp_fqs;
        fqs_item = NULL;
    }

    // ------------------------------------------------------------------------
    // validate parameters

    time_t now_s = now_realtime_sec();
    time_t expires = now_s + 1;

    if(!after_s && !before_s) {
        before_s = now_s;
        after_s = before_s - LOGS_MANAGEMENT_DEFAULT_QUERY_DURATION_IN_SEC;
    }
    else
        rrdr_relative_window_to_absolute(&after_s, &before_s, now_s);

    if(after_s > before_s) {
        time_t tmp = after_s;
        after_s = before_s;
        before_s = tmp;
    }

    if(after_s == before_s)
        after_s = before_s - LOGS_MANAGEMENT_DEFAULT_QUERY_DURATION_IN_SEC;

    if(!last)
        last = LOGS_MANAGEMENT_DEFAULT_ITEMS_PER_QUERY;


    // ------------------------------------------------------------------------
    // set query time-frame, anchors and direction

    fqs->after_ut = after_s * USEC_PER_SEC;
    fqs->before_ut = (before_s * USEC_PER_SEC) + USEC_PER_SEC - 1;
    fqs->if_modified_since = if_modified_since;
    fqs->data_only = data_only;
    fqs->delta = (fqs->data_only) ? delta : false;
    fqs->tail = (fqs->data_only && fqs->if_modified_since) ? tail : false;
    fqs->source = string_strdupz(source);
    fqs->entries = last;
    fqs->last_modified = 0;
    // fqs->filters = filters;
    fqs->query = (query && *query) ? query : NULL;
    fqs->histogram = (chart && *chart) ? chart : NULL;
    fqs->direction = direction;
    fqs->anchor.start_ut = anchor;
    fqs->anchor.stop_ut = 0;

    if(fqs->anchor.start_ut && fqs->tail) {
        // a tail request
        // we need the top X entries from BEFORE
        // but, we need to calculate the facets and the
        // histogram up to the anchor
        fqs->direction = direction = FACETS_ANCHOR_DIRECTION_BACKWARD;
        fqs->anchor.start_ut = 0;
        fqs->anchor.stop_ut = anchor;
    }

    if(anchor && anchor < fqs->after_ut) {
        // log_fqs(fqs, "received anchor is too small for query timeframe, ignoring anchor");
        anchor = 0;
        fqs->anchor.start_ut = 0;
        fqs->anchor.stop_ut = 0;
        fqs->direction = direction = FACETS_ANCHOR_DIRECTION_BACKWARD;
    }
    else if(anchor > fqs->before_ut) {
        // log_fqs(fqs, "received anchor is too big for query timeframe, ignoring anchor");
        anchor = 0;
        fqs->anchor.start_ut = 0;
        fqs->anchor.stop_ut = 0;
        fqs->direction = direction = FACETS_ANCHOR_DIRECTION_BACKWARD;
    }

    facets_set_anchor(facets, fqs->anchor.start_ut, fqs->anchor.stop_ut, fqs->direction);

    facets_set_additional_options(facets,
                                  ((fqs->data_only) ? FACETS_OPTION_DATA_ONLY : 0) |
                                  ((fqs->delta) ? FACETS_OPTION_SHOW_DELTAS : 0));

    // ------------------------------------------------------------------------
    // set the rest of the query parameters

    facets_set_items(facets, fqs->entries);
    facets_set_query(facets, fqs->query);

// #ifdef HAVE_SD_JOURNAL_RESTART_FIELDS
//     fqs->slice = slice;
//     if(slice)
//         facets_enable_slice_mode(facets);
// #else
//     fqs->slice = false;
// #endif

    if(fqs->histogram)
        facets_set_timeframe_and_histogram_by_id(facets, fqs->histogram, fqs->after_ut, fqs->before_ut);
    else
        facets_set_timeframe_and_histogram_by_name(facets, chart ? chart : "chartname", fqs->after_ut, fqs->before_ut);


    // ------------------------------------------------------------------------
    // complete the request object

    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_INFO, false);
    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_SLICE, fqs->slice);
    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_DATA_ONLY, fqs->data_only);
    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_PROGRESS, false);
    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_DELTA, fqs->delta);
    buffer_json_member_add_boolean(wb, LOGS_MANAG_FUNC_PARAM_TAIL, fqs->tail);
    buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_ID, progress_id);
    buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_SOURCE, string2str(fqs->source));
    buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_AFTER, fqs->after_ut / USEC_PER_SEC);
    buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_BEFORE, fqs->before_ut / USEC_PER_SEC);
    buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_IF_MODIFIED_SINCE, fqs->if_modified_since);
    buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_ANCHOR, anchor);
    buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_DIRECTION, 
        fqs->direction == FACETS_ANCHOR_DIRECTION_FORWARD ? "forward" : "backward");
    buffer_json_member_add_uint64(wb, LOGS_MANAG_FUNC_PARAM_LAST, fqs->entries);
    buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_QUERY, fqs->query);
    buffer_json_member_add_string(wb, LOGS_MANAG_FUNC_PARAM_HISTOGRAM, fqs->histogram);
    buffer_json_object_close(wb); // request

    // buffer_json_journal_versions(wb);

    // ------------------------------------------------------------------------
    // run the request

    if(info) {
        facets_accepted_parameters_to_json_array(facets, wb, false);
        buffer_json_member_add_array(wb, "required_params");
        {
            buffer_json_add_array_item_object(wb);
            {
                buffer_json_member_add_string(wb, "id", "source");
                buffer_json_member_add_string(wb, "name", "source");
                buffer_json_member_add_string(wb, "help", "Select the Logs Management source to query");
                buffer_json_member_add_string(wb, "type", "select");
                buffer_json_member_add_array(wb, "options");
                ret = fetch_log_sources(wb);
                buffer_json_array_close(wb); // options array
            }
            buffer_json_object_close(wb); // required params object
        }
        buffer_json_array_close(wb); // required_params array

        facets_table_config(wb);

        buffer_json_member_add_uint64(wb, "status", HTTP_RESP_OK);
        buffer_json_member_add_string(wb, "type", "table");
        buffer_json_member_add_string(wb, "help", FUNCTION_LOGSMANAGEMENT_HELP_SHORT);
        buffer_json_finalize(wb);
        goto output;
    }
    
    if(progress) {
        // TODO: Add progress function
        // function_logsmanagement_progress(wb, transaction, progress_id);
        goto cleanup;
    }

    if(!req_quota) 
        query_params.quota = LOGS_MANAG_QUERY_QUOTA_DEFAULT;
    else if(req_quota > LOGS_MANAG_QUERY_QUOTA_MAX) 
        query_params.quota = LOGS_MANAG_QUERY_QUOTA_MAX;
    else query_params.quota = req_quota;


    if(fqs->source)
        query_params.chartname[0] = (char *) string2str(fqs->source);

    query_params.order_by_asc = 0;

    
    // NOTE: Always perform descending timestamp query, req_from_ts >= req_to_ts.
    if(fqs->direction == FACETS_ANCHOR_DIRECTION_BACKWARD){
        query_params.req_from_ts = 
            (fqs->data_only && fqs->anchor.start_ut) ? fqs->anchor.start_ut / USEC_PER_MS : before_s * MSEC_PER_SEC;
        query_params.req_to_ts = 
            (fqs->data_only && fqs->anchor.stop_ut) ? fqs->anchor.stop_ut / USEC_PER_MS : after_s * MSEC_PER_SEC;
    }
    else{
        query_params.req_from_ts =
            (fqs->data_only && fqs->anchor.stop_ut) ? fqs->anchor.stop_ut / USEC_PER_MS : before_s * MSEC_PER_SEC;
        query_params.req_to_ts = 
            (fqs->data_only && fqs->anchor.start_ut) ? fqs->anchor.start_ut / USEC_PER_MS : after_s * MSEC_PER_SEC;
    }
    
    query_params.cancelled = cancelled;
    query_params.stop_monotonic_ut = now_monotonic_usec() + (timeout - 1) * USEC_PER_SEC;
    query_params.results_buff = buffer_create(query_params.quota, NULL);

    facets_rows_begin(facets);

    do{
        if(query_params.act_to_ts)
            query_params.req_from_ts = query_params.act_to_ts - 1000;

        ret = execute_logs_manag_query(&query_params);


        size_t res_off = 0;
        logs_query_res_hdr_t *p_res_hdr;
        while(query_params.results_buff->len - res_off > 0){
            p_res_hdr = (logs_query_res_hdr_t *) &query_params.results_buff->buffer[res_off];

            ssize_t remaining = p_res_hdr->text_size;
            char *ls = &query_params.results_buff->buffer[res_off] + sizeof(*p_res_hdr) + p_res_hdr->text_size - 1;
            *ls = '\0';
            int timestamp_off = p_res_hdr->matches;
            do{
                do{
                    --remaining;
                    --ls;
                } while(remaining > 0 && *ls != '\n');
                *ls = '\0';
                --remaining;
                --ls;

                usec_t timestamp = p_res_hdr->timestamp * USEC_PER_MS + --timestamp_off;

                if(unlikely(!fqs->last_modified)) {
                    if(timestamp == if_modified_since){
                        ret = &logs_qry_res_err[LOGS_QRY_RES_ERR_CODE_UNMODIFIED];
                        goto output;
                    }
                    else                    
                        fqs->last_modified = timestamp;
                }

                facets_add_key_value(facets, "log_source", p_res_hdr->log_source[0] ? p_res_hdr->log_source : "-");

                facets_add_key_value(facets, "log_type", p_res_hdr->log_type[0] ? p_res_hdr->log_type : "-");

                facets_add_key_value(facets, "filename", p_res_hdr->filename[0] ? p_res_hdr->filename : "-");

                facets_add_key_value(facets, "basename", p_res_hdr->basename[0] ? p_res_hdr->basename : "-");

                facets_add_key_value(facets, "chartname", p_res_hdr->chartname[0] ? p_res_hdr->chartname : "-");

                size_t ls_len = strlen(ls + 2);
                facets_add_key_value_length(facets, "message", sizeof("message") - 1, 
                                            ls + 2, ls_len <= FACET_MAX_VALUE_LENGTH ? ls_len : FACET_MAX_VALUE_LENGTH);

                facets_row_finished(facets, timestamp);

            } while(remaining > 0);

            res_off += sizeof(*p_res_hdr) + p_res_hdr->text_size;

        }

        buffer_flush(query_params.results_buff);

    } while(query_params.act_to_ts > query_params.req_to_ts);

    m_assert(query_params.req_from_ts == query_params.act_from_ts, "query_params.req_from_ts != query_params.act_from_ts");
    m_assert(query_params.req_to_ts   == query_params.act_to_ts  , "query_params.req_to_ts != query_params.act_to_ts");
    

    getrusage(RUSAGE_THREAD, &end);
    time_t user_time =  end.ru_utime.tv_sec * USEC_PER_SEC + end.ru_utime.tv_usec - 
                        start.ru_utime.tv_sec * USEC_PER_SEC - start.ru_utime.tv_usec;
    time_t sys_time =   end.ru_stime.tv_sec * USEC_PER_SEC + end.ru_stime.tv_usec - 
                        start.ru_stime.tv_sec * USEC_PER_SEC - start.ru_stime.tv_usec;

    buffer_json_member_add_object(wb, "logs_management_meta");
    buffer_json_member_add_string(wb, "api_version", LOGS_QRY_VERSION);
    buffer_json_member_add_uint64(wb, "num_lines", query_params.num_lines);
    buffer_json_member_add_uint64(wb, "user_time", user_time);
    buffer_json_member_add_uint64(wb, "system_time", sys_time);
    buffer_json_member_add_uint64(wb, "total_time", user_time + sys_time);
    buffer_json_member_add_uint64(wb, "error_code", (uint64_t) ret->err_code);
    buffer_json_member_add_string(wb, "error_string", ret->err_str);
    buffer_json_object_close(wb); // logs_management_meta

    buffer_json_member_add_uint64(wb, "status", ret->http_code);
    buffer_json_member_add_boolean(wb, "partial", ret->http_code != HTTP_RESP_OK || 
                                                  ret->err_code == LOGS_QRY_RES_ERR_CODE_TIMEOUT);
    buffer_json_member_add_string(wb, "type", "table");


    if(!fqs->data_only) {
        buffer_json_member_add_time_t(wb, "update_every", 1);
        buffer_json_member_add_string(wb, "help", FUNCTION_LOGSMANAGEMENT_HELP_SHORT);
    }

    if(!fqs->data_only || fqs->tail)
        buffer_json_member_add_uint64(wb, "last_modified", fqs->last_modified);

    facets_sort_and_reorder_keys(facets);
    facets_report(facets, wb, used_hashes_registry);

    buffer_json_member_add_time_t(wb, "expires", now_realtime_sec() + (fqs->data_only ? 3600 : 0));
    buffer_json_finalize(wb); // logs_management_meta


    // ------------------------------------------------------------------------
    // cleanup query params

    string_freez(fqs->source);
    fqs->source = NULL;

    // ------------------------------------------------------------------------
    // handle error response

output:
    netdata_mutex_lock(&stdout_mut);
    if(ret->http_code != HTTP_RESP_OK)
        pluginsd_function_json_error_to_stdout(transaction, ret->http_code, ret->err_str);
    else
        pluginsd_function_result_to_stdout(transaction, ret->http_code, "application/json", expires, wb);
    netdata_mutex_unlock(&stdout_mut);

cleanup:
    facets_destroy(facets);
    buffer_free(query_params.results_buff);
    buffer_free(wb);

    if(fqs_item) {
        dictionary_del(function_query_status_dict, dictionary_acquired_item_name(fqs_item));
        dictionary_acquired_item_release(function_query_status_dict, fqs_item);
        dictionary_garbage_collect(function_query_status_dict);
    }
}

struct functions_evloop_globals *logsmanagement_func_facets_init(bool *p_logsmanagement_should_exit){

    function_query_status_dict = dictionary_create_advanced(
            DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
            NULL, sizeof(FUNCTION_QUERY_STATUS));

    used_hashes_registry = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);

    netdata_mutex_lock(&stdout_mut);
    fprintf(stdout, PLUGINSD_KEYWORD_FUNCTION " GLOBAL \"%s\" %d \"%s\"\n",
                    LOGS_MANAG_FUNC_NAME, 
                    LOGS_MANAG_QUERY_TIMEOUT_DEFAULT, 
                    FUNCTION_LOGSMANAGEMENT_HELP_SHORT);
    netdata_mutex_unlock(&stdout_mut);

    struct functions_evloop_globals *wg = functions_evloop_init(1, "LGSMNGM", 
                                                                &stdout_mut, 
                                                                p_logsmanagement_should_exit);

    functions_evloop_add_function(  wg, LOGS_MANAG_FUNC_NAME, 
                                    logsmanagement_function_facets,
                                    LOGS_MANAG_QUERY_TIMEOUT_DEFAULT);
    
    return wg;
}