summaryrefslogtreecommitdiffstats
path: root/tools/omfwd.c
diff options
context:
space:
mode:
Diffstat (limited to 'tools/omfwd.c')
-rw-r--r--tools/omfwd.c447
1 files changed, 353 insertions, 94 deletions
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 812da109..2a19aae4 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -4,9 +4,6 @@
* NOTE: read comments in module-template.h to understand how this file
* works!
*
- * File begun on 2007-07-20 by RGerhards (extracted from syslogd.c, which
- * at the time of rsyslog fork from sysklogd was under BSD license)
- *
* Copyright 2007-2012 Adiscon GmbH.
*
* This file is part of rsyslog.
@@ -24,6 +21,9 @@
* 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"
@@ -58,9 +58,11 @@
#include "module-template.h"
#include "glbl.h"
#include "errmsg.h"
+#include "unicode-helper.h"
MODULE_TYPE_OUTPUT
MODULE_TYPE_NOKEEP
+MODULE_CNFNAME("omfwd")
/* internal structures
*/
@@ -73,38 +75,79 @@ DEFobjCurrIf(netstrm)
DEFobjCurrIf(tcpclt)
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 *f_hname;
+ 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 iUDPRebindInterval; /* rebind interval */
- int iTCPRebindInterval; /* rebind interval */
+ 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? */
tcpclt_t *pTCPClt; /* our tcpclt object */
+ uchar sndBuf[16*1024]; /* this is intensionally fixed -- see no good reason to make configurable */
+ unsigned offsSndBuf; /* next free spot in send buffer */
} instanceData;
/* config data */
-static uchar *pszTplName = NULL; /* name of the default template to use */
-static uchar *pszStrmDrvr = NULL; /* name of the stream driver to use */
-static int iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
-static int bResendLastOnRecon = 0; /* should the last message be re-sent on a successful reconnect? */
-static uchar *pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
-static int iUDPRebindInterval = 0; /* support for automatic re-binding (load balancers!). 0 - no rebind */
-static int iTCPRebindInterval = 0; /* support for automatic re-binding (load balancers!). 0 - no rebind */
+typedef struct configSettings_s {
+ uchar *pszTplName; /* name of the default template to use */
+ uchar *pszStrmDrvr; /* name of the stream driver to use */
+ int iStrmDrvrMode; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
+ int bResendLastOnRecon; /* should the last message be re-sent on a successful reconnect? */
+ uchar *pszStrmDrvrAuthMode; /* authentication mode to use */
+ int iTCPRebindInterval; /* support for automatic re-binding (load balancers!). 0 - no rebind */
+ int iUDPRebindInterval; /* support for automatic re-binding (load balancers!). 0 - no rebind */
+ permittedPeers_t *pPermPeers;
+} configSettings_t;
+static configSettings_t cs;
+
+/* tables for interfacing with the v6 config system */
+/* action (instance) parameters */
+static struct cnfparamdescr actpdescr[] = {
+ { "target", eCmdHdlrGetWord, 0 },
+ { "port", eCmdHdlrGetWord, 0 },
+ { "protocol", eCmdHdlrGetWord, 0 },
+ { "tcp_framing", eCmdHdlrGetWord, 0 },
+ { "ziplevel", eCmdHdlrInt, 0 },
+ { "rebindinterval", eCmdHdlrInt, 0 },
+ { "streamdriver", eCmdHdlrGetWord, 0 },
+ { "streamdrivermode", eCmdHdlrInt, 0 },
+ { "streamdriverauthmode", eCmdHdlrGetWord, 0 },
+ { "streamdriverpermittedpeers", eCmdHdlrGetWord, 0 },
+ { "resendlastmsgonreconnect", eCmdHdlrBinary, 0 },
+};
+static struct cnfparamblk actpblk =
+ { CNFPARAMBLK_VERSION,
+ sizeof(actpdescr)/sizeof(struct cnfparamdescr),
+ actpdescr
+ };
+
+
+BEGINinitConfVars /* (re)set config variables to default values */
+CODESTARTinitConfVars
+ cs.pszTplName = NULL; /* name of the default template to use */
+ cs.pszStrmDrvr = NULL; /* name of the stream driver to use */
+ cs.iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
+ cs.bResendLastOnRecon = 0; /* should the last message be re-sent on a successful reconnect? */
+ cs.pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
+ cs.iUDPRebindInterval = 0; /* support for automatic re-binding (load balancers!). 0 - no rebind */
+ cs.iTCPRebindInterval = 0; /* support for automatic re-binding (load balancers!). 0 - no rebind */
+ cs.pPermPeers = NULL;
+ENDinitConfVars
-static permittedPeers_t *pPermPeers = NULL;
static rsRetVal doTryResume(instanceData *pData);
@@ -131,6 +174,10 @@ pData->bIsConnected = 0; // TODO: remove this variable altogether
* This, for example, is needed after something went wrong.
* This function is void because it "can not" fail.
* 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
+ * loose data.
*/
static inline void
DestructTCPInstanceData(instanceData *pData)
@@ -144,6 +191,7 @@ DestructTCPInstanceData(instanceData *pData)
BEGINcreateInstance
CODESTARTcreateInstance
+ pData->offsSndBuf = 0;
ENDcreateInstance
@@ -165,7 +213,7 @@ CODESTARTfreeInstance
}
free(pData->port);
- free(pData->f_hname);
+ free(pData->target);
free(pData->pszStrmDrvr);
free(pData->pszStrmDrvrAuthMode);
net.DestructPermittedPeers(&pData->pPermPeers);
@@ -174,7 +222,7 @@ ENDfreeInstance
BEGINdbgPrintInstInfo
CODESTARTdbgPrintInstInfo
- dbgprintf("%s", pData->f_hname);
+ dbgprintf("%s", pData->target);
ENDdbgPrintInstInfo
@@ -189,7 +237,7 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
unsigned lsent = 0;
int bSendSuccess;
- if(pData->iUDPRebindInterval && (pData->nXmit++ % pData->iUDPRebindInterval == 0)) {
+ if(pData->iRebindInterval && (pData->nXmit++ % 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));
@@ -242,7 +290,7 @@ static rsRetVal
setPermittedPeer(void __attribute__((unused)) *pVal, uchar *pszID)
{
DEFiRet;
- CHKiRet(net.AddPermittedPeer(&pPermPeers, pszID));
+ CHKiRet(net.AddPermittedPeer(&cs.pPermPeers, pszID));
free(pszID); /* no longer needed, but we must free it as of interface def */
finalize_it:
RETiRet;
@@ -253,30 +301,59 @@ finalize_it:
/* CODE FOR SENDING TCP MESSAGES */
-/* Send a frame via plain TCP protocol
- * rgerhards, 2007-12-28
+/* 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 TCPSendFrame(void *pvData, char *msg, size_t len)
+static rsRetVal
+TCPSendBuf(instanceData *pData, uchar *buf, unsigned len)
{
DEFiRet;
+ unsigned alreadySent;
ssize_t lenSend;
- instanceData *pData = (instanceData *) pvData;
- lenSend = len;
+ alreadySent = 0;
netstrm.CheckConnection(pData->pNetstrm); /* hack for plain tcp syslog - see ptcp driver for details */
- CHKiRet(netstrm.Send(pData->pNetstrm, (uchar*)msg, &lenSend));
- dbgprintf("TCP sent %ld bytes, requested %ld\n", (long) lenSend, (long) len);
+ while(alreadySent != len) {
+ lenSend = len - alreadySent;
+ CHKiRet(netstrm.Send(pData->pNetstrm, buf+alreadySent, &lenSend));
+ DBGPRINTF("omfwd: TCP sent %ld bytes, requested %u\n", (long) lenSend, len - alreadySent);
+ alreadySent += lenSend;
+ }
- if(lenSend != (ssize_t) len) {
- /* no real error, could "just" not send everything...
- * For the time being, we ignore this...
- * rgerhards, 2005-10-25
- */
- dbgprintf("message not completely (tcp)send, ignoring %ld\n", (long) lenSend);
- usleep(1000); /* experimental - might be benefitial in this situation */
- /* TODO: we need to revisit this code -- rgerhards, 2007-12-28 */
+finalize_it:
+ 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;
+
+ 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)) {
+ /* no buffer space left, need to commit previous records */
+ CHKiRet(TCPSendBuf(pData, pData->sndBuf, pData->offsSndBuf));
+ pData->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));
+ 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;
+ iRet = RS_RET_DEFER_COMMIT;
+
finalize_it:
RETiRet;
}
@@ -309,7 +386,7 @@ static rsRetVal TCPSendInit(void *pvData)
if(pData->pNetstrm == NULL) {
CHKiRet(netstrms.Construct(&pData->pNS));
/* the stream driver must be set before the object is finalized! */
- CHKiRet(netstrms.SetDrvrName(pData->pNS, pszStrmDrvr));
+ CHKiRet(netstrms.SetDrvrName(pData->pNS, pData->pszStrmDrvr));
CHKiRet(netstrms.ConstructFinalize(pData->pNS));
/* now create the actual stream and connect to the server */
@@ -325,7 +402,7 @@ static rsRetVal TCPSendInit(void *pvData)
}
/* params set, now connect */
CHKiRet(netstrm.Connect(pData->pNetstrm, glbl.GetDefPFFamily(),
- (uchar*)pData->port, (uchar*)pData->f_hname));
+ (uchar*)pData->port, (uchar*)pData->target));
}
finalize_it:
@@ -351,23 +428,23 @@ static rsRetVal doTryResume(instanceData *pData)
FINALIZE;
/* The remote address is not yet known and needs to be obtained */
- dbgprintf(" %s\n", pData->f_hname);
+ dbgprintf(" %s\n", pData->target);
if(pData->protocol == FORW_UDP) {
memset(&hints, 0, sizeof(hints));
/* port must be numeric, because config file syntax requires this */
hints.ai_flags = AI_NUMERICSERV;
hints.ai_family = glbl.GetDefPFFamily();
hints.ai_socktype = SOCK_DGRAM;
- if((iErr = (getaddrinfo(pData->f_hname, pData->port, &hints, &res))) != 0) {
+ if((iErr = (getaddrinfo(pData->target, pData->port, &hints, &res))) != 0) {
dbgprintf("could not get addrinfo for hostname '%s':'%s': %d%s\n",
- pData->f_hname, pData->port, iErr, gai_strerror(iErr));
+ pData->target, pData->port, iErr, gai_strerror(iErr));
ABORT_FINALIZE(RS_RET_SUSPENDED);
}
- dbgprintf("%s found, resuming.\n", pData->f_hname);
+ 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->f_hname, NULL, 0);
+ pData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0);
}
} else {
CHKiRet(TCPSendInit((void*)pData));
@@ -391,6 +468,13 @@ CODESTARTtryResume
iRet = doTryResume(pData);
ENDtryResume
+
+BEGINbeginTransaction
+CODESTARTbeginTransaction
+dbgprintf("omfwd: beginTransaction\n");
+ENDbeginTransaction
+
+
BEGINdoAction
char *psz; /* temporary buffering */
register unsigned l;
@@ -403,7 +487,7 @@ CODESTARTdoAction
iMaxLine = glbl.GetMaxLine();
- dbgprintf(" %s:%s/%s\n", pData->f_hname, pData->port,
+ dbgprintf(" %s:%s/%s\n", pData->target, pData->port,
pData->protocol == FORW_UDP ? "udp" : "tcp");
psz = (char*) ppString[0];
@@ -456,9 +540,8 @@ CODESTARTdoAction
CHKiRet(UDPSend(pData, psz, l));
} else {
/* forward via TCP */
- rsRetVal ret;
- ret = tcpclt.Send(pData->pTCPClt, pData, psz, l);
- if(ret != RS_RET_OK) {
+ iRet = tcpclt.Send(pData->pTCPClt, pData, 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);
@@ -472,6 +555,17 @@ finalize_it:
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;
+ }
+ENDendTransaction
+
+
+
/* This function loads TCP support, if not already loaded. It will be called
* during config processing. To server ressources, TCP support will only
* be loaded if it actually is used. -- rgerhard, 2008-04-17
@@ -489,6 +583,183 @@ finalize_it:
}
+/* initialize TCP structures (if necessary) after the instance has been
+ * created.
+ */
+static rsRetVal
+initTCP(instanceData *pData)
+{
+ DEFiRet;
+ if(pData->protocol == FORW_TCP) {
+ /* create our tcpclt */
+ CHKiRet(tcpclt.Construct(&pData->pTCPClt));
+ CHKiRet(tcpclt.SetResendLastOnRecon(pData->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));
+ }
+finalize_it:
+ RETiRet;
+}
+
+
+static inline void
+setInstParamDefaults(instanceData *pData)
+{
+ pData->tplName = NULL;
+ pData->protocol = FORW_UDP;
+ pData->tcp_framing = TCP_FRAMING_OCTET_STUFFING;
+ pData->pszStrmDrvr = NULL;
+ pData->pszStrmDrvrAuthMode = NULL;
+ pData->iStrmDrvrMode = 0;
+ pData->iRebindInterval = 0;
+ pData->bResendLastOnRecon = 0;
+ pData->pPermPeers = NULL;
+ pData->compressionLevel = 0;
+}
+
+BEGINnewActInst
+ struct cnfparamvals *pvals;
+ int i;
+ rsRetVal localRet;
+CODESTARTnewActInst
+ DBGPRINTF("newActInst (omfwd)\n");
+
+ pvals = nvlstGetParams(lst, &actpblk, NULL);
+ if(pvals == NULL) {
+ errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "omfwd: either the \"file\" or "
+ "\"dynfile\" parameter must be given");
+ ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);
+ }
+
+ if(Debug) {
+ dbgprintf("action param blk in omfwd:\n");
+ cnfparamsPrint(&actpblk, pvals);
+ }
+
+ CHKiRet(createInstance(&pData));
+ setInstParamDefaults(pData);
+
+ for(i = 0 ; i < actpblk.nParams ; ++i) {
+ if(!pvals[i].bUsed)
+ continue;
+ if(!strcmp(actpblk.descr[i].name, "target")) {
+ pData->target = es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(actpblk.descr[i].name, "port")) {
+ pData->port = es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(actpblk.descr[i].name, "protocol")) {
+ if(!es_strcasebufcmp(pvals[i].val.d.estr, (uchar*)"udp", 3)) {
+ pData->protocol = FORW_UDP;
+ } else if(!es_strcasebufcmp(pvals[i].val.d.estr, (uchar*)"tcp", 3)) {
+ localRet = loadTCPSupport();
+ if(localRet != RS_RET_OK) {
+ errmsg.LogError(0, localRet, "could not activate network stream modules for TCP "
+ "(internal error %d) - are modules missing?", localRet);
+ ABORT_FINALIZE(localRet);
+ }
+ pData->protocol = FORW_TCP;
+ } else {
+ uchar *str;
+ str = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ errmsg.LogError(0, RS_RET_INVLD_PROTOCOL,
+ "omfwd: invalid protocol \"%s\"", str);
+ free(str);
+ ABORT_FINALIZE(RS_RET_INVLD_PROTOCOL);
+ }
+ } else if(!strcmp(actpblk.descr[i].name, "tcp_framing")) {
+ if(!es_strcasebufcmp(pvals[i].val.d.estr, (uchar*)"traditional", 11)) {
+ pData->tcp_framing = TCP_FRAMING_OCTET_STUFFING;
+ } else if(!es_strcasebufcmp(pvals[i].val.d.estr, (uchar*)"octet-counted", 13)) {
+ pData->tcp_framing = TCP_FRAMING_OCTET_COUNTING;
+ } else {
+ uchar *str;
+ str = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ errmsg.LogError(0, RS_RET_CNF_INVLD_FRAMING,
+ "omfwd: invalid framing \"%s\"", str);
+ free(str);
+ ABORT_FINALIZE(RS_RET_CNF_INVLD_FRAMING );
+ }
+ } else if(!strcmp(actpblk.descr[i].name, "rebindinterval")) {
+ pData->iRebindInterval = (int) pvals[i].val.d.n;
+ } else if(!strcmp(actpblk.descr[i].name, "streamdriver")) {
+ pData->pszStrmDrvr = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(actpblk.descr[i].name, "streamdrivermode")) {
+ pData->iStrmDrvrMode = pvals[i].val.d.n;
+ } else if(!strcmp(actpblk.descr[i].name, "streamdriverauthmode")) {
+ pData->pszStrmDrvrAuthMode = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ } else if(!strcmp(actpblk.descr[i].name, "streamdriverpermittedpeers")) {
+ uchar *start, *str;
+ uchar save;
+ uchar *p;
+ int lenStr;
+ str = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+ start = str;
+ lenStr = ustrlen(start); /* we need length after '\0' has been dropped... */
+ while(lenStr > 0) {
+ p = start;
+ while(*p && *p != ',' && lenStr--)
+ p++;
+ if(*p == ',') {
+ *p = '\0';
+ }
+ save = *(p+1); /* we always have this, at least the \0 byte at EOS */
+ *(p+1) = '\0';
+ if(*start == '\0') {
+ DBGPRINTF("omfwd: ignoring empty permitted peer\n");
+ } else {
+ dbgprintf("omfwd: adding permitted peer: '%s'\n", start);
+ CHKiRet(net.AddPermittedPeer(&(pData->pPermPeers), start));
+ }
+ start = p+1;
+ if(lenStr)
+ --lenStr;
+ *(p+1) = save;
+ }
+ free(str);
+ } else if(!strcmp(actpblk.descr[i].name, "ziplevel")) {
+# ifdef USE_NETZIP
+ int complevel = pvals[i].val.d.n;
+ if(complevel >= 0 && complevel <= 10) {
+ pData->compressionLevel = complevel;
+ } else {
+ errmsg.LogError(0, NO_ERRCODE, "Invalid ziplevel %d specified in "
+ "forwardig action - NOT turning on compression.",
+ complevel);
+ }
+# else
+ 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, "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 {
+ DBGPRINTF("omfwd: program error, non-handled "
+ "param '%s'\n", actpblk.descr[i].name);
+ }
+ }
+ CODE_STD_STRING_REQUESTnewActInst(1)
+
+ CHKiRet(OMSRsetEntry(*ppOMSR, 0, ustrdup((pData->tplName == NULL) ?
+ (uchar*)"RSYSLOG_TraditionalForwardFormat" : (uchar*)pData->tplName),
+ OMSR_NO_RQD_TPL_OPTS));
+
+ CHKiRet(initTCP(pData));
+CODE_STD_FINALIZERnewActInst
+ cnfparamvalsDestruct(pvals, &actpblk);
+ENDnewActInst
+
+
BEGINparseSelectorAct
uchar *q;
int i;
@@ -594,6 +865,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
/* JUST SKIP */;
}
+ pData->tcp_framing = tcp_framing;
pData->port = NULL;
if(*p == ':') { /* process port */
uchar * tmp;
@@ -625,42 +897,34 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
if(*p == ';' || *p == '#' || isspace(*p)) {
uchar cTmp = *p;
*p = '\0'; /* trick to obtain hostname (later)! */
- CHKmalloc(pData->f_hname = strdup((char*) q));
+ CHKmalloc(pData->target = strdup((char*) q));
*p = cTmp;
} else {
- CHKmalloc(pData->f_hname = strdup((char*) q));
+ CHKmalloc(pData->target = strdup((char*) q));
}
/* copy over config data as needed */
- pData->iUDPRebindInterval = iUDPRebindInterval;
- pData->iTCPRebindInterval = iTCPRebindInterval;
+ pData->iRebindInterval = (pData->protocol == FORW_TCP) ?
+ cs.iTCPRebindInterval : cs.iUDPRebindInterval;
/* process template */
CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_NO_RQD_TPL_OPTS,
- (pszTplName == NULL) ? (uchar*)"RSYSLOG_TraditionalForwardFormat" : pszTplName));
+ (cs.pszTplName == NULL) ? (uchar*)"RSYSLOG_TraditionalForwardFormat" : cs.pszTplName));
if(pData->protocol == FORW_TCP) {
- /* create our tcpclt */
- CHKiRet(tcpclt.Construct(&pData->pTCPClt));
- CHKiRet(tcpclt.SetResendLastOnRecon(pData->pTCPClt, 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, tcp_framing));
- CHKiRet(tcpclt.SetRebindInterval(pData->pTCPClt, pData->iTCPRebindInterval));
- pData->iStrmDrvrMode = iStrmDrvrMode;
- if(pszStrmDrvr != NULL)
- CHKmalloc(pData->pszStrmDrvr = (uchar*)strdup((char*)pszStrmDrvr));
- if(pszStrmDrvrAuthMode != NULL)
+ pData->bResendLastOnRecon = cs.bResendLastOnRecon;
+ 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*)pszStrmDrvrAuthMode));
- if(pPermPeers != NULL) {
- pData->pPermPeers = pPermPeers;
- pPermPeers = NULL;
+ (uchar*)strdup((char*)cs.pszStrmDrvrAuthMode));
+ if(cs.pPermPeers != NULL) {
+ pData->pPermPeers = cs.pPermPeers;
+ cs.pPermPeers = NULL;
}
}
-
+ CHKiRet(initTCP(pData));
CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct
@@ -671,21 +935,14 @@ ENDparseSelectorAct
static void
freeConfigVars(void)
{
- if(pszTplName != NULL) {
- free(pszTplName);
- pszTplName = NULL;
- }
- if(pszStrmDrvr != NULL) {
- free(pszStrmDrvr);
- pszStrmDrvr = NULL;
- }
- if(pszStrmDrvrAuthMode != NULL) {
- free(pszStrmDrvrAuthMode);
- pszStrmDrvrAuthMode = NULL;
- }
- if(pPermPeers != NULL) {
- free(pPermPeers);
- }
+ free(cs.pszTplName);
+ cs.pszTplName = NULL;
+ free(cs.pszStrmDrvr);
+ cs.pszStrmDrvr = NULL;
+ free(cs.pszStrmDrvrAuthMode);
+ cs.pszStrmDrvrAuthMode = NULL;
+ free(cs.pPermPeers);
+ cs.pPermPeers = NULL; /* TODO: fix in older builds! */
}
@@ -698,7 +955,6 @@ CODESTARTmodExit
objRelease(netstrm, LM_NETSTRMS_FILENAME);
objRelease(netstrms, LM_NETSTRMS_FILENAME);
objRelease(tcpclt, LM_TCPCLT_FILENAME);
-
freeConfigVars();
ENDmodExit
@@ -706,6 +962,8 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
+CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */
ENDqueryEtryPt
@@ -717,10 +975,10 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
freeConfigVars();
/* we now must reset all non-string values */
- iStrmDrvrMode = 0;
- bResendLastOnRecon = 0;
- iUDPRebindInterval = 0;
- iTCPRebindInterval = 0;
+ cs.iStrmDrvrMode = 0;
+ cs.bResendLastOnRecon = 0;
+ cs.iUDPRebindInterval = 0;
+ cs.iTCPRebindInterval = 0;
return RS_RET_OK;
}
@@ -728,20 +986,21 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
BEGINmodInit(Fwd)
CODESTARTmodInit
+INITLegCnfVars
*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(objUse(glbl, CORE_COMPONENT));
CHKiRet(objUse(errmsg, CORE_COMPONENT));
CHKiRet(objUse(net,LM_NET_FILENAME));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionforwarddefaulttemplate", 0, eCmdHdlrGetWord, NULL, &pszTplName, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendtcprebindinterval", 0, eCmdHdlrInt, NULL, &iTCPRebindInterval, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendudprebindinterval", 0, eCmdHdlrInt, NULL, &iUDPRebindInterval, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriver", 0, eCmdHdlrGetWord, NULL, &pszStrmDrvr, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdrivermode", 0, eCmdHdlrInt, NULL, &iStrmDrvrMode, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriverauthmode", 0, eCmdHdlrGetWord, NULL, &pszStrmDrvrAuthMode, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionforwarddefaulttemplate", 0, eCmdHdlrGetWord, NULL, &cs.pszTplName, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendtcprebindinterval", 0, eCmdHdlrInt, NULL, &cs.iTCPRebindInterval, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendudprebindinterval", 0, eCmdHdlrInt, NULL, &cs.iUDPRebindInterval, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriver", 0, eCmdHdlrGetWord, NULL, &cs.pszStrmDrvr, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdrivermode", 0, eCmdHdlrInt, NULL, &cs.iStrmDrvrMode, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriverauthmode", 0, eCmdHdlrGetWord, NULL, &cs.pszStrmDrvrAuthMode, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriverpermittedpeer", 0, eCmdHdlrGetWord, setPermittedPeer, NULL, NULL));
- CHKiRet(regCfSysLineHdlr((uchar *)"actionsendresendlastmsgonreconnect", 0, eCmdHdlrBinary, NULL, &bResendLastOnRecon, NULL));
+ CHKiRet(regCfSysLineHdlr((uchar *)"actionsendresendlastmsgonreconnect", 0, eCmdHdlrBinary, NULL, &cs.bResendLastOnRecon, NULL));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
ENDmodInit
generated by cgit (git 2.34.1) at 2024-05-07 04:19:52 +0000