// 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; }