in_tail: make read_from_head after the initial discovery optional

Additionally, when a file is ignored due to ignore_older its size is
stored to ensure that if the file is modified we are able to ingest the
data that triggered the timestamp update.

Signed-off-by: Leonardo Alminana <[email protected]>

Author: leonardo-albertovich

plugins/in_tail/tail.c

@@ -396,13 +396,15 @@ static int in_tail_init(struct flb_input_instance *in,
     }
 #endif
 
-    /*
-     * After the first scan (on start time), all new files discovered needs to be
-     * read from head, so we switch the 'read_from_head' flag to true so any
-     * other file discovered after a scan or a rotation are read from the
-     * beginning.
-     */
-    ctx->read_from_head = FLB_TRUE;
+    if (ctx->read_newly_discovered_files_from_head) {
+        /*
+        * After the first scan (on start time), all new files discovered needs to be
+        * read from head, so we switch the 'read_from_head' flag to true so any
+        * other file discovered after a scan or a rotation are read from the
+        * beginning.
+        */
+        ctx->read_from_head = FLB_TRUE;
+    }
 
     /* Set plugin context */
     flb_input_set_context(in, ctx);
@@ -594,6 +596,12 @@ static struct flb_config_map config_map[] = {
      "For new discovered files on start (without a database offset/position), read the "
      "content from the head of the file, not tail."
     },
+    {
+     FLB_CONFIG_MAP_BOOL, "read_newly_discovered_files_from_head", "true",
+     0, FLB_TRUE, offsetof(struct flb_tail_config, read_newly_discovered_files_from_head),
+     "For new discovered files after start (without a database offset/position), read the "
+     "content from the head of the file, not tail."
+    },
     {
      FLB_CONFIG_MAP_STR, "refresh_interval", "60",
      0, FLB_FALSE, 0,

plugins/in_tail/tail_config.c

@@ -279,6 +279,14 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
         return NULL;
     }
 
+    /* hash table for files lookups */
+    ctx->ignored_file_sizes = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 1000, 0);
+    if (ctx->ignored_file_sizes == NULL) {
+        flb_plg_error(ctx->ins, "could not create ignored file size hash table");
+        flb_tail_config_destroy(ctx);
+        return NULL;
+    }
+
 #ifdef FLB_HAVE_SQLDB
     ctx->db = NULL;
 #endif
@@ -507,10 +515,16 @@ int flb_tail_config_destroy(struct flb_tail_config *config)
     if (config->static_hash) {
         flb_hash_table_destroy(config->static_hash);
     }
+
     if (config->event_hash) {
         flb_hash_table_destroy(config->event_hash);
     }
 
+    if (config->ignored_file_sizes != NULL) {
+        flb_hash_table_destroy(config->ignored_file_sizes);
+    }
+
     flb_free(config);
+
     return 0;
 }

plugins/in_tail/tail_config.h

