in_tail: Preventing incorrect inode usage from db. #8025

plugins/in_tail/tail.c

@@ -372,6 +372,15 @@ static int in_tail_init(struct flb_input_instance *in,
     /* Scan path */
     flb_tail_scan(ctx->path_list, ctx);
 
+#ifdef FLB_HAVE_SQLDB
+    /* Delete stale files that are not monitored from the database */
+    ret = flb_tail_db_stale_file_delete(in, config, ctx);
+    if (ret == -1) {
+        flb_tail_config_destroy(ctx);
+        return -1;
+    }
+#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
@@ -725,6 +734,14 @@ static struct flb_config_map config_map[] = {
      "provides higher performance. Note that WAL is not compatible with "
      "shared network file systems."
     },
+    {
+     FLB_CONFIG_MAP_BOOL, "db.compare_filename", "false",
+     0, FLB_TRUE, offsetof(struct flb_tail_config, compare_filename),
+     "This option determines whether to check both the inode and the filename "
+     "when retrieving file information from the db."
+     "'true' verifies both the inode and filename, while 'false' checks only "
+     "the inode (default)."
+    },
 #endif
 
     /* Multiline Options */

plugins/in_tail/tail_config.h

@@ -107,6 +107,7 @@ struct flb_tail_config {
     struct flb_sqldb *db;
     int db_sync;
     int db_locking;
+    int compare_filename;
     flb_sds_t db_journal_mode;
     sqlite3_stmt *stmt_get_file;
     sqlite3_stmt *stmt_insert_file;

plugins/in_tail/tail_db.c

@@ -95,16 +95,46 @@ int flb_tail_db_close(struct flb_sqldb *db)
     return 0;
 }
 
+static int flb_tail_db_file_delete_by_id(struct flb_tail_config *ctx,
+                                         uint64_t id)
+{
+    int ret;
+
+    /* Bind parameters */
+    ret = sqlite3_bind_int64(ctx->stmt_delete_file, 1, id);
+    if (ret != SQLITE_OK) {
+        flb_plg_error(ctx->ins, "db: error binding id=%"PRIu64", ret=%d", id, ret);
+        return -1;
+    }
+
+    ret = sqlite3_step(ctx->stmt_delete_file);
+
+    sqlite3_clear_bindings(ctx->stmt_delete_file);
+    sqlite3_reset(ctx->stmt_delete_file);
+
+    if (ret != SQLITE_DONE) {
+        flb_plg_error(ctx->ins, "db: error deleting stale entry from database:"
+                      " id=%"PRIu64, id);
+        return -1;
+    }
+
+    flb_plg_info(ctx->ins, "db: stale file deleted from database:"
+                 " id=%"PRIu64, id);
+    return 0;
+}
+
 /*
- * Check if an file inode exists in the database. Return FLB_TRUE or
- * FLB_FALSE
+ * Check if an file inode exists in the database.
+ * If the 'compare_filename' option is enabled,
+ * it checks along with the filename. Return FLB_TRUE or FLB_FALSE
  */
 static int db_file_exists(struct flb_tail_file *file,
                           struct flb_tail_config *ctx,
                           uint64_t *id, uint64_t *inode, off_t *offset)
 {
     int ret;
     int exists = FLB_FALSE;
+    const unsigned char *name;
 
     /* Bind parameters */
     sqlite3_bind_int64(ctx->stmt_get_file, 1, file->inode);
@@ -116,11 +146,30 @@ static int db_file_exists(struct flb_tail_file *file,
         /* id: column 0 */
         *id = sqlite3_column_int64(ctx->stmt_get_file, 0);
 
+        /* name: column 1 */
+        name = sqlite3_column_text(ctx->stmt_get_file, 1);
+        if (ctx->compare_filename && name == NULL) {
+            flb_plg_error(ctx->ins, "db: error getting name: id=%"PRIu64, *id);
+            return -1;
+        }
+
         /* offset: column 2 */
         *offset = sqlite3_column_int64(ctx->stmt_get_file, 2);
 
         /* inode: column 3 */
         *inode = sqlite3_column_int64(ctx->stmt_get_file, 3);
+
+        /* 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
+                             " name=%s file_inode=%"PRIu64" file_name=%s",
+                             *id, *inode, *offset, name, file->inode,
+                             file->name);
+            }
+        }
     }
     else if (ret == SQLITE_DONE) {
         /* all good */
@@ -168,6 +217,42 @@ static int db_file_insert(struct flb_tail_file *file, struct flb_tail_config *ct
     return flb_sqldb_last_id(ctx->db);
 }
 
+static int stmt_add_param_concat(struct flb_tail_config *ctx,
+                                 flb_sds_t *stmt_sql, uint64_t count)
+{
+    uint64_t idx;
+    flb_sds_t sds_tmp;
+
+    sds_tmp = flb_sds_cat(*stmt_sql, SQL_STMT_START_PARAM,
+                          SQL_STMT_START_PARAM_LEN);
+    if (sds_tmp == NULL) {
+        flb_plg_debug(ctx->ins, "error concatenating stmt_sql: param start");
+        return -1;
+    }
+    *stmt_sql = sds_tmp;
+
+    for (idx = 1; idx < count; idx++) {
+        sds_tmp = flb_sds_cat(*stmt_sql, SQL_STMT_ADD_PARAM,
+                              SQL_STMT_ADD_PARAM_LEN);
+        if (sds_tmp == NULL) {
+            flb_plg_debug(ctx->ins, "error concatenating stmt_sql: add param");
+            return -1;
+        }
+
+        *stmt_sql = sds_tmp;
+    }
+
+    sds_tmp = flb_sds_cat(*stmt_sql, SQL_STMT_PARAM_END,
+                          SQL_STMT_PARAM_END_LEN);
+    if (sds_tmp == NULL) {
+        flb_plg_debug(ctx->ins, "error concatenating stmt_sql: param end");
+        return -1;
+    }
+    *stmt_sql = sds_tmp;
+
+    return 0;
+}
+
 int flb_tail_db_file_set(struct flb_tail_file *file,
                          struct flb_tail_config *ctx)
 {
@@ -185,6 +270,11 @@ int flb_tail_db_file_set(struct flb_tail_file *file,
     }
 
     if (ret == FLB_FALSE) {
+        /* Delete stale file of same inode */
+        if (ctx->compare_filename && id > 0) {
+            flb_tail_db_file_delete_by_id(ctx, id);
+        }
+
         /* Get the database ID for this file */
         file->db_id = db_file_insert(file, ctx);
     }
@@ -275,3 +365,128 @@ int flb_tail_db_file_delete(struct flb_tail_file *file,
     flb_plg_debug(ctx->ins, "db: file deleted from database: %s", file->name);
     return 0;
 }
+
+/*
+ * Delete stale file from database
+ */
+int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
+                                  struct flb_config *config,
+                                  struct flb_tail_config *ctx)
+{
+    int ret = -1;
+    size_t sql_size;
+    uint64_t idx;
+    uint64_t file_count = ctx->files_static_count;
+    flb_sds_t stale_delete_sql;
+    flb_sds_t sds_tmp;
+    sqlite3_stmt *stmt_delete_inodes = NULL;
+    struct mk_list *tmp;
+    struct mk_list *head;
+    struct flb_tail_file *file;
+
+    if (!ctx->db) {
+        return 0;
+    }
+
+    /* Create a stmt sql buffer */
+    sql_size = SQL_DELETE_STALE_FILE_START_LEN;
+    sql_size += SQL_DELETE_STALE_FILE_WHERE_LEN;
+    sql_size += SQL_STMT_START_PARAM_LEN;
+    sql_size += SQL_STMT_PARAM_END_LEN;
+    sql_size += SQL_STMT_END_LEN;
+    if (file_count > 0) {
+        sql_size += (SQL_STMT_ADD_PARAM_LEN * file_count);
+    }
+
+    stale_delete_sql = flb_sds_create_size(sql_size + 1);
+    if (!stale_delete_sql) {
+        flb_plg_error(ctx->ins, "cannot allocate buffer for stale_delete_sql:"
+                      " size: %zu", sql_size);
+        return -1;
+    }
+
+    /* Create a stmt sql */
+    sds_tmp = flb_sds_cat(stale_delete_sql, SQL_DELETE_STALE_FILE_START,
+                          SQL_DELETE_STALE_FILE_START_LEN);
+    if (sds_tmp == NULL) {
+        flb_plg_error(ctx->ins,
+                      "error concatenating stale_delete_sql: start");
+        flb_sds_destroy(stale_delete_sql);
+        return -1;
+    }
+    stale_delete_sql = sds_tmp;
+
+    if (file_count > 0) {
+        sds_tmp = flb_sds_cat(stale_delete_sql, SQL_DELETE_STALE_FILE_WHERE,
+                              SQL_DELETE_STALE_FILE_WHERE_LEN);
+        if (sds_tmp == NULL) {
+            flb_plg_error(ctx->ins,
+                          "error concatenating stale_delete_sql: where");
+            flb_sds_destroy(stale_delete_sql);
+            return -1;
+        }
+        stale_delete_sql = sds_tmp;
+
+        ret = stmt_add_param_concat(ctx, &stale_delete_sql, file_count);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins,
+                          "error concatenating stale_delete_sql: param");
+            flb_sds_destroy(stale_delete_sql);
+            return -1;
+        }
+    }
+
+    sds_tmp = flb_sds_cat(stale_delete_sql, SQL_STMT_END, SQL_STMT_END_LEN);
+    if (sds_tmp == NULL) {
+        flb_plg_error(ctx->ins,
+                      "error concatenating stale_delete_sql: end");
+        flb_sds_destroy(stale_delete_sql);
+        return -1;
+    }
+    stale_delete_sql = sds_tmp;
+
+    /* Prepare stmt */
+    ret = sqlite3_prepare_v2(ctx->db->handler, stale_delete_sql, -1,
+                             &stmt_delete_inodes, 0);
+    if (ret != SQLITE_OK) {
+        flb_plg_error(ctx->ins, "error preparing database SQL statement:"
+                      " stmt_delete_inodes sql:%s, ret=%d", stale_delete_sql,
+                      ret);
+        flb_sds_destroy(stale_delete_sql);
+        return -1;
+    }
+
+    /* Bind parameters */
+    idx = 1;
+    mk_list_foreach_safe(head, tmp, &ctx->files_static) {
+        file = mk_list_entry(head, struct flb_tail_file, _head);
+        ret = sqlite3_bind_int64(stmt_delete_inodes, idx, file->inode);
+        if (ret != SQLITE_OK) {
+            flb_plg_error(ctx->ins, "error binding to stmt_delete_inodes:"
+                          " inode=%lu, ret=%d", file->inode, ret);
+            sqlite3_finalize(stmt_delete_inodes);
+            flb_sds_destroy(stale_delete_sql);
+            return -1;
+        }
+        idx++;
+    }
+
+    /* Run the delete inodes */
+    ret = sqlite3_step(stmt_delete_inodes);
+    if (ret != SQLITE_DONE) {
+        sqlite3_finalize(stmt_delete_inodes);
+        flb_sds_destroy(stale_delete_sql);
+        flb_plg_error(ctx->ins, "cannot execute delete stale inodes: ret=%d",
+                      ret);
+        return -1;
+    }
+
+    ret = sqlite3_changes(ctx->db->handler);
+    flb_plg_info(ctx->ins, "db: delete unmonitored stale inodes from the"
+                 " database: count=%d", ret);
+
+    sqlite3_finalize(stmt_delete_inodes);
+    flb_sds_destroy(stale_delete_sql);
+
+    return 0;
+}

plugins/in_tail/tail_db.h

@@ -40,4 +40,7 @@ int flb_tail_db_file_rotate(const char *new_name,
                             struct flb_tail_config *ctx);
 int flb_tail_db_file_delete(struct flb_tail_file *file,
                             struct flb_tail_config *ctx);
+int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
+                                  struct flb_config *config,
+                                  struct flb_tail_config *ctx);
 #endif

