From 55256ac96815d6e13fc9df7206d50ef7dcaca4fe Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 14:51:43 +0200 Subject: added imptcp imptcp is a simplified, Linux-specific and potentielly fast syslog plain tcp input plugin (NOT supporting TLS!) --- ChangeLog | 2 + Makefile.am | 7 +- configure.ac | 15 + doc/Makefile.am | 1 + doc/imptcp.html | 84 +++ doc/rsyslog_conf_modules.html | 3 +- plugins/imptcp/Makefile.am | 6 + plugins/imptcp/imptcp.c | 1146 +++++++++++++++++++++++++++++++++++++++++ runtime/rsyslog.h | 4 + tools/syslogd.c | 2 +- 10 files changed, 1267 insertions(+), 3 deletions(-) create mode 100644 doc/imptcp.html create mode 100644 plugins/imptcp/Makefile.am create mode 100644 plugins/imptcp/imptcp.c diff --git a/ChangeLog b/ChangeLog index df29694c..b8cbdb42 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,5 @@ +- added imptcp, a simplified, Linux-specific and potentielly fast + syslog plain tcp input plugin (NOT supporting TLS!) --------------------------------------------------------------------------- Version 4.6.4 [v4-stable] (rgerhards), 2010-08.05 - bugfix: zero-sized (empty) messages were processed by imtcp diff --git a/Makefile.am b/Makefile.am index f5f9a6ed..439cb417 100644 --- a/Makefile.am +++ b/Makefile.am @@ -103,6 +103,10 @@ if ENABLE_IMFILE SUBDIRS += plugins/imfile endif +if ENABLE_IMPTCP +SUBDIRS += plugins/imptcp +endif + if ENABLE_IMDIAG SUBDIRS += plugins/imdiag endif @@ -134,5 +138,6 @@ SUBDIRS += tests # temporarily be removed below. The intent behind forcing everthing to compile # in a make distcheck is so that we detect code that accidently was not updated # when some global update happened. -DISTCHECK_CONFIGURE_FLAGS=--enable-gssapi_krb5 --enable-imfile --enable-snmp --enable-pgsql --enable-libdbi --enable-mysql --enable-omtemplate --enable-imtemplate --enable-relp --enable-rsyslogd --enable-mail --enable-klog --enable-diagtools --enable-gnutls --enable-omstdout --enable-omprog --enable-imdiag --enable-shave --enable-extended-tests +DISTCHECK_CONFIGURE_FLAGS=--enable-gssapi_krb5 --enable-imfile --enable-snmp --enable-pgsql --enable-libdbi --enable-mysql --enable-omtemplate --enable-imtemplate --enable-relp --enable-rsyslogd --enable-mail --enable-klog --enable-diagtools --enable-gnutls --enable-omstdout --enable-omprog --enable-imdiag --enable-shave --enable-extended-tests \ + --enable-imptcp ACLOCAL_AMFLAGS = -I m4 diff --git a/configure.ac b/configure.ac index bc3965d5..ce34e53d 100644 --- a/configure.ac +++ b/configure.ac @@ -727,6 +727,19 @@ AC_ARG_ENABLE(imfile, AM_CONDITIONAL(ENABLE_IMFILE, test x$enable_imfile = xyes) +# settings for the ptcp input module +AC_ARG_ENABLE(imptcp, + [AS_HELP_STRING([--enable-imptcp],[plain tcp input module enabled @<:@default=no@:>@])], + [case "${enableval}" in + yes) enable_imptcp="yes" ;; + no) enable_imptcp="no" ;; + *) AC_MSG_ERROR(bad value ${enableval} for --enable-imptcp) ;; + esac], + [enable_imptcp=no] +) +AM_CONDITIONAL(ENABLE_IMPTCP, test x$enable_imptcp = xyes) + + # settings for the omprog output module AC_ARG_ENABLE(omprog, [AS_HELP_STRING([--enable-omprog],[Compiles omprog module @<:@default=no@:>@])], @@ -828,6 +841,7 @@ AC_CONFIG_FILES([Makefile \ plugins/omprog/Makefile \ plugins/omstdout/Makefile \ plugins/imfile/Makefile \ + plugins/imptcp/Makefile \ plugins/imrelp/Makefile \ plugins/imdiag/Makefile \ plugins/omtesting/Makefile \ @@ -857,6 +871,7 @@ echo " custom module 1 will be built: $enable_cust1" echo echo "---{ input plugins }---" echo " Klog functionality enabled: $enable_klog ($os_type)" +echo " plain tcp input module enabled: $enable_imptcp" echo " imdiag enabled: $enable_imdiag" echo " file input module enabled: $enable_imfile" echo " input template module will be compiled: $enable_imtemplate" diff --git a/doc/Makefile.am b/doc/Makefile.am index ca2ee71c..7ba86ad4 100644 --- a/doc/Makefile.am +++ b/doc/Makefile.am @@ -37,6 +37,7 @@ html_files = \ omlibdbi.html \ imfile.html \ imtcp.html \ + imptcp.html \ imgssapi.html \ imrelp.html \ imuxsock.html \ diff --git a/doc/imptcp.html b/doc/imptcp.html new file mode 100644 index 00000000..913563a5 --- /dev/null +++ b/doc/imptcp.html @@ -0,0 +1,84 @@ + + + +Plain TCP Syslog Input Module (imptcp) + +back + +

Plain TCP Syslog Input Module

+

Module Name:    imptcp

+

Available since: 4.7.3+, 5.5.8+? +

Author: Rainer Gerhards +<rgerhards@adiscon.com>

+

Description:

+

Provides the ability to receive syslog messages via plain TCP syslog. +This is a specialised input plugin tailored for high performance on Linux. It will +probably not run on any other platform. Also, it does no provide TLS services. +Encryption can be provided by using stunnel. +

This module has no limit on the number of listeners and sessions that can be used. +

Multiple receivers may be configured by +specifying $InputPTCPServerRun multiple times. +

+

Configuration Directives:

+

This plugin has config directives similar named as imtcp, but they all have PTCP in +their name instead of just TCP. Note that only a subset of the parameters are supported. +

+Caveats/Known Bugs: + +

Sample:

+

This sets up a TCP server on port 514:
+

+ +

[rsyslog.conf overview] +[manual index] [rsyslog site]

+

This documentation is part of the +rsyslog +project.
+Copyright © 2010 by Rainer +Gerhards and +Adiscon. +Released under the GNU GPL version 3 or higher.

