sqlite3: initial support extracted from Holger Eitzenberger's work

This patch imports the sqlite3 support from Holger.

I applied folded the following patches, that Harald passed to me
during the last Netfilter workshop, into one:

0001-ulog2-sqlite-port-to-v2.diff.patch
0002-ulog2-sqlite-prepare-fix.diff.patch
0003-ulogd-sqlite-new-logging.diff.patch
0004-ulogd-accounting-add-flow-start-day.patch
0005-ulogd-sqlite3-err-codes-fix.dif.patch
0006-ulogd-sqlite3-tbl-corrupt-fix.diff.patch
0007-ulogd-sqlite3-handle-schema-change.diff.patch
0008-SQLITE3-count-instead-of-log-table-busy-messags.patch
0009-ulogd-SQLITE-Added-flowstartsec.diff.patch

Harald passed them to me with no description, so applying them
separately does not provide more information.

I'll start adding patches on top on these so Holger can get in
sync with my work. This also can help him to take my patches
and to integrate them to his tree.

Signed-off-by: Pablo Neira Ayuso <pablo@netfilter.org>

1 files changed, 350 insertions, 259 deletions

diff --git a/output/sqlite3/ulogd_output_SQLITE3.c b/output/sqlite3/ulogd_output_SQLITE3.c
index ba1b594..0de8145 100644
--- a/output/sqlite3/ulogd_output_SQLITE3.c
+++ b/output/sqlite3/ulogd_output_SQLITE3.c
@@ -1,4 +1,3 @@
-#if 0
 /*
  * ulogd output plugin for logging to a SQLITE database
  *
@@ -26,6 +25,9 @@
  *
  *  2005-02-09 Harald Welte <laforge@gnumonks.org>:
  *  	- port to ulogd-1.20 
+ *
+ *  2006-10-09 Holger Eitzenberger <holger@my-eitzenberger.de>
+ *  	- port to ulogd-2.00
  */
 
 #include <stdlib.h>
@@ -34,381 +36,470 @@
 #include <ulogd/ulogd.h>
 #include <ulogd/conffile.h>
 #include <sqlite3.h>
+#include <sys/queue.h>
+
+#define CFG_BUFFER_DEFAULT		10
+
+/* number of colums we have (really should be configurable) */
+#define DB_NUM_COLS	11
 
-#ifdef DEBUG_SQLITE3
+#if 0
 #define DEBUGP(x, args...)	fprintf(stderr, x, ## args)
 #else
 #define DEBUGP(x, args...)
 #endif
 
