Blame SOURCES/rsyslog-8.24.0-rhbz1538372-imjournal-duplicates.patch

fa2733
From: Jiri Vymazal <jvymazal@redhat.com>
f656cf
Date: Wed, 25 Jul 2018 15:05:01 -0500
fa2733
fa2733
modification and merge of below patches for RHEL consumers, 
fa2733
also modified journal invalidate/rotation handling to keep possibility
fa2733
to continue after switch of persistent journal
fa2733
original:
f656cf
%From 3bede5ba768975c8b6fe3d1f3e11075910f52fdd Mon Sep 17 00:00:00 2001
f656cf
%From: Jiri Vymazal <jvymazal@redhat.com>
f656cf
%Date: Wed, 7 Mar 2018 11:57:29 +0100
f656cf
%Subject: [PATCH] Fetching cursor on readJournal() and simplified pollJournal()
f656cf
%
f656cf
%Fetching journal cursor in persistJournal could cause us to save
f656cf
%invalid cursor leading to duplicating messages further on, now we are
f656cf
%saving it on each readJournal() where we now that the state is good.
f656cf
%This result in simplyfing persisJournalState() a bit as well.
f656cf
%
f656cf
%pollJournal() is now cleaner and faster, correctly handles INVALIDATE
f656cf
%status from journald and is able to continue polling after journal
f656cf
%flush. Also reduced POLL_TIMEOUT a bit as it caused rsyslog to exit
f656cf
%with error in corner cases for some ppc when left at full second.
f656cf
plus
fa2733
%
fa2733
%From a99f9b4b42d261c384aee09306fc421df2cca7a5 Mon Sep 17 00:00:00 2001
fa2733
%From: Peter Portante <peter.a.portante@gmail.com>
fa2733
%Date: Wed, 24 Jan 2018 19:34:41 -0500
fa2733
%Subject: [PATCH] Proposed fix for handling journal correctly
fa2733
%
fa2733
%The fix is to immediately setup the inotify file descriptor via
fa2733
%`sd_journal_get_fd()` right after a journal open, and then
fa2733
%periodically call `sd_journal_process()` to give the client API
fa2733
%library a chance to detect deleted journal files on disk that need to
fa2733
%be closed so they can be properly erased by the file system.
fa2733
%
fa2733
%We remove the open/close dance and simplify that code as a result.
fa2733
%
fa2733
%Fixes issue #2436.
fa2733
and also:
fa2733
%From 27f96c84d34ee000fbb5d45b00233f2ec3cf2d8a Mon Sep 17 00:00:00 2001
fa2733
%From: Rainer Gerhards <rgerhards@adiscon.com>
fa2733
%Date: Tue, 24 Oct 2017 16:14:13 +0200
fa2733
%Subject: [PATCH] imjournal bugfix: do not disable itself on error
fa2733
%
fa2733
%If some functions calls inside the main loop failed, imjournal exited
fa2733
%with an error code, actually disabling all logging from the journal.
fa2733
%This was probably never intended.
fa2733
%
fa2733
%This patch makes imjournal recover the situation instead.
fa2733
%
fa2733
%closes https://github.com/rsyslog/rsyslog/issues/1895
fa2733
---
f656cf
 plugins/imjournal/imjournal.c | 211 ++++++++++++++++++++++--------------------
f656cf
 1 file changed, 110 insertions(+), 102 deletions(-)
fa2733
fa2733
--- a/plugins/imjournal/imjournal.c
fa2733
+++ b/plugins/imjournal/imjournal.c
f656cf
@@ -80,6 +80,7 @@ static struct configSettings_s {
f656cf
 	int iDfltFacility;
f656cf
 	int bUseJnlPID;
f656cf
	char *dfltTag;
f656cf
+	int bWorkAroundJournalBug;
f656cf
 } cs;
f656cf
 
f656cf
 static rsRetVal facilityHdlr(uchar **pp, void *pVal);
