diff --git a/plugins/in_tail/tail.c b/plugins/in_tail/tail.c index eb81a813d64..e7e8dbb716c 100644 --- a/plugins/in_tail/tail.c +++ b/plugins/in_tail/tail.c @@ -361,6 +361,10 @@ static int in_tail_watcher_callback(struct flb_input_instance *ins, flb_tail_file_rotated(file); } } + + /* Files that replaced a rotated file after the rotation was handled */ + flb_tail_file_discover_rotated_replacements(ctx); + return ret; } @@ -631,7 +635,8 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_TIME, "watcher_interval", "2s", 0, FLB_TRUE, offsetof(struct flb_tail_config, watcher_interval), - "interval to check symbolic link rotation and retry transient file read failures." + "interval to check symbolic link rotation, files that replaced a rotated " + "file and retry transient file read failures." }, { FLB_CONFIG_MAP_TIME, "progress_check_interval", "2s", diff --git a/plugins/in_tail/tail_file.c b/plugins/in_tail/tail_file.c index 44e667f315b..94323802014 100644 --- a/plugins/in_tail/tail_file.c +++ b/plugins/in_tail/tail_file.c @@ -2536,6 +2536,7 @@ int flb_tail_file_rotated(struct flb_tail_file *file) flb_tail_scan(ctx->path_list, ctx); } else { + file->rotate_replaced = FLB_TRUE; tail_signal_manager(file->config); } } @@ -2547,6 +2548,60 @@ int flb_tail_file_rotated(struct flb_tail_file *file) return 0; } +/* + * Register the files that replaced rotated files but did not exist yet when + * the rotation was handled. + * + * flb_tail_file_rotated() looks for the replacement right after the rename + * is detected. A container runtime creates the new file only after the + * rename (kubelet renames the log, then asks the runtime to reopen it), so + * that lookup can come too early. The new file is then found by the next + * path scan, which runs every 'refresh_interval' (60 seconds by default), + * and if it is rotated again before that scan it is never read at all. + * + * A rotated file is checked until its replacement is registered, either here + * or by flb_tail_file_rotated(), and skipped afterwards. + * + * Returns the number of files registered. + */ +int flb_tail_file_discover_rotated_replacements(struct flb_tail_config *ctx) +{ + int ret; + int count = 0; + struct stat st; + struct mk_list *head; + struct flb_tail_file *file; + + mk_list_foreach(head, &ctx->files_rotated) { + file = mk_list_entry(head, struct flb_tail_file, _rotate_head); + if (file->rotate_replaced == FLB_TRUE) { + continue; + } + + /* Check if a new file has been created with the original name */ + ret = tail_file_stat(ctx, file->orig_name, &st); + if (ret == 0 && st.st_ino != file->inode) { + if (flb_tail_file_exists(&st, ctx) == FLB_FALSE) { + flb_plg_debug(ctx->ins, "inode=%"PRIu64" replaced rotated " + "inode=%"PRIu64" as %s", (uint64_t) st.st_ino, + file->inode, file->orig_name); + ret = flb_tail_file_append(file->orig_name, &st, + FLB_TAIL_STATIC, -1, ctx); + if (ret == 0) { + file->rotate_replaced = FLB_TRUE; + count++; + } + } + } + } + + if (count > 0) { + tail_signal_manager(ctx); + } + + return count; +} + static int check_purge_deleted_file(struct flb_tail_config *ctx, struct flb_tail_file *file, time_t ts) { diff --git a/plugins/in_tail/tail_file.h b/plugins/in_tail/tail_file.h index 156fa6a85c5..0ee475dcca3 100644 --- a/plugins/in_tail/tail_file.h +++ b/plugins/in_tail/tail_file.h @@ -127,6 +127,7 @@ char *flb_tail_file_name(struct flb_tail_file *file); int flb_tail_file_is_rotated(struct flb_tail_config *ctx, struct flb_tail_file *file); int flb_tail_file_rotated(struct flb_tail_file *file); +int flb_tail_file_discover_rotated_replacements(struct flb_tail_config *ctx); int flb_tail_file_purge(struct flb_input_instance *ins, struct flb_config *config, void *context); int flb_tail_pack_line_map(struct flb_time *time, char **data, diff --git a/plugins/in_tail/tail_file_internal.h b/plugins/in_tail/tail_file_internal.h index 1a2c1dd2f03..026d533b3c2 100644 --- a/plugins/in_tail/tail_file_internal.h +++ b/plugins/in_tail/tail_file_internal.h @@ -54,6 +54,8 @@ struct flb_tail_file { size_t name_len; size_t orig_name_len; time_t rotated; + int rotate_replaced; /* the file that replaced this rotated file + has been registered */ int64_t pending_bytes; size_t stream_offset; /* this represents the logical data offset which for compressed files could be higher diff --git a/tests/runtime/in_tail.c b/tests/runtime/in_tail.c index 967596ac1b6..2f35fdccecc 100644 --- a/tests/runtime/in_tail.c +++ b/tests/runtime/in_tail.c @@ -3140,6 +3140,139 @@ void flb_test_db_compare_filename() #endif /* FLB_HAVE_SQLDB */ /* Test list */ +#ifndef FLB_SYSTEM_WINDOWS +static pthread_mutex_t replacement_mutex = PTHREAD_MUTEX_INITIALIZER; +static flb_sds_t replacement_output = NULL; + +static int cb_replacement_collect(void *record, size_t size, void *data) +{ + (void) data; + + pthread_mutex_lock(&replacement_mutex); + if (replacement_output == NULL) { + replacement_output = flb_sds_create_size(1024); + } + if (replacement_output != NULL) { + flb_sds_cat_safe(&replacement_output, (char *) record, size); + } + pthread_mutex_unlock(&replacement_mutex); + + flb_free(record); + return 0; +} + +static int replacement_output_has(const char *needle) +{ + int ret = FLB_FALSE; + + pthread_mutex_lock(&replacement_mutex); + if (replacement_output != NULL && strstr(replacement_output, needle)) { + ret = FLB_TRUE; + } + pthread_mutex_unlock(&replacement_mutex); + + return ret; +} + +static void replacement_output_reset() +{ + pthread_mutex_lock(&replacement_mutex); + if (replacement_output != NULL) { + flb_sds_destroy(replacement_output); + replacement_output = NULL; + } + pthread_mutex_unlock(&replacement_mutex); +} + +/* + * The file that replaces a rotated file is created after the rename was + * handled (kubelet renames the log, then the runtime reopens it). With a + * long refresh_interval the path scan does not find it in time; the watcher + * must register it instead. + */ +void flb_test_rotated_replacement_discovery() +{ + int i; + int ret; + int fd_new; + ssize_t w; + struct flb_lib_out_cb cb_data; + struct test_tail_ctx *ctx; + char *file[] = {"rotated_replacement.log"}; + char *rotated = "rotated_replacement.log.1"; + char *line_old = "OLD_LINE\n"; + char *line_new = "NEW_LINE\n"; + + replacement_output_reset(); + unlink(rotated); + + cb_data.cb = cb_replacement_collect; + cb_data.data = NULL; + + ctx = test_tail_ctx_create(&cb_data, &file[0], 1, FLB_TRUE); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + exit(EXIT_FAILURE); + } + + ret = flb_input_set(ctx->flb, ctx->i_ffd, + "path", file[0], + "read_from_head", "true", + "refresh_interval", "600", + "watcher_interval", "1", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_output_set(ctx->flb, ctx->o_ffd, + "format", "json", + NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + flb_time_msleep(1000); + + w = write(ctx->fds[0], line_old, strlen(line_old)); + TEST_CHECK(w == (ssize_t) strlen(line_old)); + fsync(ctx->fds[0]); + + /* + * Rename, let the rotation be handled, then create the replacement. The + * stat backend detects the rename only on its 2.5 second check, so wait + * longer than that to make sure flb_tail_file_rotated() runs first and + * finds no replacement yet. + */ + ret = rename(file[0], rotated); + TEST_CHECK(ret == 0); + flb_time_msleep(3000); + + fd_new = open(file[0], O_RDWR | O_CREAT | O_APPEND, + S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP); + TEST_CHECK(fd_new >= 0); + w = write(fd_new, line_new, strlen(line_new)); + TEST_CHECK(w == (ssize_t) strlen(line_new)); + fsync(fd_new); + + /* well below refresh_interval, above watcher_interval */ + for (i = 0; i < 50; i++) { + if (replacement_output_has("\"log\":\"NEW_LINE\"") == FLB_TRUE) { + break; + } + flb_time_msleep(100); + } + + TEST_CHECK(replacement_output_has("\"log\":\"OLD_LINE\"") == FLB_TRUE); + if (!TEST_CHECK(replacement_output_has("\"log\":\"NEW_LINE\"") == FLB_TRUE)) { + TEST_MSG("replacement file was not registered by the watcher"); + } + + close(fd_new); + unlink(rotated); + test_tail_ctx_destroy(ctx); + replacement_output_reset(); +} +#endif /* !FLB_SYSTEM_WINDOWS */ + TEST_LIST = { {"issue_3943", flb_test_in_tail_issue_3943}, /* Properties */ @@ -3162,6 +3295,9 @@ TEST_LIST = { {"ignore_active_older_files", flb_test_in_tail_ignore_active_older_files}, {"ignore_active_older_files_reread_on_update", flb_test_in_tail_ignore_active_older_files_reread_on_update}, {"ignore_active_older_files_reread_on_update_default_read_from_head", flb_test_in_tail_ignore_active_older_files_reread_on_update_default_read_from_head}, +#ifndef FLB_SYSTEM_WINDOWS + {"rotated_replacement_discovery", flb_test_rotated_replacement_discovery}, +#endif #ifdef FLB_HAVE_INOTIFY {"inotify_watcher_false", flb_test_inotify_watcher_false}, {"inotify_pause_collectors", flb_test_inotify_pause_collectors},