in_tail: move all database queries to prepared statements (#2030)

edsiper · edsiper · commit c0d80999869a · 2020-08-22T16:33:57.000-06:00
A problem reported in #2030 lead to find solutions that avoids using fixed size buffers to compose SQL queries. Using the heap for database queries in our case is expensive. The solution proposed is to migrate all SQL queries to prepared SQL statements, so we avoid having local buffers and pass the variables directly to SQLite. Signed-off-by: Eduardo Silva <eduardo@treasure-data.com>
diff --git a/plugins/in_tail/tail_config.c b/plugins/in_tail/tail_config.c
@@ -239,6 +239,43 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
 
     /* Prepare Statement */
     if (ctx->db) {
+        /* SQL_GET_FILE */
+        ret = sqlite3_prepare_v2(ctx->db->handler,
+                                 SQL_GET_FILE,
+                                 -1,
+                                 &ctx->stmt_get_file,
+                                 0);
+        if (ret != SQLITE_OK) {
+            flb_plg_error(ctx->ins, "error preparing database SQL statement");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+
+        /* SQL_INSERT_FILE */
+        ret = sqlite3_prepare_v2(ctx->db->handler,
+                                 SQL_INSERT_FILE,
+                                 -1,
+                                 &ctx->stmt_insert_file,
+                                 0);
+        if (ret != SQLITE_OK) {
+            flb_plg_error(ctx->ins, "error preparing database SQL statement");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+
+        /* SQL_ROTATE_FILE */
+        ret = sqlite3_prepare_v2(ctx->db->handler,
+                                 SQL_ROTATE_FILE,
+                                 -1,
+                                 &ctx->stmt_rotate_file,
+                                 0);
+        if (ret != SQLITE_OK) {
+            flb_plg_error(ctx->ins, "error preparing database SQL statement");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+
+        /* SQL_UPDATE_OFFSET */
         ret = sqlite3_prepare_v2(ctx->db->handler,
                                  SQL_UPDATE_OFFSET,
                                  -1,
@@ -249,6 +286,19 @@ struct flb_tail_config *flb_tail_config_create(struct flb_input_instance *ins,
             flb_tail_config_destroy(ctx);
             return NULL;
         }
+
+        /* SQL_DELETE_FILE */
+        ret = sqlite3_prepare_v2(ctx->db->handler,
+                                 SQL_DELETE_FILE,
+                                 -1,
+                                 &ctx->stmt_delete_file,
+                                 0);
+        if (ret != SQLITE_OK) {
+            flb_plg_error(ctx->ins, "error preparing database SQL statement");
+            flb_tail_config_destroy(ctx);
+            return NULL;
+        }
+
     }
 #endif
 
@@ -285,6 +335,10 @@ int flb_tail_config_destroy(struct flb_tail_config *config)
 
 #ifdef FLB_HAVE_SQLDB
     if (config->db != NULL) {
+        sqlite3_finalize(config->stmt_get_file);
+        sqlite3_finalize(config->stmt_insert_file);
+        sqlite3_finalize(config->stmt_delete_file);
+        sqlite3_finalize(config->stmt_rotate_file);
         sqlite3_finalize(config->stmt_offset);
         flb_tail_db_close(config->db);
     }
diff --git a/plugins/in_tail/tail_config.h b/plugins/in_tail/tail_config.h
@@ -87,6 +87,10 @@ struct flb_tail_config {
 #ifdef FLB_HAVE_SQLDB
     struct flb_sqldb *db;
     int db_sync;
+    sqlite3_stmt *stmt_get_file;
+    sqlite3_stmt *stmt_insert_file;
+    sqlite3_stmt *stmt_delete_file;
+    sqlite3_stmt *stmt_rotate_file;
     sqlite3_stmt *stmt_offset;
 #endif
 
diff --git a/plugins/in_tail/tail_db.c b/plugins/in_tail/tail_db.c
@@ -84,55 +84,99 @@ int flb_tail_db_close(struct flb_sqldb *db)
     return 0;
 }
 
+/*
+ * Check if an file inode exists in the database. 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;
+
+    /* Bind parameters */
+    sqlite3_bind_int64(ctx->stmt_get_file, 1, file->inode);
+    ret = sqlite3_step(ctx->stmt_get_file);
+
+    if (ret == SQLITE_ROW) {
+        exists = FLB_TRUE;
+
+        /* id: column 0 */
+        *id = sqlite3_column_int64(ctx->stmt_get_file, 0);
+
+        /* offset: column 2 */
+        *offset = sqlite3_column_int64(ctx->stmt_get_file, 2);
+
+        /* inode: column 3 */
+        *inode = sqlite3_column_int64(ctx->stmt_get_file, 3);
+    }
+    else if (ret == SQLITE_DONE) {
+        /* all good */
+    }
+    else {
+        exists = -1;
+    }
+
+    sqlite3_clear_bindings(ctx->stmt_get_file);
+    sqlite3_reset(ctx->stmt_get_file);
+
+    return exists;
+
+}
+
+static int db_file_insert(struct flb_tail_file *file, struct flb_tail_config *ctx)
 
-static int cb_file_check(void *data, int argc, char **argv, char **cols)
 {
-    struct query_status *qs = data;
+    int ret;
+    time_t created;
 
-    qs->id = atoi(argv[0]);      /* id column */
-    qs->offset = atoll(argv[2]); /* offset column */
+    /* Register the file */
+    created = time(NULL);
 
-    qs->rows++;
-    return 0;
+    /* Bind parameters */
+    sqlite3_bind_text(ctx->stmt_insert_file, 1, file->name, -1, 0);
+    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);
+
+    /* Run the insert */
+    ret = sqlite3_step(ctx->stmt_insert_file);
+    if (ret != SQLITE_DONE) {
+        flb_plg_error(ctx->ins, "cannot execute insert file %s inode=%lu",
+                      file->name, file->inode);
+        return -1;
+    }
+
+    /* Get the database ID for this file */
+    return flb_sqldb_last_id(ctx->db);
 }
 
 int flb_tail_db_file_set(struct flb_tail_file *file,
                          struct flb_tail_config *ctx)
 {
     int ret;
-    char query[PATH_MAX];
-    struct query_status qs = {0};
-    uint64_t created;
+    uint64_t id = 0;
+    off_t offset = 0;
+    uint64_t inode = 0;
 
     /* Check if the file exists */
-    snprintf(query, sizeof(query) - 1, SQL_GET_FILE, (uint64_t) file->inode);
-    memset(&qs, '\0', sizeof(qs));
-    ret = flb_sqldb_query(ctx->db,
-                          query, cb_file_check, &qs);
-    if (ret == FLB_ERROR) {
-        flb_plg_error(ctx->ins, "cannot execute SQL: %s", query);
+    ret = db_file_exists(file, ctx, &id, &inode, &offset);
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "cannot execute query to check inode: %lu",
+                      file->inode);
         return -1;
     }
 
-    if (qs.rows == 0) {
-        /* Register the file */
-        created = time(NULL);
-        snprintf(query, sizeof(query) - 1,
-                 SQL_INSERT_FILE,
-                 file->name, (uint64_t) 0, (uint64_t) file->inode, created);
-        ret = flb_sqldb_query(ctx->db, query, NULL, NULL);
-        if (ret == FLB_ERROR) {
-            flb_plg_error(ctx->ins, "cannot execute SQL: %s", query);
-            return -1;
-        }
-
+    if (ret == FLB_FALSE) {
         /* Get the database ID for this file */
-        file->db_id = flb_sqldb_last_id(ctx->db);
-        return 0;
+        file->db_id = db_file_insert(file, ctx);
+    }
+    else {
+        file->db_id = id;
+        file->offset = offset;
     }
 
-    file->db_id  = qs.id;
-    file->offset = qs.offset;
     return 0;
 }
 
@@ -148,32 +192,45 @@ int flb_tail_db_file_offset(struct flb_tail_file *file,
 
     ret = sqlite3_step(ctx->stmt_offset);
 
-    sqlite3_clear_bindings(ctx->stmt_offset);
-    sqlite3_reset(ctx->stmt_offset);
-
     if (ret != SQLITE_DONE) {
+        sqlite3_clear_bindings(ctx->stmt_offset);
+        sqlite3_reset(ctx->stmt_offset);
         return -1;
     }
 
+    /* Verify number of updated rows */
+    ret = sqlite3_changes(ctx->db->handler);
+    if (ret == 0) {
+        /*
+         * 'someone' like you 'the reader' or another user has deleted the database
+         * entry, just restore it.
+         */
+        file->db_id = db_file_insert(file, ctx);
+    }
+
+    sqlite3_clear_bindings(ctx->stmt_offset);
+    sqlite3_reset(ctx->stmt_offset);
+
     return 0;
 }
 
-/* Mark a file as rotated */
+/* Mark a file as rotated v2 */
 int flb_tail_db_file_rotate(const char *new_name,
                             struct flb_tail_file *file,
                             struct flb_tail_config *ctx)
 {
     int ret;
-    char query[PATH_MAX];
-    struct query_status qs = {0};
 
-    snprintf(query, sizeof(query) - 1,
-             SQL_ROTATE_FILE,
-             new_name, file->db_id);
+    /* 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);
 
-    ret = flb_sqldb_query(ctx->db,
-                          query, cb_file_check, &qs);
-    if (ret != FLB_OK) {
+    ret = sqlite3_step(ctx->stmt_rotate_file);
+
+    sqlite3_clear_bindings(ctx->stmt_rotate_file);
+    sqlite3_reset(ctx->stmt_rotate_file);
+
+    if (ret != SQLITE_DONE) {
         return -1;
     }
 
@@ -185,12 +242,15 @@ int flb_tail_db_file_delete(struct flb_tail_file *file,
                             struct flb_tail_config *ctx)
 {
     int ret;
-    char query[PATH_MAX];
 
-    /* Check if the file exists */
-    snprintf(query, sizeof(query) - 1, SQL_DELETE_FILE, file->db_id);
-    ret = flb_sqldb_query(ctx->db, query, NULL, NULL);
-    if (ret != FLB_OK) {
+    /* Bind parameters */
+    sqlite3_bind_int64(ctx->stmt_delete_file, 1, file->db_id);
+    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 entry from database: %s",
                       file->name);
         return -1;
diff --git a/plugins/in_tail/tail_sql.h b/plugins/in_tail/tail_sql.h
@@ -38,20 +38,20 @@
     "  rotated INTEGER DEFAULT 0"                                       \
     ");"
 
-#define SQL_GET_FILE "SELECT * from in_tail_files WHERE inode=%"PRIu64";"
+#define SQL_GET_FILE "SELECT * from in_tail_files WHERE inode=@inode;"
 
 #define SQL_INSERT_FILE                                             \
     "INSERT INTO in_tail_files (name, offset, inode, created)"      \
-    "  VALUES ('%s', %"PRIu64", %"PRIu64", %"PRIu64");"
+    "  VALUES (@name, @offset, @inode, @created);"
+
+#define SQL_ROTATE_FILE                                                 \
+    "UPDATE in_tail_files set name=@name,rotated=1 WHERE id=@id;"
 
 #define SQL_UPDATE_OFFSET                                   \
     "UPDATE in_tail_files set offset=@offset WHERE id=@id;"
 
-#define SQL_ROTATE_FILE                                                 \
-    "UPDATE in_tail_files set name='%s',rotated=1 WHERE id=%"PRId64";"
-
 #define SQL_DELETE_FILE                                                 \
-    "DELETE FROM in_tail_files WHERE id=%"PRId64";"
+    "DELETE FROM in_tail_files WHERE id=@id;"
 
 #define SQL_PRAGMA_SYNC                         \
     "PRAGMA synchronous=%i;"
