summaryrefslogtreecommitdiffstats
path: root/plugins
diff options
context:
space:
mode:
Diffstat (limited to 'plugins')
-rw-r--r--plugins/imfile/imfile.c23
-rw-r--r--plugins/imklog/bsd.c6
-rw-r--r--plugins/imklog/imklog.c52
-rw-r--r--plugins/imklog/imklog.h4
-rw-r--r--plugins/imklog/linux.c113
-rw-r--r--plugins/imklog/solaris.c68
-rw-r--r--plugins/imptcp/imptcp.c105
-rw-r--r--plugins/imtcp/imtcp.c12
-rw-r--r--plugins/imuxsock/imuxsock.c74
-rw-r--r--plugins/mmsnmptrapd/mmsnmptrapd.c2
-rw-r--r--plugins/omhdfs/javaenv.sh14
-rw-r--r--plugins/omhdfs/omhdfs.c85
12 files changed, 426 insertions, 132 deletions
diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c
index cac3a55d..9bc84220 100644
--- a/plugins/imfile/imfile.c
+++ b/plugins/imfile/imfile.c
@@ -64,6 +64,7 @@ DEFobjCurrIf(strm)
DEFobjCurrIf(prop)
DEFobjCurrIf(ruleset)
+#define NUM_MULTISUB 1024 /* max number of submits -- TODO: make configurable */
typedef struct fileInfo_s {
uchar *pszFileName;
uchar *pszTag;
@@ -71,11 +72,13 @@ typedef struct fileInfo_s {
uchar *pszStateFile; /* file in which state between runs is to be stored */
int iFacility;
int iSeverity;
+ int maxLinesAtOnce;
int nRecords; /**< How many records did we process before persisting the stream? */
int iPersistStateInterval; /**< how often should state be persisted? (0=on close only) */
strm_t *pStrm; /* its stream (NULL if not assigned) */
int readMode; /* which mode to use in ReadMulteLine call? */
ruleset_t *pRuleset; /* ruleset to bind listener to (use system default if unspecified) */
+ multi_submit_t multiSub;
} fileInfo_t;
@@ -91,6 +94,7 @@ static int iPersistStateInterval = 0; /* how often if state file to be persisted
static int iFacility = 128; /* local0 */
static int iSeverity = 5; /* notice, as of rfc 3164 */
static int readMode = 0; /* mode to use for ReadMultiLine call */
+static int maxLinesAtOnce = 10240; /* how many lines to process in a row? */
static ruleset_t *pBindRuleset = NULL; /* ruleset to bind listener to (use system default if unspecified) */
static int iFilPtr = 0; /* number of files to be monitored; pointer to next free spot during config */
@@ -122,7 +126,9 @@ static rsRetVal enqLine(fileInfo_t *pInfo, cstr_t *cstrLine)
pMsg->iFacility = LOG_FAC(pInfo->iFacility);
pMsg->iSeverity = LOG_PRI(pInfo->iSeverity);
MsgSetRuleset(pMsg, pInfo->pRuleset);
- CHKiRet(submitMsg(pMsg));
+ pInfo->multiSub.ppMsgs[pInfo->multiSub.nElem++] = pMsg;
+ if(pInfo->multiSub.nElem == pInfo->multiSub.maxElem)
+ CHKiRet(multiSubmitMsg(&pInfo->multiSub));
finalize_it:
RETiRet;
}
@@ -206,6 +212,7 @@ static void pollFileCancelCleanup(void *pArg)
static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
{
cstr_t *pCStr = NULL;
+ int nProcessed = 0;
DEFiRet;
ASSERT(pbHadFileData != NULL);
@@ -220,7 +227,10 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
/* loop below will be exited when strmReadLine() returns EOF */
while(glbl.GetGlobalInputTermState() == 0) {
+ if(pThis->maxLinesAtOnce != 0 && nProcessed >= pThis->maxLinesAtOnce)
+ break;
CHKiRet(strm.ReadLine(pThis->pStrm, &pCStr, pThis->readMode));
+ ++nProcessed;
*pbHadFileData = 1; /* this is just a flag, so set it and forget it */
CHKiRet(enqLine(pThis, pCStr)); /* process line */
rsCStrDestruct(&pCStr); /* discard string (must be done by us!) */
@@ -231,6 +241,10 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData)
}
finalize_it:
+ if(pThis->multiSub.nElem > 0) {
+ /* submit everything that was not yet submitted */
+ CHKiRet(multiSubmitMsg(&pThis->multiSub));
+ }
; /*EMPTY STATEMENT - needed to keep compiler happy - see below! */
/* Note: the problem above is that pthread:cleanup_pop() is a macro which
* evaluates to something like "} while(0);". So the code would become
@@ -475,6 +489,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
iSeverity = 5; /* notice, as of rfc 3164 */
readMode = 0;
pBindRuleset = NULL;
+ maxLinesAtOnce = 10240;
RETiRet;
}
@@ -513,8 +528,12 @@ static rsRetVal addMonitor(void __attribute__((unused)) *pVal, uchar *pNewVal)
pThis->pszStateFile = (uchar*) strdup((char*) pszStateFile);
}
+ CHKmalloc(pThis->multiSub.ppMsgs = MALLOC(NUM_MULTISUB * sizeof(msg_t*)));
+ pThis->multiSub.maxElem = NUM_MULTISUB;
+ pThis->multiSub.nElem = 0;
pThis->iSeverity = iSeverity;
pThis->iFacility = iFacility;
+ pThis->maxLinesAtOnce = maxLinesAtOnce;
pThis->iPersistStateInterval = iPersistStateInterval;
pThis->nRecords = 0;
pThis->readMode = readMode;
@@ -592,6 +611,8 @@ CODEmodInit_QueryRegCFSLineHdlr
NULL, &iPollInterval, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilereadmode", 0, eCmdHdlrInt,
NULL, &readMode, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilemaxlinesatonce", 0, eCmdHdlrSize,
+ NULL, &maxLinesAtOnce, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilepersiststateinterval", 0, eCmdHdlrInt,
NULL, &iPersistStateInterval, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputfilebindruleset", 0, eCmdHdlrGetWord,
diff --git a/plugins/imklog/bsd.c b/plugins/imklog/bsd.c
index 0a4c7cd4..930bbd11 100644
--- a/plugins/imklog/bsd.c
+++ b/plugins/imklog/bsd.c
@@ -155,18 +155,18 @@ readklog(void)
for (p = (char*)pRcv; (q = strchr(p, '\n')) != NULL; p = q + 1) {
*q = '\0';
- Syslog(LOG_INFO, (uchar*) p);
+ Syslog(LOG_INFO, (uchar*) p, NULL);
}
len = strlen(p);
if (len >= iMaxLine - 1) {
- Syslog(LOG_INFO, (uchar*)p);
+ Syslog(LOG_INFO, (uchar*)p, NULL);
len = 0;
}
if (len > 0)
memmove(pRcv, p, len + 1);
}
if (len > 0)
- Syslog(LOG_INFO, pRcv);
+ Syslog(LOG_INFO, pRcv, NULL);
if(pRcv != NULL && (size_t) iMaxLine >= sizeof(bufRcv) - 1)
free(pRcv);
diff --git a/plugins/imklog/imklog.c b/plugins/imklog/imklog.c
index 69c8cd1a..cb28e68e 100644
--- a/plugins/imklog/imklog.c
+++ b/plugins/imklog/imklog.c
@@ -18,7 +18,10 @@
* Please note that this file replaces the klogd daemon that was
* also present in pre-v3 versions of rsyslog.
*
- * Copyright (C) 2008, 2009 by Rainer Gerhards and Adiscon GmbH
+ * To test under Linux:
+ * echo test1 > /dev/kmsg
+ *
+ * Copyright (C) 2008-2011 by Rainer Gerhards and Adiscon GmbH
*
* This file is part of rsyslog.
*
@@ -93,15 +96,21 @@ static prop_t *pLocalHostIP = NULL; /* a pseudo-constant propterty for 127.0.0.1
* rgerhards, 2008-04-12
*/
static rsRetVal
-enqMsg(uchar *msg, uchar* pszTag, int iFacility, int iSeverity)
+enqMsg(uchar *msg, uchar* pszTag, int iFacility, int iSeverity, struct timeval *tp)
{
- DEFiRet;
+ struct syslogTime st;
msg_t *pMsg;
+ DEFiRet;
assert(msg != NULL);
assert(pszTag != NULL);
- CHKiRet(msgConstruct(&pMsg));
+ if(tp == NULL) {
+ CHKiRet(msgConstruct(&pMsg));
+ } else {
+ datetime.timeval2syslogTime(tp, &st);
+ CHKiRet(msgConstructWithTime(&pMsg, &st, tp->tv_sec));
+ }
MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
MsgSetInputName(pMsg, pInputName);
MsgSetRawMsgWOSize(pMsg, (char*)msg);
@@ -174,31 +183,48 @@ rsRetVal imklogLogIntMsg(int priority, char *fmt, ...)
va_end(ap);
iRet = enqMsg((uchar*)pLogMsg, (uchar*) ((iFacilIntMsg == LOG_KERN) ? "kernel:" : "imklog:"),
- iFacilIntMsg, LOG_PRI(priority));
+ iFacilIntMsg, LOG_PRI(priority), NULL);
RETiRet;
}
-/* log a kernel message
+/* log a kernel message. If tp is non-NULL, it contains the message creation
+ * time to use.
* rgerhards, 2008-04-14
*/
-rsRetVal Syslog(int priority, uchar *pMsg)
+rsRetVal Syslog(int priority, uchar *pMsg, struct timeval *tp)
{
- DEFiRet;
+ int pri = -1;
rsRetVal localRet;
+ DEFiRet;
- /* Output using syslog */
- localRet = parsePRI(&pMsg, &priority);
- if(localRet != RS_RET_INVALID_PRI && localRet != RS_RET_OK)
- FINALIZE;
+ /* then check if we have two PRIs. This can happen in case of systemd,
+ * in which case the second PRI is the rigth one.
+ * TODO: added kernel timestamp support to this PoC. -- rgerhards, 2011-03-18
+ */
+ if(pMsg[3] == '<') { /* could be a pri... */
+ uchar *pMsgTmp = pMsg + 3;
+ localRet = parsePRI(&pMsgTmp, &pri);
+ if(localRet == RS_RET_OK && pri >= 8 && pri <= 192) {
+ /* *this* is our PRI */
+ DBGPRINTF("imklog detected secondary PRI in klog msg\n");
+ pMsg = pMsgTmp;
+ priority = pri;
+ }
+ }
+ if(pri == -1) {
+ localRet = parsePRI(&pMsg, &priority);
+ if(localRet != RS_RET_INVALID_PRI && localRet != RS_RET_OK)
+ FINALIZE;
+ }
/* if we don't get the pri, we use whatever we were supplied */
/* ignore non-kernel messages if not permitted */
if(bPermitNonKernel == 0 && LOG_FAC(priority) != LOG_KERN)
FINALIZE; /* silently ignore */
- iRet = enqMsg((uchar*)pMsg, (uchar*) "kernel:", LOG_FAC(priority), LOG_PRI(priority));
+ iRet = enqMsg((uchar*)pMsg, (uchar*) "kernel:", LOG_FAC(priority), LOG_PRI(priority), tp);
finalize_it:
RETiRet;
diff --git a/plugins/imklog/imklog.h b/plugins/imklog/imklog.h
index c183026d..39bdeb5c 100644
--- a/plugins/imklog/imklog.h
+++ b/plugins/imklog/imklog.h
@@ -5,7 +5,7 @@
* Major change: 2008-04-09: switched to a driver interface for
* several platforms
*
- * Copyright 2007-2008 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -56,7 +56,7 @@ extern uchar *pszPath;
/* the functions below may be called by the drivers */
rsRetVal imklogLogIntMsg(int priority, char *fmt, ...) __attribute__((format(printf,2, 3)));
-rsRetVal Syslog(int priority, uchar *msg);
+rsRetVal Syslog(int priority, uchar *msg, struct timeval *tp);
/* prototypes */
extern int klog_getMaxLine(void); /* work-around for klog drivers to get configured max line size */
diff --git a/plugins/imklog/linux.c b/plugins/imklog/linux.c
index 727708a5..b44619f5 100644
--- a/plugins/imklog/linux.c
+++ b/plugins/imklog/linux.c
@@ -28,6 +28,8 @@
#include "rsyslog.h"
#include <stdlib.h>
#include <stdio.h>
+#include <ctype.h>
+#include <time.h>
#include <assert.h>
#include <signal.h>
#include <string.h>
@@ -181,6 +183,93 @@ static int copyin( uchar *line, int space,
return(i);
}
+
+/* submit a message to imklog Syslog() API. In this function, we check if
+ * a kernel timestamp is present and, if so, extract and strip it.
+ * Note: this is an extra processing step. We should revisit the whole
+ * idea in v6 and remove all that old stuff that we do not longer need
+ * (like symbol resolution). <-- TODO
+ * Special thanks to Lennart Poettering for suggesting on how to convert
+ * the kernel timestamp to a realtime timestamp. This method depends on
+ * the fact the the kernel timestamp is written using the monotonic clock.
+ * Shall that change (very unlikely), this code must be changed as well. Note
+ * that due to the way we generate the delta, we are unable to write the
+ * absolutely correc timestamp (system call overhead of the clock calls
+ * prevents us from doing so). However, the difference is very minor.
+ * rgerhards, 201106-24
+ */
+static void
+submitSyslog(int pri, uchar *buf)
+{
+ long secs;
+ long nsecs;
+ long secOffs;
+ long nsecOffs;
+ unsigned i;
+ unsigned bufsize;
+ struct timespec monotonic, realtime;
+ struct timeval tv;
+ struct timeval *tp = NULL;
+
+ if(buf[3] != '[')
+ goto done;
+ DBGPRINTF("imklog: kernel timestamp detected, extracting it\n");
+
+ /* we now try to parse the timestamp. iff it parses, we assume
+ * it is a timestamp. Otherwise we know for sure it is no ts ;)
+ */
+ i = 4; /* first digit after '[' */
+ secs = 0;
+ while(buf[i] && isdigit(buf[i])) {
+ secs = secs * 10 + buf[i] - '0';
+ ++i;
+ }
+ if(buf[i] != '.') {
+ DBGPRINTF("no dot --> no kernel timestamp\n");
+ goto done; /* no TS! */
+ }
+
+ ++i; /* skip dot */
+ nsecs = 0;
+ while(buf[i] && isdigit(buf[i])) {
+ nsecs = nsecs * 10 + buf[i] - '0';
+ ++i;
+ }
+ if(buf[i] != ']') {
+ DBGPRINTF("no trailing ']' --> no kernel timestamp\n");
+ goto done; /* no TS! */
+ }
+ ++i; /* skip ']' */
+
+ /* we have a timestamp */
+ DBGPRINTF("kernel timestamp is %ld %ld\n", secs, nsecs);
+ bufsize= strlen((char*)buf);
+ memcpy(buf+3, buf+i, bufsize - i + 1);
+
+ clock_gettime(CLOCK_MONOTONIC, &monotonic);
+ clock_gettime(CLOCK_REALTIME, &realtime);
+ secOffs = realtime.tv_sec - monotonic.tv_sec;
+ nsecOffs = realtime.tv_nsec - monotonic.tv_nsec;
+ if(nsecOffs < 0) {
+ secOffs--;
+ nsecOffs += 1000000000l;
+ }
+
+ nsecs +=nsecOffs;
+ if(nsecs > 999999999l) {
+ secs++;
+ nsecs -= 1000000000l;
+ }
+ secs += secOffs;
+ tv.tv_sec = secs;
+ tv.tv_usec = nsecs / 1000;
+ tp = &tv;
+
+done:
+ Syslog(pri, buf, tp);
+}
+
+
/*
* Messages are separated by "\n". Messages longer than
* LOG_LINE_LENGTH are broken up.
@@ -235,7 +324,7 @@ static void LogLine(char *ptr, int len)
//dbgprintf("Line buffer full:\n");
//dbgprintf("\tLine: %s\n", line);
- Syslog(LOG_INFO, line_buff);
+ submitSyslog(LOG_INFO, line_buff);
line = line_buff;
space = sizeof(line_buff)-1;
parse_state = PARSING_TEXT;
@@ -254,28 +343,24 @@ static void LogLine(char *ptr, int len)
space -= delta;
len -= delta;
- if( space == 0 || len == 0 )
- {
+ if( space == 0 || len == 0 ) {
break; /* full line_buff or end of input buffer */
}
- if( *ptr == '\0' ) /* zero byte */
- {
+ if( *ptr == '\0' ) /* zero byte */ {
ptr++; /* skip zero byte */
space -= 1;
len -= 1;
-
break;
}
- if( *ptr == '\n' ) /* newline */
- {
+ if( *ptr == '\n' ) /* newline */ {
ptr++; /* skip newline */
space -= 1;
len -= 1;
*line = 0; /* force null terminator */
- Syslog(LOG_INFO, line_buff);
+ submitSyslog(LOG_INFO, line_buff);
line = line_buff;
space = sizeof(line_buff)-1;
if (symbols_twice) {
@@ -285,9 +370,7 @@ static void LogLine(char *ptr, int len)
skip_symbol_lookup = 1;
ptr = save_ptr;
len = save_len;
- }
- else
- {
+ } else {
skip_symbol_lookup = 0;
save_ptr = ptr;
save_len = len;
@@ -295,8 +378,7 @@ static void LogLine(char *ptr, int len)
}
break;
}
- if( *ptr == '[' ) /* possible kernel symbol */
- {
+ if( *ptr == '[' ) /* possible kernel symbol */ {
*line++ = *ptr++;
space -= 1;
len -= 1;
@@ -310,8 +392,7 @@ static void LogLine(char *ptr, int len)
break;
case PARSING_SYMSTART:
- if( *ptr != '<' )
- {
+ if( *ptr != '<' ) {
parse_state = PARSING_TEXT; /* not a symbol */
break;
}
diff --git a/plugins/imklog/solaris.c b/plugins/imklog/solaris.c
index 8a6d5af1..0a169cdd 100644
--- a/plugins/imklog/solaris.c
+++ b/plugins/imklog/solaris.c
@@ -80,74 +80,6 @@ klogWillRun(void)
}
-#if 0
-/* Read /dev/klog while data are available, split into lines.
- * Contrary to standard BSD syslogd, we do a blocking read. We can
- * afford this as imklog is running on its own threads. So if we have
- * a single file, it really doesn't matter if we wait inside a 1-file
- * select or the read() directly.
- */
-static void
-readklog(void)
-{
- char *p, *q;
- int len, i;
- int iMaxLine;
- uchar bufRcv[4096+1];
- uchar *pRcv = NULL; /* receive buffer */
-
- iMaxLine = klog_getMaxLine();
-
- /* we optimize performance: if iMaxLine is below 4K (which it is in almost all
- * cases, we use a fixed buffer on the stack. Only if it is higher, heap memory
- * is used. We could use alloca() to achive a similar aspect, but there are so
- * many issues with alloca() that I do not want to take that route.
- * rgerhards, 2008-09-02
- */
- if((size_t) iMaxLine < sizeof(bufRcv) - 1) {
- pRcv = bufRcv;
- } else {
- if((pRcv = (uchar*) malloc(sizeof(uchar) * (iMaxLine + 1))) == NULL)
- iMaxLine = sizeof(bufRcv) - 1; /* better this than noting */
- }
-
- len = 0;
- for (;;) {
- dbgprintf("----------imklog(BSD) waiting for kernel log line\n");
- i = read(fklog, pRcv + len, iMaxLine - len);
- if (i > 0) {
- pRcv[i + len] = '\0';
- } else {
- if (i < 0 && errno != EINTR && errno != EAGAIN) {
- imklogLogIntMsg(LOG_ERR,
- "imklog error %d reading kernel log - shutting down imklog",
- errno);
- fklog = -1;
- }
- break;
- }
-
- for(p = pRcv; (q = strchr(p, '\n')) != NULL; p = q + 1) {
- *q = '\0';
- Syslog(LOG_INFO, (uchar*) p);
- }
- len = strlen(p);
- if (len >= iMaxLine - 1) {
- Syslog(LOG_INFO, (uchar*)p);
- len = 0;
- }
- if (len > 0)
- memmove(pRcv, p, len + 1);
- }
- if (len > 0)
- Syslog(LOG_INFO, pRcv);
-
- if(pRcv != NULL && (size_t) iMaxLine >= sizeof(bufRcv) - 1)
- free(pRcv);
-}
-#endif
-
-
/* to be called in the module's AfterRun entry point
* rgerhards, 2008-04-09
*/
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index e8cffbb0..8751637d 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -50,6 +50,7 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
+#include <netinet/tcp.h>
#if HAVE_FCNTL_H
#include <fcntl.h>
#endif
@@ -88,6 +89,10 @@ DEFobjCurrIf(ruleset)
/* config settings */
typedef struct configSettings_s {
+ int bKeepAlive; /* support keep-alive packets */
+ int iKeepAliveIntvl;
+ int iKeepAliveProbes;
+ int iKeepAliveTime;
int bEmitMsgOnClose; /* emit an informational message on close by remote peer */
int iAddtlFrameDelim; /* addtl frame delimiter, e.g. for netscreen, default none */
uchar *pszInputName; /* value for inputname property, NULL is OK and handled by core engine */
@@ -111,13 +116,17 @@ struct ptcpsrv_s {
ptcpsrv_t *pNext; /* linked list maintenance */
uchar *port; /* Port to listen to */
uchar *lstnIP; /* which IP we should listen on? */
- int bEmitMsgOnClose;
int iAddtlFrameDelim;
+ int iKeepAliveIntvl;
+ int iKeepAliveProbes;
+ int iKeepAliveTime;
uchar *pszInputName;
prop_t *pInputName; /* InputName in (fast to process) property format */
ruleset_t *pRuleset;
ptcplstn_t *pLstn; /* root of our listeners */
ptcpsess_t *pSess; /* root of our sessions */
+ sbool bKeepAlive; /* support keep-alive packets */
+ sbool bEmitMsgOnClose;
};
/* the ptcp session object. Describes a single active session.
@@ -429,12 +438,80 @@ finalize_it:
}
+/* Enable KEEPALIVE handling on the socket. */
+static inline rsRetVal
+EnableKeepAlive(ptcplstn_t *pLstn, int sock)
+{
+ int ret;
+ int optval;
+ socklen_t optlen;
+ DEFiRet;
+
+ optval = 1;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen);
+ if(ret < 0) {
+ dbgprintf("EnableKeepAlive socket call returns error %d\n", ret);
+ ABORT_FINALIZE(RS_RET_ERR);
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveProbes > 0) {
+ optval = pLstn->pSrv->iKeepAliveProbes;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPCNT, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(ret, NO_ERRCODE, "imptcp cannot set keepalive probes - ignored");
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveTime > 0) {
+ optval = pLstn->pSrv->iKeepAliveTime;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPIDLE, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(ret, NO_ERRCODE, "imptcp cannot set keepalive time - ignored");
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveIntvl > 0) {
+ optval = pLstn->pSrv->iKeepAliveIntvl;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPINTVL, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(errno, NO_ERRCODE, "imptcp cannot set keepalive intvl - ignored");
+ }
+
+ dbgprintf("KEEPALIVE enabled for socket %d\n", sock);
+
+finalize_it:
+ RETiRet;
+}
+
/* accept an incoming connection request
* rgerhards, 2008-04-22
*/
static rsRetVal
-AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP)
+AcceptConnReq(ptcplstn_t *pLstn, int *newSock, prop_t **peerName, prop_t **peerIP)
{
int sockflags;
struct sockaddr_storage addr;
@@ -443,13 +520,17 @@ AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP)
DEFiRet;
- iNewSock = accept(sock, (struct sockaddr*) &addr, &addrlen);
+ iNewSock = accept(pLstn->sock, (struct sockaddr*) &addr, &addrlen);
if(iNewSock < 0) {
if(errno == EAGAIN || errno == EWOULDBLOCK)
ABORT_FINALIZE(RS_RET_NO_MORE_DATA);
ABORT_FINALIZE(RS_RET_ACCEPT_ERR);
}
+ if(pLstn->pSrv->bKeepAlive)
+ EnableKeepAlive(pLstn, iNewSock);/* we ignore errors, best to do! */
+
+
CHKiRet(getPeerNames(peerName, peerIP, (struct sockaddr*) &addr));
/* set the new socket to non-blocking IO */
@@ -883,6 +964,10 @@ static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVa
CHKmalloc(pSrv = malloc(sizeof(ptcpsrv_t)));
pSrv->pSess = NULL;
pSrv->pLstn = NULL;
+ pSrv->bKeepAlive = cs.bKeepAlive;
+ pSrv->iKeepAliveIntvl = cs.iKeepAliveTime;
+ pSrv->iKeepAliveProbes = cs.iKeepAliveProbes;
+ pSrv->iKeepAliveTime = cs.iKeepAliveTime;
pSrv->bEmitMsgOnClose = cs.bEmitMsgOnClose;
pSrv->port = pNewVal;
pSrv->iAddtlFrameDelim = cs.iAddtlFrameDelim;
@@ -946,7 +1031,7 @@ lstnActivity(ptcplstn_t *pLstn)
DBGPRINTF("imptcp: new connection on listen socket %d\n", pLstn->sock);
while(1) {
- localRet = AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP);
+ localRet = AcceptConnReq(pLstn, &newSock, &peerName, &peerIP);
if(localRet == RS_RET_NO_MORE_DATA)
break;
CHKiRet(localRet);
@@ -1145,6 +1230,10 @@ static rsRetVal
resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
{
cs.bEmitMsgOnClose = 0;
+ cs.bKeepAlive = 0;
+ cs.iKeepAliveProbes = 0;
+ cs.iKeepAliveTime = 0;
+ cs.iKeepAliveIntvl = 0;
cs.iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
free(cs.pszInputName);
cs.pszInputName = NULL;
@@ -1177,6 +1266,14 @@ CODEmodInit_QueryRegCFSLineHdlr
/* register config file handlers */
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverrun"), 0, eCmdHdlrGetWord,
addTCPListener, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive"), 0, eCmdHdlrBinary,
+ NULL, &cs.bKeepAlive, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_probes"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveProbes, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_time"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveTime, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_intvl"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveIntvl, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpservernotifyonconnectionclose"), 0,
eCmdHdlrBinary, NULL, &cs.bEmitMsgOnClose, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserveraddtlframedelimiter"), 0, eCmdHdlrInt,
diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c
index d3e9cabe..c939e1d6 100644
--- a/plugins/imtcp/imtcp.c
+++ b/plugins/imtcp/imtcp.c
@@ -3,7 +3,7 @@
*
* File begun on 2007-12-21 by RGerhards (extracted from syslogd.c)
*
- * Copyright 2007, 2009 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -82,12 +82,14 @@ static permittedPeers_t *pPermPeersRoot = NULL;
/* config settings */
+static int bKeepAlive = 0; /* support keep-alive packets */
static int iTCPSessMax = 200; /* max number of sessions */
static int iTCPLstnMax = 20; /* max number of sessions */
static int iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
static int bEmitMsgOnClose = 0; /* emit an informational message on close by remote peer */
static int iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER; /* addtl frame delimiter, e.g. for netscreen, default none */
static int bDisableLFDelim = 0; /* disbale standard LF delimiter */
+static int bUseFlowControl = 1; /* use flow control, what means indicate ourselfs a "light delayable" */
static uchar *pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
static uchar *pszInputName = NULL; /* value for inputname property, NULL is OK and handled by core engine */
static ruleset_t *pBindRuleset = NULL; /* ruleset to bind listener to (use system default if unspecified) */
@@ -191,6 +193,7 @@ static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVa
if(pOurTcpsrv == NULL) {
CHKiRet(tcpsrv.Construct(&pOurTcpsrv));
+ CHKiRet(tcpsrv.SetKeepAlive(pOurTcpsrv, bKeepAlive));
CHKiRet(tcpsrv.SetSessMax(pOurTcpsrv, iTCPSessMax));
CHKiRet(tcpsrv.SetLstnMax(pOurTcpsrv, iTCPLstnMax));
CHKiRet(tcpsrv.SetCBIsPermittedHost(pOurTcpsrv, isPermittedHost));
@@ -199,6 +202,7 @@ static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVa
CHKiRet(tcpsrv.SetCBOnRegularClose(pOurTcpsrv, onRegularClose));
CHKiRet(tcpsrv.SetCBOnErrClose(pOurTcpsrv, onErrClose));
CHKiRet(tcpsrv.SetDrvrMode(pOurTcpsrv, iStrmDrvrMode));
+ CHKiRet(tcpsrv.SetUseFlowControl(pOurTcpsrv, bUseFlowControl));
CHKiRet(tcpsrv.SetAddtlFrameDelim(pOurTcpsrv, iAddtlFrameDelim));
CHKiRet(tcpsrv.SetbDisableLFDelim(pOurTcpsrv, bDisableLFDelim));
CHKiRet(tcpsrv.SetNotificationOnRemoteClose(pOurTcpsrv, bEmitMsgOnClose));
@@ -287,8 +291,10 @@ static rsRetVal
resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
{
iTCPSessMax = 200;
+ bKeepAlive = 0;
iTCPLstnMax = 20;
iStrmDrvrMode = 0;
+ bUseFlowControl = 0;
bEmitMsgOnClose = 0;
iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
bDisableLFDelim = 0;
@@ -324,6 +330,8 @@ CODEmodInit_QueryRegCFSLineHdlr
/* register config file handlers */
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpserverrun"), 0, eCmdHdlrGetWord,
addTCPListener, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpserverkeepalive"), 0, eCmdHdlrBinary,
+ NULL, &bKeepAlive, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpmaxsessions"), 0, eCmdHdlrInt,
NULL, &iTCPSessMax, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpmaxlisteners"), 0, eCmdHdlrInt,
@@ -344,6 +352,8 @@ CODEmodInit_QueryRegCFSLineHdlr
eCmdHdlrGetWord, NULL, &pszInputName, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpserverbindruleset"), 0,
eCmdHdlrGetWord, setRuleset, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputtcpflowcontrol"), 0,
+ eCmdHdlrBinary, NULL, &bUseFlowControl, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("resetconfigvariables"), 1, eCmdHdlrCustomHandler,
resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
ENDmodInit
diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c
index af034b9f..da9eb993 100644
--- a/plugins/imuxsock/imuxsock.c
+++ b/plugins/imuxsock/imuxsock.c
@@ -6,7 +6,7 @@
*
* File begun on 2007-12-20 by RGerhards (extracted from syslogd.c)
*
- * Copyright 2007-2010 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2011 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -136,6 +136,7 @@ typedef struct lstn_s {
sbool bCreatePath; /* auto-creation of socket directory? */
sbool bUseCreds; /* pull original creator credentials from socket */
sbool bWritePid; /* write original PID into tag */
+ sbool bUseSysTimeStamp; /* use timestamp from system (instead of from message) */
} lstn_t;
static lstn_t listeners[MAXFUNIX];
@@ -156,6 +157,8 @@ static int bUseFlowCtl = 0; /* use flow control or not (if yes, only LIGHT is u
static int bIgnoreTimestamp = 1; /* ignore timestamps present in the incoming message? */
static int bWritePid = 0; /* use credentials from recvmsg() and fixup PID in TAG */
static int bWritePidSysSock = 0; /* use credentials from recvmsg() and fixup PID in TAG */
+static int bUseSysTimeStamp = 1; /* use timestamp from system (rather than from message) */
+static int bUseSysTimeStampSysSock = 1; /* same, for system log socket */
#define DFLT_bCreatePath 0
static int bCreatePath = DFLT_bCreatePath; /* auto-create socket path? */
#define DFLT_ratelimitInterval 5
@@ -302,6 +305,7 @@ addLstnSocketName(void __attribute__((unused)) *pVal, uchar *pNewVal)
listeners[nfd].sockName = pNewVal;
listeners[nfd].bUseCreds = (bWritePid || ratelimitInterval) ? 1 : 0;
listeners[nfd].bWritePid = bWritePid;
+ listeners[nfd].bUseSysTimeStamp = bUseSysTimeStamp;
nfd++;
} else {
errmsg.LogError(0, NO_ERRCODE, "Out of unix socket name descriptors, ignoring %s\n",
@@ -415,6 +419,10 @@ openLogSocket(lstn_t *pLstn)
errmsg.LogError(errno, NO_ERRCODE, "set SCM_CREDENTIALS failed on '%s'", pLstn->sockName);
pLstn->bUseCreds = 0;
}
+// TODO: move to its own #if
+ if(setsockopt(pLstn->fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) != 0) {
+ errmsg.LogError(errno, NO_ERRCODE, "set SO_TIMESTAMP failed on '%s'", pLstn->sockName);
+ }
}
# else /* HAVE_SCM_CREDENTIALS */
pLstn->bUseCreds = 0;
@@ -505,7 +513,7 @@ fixPID(uchar *bufTAG, int *lenTag, struct ucred *cred)
* can also mangle it if necessary.
*/
static inline rsRetVal
-SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred)
+SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct timeval *ts)
{
msg_t *pMsg;
int lenMsg;
@@ -544,7 +552,13 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred)
findRatelimiter(pLstn, cred, &ratelimiter); /* ignore error, better so than others... */
}
- datetime.getCurrTime(&st, &tt);
+ if(ts == NULL) {
+ datetime.getCurrTime(&st, &tt);
+ } else {
+ datetime.timeval2syslogTime(ts, &st);
+ tt = ts->tv_sec;
+ }
+
if(ratelimiter != NULL && !withinRatelimit(ratelimiter, tt, cred->pid)) {
STATSCOUNTER_INC(ctrLostRatelimit, mutCtrLostRatelimit);
FINALIZE;
@@ -564,8 +578,26 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred)
parse++; lenMsg--; /* '>' */
- if(datetime.ParseTIMESTAMP3164(&(pMsg->tTIMESTAMP), &parse, &lenMsg) != RS_RET_OK) {
- DBGPRINTF("we have a problem, invalid timestamp in msg!\n");
+ if(ts == NULL) {
+ if(datetime.ParseTIMESTAMP3164(&(pMsg->tTIMESTAMP), &parse, &lenMsg) != RS_RET_OK) {
+ DBGPRINTF("we have a problem, invalid timestamp in msg!\n");
+ }
+ } else { /* if we pulled the time from the system, we need to update the message text */
+ if(lenMsg >= 16) {
+ /* RFC3164 timestamp is 16 bytes long, so assuming a valid stamp,
+ * we can fixup the message. If the part is smaller, the stamp can
+ * not be valid and we do not touch the message. Note that there may
+ * be some scenarios where the message is larg enough but the stamp is
+ * still invalid. In those cases we will destruct part of the message,
+ * but this case is considered extremely unlikely and thus not handled
+ * specifically. -- rgerhards, 2011-06-20
+ */
+ datetime.formatTimestamp3164(&st, (char*)parse, 0);
+ parse[15] = ' '; /* re-write \0 from fromatTimestamp3164 by SP */
+ /* update "counters" to reflect processed timestamp */
+ parse += 16;
+ lenMsg -= 16;
+ }
}
/* pull tag */
@@ -616,6 +648,7 @@ static rsRetVal readSocket(lstn_t *pLstn)
struct cmsghdr *cm;
# endif
struct ucred *cred;
+ struct timeval *ts;
uchar bufRcv[4096+1];
char aux[128];
uchar *pRcv = NULL; /* receive buffer */
@@ -654,21 +687,32 @@ static rsRetVal readSocket(lstn_t *pLstn)
dbgprintf("Message from UNIX socket: #%d\n", pLstn->fd);
if(iRcvd > 0) {
cred = NULL;
-# if HAVE_SCM_CREDENTIALS
- if(pLstn->bUseCreds) {
+ ts = NULL;
+ if(pLstn->bUseCreds || pLstn->bUseSysTimeStamp) {
dbgprintf("XXX: pre CM loop, length of control message %d\n", (int) msgh.msg_controllen);
- for (cm = CMSG_FIRSTHDR(&msgh); cm; cm = CMSG_NXTHDR(&msgh, cm)) {
+ for(cm = CMSG_FIRSTHDR(&msgh); cm; cm = CMSG_NXTHDR(&msgh, cm)) {
dbgprintf("XXX: in CM loop, %d, %d\n", cm->cmsg_level, cm->cmsg_type);
- if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_CREDENTIALS) {
+# if HAVE_SCM_CREDENTIALS
+ if( pLstn->bUseCreds
+ && cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_CREDENTIALS) {
cred = (struct ucred*) CMSG_DATA(cm);
dbgprintf("XXX: got credentials pid %d\n", (int) cred->pid);
break;
}
+# endif /* HAVE_SCM_CREDENTIALS */
+# if HAVE_SO_TIMESTAMP
+ if( pLstn->bUseSysTimeStamp
+ && cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SO_TIMESTAMP) {
+ ts = (struct timeval *)CMSG_DATA(cm);
+ dbgprintf("XXX: got timestamp %ld.%ld\n",
+ (long) ts->tv_sec, (long) ts->tv_usec);
+ break;
+ }
+# endif /* HAVE_SO_TIMESTAMP */
}
dbgprintf("XXX: post CM loop\n");
}
-# endif /* HAVE_SCM_CREDENTIALS */
- CHKiRet(SubmitMsg(pRcv, iRcvd, pLstn, cred));
+ CHKiRet(SubmitMsg(pRcv, iRcvd, pLstn, cred, ts));
} else if(iRcvd < 0 && errno != EINTR) {
char errStr[1024];
rs_strerror_r(errno, errStr, sizeof(errStr));
@@ -780,6 +824,7 @@ CODESTARTwillRun
listeners[0].ratelimitSev = ratelimitSeveritySysSock;
listeners[0].bUseCreds = (bWritePidSysSock || ratelimitIntervalSysSock) ? 1 : 0;
listeners[0].bWritePid = bWritePidSysSock;
+ listeners[0].bUseSysTimeStamp = bUseSysTimeStampSysSock;
sd_fds = sd_listen_fds(0);
if (sd_fds < 0) {
@@ -892,6 +937,8 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
bUseFlowCtl = 0;
bWritePid = 0;
bWritePidSysSock = 0;
+ bUseSysTimeStamp = 1;
+ bUseSysTimeStampSysSock = 1;
bCreatePath = DFLT_bCreatePath;
ratelimitInterval = DFLT_ratelimitInterval;
ratelimitIntervalSysSock = DFLT_ratelimitInterval;
@@ -927,6 +974,7 @@ CODEmodInit_QueryRegCFSLineHdlr
listeners[0].bParseHost = 0;
listeners[0].bUseCreds = 0;
listeners[0].bCreatePath = 0;
+ listeners[0].bUseSysTimeStamp = 1;
/* initialize socket names */
for(i = 1 ; i < MAXFUNIX ; ++i) {
@@ -958,6 +1006,8 @@ CODEmodInit_QueryRegCFSLineHdlr
NULL, &bCreatePath, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputunixlistensocketusepidfromsystem", 0, eCmdHdlrBinary,
NULL, &bWritePid, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"inputunixlistensocketusesystimestamp", 0, eCmdHdlrBinary,
+ NULL, &bUseSysTimeStamp, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"addunixlistensocket", 0, eCmdHdlrGetWord,
addLstnSocketName, NULL, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"imuxsockratelimitinterval", 0, eCmdHdlrInt,
@@ -978,6 +1028,8 @@ CODEmodInit_QueryRegCFSLineHdlr
setSystemLogTimestampIgnore, NULL, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"systemlogsocketflowcontrol", 0, eCmdHdlrBinary,
setSystemLogFlowControl, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"systemlogusesystimestamp", 0, eCmdHdlrBinary,
+ NULL, &bUseSysTimeStampSysSock, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"systemlogusepidfromsystem", 0, eCmdHdlrBinary,
NULL, &bWritePidSysSock, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"systemlogratelimitinterval", 0, eCmdHdlrInt,
diff --git a/plugins/mmsnmptrapd/mmsnmptrapd.c b/plugins/mmsnmptrapd/mmsnmptrapd.c
index 767829d6..b78046ee 100644
--- a/plugins/mmsnmptrapd/mmsnmptrapd.c
+++ b/plugins/mmsnmptrapd/mmsnmptrapd.c
@@ -418,7 +418,7 @@ CODEmodInit_QueryRegCFSLineHdlr
cs.pszTagName = NULL;
cs.pszSeverityMapping = NULL;
- CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmsnmptrapdtag", 0, eCmdHdlrInt,
+ CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmsnmptrapdtag", 0, eCmdHdlrGetWord,
NULL, &cs.pszTagName, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmsnmptrapdseveritymapping", 0, eCmdHdlrGetWord,
NULL, &cs.pszSeverityMapping, STD_LOADABLE_MODULE_ID));
diff --git a/plugins/omhdfs/javaenv.sh b/plugins/omhdfs/javaenv.sh
new file mode 100644
index 00000000..d07a8473
--- /dev/null
+++ b/plugins/omhdfs/javaenv.sh
@@ -0,0 +1,14 @@
+# This is a sample file for environment settings on Fedora 13
+# that made me compile & run omhdfs. I really *hate* the way
+# java uses environment variables... Hopefully this file will
+# help building and testing omhdfs in the future (there is also
+# some more information in the rsyslog wiki).
+# rgerhards, 2011-03-11
+# this now works, but don't ask my why ;)
+#export JAVA_HOME=/usr/java/jdk1.6.0_21/bin/java
+export PATH=/usr/java/jdk1.6.0_21/bin:$PATH
+export JAVA_INCLUDES="-I/usr/java/jdk1.6.0_21/include -I/usr/java/jdk1.6.0_21/include/linux"
+export JAVA_LIBS="-L/usr/java/jdk1.6.0_21/jre/lib/i386 -ljava -ljvm -lverify"
+export HADOOP_HOME=/usr/lib/hadoop
+export CLASSPATH=/usr/lib/jvm/java-6-sun/lib:/usr/lib/hadoop/lib:/usr/lib/hadoop/hadoop-ant-0.20.2+320.jar:/usr/lib/hadoop/hadoop-core-0.20.2+320.jar:/usr/lib/hadoop/hadoop-examples-0.20.2+320.jar:/usr/lib/hadoop/hadoop-test-0.20.2+320.jar:/usr/lib/hadoop/hadoop-tools-0.20.2+320.jar/usr/lib/hadoop/lib/commons-cli-1.2.jar:/usr/lib/hadoop/lib/commons-codec-1.3.jar:/usr/lib/hadoop/lib/commons-el-1.0.jar:/usr/lib/hadoop/lib/commons-httpclient-3.0.1.jar:/usr/lib/hadoop/lib/commons-logging-1.0.4.jar:/usr/lib/hadoop/lib/commons-logging-api-1.0.4.jar:/usr/lib/hadoop/lib/commons-net-1.4.1.jar:/usr/lib/hadoop/lib/core-3.1.1.jar:/usr/lib/hadoop/lib/hadoop-fairscheduler-0.20.2+320.jar:/usr/lib/hadoop/lib/hadoop-scribe-log4j-0.20.2+320.jar:/usr/lib/hadoop/lib/hsqldb-1.8.0.10.jar:/usr/lib/hadoop/lib/hsqldb.jar:/usr/lib/hadoop/lib/jackson-core-asl-1.0.1.jar:/usr/lib/hadoop/lib/jackson-mapper-asl-1.0.1.jar:/usr/lib/hadoop/lib/jasper-compiler-5.5.12.jar:/usr/lib/hadoop/lib/jasper-runtime-5.5.12.jar:/usr/lib/hadoop/lib/jets3t-0.6.1.jar:/usr/lib/hadoop/lib/jetty-6.1.14.jar:/usr/lib/hadoop/lib/jetty-util-6.1.14.jar:/usr/lib/hadoop/lib/junit-4.5.jar:/usr/lib/hadoop/lib/kfs-0.2.2.jar:/usr/lib/hadoop/lib/libfb303.jar:/usr/lib/hadoop/lib/libthrift.jar:/usr/lib/hadoop/lib/log4j-1.2.15.jar:/usr/lib/hadoop/lib/mockito-all-1.8.2.jar:/usr/lib/hadoop/lib/mysql-connector-java-5.0.8-bin.jar:/usr/lib/hadoop/lib/oro-2.0.8.jar:/usr/lib/hadoop/lib/servlet-api-2.5-6.1.14.jar:/usr/lib/hadoop/lib/slf4j-api-1.4.3.jar:/usr/lib/hadoop/lib/slf4j-log4j12-1.4.3.jar:/usr/lib/hadoop/lib/xmlenc-0.52.jar:/etc/hadoop/conf
+###export CLASSPATH="/usr/lib/hadoop/hadoop-0.20.2+320-ant.jar: /usr/lib/hadoop/hadoop-0.20.2+320-core.jar: /usr/lib/hadoop/hadoop-0.20.2+320-examples.jar: /usr/lib/hadoop/hadoop-0.20.2+320-test.jar: /usr/lib/hadoop/hadoop-0.20.2+320-tools.jar: /usr/lib/hadoop/hadoop-ant-0.20.2+320.jar: /usr/lib/hadoop/hadoop-core-0.20.2+320.jar: /usr/lib/hadoop/hadoop-examples-0.20.2+320.jar: /usr/lib/hadoop/hadoop-test-0.20.2+320.jar: /usr/lib/hadoop/hadoop-tools-0.20.2+320.jar:/usr/lib/hadoop/lib: /usr/lib/hadoop/lib/commons-cli-1.2.jar: /usr/lib/hadoop/lib/commons-codec-1.3.jar: /usr/lib/hadoop/lib/commons-el-1.0.jar: /usr/lib/hadoop/lib/commons-httpclient-3.0.1.jar: /usr/lib/hadoop/lib/commons-logging-1.0.4.jar: /usr/lib/hadoop/lib/commons-logging-api-1.0.4.jar: /usr/lib/hadoop/lib/commons-net-1.4.1.jar: /usr/lib/hadoop/lib/core-3.1.1.jar: /usr/lib/hadoop/lib/hadoop-fairscheduler-0.20.2+320.jar: /usr/lib/hadoop/lib/hadoop-scribe-log4j-0.20.2+320.jar: /usr/lib/hadoop/lib/hsqldb-1.8.0.10.jar: /usr/lib/hadoop/lib/hsqldb.jar: /usr/lib/hadoop/lib/jackson-core-asl-1.0.1.jar: /usr/lib/hadoop/lib/jackson-mapper-asl-1.0.1.jar: /usr/lib/hadoop/lib/jasper-compiler-5.5.12.jar: /usr/lib/hadoop/lib/jasper-runtime-5.5.12.jar: /usr/lib/hadoop/lib/jets3t-0.6.1.jar: /usr/lib/hadoop/lib/jetty-6.1.14.jar: /usr/lib/hadoop/lib/jetty-util-6.1.14.jar: /usr/lib/hadoop/lib/junit-4.5.jar: /usr/lib/hadoop/lib/kfs-0.2.2.jar: /usr/lib/hadoop/lib/libfb303.jar: /usr/lib/hadoop/lib/libthrift.jar: /usr/lib/hadoop/lib/log4j-1.2.15.jar: /usr/lib/hadoop/lib/mockito-all-1.8.2.jar: /usr/lib/hadoop/lib/mysql-connector-java-5.0.8-bin.jar: /usr/lib/hadoop/lib/oro-2.0.8.jar: /usr/lib/hadoop/lib/servlet-api-2.5-6.1.14.jar: /usr/lib/hadoop/lib/slf4j-api-1.4.3.jar: /usr/lib/hadoop/lib/slf4j-log4j12-1.4.3.jar: /usr/lib/hadoop/lib/xmlenc-0.52.jar:/etc/hadoop/conf:/usr/lib/hadoop/lib"
diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c
index 8b72747f..76128a4e 100644
--- a/plugins/omhdfs/omhdfs.c
+++ b/plugins/omhdfs/omhdfs.c
@@ -80,6 +80,8 @@ typedef struct {
typedef struct _instanceData {
file_t *pFile;
+ uchar ioBuf[64*1024];
+ unsigned offsBuf;
} instanceData;
/* forward definitions (down here, need data types) */
@@ -260,7 +262,8 @@ fileOpen(file_t *pFile)
if(errno == ENOENT) {
DBGPRINTF("omhdfs: ENOENT trying to append to '%s', now trying create\n",
pFile->name);
- pFile->fh = hdfsOpenFile(pFile->fs, (char*)pFile->name, O_WRONLY|O_CREAT, 0, 0, 0);
+ pFile->fh = hdfsOpenFile(pFile->fs,
+ (char*)pFile->name, O_WRONLY|O_CREAT, 0, 0, 0);
}
}
if(pFile->fh == NULL) {
@@ -275,12 +278,15 @@ finalize_it:
}
+/* Note: lenWrite is reset to zero on successful write! */
static inline rsRetVal
-fileWrite(file_t *pFile, uchar *buf)
+fileWrite(file_t *pFile, uchar *buf, size_t *lenWrite)
{
- size_t lenWrite;
DEFiRet;
+ if(*lenWrite == 0)
+ FINALIZE;
+
if(pFile->nUsers > 1)
d_pthread_mutex_lock(&pFile->mut);
@@ -294,18 +300,18 @@ fileWrite(file_t *pFile, uchar *buf)
}
}
- lenWrite = strlen((char*) buf);
- tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, lenWrite);
- if((unsigned) num_written_bytes != lenWrite) {
- errmsg.LogError(errno, RS_RET_ERR_HDFS_WRITE, "omhdfs: failed to write %s, expected %lu bytes, "
- "written %lu\n", pFile->name, (unsigned long) lenWrite,
+dbgprintf("XXXXX: omhdfs writing %u bytes\n", *lenWrite);
+ tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, *lenWrite);
+ if((unsigned) num_written_bytes != *lenWrite) {
+ errmsg.LogError(errno, RS_RET_ERR_HDFS_WRITE,
+ "omhdfs: failed to write %s, expected %lu bytes, "
+ "written %lu\n", pFile->name, (unsigned long) *lenWrite,
(unsigned long) num_written_bytes);
ABORT_FINALIZE(RS_RET_SUSPENDED);
}
+ *lenWrite = 0;
finalize_it:
- if(pFile->nUsers > 1)
- d_pthread_mutex_unlock(&pFile->mut);
RETiRet;
}
@@ -333,6 +339,40 @@ finalize_it:
/* ---END FILE OBJECT---------------------------------------------------- */
+/* This adds data to the output buffer and performs an actual write
+ * if the new data does not fit into the buffer. Note that we never write
+ * partial data records. Other actions may write into the same file, and if
+ * we would write partial records, data could become severely mixed up.
+ * Note that we must check of some new data arrived is large than our
+ * buffer. In that case, the new data will written with its own
+ * write operation.
+ */
+static inline rsRetVal
+addData(instanceData *pData, uchar *buf)
+{
+ unsigned len;
+ DEFiRet;
+
+ len = strlen((char*)buf);
+ if(pData->offsBuf + len < sizeof(pData->ioBuf)) {
+ /* new data fits into remaining buffer */
+ memcpy((char*) pData->ioBuf + pData->offsBuf, buf, len);
+ pData->offsBuf += len;
+ } else {
+dbgprintf("XXXXX: not enough room, need to flush\n");
+ CHKiRet(fileWrite(pData->pFile, pData->ioBuf, &pData->offsBuf));
+ if(len >= sizeof(pData->ioBuf)) {
+ CHKiRet(fileWrite(pData->pFile, buf, &len));
+ } else {
+ memcpy((char*) pData->ioBuf + pData->offsBuf, buf, len);
+ pData->offsBuf += len;
+ }
+ }
+
+ iRet = RS_RET_DEFER_COMMIT;
+finalize_it:
+ RETiRet;
+}
BEGINcreateInstance
CODESTARTcreateInstance
@@ -358,13 +398,31 @@ CODESTARTtryResume
}
ENDtryResume
+
+BEGINbeginTransaction
+CODESTARTbeginTransaction
+dbgprintf("omhdfs: beginTransaction\n");
+ENDbeginTransaction
+
+
BEGINdoAction
CODESTARTdoAction
- DBGPRINTF("omuxsock: action to to write to %s\n", pData->pFile->name);
- iRet = fileWrite(pData->pFile, ppString[0]);
+ DBGPRINTF("omhdfs: action to to write to %s\n", pData->pFile->name);
+ iRet = addData(pData, ppString[0]);
+dbgprintf("omhdfs: done doAction\n");
ENDdoAction
+BEGINendTransaction
+CODESTARTendTransaction
+dbgprintf("omhdfs: endTransaction\n");
+ if(pData->offsBuf != 0) {
+ DBGPRINTF("omhdfs: data unwritten at end of transaction, persisting...\n");
+ iRet = fileWrite(pData->pFile, pData->ioBuf, &pData->offsBuf);
+ }
+ENDendTransaction
+
+
BEGINparseSelectorAct
file_t *pFile;
int r;
@@ -409,6 +467,7 @@ CODESTARTparseSelectorAct
}
fileObjAddUser(pFile);
pData->pFile = pFile;
+ pData->offsBuf = 0;
CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct
@@ -455,6 +514,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */
CODEqueryEtryPt_doHUP
ENDqueryEtryPt
@@ -472,5 +532,6 @@ CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(regCfSysLineHdlr((uchar *)"omhdfsport", 0, eCmdHdlrInt, NULL, &hdfsPort, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"omhdfsdefaulttemplate", 0, eCmdHdlrGetWord, NULL, &dfltTplName, NULL));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
+ DBGPRINTF("omhdfs: module compiled with rsyslog version %s.\n", VERSION);
CODEmodInit_QueryRegCFSLineHdlr
ENDmodInit
generated by cgit (git 2.34.1) at 2024-06-09 14:44:38 +0000