From: Jiri Vymazal <jvymazal@redhat.com>

Date: Wed, 25 Jul 2018 15:05:01 -0500

modification and merge of below patches for RHEL consumers, 

also modified journal invalidate/rotation handling to keep possibility

to continue after switch of persistent journal

original:

%From 3bede5ba768975c8b6fe3d1f3e11075910f52fdd Mon Sep 17 00:00:00 2001

%From: Jiri Vymazal <jvymazal@redhat.com>

%Date: Wed, 7 Mar 2018 11:57:29 +0100

%Subject: [PATCH] Fetching cursor on readJournal() and simplified pollJournal()

%

%Fetching journal cursor in persistJournal could cause us to save

%invalid cursor leading to duplicating messages further on, now we are

%saving it on each readJournal() where we now that the state is good.

%This result in simplyfing persisJournalState() a bit as well.

%

%pollJournal() is now cleaner and faster, correctly handles INVALIDATE

%status from journald and is able to continue polling after journal

%flush. Also reduced POLL_TIMEOUT a bit as it caused rsyslog to exit

%with error in corner cases for some ppc when left at full second.

plus

%

%From a99f9b4b42d261c384aee09306fc421df2cca7a5 Mon Sep 17 00:00:00 2001

%From: Peter Portante <peter.a.portante@gmail.com>

%Date: Wed, 24 Jan 2018 19:34:41 -0500

%Subject: [PATCH] Proposed fix for handling journal correctly

%

%The fix is to immediately setup the inotify file descriptor via

%`sd_journal_get_fd()` right after a journal open, and then

%periodically call `sd_journal_process()` to give the client API

%library a chance to detect deleted journal files on disk that need to

%be closed so they can be properly erased by the file system.

%

%We remove the open/close dance and simplify that code as a result.

%

%Fixes issue #2436.

and also:

%From 27f96c84d34ee000fbb5d45b00233f2ec3cf2d8a Mon Sep 17 00:00:00 2001

%From: Rainer Gerhards <rgerhards@adiscon.com>

%Date: Tue, 24 Oct 2017 16:14:13 +0200

%Subject: [PATCH] imjournal bugfix: do not disable itself on error

%

%If some functions calls inside the main loop failed, imjournal exited

%with an error code, actually disabling all logging from the journal.

%This was probably never intended.

%

%This patch makes imjournal recover the situation instead.

%

%closes https://github.com/rsyslog/rsyslog/issues/1895

---

 plugins/imjournal/imjournal.c | 211 ++++++++++++++++++++++--------------------

 1 file changed, 110 insertions(+), 102 deletions(-)

--- a/plugins/imjournal/imjournal.c

+++ b/plugins/imjournal/imjournal.c

@@ -80,6 +80,7 @@ static struct configSettings_s {

 	int iDfltFacility;

 	int bUseJnlPID;

	char *dfltTag;

+	int bWorkAroundJournalBug;

 } cs;

 static rsRetVal facilityHdlr(uchar **pp, void *pVal);

@@ -95,6 +96,7 @@ static struct cnfparamdescr modpdescr[] = {

 	{ "defaultfacility", eCmdHdlrString, 0 },

 	{ "usepidfromsystem", eCmdHdlrBinary, 0 },

	{ "defaulttag", eCmdHdlrGetWord, 0 },

+	{ "workaroundjournalbug", eCmdHdlrBinary, 0 }

 };

 static struct cnfparamblk modpblk =

 	{ CNFPARAMBLK_VERSION,

@@ -114,6 +114,10 @@ /* module-global parameters */

 static const char *pid_field_name;	/* read-only after startup */

 static ratelimit_t *ratelimiter = NULL;

 static sd_journal *j;

+static int j_inotify_fd;

+static char *last_cursor = NULL;

+

+#define J_PROCESS_PERIOD 1024  /* Call sd_journal_process() every 1,024 records */

 static rsRetVal persistJournalState(void);

 static rsRetVal loadJournalState(void);

@@ -123,6 +127,14 @@ openJournal(sd_journal** jj)

 	if (sd_journal_open(jj, SD_JOURNAL_LOCAL_ONLY) < 0)

 		iRet = RS_RET_IO_ERROR;

+	int r;

+

+	if ((r = sd_journal_get_fd(j)) < 0) {

+		errmsg.LogError(-r, RS_RET_IO_ERROR, "imjournal: sd_journal_get_fd() failed");

+		iRet = RS_RET_IO_ERROR;

+	} else {

+		j_inotify_fd = r;

+	}	

 	RETiRet;

 }