-struct _field {
+struct field {
+	TAILQ_ENTRY(field) link;
 	char name[ULOGD_MAX_KEYLEN];
-	unsigned int id;
-	struct _field *next;
+	struct ulogd_key *key;
 };
 
-/* the database handle we are using */
-static sqlite3 *dbh;
+TAILQ_HEAD(field_lh, field);
+
+#define tailq_for_each(pos, head, link) \
+        for (pos = (head).tqh_first; pos != NULL; pos = pos->link.tqe_next)
 
-/* a linked list of the fields the table has */
-static struct _field *fields;
 
-/* buffer for our insert statement */
-static char *stmt;
+struct sqlite3_priv {
+	sqlite3 *dbh;				/* database handle we are using */
+	struct field_lh fields;
+	char *stmt;
+	sqlite3_stmt *p_stmt;
+	int buffer_size;
+	int buffer_curr;
+	struct {
+		unsigned err_tbl_busy;	/* "Table busy" */
+	} stats;
+};
 
-/* pointer to the final prepared statement */
-static sqlite3_stmt *p_stmt;
+static struct config_keyset sqlite3_kset = {
+	.num_ces = 3,
+	.ces = {
+		{
+			.key = "db",
+			.type = CONFIG_TYPE_STRING,
+			.options = CONFIG_OPT_MANDATORY,
+		},
+		{
+			.key = "table",
+			.type = CONFIG_TYPE_STRING,
+			.options = CONFIG_OPT_MANDATORY,
+		},
+		{
+			.key = "buffer",
+			.type = CONFIG_TYPE_INT,
+			.options = CONFIG_OPT_NONE,
+			.u.value = CFG_BUFFER_DEFAULT,
+		},
+	},
+};
 
-/* number of statements to buffer before we commit */
-static int buffer_size;
+#define db_ce(pi)		(pi)->config_kset->ces[0].u.string
+#define table_ce(pi)	(pi)->config_kset->ces[1].u.string
+#define buffer_ce(pi)	(pi)->config_kset->ces[2].u.value
 
-/* number of statements currently in the buffer */
-static int buffer_ctr;
+/* forward declarations */
+static int sqlite3_createstmt(struct ulogd_pluginstance *);
 
-/* our configuration directives */
-static config_entry_t db_ce = { 
-	.key = "db", 
-	.type = CONFIG_TYPE_STRING,
-	.options = CONFIG_OPT_MANDATORY,
-};
 
-static config_entry_t table_ce = { 
-	.next = &db_ce, 
-	.key = "table",
-	.type = CONFIG_TYPE_STRING,
-	.options = CONFIG_OPT_MANDATORY,
-};
+static int
+add_row(struct ulogd_pluginstance *pi)
+{
+	struct sqlite3_priv *priv = (void *)pi->private;
+	int ret;
+
+	ret = sqlite3_step(priv->p_stmt);
+	if (ret == SQLITE_DONE)
+		priv->buffer_curr++;
+	else if (ret == SQLITE_BUSY)
+		priv->stats.err_tbl_busy++;
+	else if (ret == SQLITE_ERROR) {
+		ret = sqlite3_finalize(priv->p_stmt);
+		priv->p_stmt = NULL;
+
+		if (ret == SQLITE_SCHEMA)
+			sqlite3_createstmt(pi);
+		else {
+			ulogd_log(ULOGD_ERROR, "SQLITE3: step: %s\n",
+					  sqlite3_errmsg(priv->dbh));
+			goto err_reset;
+		}
+	}
+
+	ret = sqlite3_reset(priv->p_stmt);
+
+	return 0;
+
+ err_reset:
+	sqlite3_reset(priv->p_stmt);
+
+	return -1;
+}
 
-static config_entry_t buffer_ce = { 
-	.next = &table_ce,
-	.key = "buffer",
-	.type = CONFIG_TYPE_INT,
-	.options = CONFIG_OPT_MANDATORY,
-};
 
 /* our main output function, called by ulogd */
-static int _sqlite3_output(ulog_iret_t *result)
+static int
+sqlite3_interp(struct ulogd_pluginstance *pi)
 {
-	struct _field *f;
-	ulog_iret_t *res;
-	int col_counter;
-#ifdef IP_AS_STRING
-	char *ipaddr;
-	struct in_addr addr;
-#endif
+	struct sqlite3_priv *priv = (void *)pi->private;
+	struct field *f;
+	int ret, i = 1;
 
-	col_counter = 1;
-	for (f = fields; f; f = f->next) {
-		res = keyh_getres(f->id);
+	tailq_for_each(f, priv->fields, link) {
+		struct ulogd_key *k_ret = f->key->u.source;
 
-		if (!res) {
-			ulogd_log(ULOGD_NOTICE,
-				"no result for %s ?!?\n", f->name);
+		if (f->key == NULL || !IS_VALID(*k_ret)) {
+			sqlite3_bind_null(priv->p_stmt, i);
+			goto next_field;
 		}
+
+		switch (f->key->type) {
+		case ULOGD_RET_INT8:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.i8);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_INT16:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.i16);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_INT32:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.i32);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_INT64:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.i64);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+			
+		case ULOGD_RET_UINT8:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.ui8);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
 			
-		if (!res || !IS_VALID((*res))) {
-			/* no result, pass a null */
-			sqlite3_bind_null(p_stmt, col_counter);
-			col_counter++;
-			continue;
+		case ULOGD_RET_UINT16:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.ui16);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_UINT32:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.ui32);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_IPADDR:
+		case ULOGD_RET_UINT64:
+			ret = sqlite3_bind_int64(priv->p_stmt, i, k_ret->u.value.ui64);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_BOOL:
+			ret = sqlite3_bind_int(priv->p_stmt, i, k_ret->u.value.b);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		case ULOGD_RET_STRING:
+			ret = sqlite3_bind_text(priv->p_stmt, i, k_ret->u.value.ptr,
+									strlen(k_ret->u.value.ptr), SQLITE_STATIC);
+			if (ret != SQLITE_OK)
+				goto err_bind;
+			break;
+
+		default:
+			ulogd_log(ULOGD_NOTICE, "unknown type %d for %s\n",
+					  f->key->type, f->key->name);
 		}
-		
-		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;
+	next_field:
+		i++;
 	}
 
