summaryrefslogtreecommitdiffstats
path: root/plugins/imptcp/imptcp.c
diff options
context:
space:
mode:
Diffstat (limited to 'plugins/imptcp/imptcp.c')
-rw-r--r--plugins/imptcp/imptcp.c179
1 files changed, 158 insertions, 21 deletions
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index 91bdf8b3..008f7920 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -49,6 +49,7 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/epoll.h>
+#include <netinet/tcp.h>
#if HAVE_FCNTL_H
#include <fcntl.h>
#endif
@@ -65,6 +66,7 @@
#include "datetime.h"
#include "ruleset.h"
#include "msg.h"
+#include "statsobj.h"
#include "net.h" /* for permittedPeers, may be removed when this is removed */
/* the define is from tcpsrv.h, we need to find a new (but easier!!!) abstraction layer some time ... */
@@ -82,12 +84,17 @@ DEFobjCurrIf(prop)
DEFobjCurrIf(datetime)
DEFobjCurrIf(errmsg)
DEFobjCurrIf(ruleset)
-
+DEFobjCurrIf(statsobj)
/* config settings */
typedef struct configSettings_s {
+ int bKeepAlive; /* support keep-alive packets */
+ int iKeepAliveIntvl;
+ int iKeepAliveProbes;
+ int iKeepAliveTime;
int bEmitMsgOnClose; /* emit an informational message on close by remote peer */
+ int bSuppOctetFram; /* support octet-counted framing? */
int iAddtlFrameDelim; /* addtl frame delimiter, e.g. for netscreen, default none */
uchar *pszInputName; /* value for inputname property, NULL is OK and handled by core engine */
uchar *lstnIP; /* which IP we should listen on? */
@@ -110,26 +117,33 @@ struct ptcpsrv_s {
ptcpsrv_t *pNext; /* linked list maintenance */
uchar *port; /* Port to listen to */
uchar *lstnIP; /* which IP we should listen on? */
- int bEmitMsgOnClose;
int iAddtlFrameDelim;
+ int iKeepAliveIntvl;
+ int iKeepAliveProbes;
+ int iKeepAliveTime;
uchar *pszInputName;
prop_t *pInputName; /* InputName in (fast to process) property format */
ruleset_t *pRuleset;
ptcplstn_t *pLstn; /* root of our listeners */
ptcpsess_t *pSess; /* root of our sessions */
+ sbool bKeepAlive; /* support keep-alive packets */
+ sbool bEmitMsgOnClose;
+ sbool bSuppOctetFram;
};
/* the ptcp session object. Describes a single active session.
* includes support for doubly-linked list.
*/
struct ptcpsess_s {
- ptcpsrv_t *pSrv; /* our server */
+// ptcpsrv_t *pSrv; /* our server TODO: check remove! */
+ ptcplstn_t *pLstn; /* our listener */
ptcpsess_t *prev, *next;
int sock;
epolld_t *epd;
//--- 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? */
+ sbool bSuppOctetFram; /**< copy from listener, to speed up access */
enum {
eAtStrtFram,
eInOctetCnt,
@@ -150,7 +164,10 @@ struct ptcplstn_s {
ptcpsrv_t *pSrv; /* our server */
ptcplstn_t *prev, *next;
int sock;
+ sbool bSuppOctetFram;
epolld_t *epd;
+ statsobj_t *stats; /* listener stats */
+ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
};
@@ -188,7 +205,7 @@ static char rcvBuf[128*1024];
/* forward definitions */
static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal);
-static rsRetVal addLstn(ptcpsrv_t *pSrv, int sock);
+static rsRetVal addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6);
/* some simple constructors/destructors */
@@ -233,6 +250,7 @@ startupSrv(ptcpsrv_t *pSrv)
int sockflags;
struct addrinfo hints, *res = NULL, *r;
uchar *lstnIP;
+ int isIPv6 = 0;
lstnIP = pSrv->lstnIP == NULL ? UCHAR_CONSTANT("") : pSrv->lstnIP;
@@ -265,8 +283,9 @@ startupSrv(ptcpsrv_t *pSrv)
continue;
}
-#ifdef IPV6_V6ONLY
if(r->ai_family == AF_INET6) {
+ isIPv6 = 1;
+#ifdef IPV6_V6ONLY
int iOn = 1;
if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY,
(char *)&iOn, sizeof (iOn)) < 0) {
@@ -274,8 +293,10 @@ startupSrv(ptcpsrv_t *pSrv)
sock = -1;
continue;
}
- }
#endif
+ } else {
+ isIPv6 = 0;
+ }
if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) {
DBGPRINTF("error %d setting tcp socket option\n", errno);
close(sock);
@@ -337,7 +358,7 @@ startupSrv(ptcpsrv_t *pSrv)
/* if we reach this point, we were able to obtain a valid socket, so we can
* create our listener object. -- rgerhards, 2010-08-10
*/
- CHKiRet(addLstn(pSrv, sock));
+ CHKiRet(addLstn(pSrv, sock, isIPv6));
++numSocks;
}
@@ -428,12 +449,80 @@ finalize_it:
}
+/* Enable KEEPALIVE handling on the socket. */
+static inline rsRetVal
+EnableKeepAlive(ptcplstn_t *pLstn, int sock)
+{
+ int ret;
+ int optval;
+ socklen_t optlen;
+ DEFiRet;
+
+ optval = 1;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen);
+ if(ret < 0) {
+ dbgprintf("EnableKeepAlive socket call returns error %d\n", ret);
+ ABORT_FINALIZE(RS_RET_ERR);
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveProbes > 0) {
+ optval = pLstn->pSrv->iKeepAliveProbes;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPCNT, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(ret, NO_ERRCODE, "imptcp cannot set keepalive probes - ignored");
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveTime > 0) {
+ optval = pLstn->pSrv->iKeepAliveTime;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPIDLE, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(ret, NO_ERRCODE, "imptcp cannot set keepalive time - ignored");
+ }
+
+# if defined(TCP_KEEPCNT)
+ if(pLstn->pSrv->iKeepAliveIntvl > 0) {
+ optval = pLstn->pSrv->iKeepAliveIntvl;
+ optlen = sizeof(optval);
+ ret = setsockopt(sock, SOL_TCP, TCP_KEEPINTVL, &optval, optlen);
+ } else {
+ ret = 0;
+ }
+# else
+ ret = -1;
+# endif
+ if(ret < 0) {
+ errmsg.LogError(errno, NO_ERRCODE, "imptcp cannot set keepalive intvl - ignored");
+ }
+
+ dbgprintf("KEEPALIVE enabled for socket %d\n", sock);
+
+finalize_it:
+ RETiRet;
+}
+
/* accept an incoming connection request
* rgerhards, 2008-04-22
*/
static rsRetVal
-AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP)
+AcceptConnReq(ptcplstn_t *pLstn, int *newSock, prop_t **peerName, prop_t **peerIP)
{
int sockflags;
struct sockaddr_storage addr;
@@ -442,13 +531,17 @@ AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP)
DEFiRet;
- iNewSock = accept(sock, (struct sockaddr*) &addr, &addrlen);
+ iNewSock = accept(pLstn->sock, (struct sockaddr*) &addr, &addrlen);
if(iNewSock < 0) {
if(errno == EAGAIN || errno == EWOULDBLOCK)
ABORT_FINALIZE(RS_RET_NO_MORE_DATA);
ABORT_FINALIZE(RS_RET_ACCEPT_ERR);
}
+ if(pLstn->pSrv->bKeepAlive)
+ EnableKeepAlive(pLstn, iNewSock);/* we ignore errors, best to do! */
+
+
CHKiRet(getPeerNames(peerName, peerIP, (struct sockaddr*) &addr));
/* set the new socket to non-blocking IO */
@@ -492,22 +585,25 @@ static rsRetVal
doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub)
{
msg_t *pMsg;
+ ptcpsrv_t *pSrv;
DEFiRet;
if(pThis->iMsg == 0) {
DBGPRINTF("discarding zero-sized message\n");
FINALIZE;
}
+ pSrv = pThis->pLstn->pSrv;
/* we now create our own message object and submit it to the queue */
CHKiRet(msgConstructWithTime(&pMsg, stTime, ttGenTime));
MsgSetRawMsg(pMsg, (char*)pThis->pMsg, pThis->iMsg);
- MsgSetInputName(pMsg, pThis->pSrv->pInputName);
+ MsgSetInputName(pMsg, pSrv->pInputName);
MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME;
MsgSetRcvFrom(pMsg, pThis->peerName);
CHKiRet(MsgSetRcvFromIP(pMsg, pThis->peerIP));
- MsgSetRuleset(pMsg, pThis->pSrv->pRuleset);
+ MsgSetRuleset(pMsg, pSrv->pRuleset);
+ STATSCOUNTER_INC(pThis->pLstn->ctrSubmit, pThis->pLstn->mutCtrSubmit);
if(pMultiSub == NULL) {
CHKiRet(submitMsg(pMsg));
@@ -540,7 +636,7 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG
DEFiRet;
if(pThis->inputState == eAtStrtFram) {
- if(isdigit((int) c)) {
+ if(pThis->bSuppOctetFram && isdigit((int) c)) {
pThis->inputState = eInOctetCnt;
pThis->iOctetsRemain = 0;
pThis->eFraming = TCP_FRAMING_OCTET_COUNTING;
@@ -589,7 +685,8 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG
}
if(( (c == '\n')
- || ((pThis->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->iAddtlFrameDelim))
+ || ((pThis->pLstn->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER)
+ && (c == pThis->pLstn->pSrv->iAddtlFrameDelim))
) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */
doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub);
pThis->inputState = eAtStrtFram;
@@ -678,6 +775,7 @@ static inline void
initConfigSettings(void)
{
cs.bEmitMsgOnClose = 0;
+ cs.bSuppOctetFram = 1;
cs.iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
cs.pszInputName = NULL;
cs.pRuleset = NULL;
@@ -693,7 +791,7 @@ addEPollSock(epolld_type_t typ, void *ptr, int sock, epolld_t **pEpd)
DEFiRet;
epolld_t *epd = NULL;
- CHKmalloc(epd = malloc(sizeof(epolld_t)));
+ CHKmalloc(epd = calloc(sizeof(epolld_t), 1));
epd->typ = typ;
epd->ptr = ptr;
*pEpd = epd;
@@ -747,14 +845,27 @@ finalize_it:
/* add a listener to the server
*/
static rsRetVal
-addLstn(ptcpsrv_t *pSrv, int sock)
+addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6)
{
DEFiRet;
ptcplstn_t *pLstn;
+ uchar statname[64];
CHKmalloc(pLstn = malloc(sizeof(ptcplstn_t)));
pLstn->pSrv = pSrv;
+ pLstn->bSuppOctetFram = pSrv->bSuppOctetFram;
pLstn->sock = sock;
+ /* support statistics gathering */
+ CHKiRet(statsobj.Construct(&(pLstn->stats)));
+ snprintf((char*)statname, sizeof(statname), "imptcp(%s/%s/%s)",
+ (pSrv->lstnIP == NULL) ? "*" : (char*)pSrv->lstnIP, pSrv->port,
+ isIPv6 ? "IPv6" : "IPv4");
+ statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
+ CHKiRet(statsobj.SetName(pLstn->stats, statname));
+ STATSCOUNTER_INIT(pLstn->ctrSubmit, pLstn->mutCtrSubmit);
+ CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("submitted"),
+ ctrType_IntCtr, &(pLstn->ctrSubmit)));
+ CHKiRet(statsobj.ConstructFinalize(pLstn->stats));
/* add to start of server's listener list */
pLstn->prev = NULL;
@@ -773,15 +884,17 @@ finalize_it:
/* add a session to the server
*/
static rsRetVal
-addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP)
+addSess(ptcplstn_t *pLstn, int sock, prop_t *peerName, prop_t *peerIP)
{
DEFiRet;
ptcpsess_t *pSess = NULL;
+ ptcpsrv_t *pSrv = pLstn->pSrv;
CHKmalloc(pSess = malloc(sizeof(ptcpsess_t)));
CHKmalloc(pSess->pMsg = malloc(iMaxLine * sizeof(uchar)));
- pSess->pSrv = pSrv;
+ pSess->pLstn = pLstn;
pSess->sock = sock;
+ pSess->bSuppOctetFram = pLstn->bSuppOctetFram;
pSess->inputState = eAtStrtFram;
pSess->iMsg = 0;
pSess->bAtStrtOfFram = 1;
@@ -824,7 +937,7 @@ closeSess(ptcpsess_t *pSess)
pSess->next->prev = pSess->prev;
if(pSess->prev == NULL) {
/* need to update root! */
- pSess->pSrv->pSess = pSess->next;
+ pSess->pLstn->pSrv->pSess = pSess->next;
} else {
pSess->prev->next = pSess->next;
}
@@ -882,6 +995,11 @@ static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVa
CHKmalloc(pSrv = malloc(sizeof(ptcpsrv_t)));
pSrv->pSess = NULL;
pSrv->pLstn = NULL;
+ pSrv->bKeepAlive = cs.bKeepAlive;
+ pSrv->iKeepAliveIntvl = cs.iKeepAliveTime;
+ pSrv->iKeepAliveProbes = cs.iKeepAliveProbes;
+ pSrv->iKeepAliveTime = cs.iKeepAliveTime;
+ pSrv->bSuppOctetFram = cs.bSuppOctetFram;
pSrv->bEmitMsgOnClose = cs.bEmitMsgOnClose;
pSrv->port = pNewVal;
pSrv->iAddtlFrameDelim = cs.iAddtlFrameDelim;
@@ -945,11 +1063,11 @@ lstnActivity(ptcplstn_t *pLstn)
DBGPRINTF("imptcp: new connection on listen socket %d\n", pLstn->sock);
while(1) {
- localRet = AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP);
+ localRet = AcceptConnReq(pLstn, &newSock, &peerName, &peerIP);
if(localRet == RS_RET_NO_MORE_DATA)
break;
CHKiRet(localRet);
- CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP));
+ CHKiRet(addSess(pLstn, newSock, peerName, peerIP));
}
finalize_it:
@@ -979,7 +1097,7 @@ sessActivity(ptcpsess_t *pSess)
CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv));
} else if (lenRcv == 0) {
/* session was closed, do clean-up */
- if(pSess->pSrv->bEmitMsgOnClose) {
+ if(pSess->pLstn->pSrv->bEmitMsgOnClose) {
uchar *peerName;
int lenPeer;
prop.GetString(pSess->peerName, &peerName, &lenPeer);
@@ -1085,6 +1203,8 @@ shutdownSrv(ptcpsrv_t *pSrv)
pLstn = pSrv->pLstn;
while(pLstn != NULL) {
close(pLstn->sock);
+ statsobj.Destruct(&(pLstn->stats));
+ /* now unlink listner */
lstnDel = pLstn;
pLstn = pLstn->next;
DBGPRINTF("imptcp shutdown listen socket %d\n", lstnDel->sock);
@@ -1132,6 +1252,7 @@ CODESTARTmodExit
/* release objects we used */
objRelease(glbl, CORE_COMPONENT);
+ objRelease(statsobj, CORE_COMPONENT);
objRelease(prop, CORE_COMPONENT);
objRelease(net, LM_NET_FILENAME);
objRelease(datetime, CORE_COMPONENT);
@@ -1144,6 +1265,11 @@ static rsRetVal
resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal)
{
cs.bEmitMsgOnClose = 0;
+ cs.bKeepAlive = 0;
+ cs.iKeepAliveProbes = 0;
+ cs.iKeepAliveTime = 0;
+ cs.iKeepAliveIntvl = 0;
+ cs.bSuppOctetFram = 1;
cs.iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER;
free(cs.pszInputName);
cs.pszInputName = NULL;
@@ -1167,6 +1293,7 @@ CODEmodInit_QueryRegCFSLineHdlr
initConfigSettings();
/* request objects we use */
CHKiRet(objUse(glbl, CORE_COMPONENT));
+ CHKiRet(objUse(statsobj, CORE_COMPONENT));
CHKiRet(objUse(prop, CORE_COMPONENT));
CHKiRet(objUse(net, LM_NET_FILENAME));
CHKiRet(objUse(errmsg, CORE_COMPONENT));
@@ -1176,6 +1303,16 @@ CODEmodInit_QueryRegCFSLineHdlr
/* register config file handlers */
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverrun"), 0, eCmdHdlrGetWord,
addTCPListener, NULL, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive"), 0, eCmdHdlrBinary,
+ NULL, &cs.bKeepAlive, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_probes"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveProbes, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_time"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveTime, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverkeepalive_intvl"), 0, eCmdHdlrInt,
+ NULL, &cs.iKeepAliveIntvl, STD_LOADABLE_MODULE_ID));
+ CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserversupportoctetcountedframing"), 0, eCmdHdlrBinary,
+ NULL, &cs.bSuppOctetFram, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpservernotifyonconnectionclose"), 0,
eCmdHdlrBinary, NULL, &cs.bEmitMsgOnClose, STD_LOADABLE_MODULE_ID));
CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserveraddtlframedelimiter"), 0, eCmdHdlrInt,
generated by cgit (git 2.34.1) at 2024-05-03 22:32:17 +0000