+ diff --git a/doc/rsyslog_conf_modules.html b/doc/rsyslog_conf_modules.html index 675b8bb3..19f69da6 100644 --- a/doc/rsyslog_conf_modules.html +++ b/doc/rsyslog_conf_modules.html @@ -27,7 +27,8 @@ to message generators.
  • imfile -  input module for text files
  • imrelp - RELP input module
  • imudp - udp syslog message input
    • -
  • imtcp - input plugin for plain tcp syslog
    • +
  • imtcp - input plugin for tcp syslog
    • +
  • imptcp - input plugin for plain tcp syslog (no TLS but faster)
  • imgssapi - input plugin for plain tcp and GSS-enabled syslog
  • immark - support for mark messages
  • imklog - kernel logging
    • diff --git a/plugins/imptcp/Makefile.am b/plugins/imptcp/Makefile.am new file mode 100644 index 00000000..bfacc884 --- /dev/null +++ b/plugins/imptcp/Makefile.am @@ -0,0 +1,6 @@ +pkglib_LTLIBRARIES = imptcp.la + +imptcp_la_SOURCES = imptcp.c +imptcp_la_CPPFLAGS = -I$(top_srcdir) $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) +imptcp_la_LDFLAGS = -module -avoid-version +imptcp_la_LIBADD = diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c new file mode 100644 index 00000000..5aeb0192 --- /dev/null +++ b/plugins/imptcp/imptcp.c @@ -0,0 +1,1146 @@ +/* imptcp.c + * This is a native implementation of plain tcp. It is intentionally + * duplicate work (imtcp). The intent is to gain very fast and simple + * native ptcp support, utilizing the best interfaces Linux (no cross- + * platform intended!) has to offer. + * + * Note that in this module we try out some new naming conventions, + * so it may look a bit "different" from the other modules. We are + * investigating if removing prefixes can help make code more readable. + * + * File begun on 2010-08-10 by RGerhards + * + * Copyright 2007-2010 Rainer Gerhards and 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 . + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + */ +#include "config.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#if HAVE_FCNTL_H +#include +#endif +#include "rsyslog.h" +#include "cfsysline.h" +#include "prop.h" +#include "dirty.h" +#include "module-template.h" +#include "unicode-helper.h" +#include "glbl.h" +#include "prop.h" +#include "errmsg.h" +#include "srUtils.h" +#include "datetime.h" +#include "ruleset.h" +#include "msg.h" +#include "net.h" /* for permittedPeers, may be removed when this is removed */ +//#include "tcpsrv.h" /* NOTE: we use some defines from this module -- TODO: re-think! */ + +MODULE_TYPE_INPUT + +/* static data */ +DEF_IMOD_STATIC_DATA +DEFobjCurrIf(glbl) +DEFobjCurrIf(net) +DEFobjCurrIf(prop) +DEFobjCurrIf(datetime) +DEFobjCurrIf(errmsg) +DEFobjCurrIf(ruleset) + + + +/* config settings */ +typedef struct configSettings_s { + int bEmitMsgOnClose; /* emit an informational message on close by remote peer */ + 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? */ + ruleset_t *pRuleset; /* ruleset to bind listener to (use system default if unspecified) */ +} configSettings_t; + +static configSettings_t cs; + +/* data elements describing our running config */ +typedef struct ptcpsrv_s ptcpsrv_t; +typedef struct ptcplstn_s ptcplstn_t; +typedef struct ptcpsess_s ptcpsess_t; +typedef struct epolld_s epolld_t; + +/* the ptcp server (listener) object + * Note that the object contains support for forming a linked list + * of them. It does not make sense to do this seperately. + */ +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; + 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 */ +}; + +/* the ptcp session object. Describes a single active session. + * includes support for doubly-linked list. + */ +struct ptcpsess_s { + ptcpsrv_t *pSrv; /* our server */ + 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? */ + enum { + eAtStrtFram, + eInOctetCnt, + eInMsg + } inputState; /* our current state */ + int iOctetsRemain; /* Number of Octets remaining in message */ + TCPFRAMINGMODE eFraming; + uchar *pMsg; /* message (fragment) received */ + prop_t *peerName; /* host name we received messages from */ + prop_t *peerIP; +//--- END from tcps_sess.h +}; + + +/* the ptcp listener object. Describes a single active listener. + */ +struct ptcplstn_s { + ptcpsrv_t *pSrv; /* our server */ + ptcplstn_t *prev, *next; + int sock; + epolld_t *epd; +}; + + +/* type of object stored in epoll descriptor */ +typedef enum { + epolld_lstn, + epolld_sess +} epolld_type_t; + +/* an epoll descriptor. contains all information necessary to process + * the result of epoll. + */ +struct epolld_s { + epolld_type_t typ; + void *ptr; + struct epoll_event ev; +}; + + +/* global data */ +//static permittedPeers_t *pPermPeersRoot = NULL; +static ptcpsrv_t *pSrvRoot = NULL; +static int epollfd = -1; /* (sole) descriptor for epoll */ +static int iMaxLine; /* maximum size of a single message */ + +/* forward definitions */ +static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal); +static rsRetVal addLstn(ptcpsrv_t *pSrv, int sock); + + +/* some simple constructors/destructors */ +static void +destructSess(ptcpsess_t *pSess) +{ + free(pSess->pMsg); + free(pSess->epd); + prop.Destruct(&pSess->peerName); + prop.Destruct(&pSess->peerIP); + /* TODO: make these inits compile-time switch depending: */ + pSess->pMsg = NULL; + pSess->epd = NULL; + free(pSess); +} + +static void +destructSrv(ptcpsrv_t *pSrv) +{ + prop.Destruct(&pSrv->pInputName); + free(pSrv->port); + free(pSrv); +} + +/****************************************** TCP SUPPORT FUNCTIONS ***********************************/ +/* We may later think about moving this into a helper library again. But the whole point + * so far was to keep everything related close togehter. -- rgerhards, 2010-08-10 + */ + + +/* Start up a server. That means all of its listeners are created. + * Does NOT yet accept/process any incoming data (but binds ports). Hint: this + * code is to be executed before dropping privileges. + */ +static rsRetVal +startupSrv(ptcpsrv_t *pSrv) +{ + DEFiRet; + int iSessMax = 200; /* TODO: Make configurable or remove? */ + int error, maxs, on = 1; + int sock = -1; + int numSocks; + int sockflags; + struct addrinfo hints, *res = NULL, *r; + uchar *lstnIP; + + lstnIP = pSrv->lstnIP == NULL ? UCHAR_CONSTANT("") : pSrv->lstnIP; + + dbgprintf("imptcp creating listen socket on server '%s', port %s\n", lstnIP, pSrv->port); + + memset(&hints, 0, sizeof(hints)); + hints.ai_flags = AI_PASSIVE; + hints.ai_family = glbl.GetDefPFFamily(); + hints.ai_socktype = SOCK_STREAM; + + error = getaddrinfo((char*)pSrv->lstnIP, (char*) pSrv->port, &hints, &res); + if(error) { + dbgprintf("error %d querying server '%s', port '%s'\n", error, pSrv->lstnIP, pSrv->port); + ABORT_FINALIZE(RS_RET_INVALID_PORT); + } + + /* Count max number of sockets we may open */ + for(maxs = 0, r = res; r != NULL ; r = r->ai_next, maxs++) + /* EMPTY */; + + numSocks = 0; /* num of sockets counter at start of array */ + for(r = res; r != NULL ; r = r->ai_next) { + sock = socket(r->ai_family, r->ai_socktype, r->ai_protocol); + if(sock < 0) { + if(!(r->ai_family == PF_INET6 && errno == EAFNOSUPPORT)) + dbgprintf("error %d creating tcp listen socket", errno); + /* it is debatable if PF_INET with EAFNOSUPPORT should + * also be ignored... + */ + continue; + } + +#ifdef IPV6_V6ONLY + if(r->ai_family == AF_INET6) { + int iOn = 1; + if(setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, + (char *)&iOn, sizeof (iOn)) < 0) { + close(sock); + sock = -1; + continue; + } + } +#endif + if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) { + dbgprintf("error %d setting tcp socket option\n", errno); + close(sock); + sock = -1; + continue; + } + + /* We use non-blocking IO! */ + if((sockflags = fcntl(sock, F_GETFL)) != -1) { + sockflags |= O_NONBLOCK; + /* SETFL could fail too, so get it caught by the subsequent + * error check. + */ + sockflags = fcntl(sock, F_SETFL, sockflags); + } + if(sockflags == -1) { + dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket", errno); + close(sock); + sock = -1; + continue; + } + + + + /* We need to enable BSD compatibility. Otherwise an attacker + * could flood our log files by sending us tons of ICMP errors. + */ +#ifndef BSD + if(net.should_use_so_bsdcompat()) { + if (setsockopt(sock, SOL_SOCKET, SO_BSDCOMPAT, + (char *) &on, sizeof(on)) < 0) { + errmsg.LogError(errno, NO_ERRCODE, "TCP setsockopt(BSDCOMPAT)"); + close(sock); + sock = -1; + continue; + } + } +#endif + + if( (bind(sock, r->ai_addr, r->ai_addrlen) < 0) +#ifndef IPV6_V6ONLY + && (errno != EADDRINUSE) +#endif + ) { + /* TODO: check if *we* bound the socket - else we *have* an error! */ + dbgprintf("error %d while binding tcp socket", errno); + close(sock); + sock = -1; + continue; + } + + if(listen(sock, iSessMax / 10 + 5) < 0) { + /* If the listen fails, it most probably fails because we ask + * for a too-large backlog. So in this case we first set back + * to a fixed, reasonable, limit that should work. Only if + * that fails, too, we give up. + */ + dbgprintf("listen with a backlog of %d failed - retrying with default of 32.", + iSessMax / 10 + 5); + if(listen(sock, 32) < 0) { + dbgprintf("tcp listen error %d, suspending\n", errno); + close(sock); + sock = -1; + continue; + } + } + + /* 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)); + ++numSocks; + } + + if(numSocks != maxs) + dbgprintf("We could initialize %d TCP listen sockets out of %d we received " + "- this may or may not be an error indication.\n", numSocks, maxs); + + if(numSocks == 0) { + dbgprintf("No TCP listen sockets could successfully be initialized"); + ABORT_FINALIZE(RS_RET_COULD_NOT_BIND); + } + +finalize_it: + if(res != NULL) + freeaddrinfo(res); + + if(iRet != RS_RET_OK) { + if(sock != -1) + close(sock); + } + + RETiRet; +} + + +/* Set pRemHost based on the address provided. This is to be called upon accept()ing + * a connection request. It must be provided by the socket we received the + * message on as well as a NI_MAXHOST size large character buffer for the FQDN. + * Please see http://www.hmug.org/man/3/getnameinfo.php (under Caveats) + * for some explanation of the code found below. If we detect a malicious + * hostname, we return RS_RET_MALICIOUS_HNAME and let the caller decide + * on how to deal with that. + * rgerhards, 2008-03-31 + */ +static rsRetVal +getPeerNames(prop_t **peerName, prop_t **peerIP, struct sockaddr *pAddr) +{ + int error; + uchar szIP[NI_MAXHOST] = ""; + uchar szHname[NI_MAXHOST] = ""; + struct addrinfo hints, *res; + + DEFiRet; + + error = getnameinfo(pAddr, SALEN(pAddr), (char*)szIP, sizeof(szIP), NULL, 0, NI_NUMERICHOST); + + if(error) { + dbgprintf("Malformed from address %s\n", gai_strerror(error)); + strcpy((char*)szHname, "???"); + strcpy((char*)szIP, "???"); + ABORT_FINALIZE(RS_RET_INVALID_HNAME); + } + + if(!glbl.GetDisableDNS()) { + error = getnameinfo(pAddr, SALEN(pAddr), (char*)szHname, NI_MAXHOST, NULL, 0, NI_NAMEREQD); + if(error == 0) { + memset (&hints, 0, sizeof (struct addrinfo)); + hints.ai_flags = AI_NUMERICHOST; + hints.ai_socktype = SOCK_STREAM; + /* we now do a lookup once again. This one should fail, + * because we should not have obtained a non-numeric address. If + * we got a numeric one, someone messed with DNS! + */ + if(getaddrinfo((char*)szHname, NULL, &hints, &res) == 0) { + freeaddrinfo (res); + /* OK, we know we have evil, so let's indicate this to our caller */ + snprintf((char*)szHname, NI_MAXHOST, "[MALICIOUS:IP=%s]", szIP); + dbgprintf("Malicious PTR record, IP = \"%s\" HOST = \"%s\"", szIP, szHname); + iRet = RS_RET_MALICIOUS_HNAME; + } + } else { + strcpy((char*)szHname, (char*)szIP); + } + } else { + strcpy((char*)szHname, (char*)szIP); + } + + /* We now have the names, so now let's allocate memory and store them permanently. */ + CHKiRet(prop.Construct(peerName)); + CHKiRet(prop.SetString(*peerName, szHname, ustrlen(szHname))); + CHKiRet(prop.ConstructFinalize(*peerName)); + CHKiRet(prop.Construct(peerIP)); + CHKiRet(prop.SetString(*peerIP, szIP, ustrlen(szIP))); + CHKiRet(prop.ConstructFinalize(*peerIP)); + +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) +{ + int sockflags; + struct sockaddr_storage addr; + socklen_t addrlen = sizeof(addr); + int iNewSock = -1; + + DEFiRet; + + iNewSock = accept(sock, (struct sockaddr*) &addr, &addrlen); + if(iNewSock < 0) { + ABORT_FINALIZE(RS_RET_ACCEPT_ERR); + } + + CHKiRet(getPeerNames(peerName, peerIP, (struct sockaddr*) &addr)); + + /* set the new socket to non-blocking IO */ + if((sockflags = fcntl(iNewSock, F_GETFL)) != -1) { + sockflags |= O_NONBLOCK; + /* SETFL could fail too, so get it caught by the subsequent + * error check. + */ + sockflags = fcntl(iNewSock, F_SETFL, sockflags); + } + if(sockflags == -1) { + dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket %d", errno, iNewSock); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } + + *newSock = iNewSock; + +finalize_it: + if(iRet != RS_RET_OK) { + /* the close may be redundant, but that doesn't hurt... */ + close(iNewSock); + } + + RETiRet; +} + + +/* This is a helper for submitting the message to the rsyslog core. + * It does some common processing, including resetting the various + * state variables to a "processed" state. + * Note that this function is also called if we had a buffer overflow + * due to a too-long message. So far, there is no indication this + * happened and it may be worth thinking about different handling + * of this case (what obviously would require a change to this + * function or some related code). + * rgerhards, 2009-04-23 + * EXTRACT from tcps_sess.c + */ +static rsRetVal +doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub) +{ + msg_t *pMsg; + DEFiRet; + + if(pThis->iMsg == 0) { + DBGPRINTF("discarding zero-sized message\n"); + FINALIZE; + } + + /* 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); + MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY); + pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME; + pMsg->bParseHOSTNAME = 1; + MsgSetRcvFrom(pMsg, pThis->peerName); + CHKiRet(MsgSetRcvFromIP(pMsg, pThis->peerIP)); + MsgSetRuleset(pMsg, pThis->pSrv->pRuleset); + + if(pMultiSub == NULL) { + CHKiRet(submitMsg(pMsg)); + } else { + pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg; + if(pMultiSub->nElem == pMultiSub->maxElem) + CHKiRet(multiSubmitMsg(pMultiSub)); + } + + +finalize_it: + /* reset status variables */ + pThis->bAtStrtOfFram = 1; + pThis->iMsg = 0; + + RETiRet; +} + + +/* process the data received. As TCP is stream based, we need to process the + * data inside a state machine. The actual data received is passed in byte-by-byte + * from DataRcvd, and this function here compiles messages from them and submits + * the end result to the queue. Introducing this function fixes a long-term bug ;) + * rgerhards, 2008-03-14 + * EXTRACT from tcps_sess.c + */ +static inline rsRetVal +processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub) +{ + DEFiRet; + + if(pThis->inputState == eAtStrtFram) { + if(isdigit((int) c)) { + pThis->inputState = eInOctetCnt; + pThis->iOctetsRemain = 0; + pThis->eFraming = TCP_FRAMING_OCTET_COUNTING; + } else { + pThis->inputState = eInMsg; + pThis->eFraming = TCP_FRAMING_OCTET_STUFFING; + } + } + + if(pThis->inputState == eInOctetCnt) { + if(isdigit(c)) { + pThis->iOctetsRemain = pThis->iOctetsRemain * 10 + c - '0'; + } else { /* done with the octet count, so this must be the SP terminator */ + dbgprintf("TCP Message with octet-counter, size %d.\n", pThis->iOctetsRemain); + if(c != ' ') { + errmsg.LogError(0, NO_ERRCODE, "Framing Error in received TCP message: " + "delimiter is not SP but has ASCII value %d.\n", c); + } + if(pThis->iOctetsRemain < 1) { + /* TODO: handle the case where the octet count is 0! */ + dbgprintf("Framing Error: invalid octet count\n"); + errmsg.LogError(0, NO_ERRCODE, "Framing Error in received TCP message: " + "invalid octet count %d.\n", pThis->iOctetsRemain); + } else if(pThis->iOctetsRemain > iMaxLine) { + /* while we can not do anything against it, we can at least log an indication + * that something went wrong) -- rgerhards, 2008-03-14 + */ + dbgprintf("truncating message with %d octets - max msg size is %d\n", + pThis->iOctetsRemain, iMaxLine); + errmsg.LogError(0, NO_ERRCODE, "received oversize message: size is %d bytes, " + "max msg size is %d, truncating...\n", pThis->iOctetsRemain, iMaxLine); + } + pThis->inputState = eInMsg; + } + } else { + assert(pThis->inputState == eInMsg); + if(pThis->iMsg >= iMaxLine) { + /* emergency, we now need to flush, no matter if we are at end of message or not... */ + dbgprintf("error: message received is larger than max msg size, we split it\n"); + doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub); + /* we might think if it is better to ignore the rest of the + * message than to treat it as a new one. Maybe this is a good + * candidate for a configuration parameter... + * rgerhards, 2006-12-04 + */ + } + + if(( (c == '\n') + //|| ((pThis->pSrv->addtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->addtlFrameDelim)) + ) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */ + doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub); + pThis->inputState = eAtStrtFram; + } else { + /* IMPORTANT: here we copy the actual frame content to the message - for BOTH framing modes! + * If we have a message that is larger than the max msg size, we truncate it. This is the best + * we can do in light of what the engine supports. -- rgerhards, 2008-03-14 + */ + if(pThis->iMsg < iMaxLine) { + *(pThis->pMsg + pThis->iMsg++) = c; + } + } + + if(pThis->eFraming == TCP_FRAMING_OCTET_COUNTING) { + /* do we need to find end-of-frame via octet counting? */ + pThis->iOctetsRemain--; + if(pThis->iOctetsRemain < 1) { + /* we have end of frame! */ + doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub); + pThis->inputState = eAtStrtFram; + } + } + } + + RETiRet; +} + + +/* Processes the data received via a TCP session. If there + * is no other way to handle it, data is discarded. + * Input parameter data is the data received, iLen is its + * len as returned from recv(). iLen must be 1 or more (that + * is errors must be handled by caller!). iTCPSess must be + * the index of the TCP session that received the data. + * rgerhards 2005-07-04 + * And another change while generalizing. We now return either + * RS_RET_OK, which means the session should be kept open + * or anything else, which means it must be closed. + * rgerhards, 2008-03-01 + * As a performance optimization, we pick up the timestamp here. Acutally, + * this *is* the *correct* reception step for all the data we received, because + * we have just received a bunch of data! -- rgerhards, 2009-06-16 + * EXTRACT from tcps_sess.c + */ +#define NUM_MULTISUB 1024 +static rsRetVal +DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) +{ + multi_submit_t multiSub; + msg_t *pMsgs[NUM_MULTISUB]; + struct syslogTime stTime; + time_t ttGenTime; + char *pEnd; + DEFiRet; + + assert(pData != NULL); + assert(iLen > 0); + + datetime.getCurrTime(&stTime, &ttGenTime); + multiSub.ppMsgs = pMsgs; + multiSub.maxElem = NUM_MULTISUB; + multiSub.nElem = 0; + + /* We now copy the message to the session buffer. */ + pEnd = pData + iLen; /* this is one off, which is intensional */ + + while(pData < pEnd) { + CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub)); + } + + if(multiSub.nElem > 0) { + /* submit anything that was not yet submitted */ + CHKiRet(multiSubmitMsg(&multiSub)); + } + +finalize_it: + RETiRet; +} +#undef NUM_MULTISUB + + +/****************************************** --END-- TCP SUPPORT FUNCTIONS ***********************************/ + + +static inline void +initConfigSettings(void) +{ + cs.bEmitMsgOnClose = 0; + //cs.iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER; + cs.pszInputName = NULL; + cs.pRuleset = NULL; + cs.lstnIP = NULL; +} + + +/* add socket to the epoll set + */ +static inline rsRetVal +addEPollSock(epolld_type_t typ, void *ptr, int sock, epolld_t **pEpd) +{ + DEFiRet; + epolld_t *epd = NULL; + + CHKmalloc(epd = malloc(sizeof(epolld_t))); + epd->typ = typ; + epd->ptr = ptr; + *pEpd = epd; + epd->ev.events = EPOLLIN|EPOLLET; + epd->ev.data.ptr = (void*) epd; + + if(epoll_ctl(epollfd, EPOLL_CTL_ADD, sock, &(epd->ev)) != 0) { + char errStr[1024]; + int eno = errno; + errmsg.LogError(0, RS_RET_EPOLL_CTL_FAILED, "os error (%d) during epoll ADD: %s", + eno, rs_strerror_r(eno, errStr, sizeof(errStr))); + ABORT_FINALIZE(RS_RET_EPOLL_CTL_FAILED); + } + + DBGPRINTF("imptcp: added socket %d to epoll[%d] set\n", sock, epollfd); + +finalize_it: + if(iRet != RS_RET_OK) { + free(epd); + } + RETiRet; +} + + +/* remove a socket from the epoll set. Note that the epd parameter + * is not really required -- it is used to satisfy older kernels where + * epoll_ctl() required a non-NULL pointer even though the ptr is never used. + * For simplicity, we supply the same pointer we had when we created the + * event (it's simple because we have it at hand). + */ +static inline rsRetVal +removeEPollSock(int sock, epolld_t *epd) +{ + DEFiRet; + + DBGPRINTF("imptcp: removing socket %d from epoll[%d] set\n", sock, epollfd); + + if(epoll_ctl(epollfd, EPOLL_CTL_DEL, sock, &(epd->ev)) != 0) { + char errStr[1024]; + int eno = errno; + errmsg.LogError(0, RS_RET_EPOLL_CTL_FAILED, "os error (%d) during epoll DEL: %s", + eno, rs_strerror_r(eno, errStr, sizeof(errStr))); + ABORT_FINALIZE(RS_RET_EPOLL_CTL_FAILED); + } + +finalize_it: + RETiRet; +} + + +/* add a listener to the server + */ +static rsRetVal +addLstn(ptcpsrv_t *pSrv, int sock) +{ + DEFiRet; + ptcplstn_t *pLstn; + + CHKmalloc(pLstn = malloc(sizeof(ptcplstn_t))); + pLstn->pSrv = pSrv; + pLstn->sock = sock; + + /* add to start of server's listener list */ + pLstn->prev = NULL; + pLstn->next = pSrv->pLstn; + pSrv->pLstn = pLstn; + + iRet = addEPollSock(epolld_lstn, pLstn, sock, &pLstn->epd); + +finalize_it: + RETiRet; +} + + +/* add a session to the server + */ +static rsRetVal +addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP) +{ + DEFiRet; + ptcpsess_t *pSess = NULL; + + CHKmalloc(pSess = malloc(sizeof(ptcpsess_t))); + CHKmalloc(pSess->pMsg = malloc(iMaxLine * sizeof(uchar))); + pSess->pSrv = pSrv; + pSess->sock = sock; + pSess->inputState = eAtStrtFram; + pSess->iMsg = 0; + pSess->bAtStrtOfFram = 1; + pSess->peerName = peerName; + pSess->peerIP = peerIP; + + /* add to start of server's listener list */ + pSess->prev = NULL; + pSess->next = pSrv->pSess; + pSrv->pSess = pSess; + + iRet = addEPollSock(epolld_sess, pSess, sock, &pSess->epd); + +finalize_it: + RETiRet; +} + + +/* close/remove a session + * NOTE: we must first remove the fd from the epoll set and then close it -- else we + * get an error "bad file descriptor" from epoll. + */ +static rsRetVal +closeSess(ptcpsess_t *pSess) +{ + int sock; + DEFiRet; + + sock = pSess->sock; + CHKiRet(removeEPollSock(sock, pSess->epd)); + close(sock); + + /* finally unlink session from structures */ + if(pSess->next != NULL) + pSess->next->prev = pSess->prev; + if(pSess->prev == NULL) { + /* need to update root! */ + pSess->pSrv->pSess = pSess->next; + } else { + pSess->prev->next = pSess->next; + } + + /* unlinked, now remove structure */ + destructSess(pSess); + +finalize_it: + DBGPRINTF("imtcp: session on socket %d closed with iRet %d.\n", sock, iRet); + RETiRet; +} + + +#if 0 +/* set permitted peer -- rgerhards, 2008-05-19 + */ +static rsRetVal +setPermittedPeer(void __attribute__((unused)) *pVal, uchar *pszID) +{ + DEFiRet; + CHKiRet(net.AddPermittedPeer(&pPermPeersRoot, pszID)); + free(pszID); /* no longer needed, but we need to free as of interface def */ +finalize_it: + RETiRet; +} +#endif + + +/* accept a new ruleset to bind. Checks if it exists and complains, if not */ +static rsRetVal setRuleset(void __attribute__((unused)) *pVal, uchar *pszName) +{ + ruleset_t *pRuleset; + rsRetVal localRet; + DEFiRet; + + localRet = ruleset.GetRuleset(&pRuleset, pszName); + if(localRet == RS_RET_NOT_FOUND) { + errmsg.LogError(0, NO_ERRCODE, "error: ruleset '%s' not found - ignored", pszName); + } + CHKiRet(localRet); + cs.pRuleset = pRuleset; + DBGPRINTF("imptcp current bind ruleset %p: '%s'\n", pRuleset, pszName); + +finalize_it: + free(pszName); /* no longer needed */ + RETiRet; +} + + +static rsRetVal addTCPListener(void __attribute__((unused)) *pVal, uchar *pNewVal) +{ + DEFiRet; + ptcpsrv_t *pSrv; + + CHKmalloc(pSrv = malloc(sizeof(ptcpsrv_t))); + pSrv->pSess = NULL; + pSrv->pLstn = NULL; + pSrv->bEmitMsgOnClose = cs.bEmitMsgOnClose; + pSrv->port = pNewVal; + //pSrv->iAddtlFrameDelim = cs.iAddtlFrameDelim; + cs.pszInputName = NULL; /* moved over to pSrv, we do not own */ + pSrv->lstnIP = cs.lstnIP; + cs.lstnIP = NULL; /* moved over to pSrv, we do not own */ + pSrv->pRuleset = cs.pRuleset; + pSrv->pszInputName = (cs.pszInputName == NULL) ? UCHAR_CONSTANT("imptcp") : cs.pszInputName; + CHKiRet(prop.Construct(&pSrv->pInputName)); + CHKiRet(prop.SetString(pSrv->pInputName, pSrv->pszInputName, ustrlen(pSrv->pszInputName))); + CHKiRet(prop.ConstructFinalize(pSrv->pInputName)); + + /* add to linked list */ + pSrv->pNext = pSrvRoot; + pSrvRoot = pSrv; + + /* all config vars are auto-reset -- this also is very useful with the + * new config format effort (v6). + */ + resetConfigVariables(NULL, NULL); + +finalize_it: + if(iRet != RS_RET_OK) { + errmsg.LogError(0, NO_ERRCODE, "error %d trying to add listener", iRet); + } + RETiRet; +} + + +/* start up all listeners + * This is a one-time stop once the module is set to start. + */ +static inline rsRetVal +startupServers() +{ + DEFiRet; + ptcpsrv_t *pSrv; + + pSrv = pSrvRoot; + while(pSrv != NULL) { + DBGPRINTF("Starting up ptcp server for port %s, name '%s'\n", pSrv->port, pSrv->pszInputName); + startupSrv(pSrv); + pSrv = pSrv->pNext; + } + + RETiRet; +} + + +/* process new activity on listener. This means we need to accept a new + * connection. + */ +static inline rsRetVal +lstnActivity(ptcplstn_t *pLstn) +{ + int newSock; + prop_t *peerName; + prop_t *peerIP; + DEFiRet; + + DBGPRINTF("imptcp: new connection on listen socket %d\n", pLstn->sock); + CHKiRet(AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP)); + CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP)); + +finalize_it: + RETiRet; +} + + +/* process new activity on session. This means we need to accept data + * or close the session. + */ +static inline rsRetVal +sessActivity(ptcpsess_t *pSess) +{ + int lenRcv; + int lenBuf; + char rcvBuf[4096]; + DEFiRet; + + DBGPRINTF("imptcp: new activity on session socket %d\n", pSess->sock); + + lenBuf = sizeof(rcvBuf); + lenRcv = recv(pSess->sock, rcvBuf, lenBuf, 0); + + if(lenRcv > 0) { + /* have data, process it */ + DBGPRINTF("imtcp: data(%d) on socket %d: %s\n", lenBuf, pSess->sock, rcvBuf); + CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv)); + } else if (lenRcv == 0) { + /* session was closed, do clean-up */ + CHKiRet(closeSess(pSess)); + } else { + DBGPRINTF("imtcp: error on session socket %d - closed.\n", pSess->sock); + closeSess(pSess); /* try clean-up by dropping session */ + } + +finalize_it: + RETiRet; +} + + +/* This function is called to gather input. + */ +BEGINrunInput + int i; + int nfds; + struct epoll_event events[1]; + epolld_t *epd; +CODESTARTrunInput + DBGPRINTF("imptcp now beginning to process input data\n"); + /* v5 TODO: consentual termination mode */ + while(1) { + DBGPRINTF("imptcp going on epoll_wait\n"); + nfds = epoll_wait(epollfd, events, sizeof(events)/sizeof(struct epoll_event), -1); + for(i = 0 ; i < nfds ; ++i) { /* support for larger batches (later, TODO) */ + epd = (epolld_t*) events[i].data.ptr; + switch(epd->typ) { + case epolld_lstn: + lstnActivity((ptcplstn_t *) epd->ptr); + break; + case epolld_sess: + sessActivity((ptcpsess_t *) epd->ptr); + break; + default: + errmsg.LogError(0, RS_RET_INTERNAL_ERROR, + "error: invalid epolld_type_t %d after epoll", epd->typ); + break; + } + } + } +ENDrunInput + + +/* initialize and return if will run or not */ +BEGINwillRun +CODESTARTwillRun + /* first apply some config settings */ + //net.PrintAllowedSenders(2); /* TCP */ + iMaxLine = glbl.GetMaxLine(); /* get maximum size we currently support */ + + if(pSrvRoot == NULL) { + errmsg.LogError(0, RS_RET_NO_LSTN_DEFINED, "error: no ptcp server defined, module can not run."); + ABORT_FINALIZE(RS_RET_NO_RUN); + } + + if((epollfd = epoll_create1(EPOLL_CLOEXEC)) < 0) { + errmsg.LogError(0, RS_RET_EPOLL_CR_FAILED, "error: epoll_create() failed"); + ABORT_FINALIZE(RS_RET_NO_RUN); + } + + /* start up servers, but do not yet read input data */ + CHKiRet(startupServers()); + DBGPRINTF("imptcp started up, but not yet receiving data\n"); +finalize_it: +ENDwillRun + + +/* completely shut down a server, that means closing all of its + * listeners and sessions. + */ +static inline void +shutdownSrv(ptcpsrv_t *pSrv) +{ + ptcplstn_t *pLstn, *lstnDel; + ptcpsess_t *pSess, *sessDel; + + /* listeners */ + pLstn = pSrv->pLstn; + while(pLstn != NULL) { + close(pLstn->sock); + lstnDel = pLstn; + pLstn = pLstn->next; + DBGPRINTF("imptcp shutdown listen socket %d\n", lstnDel->sock); + free(lstnDel->epd); + free(lstnDel); + } + + /* sessions */ + pSess = pSrv->pSess; + while(pSess != NULL) { + close(pSess->sock); + sessDel = pSess; + pSess = pSess->next; + DBGPRINTF("imptcp shutdown session socket %d\n", sessDel->sock); + destructSess(sessDel); + } +} + + +BEGINafterRun + ptcpsrv_t *pSrv, *srvDel; +CODESTARTafterRun + /* do cleanup here */ + //net.clearAllowedSenders(UCHAR_CONSTANT("TCP")); + /* we need to close everything that is still open */ + pSrv = pSrvRoot; + while(pSrv != NULL) { + srvDel = pSrv; + pSrv = pSrv->pNext; + shutdownSrv(srvDel); + destructSrv(srvDel); + } + + close(epollfd); +ENDafterRun + + +BEGINmodExit +CODESTARTmodExit +#if 0 + if(pPermPeersRoot != NULL) { + net.DestructPermittedPeers(&pPermPeersRoot); + } +#endif + + /* release objects we used */ + objRelease(glbl, CORE_COMPONENT); + objRelease(prop, CORE_COMPONENT); + objRelease(net, LM_NET_FILENAME); + objRelease(datetime, CORE_COMPONENT); + objRelease(errmsg, CORE_COMPONENT); + objRelease(ruleset, CORE_COMPONENT); +ENDmodExit + + +static rsRetVal +resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal) +{ + cs.bEmitMsgOnClose = 0; + //cs.iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER; + free(cs.pszInputName); + cs.pszInputName = NULL; + free(cs.lstnIP); + cs.lstnIP = NULL; + return RS_RET_OK; +} + + + +BEGINqueryEtryPt +CODESTARTqueryEtryPt +CODEqueryEtryPt_STD_IMOD_QUERIES +ENDqueryEtryPt + + +BEGINmodInit() +CODESTARTmodInit + *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ +CODEmodInit_QueryRegCFSLineHdlr + initConfigSettings(); + /* request objects we use */ + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(prop, CORE_COMPONENT)); + CHKiRet(objUse(net, LM_NET_FILENAME)); + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(datetime, CORE_COMPONENT)); + CHKiRet(objUse(ruleset, CORE_COMPONENT)); + + /* register config file handlers */ + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverrun"), 0, eCmdHdlrGetWord, + addTCPListener, NULL, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpservernotifyonconnectionclose"), 0, + eCmdHdlrBinary, NULL, &cs.bEmitMsgOnClose, STD_LOADABLE_MODULE_ID)); + //CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverstreamdriverpermittedpeer"), 0, + //eCmdHdlrGetWord, setPermittedPeer, NULL, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserveraddtlframedelimiter"), 0, eCmdHdlrInt, + NULL, &cs.iAddtlFrameDelim, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverinputname"), 0, + eCmdHdlrGetWord, NULL, &cs.pszInputName, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverlistenip"), 0, + eCmdHdlrGetWord, NULL, &cs.lstnIP, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverbindruleset"), 0, + eCmdHdlrGetWord, setRuleset, NULL, STD_LOADABLE_MODULE_ID)); + CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("resetconfigvariables"), 1, eCmdHdlrCustomHandler, + resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID)); +ENDmodInit + + +/* vim:set ai: + */ diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index 8979893a..907b9c1a 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -360,6 +360,10 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_VAR_NOT_FOUND = -2142, /**< variable not found */ RS_RET_EMPTY_MSG = -2143, /**< provided (raw) MSG is empty */ RS_RET_PEER_CLOSED_CONN = -2144, /**< remote peer closed connection (information, no error) */ + RS_RET_NO_LSTN_DEFINED = -2172, /**< no listener defined (e.g. inside an input module) */ + RS_RET_EPOLL_CR_FAILED = -2173, /**< epoll_create() failed */ + RS_RET_EPOLL_CTL_FAILED = -2174, /**< epoll_ctl() failed */ + RS_RET_INTERNAL_ERROR = -2175, /**< rsyslogd internal error, unexpected code path reached */ /* RainerScript error messages (range 1000.. 1999) */ RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */ diff --git a/tools/syslogd.c b/tools/syslogd.c index a03dcf0e..18e0c98f 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -371,7 +371,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a /* hardcoded standard templates (used for defaults) */ -static uchar template_DebugFormat[] = "\"Debug line with all properties:\nFROMHOST: '%FROMHOST%', fromhost-ip: '%fromhost-ip%', HOSTNAME: '%HOSTNAME%', PRI: %PRI%,\nsyslogtag '%syslogtag%', programname: '%programname%', APP-NAME: '%APP-NAME%', PROCID: '%PROCID%', MSGID: '%MSGID%',\nTIMESTAMP: '%TIMESTAMP%', STRUCTURED-DATA: '%STRUCTURED-DATA%',\nmsg: '%msg%'\nescaped msg: '%msg:::drop-cc%'\nrawmsg: '%rawmsg%'\n\n\""; +static uchar template_DebugFormat[] = "\"Debug line with all properties:\nFROMHOST: '%FROMHOST%', fromhost-ip: '%fromhost-ip%', HOSTNAME: '%HOSTNAME%', PRI: %PRI%,\nsyslogtag '%syslogtag%', programname: '%programname%', APP-NAME: '%APP-NAME%', PROCID: '%PROCID%', MSGID: '%MSGID%',\nTIMESTAMP: '%TIMESTAMP%', STRUCTURED-DATA: '%STRUCTURED-DATA%',\nmsg: '%msg%'\nescaped msg: '%msg:::drop-cc%'\ninputname: %inputname% rawmsg: '%rawmsg%'\n\n\""; static uchar template_SyslogProtocol23Format[] = "\"<%PRI%>1 %TIMESTAMP:::date-rfc3339% %HOSTNAME% %APP-NAME% %PROCID% %MSGID% %STRUCTURED-DATA% %msg%\n\""; static uchar template_TraditionalFileFormat[] = "\"%TIMESTAMP% %HOSTNAME% %syslogtag%%msg:::sp-if-no-1st-sp%%msg:::drop-last-lf%\n\""; static uchar template_FileFormat[] = "\"%TIMESTAMP:::date-rfc3339% %HOSTNAME% %syslogtag%%msg:::sp-if-no-1st-sp%%msg:::drop-last-lf%\n\""; -- cgit From ee4aed1713bb968afa6db992f9e2e6c00d6c9350 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 16:18:21 +0200 Subject: added tests for imptcp and fixed some problems with it it now also works reliably in edge-triggered mode --- plugins/imptcp/imptcp.c | 59 ++++++++++++++++++++++++++------------ tests/Makefile.am | 6 ++++ tests/diag.sh | 2 +- tests/imptcp_large.sh | 14 +++++++++ tests/testsuites/imptcp_large.conf | 16 +++++++++++ 5 files changed, 78 insertions(+), 19 deletions(-) create mode 100755 tests/imptcp_large.sh create mode 100644 tests/testsuites/imptcp_large.conf diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c index 5aeb0192..80df959c 100644 --- a/plugins/imptcp/imptcp.c +++ b/plugins/imptcp/imptcp.c @@ -434,6 +434,8 @@ AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP) iNewSock = accept(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); } @@ -747,6 +749,8 @@ addLstn(ptcpsrv_t *pSrv, int sock) /* add to start of server's listener list */ pLstn->prev = NULL; pLstn->next = pSrv->pLstn; + if(pSrv->pLstn != NULL) + pSrv->pLstn->prev = pLstn; pSrv->pLstn = pLstn; iRet = addEPollSock(epolld_lstn, pLstn, sock, &pLstn->epd); @@ -777,6 +781,8 @@ addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP) /* add to start of server's listener list */ pSess->prev = NULL; pSess->next = pSrv->pSess; + if(pSrv->pSess != NULL) + pSrv->pSess->prev = pSess; pSrv->pSess = pSess; iRet = addEPollSock(epolld_sess, pSess, sock, &pSess->epd); @@ -801,6 +807,9 @@ closeSess(ptcpsess_t *pSess) close(sock); /* finally unlink session from structures */ +//fprintf(stderr, "closing session %d next %p, prev %p\n", pSess->sock, pSess->next, pSess->prev); +//dbgprintf("imptcp: pSess->next %p\n", pSess->next); +//dbgprintf("imptcp: pSess->prev %p\n", pSess->prev); if(pSess->next != NULL) pSess->next->prev = pSess->prev; if(pSess->prev == NULL) { @@ -921,11 +930,19 @@ lstnActivity(ptcplstn_t *pLstn) int newSock; prop_t *peerName; prop_t *peerIP; + rsRetVal localRet; +int iac = 0; DEFiRet; DBGPRINTF("imptcp: new connection on listen socket %d\n", pLstn->sock); - CHKiRet(AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP)); - CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP)); + while(1) { + localRet = AcceptConnReq(pLstn->sock, &newSock, &peerName, &peerIP); +//if(iac++ > 0) fprintf(stderr, "%d accepts in a row!\n", iac); + if(localRet == RS_RET_NO_MORE_DATA) + break; + CHKiRet(localRet); + CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP)); + } finalize_it: RETiRet; @@ -940,24 +957,32 @@ sessActivity(ptcpsess_t *pSess) { int lenRcv; int lenBuf; - char rcvBuf[4096]; + char rcvBuf[128*1024]; DEFiRet; +int iac = 0; DBGPRINTF("imptcp: new activity on session socket %d\n", pSess->sock); - lenBuf = sizeof(rcvBuf); - lenRcv = recv(pSess->sock, rcvBuf, lenBuf, 0); - - if(lenRcv > 0) { - /* have data, process it */ - DBGPRINTF("imtcp: data(%d) on socket %d: %s\n", lenBuf, pSess->sock, rcvBuf); - CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv)); - } else if (lenRcv == 0) { - /* session was closed, do clean-up */ - CHKiRet(closeSess(pSess)); - } else { - DBGPRINTF("imtcp: error on session socket %d - closed.\n", pSess->sock); - closeSess(pSess); /* try clean-up by dropping session */ + while(1) { + lenBuf = sizeof(rcvBuf); + lenRcv = recv(pSess->sock, rcvBuf, lenBuf, 0); +//if(iac++ > 1) fprintf(stderr, "\n%d recv in a row!\n", iac-1); + + if(lenRcv > 0) { + /* have data, process it */ + DBGPRINTF("imtcp: data(%d) on socket %d: %s\n", lenBuf, pSess->sock, rcvBuf); + CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv)); + } else if (lenRcv == 0) { + /* session was closed, do clean-up */ + CHKiRet(closeSess(pSess)); + break; + } else { + if(errno == EAGAIN || errno == EWOULDBLOCK) + break; + DBGPRINTF("imtcp: error on session socket %d - closed.\n", pSess->sock); + closeSess(pSess); /* try clean-up by dropping session */ + break; + } } finalize_it: @@ -1127,8 +1152,6 @@ CODEmodInit_QueryRegCFSLineHdlr addTCPListener, NULL, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpservernotifyonconnectionclose"), 0, eCmdHdlrBinary, NULL, &cs.bEmitMsgOnClose, STD_LOADABLE_MODULE_ID)); - //CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverstreamdriverpermittedpeer"), 0, - //eCmdHdlrGetWord, setPermittedPeer, NULL, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserveraddtlframedelimiter"), 0, eCmdHdlrInt, NULL, &cs.iAddtlFrameDelim, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr(UCHAR_CONSTANT("inputptcpserverinputname"), 0, diff --git a/tests/Makefile.am b/tests/Makefile.am index 0045f00a..b9b1ede7 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -7,6 +7,8 @@ TESTS = $(TESTRUNS) cfg.sh \ diskqueue.sh \ diskqueue-fsync.sh \ manytcp.sh \ + manyptcp.sh \ + imptcp_large.sh \ sndrcv.sh \ sndrcv_gzip.sh \ asynwr_simple.sh \ @@ -136,6 +138,10 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \ testsuites/imtcp-multiport.conf \ manytcp.sh \ testsuites/manytcp.conf \ + manyptcp.sh \ + testsuites/manyptcp.conf \ + imptcp_large.sh \ + testsuites/imptcp_large.conf \ inputname.sh \ testsuites/inputname_imtcp.conf \ testsuites/1.inputname_imtcp_12514 \ diff --git a/tests/diag.sh b/tests/diag.sh index 51ad5f6a..8659aa17 100755 --- a/tests/diag.sh +++ b/tests/diag.sh @@ -10,7 +10,7 @@ #valgrind="valgrind --tool=helgrind --log-fd=1" #valgrind="valgrind --tool=exp-ptrcheck --log-fd=1" #set -o xtrace -#export RSYSLOG_DEBUG="debug nostdout printmutexaction" +#export RSYSLOG_DEBUG="debug nostdout noprintmutexaction" #export RSYSLOG_DEBUGLOG="log" case $1 in 'init') $srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason diff --git a/tests/imptcp_large.sh b/tests/imptcp_large.sh new file mode 100755 index 00000000..f6eee895 --- /dev/null +++ b/tests/imptcp_large.sh @@ -0,0 +1,14 @@ +# Test imptcp with large messages +# added 2010-08-10 by Rgerhards +# +# This file is part of the rsyslog project, released under GPLv3 +cat rsyslog.action.1.include +source $srcdir/diag.sh init +source $srcdir/diag.sh startup imptcp_large.conf +# send 4000 messages of 10.000bytes plus header max, randomized +source $srcdir/diag.sh tcpflood -c5 -m20000 -r -d10000 -P129 +sleep 2 # due to large messages, we need this time for the tcp receiver to settle... +source $srcdir/diag.sh shutdown-when-empty # shut down rsyslogd when done processing messages +source $srcdir/diag.sh wait-shutdown # and wait for it to terminate +source $srcdir/diag.sh seq-check 0 19999 -E +source $srcdir/diag.sh exit diff --git a/tests/testsuites/imptcp_large.conf b/tests/testsuites/imptcp_large.conf new file mode 100644 index 00000000..677e33f6 --- /dev/null +++ b/tests/testsuites/imptcp_large.conf @@ -0,0 +1,16 @@ +# simple async writing test +# rgerhards, 2010-03-09 +$MaxMessageSize 10k +$IncludeConfig diag-common.conf + +$ModLoad ../plugins/imptcp/.libs/imptcp +$MainMsgQueueTimeoutShutdown 10000 +$InputPTCPServerRun 13514 + +$template outfmt,"%msg:F,58:2%,%msg:F,58:3%,%msg:F,58:4%\n" +$template dynfile,"rsyslog.out.log" # trick to use relative path names! +$OMFileFlushOnTXEnd off +$OMFileFlushInterval 2 +$OMFileIOBufferSize 256k +$IncludeConfig rsyslog.action.1.include +local0.* ?dynfile;outfmt -- cgit From 83948a07fd0a9a9a3ea7db8647a0c1d032f0bc2e Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 17:22:27 +0200 Subject: some cleanup and minor optimization --- plugins/imptcp/imptcp.c | 65 ++++++++++++++++++++++++------------------------- 1 file changed, 32 insertions(+), 33 deletions(-) diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c index 80df959c..732590a9 100644 --- a/plugins/imptcp/imptcp.c +++ b/plugins/imptcp/imptcp.c @@ -165,6 +165,16 @@ struct epolld_s { static ptcpsrv_t *pSrvRoot = NULL; static int epollfd = -1; /* (sole) descriptor for epoll */ static int iMaxLine; /* maximum size of a single message */ +/* we use a single static receive buffer, as this module is not multi-threaded. Keeping + * the buffer in the data segment is probably a little bit more efficient than on the stack + * (but at least I can't believe it will ever be less efficient ;) -- rgerhards, 2010-08-10 + * Note that we do NOT (yet?) provide a config setting to set the buffer size. For usual + * syslog traffic, it should be large enough. Also keep in mind that we run under a virtual + * memory system, so if we do not use large parts of the buffer, that's no issue at + * all -- it'll just use up address space. On the other hand, it would be silly to page in + * or page out some data just to get space for the IO buffer. + */ +static char rcvBuf[128*1024]; /* forward definitions */ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal); @@ -207,7 +217,6 @@ static rsRetVal startupSrv(ptcpsrv_t *pSrv) { DEFiRet; - int iSessMax = 200; /* TODO: Make configurable or remove? */ int error, maxs, on = 1; int sock = -1; int numSocks; @@ -217,7 +226,7 @@ startupSrv(ptcpsrv_t *pSrv) lstnIP = pSrv->lstnIP == NULL ? UCHAR_CONSTANT("") : pSrv->lstnIP; - dbgprintf("imptcp creating listen socket on server '%s', port %s\n", lstnIP, pSrv->port); + DBGPRINTF("imptcp creating listen socket on server '%s', port %s\n", lstnIP, pSrv->port); memset(&hints, 0, sizeof(hints)); hints.ai_flags = AI_PASSIVE; @@ -226,7 +235,7 @@ startupSrv(ptcpsrv_t *pSrv) error = getaddrinfo((char*)pSrv->lstnIP, (char*) pSrv->port, &hints, &res); if(error) { - dbgprintf("error %d querying server '%s', port '%s'\n", error, pSrv->lstnIP, pSrv->port); + DBGPRINTF("error %d querying server '%s', port '%s'\n", error, pSrv->lstnIP, pSrv->port); ABORT_FINALIZE(RS_RET_INVALID_PORT); } @@ -239,7 +248,7 @@ startupSrv(ptcpsrv_t *pSrv) sock = socket(r->ai_family, r->ai_socktype, r->ai_protocol); if(sock < 0) { if(!(r->ai_family == PF_INET6 && errno == EAFNOSUPPORT)) - dbgprintf("error %d creating tcp listen socket", errno); + DBGPRINTF("error %d creating tcp listen socket", errno); /* it is debatable if PF_INET with EAFNOSUPPORT should * also be ignored... */ @@ -258,7 +267,7 @@ startupSrv(ptcpsrv_t *pSrv) } #endif if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0 ) { - dbgprintf("error %d setting tcp socket option\n", errno); + DBGPRINTF("error %d setting tcp socket option\n", errno); close(sock); sock = -1; continue; @@ -273,7 +282,7 @@ startupSrv(ptcpsrv_t *pSrv) sockflags = fcntl(sock, F_SETFL, sockflags); } if(sockflags == -1) { - dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket", errno); + DBGPRINTF("error %d setting fcntl(O_NONBLOCK) on tcp socket", errno); close(sock); sock = -1; continue; @@ -302,26 +311,17 @@ startupSrv(ptcpsrv_t *pSrv) #endif ) { /* TODO: check if *we* bound the socket - else we *have* an error! */ - dbgprintf("error %d while binding tcp socket", errno); + DBGPRINTF("error %d while binding tcp socket", errno); close(sock); sock = -1; continue; } - if(listen(sock, iSessMax / 10 + 5) < 0) { - /* If the listen fails, it most probably fails because we ask - * for a too-large backlog. So in this case we first set back - * to a fixed, reasonable, limit that should work. Only if - * that fails, too, we give up. - */ - dbgprintf("listen with a backlog of %d failed - retrying with default of 32.", - iSessMax / 10 + 5); - if(listen(sock, 32) < 0) { - dbgprintf("tcp listen error %d, suspending\n", errno); - close(sock); - sock = -1; - continue; - } + if(listen(sock, 511) < 0) { + DBGPRINTF("tcp listen error %d, suspending\n", errno); + close(sock); + sock = -1; + continue; } /* if we reach this point, we were able to obtain a valid socket, so we can @@ -332,11 +332,11 @@ startupSrv(ptcpsrv_t *pSrv) } if(numSocks != maxs) - dbgprintf("We could initialize %d TCP listen sockets out of %d we received " + DBGPRINTF("We could initialize %d TCP listen sockets out of %d we received " "- this may or may not be an error indication.\n", numSocks, maxs); if(numSocks == 0) { - dbgprintf("No TCP listen sockets could successfully be initialized"); + DBGPRINTF("No TCP listen sockets could successfully be initialized"); ABORT_FINALIZE(RS_RET_COULD_NOT_BIND); } @@ -375,7 +375,7 @@ getPeerNames(prop_t **peerName, prop_t **peerIP, struct sockaddr *pAddr) error = getnameinfo(pAddr, SALEN(pAddr), (char*)szIP, sizeof(szIP), NULL, 0, NI_NUMERICHOST); if(error) { - dbgprintf("Malformed from address %s\n", gai_strerror(error)); + DBGPRINTF("Malformed from address %s\n", gai_strerror(error)); strcpy((char*)szHname, "???"); strcpy((char*)szIP, "???"); ABORT_FINALIZE(RS_RET_INVALID_HNAME); @@ -395,7 +395,7 @@ getPeerNames(prop_t **peerName, prop_t **peerIP, struct sockaddr *pAddr) freeaddrinfo (res); /* OK, we know we have evil, so let's indicate this to our caller */ snprintf((char*)szHname, NI_MAXHOST, "[MALICIOUS:IP=%s]", szIP); - dbgprintf("Malicious PTR record, IP = \"%s\" HOST = \"%s\"", szIP, szHname); + DBGPRINTF("Malicious PTR record, IP = \"%s\" HOST = \"%s\"", szIP, szHname); iRet = RS_RET_MALICIOUS_HNAME; } } else { @@ -450,7 +450,7 @@ AcceptConnReq(int sock, int *newSock, prop_t **peerName, prop_t **peerIP) sockflags = fcntl(iNewSock, F_SETFL, sockflags); } if(sockflags == -1) { - dbgprintf("error %d setting fcntl(O_NONBLOCK) on tcp socket %d", errno, iNewSock); + DBGPRINTF("error %d setting fcntl(O_NONBLOCK) on tcp socket %d", errno, iNewSock); ABORT_FINALIZE(RS_RET_IO_ERROR); } @@ -544,21 +544,21 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG if(isdigit(c)) { pThis->iOctetsRemain = pThis->iOctetsRemain * 10 + c - '0'; } else { /* done with the octet count, so this must be the SP terminator */ - dbgprintf("TCP Message with octet-counter, size %d.\n", pThis->iOctetsRemain); + DBGPRINTF("TCP Message with octet-counter, size %d.\n", pThis->iOctetsRemain); if(c != ' ') { errmsg.LogError(0, NO_ERRCODE, "Framing Error in received TCP message: " "delimiter is not SP but has ASCII value %d.\n", c); } if(pThis->iOctetsRemain < 1) { /* TODO: handle the case where the octet count is 0! */ - dbgprintf("Framing Error: invalid octet count\n"); + DBGPRINTF("Framing Error: invalid octet count\n"); errmsg.LogError(0, NO_ERRCODE, "Framing Error in received TCP message: " "invalid octet count %d.\n", pThis->iOctetsRemain); } else if(pThis->iOctetsRemain > iMaxLine) { /* while we can not do anything against it, we can at least log an indication * that something went wrong) -- rgerhards, 2008-03-14 */ - dbgprintf("truncating message with %d octets - max msg size is %d\n", + DBGPRINTF("truncating message with %d octets - max msg size is %d\n", pThis->iOctetsRemain, iMaxLine); errmsg.LogError(0, NO_ERRCODE, "received oversize message: size is %d bytes, " "max msg size is %d, truncating...\n", pThis->iOctetsRemain, iMaxLine); @@ -569,7 +569,7 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG assert(pThis->inputState == eInMsg); if(pThis->iMsg >= iMaxLine) { /* emergency, we now need to flush, no matter if we are at end of message or not... */ - dbgprintf("error: message received is larger than max msg size, we split it\n"); + DBGPRINTF("error: message received is larger than max msg size, we split it\n"); doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub); /* we might think if it is better to ignore the rest of the * message than to treat it as a new one. Maybe this is a good @@ -808,8 +808,8 @@ closeSess(ptcpsess_t *pSess) /* finally unlink session from structures */ //fprintf(stderr, "closing session %d next %p, prev %p\n", pSess->sock, pSess->next, pSess->prev); -//dbgprintf("imptcp: pSess->next %p\n", pSess->next); -//dbgprintf("imptcp: pSess->prev %p\n", pSess->prev); +//DBGPRINTF("imptcp: pSess->next %p\n", pSess->next); +//DBGPRINTF("imptcp: pSess->prev %p\n", pSess->prev); if(pSess->next != NULL) pSess->next->prev = pSess->prev; if(pSess->prev == NULL) { @@ -957,7 +957,6 @@ sessActivity(ptcpsess_t *pSess) { int lenRcv; int lenBuf; - char rcvBuf[128*1024]; DEFiRet; int iac = 0; -- cgit From b5b4bd138949f09b3703e6f7f5caf46d361fa41c Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 18:29:22 +0200 Subject: small improvements to testbench --- tests/complex1.sh | 2 +- tests/imptcp_large.sh | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/complex1.sh b/tests/complex1.sh index 7f3cd994..954ddf6c 100755 --- a/tests/complex1.sh +++ b/tests/complex1.sh @@ -12,7 +12,7 @@ source $srcdir/diag.sh init source $srcdir/diag.sh startup complex1.conf # send 30,000 messages of 400 bytes plus header max, via three dest ports source $srcdir/diag.sh tcpflood -m40000 -rd400 -P129 -f5 -n3 -c15 -i1 -sleep 2 # due to large messages, we need this time for the tcp receiver to settle... +sleep 4 # due to large messages, we need this time for the tcp receiver to settle... source $srcdir/diag.sh shutdown-when-empty # shut down rsyslogd when done processing messages source $srcdir/diag.sh wait-shutdown # and wait for it to terminate ls rsyslog.out.*.log diff --git a/tests/imptcp_large.sh b/tests/imptcp_large.sh index f6eee895..b4d130bb 100755 --- a/tests/imptcp_large.sh +++ b/tests/imptcp_large.sh @@ -2,6 +2,8 @@ # added 2010-08-10 by Rgerhards # # This file is part of the rsyslog project, released under GPLv3 +echo ==================================================================================== +echo TEST: \[imptcp_large.sh\]: test imptcp with large-size messages cat rsyslog.action.1.include source $srcdir/diag.sh init source $srcdir/diag.sh startup imptcp_large.conf -- cgit From c664adfa1683763ae503f2da9e8d4d528d6634d1 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 10 Aug 2010 19:34:42 +0200 Subject: more testbench improvements and new tests --- tests/Makefile.am | 6 +++ tests/imptcp_conndrop.sh | 16 ++++++ tests/imtcp_conndrop.sh | 16 ++++++ tests/tcpflood.c | 95 ++++++++++++++--------------------- tests/testsuites/imptcp_conndrop.conf | 16 ++++++ tests/testsuites/imtcp_conndrop.conf | 16 ++++++ 6 files changed, 107 insertions(+), 58 deletions(-) create mode 100755 tests/imptcp_conndrop.sh create mode 100755 tests/imtcp_conndrop.sh create mode 100644 tests/testsuites/imptcp_conndrop.conf create mode 100644 tests/testsuites/imtcp_conndrop.conf diff --git a/tests/Makefile.am b/tests/Makefile.am index b9b1ede7..b500bc85 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -9,6 +9,8 @@ TESTS = $(TESTRUNS) cfg.sh \ manytcp.sh \ manyptcp.sh \ imptcp_large.sh \ + imptcp_conndrop.sh \ + imtcp_conndrop.sh \ sndrcv.sh \ sndrcv_gzip.sh \ asynwr_simple.sh \ @@ -142,6 +144,10 @@ EXTRA_DIST= 1.rstest 2.rstest 3.rstest err1.rstest \ testsuites/manyptcp.conf \ imptcp_large.sh \ testsuites/imptcp_large.conf \ + imptcp_conndrop.sh \ + testsuites/imptcp_conndrop.conf \ + imtcp_conndrop.sh \ + testsuites/imtcp_conndrop.conf \ inputname.sh \ testsuites/inputname_imtcp.conf \ testsuites/1.inputname_imtcp_12514 \ diff --git a/tests/imptcp_conndrop.sh b/tests/imptcp_conndrop.sh new file mode 100755 index 00000000..684de6b5 --- /dev/null +++ b/tests/imptcp_conndrop.sh @@ -0,0 +1,16 @@ +# Test imptcp with many dropping connections +# added 2010-08-10 by Rgerhards +# +# This file is part of the rsyslog project, released under GPLv3 +echo ==================================================================================== +echo TEST: \[imptcp_conndrop.sh\]: test imptcp with random connection drops +cat rsyslog.action.1.include +source $srcdir/diag.sh init +source $srcdir/diag.sh startup imptcp_large.conf +# 100 byte messages to gain more practical data use +source $srcdir/diag.sh tcpflood -c20 -m50000 -r -d100 -P129 -D +sleep 4 # due to large messages, we need this time for the tcp receiver to settle... +source $srcdir/diag.sh shutdown-when-empty # shut down rsyslogd when done processing messages +source $srcdir/diag.sh wait-shutdown # and wait for it to terminate +source $srcdir/diag.sh seq-check 0 49999 -E +source $srcdir/diag.sh exit diff --git a/tests/imtcp_conndrop.sh b/tests/imtcp_conndrop.sh new file mode 100755 index 00000000..2caa0ce2 --- /dev/null +++ b/tests/imtcp_conndrop.sh @@ -0,0 +1,16 @@ +# Test imtcp with many dropping connections +# added 2010-08-10 by Rgerhards +# +# This file is part of the rsyslog project, released under GPLv3 +echo ==================================================================================== +echo TEST: \[imtcp_conndrop.sh\]: test imtcp with random connection drops +cat rsyslog.action.1.include +source $srcdir/diag.sh init +source $srcdir/diag.sh startup imptcp_large.conf +# 100 byte messages to gain more practical data use +source $srcdir/diag.sh tcpflood -c20 -m50000 -r -d100 -P129 -D +sleep 4 # due to large messages, we need this time for the tcp receiver to settle... +source $srcdir/diag.sh shutdown-when-empty # shut down rsyslogd when done processing messages +source $srcdir/diag.sh wait-shutdown # and wait for it to terminate +source $srcdir/diag.sh seq-check 0 49999 -E +source $srcdir/diag.sh exit diff --git a/tests/tcpflood.c b/tests/tcpflood.c index d8c3f038..ee43449e 100644 --- a/tests/tcpflood.c +++ b/tests/tcpflood.c @@ -28,6 +28,8 @@ * delemiters into account. * -C when input from a file is read, this file is transmitted -C times * (C like cycle, running out of meaningful option switches ;)) + * -D randomly drop and re-establish connections. Useful for stress-testing + * the TCP receiver. * * Part of the testbench for rsyslog. * @@ -82,11 +84,13 @@ static int numConnections = 1; /* number of connections to create */ static int *sockArray; /* array of sockets to use */ static int msgNum = 0; /* initial message number to start with */ static int bShowProgress = 1; /* show progress messages */ +static int bRandConnDrop = 0; /* randomly drop connections? */ static char *MsgToSend = NULL; /* if non-null, this is the actual message to send */ static int bBinaryFile = 0; /* is -I file binary */ static char *dataFile = NULL; /* name of data file, if NULL, generate own data */ static int numFileIterations = 1;/* how often is file data to be sent? */ FILE *dataFP = NULL; /* file pointer for data file, if used */ +static long nConnDrops = 0; /* counter: number of time connection was dropped (-D option) */ /* open a single tcp connection @@ -153,8 +157,6 @@ int openConnections(void) if(i % 10 == 0) { if(bShowProgress) printf("\r%5.5d", i); - //lenMsg = sprintf(msgBuf, "\r%5.5d", i); - //write(1, msgBuf, lenMsg); } if(openConn(&(sockArray[i])) != 0) { printf("error in trying to open connection i=%d\n", i); @@ -179,6 +181,7 @@ void closeConnections(void) { int i; size_t lenMsg; + struct linger ling; char msgBuf[128]; if(bShowProgress) @@ -190,7 +193,15 @@ void closeConnections(void) write(1, msgBuf, lenMsg); } } - close(sockArray[i]); + if(sockArray[i] != -1) { + /* we try to not overrun the receiver by trying to flush buffers + * *during* close(). -- rgerhards, 2010-08-10 + */ + ling.l_onoff = 1; + ling.l_linger = 1; + setsockopt(sockArray[i], SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)); + close(sockArray[i]); + } } lenMsg = sprintf(msgBuf, "\r%5.5d close connections\n", i); write(1, msgBuf, lenMsg); @@ -287,12 +298,18 @@ int sendMessages(void) socknum = i - (numMsgsToSend - numConnections); else { int rnd = rand(); - //socknum = rand() % numConnections; socknum = rnd % numConnections; } genMsg(buf, sizeof(buf), &lenBuf); /* generate the message to send according to params */ if(lenBuf == 0) break; /* end of processing! */ + if(sockArray[socknum] == -1) { + /* connection was dropped, need to re-establish */ + if(openConn(&(sockArray[socknum])) != 0) { + printf("error in trying to re-open connection %d\n", socknum); + exit(1); + } + } lenSend = send(sockArray[socknum], buf, lenBuf, 0); if(lenSend != lenBuf) { printf("\r%5.5d\n", i); @@ -307,6 +324,16 @@ int sendMessages(void) if(bShowProgress) printf("\r%8.8d", i); } + if(bRandConnDrop) { + /* if we need to randomly drop connections, see if we + * are a victim + */ + if(rand() > (int) (RAND_MAX * 0.95)) { + ++nConnDrops; + close(sockArray[socknum]); + sockArray[socknum] = -1; + } + } ++msgNum; ++i; } @@ -316,59 +343,6 @@ int sendMessages(void) } -/* send a message via TCP - * We open the connection on the initial send, and never close it - * (let the OS do that). If a conneciton breaks, we do NOT try to - * recover, so all test after that one will fail (and the test - * driver probably hang. returns 0 if ok, something else otherwise. - * We use traditional framing '\n' at EOR for this tester. It may be - * worth considering additional framing modes. - * rgerhards, 2009-04-08 - */ -int -tcpSend(char *buf, int lenBuf) -{ - static int sock = INVALID_SOCKET; - struct sockaddr_in addr; - - if(sock == INVALID_SOCKET) { - /* first time, need to connect to target */ - if((sock=socket(AF_INET, SOCK_STREAM, 0))==-1) { - perror("socket()"); - return(1); - } - - memset((char *) &addr, 0, sizeof(addr)); - addr.sin_family = AF_INET; - addr.sin_port = htons(13514); - if(inet_aton("127.0.0.1", &addr.sin_addr)==0) { - fprintf(stderr, "inet_aton() failed\n"); - return(1); - } - if(connect(sock, (struct sockaddr*)&addr, sizeof(addr)) != 0) { - fprintf(stderr, "connect() failed\n"); - return(1); - } - } - - /* send test data */ - if(send(sock, buf, lenBuf, 0) != lenBuf) { - perror("send test data"); - fprintf(stderr, "send() failed\n"); - return(1); - } - - /* send record terminator */ - if(send(sock, "\n", 1, 0) != 1) { - perror("send record terminator"); - fprintf(stderr, "send() failed\n"); - return(1); - } - - return 0; -} - - /* Run the test. * rgerhards, 2009-04-03 */ @@ -394,7 +368,7 @@ int main(int argc, char *argv[]) if(!isatty(1)) bShowProgress = 0; - while((opt = getopt(argc, argv, "f:t:p:c:C:m:i:I:P:d:n:M:rB")) != -1) { + while((opt = getopt(argc, argv, "f:t:p:c:C:m:i:I:P:d:Dn:M:rB")) != -1) { switch (opt) { case 't': targetIP = optarg; break; @@ -419,6 +393,8 @@ int main(int argc, char *argv[]) exit(1); } break; + case 'D': bRandConnDrop = 1; + break; case 'r': bRandomizeExtraData = 1; break; case 'f': dynFileIDs = atoi(optarg); @@ -456,7 +432,10 @@ int main(int argc, char *argv[]) exit(1); } + if(nConnDrops > 0) + printf("-D option initiated %ld connection closures\n", nConnDrops); printf("End of tcpflood Run\n"); + closeConnections(); /* this is important so that we do not finish too early! */ exit(ret); } diff --git a/tests/testsuites/imptcp_conndrop.conf b/tests/testsuites/imptcp_conndrop.conf new file mode 100644 index 00000000..677e33f6 --- /dev/null +++ b/tests/testsuites/imptcp_conndrop.conf @@ -0,0 +1,16 @@ +# simple async writing test +# rgerhards, 2010-03-09 +$MaxMessageSize 10k +$IncludeConfig diag-common.conf + +$ModLoad ../plugins/imptcp/.libs/imptcp +$MainMsgQueueTimeoutShutdown 10000 +$InputPTCPServerRun 13514 + +$template outfmt,"%msg:F,58:2%,%msg:F,58:3%,%msg:F,58:4%\n" +$template dynfile,"rsyslog.out.log" # trick to use relative path names! +$OMFileFlushOnTXEnd off +$OMFileFlushInterval 2 +$OMFileIOBufferSize 256k +$IncludeConfig rsyslog.action.1.include +local0.* ?dynfile;outfmt diff --git a/tests/testsuites/imtcp_conndrop.conf b/tests/testsuites/imtcp_conndrop.conf new file mode 100644 index 00000000..b64f132b --- /dev/null +++ b/tests/testsuites/imtcp_conndrop.conf @@ -0,0 +1,16 @@ +# simple async writing test +# rgerhards, 2010-03-09 +$MaxMessageSize 10k +$IncludeConfig diag-common.conf + +$ModLoad ../plugins/imtcp/.libs/imtcp +$MainMsgQueueTimeoutShutdown 10000 +$InputTCPServerRun 13514 + +$template outfmt,"%msg:F,58:2%,%msg:F,58:3%,%msg:F,58:4%\n" +$template dynfile,"rsyslog.out.log" # trick to use relative path names! +$OMFileFlushOnTXEnd off +$OMFileFlushInterval 2 +$OMFileIOBufferSize 256k +$IncludeConfig rsyslog.action.1.include +local0.* ?dynfile;outfmt -- cgit