f656cf
@@ -95,6 +96,7 @@ static struct cnfparamdescr modpdescr[] = {
f656cf
 	{ "defaultfacility", eCmdHdlrString, 0 },
f656cf
 	{ "usepidfromsystem", eCmdHdlrBinary, 0 },
f656cf
	{ "defaulttag", eCmdHdlrGetWord, 0 },
f656cf
+	{ "workaroundjournalbug", eCmdHdlrBinary, 0 }
f656cf
 };
f656cf
 static struct cnfparamblk modpblk =
f656cf
 	{ CNFPARAMBLK_VERSION,
fa2733
@@ -114,6 +114,10 @@ /* module-global parameters */
fa2733
 static const char *pid_field_name;	/* read-only after startup */
fa2733
 static ratelimit_t *ratelimiter = NULL;
fa2733
 static sd_journal *j;
fa2733
+static int j_inotify_fd;
fa2733
+static char *last_cursor = NULL;
fa2733
+
fa2733
+#define J_PROCESS_PERIOD 1024  /* Call sd_journal_process() every 1,024 records */
fa2733
 
fa2733
 static rsRetVal persistJournalState(void);
fa2733
 static rsRetVal loadJournalState(void);
fa2733
@@ -123,6 +127,14 @@ openJournal(sd_journal** jj)
fa2733
 
fa2733
 	if (sd_journal_open(jj, SD_JOURNAL_LOCAL_ONLY) < 0)
fa2733
 		iRet = RS_RET_IO_ERROR;
fa2733
+	int r;
fa2733
+
fa2733
+	if ((r = sd_journal_get_fd(j)) < 0) {
fa2733
+		errmsg.LogError(-r, RS_RET_IO_ERROR, "imjournal: sd_journal_get_fd() failed");
fa2733
+		iRet = RS_RET_IO_ERROR;
fa2733
+	} else {
fa2733
+		j_inotify_fd = r;
fa2733
+	}	
fa2733
 	RETiRet;
fa2733
 }
fa2733
 
fa2733
@@ -132,6 +144,7 @@ closeJournal(sd_journal** jj)
fa2733
 		persistJournalState();
fa2733
 	}
fa2733
 	sd_journal_close(*jj);
fa2733
+	j_inotify_fd = 0;
fa2733
 }
fa2733
 
fa2733
 
fa2733
@@ -262,6 +275,7 @@ readjournal(void)
fa2733
 	char *message = NULL;
fa2733
 	char *sys_iden = NULL;
fa2733
 	char *sys_iden_help = NULL;
fa2733
+	char *c = NULL;
fa2733
 
fa2733
 	const void *get;
fa2733
 	const void *pidget;
f656cf
@@ -433,6 +437,15 @@ readjournal(void)
fa2733
 		tv.tv_usec = timestamp % 1000000;
fa2733
 	}
fa2733
 
f656cf
+	if (cs.bWorkAroundJournalBug) {
f656cf
+		/* save journal cursor (at this point we can be sure it is valid) */
f656cf
+		sd_journal_get_cursor(j, &c);
f656cf
+		if (c) {
f656cf
+			free(last_cursor);
f656cf
+			last_cursor = c;
f656cf
+		}
f656cf
+	}
fa2733
+
fa2733
 	/* submit message */
fa2733
 	enqMsg((uchar *)message, (uchar *) sys_iden_help, facility, severity, &tv, json, 0);
fa2733
 
f656cf
@@ -413,44 +433,49 @@ persistJournalState (void)
fa2733
 	DEFiRet;
fa2733
 	FILE *sf; /* state file */
fa2733
 	char tmp_sf[MAXFNAME];
fa2733
-	char *cursor;
f656cf
 	int ret = 0;
fa2733
 
