diff options
Diffstat (limited to 'logsmanagement/circular_buffer.c')
| -rw-r--r-- | logsmanagement/circular_buffer.c | 404 |
1 files changed, 404 insertions, 0 deletions
diff --git a/logsmanagement/circular_buffer.c b/logsmanagement/circular_buffer.c new file mode 100644 index 000000000..6459da5ea --- /dev/null +++ b/logsmanagement/circular_buffer.c @@ -0,0 +1,404 @@ +// SPDX-License-Identifier: GPL-3.0-or-later + +/** @file circular_buffer.c + * @brief This is the implementation of a circular buffer to be used + * for saving collected logs in memory, until they are stored + * into the database. + */ + +#include "circular_buffer.h" +#include "helper.h" +#include "parser.h" + +struct qsort_item { + Circ_buff_item_t *cbi; + struct File_info *pfi; +}; + +static int qsort_timestamp (const void *item_a, const void *item_b) { + return ( (int64_t)((struct qsort_item*)item_a)->cbi->timestamp - + (int64_t)((struct qsort_item*)item_b)->cbi->timestamp); +} + +static int reverse_qsort_timestamp (const void * item_a, const void * item_b) { + return -qsort_timestamp(item_a, item_b); +} + +/** + * @brief Search circular buffers according to the query_params. + * @details If multiple buffers are to be searched, the results will be sorted + * according to timestamps. + * + * Note that buff->tail can only be changed through circ_buff_read_done(), and + * circ_buff_search() and circ_buff_read_done() are mutually exclusive due + * to uv_mutex_lock() and uv_mutex_unlock() in queries and when writing to DB. + * + * @param p_query_params Query parameters to search according to. + * @param p_file_infos File_info structs to be searched. + */ +void circ_buff_search(logs_query_params_t *const p_query_params, struct File_info *const p_file_infos[]) { + + for(int pfi_off = 0; p_file_infos[pfi_off]; pfi_off++) + uv_rwlock_rdlock(&p_file_infos[pfi_off]->circ_buff->buff_realloc_rwlock); + + int buffs_size = 0, + buff_max_num_of_items = 0; + + while(p_file_infos[buffs_size]){ + if(p_file_infos[buffs_size]->circ_buff->num_of_items > buff_max_num_of_items) + buff_max_num_of_items = p_file_infos[buffs_size]->circ_buff->num_of_items; + buffs_size++; + } + + struct qsort_item items[buffs_size * buff_max_num_of_items + 1]; // worst case allocation + + int items_off = 0; + + for(int buff_off = 0; p_file_infos[buff_off]; buff_off++){ + Circ_buff_t *buff = p_file_infos[buff_off]->circ_buff; + /* TODO: The following 3 operations need to be replaced with a struct + * to gurantee atomicity. */ + int head = __atomic_load_n(&buff->head, __ATOMIC_SEQ_CST) % buff->num_of_items; + int tail = __atomic_load_n(&buff->tail, __ATOMIC_SEQ_CST) % buff->num_of_items; + int full = __atomic_load_n(&buff->full, __ATOMIC_SEQ_CST); + + if ((head == tail) && !full) continue; // Nothing to do if buff is empty + + for (int i = tail; i != head; i = (i + 1) % buff->num_of_items){ + items[items_off].cbi = &buff->items[i]; + items[items_off++].pfi = p_file_infos[buff_off]; + } + } + + items[items_off].cbi = NULL; + items[items_off].pfi = NULL; + + if(items[0].cbi) + qsort(items, items_off, sizeof(items[0]), p_query_params->order_by_asc ? qsort_timestamp : reverse_qsort_timestamp); + + + BUFFER *const res_buff = p_query_params->results_buff; + + logs_query_res_hdr_t res_hdr = { // result header + .timestamp = p_query_params->act_to_ts, + .text_size = 0, + .matches = 0, + .log_source = "", + .log_type = "" + }; + + for (int i = 0; items[i].cbi; i++) { + + /* If exceeding quota or timeout is reached and new timestamp is different than previous, + * terminate query but inform caller about act_to_ts to continue from (its next value) in next call. */ + if( (res_buff->len >= p_query_params->quota || terminate_logs_manag_query(p_query_params)) && + items[i].cbi->timestamp != res_hdr.timestamp){ + p_query_params->act_to_ts = res_hdr.timestamp; + break; + } + + res_hdr.timestamp = items[i].cbi->timestamp; + res_hdr.text_size = items[i].cbi->text_size; + strncpyz(res_hdr.log_source, log_src_t_str[items[i].pfi->log_source], sizeof(res_hdr.log_source) - 1); + strncpyz(res_hdr.log_type, log_src_type_t_str[items[i].pfi->log_type], sizeof(res_hdr.log_type) - 1); + strncpyz(res_hdr.basename, items[i].pfi->file_basename, sizeof(res_hdr.basename) - 1); + strncpyz(res_hdr.filename, items[i].pfi->filename, sizeof(res_hdr.filename) - 1); + strncpyz(res_hdr.chartname, items[i].pfi->chartname, sizeof(res_hdr.chartname) - 1); + + if (p_query_params->order_by_asc ? + ( res_hdr.timestamp >= p_query_params->req_from_ts && res_hdr.timestamp <= p_query_params->req_to_ts ) : + ( res_hdr.timestamp >= p_query_params->req_to_ts && res_hdr.timestamp <= p_query_params->req_from_ts) ){ + + /* In case of search_keyword, less than sizeof(res_hdr) + temp_msg.text_size + * space is required, but go for worst case scenario for now */ + buffer_increase(res_buff, sizeof(res_hdr) + res_hdr.text_size); + + if(!p_query_params->keyword || !*p_query_params->keyword || !strcmp(p_query_params->keyword, " ")){ + /* NOTE: relying on items[i]->cbi->num_lines to get number of log lines + * might not be 100% correct, since parsing must have taken place + * already to return correct count. Maybe an issue under heavy load. */ + res_hdr.matches = items[i].cbi->num_lines; + memcpy(&res_buff->buffer[res_buff->len + sizeof(res_hdr)], items[i].cbi->data, res_hdr.text_size); + } + else { + res_hdr.matches = search_keyword( items[i].cbi->data, res_hdr.text_size, + &res_buff->buffer[res_buff->len + sizeof(res_hdr)], + &res_hdr.text_size, p_query_params->keyword, NULL, + p_query_params->ignore_case); + + m_assert( (res_hdr.matches > 0 && res_hdr.text_size > 0) || + (res_hdr.matches == 0 && res_hdr.text_size == 0), + "res_hdr.matches and res_hdr.text_size must both be > 0 or == 0."); + + if(unlikely(res_hdr.matches < 0)) + break; /* res_hdr.matches < 0 - error during keyword search */ + } + + if(res_hdr.text_size){ + res_buff->buffer[res_buff->len + sizeof(res_hdr) + res_hdr.text_size - 1] = '\n'; // replace '\0' with '\n' + memcpy(&res_buff->buffer[res_buff->len], &res_hdr, sizeof(res_hdr)); + res_buff->len += sizeof(res_hdr) + res_hdr.text_size; + p_query_params->num_lines += res_hdr.matches; + } + + m_assert(TEST_MS_TIMESTAMP_VALID(res_hdr.timestamp), "res_hdr.timestamp is invalid"); + } + } + + for(int pfi_off = 0; p_file_infos[pfi_off]; pfi_off++) + uv_rwlock_rdunlock(&p_file_infos[pfi_off]->circ_buff->buff_realloc_rwlock); +} + +/** + * @brief Query circular buffer if there is space for item insertion. + * @param buff Circular buffer to query for available space. + * @param requested_text_space Size of raw (uncompressed) space needed. + * @note If buff->allow_dropped_logs is 0, then this function will block and + * it will only return once there is available space as requested. In this + * case, it will never return 0. + * @return \p requested_text_space if there is enough space, else 0. + */ +size_t circ_buff_prepare_write(Circ_buff_t *const buff, size_t const requested_text_space){ + + /* Calculate how much is the maximum compressed space that will + * be required on top of the requested space for the raw data. */ + buff->in->text_compressed_size = (size_t) LZ4_compressBound(requested_text_space); + m_assert(buff->in->text_compressed_size != 0, "requested text compressed space is zero"); + size_t const required_space = requested_text_space + buff->in->text_compressed_size; + + size_t available_text_space = 0; + size_t total_cached_mem_ex_in; + +try_to_acquire_space: + total_cached_mem_ex_in = 0; + for (int i = 0; i < buff->num_of_items; i++){ + total_cached_mem_ex_in += buff->items[i].data_max_size; + } + + /* If the required space is more than the allocated space of the input + * buffer, then we need to check if the input buffer can be reallocated: + * + * a) If the total memory consumption of the circular buffer plus the + * required space is less than the limit set by "circular buffer max size" + * for this log source, then the input buffer can be reallocated. + * + * b) If the total memory consumption of the circular buffer plus the + * required space is more than the limit set by "circular buffer max size" + * for this log source, we will attempt to reclaim some of the circular + * buffer allocated memory from any empty items. + * + * c) If after reclaiming the total memory consumption is still beyond the + * configuration limit, either 0 will be returned as the available space + * for raw logs in the input buffer, or the function will block and repeat + * the same process, until there is available space to be returned, depending + * of the configuration value of buff->allow_dropped_logs. + * */ + if(required_space > buff->in->data_max_size) { + if(likely(total_cached_mem_ex_in + required_space <= buff->total_cached_mem_max)){ + buff->in->data_max_size = required_space; + buff->in->data = reallocz(buff->in->data, buff->in->data_max_size); + + available_text_space = requested_text_space; + } + else if(likely(__atomic_load_n(&buff->full, __ATOMIC_SEQ_CST) == 0)){ + int head = __atomic_load_n(&buff->head, __ATOMIC_SEQ_CST) % buff->num_of_items; + int tail = __atomic_load_n(&buff->tail, __ATOMIC_SEQ_CST) % buff->num_of_items; + + for (int i = (head == tail ? (head + 1) % buff->num_of_items : head); + i != tail; i = (i + 1) % buff->num_of_items) { + + m_assert(i <= buff->num_of_items, "i > buff->num_of_items"); + buff->items[i].data_max_size = 1; + buff->items[i].data = reallocz(buff->items[i].data, buff->items[i].data_max_size); + } + + total_cached_mem_ex_in = 0; + for (int i = 0; i < buff->num_of_items; i++){ + total_cached_mem_ex_in += buff->items[i].data_max_size; + } + + if(total_cached_mem_ex_in + required_space <= buff->total_cached_mem_max){ + buff->in->data_max_size = required_space; + buff->in->data = reallocz(buff->in->data, buff->in->data_max_size); + + available_text_space = requested_text_space; + } + else available_text_space = 0; + } + } else available_text_space = requested_text_space; + + __atomic_store_n(&buff->total_cached_mem, total_cached_mem_ex_in + buff->in->data_max_size, __ATOMIC_RELAXED); + + if(unlikely(!buff->allow_dropped_logs && !available_text_space)){ + sleep_usec(CIRC_BUFF_PREP_WR_RETRY_AFTER_MS * USEC_PER_MS); + goto try_to_acquire_space; + } + + m_assert(available_text_space || buff->allow_dropped_logs, "!available_text_space == 0 && !buff->allow_dropped_logs"); + return available_text_space; +} + +/** + * @brief Insert item from temporary input buffer to circular buffer. + * @param buff Circular buffer to insert the item into + * @return 0 in case of success or -1 in case there was an error (e.g. buff + * is out of space). + */ +int circ_buff_insert(Circ_buff_t *const buff){ + + // TODO: Probably can be changed to __ATOMIC_RELAXED, but ideally a mutex should be used here. + int head = __atomic_load_n(&buff->head, __ATOMIC_SEQ_CST) % buff->num_of_items; + int tail = __atomic_load_n(&buff->tail, __ATOMIC_SEQ_CST) % buff->num_of_items; + int full = __atomic_load_n(&buff->full, __ATOMIC_SEQ_CST); + + /* If circular buffer does not have any free items, it will be expanded + * by reallocating the `items` array and adding one more item. */ + if (unlikely(( head == tail ) && full )) { + debug_log( "buff out of space! will be expanded."); + uv_rwlock_wrlock(&buff->buff_realloc_rwlock); + + + Circ_buff_item_t *items_new = callocz(buff->num_of_items + 1, sizeof(Circ_buff_item_t)); + + for(int i = 0; i < buff->num_of_items; i++){ + Circ_buff_item_t *item_old = &buff->items[head++ % buff->num_of_items]; + items_new[i] = *item_old; + } + freez(buff->items); + buff->items = items_new; + + buff->parse = buff->parse - buff->tail; + head = buff->head = buff->num_of_items++; + buff->tail = buff->read = 0; + buff->full = 0; + + __atomic_add_fetch(&buff->buff_realloc_cnt, 1, __ATOMIC_RELAXED); + + uv_rwlock_wrunlock(&buff->buff_realloc_rwlock); + } + + Circ_buff_item_t *cur_item = &buff->items[head]; + + char *tmp_data = cur_item->data; + size_t tmp_data_max_size = cur_item->data_max_size; + + cur_item->status = buff->in->status; + cur_item->timestamp = buff->in->timestamp; + cur_item->data = buff->in->data; + cur_item->text_size = buff->in->text_size; + cur_item->text_compressed = buff->in->text_compressed; + cur_item->text_compressed_size = buff->in->text_compressed_size; + cur_item->data_max_size = buff->in->data_max_size; + cur_item->num_lines = buff->in->num_lines; + + buff->in->status = CIRC_BUFF_ITEM_STATUS_UNPROCESSED; + buff->in->timestamp = 0; + buff->in->data = tmp_data; + buff->in->text_size = 0; + // buff->in->text_compressed = tmp_data; + buff->in->text_compressed_size = 0; + buff->in->data_max_size = tmp_data_max_size; + buff->in->num_lines = 0; + + __atomic_add_fetch(&buff->text_size_total, cur_item->text_size, __ATOMIC_SEQ_CST); + + if( __atomic_add_fetch(&buff->text_compressed_size_total, cur_item->text_compressed_size, __ATOMIC_SEQ_CST)){ + __atomic_store_n(&buff->compression_ratio, + __atomic_load_n(&buff->text_size_total, __ATOMIC_SEQ_CST) / + __atomic_load_n(&buff->text_compressed_size_total, __ATOMIC_SEQ_CST), + __ATOMIC_SEQ_CST); + } else __atomic_store_n( &buff->compression_ratio, 0, __ATOMIC_SEQ_CST); + + + if(unlikely(__atomic_add_fetch(&buff->head, 1, __ATOMIC_SEQ_CST) % buff->num_of_items == + __atomic_load_n(&buff->tail, __ATOMIC_SEQ_CST) % buff->num_of_items)){ + __atomic_store_n(&buff->full, 1, __ATOMIC_SEQ_CST); + } + + __atomic_or_fetch(&cur_item->status, CIRC_BUFF_ITEM_STATUS_PARSED | CIRC_BUFF_ITEM_STATUS_STREAMED, __ATOMIC_SEQ_CST); + + return 0; +} + +/** + * @brief Return pointer to next item to be read from the circular buffer. + * @param buff Circular buffer to get next item from. + * @return Pointer to the next circular buffer item to be read, or NULL + * if there are no more items to be read. + */ +Circ_buff_item_t *circ_buff_read_item(Circ_buff_t *const buff) { + + Circ_buff_item_t *item = &buff->items[buff->read % buff->num_of_items]; + + m_assert(__atomic_load_n(&item->status, __ATOMIC_RELAXED) <= CIRC_BUFF_ITEM_STATUS_DONE, "Invalid status"); + + if( /* No more records to be retrieved from the buffer - pay attention that + * there is no `% buff->num_of_items` operation, as we need to check + * the case where buff->read is exactly equal to buff->head. */ + (buff->read == (__atomic_load_n(&buff->head, __ATOMIC_SEQ_CST))) || + /* Current item either not parsed or streamed */ + (__atomic_load_n(&item->status, __ATOMIC_RELAXED) != CIRC_BUFF_ITEM_STATUS_DONE) ){ + + return NULL; + } + + __atomic_sub_fetch(&buff->text_size_total, item->text_size, __ATOMIC_SEQ_CST); + + if( __atomic_sub_fetch(&buff->text_compressed_size_total, item->text_compressed_size, __ATOMIC_SEQ_CST)){ + __atomic_store_n(&buff->compression_ratio, + __atomic_load_n(&buff->text_size_total, __ATOMIC_SEQ_CST) / + __atomic_load_n(&buff->text_compressed_size_total, __ATOMIC_SEQ_CST), + __ATOMIC_SEQ_CST); + } else __atomic_store_n( &buff->compression_ratio, 0, __ATOMIC_SEQ_CST); + + buff->read++; + + return item; +} + +/** + * @brief Complete buffer read process. + * @param buff Circular buffer to complete read process on. + */ +void circ_buff_read_done(Circ_buff_t *const buff){ + /* Even if one item was read, it means buffer cannot be full anymore */ + if(__atomic_load_n(&buff->tail, __ATOMIC_RELAXED) != buff->read) + __atomic_store_n(&buff->full, 0, __ATOMIC_SEQ_CST); + + __atomic_store_n(&buff->tail, buff->read, __ATOMIC_SEQ_CST); +} + +/** + * @brief Create a new circular buffer. + * @param num_of_items Number of Circ_buff_item_t items in the buffer. + * @param max_size Maximum memory the circular buffer can occupy. + * @param allow_dropped_logs Maximum memory the circular buffer can occupy. + * @return Pointer to the new circular buffer structure. + */ +Circ_buff_t *circ_buff_init(const int num_of_items, + const size_t max_size, + const int allow_dropped_logs ) { + Circ_buff_t *buff = callocz(1, sizeof(Circ_buff_t)); + buff->num_of_items = num_of_items; + buff->items = callocz(buff->num_of_items, sizeof(Circ_buff_item_t)); + buff->in = callocz(1, sizeof(Circ_buff_item_t)); + + uv_rwlock_init(&buff->buff_realloc_rwlock); + + buff->total_cached_mem_max = max_size; + buff->allow_dropped_logs = allow_dropped_logs; + + return buff; +} + +/** + * @brief Destroy a circular buffer. + * @param buff Circular buffer to be destroyed. + */ +void circ_buff_destroy(Circ_buff_t *buff){ + for (int i = 0; i < buff->num_of_items; i++) freez(buff->items[i].data); + freez(buff->items); + freez(buff->in->data); + freez(buff->in); + freez(buff); +}; |