diff options
Diffstat (limited to 'tools/omfwd.c')
| -rw-r--r-- | tools/omfwd.c | 649 |
1 files changed, 508 insertions, 141 deletions
diff --git a/tools/omfwd.c b/tools/omfwd.c index 10cce0e..2fd24bd 100644 --- a/tools/omfwd.c +++ b/tools/omfwd.c @@ -4,30 +4,26 @@ * 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) - * This file is under development and has not yet arrived at being fully - * self-contained and a real object. So far, it is mostly an excerpt - * 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, 2009 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2012 Adiscon GmbH. * * This file is part of rsyslog. * - * Rsyslog is free software: you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation, either version 3 of the License, or - * (at your option) any later version. - * - * Rsyslog is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with Rsyslog. If not, see <http://www.gnu.org/licenses/>. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * -or- + * see COPYING.ASL20 in the source distribution + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * 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. * - * A copy of the GPL can be found in the file "COPYING" in this distribution. + * TODO v6 config: + * - permitted peer *list* */ #include "config.h" #include "rsyslog.h" @@ -62,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 */ @@ -77,41 +75,138 @@ 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 */ +/* module-global parameters */ +static struct cnfparamdescr modpdescr[] = { + { "template", eCmdHdlrGetWord, 0 }, +}; +static struct cnfparamblk modpblk = + { CNFPARAMBLK_VERSION, + sizeof(modpdescr)/sizeof(struct cnfparamdescr), + modpdescr + }; + +/* 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 + }; + +struct modConfData_s { + rsconf_t *pConf; /* our overall config object */ + uchar *tplName; /* default template */ +}; + +static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */ +static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */ + + +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); +/* this function gets the default template. It coordinates action between + * old-style and new-style configuration parts. + */ +static inline uchar* +getDfltTpl(void) +{ + if(loadModConf != NULL && loadModConf->tplName != NULL) + return loadModConf->tplName; + else if(cs.pszTplName == NULL) + return (uchar*)"RSYSLOG_TraditionalForwardFormat"; + else + return cs.pszTplName; +} + + +/* set the default template to be used + * This is a module-global parameter, and as such needs special handling. It needs to + * be coordinated with values set via the v2 config system (rsyslog v6+). What we do + * is we do not permit this directive after the v2 config system has been used to set + * the parameter. + */ +static rsRetVal +setLegacyDfltTpl(void __attribute__((unused)) *pVal, uchar* newVal) +{ + DEFiRet; + + if(loadModConf != NULL && loadModConf->tplName != NULL) { + free(newVal); + errmsg.LogError(0, RS_RET_ERR, "omfwd default template already set via module " + "global parameter - can no longer be changed"); + ABORT_FINALIZE(RS_RET_ERR); + } + free(cs.pszTplName); + cs.pszTplName = newVal; +finalize_it: + RETiRet; +} + /* Close the UDP sockets. * rgerhards, 2009-05-29 */ @@ -131,26 +226,14 @@ pData->bIsConnected = 0; // TODO: remove this variable altogether } -/* get the syslog forward port from selector_t. The passed in - * struct must be one that is setup for forwarding. - * rgerhards, 2007-06-28 - * We may change the implementation to try to lookup the port - * if it is unspecified. So far, we use the IANA default auf 514. - */ -static char *getFwdPt(instanceData *pData) -{ - assert(pData != NULL); - if(pData->port == NULL) - return("514"); - else - return(pData->port); -} - - /* destruct the TCP helper objects * 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) @@ -162,8 +245,75 @@ DestructTCPInstanceData(instanceData *pData) netstrms.Destruct(&pData->pNS); } + +BEGINbeginCnfLoad +CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; + pModConf->tplName = NULL; +ENDbeginCnfLoad + +BEGINsetModCnf + struct cnfparamvals *pvals = NULL; + int i; +CODESTARTsetModCnf + pvals = nvlstGetParams(lst, &modpblk, NULL); + if(pvals == NULL) { + errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "error processing module " + "config parameters [module(...)]"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + + if(Debug) { + dbgprintf("module (global) param blk for omfwd:\n"); + cnfparamsPrint(&modpblk, pvals); + } + + for(i = 0 ; i < modpblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(modpblk.descr[i].name, "template")) { + loadModConf->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + if(cs.pszTplName != NULL) { + errmsg.LogError(0, RS_RET_DUP_PARAM, "omfwd: warning: default template " + "was already set via legacy directive - may lead to inconsistent " + "results."); + } + } else { + dbgprintf("omfwd: program error, non-handled " + "param '%s' in beginCnfLoad\n", modpblk.descr[i].name); + } + } +finalize_it: + if(pvals != NULL) + cnfparamvalsDestruct(pvals, &modpblk); +ENDsetModCnf + +BEGINendCnfLoad +CODESTARTendCnfLoad + loadModConf = NULL; /* done loading */ + /* free legacy config vars */ + free(cs.pszTplName); + cs.pszTplName = NULL; +ENDendCnfLoad + +BEGINcheckCnf +CODESTARTcheckCnf +ENDcheckCnf + +BEGINactivateCnf +CODESTARTactivateCnf + runModConf = pModConf; +ENDactivateCnf + +BEGINfreeCnf +CODESTARTfreeCnf + free(pModConf->tplName); +ENDfreeCnf + BEGINcreateInstance CODESTARTcreateInstance + pData->offsSndBuf = 0; ENDcreateInstance @@ -185,7 +335,7 @@ CODESTARTfreeInstance } free(pData->port); - free(pData->f_hname); + free(pData->target); free(pData->pszStrmDrvr); free(pData->pszStrmDrvrAuthMode); net.DestructPermittedPeers(&pData->pPermPeers); @@ -194,7 +344,7 @@ ENDfreeInstance BEGINdbgPrintInstInfo CODESTARTdbgPrintInstInfo - dbgprintf("%s", pData->f_hname); + dbgprintf("%s", pData->target); ENDdbgPrintInstInfo @@ -209,7 +359,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)); @@ -227,12 +377,12 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len) * call fails. Then, lsent has the error status, even though * the sendto() succeeded. -- rgerhards, 2007-06-22 */ - bSendSuccess = FALSE; + 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); if (lsent == len) { - bSendSuccess = TRUE; + bSendSuccess = RSTRUE; break; } else { int eno = errno; @@ -245,7 +395,7 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len) break; } /* finished looping */ - if (bSendSuccess == FALSE) { + if (bSendSuccess == RSFALSE) { dbgprintf("error forwarding via udp, suspending\n"); iRet = RS_RET_SUSPENDED; } @@ -262,7 +412,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; @@ -273,30 +423,65 @@ 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; + + alreadySent = 0; + CHKiRet(netstrm.CheckConnection(pData->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)); + DBGPRINTF("omfwd: TCP sent %ld bytes, requested %u\n", (long) lenSend, len - alreadySent); + alreadySent += lenSend; + } + +finalize_it: + if(iRet != RS_RET_OK) { + /* error! */ + dbgprintf("TCPSendBuf error %d, destruct TCP Connection!\n", iRet); + DestructTCPInstanceData(pData); + iRet = RS_RET_SUSPENDED; + } + 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; - lenSend = len; - 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); + 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; + } - 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 */ + /* 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; } @@ -310,6 +495,7 @@ static rsRetVal TCPSendPrepRetry(void *pvData) { DEFiRet; instanceData *pData = (instanceData *) pvData; +dbgprintf("TCPSendPrepRetry performs a DestructTCPInstanceData\n"); assert(pData != NULL); DestructTCPInstanceData(pData); @@ -327,9 +513,10 @@ static rsRetVal TCPSendInit(void *pvData) assert(pData != NULL); if(pData->pNetstrm == NULL) { + dbgprintf("TCPSendInit CREATE\n"); 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 */ @@ -345,11 +532,12 @@ static rsRetVal TCPSendInit(void *pvData) } /* params set, now connect */ CHKiRet(netstrm.Connect(pData->pNetstrm, glbl.GetDefPFFamily(), - (uchar*)getFwdPt(pData), (uchar*)pData->f_hname)); + (uchar*)pData->port, (uchar*)pData->target)); } finalize_it: if(iRet != RS_RET_OK) { + dbgprintf("TCPSendInit FAILED with %d.\n", iRet); DestructTCPInstanceData(pData); } @@ -371,23 +559,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, getFwdPt(pData), &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, getFwdPt(pData), 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)); @@ -411,16 +599,26 @@ CODESTARTtryResume iRet = doTryResume(pData); ENDtryResume + +BEGINbeginTransaction +CODESTARTbeginTransaction +dbgprintf("omfwd: beginTransaction\n"); +ENDbeginTransaction + + BEGINdoAction - char *psz = NULL; /* temporary buffering */ + char *psz; /* temporary buffering */ register unsigned l; int iMaxLine; +# ifdef USE_NETZIP + Bytef *out = NULL; /* for compression */ +# endif CODESTARTdoAction CHKiRet(doTryResume(pData)); iMaxLine = glbl.GetMaxLine(); - dbgprintf(" %s:%s/%s\n", pData->f_hname, getFwdPt(pData), + dbgprintf(" %s:%s/%s\n", pData->target, pData->port, pData->protocol == FORW_UDP ? "udp" : "tcp"); psz = (char*) ppString[0]; @@ -438,7 +636,6 @@ CODESTARTdoAction * rgerhards, 2006-11-30 */ if(pData->compressionLevel && (l > CONF_MIN_SIZE_FOR_COMPRESS)) { - Bytef *out; uLongf destLen = iMaxLine + iMaxLine/100 +12; /* recommended value from zlib doc */ uLong srcLen = l; int ret; @@ -459,14 +656,11 @@ CODESTARTdoAction * rgerhards, 2006-11-30 */ dbgprintf("Compression failed, sending uncompressed message\n"); - free(out); } else if(destLen+1 < l) { /* only use compression if there is a gain in using it! */ dbgprintf("there is gain in compression, so we do it\n"); psz = (char*) out; l = destLen + 1; /* take care for the "z" at message start! */ - } else { - free(out); } ++destLen; } @@ -477,9 +671,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); @@ -488,14 +681,22 @@ CODESTARTdoAction } finalize_it: # ifdef USE_NETZIP - if((psz != NULL) && (psz != (char*) ppString[0])) { - /* we need to free temporary buffer, alloced above - Naoya Nakazawa, 2010-01-11 */ - free(psz); - } + free(out); /* is NULL if it was never used... */ # endif 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 @@ -513,6 +714,181 @@ 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(getDfltTpl()), OMSR_NO_RQD_TPL_OPTS)); + + CHKiRet(initTCP(pData)); +CODE_STD_FINALIZERnewActInst + cnfparamvalsDestruct(pvals, &actpblk); +ENDnewActInst + + BEGINparseSelectorAct uchar *q; int i; @@ -618,6 +994,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) /* JUST SKIP */; } + pData->tcp_framing = tcp_framing; pData->port = NULL; if(*p == ':') { /* process port */ uchar * tmp; @@ -630,12 +1007,16 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) if(pData->port == NULL) { errmsg.LogError(0, NO_ERRCODE, "Could not get memory to store syslog forwarding port, " "using default port, results may not be what you intend\n"); - /* we leave f_forw.port set to NULL, this is then handled by getFwdPt(). */ + /* we leave f_forw.port set to NULL, this is then handled below */ } else { memcpy(pData->port, tmp, i); *(pData->port + i) = '\0'; } } + /* check if no port is set. If so, we use the IANA-assigned port of 514 */ + if(pData->port == NULL) { + CHKmalloc(pData->port = strdup("514")); + } /* now skip to template */ while(*p && *p != ';' && *p != '#' && !isspace((int) *p)) @@ -645,42 +1026,33 @@ 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)); + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_NO_RQD_TPL_OPTS, getDfltTpl())); 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 @@ -691,21 +1063,12 @@ 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.pszStrmDrvr); + cs.pszStrmDrvr = NULL; + free(cs.pszStrmDrvrAuthMode); + cs.pszStrmDrvrAuthMode = NULL; + free(cs.pPermPeers); + cs.pPermPeers = NULL; /* TODO: fix in older builds! */ } @@ -718,7 +1081,6 @@ CODESTARTmodExit objRelease(netstrm, LM_NETSTRMS_FILENAME); objRelease(netstrms, LM_NETSTRMS_FILENAME); objRelease(tcpclt, LM_TCPCLT_FILENAME); - freeConfigVars(); ENDmodExit @@ -726,6 +1088,10 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES +CODEqueryEtryPt_STD_CONF2_QUERIES +CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES +CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES +CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */ ENDqueryEtryPt @@ -737,10 +1103,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; } @@ -748,20 +1114,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, setLegacyDfltTpl, NULL, 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 |