fa2733
-	/* On success, sd_journal_get_cursor()  returns 1 in systemd
fa2733
-	   197 or older and 0 in systemd 198 or newer */
fa2733
-	if ((ret = sd_journal_get_cursor(j, &cursor)) >= 0) {
fa2733
-               /* we create a temporary name by adding a ".tmp"
fa2733
-                * suffix to the end of our state file's name
fa2733
-                */
fa2733
-               snprintf(tmp_sf, sizeof(tmp_sf), "%s.tmp", cs.stateFile);
fa2733
-               if ((sf = fopen(tmp_sf, "wb")) != NULL) {
fa2733
-			if (fprintf(sf, "%s", cursor) < 0) {
fa2733
-				iRet = RS_RET_IO_ERROR;
fa2733
-			}
fa2733
-			fclose(sf);
fa2733
-			free(cursor);
fa2733
-                       /* change the name of the file to the configured one */
fa2733
-                       if (iRet == RS_RET_OK && rename(tmp_sf, cs.stateFile) == -1) {
fa2733
-                               char errStr[256];
fa2733
-                               rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-                               iRet = RS_RET_IO_ERROR;
fa2733
-                               errmsg.LogError(0, iRet, "rename() failed: "
fa2733
-                                       "'%s', new path: '%s'\n", errStr, cs.stateFile);
fa2733
-                       }
f656cf
+	if (cs.bWorkAroundJournalBug) {
f656cf
+		if (!last_cursor)
f656cf
+			ABORT_FINALIZE(RS_RET_OK);
fa2733
 
fa2733
-		} else {
fa2733
-			char errStr[256];
fa2733
-			rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-			errmsg.LogError(0, RS_RET_FOPEN_FAILURE, "fopen() failed: "
fa2733
-				"'%s', path: '%s'\n", errStr, tmp_sf);
fa2733
-			iRet = RS_RET_FOPEN_FAILURE;
fa2733
-		}
fa2733
-	} else {
f656cf
+	} else if ((ret = sd_journal_get_cursor(j, &last_cursor)) < 0) {
f656cf
 		char errStr[256];
f656cf
 		rs_strerror_r(-(ret), errStr, sizeof(errStr));
f656cf
 		errmsg.LogError(0, RS_RET_ERR, "sd_journal_get_cursor() failed: '%s'\n", errStr);
fa2733
-		iRet = RS_RET_ERR;
f656cf
+		ABORT_FINALIZE(RS_RET_ERR);
f656cf
	}
fa2733
+	/* we create a temporary name by adding a ".tmp"
fa2733
+	 * suffix to the end of our state file's name
fa2733
+	 */
fa2733
+	snprintf(tmp_sf, sizeof(tmp_sf), "%s.tmp", cs.stateFile);
fa2733
+
fa2733
+	sf = fopen(tmp_sf, "wb");
fa2733
+	if (!sf) {
fa2733
+		errmsg.LogError(errno, RS_RET_FOPEN_FAILURE, "imjournal: fopen() failed for path: '%s'", tmp_sf);
fa2733
+		ABORT_FINALIZE(RS_RET_FOPEN_FAILURE);
fa2733
+	}
fa2733
+
f656cf
+	ret = fputs(last_cursor, sf);
f656cf
+	if (ret < 0) {
f656cf
+		errmsg.LogError(errno, RS_RET_IO_ERROR, "imjournal: failed to save cursor to: '%s'", tmp_sf);
f656cf
+		ret = fclose(sf);
fa2733
+		ABORT_FINALIZE(RS_RET_IO_ERROR);
fa2733
+	}
fa2733
+
f656cf
+	ret = fclose(sf);
f656cf
+	if (ret < 0) {
f656cf
+		errmsg.LogError(errno, RS_RET_IO_ERROR, "imjournal: fclose() failed for path: '%s'", tmp_sf);
fa2733
+		ABORT_FINALIZE(RS_RET_IO_ERROR);
fa2733
+	}
fa2733
+
f656cf
+	ret = rename(tmp_sf, cs.stateFile);
f656cf
+	if (ret < 0) {
fa2733
+		errmsg.LogError(errno, iRet, "imjournal: rename() failed for new path: '%s'", cs.stateFile);
fa2733
+		ABORT_FINALIZE(RS_RET_IO_ERROR);
fa2733
+	}
fa2733
+
fa2733
+finalize_it:
fa2733
 	RETiRet;
fa2733
 }
