Skip to content

in_tail: Serialize DB operation for preventing resource collisions on threaded mode #11460

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 2 commits into from
Feb 13, 2026
Merged

in_tail: Serialize DB operation for preventing resource collisions on threaded mode #11460

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eeba913
in_tail: Serialize DB operation for preventing resource collisions
cosmo0920 Feb 13, 2026
82a1d33
lock: Handle EBUSY error code as retry
cosmo0920 Feb 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions plugins/in_tail/tail_db.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,24 @@ struct query_status {
int64_t offset;
};

static inline int tail_db_lock(struct flb_tail_config *ctx)
{
if (ctx->db == NULL) {
return 0;
}

return flb_sqldb_lock(ctx->db);
}

static inline int tail_db_unlock(struct flb_tail_config *ctx)
{
if (ctx->db == NULL) {
return 0;
}

return flb_sqldb_unlock(ctx->db);
}

/* Open or create database required by tail plugin */
struct flb_sqldb *flb_tail_db_open(const char *path,
struct flb_input_instance *in,
Expand Down Expand Up @@ -261,11 +279,21 @@ int flb_tail_db_file_set(struct flb_tail_file *file,
off_t offset = 0;
uint64_t inode = 0;

flb_plg_debug(ctx->ins, "db file set called for %s inode=%"PRIu64,
file->name, file->inode);

ret = tail_db_lock(ctx);
if (ret != 0) {
flb_plg_error(ctx->ins, "db: could not acquire lock");
return -1;
}

/* Check if the file exists */
ret = db_file_exists(file, ctx, &id, &inode, &offset);
if (ret == -1) {
flb_plg_error(ctx->ins, "cannot execute query to check inode: %" PRIu64,
file->inode);
tail_db_unlock(ctx);
return -1;
}

Expand All @@ -283,6 +311,7 @@ int flb_tail_db_file_set(struct flb_tail_file *file,
file->offset = offset;
}

tail_db_unlock(ctx);
return 0;
}

Expand All @@ -292,6 +321,12 @@ int flb_tail_db_file_offset(struct flb_tail_file *file,
{
int ret;

ret = tail_db_lock(ctx);
if (ret != 0) {
flb_plg_error(ctx->ins, "db: could not acquire lock");
return -1;
}
Comment thread
cosmo0920 marked this conversation as resolved.

/* Bind parameters */
sqlite3_bind_int64(ctx->stmt_offset, 1, file->offset);
sqlite3_bind_int64(ctx->stmt_offset, 2, file->db_id);
Expand All @@ -301,6 +336,7 @@ int flb_tail_db_file_offset(struct flb_tail_file *file,
if (ret != SQLITE_DONE) {
sqlite3_clear_bindings(ctx->stmt_offset);
sqlite3_reset(ctx->stmt_offset);
tail_db_unlock(ctx);
return -1;
}

Expand All @@ -317,6 +353,7 @@ int flb_tail_db_file_offset(struct flb_tail_file *file,
sqlite3_clear_bindings(ctx->stmt_offset);
sqlite3_reset(ctx->stmt_offset);

tail_db_unlock(ctx);
return 0;
}

Expand All @@ -327,6 +364,12 @@ int flb_tail_db_file_rotate(const char *new_name,
{
int ret;

ret = tail_db_lock(ctx);
if (ret != 0) {
flb_plg_error(ctx->ins, "db: could not acquire lock");
return -1;
}

/* Bind parameters */
sqlite3_bind_text(ctx->stmt_rotate_file, 1, new_name, -1, 0);
sqlite3_bind_int64(ctx->stmt_rotate_file, 2, file->db_id);
Expand All @@ -337,9 +380,11 @@ int flb_tail_db_file_rotate(const char *new_name,
sqlite3_reset(ctx->stmt_rotate_file);

if (ret != SQLITE_DONE) {
tail_db_unlock(ctx);
return -1;
}

tail_db_unlock(ctx);
return 0;
}

Expand All @@ -349,6 +394,12 @@ int flb_tail_db_file_delete(struct flb_tail_file *file,
{
int ret;

ret = tail_db_lock(ctx);
if (ret != 0) {
flb_plg_error(ctx->ins, "db: could not acquire lock");
return -1;
}

/* Bind parameters */
sqlite3_bind_int64(ctx->stmt_delete_file, 1, file->db_id);
ret = sqlite3_step(ctx->stmt_delete_file);
Expand All @@ -359,10 +410,12 @@ int flb_tail_db_file_delete(struct flb_tail_file *file,
if (ret != SQLITE_DONE) {
flb_plg_error(ctx->ins, "db: error deleting entry from database: %s",
file->name);
tail_db_unlock(ctx);
return -1;
}

flb_plg_debug(ctx->ins, "db: file deleted from database: %s", file->name);
tail_db_unlock(ctx);
return 0;
}

Expand All @@ -388,6 +441,12 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
return 0;
}

ret = tail_db_lock(ctx);
if (ret != 0) {
flb_plg_error(ctx->ins, "db: could not acquire lock");
return -1;
}

/* Create a stmt sql buffer */
sql_size = SQL_DELETE_STALE_FILE_START_LEN;
sql_size += SQL_DELETE_STALE_FILE_WHERE_LEN;
Expand All @@ -402,6 +461,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
if (!stale_delete_sql) {
flb_plg_error(ctx->ins, "cannot allocate buffer for stale_delete_sql:"
" size: %zu", sql_size);
tail_db_unlock(ctx);
return -1;
}

Expand All @@ -412,6 +472,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
flb_plg_error(ctx->ins,
"error concatenating stale_delete_sql: start");
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}
stale_delete_sql = sds_tmp;
Expand All @@ -423,6 +484,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
flb_plg_error(ctx->ins,
"error concatenating stale_delete_sql: where");
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}
stale_delete_sql = sds_tmp;
Expand All @@ -432,6 +494,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
flb_plg_error(ctx->ins,
"error concatenating stale_delete_sql: param");
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}
}
Expand All @@ -441,6 +504,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
flb_plg_error(ctx->ins,
"error concatenating stale_delete_sql: end");
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}
stale_delete_sql = sds_tmp;
Expand All @@ -453,6 +517,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
" stmt_delete_inodes sql:%s, ret=%d", stale_delete_sql,
ret);
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}

Expand All @@ -466,6 +531,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
" inode=%" PRIu64 ", ret=%d", file->inode, ret);
sqlite3_finalize(stmt_delete_inodes);
flb_sds_destroy(stale_delete_sql);
tail_db_unlock(ctx);
return -1;
}
idx++;
Expand All @@ -478,6 +544,7 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
flb_sds_destroy(stale_delete_sql);
flb_plg_error(ctx->ins, "cannot execute delete stale inodes: ret=%d",
ret);
tail_db_unlock(ctx);
return -1;
}

Expand All @@ -488,5 +555,6 @@ int flb_tail_db_stale_file_delete(struct flb_input_instance *ins,
sqlite3_finalize(stmt_delete_inodes);
flb_sds_destroy(stale_delete_sql);

tail_db_unlock(ctx);
return 0;
}
2 changes: 1 addition & 1 deletion src/flb_lock.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ int flb_lock_acquire(flb_lock_t *lock,
result = pthread_mutex_trylock(lock);

if (result != 0) {
if (result == EAGAIN) {
if (result == EAGAIN || result == EBUSY) {
if (retry_limit != FLB_LOCK_INFINITE_RETRY_LIMIT) {
retry_count++;
}
Expand Down
Loading