|
|
|
|
@ -0,0 +1,752 @@
|
|
|
|
|
From 78976a9bc0592317be2e41b7f3703803f7971e1e Mon Sep 17 00:00:00 2001
|
|
|
|
|
From: Jiri Vymazal <jvymazal@redhat.com>
|
|
|
|
|
Date: Wed, 25 Sep 2019 15:18:57 +0200
|
|
|
|
|
Subject: [PATCH] plugin code restructuring, added remote option
|
|
|
|
|
|
|
|
|
|
Decomposed ReadJournal() a bit, also now coupling journald
|
|
|
|
|
variables in one struct, added few warning messages and debug
|
|
|
|
|
prints to help with bug hunts in future, also got rid of two
|
|
|
|
|
needless journald calls. WorkAroundJournalBug now deprecated.
|
|
|
|
|
Added option to pull journald records from outside local machine.
|
|
|
|
|
---
|
|
|
|
|
plugins/imjournal/imjournal.c | 368 ++++++++++++++++++++--------------
|
|
|
|
|
1 file changed, 214 insertions(+), 154 deletions(-)
|
|
|
|
|
|
|
|
|
|
diff --git a/plugins/imjournal/imjournal.c b/plugins/imjournal/imjournal.c
|
|
|
|
|
index 2e27922b51..8008f159c4 100644
|
|
|
|
|
--- a/plugins/imjournal/imjournal.c
|
|
|
|
|
+++ b/plugins/imjournal/imjournal.c
|
|
|
|
|
@@ -81,8 +81,9 @@ static struct configSettings_s {
|
|
|
|
|
int iDfltFacility;
|
|
|
|
|
int bUseJnlPID;
|
|
|
|
|
char *usePid;
|
|
|
|
|
- int bWorkAroundJournalBug;
|
|
|
|
|
+ int bWorkAroundJournalBug; /* deprecated, left for backwards compatibility only */
|
|
|
|
|
int bFsync;
|
|
|
|
|
+ int bRemote;
|
|
|
|
|
} cs;
|
|
|
|
|
|
|
|
|
|
static rsRetVal facilityHdlr(uchar **pp, void *pVal);
|
|
|
|
|
@@ -100,7 +101,8 @@ static struct cnfparamdescr modpdescr[] = {
|
|
|
|
|
{ "usepidfromsystem", eCmdHdlrBinary, 0 },
|
|
|
|
|
{ "usepid", eCmdHdlrString, 0 },
|
|
|
|
|
{ "workaroundjournalbug", eCmdHdlrBinary, 0 },
|
|
|
|
|
- { "fsync", eCmdHdlrBinary, 0 }
|
|
|
|
|
+ { "fsync", eCmdHdlrBinary, 0 },
|
|
|
|
|
+ { "remote", eCmdHdlrBinary, 0 }
|
|
|
|
|
};
|
|
|
|
|
static struct cnfparamblk modpblk =
|
|
|
|
|
{ CNFPARAMBLK_VERSION,
|
|
|
|
|
@@ -120,8 +122,6 @@ static prop_t *pLocalHostIP = NULL; /* a pseudo-constant propterty for 127.0.0.1
|
|
|
|
|
static const char *pidFieldName; /* read-only after startup */
|
|
|
|
|
static int bPidFallBack;
|
|
|
|
|
static ratelimit_t *ratelimiter = NULL;
|
|
|
|
|
-static sd_journal *j;
|
|
|
|
|
-static sbool reloaded = 0;
|
|
|
|
|
static struct {
|
|
|
|
|
statsobj_t *stats;
|
|
|
|
|
STATSCOUNTER_DEF(ctrSubmitted, mutCtrSubmitted)
|
|
|
|
|
@@ -134,34 +134,58 @@ static struct {
|
|
|
|
|
uint64 ratelimitDiscardedInInterval;
|
|
|
|
|
uint64 diskUsageBytes;
|
|
|
|
|
} statsCounter;
|
|
|
|
|
-static char *last_cursor = NULL;
|
|
|
|
|
+struct journalContext_s { /* structure encapsulating all the journald_API-related stuff */
|
|
|
|
|
+ sd_journal *j; /* main object encapsulating journal for us, has to be used in every sd_journal*() call */
|
|
|
|
|
+ sbool reloaded; /* we have reloaded journal after detecting rotation */
|
|
|
|
|
+ sbool atHead; /* true if we are at start of journal (no seek was done) */
|
|
|
|
|
+ char *cursor; /* should point to last valid journald entry we processed */
|
|
|
|
|
+};
|
|
|
|
|
+static struct journalContext_s journalContext = {NULL, 0, 1, NULL};
|
|
|
|
|
|
|
|
|
|
#define J_PROCESS_PERIOD 1024 /* Call sd_journal_process() every 1,024 records */
|
|
|
|
|
|
|
|
|
|
-static rsRetVal persistJournalState(int trySave);
|
|
|
|
|
+static rsRetVal persistJournalState(void);
|
|
|
|
|
static rsRetVal loadJournalState(void);
|
|
|
|
|
|
|
|
|
|
static rsRetVal openJournal(void) {
|
|
|
|
|
int r;
|
|
|
|
|
DEFiRet;
|
|
|
|
|
|
|
|
|
|
- if ((r = sd_journal_open(&j, SD_JOURNAL_LOCAL_ONLY)) < 0) {
|
|
|
|
|
- LogError(-r, RS_RET_IO_ERROR, "imjournal: sd_journal_open() failed");
|
|
|
|
|
- iRet = RS_RET_IO_ERROR;
|
|
|
|
|
+ if (journalContext.j) {
|
|
|
|
|
+ LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imjournal: opening journal when already opened.\n");
|
|
|
|
|
}
|
|
|
|
|
- if ((r = sd_journal_get_fd(j)) < 0) {
|
|
|
|
|
- LogError(-r, RS_RET_IO_ERROR, "imjournal: sd_journal_get_fd() failed");
|
|
|
|
|
+ if ((r = sd_journal_open(&journalContext.j, cs.bRemote? 0 : SD_JOURNAL_LOCAL_ONLY)) < 0) {
|
|
|
|
|
+ LogError(-r, RS_RET_IO_ERROR, "imjournal: sd_journal_open() failed");
|
|
|
|
|
iRet = RS_RET_IO_ERROR;
|
|
|
|
|
}
|
|
|
|
|
+ journalContext.atHead = 1;
|
|
|
|
|
RETiRet;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/* trySave shoulod only be true if there is no journald error preceeding this call */
|
|
|
|
|
-static void closeJournal(int trySave) {
|
|
|
|
|
- if (cs.stateFile) { /* can't persist without a state file */
|
|
|
|
|
- persistJournalState(trySave);
|
|
|
|
|
+static void closeJournal(void) {
|
|
|
|
|
+ if (!journalContext.j) {
|
|
|
|
|
+ LogMsg(0, RS_RET_OK_WARN, LOG_WARNING, "imjournal: closing NULL journal.\n");
|
|
|
|
|
}
|
|
|
|
|
- sd_journal_close(j);
|
|
|
|
|
+ sd_journal_close(journalContext.j);
|
|
|
|
|
+ journalContext.j = NULL; /* setting to NULL here as journald API will not do that for us... */
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+static int journalGetData(const char *field, const void **data, size_t *length)
|
|
|
|
|
+{
|
|
|
|
|
+ int ret;
|
|
|
|
|
+
|
|
|
|
|
+ ret = sd_journal_get_data(journalContext.j, field, data, length);
|
|
|
|
|
+ if (ret == -EADDRNOTAVAIL) {
|
|
|
|
|
+ LogError(-ret, RS_RET_ERR, "imjournal: Tried to get data without a 'next' call.\n");
|
|
|
|
|
+ if ((ret = sd_journal_next(journalContext.j)) < 0) {
|
|
|
|
|
+ LogError(-ret, RS_RET_ERR, "imjournal: sd_journal_next() failed\n");
|
|
|
|
|
+ } else {
|
|
|
|
|
+ ret = sd_journal_get_data(journalContext.j, field, data, length);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ return ret;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -223,12 +247,84 @@ sanitizeValue(const char *in, size_t len, char **out)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
+/* Read JSON part of single journald message and return it as JSON object
|
|
|
|
|
+ */
|
|
|
|
|
+static rsRetVal
|
|
|
|
|
+readJSONfromJournalMsg(struct fjson_object **json)
|
|
|
|
|
+{
|
|
|
|
|
+ DEFiRet;
|
|
|
|
|
+ const void *get;
|
|
|
|
|
+ const void *equal_sign;
|
|
|
|
|
+ struct fjson_object *jval;
|
|
|
|
|
+ size_t l;
|
|
|
|
|
+ long prefixlen = 0;
|
|
|
|
|
+
|
|
|
|
|
+ CHKmalloc(*json = fjson_object_new_object());
|
|
|
|
|
+
|
|
|
|
|
+ SD_JOURNAL_FOREACH_DATA(journalContext.j, get, l) {
|
|
|
|
|
+ char *data;
|
|
|
|
|
+ char *name;
|
|
|
|
|
+
|
|
|
|
|
+ /* locate equal sign, this is always present */
|
|
|
|
|
+ equal_sign = memchr(get, '=', l);
|
|
|
|
|
+
|
|
|
|
|
+ /* ... but we know better than to trust the specs */
|
|
|
|
|
+ if (equal_sign == NULL) {
|
|
|
|
|
+ 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);
|
|
|
|
|
+
|
|
|
|
|
+ CHKmalloc(name = strndup(get, prefixlen));
|
|
|
|
|
+
|
|
|
|
|
+ prefixlen++; /* remove '=' */
|
|
|
|
|
+
|
|
|
|
|
+ CHKiRet_Hdlr(sanitizeValue(((const char *)get) + prefixlen, l - prefixlen, &data)) {
|
|
|
|
|
+ free (name);
|
|
|
|
|
+ FINALIZE;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /* and save them to json object */
|
|
|
|
|
+ jval = fjson_object_new_string((char *)data);
|
|
|
|
|
+ fjson_object_object_add(*json, name, jval);
|
|
|
|
|
+ free (data);
|
|
|
|
|
+ free (name);
|
|
|
|
|
+ }
|
|
|
|
|
+finalize_it:
|
|
|
|
|
+ RETiRet;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+/* Try to obtain current journald cursor and save it to journalContext struct.
|
|
|
|
|
+ */
|
|
|
|
|
+static rsRetVal
|
|
|
|
|
+updateJournalCursor(void)
|
|
|
|
|
+{
|
|
|
|
|
+ DEFiRet;
|
|
|
|
|
+ char *c = NULL;
|
|
|
|
|
+ int r;
|
|
|
|
|
+
|
|
|
|
|
+ if ((r = sd_journal_get_cursor(journalContext.j, &c)) < 0) {
|
|
|
|
|
+ LogError(-r, RS_RET_ERR, "imjournal: Could not get journald cursor!\n");
|
|
|
|
|
+ ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
+ }
|
|
|
|
|
+ /* save journal cursor (at this point we can be sure it is valid) */
|
|
|
|
|
+ free(journalContext.cursor);
|
|
|
|
|
+ journalContext.cursor = c;
|
|
|
|
|
+finalize_it:
|
|
|
|
|
+ RETiRet;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
/* enqueue the the journal message into the message queue.
|
|
|
|
|
* The provided msg string is not freed - thus must be done
|
|
|
|
|
* by the caller.
|
|
|
|
|
*/
|
|
|
|
|
static rsRetVal
|
|
|
|
|
-enqMsg(uchar *msg, uchar *pszTag, int iFacility, int iSeverity, struct timeval *tp, struct json_object *json,
|
|
|
|
|
+enqMsg(uchar *msg, uchar *pszTag, int iFacility, int iSeverity, struct timeval *tp, struct fjson_object *json,
|
|
|
|
|
int sharedJsonProperties)
|
|
|
|
|
{
|
|
|
|
|
struct syslogTime st;
|
|
|
|
|
@@ -267,15 +363,17 @@ int sharedJsonProperties)
|
|
|
|
|
STATSCOUNTER_INC(statsCounter.ctrSubmitted, statsCounter.mutCtrSubmitted);
|
|
|
|
|
|
|
|
|
|
finalize_it:
|
|
|
|
|
- if (iRet == RS_RET_DISCARDMSG)
|
|
|
|
|
+ if (iRet == RS_RET_DISCARDMSG) {
|
|
|
|
|
STATSCOUNTER_INC(statsCounter.ctrDiscarded, statsCounter.mutCtrDiscarded);
|
|
|
|
|
+ } else if (iRet != RS_RET_OK) {
|
|
|
|
|
+ LogError(0, RS_RET_ERR, "imjournal: error during enqMsg().\n");
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
|
|
RETiRet;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
-/* Read journal log while data are available, each read() reads one
|
|
|
|
|
- * record of printk buffer.
|
|
|
|
|
+/* Read journal log while data are available, each read() reads one journald record.
|
|
|
|
|
*/
|
|
|
|
|
static rsRetVal
|
|
|
|
|
readjournal(void)
|
|
|
|
|
@@ -285,39 +383,32 @@ readjournal(void)
|
|
|
|
|
struct timeval tv;
|
|
|
|
|
uint64_t timestamp;
|
|
|
|
|
|
|
|
|
|
- struct json_object *json = NULL;
|
|
|
|
|
+ struct fjson_object *json = NULL;
|
|
|
|
|
int r;
|
|
|
|
|
|
|
|
|
|
/* Information from messages */
|
|
|
|
|
char *message = NULL;
|
|
|
|
|
char *sys_iden;
|
|
|
|
|
char *sys_iden_help = NULL;
|
|
|
|
|
- char *c = NULL;
|
|
|
|
|
|
|
|
|
|
const void *get;
|
|
|
|
|
const void *pidget;
|
|
|
|
|
size_t length;
|
|
|
|
|
size_t pidlength;
|
|
|
|
|
|
|
|
|
|
- const void *equal_sign;
|
|
|
|
|
- struct json_object *jval;
|
|
|
|
|
- size_t l;
|
|
|
|
|
-
|
|
|
|
|
- long prefixlen = 0;
|
|
|
|
|
-
|
|
|
|
|
int severity = cs.iDfltSeverity;
|
|
|
|
|
int facility = cs.iDfltFacility;
|
|
|
|
|
|
|
|
|
|
/* Get message text */
|
|
|
|
|
- if (sd_journal_get_data(j, "MESSAGE", &get, &length) < 0) {
|
|
|
|
|
- message = strdup("");
|
|
|
|
|
+ if (journalGetData("MESSAGE", &get, &length) < 0) {
|
|
|
|
|
+ CHKmalloc(message = strdup(""));
|
|
|
|
|
} else {
|
|
|
|
|
CHKiRet(sanitizeValue(((const char *)get) + 8, length - 8, &message));
|
|
|
|
|
}
|
|
|
|
|
STATSCOUNTER_INC(statsCounter.ctrRead, statsCounter.mutCtrRead);
|
|
|
|
|
|
|
|
|
|
/* Get message severity ("priority" in journald's terminology) */
|
|
|
|
|
- if (sd_journal_get_data(j, "PRIORITY", &get, &length) >= 0) {
|
|
|
|
|
+ if (journalGetData("PRIORITY", &get, &length) >= 0) {
|
|
|
|
|
if (length == 10) {
|
|
|
|
|
severity = ((char *)get)[9] - '0';
|
|
|
|
|
if (severity < 0 || 7 < severity) {
|
|
|
|
|
@@ -332,7 +423,7 @@ readjournal(void)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/* Get syslog facility */
|
|
|
|
|
- if (sd_journal_get_data(j, "SYSLOG_FACILITY", &get, &length) >= 0) {
|
|
|
|
|
+ if (journalGetData("SYSLOG_FACILITY", &get, &length) >= 0) {
|
|
|
|
|
// Note: the journal frequently contains invalid facilities!
|
|
|
|
|
if (length == 17 || length == 18) {
|
|
|
|
|
facility = ((char *)get)[16] - '0';
|
|
|
|
|
@@ -352,14 +443,14 @@ readjournal(void)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/* Get message identifier, client pid and add ':' */
|
|
|
|
|
- if (sd_journal_get_data(j, "SYSLOG_IDENTIFIER", &get, &length) >= 0) {
|
|
|
|
|
+ if (journalGetData("SYSLOG_IDENTIFIER", &get, &length) >= 0) {
|
|
|
|
|
CHKiRet(sanitizeValue(((const char *)get) + 18, length - 18, &sys_iden));
|
|
|
|
|
} else {
|
|
|
|
|
CHKmalloc(sys_iden = strdup("journal"));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/* trying to get PID, default is "SYSLOG_PID" property */
|
|
|
|
|
- if (sd_journal_get_data(j, pidFieldName, &pidget, &pidlength) >= 0) {
|
|
|
|
|
+ if (journalGetData(pidFieldName, &pidget, &pidlength) >= 0) {
|
|
|
|
|
char *sys_pid;
|
|
|
|
|
int val_ofs;
|
|
|
|
|
|
|
|
|
|
@@ -372,7 +463,7 @@ readjournal(void)
|
|
|
|
|
free (sys_pid);
|
|
|
|
|
} else {
|
|
|
|
|
/* this is fallback, "SYSLOG_PID" doesn't exist so trying to get "_PID" property */
|
|
|
|
|
- if (bPidFallBack && sd_journal_get_data(j, "_PID", &pidget, &pidlength) >= 0) {
|
|
|
|
|
+ if (bPidFallBack && journalGetData("_PID", &pidget, &pidlength) >= 0) {
|
|
|
|
|
char *sys_pid;
|
|
|
|
|
int val_ofs;
|
|
|
|
|
|
|
|
|
|
@@ -396,55 +487,15 @@ readjournal(void)
|
|
|
|
|
ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
- json = json_object_new_object();
|
|
|
|
|
-
|
|
|
|
|
- SD_JOURNAL_FOREACH_DATA(j, get, l) {
|
|
|
|
|
- char *data;
|
|
|
|
|
- char *name;
|
|
|
|
|
-
|
|
|
|
|
- /* locate equal sign, this is always present */
|
|
|
|
|
- equal_sign = memchr(get, '=', l);
|
|
|
|
|
-
|
|
|
|
|
- /* ... but we know better than to trust the specs */
|
|
|
|
|
- if (equal_sign == NULL) {
|
|
|
|
|
- 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);
|
|
|
|
|
-
|
|
|
|
|
- name = strndup(get, prefixlen);
|
|
|
|
|
- CHKmalloc(name);
|
|
|
|
|
-
|
|
|
|
|
- prefixlen++; /* remove '=' */
|
|
|
|
|
-
|
|
|
|
|
- CHKiRet_Hdlr(sanitizeValue(((const char *)get) + prefixlen, l - prefixlen, &data)) {
|
|
|
|
|
- free (name);
|
|
|
|
|
- FINALIZE;
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- /* and save them to json object */
|
|
|
|
|
- jval = json_object_new_string((char *)data);
|
|
|
|
|
- json_object_object_add(json, name, jval);
|
|
|
|
|
- free (data);
|
|
|
|
|
- free (name);
|
|
|
|
|
- }
|
|
|
|
|
+ CHKiRet(readJSONfromJournalMsg(&json));
|
|
|
|
|
|
|
|
|
|
/* calculate timestamp */
|
|
|
|
|
- if (sd_journal_get_realtime_usec(j, ×tamp) >= 0) {
|
|
|
|
|
+ if (sd_journal_get_realtime_usec(journalContext.j, ×tamp) >= 0) {
|
|
|
|
|
tv.tv_sec = timestamp / 1000000;
|
|
|
|
|
tv.tv_usec = timestamp % 1000000;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
- if (cs.bWorkAroundJournalBug) {
|
|
|
|
|
- /* save journal cursor (at this point we can be sure it is valid) */
|
|
|
|
|
- if (!sd_journal_get_cursor(j, &c)) {
|
|
|
|
|
- free(last_cursor);
|
|
|
|
|
- last_cursor = c;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
+ iRet = updateJournalCursor();
|
|
|
|
|
|
|
|
|
|
/* submit message */
|
|
|
|
|
enqMsg((uchar *)message, (uchar *) sys_iden_help, facility, severity, &tv, json, 0);
|
|
|
|
|
@@ -456,32 +507,22 @@ readjournal(void)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
-/* This function gets journal cursor and saves it into state file.
|
|
|
|
|
- * If WorkAroundJournalBug option is turned on it does use cursor saved previously.
|
|
|
|
|
- * If it is false and if "trySave" is false it skips altogether.
|
|
|
|
|
+/* This function saves journal cursor into state file.
|
|
|
|
|
+ * It must be checked that stateFile is configured prior to calling this.
|
|
|
|
|
*/
|
|
|
|
|
static rsRetVal
|
|
|
|
|
-persistJournalState(int trySave)
|
|
|
|
|
+persistJournalState(void)
|
|
|
|
|
{
|
|
|
|
|
DEFiRet;
|
|
|
|
|
FILE *sf = NULL; /* state file */
|
|
|
|
|
char tmp_sf[MAXFNAME];
|
|
|
|
|
size_t n;
|
|
|
|
|
|
|
|
|
|
- if (cs.bWorkAroundJournalBug) {
|
|
|
|
|
- /* first check that we have valid cursor */
|
|
|
|
|
- if (!last_cursor) {
|
|
|
|
|
- ABORT_FINALIZE(RS_RET_OK);
|
|
|
|
|
- }
|
|
|
|
|
- } else if (trySave) {
|
|
|
|
|
- int ret;
|
|
|
|
|
- free(last_cursor);
|
|
|
|
|
- if ((ret = sd_journal_get_cursor(j, &last_cursor))) {
|
|
|
|
|
- LogError(-ret, RS_RET_ERR, "imjournal: sd_journal_get_cursor() failed");
|
|
|
|
|
- last_cursor = NULL;
|
|
|
|
|
- ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
- }
|
|
|
|
|
- } else { /* not trying to get cursor out of invalid journal state */
|
|
|
|
|
+ DBGPRINTF("Persisting journal position, cursor: %s, at head? %d\n",
|
|
|
|
|
+ journalContext.cursor, journalContext.atHead);
|
|
|
|
|
+
|
|
|
|
|
+ /* first check that we have valid cursor */
|
|
|
|
|
+ if (!journalContext.cursor) {
|
|
|
|
|
ABORT_FINALIZE(RS_RET_OK);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -500,7 +541,7 @@ persistJournalState(int trySave)
|
|
|
|
|
ABORT_FINALIZE(RS_RET_FOPEN_FAILURE);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
- if(fputs(last_cursor, sf) == EOF) {
|
|
|
|
|
+ if(fputs(journalContext.cursor, sf) == EOF) {
|
|
|
|
|
LogError(errno, RS_RET_IO_ERROR, "imjournal: failed to save cursor to: '%s'", tmp_sf);
|
|
|
|
|
ABORT_FINALIZE(RS_RET_IO_ERROR);
|
|
|
|
|
}
|
|
|
|
|
@@ -541,43 +582,59 @@ persistJournalState(int trySave)
|
|
|
|
|
|
|
|
|
|
static rsRetVal skipOldMessages(void);
|
|
|
|
|
|
|
|
|
|
-#define POLL_TIMEOUT 900000 /* timeout for poll is 900ms */
|
|
|
|
|
-
|
|
|
|
|
static rsRetVal
|
|
|
|
|
-pollJournal(void)
|
|
|
|
|
+handleRotation(void)
|
|
|
|
|
{
|
|
|
|
|
DEFiRet;
|
|
|
|
|
- int err;
|
|
|
|
|
+ int r;
|
|
|
|
|
|
|
|
|
|
- err = sd_journal_wait(j, POLL_TIMEOUT);
|
|
|
|
|
- if (err == SD_JOURNAL_INVALIDATE && !reloaded) {
|
|
|
|
|
- STATSCOUNTER_INC(statsCounter.ctrRotations, statsCounter.mutCtrRotations);
|
|
|
|
|
- closeJournal(0);
|
|
|
|
|
+ LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal files changed, reloading...\n");
|
|
|
|
|
+ STATSCOUNTER_INC(statsCounter.ctrRotations, statsCounter.mutCtrRotations);
|
|
|
|
|
+ closeJournal();
|
|
|
|
|
|
|
|
|
|
- iRet = openJournal();
|
|
|
|
|
- if (iRet != RS_RET_OK) {
|
|
|
|
|
- ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
- }
|
|
|
|
|
+ iRet = openJournal();
|
|
|
|
|
+ if (iRet != RS_RET_OK) {
|
|
|
|
|
+ ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
|
|
- /* If we have locally saved cursor there is no need to read it from state file */
|
|
|
|
|
- if (cs.bWorkAroundJournalBug && last_cursor)
|
|
|
|
|
- {
|
|
|
|
|
- if (sd_journal_seek_cursor(j, last_cursor) != 0) {
|
|
|
|
|
- LogError(0, RS_RET_ERR, "imjournal: "
|
|
|
|
|
- "couldn't seek to cursor `%s'\n", last_cursor);
|
|
|
|
|
- iRet = RS_RET_ERR;
|
|
|
|
|
- }
|
|
|
|
|
- /* Need to advance because cursor points at last processed message */
|
|
|
|
|
- sd_journal_next(j);
|
|
|
|
|
+ /* If we have locally saved cursor there is no need to read it from state file */
|
|
|
|
|
+ if (journalContext.cursor)
|
|
|
|
|
+ {
|
|
|
|
|
+ if (sd_journal_seek_cursor(journalContext.j, journalContext.cursor) != 0) {
|
|
|
|
|
+ LogError(0, RS_RET_ERR, "imjournal: "
|
|
|
|
|
+ "couldn't seek to cursor `%s'\n", journalContext.cursor);
|
|
|
|
|
+ iRet = RS_RET_ERR;
|
|
|
|
|
}
|
|
|
|
|
- else if (cs.stateFile) {
|
|
|
|
|
- iRet = loadJournalState();
|
|
|
|
|
+ journalContext.atHead = 0;
|
|
|
|
|
+ /* Need to advance because cursor points at last processed message */
|
|
|
|
|
+ if ((r = sd_journal_next(journalContext.j)) < 0) {
|
|
|
|
|
+ LogError(-r, RS_RET_ERR, "imjournal: sd_journal_next() failed");
|
|
|
|
|
+ iRet = RS_RET_ERR;
|
|
|
|
|
}
|
|
|
|
|
- LogMsg(0, RS_RET_OK, LOG_NOTICE, "imjournal: journal reloaded...");
|
|
|
|
|
- reloaded = 1;
|
|
|
|
|
+ }
|
|
|
|
|
+ else if (cs.stateFile) {
|
|
|
|
|
+ iRet = loadJournalState();
|
|
|
|
|
+ }
|
|
|
|
|
+ journalContext.reloaded = 1;
|
|
|
|
|
+
|
|
|
|
|
+finalize_it:
|
|
|
|
|
+ RETiRet;
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+#define POLL_TIMEOUT 900000 /* timeout for poll is 900ms */
|
|
|
|
|
+
|
|
|
|
|
+static rsRetVal
|
|
|
|
|
+pollJournal(void)
|
|
|
|
|
+{
|
|
|
|
|
+ DEFiRet;
|
|
|
|
|
+ int err;
|
|
|
|
|
+
|
|
|
|
|
+ err = sd_journal_wait(journalContext.j, POLL_TIMEOUT);
|
|
|
|
|
+ if (err == SD_JOURNAL_INVALIDATE && !journalContext.reloaded) {
|
|
|
|
|
+ CHKiRet(handleRotation());
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
- reloaded = 0;
|
|
|
|
|
+ journalContext.reloaded = 0;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
finalize_it:
|
|
|
|
|
@@ -591,12 +648,13 @@ skipOldMessages(void)
|
|
|
|
|
int r;
|
|
|
|
|
DEFiRet;
|
|
|
|
|
|
|
|
|
|
- if ((r = sd_journal_seek_tail(j)) < 0) {
|
|
|
|
|
+ if ((r = sd_journal_seek_tail(journalContext.j)) < 0) {
|
|
|
|
|
LogError(-r, RS_RET_ERR,
|
|
|
|
|
"imjournal: sd_journal_seek_tail() failed");
|
|
|
|
|
ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
}
|
|
|
|
|
- if ((r = sd_journal_previous(j)) < 0) {
|
|
|
|
|
+ journalContext.atHead = 0;
|
|
|
|
|
+ if ((r = sd_journal_previous(journalContext.j)) < 0) {
|
|
|
|
|
LogError(-r, RS_RET_ERR,
|
|
|
|
|
"imjournal: sd_journal_previous() failed");
|
|
|
|
|
ABORT_FINALIZE(RS_RET_ERR);
|
|
|
|
|
@@ -615,6 +673,9 @@ loadJournalState(void)
|
|
|
|
|
int r;
|
|
|
|
|
FILE *r_sf;
|
|
|
|
|
|
|
|
|
|
+ DBGPRINTF("Loading journal position, at head? %d, reloaded? %d\n",
|
|
|
|
|
+ journalContext.atHead, journalContext.reloaded);
|
|
|
|
|
+
|
|
|
|
|
if (cs.stateFile[0] != '/') {
|
|
|
|
|
char *new_stateFile;
|
|
|
|
|
if (-1 == asprintf(&new_stateFile, "%s/%s", (char *)glbl.GetWorkDir(), cs.stateFile)) {
|
|
|
|
|
@@ -639,13 +700,14 @@ loadJournalState(void)
|
|
|
|
|
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) {
|
|
|
|
|
+ if (sd_journal_seek_cursor(journalContext.j, readCursor) != 0) {
|
|
|
|
|
LogError(0, RS_RET_ERR, "imjournal: "
|
|
|
|
|
"couldn't seek to cursor `%s'\n", readCursor);
|
|
|
|
|
iRet = RS_RET_ERR;
|
|
|
|
|
} else {
|
|
|
|
|
+ journalContext.atHead = 0;
|
|
|
|
|
char * tmp_cursor = NULL;
|
|
|
|
|
- sd_journal_next(j);
|
|
|
|
|
+ sd_journal_next(journalContext.j);
|
|
|
|
|
/*
|
|
|
|
|
* This is resolving the situation when system is after reboot and boot_id
|
|
|
|
|
* doesn't match so cursor pointing into "future".
|
|
|
|
|
@@ -658,14 +720,15 @@ loadJournalState(void)
|
|
|
|
|
* but if cursor has been intentionally compromised it could stop logging even
|
|
|
|
|
* with persistent journal.
|
|
|
|
|
* */
|
|
|
|
|
- if ((r = sd_journal_get_cursor(j, &tmp_cursor)) < 0) {
|
|
|
|
|
+ if ((r = sd_journal_get_cursor(journalContext.j, &tmp_cursor)) < 0) {
|
|
|
|
|
LogError(-r, RS_RET_IO_ERROR, "imjournal: "
|
|
|
|
|
"loaded invalid cursor, seeking to the head of journal\n");
|
|
|
|
|
- if ((r = sd_journal_seek_head(j)) < 0) {
|
|
|
|
|
+ if ((r = sd_journal_seek_head(journalContext.j)) < 0) {
|
|
|
|
|
LogError(-r, RS_RET_ERR, "imjournal: "
|
|
|
|
|
"sd_journal_seek_head() failed, when cursor is invalid\n");
|
|
|
|
|
iRet = RS_RET_ERR;
|
|
|
|
|
}
|
|
|
|
|
+ journalContext.atHead = 1;
|
|
|
|
|
}
|
|
|
|
|
free(tmp_cursor);
|
|
|
|
|
}
|
|
|
|
|
@@ -680,18 +743,15 @@ loadJournalState(void)
|
|
|
|
|
if (iRet != RS_RET_OK && cs.bIgnoreNonValidStatefile) {
|
|
|
|
|
/* ignore state file errors */
|
|
|
|
|
iRet = RS_RET_OK;
|
|
|
|
|
- LogError(0, NO_ERRCODE, "imjournal: ignoring invalid state file %s",
|
|
|
|
|
- cs.stateFile);
|
|
|
|
|
+ LogError(0, NO_ERRCODE, "imjournal: ignoring invalid state file %s", cs.stateFile);
|
|
|
|
|
if (cs.bIgnorePrevious) {
|
|
|
|
|
skipOldMessages();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
- LogError(0, RS_RET_FOPEN_FAILURE, "imjournal: "
|
|
|
|
|
- "open on state file `%s' failed\n", cs.stateFile);
|
|
|
|
|
+ LogError(0, RS_RET_FOPEN_FAILURE, "imjournal: open on state file `%s' failed\n", cs.stateFile);
|
|
|
|
|
if (cs.bIgnorePrevious) {
|
|
|
|
|
- /* Seek to the very end of the journal and ignore all
|
|
|
|
|
- * older messages. */
|
|
|
|
|
+ /* Seek to the very end of the journal and ignore all older messages. */
|
|
|
|
|
skipOldMessages();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@@ -704,7 +764,7 @@ static void
|
|
|
|
|
tryRecover(void) {
|
|
|
|
|
LogMsg(0, RS_RET_OK, LOG_INFO, "imjournal: trying to recover from journal error");
|
|
|
|
|
STATSCOUNTER_INC(statsCounter.ctrRecoveryAttempts, statsCounter.mutCtrRecoveryAttempts);
|
|
|
|
|
- closeJournal(0);
|
|
|
|
|
+ closeJournal();
|
|
|
|
|
srSleep(10, 0); // do not hammer machine with too-frequent retries
|
|
|
|
|
openJournal();
|
|
|
|
|
}
|
|
|
|
|
@@ -723,8 +783,7 @@ CODESTARTrunInput
|
|
|
|
|
/* Load our position in the journal from the state file. */
|
|
|
|
|
CHKiRet(loadJournalState());
|
|
|
|
|
} else if (cs.bIgnorePrevious) {
|
|
|
|
|
- /* Seek to the very end of the journal and ignore all
|
|
|
|
|
- * older messages. */
|
|
|
|
|
+ /* Seek to the very end of the journal and ignore all older messages. */
|
|
|
|
|
skipOldMessages();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@@ -758,7 +817,7 @@ CODESTARTrunInput
|
|
|
|
|
while (glbl.GetGlobalInputTermState() == 0) {
|
|
|
|
|
int r;
|
|
|
|
|
|
|
|
|
|
- r = sd_journal_next(j);
|
|
|
|
|
+ r = sd_journal_next(journalContext.j);
|
|
|
|
|
if (r < 0) {
|
|
|
|
|
LogError(-r, RS_RET_ERR, "imjournal: sd_journal_next() failed");
|
|
|
|
|
tryRecover();
|
|
|
|
|
@@ -766,8 +825,12 @@ CODESTARTrunInput
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (r == 0) {
|
|
|
|
|
+ if (journalContext.atHead) {
|
|
|
|
|
+ LogMsg(0, RS_RET_OK, LOG_WARNING, "imjournal: "
|
|
|
|
|
+ "Journal indicates no msgs when positioned at head.\n");
|
|
|
|
|
+ }
|
|
|
|
|
/* No new messages, wait for activity. */
|
|
|
|
|
- if (pollJournal() != RS_RET_OK && !reloaded) {
|
|
|
|
|
+ if (pollJournal() != RS_RET_OK && !journalContext.reloaded) {
|
|
|
|
|
tryRecover();
|
|
|
|
|
}
|
|
|
|
|
continue;
|
|
|
|
|
@@ -776,7 +839,7 @@ CODESTARTrunInput
|
|
|
|
|
/*
|
|
|
|
|
* update journal disk usage before reading the new message.
|
|
|
|
|
*/
|
|
|
|
|
- const int e = sd_journal_get_usage(j, (uint64_t *)&statsCounter.diskUsageBytes);
|
|
|
|
|
+ const int e = sd_journal_get_usage(journalContext.j, (uint64_t *)&statsCounter.diskUsageBytes);
|
|
|
|
|
if (e < 0) {
|
|
|
|
|
LogError(-e, RS_RET_ERR, "imjournal: sd_get_usage() failed");
|
|
|
|
|
}
|
|
|
|
|
@@ -787,21 +850,12 @@ CODESTARTrunInput
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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) {
|
|
|
|
|
- LogError(-r, RS_RET_ERR, "imjournal: sd_journal_process() failed");
|
|
|
|
|
- tryRecover();
|
|
|
|
|
- continue;
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
+ journalContext.atHead = 0;
|
|
|
|
|
|
|
|
|
|
if (cs.stateFile) { /* can't persist without a state file */
|
|
|
|
|
/* TODO: This could use some finer metric. */
|
|
|
|
|
if ((count % cs.iPersistStateInterval) == 0) {
|
|
|
|
|
- persistJournalState(1);
|
|
|
|
|
+ persistJournalState();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@@ -825,6 +879,7 @@ CODESTARTbeginCnfLoad
|
|
|
|
|
cs.usePid = NULL;
|
|
|
|
|
cs.bWorkAroundJournalBug = 1;
|
|
|
|
|
cs.bFsync = 0;
|
|
|
|
|
+ cs.bRemote = 0;
|
|
|
|
|
ENDbeginCnfLoad
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -881,7 +936,7 @@ BEGINfreeCnf
|
|
|
|
|
CODESTARTfreeCnf
|
|
|
|
|
free(cs.stateFile);
|
|
|
|
|
free(cs.usePid);
|
|
|
|
|
- free(last_cursor);
|
|
|
|
|
+ free(journalContext.cursor);
|
|
|
|
|
statsobj.Destruct(&(statsCounter.stats));
|
|
|
|
|
ENDfreeCnf
|
|
|
|
|
|
|
|
|
|
@@ -894,7 +949,10 @@ ENDwillRun
|
|
|
|
|
/* close journal */
|
|
|
|
|
BEGINafterRun
|
|
|
|
|
CODESTARTafterRun
|
|
|
|
|
- closeJournal(1);
|
|
|
|
|
+ if (cs.stateFile) { /* can't persist without a state file */
|
|
|
|
|
+ persistJournalState();
|
|
|
|
|
+ }
|
|
|
|
|
+ closeJournal();
|
|
|
|
|
ratelimitDestruct(ratelimiter);
|
|
|
|
|
ENDafterRun
|
|
|
|
|
|
|
|
|
|
@@ -966,6 +1024,8 @@ CODESTARTsetModCnf
|
|
|
|
|
cs.bWorkAroundJournalBug = (int) pvals[i].val.d.n;
|
|
|
|
|
} else if (!strcmp(modpblk.descr[i].name, "fsync")) {
|
|
|
|
|
cs.bFsync = (int) pvals[i].val.d.n;
|
|
|
|
|
+ } else if (!strcmp(modpblk.descr[i].name, "remote")) {
|
|
|
|
|
+ cs.bRemote = (int) pvals[i].val.d.n;
|
|
|
|
|
} else {
|
|
|
|
|
dbgprintf("imjournal: program error, non-handled "
|
|
|
|
|
"param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
|
|
|
|
|
diff --git a/plugins/imjournal/imjournal.c b/plugins/imjournal/imjournal.c
|
|
|
|
|
index 8008f159c4..18e4d25ebb 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-2017 Adiscon GmbH
|
|
|
|
|
+ * Copyright (C) 2008-2019 Adiscon GmbH
|
|
|
|
|
*
|
|
|
|
|
* This file is part of rsyslog.
|
|
|
|
|
*
|
|
|
|
|
@@ -676,16 +676,6 @@ loadJournalState(void)
|
|
|
|
|
DBGPRINTF("Loading journal position, at head? %d, reloaded? %d\n",
|
|
|
|
|
journalContext.atHead, journalContext.reloaded);
|
|
|
|
|
|
|
|
|
|
- if (cs.stateFile[0] != '/') {
|
|
|
|
|
- char *new_stateFile;
|
|
|
|
|
- if (-1 == asprintf(&new_stateFile, "%s/%s", (char *)glbl.GetWorkDir(), cs.stateFile)) {
|
|
|
|
|
- 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 not exists (on very first run), skip */
|
|
|
|
|
if (access(cs.stateFile, F_OK|R_OK) == -1 && errno == ENOENT) {
|
|
|
|
|
if (cs.bIgnorePrevious) {
|
|
|
|
|
@@ -885,6 +875,17 @@ ENDbeginCnfLoad
|
|
|
|
|
|
|
|
|
|
BEGINendCnfLoad
|
|
|
|
|
CODESTARTendCnfLoad
|
|
|
|
|
+ /* bad trick to handle old and new style config all in old-style var */
|
|
|
|
|
+ if(cs.stateFile != NULL && cs.stateFile[0] != '/') {
|
|
|
|
|
+ char *new_stateFile;
|
|
|
|
|
+ if (-1 == asprintf(&new_stateFile, "%s/%s", (char *)glbl.GetWorkDir(), cs.stateFile)) {
|
|
|
|
|
+ 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;
|
|
|
|
|
+ }
|
|
|
|
|
+finalize_it:
|
|
|
|
|
ENDendCnfLoad
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -1032,7 +1033,6 @@ CODESTARTsetModCnf
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
-
|
|
|
|
|
finalize_it:
|
|
|
|
|
if (pvals != NULL)
|
|
|
|
|
cnfparamvalsDestruct(pvals, &modpblk);
|