Merge branch 'master' of git://anonscm.debian.org/collab-maint/rsyslog

Conflicts:
	debian/changelog
	debian/patches/series
	debian/rules

2 files changed, 332 insertions, 80 deletions

diff --git a/plugins/imjournal/Makefile.in b/plugins/imjournal/Makefile.in
index d181e2e..5644fa5 100644
--- a/plugins/imjournal/Makefile.in
+++ b/plugins/imjournal/Makefile.in
@@ -154,7 +154,6 @@ GREP = @GREP@
 GSS_LIBS = @GSS_LIBS@
 GUARDTIME_CFLAGS = @GUARDTIME_CFLAGS@
 GUARDTIME_LIBS = @GUARDTIME_LIBS@
-HAVE_LIBGCRYPT_CONFIG = @HAVE_LIBGCRYPT_CONFIG@
 HAVE_MYSQL_CONFIG = @HAVE_MYSQL_CONFIG@
 HAVE_ORACLE_CONFIG = @HAVE_ORACLE_CONFIG@
 HAVE_PGSQL_CONFIG = @HAVE_PGSQL_CONFIG@
@@ -175,14 +174,15 @@ LEXLIB = @LEXLIB@
 LEX_OUTPUT_ROOT = @LEX_OUTPUT_ROOT@
 LIBDBI_CFLAGS = @LIBDBI_CFLAGS@
 LIBDBI_LIBS = @LIBDBI_LIBS@
-LIBEE_CFLAGS = @LIBEE_CFLAGS@
-LIBEE_LIBS = @LIBEE_LIBS@
 LIBESTR_CFLAGS = @LIBESTR_CFLAGS@
 LIBESTR_LIBS = @LIBESTR_LIBS@
 LIBGCRYPT_CFLAGS = @LIBGCRYPT_CFLAGS@
+LIBGCRYPT_CONFIG = @LIBGCRYPT_CONFIG@
 LIBGCRYPT_LIBS = @LIBGCRYPT_LIBS@
 LIBLOGGING_CFLAGS = @LIBLOGGING_CFLAGS@
 LIBLOGGING_LIBS = @LIBLOGGING_LIBS@
+LIBLOGGING_STDLOG_CFLAGS = @LIBLOGGING_STDLOG_CFLAGS@
+LIBLOGGING_STDLOG_LIBS = @LIBLOGGING_STDLOG_LIBS@
 LIBLOGNORM_CFLAGS = @LIBLOGNORM_CFLAGS@
 LIBLOGNORM_LIBS = @LIBLOGNORM_LIBS@
 LIBM = @LIBM@
@@ -207,6 +207,8 @@ NM = @NM@
 NMEDIT = @NMEDIT@
 OBJDUMP = @OBJDUMP@
 OBJEXT = @OBJEXT@
+OPENSSL_CFLAGS = @OPENSSL_CFLAGS@
+OPENSSL_LIBS = @OPENSSL_LIBS@
 ORACLE_CFLAGS = @ORACLE_CFLAGS@
 ORACLE_LIBS = @ORACLE_LIBS@
 OTOOL = @OTOOL@
@@ -388,22 +390,25 @@ distclean-compile:
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/imjournal_la-imjournal.Plo@am__quote@
 
 .c.o:
-@am__fastdepCC_TRUE@	$(AM_V_CC)$(COMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $<
-@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Po
+@am__fastdepCC_TRUE@	$(AM_V_CC)depbase=`echo $@ | sed 's|[^/]*$$|$(DEPDIR)/&|;s|\.o$$||'`;\
+@am__fastdepCC_TRUE@	$(COMPILE) -MT $@ -MD -MP -MF $$depbase.Tpo -c -o $@ $< &&\
+@am__fastdepCC_TRUE@	$(am__mv) $$depbase.Tpo $$depbase.Po
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	$(AM_V_CC)source='$<' object='$@' libtool=no @AMDEPBACKSLASH@
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
-@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(COMPILE) -c $<
+@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(COMPILE) -c -o $@ $<
 
 .c.obj:
-@am__fastdepCC_TRUE@	$(AM_V_CC)$(COMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ `$(CYGPATH_W) '$<'`
-@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Po
+@am__fastdepCC_TRUE@	$(AM_V_CC)depbase=`echo $@ | sed 's|[^/]*$$|$(DEPDIR)/&|;s|\.obj$$||'`;\
+@am__fastdepCC_TRUE@	$(COMPILE) -MT $@ -MD -MP -MF $$depbase.Tpo -c -o $@ `$(CYGPATH_W) '$<'` &&\
+@am__fastdepCC_TRUE@	$(am__mv) $$depbase.Tpo $$depbase.Po
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	$(AM_V_CC)source='$<' object='$@' libtool=no @AMDEPBACKSLASH@
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
-@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(COMPILE) -c `$(CYGPATH_W) '$<'`
+@am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(COMPILE) -c -o $@ `$(CYGPATH_W) '$<'`
 
 .c.lo:
-@am__fastdepCC_TRUE@	$(AM_V_CC)$(LTCOMPILE) -MT $@ -MD -MP -MF $(DEPDIR)/$*.Tpo -c -o $@ $<
-@am__fastdepCC_TRUE@	$(AM_V_at)$(am__mv) $(DEPDIR)/$*.Tpo $(DEPDIR)/$*.Plo
+@am__fastdepCC_TRUE@	$(AM_V_CC)depbase=`echo $@ | sed 's|[^/]*$$|$(DEPDIR)/&|;s|\.lo$$||'`;\
+@am__fastdepCC_TRUE@	$(LTCOMPILE) -MT $@ -MD -MP -MF $$depbase.Tpo -c -o $@ $< &&\
+@am__fastdepCC_TRUE@	$(am__mv) $$depbase.Tpo $$depbase.Plo
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	$(AM_V_CC)source='$<' object='$@' libtool=yes @AMDEPBACKSLASH@
 @AMDEP_TRUE@@am__fastdepCC_FALSE@	DEPDIR=$(DEPDIR) $(CCDEPMODE) $(depcomp) @AMDEPBACKSLASH@
 @am__fastdepCC_FALSE@	$(AM_V_CC@am__nodep@)$(LTCOMPILE) -c -o $@ $<
diff --git a/plugins/imjournal/imjournal.c b/plugins/imjournal/imjournal.c
index 2af1958..6a97b25 100644
--- a/plugins/imjournal/imjournal.c
+++ b/plugins/imjournal/imjournal.c
@@ -3,7 +3,7 @@
  * To test under Linux:
  * emmit log message into systemd journal
  *
- * Copyright (C) 2008-2012 Adiscon GmbH
+ * Copyright (C) 2008-2013 Adiscon GmbH
  *
  * This file is part of rsyslog.
  *
@@ -30,7 +30,10 @@
 #include <ctype.h>
 #include <stdlib.h>
 #include <time.h>
+#include <sys/poll.h>
 #include <sys/socket.h>
+#include <errno.h>
+#include <systemd/sd-journal.h>
 
 #include "dirty.h"
 #include "cfsysline.h"
@@ -43,8 +46,9 @@
 #include "glbl.h"
 #include "prop.h"
 #include "errmsg.h"
+#include "srUtils.h"
 #include "unicode-helper.h"
-#include <systemd/sd-journal.h>
+#include "ratelimit.h"
 
 MODULE_TYPE_INPUT
 MODULE_TYPE_NOKEEP
@@ -61,12 +65,24 @@ DEFobjCurrIf(errmsg)
 static struct configSettings_s {
 	char *stateFile;
 	int iPersistStateInterval;
+	int ratelimitInterval;
+	int ratelimitBurst;
+	int bIgnorePrevious;
+	int iDfltSeverity;
+	int iDfltFacility;
 } cs;
 
-/* module-gloval parameters */
+static rsRetVal facilityHdlr(uchar **pp, void *pVal);
+
+/* module-global parameters */
 static struct cnfparamdescr modpdescr[] = {
 	{ "statefile", eCmdHdlrGetWord, 0 },
-	{ "persiststateinterval", eCmdHdlrInt, 0 }
+	{ "ratelimit.interval", eCmdHdlrInt, 0 },
+	{ "ratelimit.burst", eCmdHdlrInt, 0 },
+	{ "persiststateinterval", eCmdHdlrInt, 0 },
+	{ "ignorepreviousmessages", eCmdHdlrBinary, 0 },
+	{ "defaultseverity", eCmdHdlrSeverity, 0 },
+	{ "defaultfacility", eCmdHdlrString, 0 }
 };
 static struct cnfparamblk modpblk =
 	{ CNFPARAMBLK_VERSION,
@@ -75,14 +91,51 @@ static struct cnfparamblk modpblk =
 	};
 
 #define DFLT_persiststateinterval 10
+#define DFLT_SEVERITY pr2fac(LOG_NOTICE)
+#define DFLT_FACILITY pri2sev(LOG_USER)
 
-static int bLegacyCnfModGlobalsPermitted = 0;/* are legacy module-global config parameters permitted? */
+static int bLegacyCnfModGlobalsPermitted = 1;/* are legacy module-global config parameters permitted? */
 
 static prop_t *pInputName = NULL;	/* there is only one global inputName for all messages generated by this module */
 static prop_t *pLocalHostIP = NULL;	/* a pseudo-constant propterty for 127.0.0.1 */
 
+static ratelimit_t *ratelimiter = NULL;
 static sd_journal *j;
 
+
+/* ugly workaround to handle facility numbers; values
+ * derived from names need to be eight times smaller,
+ * i.e.: 0..23
+ */
+static rsRetVal facilityHdlr(uchar **pp, void *pVal)
+{
+	DEFiRet;
+	char *p;
+
+	skipWhiteSpace(pp);
+	p = (char *) *pp;
+
+	if (isdigit((int) *p)) {
+		*((int *) pVal) = (int) strtol(p, (char **) pp, 10);
+	} else {
+		int len;
+		syslogName_t *c;
+
+		for (len = 0; p[len] && !isspace((int) p[len]); len++)
+			/* noop */;
+		for (c = syslogFacNames; c->c_name; c++) {
+			if (!strncasecmp(p, (char *) c->c_name, len)) {
+				*((int *) pVal) = pri2fac(c->c_val);
+				break;
+			}
+		}
+		*pp += len;
+	}
+
+	RETiRet;
+}
+
+
 /* enqueue the the journal message into the message queue.
  * The provided msg string is not freed - thus must be done
  * by the caller.
@@ -118,7 +171,7 @@ enqMsg(uchar *msg, uchar *pszTag, int iFacility, int iSeverity, struct timeval *
 		msgAddJSON(pMsg, (uchar*)"!", json);
 	}
 
-	CHKiRet(submitMsg2(pMsg));
+	CHKiRet(ratelimitAddMsg(ratelimiter, NULL, pMsg));
 
 finalize_it:
 	RETiRet;
@@ -136,16 +189,19 @@ readjournal() {
 	uint64_t timestamp;
 
 	struct json_object *json = NULL;
+	int r;
 
 	/* Information from messages */
 	char *message;
+	char *sys_pid;
 	char *sys_iden;
 	char *sys_iden_help;
 
 	const void *get;
+	const void *pidget;
 	char *parse;
-	char *get2;
 	size_t length;
+	size_t pidlength;
 
 	const void *equal_sign;
 	struct json_object *jval;
@@ -155,57 +211,57 @@ readjournal() {
 
 	long prefixlen = 0;
 
-	int priority = 0;
-	int facility = 0;
-
-	/* Get next journal message, if there is none, wait a second */
-	if (sd_journal_next(j) == 0) {
-		sleep(1);
-		iRet = RS_RET_OK;
-		goto ret;
-	}
+	int severity = cs.iDfltSeverity;
+	int facility = cs.iDfltFacility;
 
 	/* Get message text */
 	if (sd_journal_get_data(j, "MESSAGE", &get, &length) < 0) {
-		logmsgInternal(NO_ERRCODE, LOG_SYSLOG|LOG_INFO, (uchar *)"log message from journal doesn't have MESSAGE", 0);
-		iRet = RS_RET_OK;
-		goto ret;
-	}
-	message = strndup(get+8, length-8);
-	if (message == NULL) {
-		iRet = RS_RET_OUT_OF_MEMORY;
-		goto ret;
+		message = strdup("");
+	} else {
+		message = strndup(((const char*)get)+8, length-8);
+		if (message == NULL) {
+			iRet = RS_RET_OUT_OF_MEMORY;
+			goto ret;
+		}
 	}
 
-	/* Get message priority */
+	/* Get message severity ("priority" in journald's terminology) */
 	if (sd_journal_get_data(j, "PRIORITY", &get, &length) >= 0) {
-		get2 = strndup(get, length);
-		priority = ((char *)get2)[9] - '0';
-		free (get2);
+		if (length == 10) {
+			severity = ((char *)get)[9] - '0';
+			if (severity < 0 || 7 < severity) {
+				dbgprintf("The value of the 'PRIORITY' field is "
+					"out of bounds: %d, resetting\n", severity);
+				severity = cs.iDfltSeverity;
+			}
+		} else {
+			dbgprintf("The value of the 'PRIORITY' field has an "
+				"unexpected length: %d\n", length);
+		}
 	}
 
 	/* Get syslog facility */
 	if (sd_journal_get_data(j, "SYSLOG_FACILITY", &get, &length) >= 0) {
-		get2 = strndup(get, length);
-		char f = ((char *)get2)[16];
-		if (f >= '0' && f <= '9') {
-			facility += f - '0';
-		}
-		f = ((char *)get2)[17];
-		if (f >= '0' && f <= '9') {
-			facility *= 10;
-			facility += (f - '0');
+		if (length == 17 || length == 18) {
+			facility = ((char *)get)[16] - '0';
+			if (length == 18) {
+				facility *= 10;
+				facility += ((char *)get)[17] - '0';
+			}
+			if (facility < 0 || 23 < facility) {
+				dbgprintf("The value of the 'FACILITY' field is "
+					"out of bounds: %d, resetting\n", facility);
+				facility = cs.iDfltFacility;
+			}
+		} else {
+			dbgprintf("The value of the 'FACILITY' field has an "
+				"unexpected length: %d\n", length);
 		}
-		free (get2);
-	} else {
-		/* message is missing facility -> internal systemd journal msg, drop */
-		iRet = RS_RET_OK;
-		goto free_message;
 	}
 
-	/* Get message identifier and add ':' */
+	/* Get message identifier, client pid and add ':' */
 	if (sd_journal_get_data(j, "SYSLOG_IDENTIFIER", &get, &length) >= 0) {
-		sys_iden = strndup(get+18, length-18);
+		sys_iden = strndup(((const char*)get)+18, length-18);
 	} else {
 		sys_iden = strdup("journal");
 	}
@@ -214,19 +270,43 @@ readjournal() {
 		goto free_message;
 	}
 
-	asprintf(&sys_iden_help, "%s:", sys_iden);
-	if (sys_iden_help == NULL) {
+	if (sd_journal_get_data(j, "SYSLOG_PID", &pidget, &pidlength) >= 0) {
+		sys_pid = strndup(((const char*)pidget)+11, pidlength-11);
+		if (sys_pid == NULL) {
+			iRet = RS_RET_OUT_OF_MEMORY;
+			free (sys_iden);
+			goto free_message;
+		}
+	} else {
+		sys_pid = NULL;
+	}
+
+	if (sys_pid) {
+		r = asprintf(&sys_iden_help, "%s[%s]:", sys_iden, sys_pid);
+	} else {
+		r = asprintf(&sys_iden_help, "%s:", sys_iden);
+	}
+
+	free (sys_iden);
+	free (sys_pid);
+
+	if (-1 == r) {
 		iRet = RS_RET_OUT_OF_MEMORY;
 		goto finalize_it;
 	}
-	free (sys_iden);
 
 	json = json_object_new_object();
 
 	SD_JOURNAL_FOREACH_DATA(j, get, l) {
 		/* locate equal sign, this is always present */
 		equal_sign = memchr(get, '=', l);
-		assert (equal_sign != NULL);
+
+		/* ... but we know better than to trust the specs */
+		if (equal_sign == NULL) {
+			errmsg.LogError(0, RS_RET_ERR, "SD_JOURNAL_FOREACH_DATA()"
+				"returned a malformed field (has no '='): '%s'", (char*)get);
+			continue; /* skip the entry */
+		}
 
 		/* get length of journal data prefix */
 		prefixlen = ((char *)equal_sign - (char *)get);
@@ -296,7 +376,7 @@ readjournal() {
 
 		prefixlen++; /* remove '=' */
 
-		data = strndup(get + prefixlen, l - prefixlen);
+		data = strndup(((const char*)get) + prefixlen, l - prefixlen);
 		if (data == NULL) {
 			iRet = RS_RET_OUT_OF_MEMORY;
 			free (name);
@@ -317,7 +397,7 @@ readjournal() {
 	}
 
 	/* submit message */
-	enqMsg((uchar *)message, (uchar *) sys_iden_help, facility, priority, &tv, json);
+	enqMsg((uchar *)message, (uchar *) sys_iden_help, facility, severity, &tv, json);
 
 finalize_it:
 	free(sys_iden_help);
@@ -337,7 +417,9 @@ persistJournalState () {
 	char *cursor;
 	int ret = 0;
 
-	if ((ret = sd_journal_get_cursor(j, &cursor)) > 0) {
+	/* On success, sd_journal_get_cursor()  returns 1 in systemd
+	   197 or older and 0 in systemd 198 or newer */
+	if ((ret = sd_journal_get_cursor(j, &cursor)) >= 0) {
 		if ((sf = fopen(cs.stateFile, "wb")) != NULL) {
 			if (fprintf(sf, "%s", cursor) < 0) {
 				iRet = RS_RET_IO_ERROR;
@@ -345,29 +427,93 @@ persistJournalState () {
 			fclose(sf);
 			free(cursor);
 		} else {
+			char errStr[256];
+			rs_strerror_r(errno, errStr, sizeof(errStr));
+			errmsg.LogError(0, RS_RET_FOPEN_FAILURE, "fopen() failed: "
+				"'%s', path: '%s'\n", errStr, cs.stateFile);
 			iRet = RS_RET_FOPEN_FAILURE;
 		}
 	} else {
+		char errStr[256];
+		rs_strerror_r(-(ret), errStr, sizeof(errStr));
+		errmsg.LogError(0, RS_RET_ERR, "sd_journal_get_cursor() failed: '%s'\n", errStr);
 		iRet = RS_RET_ERR;
 	}
 	RETiRet;
 }
 
 
-BEGINrunInput
-CODESTARTrunInput
-	/* this is an endless loop - it is terminated when the thread is
-	 * signalled to do so. This, however, is handled by the framework,
-	 * right into the sleep below.
-	 */
-	int count = 0;
+/* Polls the journal for new messages. Similar to sd_journal_wait()
+ * except for the special handling of EINTR.
+ */
+static rsRetVal
+pollJournal()
+{
+	DEFiRet;
+	struct pollfd pollfd;
+	int r;
+
+	pollfd.fd = sd_journal_get_fd(j);
+	pollfd.events = sd_journal_get_events(j);
+	r = poll(&pollfd, 1, -1);
+	if (r == -1) {
+		if (errno == EINTR) {
+			/* EINTR is also received during termination
+			 * so return now to check the term state.
+			 */
+			ABORT_FINALIZE(RS_RET_OK);
+		} else {
+			char errStr[256];
+
+			rs_strerror_r(errno, errStr, sizeof(errStr));
+			errmsg.LogError(0, RS_RET_ERR,
+				"poll() failed: '%s'", errStr);
+			ABORT_FINALIZE(RS_RET_ERR);
+		}
+	}
+
+	assert(r == 1);
+
+	r = sd_journal_process(j);
+	if (r < 0) {
+		char errStr[256];
+
+		rs_strerror_r(errno, errStr, sizeof(errStr));
+		errmsg.LogError(0, RS_RET_ERR,
+			"sd_journal_process() failed: '%s'", errStr);
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* This function loads a journal cursor from the state file.
+ */
+static rsRetVal
+loadJournalState()
+{
+	DEFiRet;
 
-	char readCursor[128 + 1];
-	FILE *r_sf;
+	if (cs.stateFile[0] != '/') {
+		char *new_stateFile;
+
+		if (-1 == asprintf(&new_stateFile, "%s/%s", (char *)glbl.GetWorkDir(), cs.stateFile)) {
+			errmsg.LogError(0, RS_RET_OUT_OF_MEMORY, "imjournal: asprintf failed\n");
+			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+		}
+		free (cs.stateFile);
+		cs.stateFile = new_stateFile;
+	}
 
 	/* if state file exists, set cursor to appropriate position */
 	if (access(cs.stateFile, F_OK|R_OK) != -1) {
+		FILE *r_sf;
+
 		if ((r_sf = fopen(cs.stateFile, "rb")) != NULL) {
+			char readCursor[128 + 1];
+
 			if (fscanf(r_sf, "%128s\n", readCursor) != EOF) {
 				if (sd_journal_seek_cursor(j, readCursor) != 0) {
 					errmsg.LogError(0, RS_RET_ERR, "imjournal: "
@@ -387,17 +533,77 @@ CODESTARTrunInput
 			errmsg.LogError(0, RS_RET_FOPEN_FAILURE, "imjournal: "
 					"open on state file `%s' failed\n", cs.stateFile);
 		}
+	} else {
+		/* when IgnorePrevious, seek to the end of journal */
+		if (cs.bIgnorePrevious) {
+			if (sd_journal_seek_tail(j) < 0) {
+				char errStr[256];
+
+				rs_strerror_r(errno, errStr, sizeof(errStr));
+				errmsg.LogError(0, RS_RET_ERR,
+					"sd_journal_seek_tail() failed: '%s'", errStr);
+				ABORT_FINALIZE(RS_RET_ERR);
+			}
+
+			if (sd_journal_previous(j) < 0) {
+				char errStr[256];
+
+				rs_strerror_r(errno, errStr, sizeof(errStr));
+				errmsg.LogError(0, RS_RET_ERR,
+					"sd_journal_previous() failed: '%s'", errStr);
+				ABORT_FINALIZE(RS_RET_ERR);
+			}
+		}
+	} 
+
+finalize_it:
+	RETiRet;
+}
+
+BEGINrunInput
+CODESTARTrunInput
+	CHKiRet(ratelimitNew(&ratelimiter, "imjournal", NULL));
+	dbgprintf("imjournal: ratelimiting burst %d, interval %d\n", cs.ratelimitBurst,
+		  cs.ratelimitInterval);
+	ratelimitSetLinuxLike(ratelimiter, cs.ratelimitInterval, cs.ratelimitBurst);
+	ratelimitSetNoTimeCache(ratelimiter);
+
+	if (cs.stateFile) {
+		CHKiRet(loadJournalState());
 	}
 
+	/* this is an endless loop - it is terminated when the thread is
+	 * signalled to do so. This, however, is handled by the framework.
+	 */
 	while (glbl.GetGlobalInputTermState() == 0) {
+		int count = 0, r;
+
+		r = sd_journal_next(j);
+		if (r < 0) {
+			char errStr[256];
+
+			rs_strerror_r(errno, errStr, sizeof(errStr));
+			errmsg.LogError(0, RS_RET_ERR,
+				"sd_journal_next() failed: '%s'", errStr);
+			ABORT_FINALIZE(RS_RET_ERR);
+		}
+
+		if (r == 0) {
+			/* No new messages, wait for activity. */
+			CHKiRet(pollJournal());
+			continue;
+		}
+
 		CHKiRet(readjournal());
-		count++;
-		if (count == cs.iPersistStateInterval) {
-			count = 0;
-			persistJournalState();
+		if (cs.stateFile) { /* can't persist without a state file */
+			/* TODO: This could use some finer metric. */
+			count++;
+			if (count == cs.iPersistStateInterval) {
+				count = 0;
+				persistJournalState();
+			}
 		}
 	}
-	persistJournalState();
 
 finalize_it:
 ENDrunInput
@@ -409,6 +615,10 @@ CODESTARTbeginCnfLoad
 
 	cs.iPersistStateInterval = DFLT_persiststateinterval;
 	cs.stateFile = NULL;
+	cs.ratelimitBurst = 20000;
+	cs.ratelimitInterval = 600;
+	cs.iDfltSeverity = DFLT_SEVERITY;
+	cs.iDfltFacility = DFLT_FACILITY;
 ENDbeginCnfLoad
 
 
@@ -444,7 +654,11 @@ ENDwillRun
 /* close journal */
 BEGINafterRun
 CODESTARTafterRun
+	if (cs.stateFile) { /* can't persist without a state file */
+		persistJournalState();
+	}
 	sd_journal_close(j);
+	ratelimitDestruct(ratelimiter);
 ENDafterRun
 
 
@@ -487,6 +701,23 @@ CODESTARTsetModCnf
 			cs.iPersistStateInterval = (int) pvals[i].val.d.n;
 		} else if (!strcmp(modpblk.descr[i].name, "statefile")) {
 			cs.stateFile = (char *)es_str2cstr(pvals[i].val.d.estr, NULL);
+		} else if(!strcmp(modpblk.descr[i].name, "ratelimit.burst")) {
+			cs.ratelimitBurst = (int) pvals[i].val.d.n;
+		} else if(!strcmp(modpblk.descr[i].name, "ratelimit.interval")) {
+			cs.ratelimitInterval = (int) pvals[i].val.d.n;
+		} else if (!strcmp(modpblk.descr[i].name, "ignorepreviousmessages")) {
+			cs.bIgnorePrevious = (int) pvals[i].val.d.n; 
+		} else if (!strcmp(modpblk.descr[i].name, "defaultseverity")) {
+			cs.iDfltSeverity = (int) pvals[i].val.d.n;
+		} else if (!strcmp(modpblk.descr[i].name, "defaultfacility")) {
+			/* ugly workaround to handle facility numbers; values
+			   derived from names need to be eight times smaller */
+
+			char *fac, *p;
+
+			fac = p = es_str2cstr(pvals[i].val.d.estr, NULL);
+			facilityHdlr((uchar **) &p, (void *) &cs.iDfltFacility);
+			free(fac);
 		} else {
 			dbgprintf("imjournal: program error, non-handled "
 				"param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
@@ -500,16 +731,22 @@ finalize_it:
 ENDsetModCnf
 
 
+BEGINisCompatibleWithFeature
+CODESTARTisCompatibleWithFeature
+	if(eFeat == sFEATURENonCancelInputTermination)
+		iRet = RS_RET_OK;
+ENDisCompatibleWithFeature
+
+
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_IMOD_QUERIES
 CODEqueryEtryPt_STD_CONF2_QUERIES
 CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
+CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES
 ENDqueryEtryPt
 
 
-
-
 BEGINmodInit()
 CODESTARTmodInit
 	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
@@ -524,10 +761,20 @@ CODEmodInit_QueryRegCFSLineHdlr
 	CHKiRet(prop.CreateStringProp(&pInputName, UCHAR_CONSTANT("imjournal"), sizeof("imjournal") - 1));
 	CHKiRet(prop.CreateStringProp(&pLocalHostIP, UCHAR_CONSTANT("127.0.0.1"), sizeof("127.0.0.1") - 1));
 
-	CHKiRet(omsdRegCFSLineHdlr((uchar *)"persiststateinterval", 0, eCmdHdlrInt,
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournalpersiststateinterval", 0, eCmdHdlrInt,
 		NULL, &cs.iPersistStateInterval, STD_LOADABLE_MODULE_ID));
-	CHKiRet(omsdRegCFSLineHdlr((uchar *)"statefile", 0, eCmdHdlrGetWord,
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournalratelimitinterval", 0, eCmdHdlrInt,
+		NULL, &cs.ratelimitInterval, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournalratelimitburst", 0, eCmdHdlrInt,
+		NULL, &cs.ratelimitBurst, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournalstatefile", 0, eCmdHdlrGetWord,
 		NULL, &cs.stateFile, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournalignorepreviousmessages", 0, eCmdHdlrBinary,
+		NULL, &cs.bIgnorePrevious, STD_LOADABLE_MODULE_ID)); 
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournaldefaultseverity", 0, eCmdHdlrSeverity,
+		NULL, &cs.iDfltSeverity, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournaldefaultfacility", 0, eCmdHdlrCustomHandler,
+		facilityHdlr, &cs.iDfltFacility, STD_LOADABLE_MODULE_ID));
 
 
 ENDmodInit