fa2733
 
f656cf
@@ -473,64 +473,26 @@
fa2733
  * except for the special handling of EINTR.
fa2733
  */
fa2733
 
fa2733
-#define POLL_TIMEOUT 1000 /* timeout for poll is 1s */
fa2733
+#define POLL_TIMEOUT 900000 /* timeout for poll is 900ms */
fa2733
 
fa2733
 static rsRetVal
fa2733
 pollJournal(void)
fa2733
 {
fa2733
 	DEFiRet;
fa2733
-	struct pollfd pollfd;
fa2733
-	int pr = 0;
fa2733
-	int jr = 0;
fa2733
-
fa2733
-	pollfd.fd = sd_journal_get_fd(j);
fa2733
-	pollfd.events = sd_journal_get_events(j);
fa2733
-	pr = poll(&pollfd, 1, POLL_TIMEOUT);
fa2733
-	if (pr == -1) {
fa2733
-		if (errno == EINTR) {
fa2733
-			/* EINTR is also received during termination
fa2733
-			 * so return now to check the term state.
fa2733
-			 */
fa2733
-			ABORT_FINALIZE(RS_RET_OK);
fa2733
-		} else {
fa2733
-			char errStr[256];
fa2733
-
fa2733
-			rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-			errmsg.LogError(0, RS_RET_ERR,
fa2733
-				"poll() failed: '%s'", errStr);
fa2733
-			ABORT_FINALIZE(RS_RET_ERR);
fa2733
-		}
fa2733
-	}
fa2733
+	int r;
fa2733
 
f656cf
+	r = sd_journal_wait(j, POLL_TIMEOUT);
fa2733
 
fa2733
-	jr = sd_journal_process(j);
fa2733
-	
fa2733
-	if (pr == 1 && jr == SD_JOURNAL_INVALIDATE) {
fa2733
-		/* do not persist stateFile sd_journal_get_cursor will fail! */
fa2733
-		char* tmp = cs.stateFile;
fa2733
-		cs.stateFile = NULL;
fa2733
+	if (r == SD_JOURNAL_INVALIDATE) {
fa2733
 		closeJournal(&j);
fa2733
-		cs.stateFile = tmp;
fa2733
 
fa2733
 		iRet = openJournal(&j);
fa2733
-		if (iRet != RS_RET_OK) {
fa2733
-			char errStr[256];
fa2733
-			rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-			errmsg.LogError(0, RS_RET_IO_ERROR,
fa2733
-				"sd_journal_open() failed: '%s'", errStr);
fa2733
+		if (iRet != RS_RET_OK)
fa2733
 			ABORT_FINALIZE(RS_RET_ERR);
fa2733
-		}
fa2733
 
fa2733
-		if(cs.stateFile != NULL){
fa2733
+		if (cs.stateFile)
fa2733
 			iRet = loadJournalState();
fa2733
-		}
fa2733
-		LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal reloaded...");
fa2733
-	} else if (jr < 0) {
fa2733
-		char errStr[256];
fa2733
-		rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-		errmsg.LogError(0, RS_RET_ERR,
fa2733
-			"sd_journal_process() failed: '%s'", errStr);
fa2733
-		ABORT_FINALIZE(RS_RET_ERR);
fa2733
+		errmsg.LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal reloaded...");
fa2733
 	}
fa2733
 
fa2733
 finalize_it:
fa2733
@@ -631,8 +612,17 @@ loadJournalState(void)
fa2733
 	RETiRet;
fa2733
 }
fa2733
 
fa2733
+static void
fa2733
+tryRecover(void) {
fa2733
+	errmsg.LogMsg(0, RS_RET_OK, LOG_INFO, "imjournal: trying to recover from unexpected "
fa2733
+		"journal error");
fa2733
+	closeJournal(&j);
fa2733
+	srSleep(10, 0);	// do not hammer machine with too-frequent retries
fa2733
+	openJournal(&j);
fa2733
+}
fa2733
+
fa2733
 BEGINrunInput
fa2733
-	int count = 0;
fa2733
+	uint64_t count = 0;
fa2733
 CODESTARTrunInput
fa2733
 	CHKiRet(ratelimitNew(&ratelimiter, "imjournal", NULL));
fa2733
 	dbgprintf("imjournal: ratelimiting burst %d, interval %d\n", cs.ratelimitBurst,
fa2733
@@ -665,26 +655,38 @@ CODESTARTrunInput
fa2733
 
fa2733
 		r = sd_journal_next(j);
fa2733
 		if (r < 0) {
fa2733
-			char errStr[256];
fa2733
-
fa2733
-			rs_strerror_r(errno, errStr, sizeof(errStr));
fa2733
-			errmsg.LogError(0, RS_RET_ERR,
fa2733
-				"sd_journal_next() failed: '%s'", errStr);
fa2733
-			ABORT_FINALIZE(RS_RET_ERR);
fa2733
+			tryRecover();
fa2733
+			continue;
fa2733
 		}
fa2733
 
fa2733
 		if (r == 0) {
fa2733
 			/* No new messages, wait for activity. */
fa2733
-			CHKiRet(pollJournal());
fa2733
+			if (pollJournal() != RS_RET_OK) {
fa2733
+ 				tryRecover();
fa2733
+ 			}
fa2733
 			continue;
fa2733
 		}
fa2733
 
fa2733
-		CHKiRet(readjournal());
fa2733
+		if (readjournal() != RS_RET_OK) {
fa2733
+ 			tryRecover();
fa2733
+ 			continue;
fa2733
+ 		}
fa2733
+
fa2733
+		count++;
fa2733
+
fa2733
+		if ((count % J_PROCESS_PERIOD) == 0) {
fa2733
+			/* Give the journal a periodic chance to detect rotated journal files to be cleaned up. */
fa2733
+			r = sd_journal_process(j);
fa2733
+			if (r < 0) {
fa2733
+				errmsg.LogError(-r, RS_RET_ERR, "imjournal: sd_journal_process() failed");
fa2733
+				tryRecover();
fa2733
+				continue;
fa2733
+			}
fa2733
+		}
fa2733
+
fa2733
 		if (cs.stateFile) { /* can't persist without a state file */
fa2733
 			/* TODO: This could use some finer metric. */
fa2733
-			count++;
fa2733
-			if (count == cs.iPersistStateInterval) {
fa2733
-				count = 0;
fa2733
+			if ((count % cs.iPersistStateInterval) == 0) {
fa2733
 				persistJournalState();
fa2733
 			}
fa2733
 		}
f656cf
@@ -901,6 +909,8 @@ CODESTARTsetModCnf
f656cf
 			cs.bUseJnlPID = (int) pvals[i].val.d.n;
f656cf
		} else if (!strcmp(modpblk.descr[i].name, "defaulttag")) {
f656cf
			cs.dfltTag = (char *)es_str2cstr(pvals[i].val.d.estr, NULL);
f656cf
+		} else if (!strcmp(modpblk.descr[i].name, "workaroundjournalbug")) {
f656cf
+			cs.bWorkAroundJournalBug = (int) pvals[i].val.d.n;
f656cf
 		} else {
f656cf
 			dbgprintf("imjournal: program error, non-handled "
f656cf
 				"param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
f656cf
@@ -961,6 +971,8 @@ CODEmodInit_QueryRegCFSLineHdlr
f656cf
		NULL, &cs.bUseJnlPID, STD_LOADABLE_MODULE_ID));
f656cf
	CHKiRet(omsdRegCFSLineHdlr((uchar *)"imjournaldefaulttag", 0, eCmdHdlrGetWord,
f656cf
		NULL, &cs.dfltTag, STD_LOADABLE_MODULE_ID));
f656cf
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"workaroundjournalbug", 0, eCmdHdlrBinary,
f656cf
+		NULL, &cs.bWorkAroundJournalBug, STD_LOADABLE_MODULE_ID));
f656cf
 ENDmodInit
f656cf
 /* vim:set ai:
f656cf
  */