@@ -132,6 +144,7 @@ closeJournal(sd_journal** jj)

 		persistJournalState();

 	}

 	sd_journal_close(*jj);

+	j_inotify_fd = 0;

 }

@@ -262,6 +275,7 @@ readjournal(void)

 	char *message = NULL;

 	char *sys_iden = NULL;

 	char *sys_iden_help = NULL;

+	char *c = NULL;

 	const void *get;

 	const void *pidget;

@@ -433,6 +437,15 @@ readjournal(void)

 		tv.tv_usec = timestamp % 1000000;

 	}

+	if (cs.bWorkAroundJournalBug) {

+		/* save journal cursor (at this point we can be sure it is valid) */

+		sd_journal_get_cursor(j, &c);

+		if (c) {

+			free(last_cursor);

+			last_cursor = c;

+		}

+	}

+

 	/* submit message */

 	enqMsg((uchar *)message, (uchar *) sys_iden_help, facility, severity, &tv, json, 0);

@@ -413,44 +433,49 @@ persistJournalState (void)

 	DEFiRet;

 	FILE *sf; /* state file */

 	char tmp_sf[MAXFNAME];

-	char *cursor;

 	int ret = 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) {

-               /* we create a temporary name by adding a ".tmp"

-                * suffix to the end of our state file's name

-                */

-               snprintf(tmp_sf, sizeof(tmp_sf), "%s.tmp", cs.stateFile);

-               if ((sf = fopen(tmp_sf, "wb")) != NULL) {

-			if (fprintf(sf, "%s", cursor) < 0) {

-				iRet = RS_RET_IO_ERROR;

-			}

-			fclose(sf);

-			free(cursor);

-                       /* change the name of the file to the configured one */

-                       if (iRet == RS_RET_OK && rename(tmp_sf, cs.stateFile) == -1) {

-                               char errStr[256];

-                               rs_strerror_r(errno, errStr, sizeof(errStr));

-                               iRet = RS_RET_IO_ERROR;

-                               errmsg.LogError(0, iRet, "rename() failed: "

-                                       "'%s', new path: '%s'\n", errStr, cs.stateFile);

-                       }

+	if (cs.bWorkAroundJournalBug) {

+		if (!last_cursor)

+			ABORT_FINALIZE(RS_RET_OK);

-		} 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, tmp_sf);

-			iRet = RS_RET_FOPEN_FAILURE;

-		}

-	} else {

+	} else if ((ret = sd_journal_get_cursor(j, &last_cursor)) < 0) {

 		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;

+		ABORT_FINALIZE(RS_RET_ERR);

	}

+	/* we create a temporary name by adding a ".tmp"

+	 * suffix to the end of our state file's name

+	 */

+	snprintf(tmp_sf, sizeof(tmp_sf), "%s.tmp", cs.stateFile);

+

+	sf = fopen(tmp_sf, "wb");

+	if (!sf) {

+		errmsg.LogError(errno, RS_RET_FOPEN_FAILURE, "imjournal: fopen() failed for path: '%s'", tmp_sf);

+		ABORT_FINALIZE(RS_RET_FOPEN_FAILURE);

+	}

+

+	ret = fputs(last_cursor, sf);

+	if (ret < 0) {

+		errmsg.LogError(errno, RS_RET_IO_ERROR, "imjournal: failed to save cursor to: '%s'", tmp_sf);

+		ret = fclose(sf);

+		ABORT_FINALIZE(RS_RET_IO_ERROR);

+	}

+

+	ret = fclose(sf);

+	if (ret < 0) {

+		errmsg.LogError(errno, RS_RET_IO_ERROR, "imjournal: fclose() failed for path: '%s'", tmp_sf);

+		ABORT_FINALIZE(RS_RET_IO_ERROR);

+	}

+

+	ret = rename(tmp_sf, cs.stateFile);

+	if (ret < 0) {

+		errmsg.LogError(errno, iRet, "imjournal: rename() failed for new path: '%s'", cs.stateFile);

+		ABORT_FINALIZE(RS_RET_IO_ERROR);

+	}

+

+finalize_it:

 	RETiRet;

 }

@@ -473,64 +473,26 @@

  * except for the special handling of EINTR.

  */

-#define POLL_TIMEOUT 1000 /* timeout for poll is 1s */

+#define POLL_TIMEOUT 900000 /* timeout for poll is 900ms */

 static rsRetVal

 pollJournal(void)

 {

 	DEFiRet;

-	struct pollfd pollfd;

-	int pr = 0;

-	int jr = 0;

-

-	pollfd.fd = sd_journal_get_fd(j);

-	pollfd.events = sd_journal_get_events(j);

-	pr = poll(&pollfd, 1, POLL_TIMEOUT);

-	if (pr == -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);

-		}