plugins/in_tail/tail_sql.h

@@ -53,6 +53,28 @@
 #define SQL_DELETE_FILE                                                 \
     "DELETE FROM in_tail_files WHERE id=@id;"
 
+#define SQL_STMT_START_PARAM "(?"
+#define SQL_STMT_START_PARAM_LEN (sizeof(SQL_STMT_START_PARAM) - 1)
+
+#define SQL_STMT_ADD_PARAM ",?"
+#define SQL_STMT_ADD_PARAM_LEN (sizeof(SQL_STMT_ADD_PARAM) - 1)
+
+#define SQL_STMT_PARAM_END ")"
+#define SQL_STMT_PARAM_END_LEN (sizeof(SQL_STMT_PARAM_END) - 1)
+
+#define SQL_STMT_END ";"
+#define SQL_STMT_END_LEN (sizeof(SQL_STMT_END) - 1)
+
+#define SQL_DELETE_STALE_FILE_START                                     \
+    "DELETE FROM in_tail_files "
+#define SQL_DELETE_STALE_FILE_START_LEN                                 \
+    (sizeof(SQL_DELETE_STALE_FILE_START) - 1)
+
+#define SQL_DELETE_STALE_FILE_WHERE                                     \
+    "WHERE inode NOT IN "
+#define SQL_DELETE_STALE_FILE_WHERE_LEN                                 \
+    (sizeof(SQL_DELETE_STALE_FILE_WHERE) - 1)
+
 #define SQL_PRAGMA_SYNC                         \
     "PRAGMA synchronous=%i;"