in_tail: fix gzip data loss on resume and update DB schema

Previously, when tailing gzip files, there was no mechanism to persistently
store the uncompressed position ('skip_bytes'). This meant that upon restart,
the plugin could not correctly locate the reading position, identifying it as
a rotation or new file case, potentially leading to data loss.

To fix this, 'skip_bytes' is now stored in the database to persist the
uncompressed offset. Additionally, 'exclude_bytes' is introduced to track
runtime skipping without interfering with the persistent value.

The SQLite schema has been updated to include 'anchor_offset' and 'skip_bytes'
columns to support these features.

Signed-off-by: jinyong.choi <[email protected]>

Author: jinyongchoi

plugins/in_tail/tail_db.c

@@ -25,11 +25,13 @@
 #include "tail_sql.h"
 #include "tail_file.h"
 
-struct query_status {
-    int id;
-    int rows;
-    int64_t offset;
-};
+/* Callback to detect if a query returned any rows */
+static int cb_column_exists(void *data, int argc, char **argv, char **cols)
+{
+    int *found = (int *)data;
+    *found = 1;
+    return 0;
+}
 
 /* Open or create database required by tail plugin */
 struct flb_sqldb *flb_tail_db_open(const char *path,
@@ -38,6 +40,7 @@ struct flb_sqldb *flb_tail_db_open(const char *path,
                                    struct flb_config *config)
 {
     int ret;
+    int column_found;
     char tmp[64];
     struct flb_sqldb *db;
 
@@ -55,6 +58,54 @@ struct flb_sqldb *flb_tail_db_open(const char *path,
         return NULL;
     }
 
+    /* Check if 'skip' column exists (migration for older databases) */
+    column_found = 0;
+    ret = flb_sqldb_query(db,
+                          "SELECT 1 FROM pragma_table_info('in_tail_files') "
+                          "WHERE name='skip';",
+                          cb_column_exists, &column_found);
+    if (ret != FLB_OK) {
+        flb_plg_error(ctx->ins, "db: could not query table info for 'skip' column");
+        flb_sqldb_close(db);
+        return NULL;
+    }
+    if (column_found == 0) {
+        flb_plg_debug(ctx->ins, "db: migrating database, adding 'skip' column");
+        ret = flb_sqldb_query(db,
+                              "ALTER TABLE in_tail_files "
+                              "ADD COLUMN skip INTEGER DEFAULT 0;",
+                              NULL, NULL);
+        if (ret != FLB_OK) {
+            flb_plg_error(ctx->ins, "db: could not add 'skip' column");
+            flb_sqldb_close(db);
+            return NULL;
+        }
+    }
+
+    /* Check if 'anchor' column exists (migration for older databases) */
+    column_found = 0;
+    ret = flb_sqldb_query(db,
+                          "SELECT 1 FROM pragma_table_info('in_tail_files') "
+                          "WHERE name='anchor';",
+                          cb_column_exists, &column_found);
+    if (ret != FLB_OK) {
+        flb_plg_error(ctx->ins, "db: could not query table info for 'anchor' column");
+        flb_sqldb_close(db);
+        return NULL;
+    }
+    if (column_found == 0) {
+        flb_plg_debug(ctx->ins, "db: migrating database, adding 'anchor' column");
+        ret = flb_sqldb_query(db,
+                              "ALTER TABLE in_tail_files "
+                              "ADD COLUMN anchor INTEGER DEFAULT 0;",
+                              NULL, NULL);
+        if (ret != FLB_OK) {
+            flb_plg_error(ctx->ins, "db: could not add 'anchor' column");
+            flb_sqldb_close(db);
+            return NULL;
+        }
+    }
+
     if (ctx->db_sync >= 0) {
         snprintf(tmp, sizeof(tmp) - 1, SQL_PRAGMA_SYNC,
                  ctx->db_sync);
@@ -130,7 +181,8 @@ static int flb_tail_db_file_delete_by_id(struct flb_tail_config *ctx,
  */
 static int db_file_exists(struct flb_tail_file *file,
                           struct flb_tail_config *ctx,
-                          uint64_t *id, uint64_t *inode, off_t *offset)
+                          uint64_t *id, uint64_t *inode,
+                          int64_t *offset, uint64_t *skip, int64_t *anchor)
 {
     int ret;
     int exists = FLB_FALSE;
@@ -149,6 +201,8 @@ static int db_file_exists(struct flb_tail_file *file,
         /* name: column 1 */
         name = sqlite3_column_text(ctx->stmt_get_file, 1);
         if (ctx->compare_filename && name == NULL) {
+            sqlite3_clear_bindings(ctx->stmt_get_file);
+            sqlite3_reset(ctx->stmt_get_file);
             flb_plg_error(ctx->ins, "db: error getting name: id=%"PRIu64, *id);
             return -1;
         }
@@ -159,12 +213,18 @@ static int db_file_exists(struct flb_tail_file *file,
         /* inode: column 3 */
         *inode = sqlite3_column_int64(ctx->stmt_get_file, 3);
 
+        /* skip: column 6 */
+        *skip = sqlite3_column_int64(ctx->stmt_get_file, 6);
+
+        /* anchor: column 7 */
+        *anchor = sqlite3_column_int64(ctx->stmt_get_file, 7);
+
         /* Checking if the file's name and inode match exactly */
         if (ctx->compare_filename) {
             if (flb_tail_target_file_name_cmp((char *) name, file) != 0) {
                 exists = FLB_FALSE;
                 flb_plg_debug(ctx->ins, "db: exists stale file from database:"
-                             " id=%"PRIu64" inode=%"PRIu64" offset=%"PRIu64
+                             " id=%"PRIu64" inode=%"PRIu64" offset=%"PRId64
                              " name=%s file_inode=%"PRIu64" file_name=%s",
                              *id, *inode, *offset, name, file->inode,
                              file->name);
@@ -199,6 +259,8 @@ static int db_file_insert(struct flb_tail_file *file, struct flb_tail_config *ct
     sqlite3_bind_int64(ctx->stmt_insert_file, 2, file->offset);
     sqlite3_bind_int64(ctx->stmt_insert_file, 3, file->inode);
     sqlite3_bind_int64(ctx->stmt_insert_file, 4, created);
+    sqlite3_bind_int64(ctx->stmt_insert_file, 5, file->skip_bytes);
+    sqlite3_bind_int64(ctx->stmt_insert_file, 6, file->anchor_offset);
 
     /* Run the insert */
     ret = sqlite3_step(ctx->stmt_insert_file);
@@ -258,11 +320,13 @@ int flb_tail_db_file_set(struct flb_tail_file *file,
 {
     int ret;
     uint64_t id = 0;
-    off_t offset = 0;
+    int64_t offset = 0;
+    uint64_t skip = 0;
+    int64_t anchor = 0;
     uint64_t inode = 0;
 
     /* Check if the file exists */
-    ret = db_file_exists(file, ctx, &id, &inode, &offset);
+    ret = db_file_exists(file, ctx, &id, &inode, &offset, &skip, &anchor);
     if (ret == -1) {
         flb_plg_error(ctx->ins, "cannot execute query to check inode: %" PRIu64,
                       file->inode);
@@ -281,6 +345,18 @@ int flb_tail_db_file_set(struct flb_tail_file *file,
     else {
         file->db_id = id;
         file->offset = offset;
+        file->skip_bytes = skip;
+        file->anchor_offset = anchor;
+
+        /* Initialize skipping mode if needed */
+        if (file->skip_bytes > 0) {
+            file->exclude_bytes = file->skip_bytes;
+            file->skipping_mode = FLB_TRUE;
+        }
+        else {
+            file->exclude_bytes = 0;
+            file->skipping_mode = FLB_FALSE;
+        }
     }
 
     return 0;
@@ -294,7 +370,9 @@ int flb_tail_db_file_offset(struct flb_tail_file *file,
 
     /* Bind parameters */
     sqlite3_bind_int64(ctx->stmt_offset, 1, file->offset);
-    sqlite3_bind_int64(ctx->stmt_offset, 2, file->db_id);
+    sqlite3_bind_int64(ctx->stmt_offset, 2, file->skip_bytes);
+    sqlite3_bind_int64(ctx->stmt_offset, 3, file->anchor_offset);
+    sqlite3_bind_int64(ctx->stmt_offset, 4, file->db_id);
 
     ret = sqlite3_step(ctx->stmt_offset);
 

plugins/in_tail/tail_file.c

@@ -44,6 +44,7 @@
 #include "tail_dockermode.h"
 #include "tail_multiline.h"
 #include "tail_scan.h"
+#include <fluent-bit/flb_compression.h>
 
 #ifdef FLB_SYSTEM_WINDOWS
 #include "win32.h"
@@ -1036,12 +1037,39 @@ static int set_file_position(struct flb_tail_config *ctx,
     if (ctx->db) {
         ret = flb_tail_db_file_set(file, ctx);
         if (ret == 0) {
-            if (file->offset > 0) {
-                ret = lseek(file->fd, file->offset, SEEK_SET);
+            int64_t seek_pos;
+            int has_db_position;
+
+            /*
+             * Determine seek position based on file type and DB state:
+             *
+             * - Gzip files with anchor/skip info: use anchor_offset
+             * - Normal files or gzip migration fallback: use offset
+             * - No DB position + read_from_head=off: seek to EOF
+             */
+            seek_pos = file->offset;
+            has_db_position = (file->offset > 0);
+
+            /* Override for gzip files with proper anchor tracking */
+            if (file->decompression_context != NULL &&
+                (file->anchor_offset > 0 || file->skip_bytes > 0)) {
+                seek_pos = file->anchor_offset;
+                has_db_position = FLB_TRUE;
+            }
+
+            if (has_db_position) {
+                ret = lseek(file->fd, seek_pos, SEEK_SET);
                 if (ret == -1) {
                     flb_errno();
                     return -1;
                 }
+                file->offset = ret;
+
+                /* Initialize skip state for gzip resume */
+                if (file->decompression_context != NULL && file->skip_bytes > 0) {
+                    file->exclude_bytes = file->skip_bytes;
+                    file->skipping_mode = FLB_TRUE;
+                }
             }
             else if (ctx->read_from_head == FLB_FALSE) {
                 ret = lseek(file->fd, 0, SEEK_END);
@@ -1050,8 +1078,23 @@ static int set_file_position(struct flb_tail_config *ctx,
                     return -1;
                 }
                 file->offset = ret;
+                file->anchor_offset = ret;
                 flb_tail_db_file_offset(file, ctx);
             }
+
+            if (file->decompression_context == NULL) {
+                file->stream_offset = file->offset;
+            }
+            else {
+                /*
+                 * For single-member gzip, stream_offset = skip_bytes is correct.
+                 * For multi-member gzip, skip_bytes only tracks bytes within the
+                 * current member (reset at member boundaries), so stream_offset
+                 * may not reflect the total decompressed bytes from prior members.
+                 */
+                file->stream_offset = file->skip_bytes;
+            }
+
             return 0;
         }
     }
@@ -1084,6 +1127,15 @@ static int set_file_position(struct flb_tail_config *ctx,
     if (file->decompression_context == NULL) {
         file->stream_offset = ret;
     }
+    else {
+        /*
+         * Compressed file without DB: no persistent state available.
+         * Initialize skip-related fields to 0 for code consistency.
+         */
+        file->anchor_offset = file->offset;
+        file->exclude_bytes = 0;
+        file->stream_offset = 0;
+    }
 
     return 0;
 }
@@ -1279,6 +1331,12 @@ int flb_tail_file_append(char *path, struct stat *st, int mode,
     file->skip_next = FLB_FALSE;
     file->skip_warn = FLB_FALSE;
 
+    /* Initialize gzip resume fields */
+    file->anchor_offset = 0;
+    file->skip_bytes = 0;
+    file->exclude_bytes = 0;
+    file->skipping_mode = FLB_FALSE;
+
     /* Multiline core mode */
     if (ctx->ml_ctx) {
         /*
@@ -1454,18 +1512,20 @@ void flb_tail_file_remove(struct flb_tail_file *file)
              * If there is data in the buffer, it means it was not processed.
              * We must rewind the offset to ensure this data is re-read on restart.
              */
-            off_t old_offset = file->offset;
+            int64_t old_offset = file->offset;
 
             if (file->offset > file->buf_len) {
                 file->offset -= file->buf_len;
-            } else {
+            }
+            else {
                 file->offset = 0;
             }
 
             flb_plg_debug(ctx->ins, "inode=%"PRIu64" rewind offset for %s: "
-                          "old=%"PRId64" new=%"PRId64" (buf_len=%lu)",
-                          file->inode, file->name, old_offset, file->offset,
-                          (unsigned long)file->buf_len);
+                          "old=%jd new=%jd (buf_len=%zu)",
+                          file->inode, file->name,
+                          (intmax_t)old_offset, (intmax_t)file->offset,
+                          file->buf_len);
 
 #ifdef FLB_HAVE_SQLDB
             if (ctx->db && file->db_id > FLB_TAIL_DB_ID_NONE) {
@@ -1474,9 +1534,16 @@ void flb_tail_file_remove(struct flb_tail_file *file)
 #endif
         }
         else {
-            flb_plg_warn(ctx->ins, "inode=%"PRIu64" cannot rewind compressed file %s; "
-                         "%lu decompressed bytes in buffer may be lost on restart",
-                         file->inode, file->name, (unsigned long)file->buf_len);
+            flb_plg_debug(ctx->ins,
+                          "inode=%"PRIu64" file=%s: buffered data (%zu bytes) "
+                          "remains on exit for gzip input; cannot rewind raw "
+                          "offset safely with streaming decompression",
+                          file->inode, file->name, file->buf_len);
+#ifdef FLB_HAVE_SQLDB
+            if (ctx->db && file->db_id > FLB_TAIL_DB_ID_NONE) {
+                flb_tail_db_file_offset(file, ctx);
+            }
+#endif
         }
     }
 
@@ -1598,6 +1665,10 @@ static int adjust_counters(struct flb_tail_config *ctx, struct flb_tail_file *fi
                       file->inode, file->name, size_delta);
         file->offset = offset;
         file->buf_len = 0;
+        file->anchor_offset = offset;
+        file->skip_bytes = 0;
+        file->exclude_bytes = 0;
+        file->skipping_mode = FLB_FALSE;
 
         /* Update offset in the database file */
 #ifdef FLB_HAVE_SQLDB
@@ -1797,6 +1868,28 @@ int flb_tail_file_chunk(struct flb_tail_file *file)
                     return FLB_TAIL_ERROR;
                 }
 
+                if (file->skipping_mode == FLB_TRUE && decompressed_data_length > 0) {
+                    flb_plg_debug(ctx->ins,
+                                 "Skipping: anchor=%jd offset=%jd "
+                                 "exclude=%zu decompressed=%zu",
+                                 (intmax_t)file->anchor_offset,
+                                 (intmax_t)file->offset,
+                                 file->exclude_bytes, decompressed_data_length);
+                    if (file->exclude_bytes >= decompressed_data_length) {
+                        file->exclude_bytes -= decompressed_data_length;
+                        decompressed_data_length = 0;
+                    }
+                    else {
+                        size_t remain = decompressed_data_length - file->exclude_bytes;
+                        memmove(&file->buf_data[file->buf_len],
+                                &file->buf_data[file->buf_len + file->exclude_bytes],
+                                remain);
+                        decompressed_data_length = remain;
+                        file->exclude_bytes = 0;
+                        file->skipping_mode = FLB_FALSE;
+                    }
+                }
+
                 stream_data_length = decompressed_data_length;
             }
         }
@@ -1836,6 +1929,29 @@ int flb_tail_file_chunk(struct flb_tail_file *file)
         file->buf_len -= processed_bytes;
         file->buf_data[file->buf_len] = '\0';
 
+        if (file->decompression_context) {
+            file->skip_bytes += processed_bytes;
+
+            /*
+             * Gzip member boundary: update anchor when we complete a member
+             * and all decompressed data is consumed.
+             */
+            if (file->decompression_context->state ==
+                    FLB_DECOMPRESSOR_STATE_EXPECTING_HEADER &&
+                file->decompression_context->input_buffer_length == 0 &&
+                file->buf_len == 0) {
+                flb_plg_debug(file->config->ins,
+                             "Gzip member completed: updating anchor "
+                             "from %jd to %jd, resetting skip from "
+                             "%"PRIu64" to 0",
+                             (intmax_t)file->anchor_offset,
+                             (intmax_t)file->offset,
+                             file->skip_bytes);
+                file->anchor_offset = file->offset;
+                file->skip_bytes = 0;
+            }
+        }
+
 #ifdef FLB_HAVE_SQLDB
         if (file->config->db) {
             flb_tail_db_file_offset(file, file->config);