summaryrefslogtreecommitdiffstats
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to 'tools')
-rw-r--r--tools/omdiscard.c17
-rw-r--r--tools/omfile.c140
-rw-r--r--tools/omfwd.c450
-rw-r--r--tools/ompipe.c25
-rw-r--r--tools/omshell.c25
-rw-r--r--tools/omusrmsg.c19
-rw-r--r--tools/pmrfc3164.c4
-rw-r--r--tools/pmrfc5424.c2
-rw-r--r--tools/rsyslogd.84
-rw-r--r--tools/syslogd.c53
10 files changed, 547 insertions, 192 deletions
diff --git a/tools/omdiscard.c b/tools/omdiscard.c
index 15c6ea82..a76bcc33 100644
--- a/tools/omdiscard.c
+++ b/tools/omdiscard.c
@@ -6,7 +6,7 @@
*
* File begun on 2007-07-24 by RGerhards
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -49,6 +49,10 @@ typedef struct _instanceData {
EMPTY_STRUCT
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
/* we do not need a createInstance()!
BEGINcreateInstance
CODESTARTcreateInstance
@@ -56,6 +60,11 @@ ENDcreateInstance
*/
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
/* do nothing */
@@ -87,6 +96,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINparseSelectorAct
CODESTARTparseSelectorAct
CODE_STD_STRING_REQUESTparseSelectorAct(0)
@@ -114,6 +128,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
ENDqueryEtryPt
diff --git a/tools/omfile.c b/tools/omfile.c
index ba9f7f70..1ee1026c 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -133,6 +133,7 @@ typedef struct s_dynaFileCacheEntry dynaFileCacheEntry;
typedef struct _instanceData {
+ pthread_mutex_t mutWrite; /* guard against multiple instances writing to single file */
uchar *f_fname; /* file or template name (display only) */
uchar *tplName; /* name of assigned template */
strm_t *pStrm; /* our output stream */
@@ -155,7 +156,6 @@ typedef struct _instanceData {
uchar *cryprovName; /* crypto provider */
uchar *cryprovNameFull;/* full internal crypto provider name */
void *cryprovData; /* opaque data ptr for provider use */
- void *cryprovFileData;/* opaque data ptr for file instance */
cryprov_if_t cryprov; /* ptr to crypto provider interface */
sbool useCryprov; /* quicker than checkig ptr (1 vs 8 bytes!) */
int iCurrElt; /* currently active cache element (-1 = none) */
@@ -182,6 +182,20 @@ typedef struct _instanceData {
STATSCOUNTER_DEF(ctrMax, mutCtrMax);
} instanceData;
+/* to build a linked list for temporary storage of lines while we cannot commit */
+typedef struct linebuf {
+ uchar *filename; /* for dynafiles, make go away */
+ uchar *ln;
+ unsigned iMsgOpts;
+ struct linebuf *pNext;
+} linebuf_t;
+
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+ linebuf_t *pRoot;
+ linebuf_t *pLast;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
int iDynaFileCacheSize; /* max cache for dynamic files */
@@ -207,6 +221,8 @@ uchar *pszFileDfltTplName; /* name of the default template to use */
struct modConfData_s {
rsconf_t *pConf; /* our overall config object */
uchar *tplName; /* default template */
+ int fCreateMode; /* default mode to use when creating files */
+ int fDirCreateMode; /* default mode to use when creating files */
};
static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
@@ -216,6 +232,8 @@ static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current ex
/* module-global parameters */
static struct cnfparamdescr modpdescr[] = {
{ "template", eCmdHdlrGetWord, 0 },
+ { "dircreatemode", eCmdHdlrFileCreateMode, 0 },
+ { "filecreatemode", eCmdHdlrFileCreateMode, 0 }
};
static struct cnfparamblk modpblk =
{ CNFPARAMBLK_VERSION,
@@ -783,7 +801,7 @@ finalize_it:
/* rgerhards 2004-11-11: write to a file output. */
static rsRetVal
-writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData)
+writeFile(instanceData *pData, linebuf_t *linebuf)
{
DEFiRet;
@@ -793,7 +811,7 @@ writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData)
* check if it still is ok or a new file needs to be created
*/
if(pData->bDynamicName) {
- CHKiRet(prepareDynFile(pData, ppString[1], iMsgOpts));
+ CHKiRet(prepareDynFile(pData, linebuf->filename, linebuf->iMsgOpts));
} else { /* "regular", non-dynafile */
if(pData->pStrm == NULL) {
CHKiRet(prepareFile(pData, pData->f_fname));
@@ -803,7 +821,7 @@ writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData)
}
}
- CHKiRet(doWrite(pData, ppString[0], strlen(CHAR_CONVERT(ppString[0]))));
+ CHKiRet(doWrite(pData, linebuf->ln, ustrlen(linebuf->ln)));
finalize_it:
RETiRet;
@@ -815,6 +833,8 @@ CODESTARTbeginCnfLoad
loadModConf = pModConf;
pModConf->pConf = pConf;
pModConf->tplName = NULL;
+ pModConf->fCreateMode = 0644;
+ pModConf->fDirCreateMode = 0700;
ENDbeginCnfLoad
BEGINsetModCnf
@@ -843,6 +863,10 @@ CODESTARTsetModCnf
"was already set via legacy directive - may lead to inconsistent "
"results.");
}
+ } else if(!strcmp(modpblk.descr[i].name, "dircreatemode")) {
+ loadModConf->fDirCreateMode = (int) pvals[i].val.d.n;
+ } else if(!strcmp(modpblk.descr[i].name, "filecreatemode")) {
+ loadModConf->fCreateMode = (int) pvals[i].val.d.n;
} else {
dbgprintf("omfile: program error, non-handled "
"param '%s' in beginCnfLoad\n", modpblk.descr[i].name);
@@ -879,9 +903,15 @@ ENDfreeCnf
BEGINcreateInstance
CODESTARTcreateInstance
pData->pStrm = NULL;
+ pthread_mutex_init(&pData->mutWrite, NULL);
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINfreeInstance
CODESTARTfreeInstance
free(pData->tplName);
@@ -904,9 +934,15 @@ CODESTARTfreeInstance
free(pData->cryprovName);
free(pData->cryprovNameFull);
}
+ pthread_mutex_destroy(&pData->mutWrite);
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINtryResume
CODESTARTtryResume
ENDtryResume
@@ -917,8 +953,70 @@ CODESTARTbeginTransaction
ENDbeginTransaction
+static rsRetVal
+bufferLine(wrkrInstanceData_t *pWrkrData, uchar *filename, uchar *line)
+{
+ linebuf_t *lb;
+ DEFiRet;
+
+ CHKmalloc(lb = (linebuf_t*) malloc(sizeof(linebuf_t)));
+dbgprintf("DDDD: filename '%s'\n", filename);
+ CHKmalloc(lb->filename = ustrdup(filename));
+ CHKmalloc(lb->ln = ustrdup(line));
+ lb->pNext = NULL;
+ if(pWrkrData->pRoot == NULL) {
+ pWrkrData->pRoot = pWrkrData->pLast = lb;
+ } else {
+ pWrkrData->pLast->pNext = lb;
+ pWrkrData->pLast = lb;
+ }
+finalize_it:
+ RETiRet;
+}
+
+static void
+submitCachedLines(wrkrInstanceData_t *pWrkrData, instanceData *pData)
+{
+ linebuf_t *curr, *todel;
+
+dbgprintf("omfile: waiting on write lock (pWrkrData %p)\n", pWrkrData);
+ pthread_mutex_lock(&pData->mutWrite);
+dbgprintf("omfile: aquired write lock (pWrkrData %p)\n", pWrkrData);
+
+ for(curr = pWrkrData->pRoot ; curr != NULL ; ) {
+ DBGPRINTF("omfile: file to log to: %s\n", curr->filename);
+ DBGPRINTF("omfile: start of data: '%.128s'\n", curr->ln);
+ STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests);
+ writeFile(pData, curr);
+
+ todel = curr;
+ curr = curr->pNext;
+ free(todel->filename);
+ free(todel->ln);
+ free(todel);
+ }
+ pthread_mutex_unlock(&pData->mutWrite);
+dbgprintf("omfile: free write lock (pWrkrData %p)\n", pWrkrData);
+ pWrkrData->pRoot = NULL;
+}
+
+
+BEGINdoAction
+ instanceData *pData;
+CODESTARTdoAction
+ pData = pWrkrData->pData;
+dbgprintf("DDDD: bDynName %d, filename '%s'\n", pData->bDynamicName, ppString[1]);
+ iRet = bufferLine(pWrkrData, (pData->bDynamicName) ? ppString[1] : pData->f_fname,
+ ppString[0]);
+ if(iRet == RS_RET_OK)
+ iRet = RS_RET_DEFER_COMMIT;
+ENDdoAction
+
BEGINendTransaction
+ instanceData *pData;
CODESTARTendTransaction
+ pData = pWrkrData->pData;
+ submitCachedLines(pWrkrData, pData);
/* Note: pStrm may be NULL if there was an error opening the stream */
if(pData->bFlushOnTXEnd && pData->pStrm != NULL) {
/* if we have an async writer, it controls the flush via
@@ -932,21 +1030,6 @@ finalize_it:
ENDendTransaction
-BEGINdoAction
-CODESTARTdoAction
- DBGPRINTF("file to log to: %s\n",
- (pData->bDynamicName) ? ppString[1] : pData->f_fname);
- DBGPRINTF("omfile: start of data: '%.128s'\n", ppString[0]);
- STATSCOUNTER_INC(pData->ctrRequests, pData->mutCtrRequests);
- CHKiRet(writeFile(ppString, iMsgOpts, pData));
- if(!bCoreSupportsBatching && pData->bFlushOnTXEnd) {
- CHKiRet(strm.Flush(pData->pStrm));
- }
-finalize_it:
- if(iRet == RS_RET_OK)
- iRet = RS_RET_DEFER_COMMIT;
-ENDdoAction
-
static inline void
setInstParamDefaults(instanceData *pData)
@@ -959,8 +1042,8 @@ setInstParamDefaults(instanceData *pData)
pData->dirGID = -1;
pData->bFailOnChown = 1;
pData->iDynaFileCacheSize = 10;
- pData->fCreateMode = 0644;
- pData->fDirCreateMode = 0700;
+ pData->fCreateMode = loadModConf->fCreateMode;
+ pData->fDirCreateMode = loadModConf->fDirCreateMode;
pData->bCreateDirs = 1;
pData->bSyncFile = 0;
pData->iZipLevel = 0;
@@ -993,19 +1076,19 @@ setupInstStatsCtrs(instanceData *pData)
CHKiRet(statsobj.SetName(pData->stats, ctrName));
STATSCOUNTER_INIT(pData->ctrRequests, pData->mutCtrRequests);
CHKiRet(statsobj.AddCounter(pData->stats, UCHAR_CONSTANT("requests"),
- ctrType_IntCtr, &(pData->ctrRequests)));
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pData->ctrRequests)));
STATSCOUNTER_INIT(pData->ctrLevel0, pData->mutCtrLevel0);
CHKiRet(statsobj.AddCounter(pData->stats, UCHAR_CONSTANT("level0"),
- ctrType_IntCtr, &(pData->ctrLevel0)));
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pData->ctrLevel0)));
STATSCOUNTER_INIT(pData->ctrMiss, pData->mutCtrMiss);
CHKiRet(statsobj.AddCounter(pData->stats, UCHAR_CONSTANT("missed"),
- ctrType_IntCtr, &(pData->ctrMiss)));
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pData->ctrMiss)));
STATSCOUNTER_INIT(pData->ctrEvict, pData->mutCtrEvict);
CHKiRet(statsobj.AddCounter(pData->stats, UCHAR_CONSTANT("evicted"),
- ctrType_IntCtr, &(pData->ctrEvict)));
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pData->ctrEvict)));
STATSCOUNTER_INIT(pData->ctrMax, pData->mutCtrMax);
CHKiRet(statsobj.AddCounter(pData->stats, UCHAR_CONSTANT("maxused"),
- ctrType_IntCtr, &(pData->ctrMax)));
+ ctrType_IntCtr, CTR_FLAG_RESETTABLE, &(pData->ctrMax)));
CHKiRet(statsobj.ConstructFinalize(pData->stats));
finalize_it:
@@ -1089,7 +1172,7 @@ initCryprov(instanceData *pData, struct nvlst *lst)
szDrvrName);
ABORT_FINALIZE(RS_RET_CRYPROV_ERR);
}
- CHKiRet(pData->cryprov.SetCnfParam(pData->cryprovData, lst));
+ CHKiRet(pData->cryprov.SetCnfParam(pData->cryprovData, lst, CRYPROV_PARAMTYPE_REGULAR));
dbgprintf("loaded crypto provider %s, data instance at %p\n",
szDrvrName, pData->cryprovData);
@@ -1327,6 +1410,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
BEGINdoHUP
CODESTARTdoHUP
+ pthread_mutex_lock(&pData->mutWrite);
if(pData->bDynamicName) {
dynaFileFreeCacheEntries(pData);
} else {
@@ -1334,6 +1418,7 @@ CODESTARTdoHUP
closeFile(pData);
}
}
+ pthread_mutex_unlock(&pData->mutWrite);
ENDdoHUP
@@ -1349,6 +1434,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 129392d2..8cf306ed 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -4,7 +4,7 @@
* NOTE: read comments in module-template.h to understand how this file
* works!
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -21,9 +21,6 @@
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
- *
- * TODO v6 config:
- * - permitted peer *list*
*/
#include "config.h"
#include "rsyslog.h"
@@ -39,6 +36,7 @@
#include <errno.h>
#include <ctype.h>
#include <unistd.h>
+#include <stdint.h>
#ifdef USE_NETZIP
#include <zlib.h>
#endif
@@ -74,32 +72,51 @@ DEFobjCurrIf(netstrms)
DEFobjCurrIf(netstrm)
DEFobjCurrIf(tcpclt)
+
+/* some local constants (just) for better readybility */
+#define IS_FLUSH 1
+#define NO_FLUSH 0
+
typedef struct _instanceData {
uchar *tplName; /* name of assigned template */
- netstrms_t *pNS; /* netstream subsystem */
- netstrm_t *pNetstrm; /* our output netstream */
uchar *pszStrmDrvr;
uchar *pszStrmDrvrAuthMode;
permittedPeers_t *pPermPeers;
int iStrmDrvrMode;
char *target;
- int *pSockArray; /* sockets to use for UDP */
- int bIsConnected; /* are we connected to remote host? 0 - no, 1 - yes, UDP means addr resolved */
- struct addrinfo *f_addr;
int compressionLevel; /* 0 - no compression, else level for zlib */
char *port;
int protocol;
int iRebindInterval; /* rebind interval */
- int nXmit; /* number of transmissions since last (re-)bind */
# define FORW_UDP 0
# define FORW_TCP 1
/* following fields for TCP-based delivery */
TCPFRAMINGMODE tcp_framing;
int bResendLastOnRecon; /* should the last message be re-sent on a successful reconnect? */
+# define COMPRESS_NEVER 0
+# define COMPRESS_SINGLE_MSG 1 /* old, single-message compression */
+ /* all other settings are for stream-compression */
+# define COMPRESS_STREAM_ALWAYS 2
+ uint8_t compressionMode;
+ int errsToReport; /* max number of errors to report (per instance) */
+ sbool strmCompFlushOnTxEnd; /* flush stream compression on transaction end? */
+} instanceData;
+
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+ netstrms_t *pNS; /* netstream subsystem */
+ netstrm_t *pNetstrm; /* our output netstream */
+ struct addrinfo *f_addr;
+ int *pSockArray; /* sockets to use for UDP */
+ int bIsConnected; /* are we connected to remote host? 0 - no, 1 - yes, UDP means addr resolved */
+ int nXmit; /* number of transmissions since last (re-)bind */
tcpclt_t *pTCPClt; /* our tcpclt object */
+ sbool bzInitDone; /* did we do an init of zstrm already? */
+ z_stream zstrm; /* zip stream to use for tcp compression */
uchar sndBuf[16*1024]; /* this is intensionally fixed -- see no good reason to make configurable */
unsigned offsSndBuf; /* next free spot in send buffer */
-} instanceData;
+ int errsToReport; /* (remaining) number of errors to report */
+} wrkrInstanceData_t;
/* config data */
typedef struct configSettings_s {
@@ -132,6 +149,9 @@ static struct cnfparamdescr actpdescr[] = {
{ "protocol", eCmdHdlrGetWord, 0 },
{ "tcp_framing", eCmdHdlrGetWord, 0 },
{ "ziplevel", eCmdHdlrInt, 0 },
+ { "compression.mode", eCmdHdlrGetWord, 0 },
+ { "compression.stream.flushontxend", eCmdHdlrBinary, 0 },
+ { "maxerrormessages", eCmdHdlrInt, 0 },
{ "rebindinterval", eCmdHdlrInt, 0 },
{ "streamdriver", eCmdHdlrGetWord, 0 },
{ "streamdrivermode", eCmdHdlrInt, 0 },
@@ -155,6 +175,9 @@ static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current l
static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */
+static rsRetVal initTCP(wrkrInstanceData_t *pWrkrData);
+
+
BEGINinitConfVars /* (re)set config variables to default values */
CODESTARTinitConfVars
cs.pszTplName = NULL; /* name of the default template to use */
@@ -168,7 +191,8 @@ CODESTARTinitConfVars
ENDinitConfVars
-static rsRetVal doTryResume(instanceData *pData);
+static rsRetVal doTryResume(wrkrInstanceData_t *);
+static rsRetVal doZipFinish(wrkrInstanceData_t *);
/* this function gets the default template. It coordinates action between
* old-style and new-style configuration parts.
@@ -212,17 +236,16 @@ finalize_it:
* rgerhards, 2009-05-29
*/
static rsRetVal
-closeUDPSockets(instanceData *pData)
+closeUDPSockets(wrkrInstanceData_t *pWrkrData)
{
DEFiRet;
- assert(pData != NULL);
- if(pData->pSockArray != NULL) {
- net.closeUDPListenSockets(pData->pSockArray);
- pData->pSockArray = NULL;
- freeaddrinfo(pData->f_addr);
- pData->f_addr = NULL;
+ if(pWrkrData->pSockArray != NULL) {
+ net.closeUDPListenSockets(pWrkrData->pSockArray);
+ pWrkrData->pSockArray = NULL;
+ freeaddrinfo(pWrkrData->f_addr);
+ pWrkrData->f_addr = NULL;
}
-pData->bIsConnected = 0; // TODO: remove this variable altogether
+pWrkrData->bIsConnected = 0; // TODO: remove this variable altogether
RETiRet;
}
@@ -233,17 +256,17 @@ pData->bIsConnected = 0; // TODO: remove this variable altogether
* rgerhards, 2008-06-04
* Note that we DO NOT discard the current buffer contents
* (if any). This permits us to save data between sessions. In
- * the wort case, some duplication occurs, but we do not
+ * the worst case, some duplication occurs, but we do not
* loose data.
*/
static inline void
-DestructTCPInstanceData(instanceData *pData)
+DestructTCPInstanceData(wrkrInstanceData_t *pWrkrData)
{
- assert(pData != NULL);
- if(pData->pNetstrm != NULL)
- netstrm.Destruct(&pData->pNetstrm);
- if(pData->pNS != NULL)
- netstrms.Destruct(&pData->pNS);
+ doZipFinish(pWrkrData);
+ if(pWrkrData->pNetstrm != NULL)
+ netstrm.Destruct(&pWrkrData->pNetstrm);
+ if(pWrkrData->pNS != NULL)
+ netstrms.Destruct(&pWrkrData->pNS);
}
@@ -314,10 +337,25 @@ ENDfreeCnf
BEGINcreateInstance
CODESTARTcreateInstance
- pData->offsSndBuf = 0;
+ pData->errsToReport = 5;
+ if(cs.pszStrmDrvr != NULL)
+ CHKmalloc(pData->pszStrmDrvr = (uchar*)strdup((char*)cs.pszStrmDrvr));
+ if(cs.pszStrmDrvrAuthMode != NULL)
+ CHKmalloc(pData->pszStrmDrvrAuthMode =
+ (uchar*)strdup((char*)cs.pszStrmDrvrAuthMode));
+finalize_it:
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ dbgprintf("DDDD: createWrkrInstance: pWrkrData %p\n", pWrkrData);
+ pWrkrData->offsSndBuf = 0;
+ pWrkrData->errsToReport = pData->errsToReport;
+ iRet = initTCP(pWrkrData);
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -327,22 +365,25 @@ ENDisCompatibleWithFeature
BEGINfreeInstance
CODESTARTfreeInstance
- /* final cleanup */
- DestructTCPInstanceData(pData);
- closeUDPSockets(pData);
-
- if(pData->protocol == FORW_TCP) {
- tcpclt.Destruct(&pData->pTCPClt);
- }
-
- free(pData->port);
- free(pData->target);
free(pData->pszStrmDrvr);
free(pData->pszStrmDrvrAuthMode);
+ free(pData->port);
+ free(pData->target);
net.DestructPermittedPeers(&pData->pPermPeers);
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ DestructTCPInstanceData(pWrkrData);
+ closeUDPSockets(pWrkrData);
+
+ if(pWrkrData->pData->protocol == FORW_TCP) {
+ tcpclt.Destruct(&pWrkrData->pTCPClt);
+ }
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
dbgprintf("%s", pData->target);
@@ -352,25 +393,27 @@ ENDdbgPrintInstInfo
/* Send a message via UDP
* rgehards, 2007-12-20
*/
-static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
+static rsRetVal UDPSend(wrkrInstanceData_t *pWrkrData, char *msg, size_t len)
{
DEFiRet;
struct addrinfo *r;
int i;
unsigned lsent = 0;
- int bSendSuccess;
+ sbool bSendSuccess;
+ int lasterrno;
+ char errStr[1024];
- if(pData->iRebindInterval && (pData->nXmit++ % pData->iRebindInterval == 0)) {
+ if(pWrkrData->pData->iRebindInterval && (pWrkrData->nXmit++ % pWrkrData->pData->iRebindInterval == 0)) {
dbgprintf("omfwd dropping UDP 'connection' (as configured)\n");
- pData->nXmit = 1; /* else we have an addtl wrap at 2^31-1 */
- CHKiRet(closeUDPSockets(pData));
+ pWrkrData->nXmit = 1; /* else we have an addtl wrap at 2^31-1 */
+ CHKiRet(closeUDPSockets(pWrkrData));
}
- if(pData->pSockArray == NULL) {
- CHKiRet(doTryResume(pData));
+ if(pWrkrData->pSockArray == NULL) {
+ CHKiRet(doTryResume(pWrkrData));
}
- if(pData->pSockArray != NULL) {
+ if(pWrkrData->pSockArray != NULL) {
/* we need to track if we have success sending to the remote
* peer. Success is indicated by at least one sendto() call
* succeeding. We track this be bSendSuccess. We can not simply
@@ -379,25 +422,37 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
* the sendto() succeeded. -- rgerhards, 2007-06-22
*/
bSendSuccess = RSFALSE;
- for (r = pData->f_addr; r; r = r->ai_next) {
- for (i = 0; i < *pData->pSockArray; i++) {
- lsent = sendto(pData->pSockArray[i+1], msg, len, 0, r->ai_addr, r->ai_addrlen);
+ for (r = pWrkrData->f_addr; r; r = r->ai_next) {
+ for (i = 0; i < *pWrkrData->pSockArray; i++) {
+ lsent = sendto(pWrkrData->pSockArray[i+1], msg, len, 0, r->ai_addr, r->ai_addrlen);
if (lsent == len) {
bSendSuccess = RSTRUE;
break;
} else {
- int eno = errno;
- char errStr[1024];
- dbgprintf("sendto() error: %d = %s.\n",
- eno, rs_strerror_r(eno, errStr, sizeof(errStr)));
+ lasterrno = errno;
+ DBGPRINTF("sendto() error: %d = %s.\n",
+ lasterrno,
+ rs_strerror_r(lasterrno, errStr, sizeof(errStr)));
}
}
if (lsent == len && !send_to_all)
break;
}
/* finished looping */
- if (bSendSuccess == RSFALSE) {
+ if(bSendSuccess == RSFALSE) {
dbgprintf("error forwarding via udp, suspending\n");
+ if(pWrkrData->errsToReport > 0) {
+ rs_strerror_r(lasterrno, errStr, sizeof(errStr));
+ errmsg.LogError(0, RS_RET_ERR_UDPSEND, "omfwd: error sending "
+ "via udp: %s", errStr);
+ if(pWrkrData->errsToReport == 1) {
+ errmsg.LogError(0, RS_RET_LAST_ERRREPORT, "omfwd: "
+ "max number of error message emitted "
+ "- further messages will be "
+ "suppressed");
+ }
+ --pWrkrData->errsToReport;
+ }
iRet = RS_RET_SUSPENDED;
}
}
@@ -423,24 +478,19 @@ finalize_it:
/* CODE FOR SENDING TCP MESSAGES */
-
-/* Send a buffer via TCP. Usually, this is used to send the current
- * send buffer, but if a message is larger than the buffer, we need to
- * have the capability to send the message buffer directly.
- * rgerhards, 2011-04-04
- */
static rsRetVal
-TCPSendBuf(instanceData *pData, uchar *buf, unsigned len)
+TCPSendBufUncompressed(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len)
{
DEFiRet;
unsigned alreadySent;
ssize_t lenSend;
alreadySent = 0;
- CHKiRet(netstrm.CheckConnection(pData->pNetstrm)); /* hack for plain tcp syslog - see ptcp driver for details */
+ CHKiRet(netstrm.CheckConnection(pWrkrData->pNetstrm)); /* hack for plain tcp syslog - see ptcp driver for details */
+
while(alreadySent != len) {
lenSend = len - alreadySent;
- CHKiRet(netstrm.Send(pData->pNetstrm, buf+alreadySent, &lenSend));
+ CHKiRet(netstrm.Send(pWrkrData->pNetstrm, buf+alreadySent, &lenSend));
DBGPRINTF("omfwd: TCP sent %ld bytes, requested %u\n", (long) lenSend, len - alreadySent);
alreadySent += lenSend;
}
@@ -449,38 +499,136 @@ finalize_it:
if(iRet != RS_RET_OK) {
/* error! */
dbgprintf("TCPSendBuf error %d, destruct TCP Connection!\n", iRet);
- DestructTCPInstanceData(pData);
+ DestructTCPInstanceData(pWrkrData);
iRet = RS_RET_SUSPENDED;
}
RETiRet;
}
+static rsRetVal
+TCPSendBufCompressed(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len, sbool bIsFlush)
+{
+ int zRet; /* zlib return state */
+ unsigned outavail;
+ uchar zipBuf[32*1024];
+ int op;
+ DEFiRet;
+
+ if(!pWrkrData->bzInitDone) {
+ /* allocate deflate state */
+ pWrkrData->zstrm.zalloc = Z_NULL;
+ pWrkrData->zstrm.zfree = Z_NULL;
+ pWrkrData->zstrm.opaque = Z_NULL;
+ /* see note in file header for the params we use with deflateInit2() */
+ zRet = deflateInit(&pWrkrData->zstrm, 9);
+ if(zRet != Z_OK) {
+ DBGPRINTF("error %d returned from zlib/deflateInit()\n", zRet);
+ ABORT_FINALIZE(RS_RET_ZLIB_ERR);
+ }
+ pWrkrData->bzInitDone = RSTRUE;
+ }
+
+ /* now doing the compression */
+ pWrkrData->zstrm.next_in = (Bytef*) buf;
+ pWrkrData->zstrm.avail_in = len;
+ if(pWrkrData->pData->strmCompFlushOnTxEnd && bIsFlush)
+ op = Z_SYNC_FLUSH;
+ else
+ op = Z_NO_FLUSH;
+ /* run deflate() on buffer until everything has been compressed */
+ do {
+ DBGPRINTF("omfwd: in deflate() loop, avail_in %d, total_in %ld, isFlush %d\n", pWrkrData->zstrm.avail_in, pWrkrData->zstrm.total_in, bIsFlush);
+ pWrkrData->zstrm.avail_out = sizeof(zipBuf);
+ pWrkrData->zstrm.next_out = zipBuf;
+ zRet = deflate(&pWrkrData->zstrm, op); /* no bad return value */
+ DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pWrkrData->zstrm.avail_out);
+ outavail = sizeof(zipBuf) - pWrkrData->zstrm.avail_out;
+ if(outavail != 0) {
+ CHKiRet(TCPSendBufUncompressed(pWrkrData, zipBuf, outavail));
+ }
+ } while (pWrkrData->zstrm.avail_out == 0);
+
+finalize_it:
+ RETiRet;
+}
+
+static rsRetVal
+TCPSendBuf(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len, sbool bIsFlush)
+{
+ DEFiRet;
+ if(pWrkrData->pData->compressionMode >= COMPRESS_STREAM_ALWAYS)
+ iRet = TCPSendBufCompressed(pWrkrData, buf, len, bIsFlush);
+ else
+ iRet = TCPSendBufUncompressed(pWrkrData, buf, len);
+ RETiRet;
+}
+
+/* finish zlib buffer, to be called before closing the ZIP file (if
+ * running in stream mode).
+ */
+static rsRetVal
+doZipFinish(wrkrInstanceData_t *pWrkrData)
+{
+ int zRet; /* zlib return state */
+ DEFiRet;
+ unsigned outavail;
+ uchar zipBuf[32*1024];
+
+ if(!pWrkrData->bzInitDone)
+ goto done;
+
+// TODO: can we get this into a single common function?
+dbgprintf("DDDD: in doZipFinish()\n");
+ pWrkrData->zstrm.avail_in = 0;
+ /* run deflate() on buffer until everything has been compressed */
+ do {
+ DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", pWrkrData->zstrm.avail_in, pWrkrData->zstrm.total_in);
+ pWrkrData->zstrm.avail_out = sizeof(zipBuf);
+ pWrkrData->zstrm.next_out = zipBuf;
+ zRet = deflate(&pWrkrData->zstrm, Z_FINISH); /* no bad return value */
+ DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pWrkrData->zstrm.avail_out);
+ outavail = sizeof(zipBuf) - pWrkrData->zstrm.avail_out;
+ if(outavail != 0) {
+ CHKiRet(TCPSendBufUncompressed(pWrkrData, zipBuf, outavail));
+ }
+ } while (pWrkrData->zstrm.avail_out == 0);
+
+finalize_it:
+ zRet = deflateEnd(&pWrkrData->zstrm);
+ if(zRet != Z_OK) {
+ DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet);
+ }
+
+ pWrkrData->bzInitDone = 0;
+done: RETiRet;
+}
+
/* Add frame to send buffer (or send, if requried)
*/
static rsRetVal TCPSendFrame(void *pvData, char *msg, size_t len)
{
DEFiRet;
- instanceData *pData = (instanceData *) pvData;
+ wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
DBGPRINTF("omfwd: add %u bytes to send buffer (curr offs %u)\n",
- (unsigned) len, pData->offsSndBuf);
- if(pData->offsSndBuf != 0 && pData->offsSndBuf + len >= sizeof(pData->sndBuf)) {
+ (unsigned) len, pWrkrData->offsSndBuf);
+ if(pWrkrData->offsSndBuf != 0 && pWrkrData->offsSndBuf + len >= sizeof(pWrkrData->sndBuf)) {
/* no buffer space left, need to commit previous records */
- CHKiRet(TCPSendBuf(pData, pData->sndBuf, pData->offsSndBuf));
- pData->offsSndBuf = 0;
+ CHKiRet(TCPSendBuf(pWrkrData, pWrkrData->sndBuf, pWrkrData->offsSndBuf, NO_FLUSH));
+ pWrkrData->offsSndBuf = 0;
iRet = RS_RET_PREVIOUS_COMMITTED;
}
/* check if the message is too large to fit into buffer */
- if(len > sizeof(pData->sndBuf)) {
- CHKiRet(TCPSendBuf(pData, (uchar*)msg, len));
+ if(len > sizeof(pWrkrData->sndBuf)) {
+ CHKiRet(TCPSendBuf(pWrkrData, (uchar*)msg, len, NO_FLUSH));
ABORT_FINALIZE(RS_RET_OK); /* committed everything so far */
}
/* we now know the buffer has enough free space */
- memcpy(pData->sndBuf + pData->offsSndBuf, msg, len);
- pData->offsSndBuf += len;
+ memcpy(pWrkrData->sndBuf + pWrkrData->offsSndBuf, msg, len);
+ pWrkrData->offsSndBuf += len;
iRet = RS_RET_DEFER_COMMIT;
finalize_it:
@@ -495,11 +643,10 @@ finalize_it:
static rsRetVal TCPSendPrepRetry(void *pvData)
{
DEFiRet;
- instanceData *pData = (instanceData *) pvData;
-dbgprintf("TCPSendPrepRetry performs a DestructTCPInstanceData\n");
+ wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
- assert(pData != NULL);
- DestructTCPInstanceData(pData);
+ assert(pWrkrData != NULL);
+ DestructTCPInstanceData(pWrkrData);
RETiRet;
}
@@ -510,36 +657,39 @@ dbgprintf("TCPSendPrepRetry performs a DestructTCPInstanceData\n");
static rsRetVal TCPSendInit(void *pvData)
{
DEFiRet;
- instanceData *pData = (instanceData *) pvData;
+ wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
+ instanceData *pData;
+
+ assert(pWrkrData != NULL);
+ pData = pWrkrData->pData;
- assert(pData != NULL);
- if(pData->pNetstrm == NULL) {
+ if(pWrkrData->pNetstrm == NULL) {
dbgprintf("TCPSendInit CREATE\n");
- CHKiRet(netstrms.Construct(&pData->pNS));
+ CHKiRet(netstrms.Construct(&pWrkrData->pNS));
/* the stream driver must be set before the object is finalized! */
- CHKiRet(netstrms.SetDrvrName(pData->pNS, pData->pszStrmDrvr));
- CHKiRet(netstrms.ConstructFinalize(pData->pNS));
+ CHKiRet(netstrms.SetDrvrName(pWrkrData->pNS, pData->pszStrmDrvr));
+ CHKiRet(netstrms.ConstructFinalize(pWrkrData->pNS));
/* now create the actual stream and connect to the server */
- CHKiRet(netstrms.CreateStrm(pData->pNS, &pData->pNetstrm));
- CHKiRet(netstrm.ConstructFinalize(pData->pNetstrm));
- CHKiRet(netstrm.SetDrvrMode(pData->pNetstrm, pData->iStrmDrvrMode));
+ CHKiRet(netstrms.CreateStrm(pWrkrData->pNS, &pWrkrData->pNetstrm));
+ CHKiRet(netstrm.ConstructFinalize(pWrkrData->pNetstrm));
+ CHKiRet(netstrm.SetDrvrMode(pWrkrData->pNetstrm, pData->iStrmDrvrMode));
/* now set optional params, but only if they were actually configured */
if(pData->pszStrmDrvrAuthMode != NULL) {
- CHKiRet(netstrm.SetDrvrAuthMode(pData->pNetstrm, pData->pszStrmDrvrAuthMode));
+ CHKiRet(netstrm.SetDrvrAuthMode(pWrkrData->pNetstrm, pData->pszStrmDrvrAuthMode));
}
if(pData->pPermPeers != NULL) {
- CHKiRet(netstrm.SetDrvrPermPeers(pData->pNetstrm, pData->pPermPeers));
+ CHKiRet(netstrm.SetDrvrPermPeers(pWrkrData->pNetstrm, pData->pPermPeers));
}
/* params set, now connect */
- CHKiRet(netstrm.Connect(pData->pNetstrm, glbl.GetDefPFFamily(),
+ CHKiRet(netstrm.Connect(pWrkrData->pNetstrm, glbl.GetDefPFFamily(),
(uchar*)pData->port, (uchar*)pData->target));
}
finalize_it:
if(iRet != RS_RET_OK) {
dbgprintf("TCPSendInit FAILED with %d.\n", iRet);
- DestructTCPInstanceData(pData);
+ DestructTCPInstanceData(pWrkrData);
}
RETiRet;
@@ -549,15 +699,17 @@ finalize_it:
/* try to resume connection if it is not ready
* rgerhards, 2007-08-02
*/
-static rsRetVal doTryResume(instanceData *pData)
+static rsRetVal doTryResume(wrkrInstanceData_t *pWrkrData)
{
int iErr;
struct addrinfo *res;
struct addrinfo hints;
+ instanceData *pData;
DEFiRet;
- if(pData->bIsConnected)
+ if(pWrkrData->bIsConnected)
FINALIZE;
+ pData = pWrkrData->pData;
/* The remote address is not yet known and needs to be obtained */
dbgprintf(" %s\n", pData->target);
@@ -573,20 +725,20 @@ static rsRetVal doTryResume(instanceData *pData)
ABORT_FINALIZE(RS_RET_SUSPENDED);
}
dbgprintf("%s found, resuming.\n", pData->target);
- pData->f_addr = res;
- pData->bIsConnected = 1;
- if(pData->pSockArray == NULL) {
- pData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0);
+ pWrkrData->f_addr = res;
+ pWrkrData->bIsConnected = 1;
+ if(pWrkrData->pSockArray == NULL) {
+ pWrkrData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0, 0);
}
} else {
- CHKiRet(TCPSendInit((void*)pData));
+ CHKiRet(TCPSendInit((void*)pWrkrData));
}
finalize_it:
if(iRet != RS_RET_OK) {
- if(pData->f_addr != NULL) {
- freeaddrinfo(pData->f_addr);
- pData->f_addr = NULL;
+ if(pWrkrData->f_addr != NULL) {
+ freeaddrinfo(pWrkrData->f_addr);
+ pWrkrData->f_addr = NULL;
}
iRet = RS_RET_SUSPENDED;
}
@@ -597,7 +749,8 @@ finalize_it:
BEGINtryResume
CODESTARTtryResume
- iRet = doTryResume(pData);
+ dbgprintf("DDDD: tryResume: pWrkrData %p\n", pWrkrData);
+ iRet = doTryResume(pWrkrData);
ENDtryResume
@@ -614,8 +767,11 @@ BEGINdoAction
# ifdef USE_NETZIP
Bytef *out = NULL; /* for compression */
# endif
+ instanceData *pData;
CODESTARTdoAction
- CHKiRet(doTryResume(pData));
+ dbgprintf("DDDD: doAction: pWrkrData %p\n", pWrkrData);
+ pData = pWrkrData->pData;
+ CHKiRet(doTryResume(pWrkrData));
iMaxLine = glbl.GetMaxLine();
@@ -636,11 +792,10 @@ CODESTARTdoAction
* hard-coded but this may be changed to a config parameter.
* rgerhards, 2006-11-30
*/
- if(pData->compressionLevel && (l > CONF_MIN_SIZE_FOR_COMPRESS)) {
+ if(pData->compressionMode == COMPRESS_SINGLE_MSG && (l > CONF_MIN_SIZE_FOR_COMPRESS)) {
uLongf destLen = iMaxLine + iMaxLine/100 +12; /* recommended value from zlib doc */
uLong srcLen = l;
int ret;
- /* TODO: optimize malloc sequence? -- rgerhards, 2008-09-02 */
CHKmalloc(out = (Bytef*) MALLOC(destLen));
out[0] = 'z';
out[1] = '\0';
@@ -669,16 +824,19 @@ CODESTARTdoAction
if(pData->protocol == FORW_UDP) {
/* forward via UDP */
- CHKiRet(UDPSend(pData, psz, l));
+ CHKiRet(UDPSend(pWrkrData, psz, l));
} else {
/* forward via TCP */
- iRet = tcpclt.Send(pData->pTCPClt, pData, psz, l);
+ iRet = tcpclt.Send(pWrkrData->pTCPClt, pWrkrData, psz, l);
if(iRet != RS_RET_OK && iRet != RS_RET_DEFER_COMMIT && iRet != RS_RET_PREVIOUS_COMMITTED) {
/* error! */
dbgprintf("error forwarding via tcp, suspending\n");
- DestructTCPInstanceData(pData);
+ DestructTCPInstanceData(pWrkrData);
iRet = RS_RET_SUSPENDED;
}
+ if(pData->compressionMode >= COMPRESS_STREAM_ALWAYS && pData->strmCompFlushOnTxEnd)
+ /* mimic not committed, as we need the EndTx entry point to be called */
+ iRet = RS_RET_DEFER_COMMIT;
}
finalize_it:
# ifdef USE_NETZIP
@@ -689,10 +847,10 @@ ENDdoAction
BEGINendTransaction
CODESTARTendTransaction
-dbgprintf("omfwd: endTransaction, offsSndBuf %u\n", pData->offsSndBuf);
- if(pData->offsSndBuf != 0) {
- iRet = TCPSendBuf(pData, pData->sndBuf, pData->offsSndBuf);
- pData->offsSndBuf = 0;
+dbgprintf("omfwd: endTransaction, offsSndBuf %u\n", pWrkrData->offsSndBuf);
+ if(pWrkrData->offsSndBuf != 0) {
+ iRet = TCPSendBuf(pWrkrData, pWrkrData->sndBuf, pWrkrData->offsSndBuf, IS_FLUSH);
+ pWrkrData->offsSndBuf = 0;
}
ENDendTransaction
@@ -719,25 +877,22 @@ finalize_it:
* created.
*/
static rsRetVal
-initTCP(instanceData *pData)
+initTCP(wrkrInstanceData_t *pWrkrData)
{
+ instanceData *pData;
DEFiRet;
+
+ pData = pWrkrData->pData;
if(pData->protocol == FORW_TCP) {
/* create our tcpclt */
- CHKiRet(tcpclt.Construct(&pData->pTCPClt));
- CHKiRet(tcpclt.SetResendLastOnRecon(pData->pTCPClt, pData->bResendLastOnRecon));
+ CHKiRet(tcpclt.Construct(&pWrkrData->pTCPClt));
+ CHKiRet(tcpclt.SetResendLastOnRecon(pWrkrData->pTCPClt, pData->bResendLastOnRecon));
/* and set callbacks */
- CHKiRet(tcpclt.SetSendInit(pData->pTCPClt, TCPSendInit));
- CHKiRet(tcpclt.SetSendFrame(pData->pTCPClt, TCPSendFrame));
- CHKiRet(tcpclt.SetSendPrepRetry(pData->pTCPClt, TCPSendPrepRetry));
- CHKiRet(tcpclt.SetFraming(pData->pTCPClt, pData->tcp_framing));
- CHKiRet(tcpclt.SetRebindInterval(pData->pTCPClt, pData->iRebindInterval));
- pData->iStrmDrvrMode = cs.iStrmDrvrMode;
- if(cs.pszStrmDrvr != NULL)
- CHKmalloc(pData->pszStrmDrvr = (uchar*)strdup((char*)cs.pszStrmDrvr));
- if(cs.pszStrmDrvrAuthMode != NULL)
- CHKmalloc(pData->pszStrmDrvrAuthMode =
- (uchar*)strdup((char*)cs.pszStrmDrvrAuthMode));
+ CHKiRet(tcpclt.SetSendInit(pWrkrData->pTCPClt, TCPSendInit));
+ CHKiRet(tcpclt.SetSendFrame(pWrkrData->pTCPClt, TCPSendFrame));
+ CHKiRet(tcpclt.SetSendPrepRetry(pWrkrData->pTCPClt, TCPSendPrepRetry));
+ CHKiRet(tcpclt.SetFraming(pWrkrData->pTCPClt, pData->tcp_framing));
+ CHKiRet(tcpclt.SetRebindInterval(pWrkrData->pTCPClt, pData->iRebindInterval));
}
finalize_it:
RETiRet;
@@ -756,14 +911,19 @@ setInstParamDefaults(instanceData *pData)
pData->iRebindInterval = 0;
pData->bResendLastOnRecon = 0;
pData->pPermPeers = NULL;
- pData->compressionLevel = 0;
+ pData->compressionLevel = 9;
+ pData->strmCompFlushOnTxEnd = 1;
+ pData->compressionMode = COMPRESS_NEVER;
+ pData->errsToReport = 5;
}
BEGINnewActInst
struct cnfparamvals *pvals;
uchar *tplToUse;
+ char *cstr;
int i;
rsRetVal localRet;
+ int complevel = -1;
CODESTARTnewActInst
DBGPRINTF("newActInst (omfwd)\n");
@@ -860,9 +1020,10 @@ CODESTARTnewActInst
free(str);
} else if(!strcmp(actpblk.descr[i].name, "ziplevel")) {
# ifdef USE_NETZIP
- int complevel = pvals[i].val.d.n;
+ complevel = pvals[i].val.d.n;
if(complevel >= 0 && complevel <= 10) {
pData->compressionLevel = complevel;
+ pData->compressionMode = COMPRESS_SINGLE_MSG;
} else {
errmsg.LogError(0, NO_ERRCODE, "Invalid ziplevel %d specified in "
"forwardig action - NOT turning on compression.",
@@ -872,21 +1033,50 @@ CODESTARTnewActInst
errmsg.LogError(0, NO_ERRCODE, "Compression requested, but rsyslogd is not compiled "
"with compression support - request ignored.");
# endif /* #ifdef USE_NETZIP */
+ } else if(!strcmp(actpblk.descr[i].name, "maxerrormessages")) {
+ pData->errsToReport = (int) pvals[i].val.d.n;
} else if(!strcmp(actpblk.descr[i].name, "resendlastmsgonreconnect")) {
pData->bResendLastOnRecon = (int) pvals[i].val.d.n;
} else if(!strcmp(actpblk.descr[i].name, "template")) {
pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(actpblk.descr[i].name, "compression.stream.flushontxend")) {
+ pData->strmCompFlushOnTxEnd = (sbool) pvals[i].val.d.n;
+ } else if(!strcmp(actpblk.descr[i].name, "compression.mode")) {
+ cstr = es_str2cstr(pvals[i].val.d.estr, NULL);
+ if(!strcasecmp(cstr, "stream:always")) {
+ pData->compressionMode = COMPRESS_STREAM_ALWAYS;
+ } else if(!strcasecmp(cstr, "none")) {
+ pData->compressionMode = COMPRESS_NEVER;
+ } else if(!strcasecmp(cstr, "single")) {
+ pData->compressionMode = COMPRESS_SINGLE_MSG;
+ } else {
+ errmsg.LogError(0, RS_RET_PARAM_ERROR, "omfwd: invalid value for 'compression.mode' "
+ "parameter (given is '%s')", cstr);
+ free(cstr);
+ ABORT_FINALIZE(RS_RET_PARAM_ERROR);
+ }
+ free(cstr);
} else {
DBGPRINTF("omfwd: program error, non-handled "
"param '%s'\n", actpblk.descr[i].name);
}
}
+
+ if(complevel != -1) {
+ pData->compressionLevel = complevel;
+ if(pData->compressionMode == COMPRESS_NEVER) {
+ /* to keep compatible with pre-7.3.11, only setting the
+ * compresion level means old-style single-message mode.
+ */
+ pData->compressionMode = COMPRESS_SINGLE_MSG;
+ }
+ }
+
CODE_STD_STRING_REQUESTnewActInst(1)
tplToUse = ustrdup((pData->tplName == NULL) ? getDfltTpl() : pData->tplName);
CHKiRet(OMSRsetEntry(*ppOMSR, 0, tplToUse, OMSR_NO_RQD_TPL_OPTS));
- CHKiRet(initTCP(pData));
CODE_STD_FINALIZERnewActInst
cnfparamvalsDestruct(pvals, &actpblk);
ENDnewActInst
@@ -948,6 +1138,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
iLevel = *p - '0';
++p; /* eat */
pData->compressionLevel = iLevel;
+ pData->compressionMode = COMPRESS_SINGLE_MSG;
} else {
errmsg.LogError(0, NO_ERRCODE, "Invalid compression level '%c' specified in "
"forwardig action - NOT turning on compression.",
@@ -1025,7 +1216,6 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
while(*p && *p != ';' && *p != '#' && !isspace((int) *p))
++p; /*JUST SKIP*/
- /* TODO: make this if go away! */
if(*p == ';' || *p == '#' || isspace(*p)) {
uchar cTmp = *p;
*p = '\0'; /* trick to obtain hostname (later)! */
@@ -1055,7 +1245,6 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
cs.pPermPeers = NULL;
}
}
- CHKiRet(initTCP(pData));
CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct
@@ -1091,6 +1280,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
diff --git a/tools/ompipe.c b/tools/ompipe.c
index df8066b1..79f3ae84 100644
--- a/tools/ompipe.c
+++ b/tools/ompipe.c
@@ -12,7 +12,7 @@
* NOTE: read comments in module-template.h to understand how this pipe
* works!
*
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -69,9 +69,14 @@ typedef struct _instanceData {
uchar *pipe; /* pipe or template name (display only) */
uchar *tplName; /* format template to use */
short fd; /* pipe descriptor for (current) pipe */
+ pthread_mutex_t mutWrite; /* guard against multiple instances writing to same pipe */
sbool bHadError; /* did we already have/report an error on this pipe? */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
EMPTY_STRUCT
} configSettings_t;
@@ -276,25 +281,42 @@ CODESTARTcreateInstance
pData->pipe = NULL;
pData->fd = -1;
pData->bHadError = 0;
+ pthread_mutex_init(&pData->mutWrite, NULL);
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINfreeInstance
CODESTARTfreeInstance
+ pthread_mutex_destroy(&pData->mutWrite);
free(pData->pipe);
if(pData->fd != -1)
close(pData->fd);
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINtryResume
CODESTARTtryResume
ENDtryResume
BEGINdoAction
+ instanceData *pData;
CODESTARTdoAction
+ pData = pWrkrData->pData;
DBGPRINTF(" (%s)\n", pData->pipe);
+ /* this module is single-threaded by nature */
+ pthread_mutex_lock(&pData->mutWrite);
iRet = writePipe(ppString, pData);
+ pthread_mutex_unlock(&pData->mutWrite);
ENDdoAction
@@ -390,6 +412,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_doHUP
CODEqueryEtryPt_STD_CONF2_QUERIES
CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES
diff --git a/tools/omshell.c b/tools/omshell.c
index ac62fa62..ad6e979f 100644
--- a/tools/omshell.c
+++ b/tools/omshell.c
@@ -19,7 +19,7 @@
* of the "old" message code without any modifications. However, it
* helps to have things at the right place one we go to the meat of it.
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -63,11 +63,20 @@ typedef struct _instanceData {
uchar progName[MAXFNAME]; /* program to execute */
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
BEGINcreateInstance
CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -80,6 +89,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
printf("%s", pData->progName);
@@ -92,13 +106,9 @@ ENDtryResume
BEGINdoAction
CODESTARTdoAction
- /* TODO: using pData->progName is not clean from the point of
- * modularization. We'll change that as we go ahead with modularization.
- * rgerhards, 2007-07-20
- */
dbgprintf("\n");
- if(execProg((uchar*) pData->progName, 1, ppString[0]) == 0)
- errmsg.LogError(0, NO_ERRCODE, "Executing program '%s' failed", (char*)pData->progName);
+ if(execProg((uchar*) pWrkrData->pData->progName, 1, ppString[0]) == 0)
+ errmsg.LogError(0, NO_ERRCODE, "Executing program '%s' failed", (char*)pWrkrData->pData->progName);
ENDdoAction
@@ -139,6 +149,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
ENDqueryEtryPt
diff --git a/tools/omusrmsg.c b/tools/omusrmsg.c
index f4cc4094..5d0b088f 100644
--- a/tools/omusrmsg.c
+++ b/tools/omusrmsg.c
@@ -8,7 +8,7 @@
* File begun on 2007-07-20 by RGerhards (extracted from syslogd.c, which at the
* time of the fork from sysklogd was under BSD license)
*
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -87,6 +87,10 @@ typedef struct _instanceData {
uchar *tplName;
} instanceData;
+typedef struct wrkrInstanceData {
+ instanceData *pData;
+} wrkrInstanceData_t;
+
typedef struct configSettings_s {
EMPTY_STRUCT
} configSettings_t;
@@ -115,6 +119,11 @@ CODESTARTcreateInstance
ENDcreateInstance
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
BEGINisCompatibleWithFeature
CODESTARTisCompatibleWithFeature
if(eFeat == sFEATURERepeatedMsgReduction)
@@ -128,6 +137,11 @@ CODESTARTfreeInstance
ENDfreeInstance
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
BEGINdbgPrintInstInfo
register int i;
CODESTARTdbgPrintInstInfo
@@ -276,7 +290,7 @@ ENDtryResume
BEGINdoAction
CODESTARTdoAction
dbgprintf("\n");
- iRet = wallmsg(ppString[0], pData);
+ iRet = wallmsg(ppString[0], pWrkrData->pData);
ENDdoAction
@@ -435,6 +449,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
ENDqueryEtryPt
diff --git a/tools/pmrfc3164.c b/tools/pmrfc3164.c
index 5dfa74f0..25964702 100644
--- a/tools/pmrfc3164.c
+++ b/tools/pmrfc3164.c
@@ -84,7 +84,7 @@ CODESTARTparse
assert(pMsg->pszRawMsg != NULL);
lenMsg = pMsg->iLenRawMsg - pMsg->offAfterPRI; /* note: offAfterPRI is already the number of PRI chars (do not add one!) */
p2parse = pMsg->pszRawMsg + pMsg->offAfterPRI; /* point to start of text, after PRI */
- setProtocolVersion(pMsg, 0);
+ setProtocolVersion(pMsg, MSG_LEGACY_PROTOCOL);
/* Check to see if msg contains a timestamp. We start by assuming
* that the message timestamp is the time of reception (which we
@@ -95,6 +95,8 @@ CODESTARTparse
if(datetime.ParseTIMESTAMP3339(&(pMsg->tTIMESTAMP), &p2parse, &lenMsg) == RS_RET_OK) {
/* we are done - parse pointer is moved by ParseTIMESTAMP3339 */;
} else if(datetime.ParseTIMESTAMP3164(&(pMsg->tTIMESTAMP), &p2parse, &lenMsg) == RS_RET_OK) {
+ if(pMsg->dfltTZ[0] != '\0')
+ applyDfltTZ(&pMsg->tTIMESTAMP, pMsg->dfltTZ);
/* we are done - parse pointer is moved by ParseTIMESTAMP3164 */;
} else if(*p2parse == ' ' && lenMsg > 1) { /* try to see if it is slighly malformed - HP procurve seems to do that sometimes */
++p2parse; /* move over space */
diff --git a/tools/pmrfc5424.c b/tools/pmrfc5424.c
index 9b5c6165..8e9510f3 100644
--- a/tools/pmrfc5424.c
+++ b/tools/pmrfc5424.c
@@ -227,7 +227,7 @@ CODESTARTparse
ABORT_FINALIZE(RS_RET_COULD_NOT_PARSE);
}
DBGPRINTF("Message has RFC5424/syslog-protocol format.\n");
- setProtocolVersion(pMsg, 1);
+ setProtocolVersion(pMsg, MSG_RFC5424_PROTOCOL);
p2parse += 2;
lenMsg -= 2;
diff --git a/tools/rsyslogd.8 b/tools/rsyslogd.8
index 620006f2..ac732b88 100644
--- a/tools/rsyslogd.8
+++ b/tools/rsyslogd.8
@@ -191,6 +191,10 @@ is specified and the host logging resolves to satu.infodrom.north.de
no domain would be cut, you will have to specify two domains like:
.BR "\-s north.de:infodrom.north.de" .
.TP
+.BI "\-S ip_address" "local client source IP"
+rsyslogd uses ip_address as local client address while connecting
+to remote logserver. Currently used by omrelp only and only with tcp.
+.TP
.BI "\-u " "userlevel"
This is a "catch all" option for some very seldomly-used user settings.
The "userlevel" variable selects multiple things. Add the specific values
diff --git a/tools/syslogd.c b/tools/syslogd.c
index a8a733d6..052a9329 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -497,24 +497,19 @@ finalize_it:
* rgerhards, 2010-06-09
*/
static inline rsRetVal
-preprocessBatch(batch_t *pBatch) {
+preprocessBatch(batch_t *pBatch, int *pbShutdownImmediate) {
prop_t *ip;
prop_t *fqdn;
prop_t *localName;
prop_t *propFromHost = NULL;
prop_t *propFromHostIP = NULL;
- int bSingleRuleset;
- ruleset_t *batchRuleset; /* the ruleset used for all message inside the batch, if there is a single one */
int bIsPermitted;
msg_t *pMsg;
int i;
rsRetVal localRet;
DEFiRet;
- bSingleRuleset = 1;
- batchRuleset = (pBatch->nElem > 0) ? pBatch->pElem[0].pMsg->pRuleset : NULL;
-
- for(i = 0 ; i < pBatch->nElem && !*(pBatch->pbShutdownImmediate) ; i++) {
+ for(i = 0 ; i < pBatch->nElem && !*pbShutdownImmediate ; i++) {
pMsg = pBatch->pElem[i].pMsg;
if((pMsg->msgFlags & NEEDS_ACLCHK_U) != 0) {
DBGPRINTF("msgConsumer: UDP ACL must be checked for message (hostname-based)\n");
@@ -539,12 +534,8 @@ preprocessBatch(batch_t *pBatch) {
pBatch->eltState[i] = BATCH_STATE_DISC;
}
}
- if(pMsg->pRuleset != batchRuleset)
- bSingleRuleset = 0;
}
- batchSetSingleRuleset(pBatch, bSingleRuleset);
-
finalize_it:
if(propFromHost != NULL)
prop.Destruct(&propFromHost);
@@ -560,17 +551,16 @@ finalize_it:
* for the main queue.
*/
static rsRetVal
-msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, int *pbShutdownImmediate)
+msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, wti_t *pWti)
{
DEFiRet;
assert(pBatch != NULL);
- pBatch->pbShutdownImmediate = pbShutdownImmediate; /* TODO: move this to batch creation! */
- preprocessBatch(pBatch);
- ruleset.ProcessBatch(pBatch);
+ preprocessBatch(pBatch, pWti->pbShutdownImmediate);
+ ruleset.ProcessBatch(pBatch, pWti);
//TODO: the BATCH_STATE_COMM must be set somewhere down the road, but we
//do not have this yet and so we emulate -- 2010-06-10
int i;
- for(i = 0 ; i < pBatch->nElem && !*pbShutdownImmediate ; i++) {
+ for(i = 0 ; i < pBatch->nElem && !*pWti->pbShutdownImmediate ; i++) {
pBatch->eltState[i] = BATCH_STATE_COMM;
}
RETiRet;
@@ -1056,7 +1046,7 @@ finalize_it:
* the time being (remember that we want to restructure config processing at large!).
* rgerhards, 2009-10-27
*/
-rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName, struct cnfparamvals *queueParams)
+rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName, struct nvlst *lst)
{
struct queuefilenames_s *qfn;
uchar *qfname = NULL;
@@ -1072,7 +1062,7 @@ rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName, struct cnfpara
/* name our main queue object (it's not fatal if it fails...) */
obj.SetName((obj_t*) (*ppQueue), pszQueueName);
- if(queueParams == NULL) { /* use legacy parameters? */
+ if(lst == NULL) { /* use legacy parameters? */
/* ... set some properties ... */
# define setQPROP(func, directive, data) \
CHKiRet_Hdlr(func(*ppQueue, data)) { \
@@ -1129,11 +1119,16 @@ rsRetVal createMainQueue(qqueue_t **ppQueue, uchar *pszQueueName, struct cnfpara
# undef setQPROPstr
} else { /* use new style config! */
qqueueSetDefaultsRulesetQueue(*ppQueue);
- qqueueApplyCnfParam(*ppQueue, queueParams);
+ qqueueApplyCnfParam(*ppQueue, lst);
}
+ RETiRet;
+}
- /* ... and finally start the queue! */
- CHKiRet_Hdlr(qqueueStart(*ppQueue)) {
+rsRetVal
+startMainQueue(qqueue_t *pQueue)
+{
+ DEFiRet;
+ CHKiRet_Hdlr(qqueueStart(pQueue)) {
/* no queue is fatal, we need to give up in that case... */
errmsg.LogError(0, iRet, "could not start (ruleset) main message queue"); \
}
@@ -1253,6 +1248,7 @@ doHUP(void)
queryLocalHostname(); /* re-read our name */
ruleset.IterateAllActions(ourConf, doHUPActions, NULL);
+ lookupDoHUP();
}
@@ -1342,6 +1338,11 @@ static void printVersion(void)
#else
printf("\tuuid support:\t\t\t\tNo\n");
#endif
+#ifdef HAVE_JSON_OBJECT_NEW_INT64
+ printf("\tNumber of Bits in RainerScript integers: 64\n");
+#else
+ printf("\tNumber of Bits in RainerScript integers: 32 (due to too-old json-c lib)\n");
+#endif
printf("\nSee http://www.rsyslog.com for more information.\n");
}
@@ -1772,7 +1773,7 @@ int realMain(int argc, char **argv)
* of other options, we do this during the inital option processing.
* rgerhards, 2008-04-04
*/
- while((ch = getopt(argc, argv, "46a:Ac:dDef:g:hi:l:m:M:nN:op:qQr::s:t:T:u:vwx")) != EOF) {
+ while((ch = getopt(argc, argv, "46a:Ac:dDef:g:hi:l:m:M:nN:op:qQr::s:S:t:T:u:vwx")) != EOF) {
switch((char)ch) {
case '4':
case '6':
@@ -1790,6 +1791,7 @@ int realMain(int argc, char **argv)
case 'q': /* add hostname if DNS resolving has failed */
case 'Q': /* dont resolve hostnames in ACL to IPs */
case 's':
+ case 'S': /* Source IP for local client to be used on multihomed host */
case 'T': /* chroot on startup (primarily for testing) */
case 'u': /* misc user settings */
case 'w': /* disable disallowed host warnings */
@@ -1881,6 +1883,13 @@ int realMain(int argc, char **argv)
case 'a':
fprintf(stderr, "rsyslogd: error -a is no longer supported, use module imuxsock instead");
break;
+ case 'S': /* Source IP for local client to be used on multihomed host */
+ if(glbl.GetSourceIPofLocalClient() != NULL) {
+ fprintf (stderr, "rsyslogd: Only one -S argument allowed, the first one is taken.\n");
+ } else {
+ glbl.SetSourceIPofLocalClient((uchar*)arg);
+ }
+ break;
case 'f': /* configuration file */
ConfFile = (uchar*) arg;
break;
generated by cgit v1.2.3 (git 2.25.1) at 2025-07-08 23:45:17 +0000