summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--plugins/imptcp/imptcp.c45
-rw-r--r--tools/omfwd.c62
2 files changed, 91 insertions, 16 deletions
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index 7481611d..d3551225 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -10,7 +10,7 @@
*
* File begun on 2010-08-10 by RGerhards
*
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH.
*
* This file is part of rsyslog.
*
@@ -50,6 +50,7 @@
#include <sys/socket.h>
#include <sys/epoll.h>
#include <netinet/tcp.h>
+#include <stdint.h>
#include <zlib.h>
#if HAVE_FCNTL_H
#include <fcntl.h>
@@ -94,6 +95,11 @@ static void * wrkr(void *myself);
#define DFLT_wrkrMax 2
+#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
+
/* config settings */
typedef struct configSettings_s {
int bKeepAlive; /* support keep-alive packets */
@@ -118,6 +124,7 @@ struct instanceConf_s {
int bEmitMsgOnClose;
int bSuppOctetFram; /* support octet-counted framing? */
int iAddtlFrameDelim;
+ uint8_t compressionMode;
uchar *pszBindPort; /* port to bind to */
uchar *pszBindAddr; /* IP to bind socket to */
uchar *pszBindRuleset; /* name of ruleset to bind to */
@@ -157,6 +164,7 @@ static struct cnfparamdescr inppdescr[] = {
{ "ruleset", eCmdHdlrString, 0 },
{ "supportoctetcountedframing", eCmdHdlrBinary, 0 },
{ "notifyonconnectionclose", eCmdHdlrBinary, 0 },
+ { "compression.mode", eCmdHdlrGetWord, 0 },
{ "keepalive", eCmdHdlrBinary, 0 },
{ "keepalive.probes", eCmdHdlrInt, 0 },
{ "keepalive.time", eCmdHdlrInt, 0 },
@@ -192,6 +200,7 @@ struct ptcpsrv_s {
int iKeepAliveIntvl;
int iKeepAliveProbes;
int iKeepAliveTime;
+ uint8_t compressionMode;
uchar *pszInputName;
prop_t *pInputName; /* InputName in (fast to process) property format */
ruleset_t *pRuleset;
@@ -214,6 +223,7 @@ struct ptcpsess_s {
epolld_t *epd;
sbool bzInitDone; /* did we do an init of zstrm already? */
z_stream zstrm; /* zip stream to use for tcp compression */
+ uint8_t compressionMode;
//--- from tcps_sess.h
int iMsg; /* index of next char to store in msg */
int bAtStrtOfFram; /* are we at the very beginning of a new frame? */
@@ -839,7 +849,7 @@ finalize_it:
}
static rsRetVal
-DataRcvdCompress(ptcpsess_t *pThis, char *buf, size_t len)
+DataRcvdCompressed(ptcpsess_t *pThis, char *buf, size_t len)
{
struct syslogTime stTime;
time_t ttGenTime;
@@ -894,7 +904,12 @@ finalize_it:
static rsRetVal
DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen)
{
- return DataRcvdCompress(pThis, pData, iLen);
+ DEFiRet;
+ if(pThis->compressionMode >= COMPRESS_STREAM_ALWAYS)
+ iRet = DataRcvdCompressed(pThis, pData, iLen);
+ else
+ iRet = DataRcvdUncompressed(pThis, pData, iLen, 0);
+ RETiRet;
}
@@ -1032,6 +1047,7 @@ addSess(ptcplstn_t *pLstn, int sock, prop_t *peerName, prop_t *peerIP)
pSess->bAtStrtOfFram = 1;
pSess->peerName = peerName;
pSess->peerIP = peerIP;
+ pSess->compressionMode = pLstn->pSrv->compressionMode;
/* add to start of server's listener list */
pSess->prev = NULL;
@@ -1062,10 +1078,11 @@ doZipFinish(ptcpsess_t *pSess)
if(!pSess->bzInitDone)
goto done;
+dbgprintf("DDDD: enter doZipFinish\n");
pSess->zstrm.avail_in = 0;
/* run inflate() on buffer until everything has been compressed */
do {
- DBGPRINTF("in inflate() loop, avail_in %d, total_in %ld\n", pSess->zstrm.avail_in, pSess->zstrm.total_in);
+ DBGPRINTF("doZipFinish: in inflate() loop, avail_in %d, total_in %ld\n", pSess->zstrm.avail_in, pSess->zstrm.total_in);
pSess->zstrm.avail_out = sizeof(zipBuf);
pSess->zstrm.next_out = zipBuf;
zRet = inflate(&pSess->zstrm, Z_FINISH); /* no bad return value */
@@ -1077,9 +1094,9 @@ doZipFinish(ptcpsess_t *pSess)
} while (pSess->zstrm.avail_out == 0);
finalize_it:
- zRet = deflateEnd(&pSess->zstrm);
+ zRet = inflateEnd(&pSess->zstrm);
if(zRet != Z_OK) {
- DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet);
+ DBGPRINTF("imptcp: error %d returned from zlib/inflateEnd()\n", zRet);
}
pSess->bzInitDone = 0;
@@ -1146,6 +1163,7 @@ createInstance(instanceConf_t **pinst)
inst->pBindRuleset = NULL;
inst->ratelimitBurst = 10000; /* arbitrary high limit */
inst->ratelimitInterval = 0; /* off */
+ inst->compressionMode = COMPRESS_SINGLE_MSG;
/* node created, let's add to config */
if(loadModConf->tail == NULL) {
@@ -1225,6 +1243,7 @@ addListner(modConfData_t __attribute__((unused)) *modConf, instanceConf_t *inst)
pSrv->iKeepAliveProbes = inst->iKeepAliveProbes;
pSrv->iKeepAliveTime = inst->iKeepAliveTime;
pSrv->bEmitMsgOnClose = inst->bEmitMsgOnClose;
+ pSrv->compressionMode = inst->compressionMode;
CHKiRet(ratelimitNew(&pSrv->ratelimiter, "imtcp", (char*)inst->pszBindPort));
ratelimitSetLinuxLike(pSrv->ratelimiter, inst->ratelimitInterval, inst->ratelimitBurst);
ratelimitSetThreadSafe(pSrv->ratelimiter);
@@ -1519,6 +1538,7 @@ wrkr(void *myself)
BEGINnewInpInst
struct cnfparamvals *pvals;
instanceConf_t *inst;
+ char *cstr;
int i;
CODESTARTnewInpInst
DBGPRINTF("newInpInst (imptcp)\n");
@@ -1550,6 +1570,19 @@ CODESTARTnewInpInst
inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
} else if(!strcmp(inppblk.descr[i].name, "supportoctetcountedframing")) {
inst->bSuppOctetFram = (int) pvals[i].val.d.n;
+ } else if(!strcmp(inppblk.descr[i].name, "compression.mode")) {
+ cstr = es_str2cstr(pvals[i].val.d.estr, NULL);
+ if(!strcasecmp(cstr, "stream:always")) {
+ inst->compressionMode = COMPRESS_STREAM_ALWAYS;
+ } else if(!strcasecmp(cstr, "none")) {
+ inst->compressionMode = COMPRESS_NEVER;
+ } 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 if(!strcmp(inppblk.descr[i].name, "keepalive")) {
inst->bKeepAlive = (int) pvals[i].val.d.n;
} else if(!strcmp(inppblk.descr[i].name, "keepalive.probes")) {
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 61a5d57b..179e9614 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.
*
@@ -39,6 +39,7 @@
#include <errno.h>
#include <ctype.h>
#include <unistd.h>
+#include <stdint.h>
#ifdef USE_NETZIP
#include <zlib.h>
#endif
@@ -97,6 +98,11 @@ typedef struct _instanceData {
TCPFRAMINGMODE tcp_framing;
int bResendLastOnRecon; /* should the last message be re-sent on a successful reconnect? */
tcpclt_t *pTCPClt; /* our tcpclt object */
+# 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;
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 */
@@ -134,6 +140,7 @@ static struct cnfparamdescr actpdescr[] = {
{ "protocol", eCmdHdlrGetWord, 0 },
{ "tcp_framing", eCmdHdlrGetWord, 0 },
{ "ziplevel", eCmdHdlrInt, 0 },
+ { "compression.mode", eCmdHdlrGetWord, 0 },
{ "rebindinterval", eCmdHdlrInt, 0 },
{ "streamdriver", eCmdHdlrGetWord, 0 },
{ "streamdrivermode", eCmdHdlrInt, 0 },
@@ -428,7 +435,7 @@ finalize_it:
/* CODE FOR SENDING TCP MESSAGES */
static rsRetVal
-TCPSendUncompressedBuf(instanceData *pData, uchar *buf, unsigned len)
+TCPSendBufUncompressed(instanceData *pData, uchar *buf, unsigned len)
{
DEFiRet;
unsigned alreadySent;
@@ -488,7 +495,7 @@ TCPSendBufCompressed(instanceData *pData, uchar *buf, unsigned len)
DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pData->zstrm.avail_out);
outavail = sizeof(zipBuf) - pData->zstrm.avail_out;
if(outavail != 0) {
- CHKiRet(TCPSendUncompressedBuf(pData, zipBuf, outavail));
+ CHKiRet(TCPSendBufUncompressed(pData, zipBuf, outavail));
}
} while (pData->zstrm.avail_out == 0);
@@ -499,7 +506,12 @@ finalize_it:
static rsRetVal
TCPSendBuf(instanceData *pData, uchar *buf, unsigned len)
{
- return TCPSendBufCompressed(pData, buf, len);
+ DEFiRet;
+ if(pData->compressionMode >= COMPRESS_STREAM_ALWAYS)
+ iRet = TCPSendBufCompressed(pData, buf, len);
+ else
+ iRet = TCPSendBufUncompressed(pData, buf, len);
+ RETiRet;
}
/* finish zlib buffer, to be called before closing the ZIP file (if
@@ -516,6 +528,7 @@ doZipFinish(instanceData *pData)
if(!pData->bzInitDone)
goto done;
+// TODO: can we get this into a single common function?
dbgprintf("DDDD: in doZipFinish()\n");
pData->zstrm.avail_in = 0;
/* run deflate() on buffer until everything has been compressed */
@@ -527,7 +540,7 @@ dbgprintf("DDDD: in doZipFinish()\n");
DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pData->zstrm.avail_out);
outavail = sizeof(zipBuf) - pData->zstrm.avail_out;
if(outavail != 0) {
- CHKiRet(TCPSendUncompressedBuf(pData, zipBuf, outavail));
+ CHKiRet(TCPSendBufUncompressed(pData, zipBuf, outavail));
}
} while (pData->zstrm.avail_out == 0);
@@ -722,11 +735,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';
@@ -842,14 +854,17 @@ setInstParamDefaults(instanceData *pData)
pData->iRebindInterval = 0;
pData->bResendLastOnRecon = 0;
pData->pPermPeers = NULL;
- pData->compressionLevel = 0;
+ pData->compressionLevel = 9;
+ pData->compressionMode = COMPRESS_NEVER;
}
BEGINnewActInst
struct cnfparamvals *pvals;
uchar *tplToUse;
+ char *cstr;
int i;
rsRetVal localRet;
+ int complevel = -1;
CODESTARTnewActInst
DBGPRINTF("newActInst (omfwd)\n");
@@ -946,9 +961,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.",
@@ -962,11 +978,37 @@ CODESTARTnewActInst
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.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);
@@ -1034,6 +1076,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.",
@@ -1111,7 +1154,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)! */
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-21 18:59:50 +0000