-	}

+	int r;

+	r = sd_journal_wait(j, POLL_TIMEOUT);

-	jr = sd_journal_process(j);

-	

-	if (pr == 1 && jr == SD_JOURNAL_INVALIDATE) {

-		/* do not persist stateFile sd_journal_get_cursor will fail! */

-		char* tmp = cs.stateFile;

-		cs.stateFile = NULL;

+	if (r == SD_JOURNAL_INVALIDATE) {

 		closeJournal(&j);

-		cs.stateFile = tmp;

 		iRet = openJournal(&j);

-		if (iRet != RS_RET_OK) {

-			char errStr[256];

-			rs_strerror_r(errno, errStr, sizeof(errStr));

-			errmsg.LogError(0, RS_RET_IO_ERROR,

-				"sd_journal_open() failed: '%s'", errStr);

+		if (iRet != RS_RET_OK)

 			ABORT_FINALIZE(RS_RET_ERR);

-		}

-		if(cs.stateFile != NULL){

+		if (cs.stateFile)

 			iRet = loadJournalState();

-		}

-		LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal reloaded...");

-	} else if (jr < 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);

+		errmsg.LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal reloaded...");

 	}

 finalize_it:

@@ -631,8 +612,17 @@ loadJournalState(void)

 	RETiRet;

 }

+static void

+tryRecover(void) {

+	errmsg.LogMsg(0, RS_RET_OK, LOG_INFO, "imjournal: trying to recover from unexpected "

+		"journal error");

+	closeJournal(&j);

+	srSleep(10, 0);	// do not hammer machine with too-frequent retries

+	openJournal(&j);

+}

+

 BEGINrunInput

-	int count = 0;

+	uint64_t count = 0;

 CODESTARTrunInput

 	CHKiRet(ratelimitNew(&ratelimiter, "imjournal", NULL));

 	dbgprintf("imjournal: ratelimiting burst %d, interval %d\n", cs.ratelimitBurst,

@@ -665,26 +655,38 @@ CODESTARTrunInput

 		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);

+			tryRecover();

+			continue;

 		}

 		if (r == 0) {

 			/* No new messages, wait for activity. */

-			CHKiRet(pollJournal());

+			if (pollJournal() != RS_RET_OK) {

+ 				tryRecover();

+ 			}

 			continue;

 		}

-		CHKiRet(readjournal());

+		if (readjournal() != RS_RET_OK) {

+ 			tryRecover();

+ 			continue;

+ 		}

+

+		count++;

+

+		if ((count % J_PROCESS_PERIOD) == 0) {

+			/* Give the journal a periodic chance to detect rotated journal files to be cleaned up. */

+			r = sd_journal_process(j);

+			if (r < 0) {

+				errmsg.LogError(-r, RS_RET_ERR, "imjournal: sd_journal_process() failed");

+				tryRecover();

+				continue;

+			}

+		}

+

 		if (cs.stateFile) { /* can't persist without a state file */

 			/* TODO: This could use some finer metric. */

-			count++;

-			if (count == cs.iPersistStateInterval) {

-				count = 0;

+			if ((count % cs.iPersistStateInterval) == 0) {

 				persistJournalState();

 			}

 		}

@@ -901,6 +909,8 @@ CODESTARTsetModCnf

 			cs.bUseJnlPID = (int) pvals[i].val.d.n;

		} else if (!strcmp(modpblk.descr[i].name, "defaulttag")) {

			cs.dfltTag = (char *)es_str2cstr(pvals[i].val.d.estr, NULL);

+		} else if (!strcmp(modpblk.descr[i].name, "workaroundjournalbug")) {

+			cs.bWorkAroundJournalBug = (int) pvals[i].val.d.n;

 		} else {

 			dbgprintf("imjournal: program error, non-handled "

 				"param '%s' in beginCnfLoad\n", modpblk.descr[i].name);

@@ -961,6 +971,8 @@ CODEmodInit_QueryRegCFSLineHdlr

		NULL, &cs.bUseJnlPID, STD_LOADABLE_MODULE_ID));

	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournaldefaulttag", 0, eCmdHdlrGetWord,

		NULL, &cs.dfltTag, STD_LOADABLE_MODULE_ID));

+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"workaroundjournalbug", 0, eCmdHdlrBinary,

+		NULL, &cs.bWorkAroundJournalBug, STD_LOADABLE_MODULE_ID));

 ENDmodInit

 /* vim:set ai:

  */