-	/* 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.");
+	if (add_row(pi) < 0)
+		return ULOGD_IRET_ERR;
 
-		buffer_ctr = 0;
-		DEBUGP("committing.\n");
-	}
+	return ULOGD_IRET_OK;
 
-	return 0;
+ err_bind:
+	ulogd_log(ULOGD_ERROR, "SQLITE: bind: %s\n", sqlite3_errmsg(priv->dbh));
+	
+	return ULOGD_IRET_ERR;
 }
 
 #define _SQLITE3_INSERTTEMPL   "insert into X (Y) values (Z)"
 
 /* create the static part of our insert statement */
-static int _sqlite3_createstmt(void)
+static int
+sqlite3_createstmt(struct ulogd_pluginstance *pi)
 {
-	struct _field *f;
-	unsigned int size;
+	struct sqlite3_priv *priv = (void *)pi->private;
+	struct field *f;
 	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);
+	int i, cols = 0;
 
-	DEBUGP("initial size: %u\n", size);
+	if (priv->stmt != NULL)
+		free(priv->stmt);
 
-	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++;
+	if ((priv->stmt = calloc(1, 1024)) == NULL) {
+		ulogd_log(ULOGD_ERROR, "SQLITE3: out of memory\n");
+		return -1;
 	}
 
-	DEBUGP("there were %d columns\n",col_count);
-	DEBUGP("after calc name length: %u\n",size);
+	sprintf(priv->stmt, "insert into %s (", table_ce(pi));
+	stmt_pos = priv->stmt + strlen(priv->stmt);
 
-	ulogd_log(ULOGD_DEBUG, "allocating %u bytes for statement\n", size);
+	tailq_for_each(f, priv->fields, link) {
+		strncpy(buf, f->name, ULOGD_MAX_KEYLEN);
 
-	stmt = (char *) malloc(size);
-
-	if (!stmt) {
-		ulogd_log(ULOGD_ERROR, "OOM!\n");
-		return 1;
-	}
-
-	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 = priv->stmt + strlen(priv->stmt);
+
+		cols++;
 	}
 
 	*(stmt_pos - 1) = ')';
 
 	sprintf(stmt_pos, " values (");
-	stmt_pos = stmt + strlen(stmt);
+	stmt_pos = priv->stmt + strlen(priv->stmt);
 