@@ -82,6 +82,7 @@ struct flb_tail_config {
 #endif
     int refresh_interval_sec;  /* seconds to re-scan           */
     long refresh_interval_nsec;/* nanoseconds to re-scan       */
+    int read_newly_discovered_files_from_head; /* read new files from head after startup */
     int read_from_head;        /* read new files from head     */
     int rotate_wait;           /* sec to wait on rotated files */
     int watcher_interval;      /* watcher interval             */
@@ -169,6 +170,8 @@ struct flb_tail_config {
     struct flb_hash_table *static_hash;
     struct flb_hash_table *event_hash;
 
+    struct flb_hash_table *ignored_file_sizes;
+
     struct flb_config *config;
 };
 

plugins/in_tail/tail_file.c

@@ -911,13 +911,24 @@ static int set_file_position(struct flb_tail_config *ctx,
         return 0;
     }
 
-    /* tail... */
-    ret = lseek(file->fd, 0, SEEK_END);
-    if (ret == -1) {
-        flb_errno();
-        return -1;
+    if (file->offset > 0) {
+        ret = lseek(file->fd, file->offset, SEEK_SET);
+
+        if (ret == -1) {
+            flb_errno();
+            return -1;
+        }
+    }
+    else {
+        ret = lseek(file->fd, 0, SEEK_END);
+
+        if (ret == -1) {
+            flb_errno();
+            return -1;
+        }
+
+        file->offset = ret;
     }
-    file->offset = ret;
 
     if (file->decompression_context == NULL) {
         file->stream_offset = ret;
@@ -966,6 +977,7 @@ static int ml_flush_callback(struct flb_ml_parser *parser,
 }
 
 int flb_tail_file_append(char *path, struct stat *st, int mode,
+                         ssize_t offset,
                          struct flb_tail_config *ctx)
 {
     int fd;
@@ -1055,6 +1067,10 @@ int flb_tail_file_append(char *path, struct stat *st, int mode,
     file->mult_flush_timeout = 0;
     file->mult_skipping = FLB_FALSE;
 
+    if (offset != -1) {
+        file->offset = offset;
+    }
+
     if (strlen(path) >= 3 &&
         strcasecmp(&path[strlen(path) - 3], ".gz") == 0) {
         file->decompression_context =
@@ -1937,7 +1953,7 @@ int flb_tail_file_rotated(struct flb_tail_file *file)
         ret = stat(tmp, &st);
         if (ret == 0 && st.st_ino != file->inode) {
             if (flb_tail_file_exists(&st, ctx) == FLB_FALSE) {
-                ret = flb_tail_file_append(tmp, &st, FLB_TAIL_STATIC, ctx);
+                ret = flb_tail_file_append(tmp, &st, FLB_TAIL_STATIC, -1, ctx);
                 if (ret == -1) {
                     flb_tail_scan(ctx->path_list, ctx);
                 }

plugins/in_tail/tail_file.h

@@ -119,6 +119,7 @@ int flb_tail_file_name_dup(char *path, struct flb_tail_file *file);
 int flb_tail_file_to_event(struct flb_tail_file *file);
 int flb_tail_file_chunk(struct flb_tail_file *file);
 int flb_tail_file_append(char *path, struct stat *st, int mode,
+                         ssize_t offset,
                          struct flb_tail_config *ctx);
 void flb_tail_file_remove(struct flb_tail_file *file);
 int flb_tail_file_remove_all(struct flb_tail_config *ctx);

plugins/in_tail/tail_scan.c

@@ -30,6 +30,30 @@
 #include "tail_scan_glob.c"
 #endif
 
+void flb_tail_scan_register_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length, size_t size)
+{
+    flb_hash_table_add(ctx->ignored_file_sizes, path, path_length, (void *) size, 0);
+
+}
+
+void flb_tail_scan_unregister_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length)
+{
+    flb_hash_table_del(ctx->ignored_file_sizes, path);
+}
+
+ssize_t flb_tail_scan_fetch_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length)
+{
+    ssize_t result;
+
+    result = (ssize_t) flb_hash_table_get_ptr(ctx->ignored_file_sizes, path, path_length);
+
+    if (result == 0) {
+        result = -1;
+    }
+
+    return result;
+}
+
 int flb_tail_scan(struct mk_list *path_list, struct flb_tail_config *ctx)
 {
     int ret;

plugins/in_tail/tail_scan.h

@@ -26,4 +26,8 @@ int flb_tail_scan(struct mk_list *path, struct flb_tail_config *ctx);
 int flb_tail_scan_callback(struct flb_input_instance *ins,
                            struct flb_config *config, void *context);
 
+void flb_tail_scan_register_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length, size_t size);
+void flb_tail_scan_unregister_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length);
+ssize_t flb_tail_scan_fetch_ignored_file_size(struct flb_tail_config *ctx, const char *path, size_t path_length);
+
 #endif

plugins/in_tail/tail_scan_glob.c

@@ -183,6 +183,7 @@ static inline int do_glob(const char *pattern, int flags,
     return ret;
 }
 
+
 /* Scan a path, register the entries and return how many */
 static int tail_scan_path(const char *path, struct flb_tail_config *ctx)
 {
@@ -193,6 +194,9 @@ static int tail_scan_path(const char *path, struct flb_tail_config *ctx)
     time_t now;
     int64_t mtime;
     struct stat st;
+    ssize_t ignored_file_size;
+
+    ignored_file_size = -1;
 
     flb_plg_debug(ctx->ins, "scanning path %s", path);
 
@@ -245,14 +249,36 @@ static int tail_scan_path(const char *path, struct flb_tail_config *ctx)
                     if ((now - ctx->ignore_older) > mtime) {
                         flb_plg_debug(ctx->ins, "excluded=%s (ignore_older)",
                                       globbuf.gl_pathv[i]);
+
+                        flb_tail_scan_register_ignored_file_size(
+                            ctx,
+                            globbuf.gl_pathv[i],
+                            strlen(globbuf.gl_pathv[i]),
+                            st.st_size);
+
                         continue;
                     }
                 }
             }
 
+            if (ctx->ignore_older > 0) {
+                ignored_file_size = flb_tail_scan_fetch_ignored_file_size(
+                                        ctx,
+                                        globbuf.gl_pathv[i],
+                                        strlen(globbuf.gl_pathv[i]));
+
+                flb_tail_scan_unregister_ignored_file_size(
+                    ctx,
+                    globbuf.gl_pathv[i],
+                    strlen(globbuf.gl_pathv[i]));
+            }
+
             /* Append file to list */
             ret = flb_tail_file_append(globbuf.gl_pathv[i], &st,
-                                       FLB_TAIL_STATIC, ctx);
+                                       FLB_TAIL_STATIC,
+                                       ignored_file_size,
+                                       ctx);
+
             if (ret == 0) {
                 flb_plg_debug(ctx->ins, "scan_glob add(): %s, inode %" PRIu64,
                               globbuf.gl_pathv[i], (uint64_t) st.st_ino);

plugins/in_tail/tail_scan_win32.c

@@ -65,6 +65,9 @@ static int tail_register_file(const char *target, struct flb_tail_config *ctx,
     int64_t mtime;
     struct stat st;
     char path[MAX_PATH];
+    ssize_t ignored_file_size;
+
+    ignored_file_size = -1;
 
     if (_fullpath(path, target, MAX_PATH) == NULL) {
         flb_plg_error(ctx->ins, "cannot get absolute path of %s", target);
@@ -81,6 +84,13 @@ static int tail_register_file(const char *target, struct flb_tail_config *ctx,
             if ((ts - ctx->ignore_older) > mtime) {
                 flb_plg_debug(ctx->ins, "excluded=%s (ignore_older)",
                               target);
+
+                flb_tail_scan_register_ignored_file_size(
+                    ctx,
+                    path,
+                    strlen(path),
+                    st.st_size);
+
                 return -1;
             }
         }
@@ -91,7 +101,19 @@ static int tail_register_file(const char *target, struct flb_tail_config *ctx,
         return -1;
     }
 
-    return flb_tail_file_append(path, &st, FLB_TAIL_STATIC, ctx);
+    if (ctx->ignore_older > 0) {
+        ignored_file_size = flb_tail_scan_fetch_ignored_file_size(
+                                ctx,
+                                path,
+                                strlen(path));
+
+        flb_tail_scan_unregister_ignored_file_size(
+            ctx,
+            path,
+            strlen(path));
+    }
+
+    return flb_tail_file_append(path, &st, FLB_TAIL_STATIC, ignored_file_size, ctx);
 }
 
 /*