diff options
author | Costa Tsaousis <costa@netdata.cloud> | 2023-11-26 02:22:47 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-11-26 02:22:47 +0200 |
commit | a59097ca507f01782ad4b401236770acf1175005 (patch) | |
tree | 0b7bca79c9daa1594344c6ad64026f8aa2c71195 /collectors | |
parent | e08ce9a587f48e39b3404b6aee110ed89f7d0537 (diff) |
journals management improvements (#16475)
* different approach to distribute estimatons evenly
* log estimations calculation to spot the issue
* better log
* more logs
* more logs
* more logs
* more logs
* commented logs
* process journal files and directories in reverse order, so that the system will immediately be available after start
* put the right filename in the dictionary
* fix scan dir recursion
* add missing fields when sending logs to systemd-journal-remote
* the standard datetime format is now switched to rfc3339
Diffstat (limited to 'collectors')
6 files changed, 164 insertions, 95 deletions
diff --git a/collectors/systemd-journal.plugin/systemd-internals.h b/collectors/systemd-journal.plugin/systemd-internals.h index 5ae48ed113..e1ae44d4f1 100644 --- a/collectors/systemd-journal.plugin/systemd-internals.h +++ b/collectors/systemd-journal.plugin/systemd-internals.h @@ -13,14 +13,14 @@ #define SYSTEMD_JOURNAL_FUNCTION_DESCRIPTION "View, search and analyze systemd journal entries." #define SYSTEMD_JOURNAL_FUNCTION_NAME "systemd-journal" #define SYSTEMD_JOURNAL_DEFAULT_TIMEOUT 60 +#define SYSTEMD_JOURNAL_ENABLE_ESTIMATIONS_FILE_PERCENTAGE 0.01 +#define SYSTEMD_JOURNAL_EXECUTE_WATCHER_PENDING_EVERY_MS 250 +#define SYSTEMD_JOURNAL_ALL_FILES_SCAN_EVERY_USEC (5 * 60 * USEC_PER_SEC) #define SYSTEMD_UNITS_FUNCTION_DESCRIPTION "View the status of systemd units" #define SYSTEMD_UNITS_FUNCTION_NAME "systemd-list-units" #define SYSTEMD_UNITS_DEFAULT_TIMEOUT 30 -#define EXECUTE_WATCHER_PENDING_EVERY_MS 500 -#define FULL_JOURNAL_SCAN_EVERY_USEC (5 * 60 * USEC_PER_SEC) - extern __thread size_t fstat_thread_calls; extern __thread size_t fstat_thread_cached_responses; void fstat_cache_enable_on_thread(void); @@ -28,7 +28,6 @@ void fstat_cache_disable_on_thread(void); extern netdata_mutex_t stdout_mutex; - typedef enum { ND_SD_JOURNAL_NO_FILE_MATCHED, ND_SD_JOURNAL_FAILED_TO_OPEN, @@ -97,7 +96,8 @@ int journal_file_dict_items_forward_compar(const void *a, const void *b); void buffer_json_journal_versions(BUFFER *wb); void available_journal_file_sources_to_json_array(BUFFER *wb); bool journal_files_completed_once(void); -void journal_files_updater_all_headers_sorted(void); +void journal_files_registry_update(void); +void journal_directory_scan_recursively(DICTIONARY *files, DICTIONARY *dirs, const char *dirname, int depth); FACET_ROW_SEVERITY syslog_priority_to_facet_severity(FACETS *facets, FACET_ROW *row, void *data); @@ -116,14 +116,12 @@ usec_t journal_file_update_annotation_boot_id(sd_journal *j, struct journal_file #define MAX_JOURNAL_DIRECTORIES 100 struct journal_directory { char *path; - bool logged_failure; }; extern struct journal_directory journal_directories[MAX_JOURNAL_DIRECTORIES]; void journal_init_files_and_directories(void); void journal_init_query_status(void); void function_systemd_journal(const char *transaction, char *function, int timeout, bool *cancelled); -void journal_files_registry_update(void); void journal_file_update_header(const char *filename, struct journal_file *jf); void netdata_systemd_journal_message_ids_init(void); diff --git a/collectors/systemd-journal.plugin/systemd-journal-annotations.c b/collectors/systemd-journal.plugin/systemd-journal-annotations.c index 58480c86c1..b12356110c 100644 --- a/collectors/systemd-journal.plugin/systemd-journal-annotations.c +++ b/collectors/systemd-journal.plugin/systemd-journal-annotations.c @@ -427,8 +427,8 @@ void netdata_systemd_journal_transform_boot_id(FACETS *facets __maybe_unused, BU ut = *p_ut; if(ut && ut != UINT64_MAX) { - char buffer[ISO8601_MAX_LENGTH]; - iso8601_datetime_ut(buffer, sizeof(buffer), ut, ISO8601_UTC); + char buffer[RFC3339_MAX_LENGTH]; + rfc3339_datetime_ut(buffer, sizeof(buffer), ut, 0, true); switch(scope) { default: @@ -506,8 +506,8 @@ void netdata_systemd_journal_transform_timestamp_usec(FACETS *facets __maybe_unu if(*v && isdigit(*v)) { uint64_t ut = str2ull(buffer_tostring(wb), NULL); if(ut) { - char buffer[ISO8601_MAX_LENGTH]; - iso8601_datetime_ut(buffer, sizeof(buffer), ut, ISO8601_UTC | ISO8601_MICROSECONDS); + char buffer[RFC3339_MAX_LENGTH]; + rfc3339_datetime_ut(buffer, sizeof(buffer), ut, 6, true); buffer_sprintf(wb, " (%s)", buffer); } } diff --git a/collectors/systemd-journal.plugin/systemd-journal-files.c b/collectors/systemd-journal.plugin/systemd-journal-files.c index 71dd3082cb..56496df225 100644 --- a/collectors/systemd-journal.plugin/systemd-journal-files.c +++ b/collectors/systemd-journal.plugin/systemd-journal-files.c @@ -602,17 +602,16 @@ static void files_registry_delete_cb(const DICTIONARY_ITEM *item, void *value, v string_freez(jf->source); } -void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_monotonic_ut) { +void journal_directory_scan_recursively(DICTIONARY *files, DICTIONARY *dirs, const char *dirname, int depth) { static const char *ext = ".journal"; - static const size_t ext_len = sizeof(".journal") - 1; + static const ssize_t ext_len = sizeof(".journal") - 1; if (depth > VAR_LOG_JOURNAL_MAX_DEPTH) return; DIR *dir; struct dirent *entry; - struct stat info; - char absolute_path[FILENAME_MAX]; + char full_path[FILENAME_MAX]; // Open the directory. if ((dir = opendir(dirname)) == NULL) { @@ -621,33 +620,43 @@ void journal_directory_scan(const char *dirname, int depth, usec_t last_scan_mon return; } + bool existing = false; + bool *found = dictionary_set(dirs, dirname, &existing, sizeof(existing)); + if(*found) return; + *found = true; + // Read each entry in the directory. while ((entry = readdir(dir)) != NULL) { - snprintfz(absolute_path, sizeof(absolute_path), "%s/%s", dirname, entry->d_name); - if (stat(absolute_path, &info) != 0) { - netdata_log_error("Failed to stat() '%s", absolute_path); + if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0) continue; - } - if (S_ISDIR(info.st_mode)) { - // If entry is a directory, call traverse recursively. - if (strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0) - journal_directory_scan(absolute_path, depth + 1, last_scan_monotonic_ut); + ssize_t len = snprintfz(full_path, sizeof(full_path), "%s/%s", dirname, entry->d_name); + + if (entry->d_type == DT_DIR) { + journal_directory_scan_recursively(files, dirs, full_path, depth++); + } + else if (entry->d_type == DT_REG && len > ext_len && strcmp(full_path + len - ext_len, ext) == 0) { + if(files) + dictionary_set(files, full_path, NULL, 0); + send_newline_and_flush(); } - else if (S_ISREG(info.st_mode)) { - // If entry is a regular file, check if it ends with .journal. - char *filename = entry->d_name; - size_t len = strlen(filename); - - if (len > ext_len && strcmp(filename + len - ext_len, ext) == 0) { - struct journal_file t = { - .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC + info.st_mtim.tv_nsec / NSEC_PER_USEC, - .last_scan_monotonic_ut = last_scan_monotonic_ut, - .size = info.st_size, - .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT, - }; - dictionary_set(journal_files_registry, absolute_path, &t, sizeof(t)); + else if (entry->d_type == DT_LNK) { + struct stat info; + if (stat(full_path, &info) == -1) + continue; + + if (S_ISDIR(info.st_mode)) { + // The symbolic link points to a directory + char resolved_path[FILENAME_MAX + 1]; + if (realpath(full_path, resolved_path) != NULL) { + journal_directory_scan_recursively(files, dirs, resolved_path, depth++); + } + } + else if(S_ISREG(info.st_mode) && len > ext_len && strcmp(full_path + len - ext_len, ext) == 0) { + if(files) + dictionary_set(files, full_path, NULL, 0); + send_newline_and_flush(); } } @@ -661,41 +670,38 @@ bool journal_files_completed_once(void) { return journal_files_scans > 0; } -static int journal_file_dict_items_last_modified_compar(const void *a, const void *b) { - const DICTIONARY_ITEM **ad = (const DICTIONARY_ITEM **)a, **bd = (const DICTIONARY_ITEM **)b; - struct journal_file *jfa = dictionary_acquired_item_value(*ad); - struct journal_file *jfb = dictionary_acquired_item_value(*bd); +int filenames_compar(const void *a, const void *b) { + const char *p1 = *(const char **)a; + const char *p2 = *(const char **)b; - if(jfa->file_last_modified_ut > jfb->file_last_modified_ut) + const char *at1 = strchr(p1, '@'); + const char *at2 = strchr(p2, '@'); + + if(!at1 && at2) return -1; - else if(jfa->file_last_modified_ut < jfb->file_last_modified_ut) + + if(at1 && !at2) return 1; - return 0; -} + if(!at1 && !at2) + return strcmp(p1, p2); -void journal_files_updater_all_headers_sorted(void) { - const DICTIONARY_ITEM *file_items[dictionary_entries(journal_files_registry)]; - size_t files_used = 0; + const char *dash1 = strrchr(at1, '-'); + const char *dash2 = strrchr(at2, '-'); - struct journal_file *jf; - dfe_start_write(journal_files_registry, jf){ - if(jf->last_scan_header_vs_last_modified_ut < jf->file_last_modified_ut) - file_items[files_used++] = dictionary_acquired_item_dup(journal_files_registry, jf_dfe.item); - } - dfe_done(jf); + if(!dash1 || !dash2) + return strcmp(p1, p2); - // sort them in reverse order (newer first) - qsort(file_items, files_used, sizeof(const DICTIONARY_ITEM *), - journal_file_dict_items_last_modified_compar); + uint64_t ts1 = strtoul(dash1 + 1, NULL, 16); + uint64_t ts2 = strtoul(dash2 + 1, NULL, 16); - // update the header (first and last message ut, sequence numbers, etc) - for(size_t i = 0; i < files_used ; i++) { - jf = dictionary_acquired_item_value(file_items[i]); - journal_file_update_header(jf->filename, jf); - dictionary_acquired_item_release(journal_files_registry, file_items[i]); - send_newline_and_flush(); - } + if(ts1 > ts2) + return -1; + + if(ts1 < ts2) + return 1; + + return -strcmp(p1, p2); } void journal_files_registry_update(void) { @@ -704,12 +710,45 @@ void journal_files_registry_update(void) { if(spinlock_trylock(&spinlock)) { usec_t scan_monotonic_ut = now_monotonic_usec(); + DICTIONARY *files = dictionary_create(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_DONT_OVERWRITE_VALUE); + DICTIONARY *dirs = dictionary_create(DICT_OPTION_SINGLE_THREADED|DICT_OPTION_DONT_OVERWRITE_VALUE); + for(unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES; i++) { - if(!journal_directories[i].path) - break; + if(!journal_directories[i].path) break; + journal_directory_scan_recursively(files, dirs, journal_directories[i].path, 0); + } + + const char **array = mallocz(sizeof(const char *) * dictionary_entries(files)); + size_t used = 0; + + void *x; + dfe_start_read(files, x) { + if(used >= dictionary_entries(files)) continue; + array[used++] = x_dfe.name; + } + dfe_done(x); - journal_directory_scan(journal_directories[i].path, 0, scan_monotonic_ut); + qsort(array, used, sizeof(const char *), filenames_compar); + + for(size_t i = 0; i < used ;i++) { + const char *full_path = array[i]; + + struct stat info; + if (stat(full_path, &info) == -1) + continue; + + struct journal_file t = { + .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC + info.st_mtim.tv_nsec / NSEC_PER_USEC, + .last_scan_monotonic_ut = scan_monotonic_ut, + .size = info.st_size, + .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT, + }; + struct journal_file *jf = dictionary_set(journal_files_registry, full_path, &t, sizeof(t)); + journal_file_update_header(jf->filename, jf); } + freez(array); + dictionary_destroy(files); + dictionary_destroy(dirs); struct journal_file *jf; dfe_start_write(journal_files_registry, jf){ @@ -718,8 +757,6 @@ void journal_files_registry_update(void) { } dfe_done(jf); - journal_files_updater_all_headers_sorted(); - journal_files_scans++; spinlock_unlock(&spinlock); diff --git a/collectors/systemd-journal.plugin/systemd-journal-watcher.c b/collectors/systemd-journal.plugin/systemd-journal-watcher.c index a2987cb219..ed41f62474 100644 --- a/collectors/systemd-journal.plugin/systemd-journal-watcher.c +++ b/collectors/systemd-journal.plugin/systemd-journal-watcher.c @@ -136,29 +136,31 @@ static char* get_path_from_wd(Watcher *watcher, int wd) { return NULL; } -static void watch_directory_recursively(Watcher *watcher, int inotifyFd, const char *basePath) { - // First, add a watch for the top-level directory itself - add_watch(watcher, inotifyFd, basePath); - - char path[PATH_MAX]; - struct dirent *dp; - DIR *dir = opendir(basePath); +static bool is_directory_watched(Watcher *watcher, const char *path) { + for (int i = 0; i < watcher->watchCount; ++i) { + if (watcher->watchList[i].wd != -1 && strcmp(watcher->watchList[i].path, path) == 0) { + return true; + } + } + return false; +} - if (!dir) - return; +static void watch_directory_and_subdirectories(Watcher *watcher, int inotifyFd, const char *basePath) { + DICTIONARY *dirs = dictionary_create(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_DONT_OVERWRITE_VALUE); - while ((dp = readdir(dir)) != NULL) { - if (strcmp(dp->d_name, ".") != 0 && strcmp(dp->d_name, "..") != 0) { - snprintfz(path, sizeof(path), "%s/%s", basePath, dp->d_name); + journal_directory_scan_recursively(NULL, dirs, basePath, 0); - if (dp->d_type == DT_DIR) { - // Recursively watch this directory - watch_directory_recursively(watcher, inotifyFd, path); - } + void *x; + dfe_start_read(dirs, x) { + const char *dirname = x_dfe.name; + // Check if this directory is already being watched + if (!is_directory_watched(watcher, dirname)) { + add_watch(watcher, inotifyFd, dirname); } } + dfe_done(x); - closedir(dir); + dictionary_destroy(dirs); } static bool is_subpath(const char *path, const char *subpath) { @@ -236,7 +238,7 @@ void process_event(Watcher *watcher, int inotifyFd, struct inotify_event *event) fullPath); // Start watching the new directory - recursive watch - watch_directory_recursively(watcher, inotifyFd, fullPath); + watch_directory_and_subdirectories(watcher, inotifyFd, fullPath); } else nd_log(NDLS_COLLECTORS, NDLP_WARNING, @@ -310,14 +312,14 @@ void *journal_watcher_main(void *arg __maybe_unused) { for (unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES; i++) { if (!journal_directories[i].path) break; - watch_directory_recursively(&watcher, inotifyFd, journal_directories[i].path); + watch_directory_and_subdirectories(&watcher, inotifyFd, journal_directories[i].path); } usec_t last_headers_update_ut = now_monotonic_usec(); struct buffered_reader reader; while (1) { buffered_reader_ret_t rc = buffered_reader_read_timeout( - &reader, inotifyFd, EXECUTE_WATCHER_PENDING_EVERY_MS, false); + &reader, inotifyFd, SYSTEMD_JOURNAL_EXECUTE_WATCHER_PENDING_EVERY_MS, false); if (rc != BUFFERED_READER_READ_OK && rc != BUFFERED_READER_READ_POLL_TIMEOUT) { nd_log(NDLS_COLLECTORS, NDLP_CRIT, @@ -353,7 +355,7 @@ void *journal_watcher_main(void *arg __maybe_unused) { usec_t ut = now_monotonic_usec(); if (dictionary_entries(watcher.pending) && (rc == BUFFERED_READER_READ_POLL_TIMEOUT || - last_headers_update_ut + (EXECUTE_WATCHER_PENDING_EVERY_MS * USEC_PER_MS) <= ut)) { + last_headers_update_ut + (SYSTEMD_JOURNAL_EXECUTE_WATCHER_PENDING_EVERY_MS * USEC_PER_MS) <= ut)) { process_pending(&watcher); last_headers_update_ut = ut; } diff --git a/collectors/systemd-journal.plugin/systemd-journal.c b/collectors/systemd-journal.plugin/systemd-journal.c index 5ca72d9f73..458f2492e0 100644 --- a/collectors/systemd-journal.plugin/systemd-journal.c +++ b/collectors/systemd-journal.plugin/systemd-journal.c @@ -474,9 +474,9 @@ static size_t sampling_running_file_query_estimate_remaining_lines_by_time(FUNCT size_t scanned_lines = sampling_file_lines_scanned_so_far(fqs); // Calculate the proportion of time covered - usec_t total_time_ut; + usec_t total_time_ut, remaining_start_ut, remaining_end_ut; usec_t remaining_time_ut = sampling_running_file_query_remaining_time(fqs, jf, direction, msg_ut, &total_time_ut, - NULL, NULL); + &remaining_start_ut, &remaining_end_ut); if (total_time_ut == 0) total_time_ut = 1; double proportion_by_time = (double) (total_time_ut - remaining_time_ut) / (double) total_time_ut; @@ -494,10 +494,30 @@ static size_t sampling_running_file_query_estimate_remaining_lines_by_time(FUNCT size_t remaining_logs_by_time = expected_matching_logs_by_time - scanned_lines; if (remaining_logs_by_time < 1) remaining_logs_by_time = 1; +// nd_log(NDLS_COLLECTORS, NDLP_INFO, +// "JOURNAL ESTIMATION: '%s' " +// "scanned_lines=%zu [sampled=%zu, unsampled=%zu, estimated=%zu], " +// "file [%"PRIu64" - %"PRIu64", duration %"PRId64", known lines in file %zu], " +// "query [%"PRIu64" - %"PRIu64", duration %"PRId64"], " +// "first message read from the file at %"PRIu64", current message at %"PRIu64", " +// "proportion of time %.2f %%, " +// "expected total lines in file %zu, " +// "remaining lines %zu, " +// "remaining time %"PRIu64" [%"PRIu64" - %"PRIu64", duration %"PRId64"]" +// , jf->filename +// , scanned_lines, fqs->samples_per_file.sampled, fqs->samples_per_file.unsampled, fqs->samples_per_file.estimated +// , jf->msg_first_ut, jf->msg_last_ut, jf->msg_last_ut - jf->msg_first_ut, jf->messages_in_file +// , fqs->query_file.start_ut, fqs->query_file.stop_ut, fqs->query_file.stop_ut - fqs->query_file.start_ut +// , fqs->query_file.first_msg_ut, msg_ut +// , proportion_by_time * 100.0 +// , expected_matching_logs_by_time +// , remaining_logs_by_time +// , remaining_time_ut, remaining_start_ut, remaining_end_ut, remaining_end_ut - remaining_start_ut +// ); + return remaining_logs_by_time; } - static size_t sampling_running_file_query_estimate_remaining_lines(sd_journal *j, FUNCTION_QUERY_STATUS *fqs, struct journal_file *jf, FACETS_ANCHOR_DIRECTION direction, usec_t msg_ut) { size_t expected_matching_logs_by_seqnum = 0; double proportion_by_seqnum = 0.0; @@ -613,7 +633,7 @@ static inline sampling_t is_row_in_sample(sd_journal *j, FUNCTION_QUERY_STATUS * if(fqs->samples_per_file.unsampled > fqs->samples_per_file.sampled) { double progress_by_time = sampling_running_file_query_progress_by_time(fqs, jf, direction, msg_ut); - if(progress_by_time > 0.05) + if(progress_by_time > SYSTEMD_JOURNAL_ENABLE_ESTIMATIONS_FILE_PERCENTAGE) return SAMPLING_STOP_AND_ESTIMATE; } @@ -1166,6 +1186,18 @@ static int netdata_systemd_journal_query(BUFFER *wb, FACETS *facets, FUNCTION_QU ND_SD_JOURNAL_STATUS tmp_status = netdata_systemd_journal_query_one_file(filename, wb, facets, jf, fqs); +// nd_log(NDLS_COLLECTORS, NDLP_INFO, +// "JOURNAL ESTIMATION FINAL: '%s' " +// "total lines %zu [sampled=%zu, unsampled=%zu, estimated=%zu], " +// "file [%"PRIu64" - %"PRIu64", duration %"PRId64", known lines in file %zu], " +// "query [%"PRIu64" - %"PRIu64", duration %"PRId64"], " +// , jf->filename +// , fqs->samples_per_file.sampled + fqs->samples_per_file.unsampled + fqs->samples_per_file.estimated +// , fqs->samples_per_file.sampled, fqs->samples_per_file.unsampled, fqs->samples_per_file.estimated +// , jf->msg_first_ut, jf->msg_last_ut, jf->msg_last_ut - jf->msg_first_ut, jf->messages_in_file +// , fqs->query_file.start_ut, fqs->query_file.stop_ut, fqs->query_file.stop_ut - fqs->query_file.start_ut +// ); + rows_useful = fqs->rows_useful - rows_useful; rows_read = fqs->rows_read - rows_read; bytes_read = fqs->bytes_read - bytes_read; diff --git a/collectors/systemd-journal.plugin/systemd-main.c b/collectors/systemd-journal.plugin/systemd-main.c index 74ee86dc28..d335fd82be 100644 --- a/collectors/systemd-journal.plugin/systemd-main.c +++ b/collectors/systemd-journal.plugin/systemd-main.c @@ -86,14 +86,14 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) { usec_t step_ut = 100 * USEC_PER_MS; usec_t send_newline_ut = 0; - usec_t since_last_scan_ut = FULL_JOURNAL_SCAN_EVERY_USEC * 2; // something big to trigger scanning at start + usec_t since_last_scan_ut = SYSTEMD_JOURNAL_ALL_FILES_SCAN_EVERY_USEC * 2; // something big to trigger scanning at start bool tty = isatty(fileno(stderr)) == 1; heartbeat_t hb; heartbeat_init(&hb); while(!plugin_should_exit) { - if(since_last_scan_ut > FULL_JOURNAL_SCAN_EVERY_USEC) { + if(since_last_scan_ut > SYSTEMD_JOURNAL_ALL_FILES_SCAN_EVERY_USEC) { journal_files_registry_update(); since_last_scan_ut = 0; } |