-	for (i = 0; i < col_count - 1; i++) {
+	for (i = 0; i < cols - 1; i++) {
 		sprintf(stmt_pos,"?,");
 		stmt_pos += 2;
 	}
 
 	sprintf(stmt_pos, "?)");
-	ulogd_log(ULOGD_DEBUG, "stmt='%s'\n", stmt);
+	ulogd_log(ULOGD_DEBUG, "%s: stmt='%s'\n", pi->id, priv->stmt);
 
 	DEBUGP("about to prepare statement.\n");
 
-	sqlite3_prepare(dbh,stmt,-1,&p_stmt,0);
+	sqlite3_prepare(priv->dbh, priv->stmt, -1, &priv->p_stmt, 0);
+	if (priv->p_stmt == NULL) {
+		ulogd_log(ULOGD_ERROR, "SQLITE3: prepare: %s\n",
+				  sqlite3_errmsg(priv->dbh));
+		return 1;
+	}
 
 	DEBUGP("statement prepared.\n");
 
-	if (!p_stmt) {
-		ulogd_log(ULOGD_ERROR,"unable to prepare statement");
-		return 1;
+	return 0;
+}
+
+
+static struct ulogd_key *
+ulogd_find_key(struct ulogd_pluginstance *pi, const char *name)
+{
+	int i;
+
+	for (i = 0; i < pi->input.num_keys; i++) {
+		if (strcmp(pi->input.keys[i].name, name) == 0)
+			return &pi->input.keys[i];
+	}
+
+	return NULL;
+}
+
+#define SELECT_ALL_STR			"select * from "
+#define SELECT_ALL_LEN			sizeof(SELECT_ALL_STR)
+
+static int
+db_count_cols(struct ulogd_pluginstance *pi, sqlite3_stmt **stmt)
+{
+	struct sqlite3_priv *priv = (void *)pi->private;
+	char query[SELECT_ALL_LEN + CONFIG_VAL_STRING_LEN] = SELECT_ALL_STR;
+
+	strncat(query, table_ce(pi), LINE_LEN);
+
+	if (sqlite3_prepare(priv->dbh, query, -1, stmt, 0) != SQLITE_OK)
+		return -1;
+
+	return sqlite3_column_count(*stmt);
+}
+
+
+/* FIXME make this configurable */
+#define SQL_CREATE_STR \
+		"create table daily(ip_saddr integer, ip_daddr integer, " \
+		"ip_protocol integer, l4_dport integer, raw_in_pktlen integer, " \
+		"raw_in_pktcount integer, raw_out_pktlen integer, " \
+		"raw_out_pktcount integer, flow_start_day integer, " \
+		"flow_start_sec integer, flow_duration integer)"
+
+static int
+db_create_tbl(struct ulogd_pluginstance *pi)
+{
+	struct sqlite3_priv *priv = (void *)pi->private;
+	char *errmsg;
+	int ret;
+
+	sqlite3_exec(priv->dbh, "drop table daily", NULL, NULL, NULL);
+
+	ret = sqlite3_exec(priv->dbh, SQL_CREATE_STR, NULL, NULL, &errmsg);
+	if (ret != SQLITE_OK) {
+		ulogd_log(ULOGD_ERROR, "SQLITE3: create table: %s\n", errmsg);
+		sqlite3_free(errmsg);
+
+		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)
+/* initialize DB, possibly creating it */
+static int
+sqlite3_init_db(struct ulogd_pluginstance *pi)
 {
+	struct sqlite3_priv *priv = (void *)pi->private;
 	char buf[ULOGD_MAX_KEYLEN];
-	char query[SQLITE_SELECT_LEN + CONFIG_VAL_STRING_LEN] = "select * from \0";
 	char *underscore;
-	struct _field *f;
+	struct field *f;
 	sqlite3_stmt *schema_stmt;
-	int column;
-	int result;
-	int id;
+	int col, num_cols;
 
-	if (!dbh)
-		return 1;
+	if (priv->dbh == NULL)
+		return -1;
 
-	strncat(query,table,LINE_LEN);
-	
-	result = sqlite3_prepare(dbh,query,-1,&schema_stmt,0);
-	
-	if (result != SQLITE_OK)
-		return 1;
+	num_cols = db_count_cols(pi, &schema_stmt);
+	if (num_cols != DB_NUM_COLS) {
+		if (db_create_tbl(pi) < 0)
+			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);
-		while ((underscore = strchr(buf, '_')))
-			*underscore = '.';
+		num_cols = db_count_cols(pi, &schema_stmt);
+	}
 
-		DEBUGP("field '%s' found: ", buf);
+	for (col = 0; col < num_cols; col++) {
+		strncpy(buf, sqlite3_column_name(schema_stmt, col), ULOGD_MAX_KEYLEN);
 
-		if (!(id = keyh_getid(buf))) {
-			DEBUGP(" no keyid!\n");
-			continue;
-		}
+		/* replace all underscores with dots */
+		while ((underscore = strchr(buf, '_')) != NULL)
+			*underscore = '.';
 
-		DEBUGP("keyid %u\n", id);
+		DEBUGP("field '%s' found\n", buf);
 
 		/* prepend it to the linked list */
-		f = (struct _field *) malloc(sizeof *f);
-		if (!f) {
-			ulogd_log(ULOGD_ERROR, "OOM!\n");
-			return 1;
+		if ((f = calloc(1, sizeof(struct field))) == NULL) {
+			ulogd_log(ULOGD_ERROR, "SQLITE3: out of memory\n");
+			return -1;
 		}
 		strncpy(f->name, buf, ULOGD_MAX_KEYLEN);
-		f->id = id;
-		f->next = fields;
-		fields = f;	
+
+		if ((f->key = ulogd_find_key(pi, buf)) == NULL)
+			return -1;
+
+		TAILQ_INSERT_TAIL(&priv->fields, f, link);
 	}
 
 	sqlite3_finalize(schema_stmt);
+
 	return 0;
 }
 
-/** 
- * make connection and select database 
- * returns 0 if database failed to open.
- */
-static int _sqlite3_open_db(char *db_file)
-{
-	DEBUGP("opening database.\n");
-	return sqlite3_open(db_file,&dbh);
-}
+#define SQLITE3_BUSY_TIMEOUT 300
 
-/* give us an opportunity to close the database down properly */
-static void _sqlite3_fini(void)
+static int
+sqlite3_configure(struct ulogd_pluginstance *pi,
+				  struct ulogd_pluginstance_stack *stack)
 {
-	DEBUGP("cleaning up db connection\n");
+	/* struct sqlite_priv *priv = (void *)pi->private; */
 
-	/* free up our prepared statements so we can close the db */
-	if (p_stmt) {
-		sqlite3_finalize(p_stmt);
-		DEBUGP("prepared statement finalized\n");
-	}
+	config_parse_file(pi->id, pi->config_kset);
 
-	if (dbh) {
-		int result;
-		/* flush the remaining insert statements to the database. */
-		result = sqlite3_exec(dbh,"commit",NULL,NULL,NULL);
+	if (ulogd_wildcard_inputkeys(pi) < 0)
+		return -1;
 
-		if (result != SQLITE_OK)
-			ulogd_log(ULOGD_ERROR,"unable to commit remaining records to db.");
+	DEBUGP("%s: db='%s' table='%s'\n", pi->id, db_ce(pi), table_ce(pi));
 
-		sqlite3_close(dbh);
-		DEBUGP("database file closed\n");
-	}
+	return 0;
 }
 
-#define _SQLITE3_BUSY_TIMEOUT 300
-
-static int _sqlite3_init(void)
+static int
+sqlite3_start(struct ulogd_pluginstance *pi)
 {
-	/* have the opts parsed */
-	config_parse_file("SQLITE3", &buffer_ce);
+	struct sqlite3_priv *priv = (void *)pi->private;
 
-	if (_sqlite3_open_db(db_ce.u.string)) {
-		ulogd_log(ULOGD_ERROR, "can't open the database file\n");
-		return 1;
+	TAILQ_INIT(&priv->fields);
+
+	if (sqlite3_open(db_ce(pi), &priv->dbh) != SQLITE_OK) {
+		ulogd_log(ULOGD_ERROR, "SQLITE3: %s\n", sqlite3_errmsg(priv->dbh));
+		return -1;
 	}
 
 	/* set the timeout so that we don't automatically fail
-         * if the table is busy. */
-	sqlite3_busy_timeout(dbh, _SQLITE3_BUSY_TIMEOUT);
+	   if the table is busy */
+	sqlite3_busy_timeout(priv->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");
-		return 1;
-	}
+	if (sqlite3_init_db(pi) < 0)
+		return -1;
 
 	/* initialize our buffer size and counter */
-	buffer_size = buffer_ce.u.value;
-	buffer_ctr = 0;
+	priv->buffer_size = buffer_ce(pi);
+	priv->buffer_curr = 0;
 
-	DEBUGP("Have a buffer size of : %d\n", buffer_size);
+	/* create and prepare the actual insert statement */
+	sqlite3_createstmt(pi);
 
-	if (sqlite3_exec(dbh,"begin deferred",NULL,NULL,NULL) != SQLITE_OK)
-		ulogd_log(ULOGD_ERROR,"can't create a new transaction\n");
+	return 0;
+}
 
-	/* create and prepare the actual insert statement */
-	_sqlite3_createstmt();
+/* give us an opportunity to close the database down properly */
+static int
+sqlite3_stop(struct ulogd_pluginstance *pi)
+{
+	struct sqlite3_priv *priv = (void *)pi->private;
+
+	/* free up our prepared statements so we can close the db */
+	if (priv->p_stmt) {
+		sqlite3_finalize(priv->p_stmt);
+		DEBUGP("prepared statement finalized\n");
+	}
+
+	if (priv->dbh == NULL)
+		return -1;
+
+	sqlite3_close(priv->dbh);
 
 	return 0;
 }
 
-static ulog_output_t _sqlite3_plugin = { 
-	.name = "sqlite3", 
-	.output = &_sqlite3_output, 
-	.init = &_sqlite3_init,
-	.fini = &_sqlite3_fini,
+static struct ulogd_plugin sqlite3_plugin = { 
+	.name = "SQLITE3", 
+	.input = {
+		.type = ULOGD_DTYPE_PACKET | ULOGD_DTYPE_FLOW,
+	},
+	.output = {
+		.type = ULOGD_DTYPE_SINK,
+	},
+	.config_kset = &sqlite3_kset,
+	.priv_size = sizeof(struct sqlite3_priv),
+	.configure = sqlite3_configure,
+	.start = sqlite3_start,
+	.stop = sqlite3_stop,
+	.interp = sqlite3_interp,
+	.version = ULOGD_VERSION,
 };
 
-void _init(void) 
+static void init(void) __attribute__((constructor));
+
+static void
+init(void) 
 {
-	register_output(&_sqlite3_plugin);
+	ulogd_register_plugin(&sqlite3_plugin);
 }
-
-#endif