[prev in list] [next in list] [prev in thread] [next in thread]
List: ulogd
Subject: [ulogd] ulogd2 sqlite3 plugin
From: Matthias Lay <mlay () telco-tech ! de>
Date: 2011-02-14 10:33:38
Message-ID: 4D590502.4020701 () telco-tech ! de
[Download RAW message or body]
hi,
I just rewrote the sqlite plugin to work with the new plugin stack from
ulogd2.x .
it was my first work with the sqlite API so maybe someone can have a
look at it.
Regards
M. Lay
["ulogd2_sqlite.patch" (application/octet-stream)]
diff -up ./output/sqlite3/ulogd_output_SQLITE3.c.doit ./output/sqlite3/ulogd_output_SQLITE3.c
--- ./output/sqlite3/ulogd_output_SQLITE3.c.doit 2011-02-10 16:47:03.565935807 +0100
+++ ./output/sqlite3/ulogd_output_SQLITE3.c 2011-02-14 11:05:10.549936268 +0100
@@ -1,4 +1,3 @@
-#if 0
/*
* ulogd output plugin for logging to a SQLITE database
*
@@ -25,7 +24,9 @@
* at http://www.pojo.us/ulogd/
*
* 2005-02-09 Harald Welte <laforge@gnumonks.org>:
- * - port to ulogd-1.20
+ * - port to ulogd-1.20
+ * 2011-02-14 M. Lay <mlay@telco-tech.de>:
+ * - port to ulog-2.x DB_plugin
*/
#include <stdlib.h>
@@ -34,6 +35,7 @@
#include <ulogd/ulogd.h>
#include <ulogd/conffile.h>
#include <sqlite3.h>
+#include <ulogd/db.h>
#ifdef DEBUG_SQLITE3
#define DEBUGP(x, args...) fprintf(stderr, x, ## args)
@@ -41,374 +43,197 @@
#define DEBUGP(x, args...)
#endif
-struct _field {
- char name[ULOGD_MAX_KEYLEN];
- unsigned int id;
- struct _field *next;
-};
-
-/* the database handle we are using */
-static sqlite3 *dbh;
-
-/* a linked list of the fields the table has */
-static struct _field *fields;
-/* buffer for our insert statement */
-static char *stmt;
+/* length of "select * from \0" */
+#define SQLITE3_BUSY_TIMEOUT 300
-/* pointer to the final prepared statement */
-static sqlite3_stmt *p_stmt;
-/* number of statements to buffer before we commit */
-static int buffer_size;
-/* number of statements currently in the buffer */
-static int buffer_ctr;
+struct sqlite3_instance {
+ struct db_instance db_inst;
-/* our configuration directives */
-static config_entry_t db_ce = {
- .key = "db",
- .type = CONFIG_TYPE_STRING,
- .options = CONFIG_OPT_MANDATORY,
+ /* the database handle we are using */
+ sqlite3 *dbh;
};
-static config_entry_t table_ce = {
- .next = &db_ce,
- .key = "table",
- .type = CONFIG_TYPE_STRING,
- .options = CONFIG_OPT_MANDATORY,
-};
-static config_entry_t buffer_ce = {
- .next = &table_ce,
- .key = "buffer",
- .type = CONFIG_TYPE_INT,
- .options = CONFIG_OPT_MANDATORY,
+/* our configuration directives */
+static struct config_keyset kset_sqlite3 = {
+ .num_ces = DB_CE_NUM+3,
+ .ces = {
+ DB_CES,
+ {
+ .key = "db",
+ .type = CONFIG_TYPE_STRING,
+ .options = CONFIG_OPT_MANDATORY,
+ },
+ },
};
-/* our main output function, called by ulogd */
-static int _sqlite3_output(ulog_iret_t *result)
-{
- struct _field *f;
- ulog_iret_t *res;
- int col_counter;
-#ifdef IP_AS_STRING
- char *ipaddr;
- struct in_addr addr;
-#endif
-
- col_counter = 1;
- for (f = fields; f; f = f->next) {
- res = keyh_getres(f->id);
-
- if (!res) {
- ulogd_log(ULOGD_NOTICE,
- "no result for %s ?!?\n", f->name);
- }
-
- if (!res || !IS_VALID((*res))) {
- /* no result, pass a null */
- sqlite3_bind_null(p_stmt, col_counter);
- col_counter++;
- continue;
- }
-
- switch (res->type) {
- case ULOGD_RET_INT8:
- sqlite3_bind_int(p_stmt,col_counter,res->value.i8);
- break;
- case ULOGD_RET_INT16:
- sqlite3_bind_int(p_stmt,col_counter,res->value.i16);
- break;
- case ULOGD_RET_INT32:
- sqlite3_bind_int(p_stmt,col_counter,res->value.i32);
- break;
- case ULOGD_RET_INT64:
- sqlite3_bind_int64(p_stmt,col_counter,res->value.i64);
- break;
- case ULOGD_RET_UINT8:
- sqlite3_bind_int(p_stmt,col_counter,res->value.ui8);
- break;
- case ULOGD_RET_UINT16:
- sqlite3_bind_int(p_stmt,col_counter,res->value.ui16);
- break;
- case ULOGD_RET_IPADDR:
-#ifdef IP_AS_STRING
- memset(&addr, 0, sizeof(addr));
- addr.s_addr = ntohl(res->value.ui32);
- ipaddr = inet_ntoa(addr);
- sqlite3_bind_text(p_stmt,col_counter,ipaddr,strlen(ipaddr),SQLITE_STATIC);
- break;
-#endif /* IP_AS_STRING */
- /* EVIL: fallthrough when logging IP as u_int32_t */
- case ULOGD_RET_UINT32:
- sqlite3_bind_int(p_stmt,col_counter,res->value.ui32);
- break;
- case ULOGD_RET_UINT64:
- sqlite3_bind_int64(p_stmt,col_counter,res->value.ui64);
- break;
- case ULOGD_RET_BOOL:
- sqlite3_bind_int(p_stmt,col_counter,res->value.b);
- break;
- case ULOGD_RET_STRING:
- sqlite3_bind_text(p_stmt,col_counter,res->value.ptr,strlen(res->value.ptr),SQLITE_STATIC);
- break;
- default:
- ulogd_log(ULOGD_NOTICE,
- "unknown type %d for %s\n",
- res->type, res->key);
- break;
- }
-
- col_counter++;
- }
-
- /* now we have created our statement, insert it */
- if (sqlite3_step(p_stmt) == SQLITE_DONE) {
- sqlite3_reset(p_stmt);
- buffer_ctr++;
- } else {
- ulogd_log(ULOGD_ERROR, "sql error during insert: %s\n",
- sqlite3_errmsg(dbh));
- return 1;
- }
+#define db_ce(x) (x->ces[DB_CE_NUM+0])
- /* commit all of the inserts to the database, ie flush buffer */
- if (buffer_ctr >= buffer_size) {
- if (sqlite3_exec(dbh,"commit",NULL,NULL,NULL) != SQLITE_OK)
- ulogd_log(ULOGD_ERROR,"unable to commit records to db.");
-
- if (sqlite3_exec(dbh,"begin deferred",NULL,NULL,NULL) != SQLITE_OK)
- ulogd_log(ULOGD_ERROR,"unable to begin a new transaction.");
-
- buffer_ctr = 0;
- DEBUGP("committing.\n");
- }
-
- return 0;
-}
+/* length of "select * from \0" */
+#define SQLITE_SELECT_LEN 15
-#define _SQLITE3_INSERTTEMPL "insert into X (Y) values (Z)"
-/* create the static part of our insert statement */
-static int _sqlite3_createstmt(void)
+static int get_columns_sqlite3(struct ulogd_pluginstance *upi)
{
- struct _field *f;
- unsigned int size;
- char buf[ULOGD_MAX_KEYLEN];
- char *underscore;
- char *stmt_pos;
- int col_count;
- int i;
-
- if (stmt) {
- ulogd_log(ULOGD_NOTICE, "createstmt called, but stmt"
- " already existing\n");
- return 1;
- }
-
- /* caclulate the size for the insert statement */
- size = strlen(_SQLITE3_INSERTTEMPL) + strlen(table_ce.u.string);
-
- DEBUGP("initial size: %u\n", size);
-
- col_count = 0;
- for (f = fields; f; f = f->next) {
- /* we need space for the key and a comma, and a ? */
- size += strlen(f->name) + 3;
- DEBUGP("size is now %u since adding %s\n",size,f->name);
- col_count++;
- }
-
- DEBUGP("there were %d columns\n",col_count);
- DEBUGP("after calc name length: %u\n",size);
-
- ulogd_log(ULOGD_DEBUG, "allocating %u bytes for statement\n", size);
+ struct sqlite3_instance *si = (struct sqlite_instance *) upi->private;
+ sqlite3_stmt *pStmt = NULL;
+ char querybuf[SQLITE_SELECT_LEN + strlen(table_ce(upi->config_kset).u.string) + 2];
+ int column = -1;
- stmt = (char *) malloc(size);
- if (!stmt) {
- ulogd_log(ULOGD_ERROR, "OOM!\n");
+ if (!si->dbh) {
+ ulogd_log(ULOGD_ERROR, "no database handle\n");
return 1;
}
+ snprintf(querybuf, sizeof(querybuf)-1, "select * from %s;", table_ce(upi->config_kset).u.string);
- sprintf(stmt, "insert into %s (", table_ce.u.string);
- stmt_pos = stmt + strlen(stmt);
-
- for (f = fields; f; f = f->next) {
- strncpy(buf, f->name, ULOGD_MAX_KEYLEN);
- while ((underscore = strchr(buf, '.')))
- *underscore = '_';
- sprintf(stmt_pos, "%s,", buf);
- stmt_pos = stmt + strlen(stmt);
- }
-
- *(stmt_pos - 1) = ')';
-
- sprintf(stmt_pos, " values (");
- stmt_pos = stmt + strlen(stmt);
-
- for (i = 0; i < col_count - 1; i++) {
- sprintf(stmt_pos,"?,");
- stmt_pos += 2;
- }
-
- sprintf(stmt_pos, "?)");
- ulogd_log(ULOGD_DEBUG, "stmt='%s'\n", stmt);
-
- DEBUGP("about to prepare statement.\n");
-
- sqlite3_prepare(dbh,stmt,-1,&p_stmt,0);
-
- DEBUGP("statement prepared.\n");
-
- if (!p_stmt) {
- ulogd_log(ULOGD_ERROR,"unable to prepare statement");
+
+ if ( SQLITE_OK != sqlite3_prepare(si->dbh, querybuf,-1,&pStmt,0)){
+ ulogd_log(ULOGD_ERROR, "Error preparing statement\n");
return 1;
}
-
- return 0;
-}
-
-
-/* length of "select * from \0" */
-#define SQLITE_SELECT_LEN 15
-
-/* find out which columns the table has */
-static int _sqlite3_get_columns(const char *table)
-{
- char buf[ULOGD_MAX_KEYLEN];
- char query[SQLITE_SELECT_LEN + CONFIG_VAL_STRING_LEN] = "select * from \0";
- char *underscore;
- struct _field *f;
- sqlite3_stmt *schema_stmt;
- int column;
- int result;
- int id;
-
- if (!dbh)
- return 1;
-
- strncat(query,table,LINE_LEN);
-
- result = sqlite3_prepare(dbh,query,-1,&schema_stmt,0);
- if (result != SQLITE_OK)
- return 1;
- for (column = 0; column < sqlite3_column_count(schema_stmt); column++) {
- /* replace all underscores with dots */
- strncpy(buf, sqlite3_column_name(schema_stmt,column), ULOGD_MAX_KEYLEN);
+ upi->input.num_keys = sqlite3_column_count(pStmt);
+ DEBUGP("%u fields in table\n", upi->input.num_keys);
+ upi->input.keys = malloc(sizeof(struct ulogd_key) * upi->input.num_keys);
+ if (!upi->input.keys) {
+ upi->input.num_keys = 0;
+ ulogd_log(ULOGD_ERROR, "ENOMEM\n");
+ return -1;
+ }
+ memset(upi->input.keys, 0, sizeof(struct ulogd_key) *upi->input.num_keys);
+
+ for (column = 0; column < sqlite3_column_count(pStmt); column++) {
+ char buf[ULOGD_MAX_KEYLEN+1];
+ char *underscore;
+ strncpy(buf, sqlite3_column_name(pStmt,column), ULOGD_MAX_KEYLEN);
+ /* convert "_" to "." */
while ((underscore = strchr(buf, '_')))
- *underscore = '.';
-
- DEBUGP("field '%s' found: ", buf);
+ *underscore = '.';
- if (!(id = keyh_getid(buf))) {
- DEBUGP(" no keyid!\n");
- continue;
- }
-
- DEBUGP("keyid %u\n", id);
-
- /* prepend it to the linked list */
- f = (struct _field *) malloc(sizeof *f);
- if (!f) {
- ulogd_log(ULOGD_ERROR, "OOM!\n");
- return 1;
- }
- strncpy(f->name, buf, ULOGD_MAX_KEYLEN);
- f->id = id;
- f->next = fields;
- fields = f;
+ DEBUGP("field '%s' found\n", buf);
+ strncpy(upi->input.keys[column].name, buf, ULOGD_MAX_KEYLEN);
}
-
- sqlite3_finalize(schema_stmt);
+
+ sqlite3_finalize(pStmt);
return 0;
+
}
+
/**
- * make connection and select database
- * returns 0 if database failed to open.
+ * open database
+ * returns returnvalue of sqlite3_open.
*/
-static int _sqlite3_open_db(char *db_file)
+static int open_db_sqlite3(struct ulogd_pluginstance *upi)
{
- DEBUGP("opening database.\n");
- return sqlite3_open(db_file,&dbh);
+ struct sqlite3_instance *si = (struct sqlite3_instance *) upi->private;
+ int ret = -1;
+
+ ret = sqlite3_open(db_ce(upi->config_kset).u.string, &si->dbh);
+
+ DEBUGP("opening database. ret=%i\n",ret);
+
+ //sqlite3_busy_timeout(si->dbh, SQLITE3_BUSY_TIMEOUT);
+ return ret;
}
/* give us an opportunity to close the database down properly */
-static void _sqlite3_fini(void)
+static int close_db_sqlite3(struct ulogd_pluginstance *upi)
{
- DEBUGP("cleaning up db connection\n");
+ struct sqlite3_instance *si = (struct sqlite3_instance *) upi->private;
+ int result = 0;
- /* free up our prepared statements so we can close the db */
- if (p_stmt) {
- sqlite3_finalize(p_stmt);
- DEBUGP("prepared statement finalized\n");
- }
+ DEBUGP("cleaning up db connection\n");
- if (dbh) {
- int result;
+ if (si->dbh) {
/* flush the remaining insert statements to the database. */
- result = sqlite3_exec(dbh,"commit",NULL,NULL,NULL);
+ result = sqlite3_exec(si->dbh,"commit",NULL,NULL,NULL);
- if (result != SQLITE_OK)
+ if (result != SQLITE_OK){
ulogd_log(ULOGD_ERROR,"unable to commit remaining records to db.");
+ }
- sqlite3_close(dbh);
+ result = sqlite3_close(si->dbh);
DEBUGP("database file closed\n");
}
+ return result;
}
-#define _SQLITE3_BUSY_TIMEOUT 300
-static int _sqlite3_init(void)
+
+static int execute_sqlite3(struct ulogd_pluginstance *upi,
+ const char *stmt, unsigned int len)
{
- /* have the opts parsed */
- config_parse_file("SQLITE3", &buffer_ce);
+ struct sqlite3_instance *si = (struct sqlite3_instance *) upi->private;
+ int ret;
- if (_sqlite3_open_db(db_ce.u.string)) {
- ulogd_log(ULOGD_ERROR, "can't open the database file\n");
- return 1;
- }
+ ret = sqlite3_exec(si->dbh, stmt, NULL, NULL, NULL);
- /* set the timeout so that we don't automatically fail
- * if the table is busy. */
- sqlite3_busy_timeout(dbh, _SQLITE3_BUSY_TIMEOUT);
-
- /* read the fieldnames to know which values to insert */
- if (_sqlite3_get_columns(table_ce.u.string)) {
- ulogd_log(ULOGD_ERROR, "unable to get sqlite columns\n");
+ if (SQLITE_OK != ret){
+ DEBUGP("cant create transaction. statement was %s\n", stmt);
return 1;
}
+ return 0;
+}
- /* initialize our buffer size and counter */
- buffer_size = buffer_ce.u.value;
- buffer_ctr = 0;
+static int escape_string_sqlite3(struct ulogd_pluginstance *upi,
+ char *dst, const char *src, unsigned int len)
+{
+ /* do we need it? */
+ return 0;
+}
- DEBUGP("Have a buffer size of : %d\n", buffer_size);
- if (sqlite3_exec(dbh,"begin deferred",NULL,NULL,NULL) != SQLITE_OK)
- ulogd_log(ULOGD_ERROR,"can't create a new transaction\n");
+static struct db_driver db_driver_sqlite3 = {
+ .get_columns = &get_columns_sqlite3,
+ .open_db = &open_db_sqlite3,
+ .close_db = &close_db_sqlite3,
+ .escape_string = &escape_string_sqlite3,
+ .execute = &execute_sqlite3,
+};
- /* create and prepare the actual insert statement */
- _sqlite3_createstmt();
+static int configure_sqlite3(struct ulogd_pluginstance *upi,
+ struct ulogd_pluginstance_stack *stack)
+{
+ struct db_instance *di = (struct db_instance *) &upi->private;
- return 0;
-}
+ di->driver = &db_driver_sqlite3;
+ DEBUGP("configuring_sqlite3 plugin\n");
-static ulog_output_t _sqlite3_plugin = {
- .name = "sqlite3",
- .output = &_sqlite3_output,
- .init = &_sqlite3_init,
- .fini = &_sqlite3_fini,
+ return ulogd_db_configure(upi, stack);
+}
+
+static struct ulogd_plugin plugin_sqlite3 = {
+ .name = "SQLITE3",
+ .input = {
+ .keys = NULL,
+ .num_keys = 0,
+ .type = ULOGD_DTYPE_PACKET | ULOGD_DTYPE_FLOW,
+
+ },
+ .output = {
+ .type = ULOGD_DTYPE_SINK,
+ },
+ .config_kset = &kset_sqlite3,
+ .priv_size = sizeof(struct sqlite3_instance),
+ .configure = &configure_sqlite3,
+ .start = &ulogd_db_start,
+ .stop = &ulogd_db_stop,
+ .signal = &ulogd_db_signal,
+ .interp = &ulogd_db_interp,
+ .version = ULOGD_VERSION,
};
-void _init(void)
+void __attribute__ ((constructor)) init(void);
+
+void init(void)
{
- register_output(&_sqlite3_plugin);
+ ulogd_register_plugin(&plugin_sqlite3);
}
-#endif
[prev in list] [next in list] [prev in thread] [next in thread]
Configure |
About |
News |
Add a list |
Sponsored by KoreLogic