diff options
| -rw-r--r-- | ChangeLog | 19 | ||||
| -rw-r--r-- | doc/rsyslog_conf_modules.html | 7 | ||||
| -rw-r--r-- | plugins/mmnormalize/mmnormalize.c | 24 | ||||
| -rw-r--r-- | tests/filewriter.c | 158 | ||||
| -rw-r--r-- | tests/tcpflood.c | 271 |
5 files changed, 435 insertions, 44 deletions
@@ -1,5 +1,22 @@ --------------------------------------------------------------------------- -Version 6.1.2 [DEVEL] (rgerhards), 2010-12-?? +Version 6.1.2 [DEVEL] (rgerhards), 2010-12-13 +- added experimental support for log normalizaton (via liblognorm) + support for normalizing log messages has been added in the form of + mmnormalize. The core engine (property replacer, filter engine) has + been enhanced to support properties from normalized events. + Note: this is EXPERIMENTAL code. It is currently know that + there are issues if the functionality is used with + - disk-based queues + - asynchronous action queues + You can not use the new functionality together with these features. + This limitation will be removed in later releases. However, we + preferred to release early, so that one can experiment with the new + feature set and accepted the price that this means the full set of + functionality is not yet available. If not used together with + these features, log normalizing should be pretty stable. +- enhanced testing tool tcpflood + now supports sending via UDP and the capability to run multiple + iterations and generate statistics data records - bugfix: potential abort when output modules with different parameter passing modes were used in configured output modules --------------------------------------------------------------------------- diff --git a/doc/rsyslog_conf_modules.html b/doc/rsyslog_conf_modules.html index 74aa319c..7623c114 100644 --- a/doc/rsyslog_conf_modules.html +++ b/doc/rsyslog_conf_modules.html @@ -93,12 +93,17 @@ repated n times" messages emitted by some syslogds. They can be implemented using either the output module or the parser module interface. From the rsyslog core's point of view, they actually are output or parser modules, it is their implementation that makes them special. -<p>Currently, there do not exist any such modules, but could be written with +<p>Currently, there exists only a limited set of such modules, but new ones could be written with the methods the engine provides. They could be used, for example, to: <ul> <li>anonymize message content <li>add dynamically computed content to message (fields) </ul> +<p>Currently supported modules are: +<ul> +<li><a href="mmnormalize.html">mmnormalize</a> - used to normalize +log messages. +</ul> <a name="lm"></a><h2>String Generator Modules</h2> <p>String generator modules are used, as the name implies, to generate strings based diff --git a/plugins/mmnormalize/mmnormalize.c b/plugins/mmnormalize/mmnormalize.c index 8c4efd81..49b143b6 100644 --- a/plugins/mmnormalize/mmnormalize.c +++ b/plugins/mmnormalize/mmnormalize.c @@ -64,7 +64,7 @@ typedef struct _instanceData { } instanceData; typedef struct configSettings_s { - uchar *sampdb; /**< name of sample db to use */ + uchar *rulebase; /**< name of sample db to use */ sbool bUseRawMsg; /**< use %rawmsg% instead of %msg% */ } configSettings_t; @@ -148,7 +148,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) ABORT_FINALIZE(RS_RET_CONFLINE_UNPROCESSED); } - if(cs.sampdb == NULL) { + if(cs.rulebase == NULL) { errmsg.LogError(0, RS_RET_NO_RULESET, "error: no sample database was specified, use " "$MMNormalizeSampleDB directive first!"); ABORT_FINALIZE(RS_RET_NO_RULESET); @@ -180,9 +180,9 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_INIT); } ln_setEECtx(pData->ctxln, pData->ctxee); - if(ln_loadSamples(pData->ctxln, (char*) cs.sampdb) != 0) { + if(ln_loadSamples(pData->ctxln, (char*) cs.rulebase) != 0) { errmsg.LogError(0, RS_RET_NO_RULESET, "error: sample db '%s' could not be loaded " - "cannot activate action", cs.sampdb); + "cannot activate action", cs.rulebase); ee_exitCtx(pData->ctxee); ln_exitCtx(pData->ctxln); ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD); @@ -191,8 +191,8 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) /* all config vars auto-reset! */ cs.bUseRawMsg = 0; - free(cs.sampdb); - cs.sampdb = NULL; + free(cs.rulebase); + cs.rulebase = NULL; CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct @@ -215,17 +215,17 @@ ENDqueryEtryPt static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal) { DEFiRet; - cs.sampdb = NULL; + cs.rulebase = NULL; cs.bUseRawMsg = 0; RETiRet; } -/* set the sampdb name */ +/* set the rulebase name */ static rsRetVal -setSampDB(void __attribute__((unused)) *pVal, uchar *pszName) +setRuleBase(void __attribute__((unused)) *pVal, uchar *pszName) { DEFiRet; - cs.sampdb = pszName; + cs.rulebase = pszName; pszName = NULL; RETiRet; } @@ -261,8 +261,8 @@ CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmnormalizesampledb", 0, eCmdHdlrGetWord, - setSampDB, NULL, STD_LOADABLE_MODULE_ID, eConfObjAction)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmnormalizerulebase", 0, eCmdHdlrGetWord, + setRuleBase, NULL, STD_LOADABLE_MODULE_ID, eConfObjAction)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"mmnormalizeuserawmsg", 0, eCmdHdlrInt, NULL, &cs.bUseRawMsg, STD_LOADABLE_MODULE_ID, eConfObjAction)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, diff --git a/tests/filewriter.c b/tests/filewriter.c new file mode 100644 index 00000000..07991b1d --- /dev/null +++ b/tests/filewriter.c @@ -0,0 +1,158 @@ +/* This program expands the input file several times. This + * is done in order to obtain large (and maybe huge) files for + * testing. Note that the input file is stored in memory. It's + * last line must properly be terminated. + * Max input line size is 10K. + * + * command line options: + * -i file to be used for input (else stdin) + * -o file to be used for output (else stdout) + * -c number of times the file is to be copied + * -n add line numbers (default: off) + * -w wait nbr of microsecs between batches + * -W number of file lines to generate in a batch + * This is useful only if -w is specified as well, + * default is 1000. + * + * Copyright 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 <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + */ +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <getopt.h> + +/* input file is stored in a single-linked list */ +struct line { + struct line *next; + char *ln; +} *root, *tail; + +static FILE *fpIn; +static FILE *fpOut; +static long long nCopies = 1; +static int linenbrs = 0; +static int waitusecs = 0; +static int batchsize = 1000; + + +/* read the input file and create in-memory representation + */ +static inline void +readFile() +{ + char *r; + char lnBuf[10240]; + struct line *node; + + root = tail = NULL; + r = fgets(lnBuf, sizeof(lnBuf), fpIn); + while(r != NULL) { + node = malloc(sizeof(struct line)); + if(node == NULL) { + perror("malloc node"); + exit(1); + } + node->next = NULL; + node->ln = strdup(lnBuf); + if(node->ln == NULL) { + perror("malloc node"); + exit(1); + } + if(tail == NULL) { + tail = root = node; + } else { + tail->next = node; + tail = node; + } + r = fgets(lnBuf, sizeof(lnBuf), fpIn); + } + if(!feof(fpIn)) { + perror("fgets"); + fprintf(stderr, "end of read loop, but not end of file!"); + exit(1); + } +} + + +static void +genCopies() +{ + long long i; + long long unsigned lnnbr; + struct line *node; + + lnnbr = 1; + for(i = 0 ; i < nCopies ; ++i) { + if(i % 10000 == 0) + fprintf(stderr, "copyrun %d\n", i); + if(waitusecs && (i % batchsize == 0)) { + usleep(waitusecs); + } + for(node = root ; node != NULL ; node = node->next) { + if(linenbrs) + fprintf(fpOut, "%12.12llu:%s", lnnbr, node->ln); + else + fprintf(fpOut, "%s", node->ln); + ++lnnbr; + } + } +} + +void main(int argc, char *argv[]) +{ + int opt; + fpIn = stdin; + fpOut = stdout; + + while((opt = getopt(argc, argv, "i:o:c:nw:W:")) != -1) { + switch (opt) { + case 'i': /* input file */ + if((fpIn = fopen(optarg, "r")) == NULL) { + perror(optarg); + exit(1); + } + break; + case 'o': /* output file */ + if((fpOut = fopen(optarg, "w")) == NULL) { + perror(optarg); + exit(1); + } + break; + case 'c': + nCopies = atoll(optarg); + break; + case 'n': + linenbrs = 1; + break; + case 'w': + waitusecs = atoi(optarg); + break; + case 'W': + batchsize = atoi(optarg); + break; + default: printf("invalid option '%c' or value missing - terminating...\n", opt); + exit (1); + break; + } + } + + readFile(); + genCopies(); +} diff --git a/tests/tcpflood.c b/tests/tcpflood.c index a37845a3..acca4452 100644 --- a/tests/tcpflood.c +++ b/tests/tcpflood.c @@ -16,6 +16,7 @@ * bytes as forth. Add -r to randomize the amount of extra * data included in the range 1..(value of -d). * -r randomize amount of extra data added (-d must be > 0) + * -s (silent) do not show progress indicator (never done on non-tty) * -f support for testing dynafiles. If given, include a dynafile ID * in the range 0..(f-1) as the SECOND field, shifting all field values * one field to the right. Zero (default) disables this functionality. @@ -32,6 +33,19 @@ * -D randomly drop and re-establish connections. Useful for stress-testing * the TCP receiver. * -F USASCII value for frame delimiter (in octet-stuffing mode), default LF + * -R number of times the test shall be run (very useful for gathering performance + * data and other repetitive things). Default: 1 + * -S number of seconds to sleep between different runs (-R) Default: 30 + * -X generate sTats data records. Default: off + * -e encode output in CSV (not yet everywhere supported) + * for performance data: + * each inidividual line has the runtime of one test + * the last line has 0 in field 1, followed by numberRuns,TotalRuntime, + * Average,min,max + * -T transport to use. Currently supported: "udp", "tcp" (default) + * Note: UDP supports a single target port, only + * -W wait time between sending batches of messages, in microseconds (Default: 0) + * -b number of messages within a batch (default: 100,000,000 millions) * * Part of the testbench for rsyslog. * @@ -67,6 +81,7 @@ #include <string.h> #include <netinet/in.h> #include <sys/resource.h> +#include <sys/time.h> #define EXIT_FAILURE 1 #define INVALID_SOCKET -1 @@ -87,6 +102,7 @@ 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 bSilent = 0; /* completely silent operation */ 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 */ @@ -95,6 +111,44 @@ static int numFileIterations = 1;/* how often is file data to be sent? */ static char frameDelim = '\n'; /* default frame delimiter */ FILE *dataFP = NULL; /* file pointer for data file, if used */ static long nConnDrops = 0; /* counter: number of time connection was dropped (-D option) */ +static int numRuns = 1; /* number of times the test shall be run */ +static int sleepBetweenRuns = 30; /* number of seconds to sleep between runs */ +static int bStatsRecords = 0; /* generate stats records */ +static int bCSVoutput = 0; /* generate output in CSV (where applicable) */ +static long long batchsize = 100000000ll; +static int waittime = 0; + + +/* the following structure is used to gather performance data */ +struct runstats { + unsigned long long totalRuntime; + unsigned long minRuntime; + unsigned long maxRuntime; + int numRuns; +}; + +static int udpsock; /* socket for sending in UDP mode */ +static struct sockaddr_in udpRcvr; /* remote receiver in UDP mode */ + +static enum { TP_UDP, TP_TCP } transport = TP_TCP; + +/* prepare send subsystem for UDP send */ +static inline int +setupUDP(void) +{ + if((udpsock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) + return 1; + + memset((char *) &udpRcvr, 0, sizeof(udpRcvr)); + udpRcvr.sin_family = AF_INET; + udpRcvr.sin_port = htons(targetPort); + if(inet_aton(targetIP, &udpRcvr.sin_addr)==0) { + fprintf(stderr, "inet_aton() failed\n"); + return(1); + } + + return 0; +} /* open a single tcp connection @@ -154,6 +208,9 @@ int openConnections(void) char msgBuf[128]; size_t lenMsg; + if(transport == TP_UDP) + return setupUDP(); + if(bShowProgress) write(1, " open connections", sizeof(" open connections")-1); sockArray = calloc(numConnections, sizeof(int)); @@ -167,8 +224,10 @@ int openConnections(void) return 1; } } - lenMsg = sprintf(msgBuf, "\r%5.5d open connections\n", i); - write(1, msgBuf, lenMsg); + if(bShowProgress) { + lenMsg = sprintf(msgBuf, "\r%5.5d open connections\n", i); + write(1, msgBuf, lenMsg); + } return 0; } @@ -188,6 +247,9 @@ void closeConnections(void) struct linger ling; char msgBuf[128]; + if(transport != TP_TCP) + return; + if(bShowProgress) write(1, " close connections", sizeof(" close connections")-1); for(i = 0 ; i < numConnections ; ++i) { @@ -207,8 +269,10 @@ void closeConnections(void) close(sockArray[i]); } } - lenMsg = sprintf(msgBuf, "\r%5.5d close connections\n", i); - write(1, msgBuf, lenMsg); + if(bShowProgress) { + lenMsg = sprintf(msgBuf, "\r%5.5d close connections\n", i); + write(1, msgBuf, lenMsg); + } } @@ -218,12 +282,11 @@ void closeConnections(void) * of constructing test messages. -- rgerhards, 2010-03-31 */ static inline void -genMsg(char *buf, size_t maxBuf, int *pLenBuf) +genMsg(char *buf, size_t maxBuf, int *pLenBuf, long long *numMsgsGen) { int edLen; /* actual extra data length to use */ char extraData[MAX_EXTRADATA_LEN + 1]; char dynFileIDBuf[128] = ""; - static int numMsgsGen = 0; int done; if(dataFP != NULL) { @@ -263,7 +326,7 @@ genMsg(char *buf, size_t maxBuf, int *pLenBuf) *pLenBuf = snprintf(buf, maxBuf, "%s\n", MsgToSend); } - if(numMsgsGen++ >= numMsgsToSend) + if((*numMsgsGen)++ >= (unsigned) numMsgsToSend) *pLenBuf = 0; /* indicate end of run */ finalize_it: ; @@ -284,14 +347,18 @@ int sendMessages(void) int lenBuf; int lenSend; char *statusText; + long long numSent = 0; /* number of messages sent in this test */ char buf[MAX_EXTRADATA_LEN + 1024]; - if(dataFile == NULL) { - printf("Sending %d messages.\n", numMsgsToSend); - statusText = "messages"; - } else { - printf("Sending file '%s' %d times.\n", dataFile, numFileIterations); - statusText = "kb"; + if(!bSilent) { + if(dataFile == NULL) { + printf("Sending %d messages.\n", numMsgsToSend); + statusText = "messages"; + } else { + printf("Sending file '%s' %d times.\n", dataFile, + numFileIterations); + statusText = "kb"; + } } if(bShowProgress) printf("\r%8.8d %s sent", 0, statusText); @@ -304,23 +371,27 @@ int sendMessages(void) int rnd = rand(); socknum = rnd % numConnections; } - genMsg(buf, sizeof(buf), &lenBuf); /* generate the message to send according to params */ + genMsg(buf, sizeof(buf), &lenBuf, &numSent); /* 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); + if(transport == TP_TCP) { + 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); + } else if(transport == TP_UDP) { + lenSend = sendto(udpsock, buf, lenBuf, 0, &udpRcvr, sizeof(udpRcvr)); } - lenSend = send(sockArray[socknum], buf, lenBuf, 0); if(lenSend != lenBuf) { printf("\r%5.5d\n", i); fflush(stdout); perror("send test data"); - printf("send() failed at socket %d, index %d, msgNum %d\n", - sockArray[socknum], i, msgNum); + printf("send() failed at socket %d, index %d, msgNum %lld\n", + sockArray[socknum], i, numSent); fflush(stderr); return(1); } @@ -338,10 +409,121 @@ int sendMessages(void) sockArray[socknum] = -1; } } + if(numSent % batchsize == 0) { + usleep(waittime); + } ++msgNum; ++i; } - printf("\r%8.8d %s sent\n", i, statusText); + if(!bSilent) + printf("\r%8.8d %s sent\n", i, statusText); + + return 0; +} + + +/* functions related to computing statistics on the runtime of a test. This is + * a separate function primarily not to mess up the test driver. + * rgerhards, 2010-12-08 + */ +static inline void +endTiming(struct timeval *tvStart, struct runstats *stats) +{ + long sec, usec; + unsigned long runtime; + struct timeval tvEnd; + + gettimeofday(&tvEnd, NULL); + if(tvStart->tv_usec > tvEnd.tv_usec) { + tvEnd.tv_sec--; + tvEnd.tv_usec += 1000000; + } + + sec = tvEnd.tv_sec - tvStart->tv_sec; + usec = tvEnd.tv_usec - tvStart->tv_usec; + + runtime = sec * 1000 + (usec / 1000); + stats->totalRuntime += runtime; + if(runtime < stats->minRuntime) + stats->minRuntime = runtime; + if(runtime > stats->maxRuntime) + stats->maxRuntime = runtime; + + if(!bSilent || bStatsRecords) { + if(bCSVoutput) { + printf("%ld.%4.4ld\n", runtime / 1000, runtime % 1000); + } else { + printf("runtime: %ld.%4.4ld\n", runtime / 1000, runtime % 1000); + } + } +} + + +/* generate stats summary record at end of run + */ +static inline void +genStats(struct runstats *stats) +{ + long unsigned avg; + avg = stats->totalRuntime / stats->numRuns; + + if(bCSVoutput) { + printf("#numRuns,TotalRuntime,AvgRuntime,MinRuntime,MaxRuntime\n"); + printf("%d,%llu.%4.4d,%lu.%4.4lu,%lu.%4.4lu,%lu.%4.4lu\n", + stats->numRuns, + stats->totalRuntime / 1000, (int) stats->totalRuntime % 1000, + avg / 1000, avg % 1000, + stats->minRuntime / 1000, stats->minRuntime % 1000, + stats->maxRuntime / 1000, stats->maxRuntime % 1000); + } else { + printf("Runs: %d\n", stats->numRuns); + printf("Runtime:\n"); + printf(" total: %llu.%4.4d\n", stats->totalRuntime / 1000, + (int) stats->totalRuntime % 1000); + printf(" avg: %lu.%4.4lu\n", avg / 1000, avg % 1000); + printf(" min: %lu.%4.4lu\n", stats->minRuntime / 1000, stats->minRuntime % 1000); + printf(" max: %lu.%4.4lu\n", stats->maxRuntime / 1000, stats->maxRuntime % 1000); + printf("All times are wallclock time.\n"); + } +} + + +/* Run the actual test. This function handles various meta-parameters, like + * a specified number of iterations, performance measurement and so on... + * rgerhards, 2010-12-08 + */ +static int +runTests(void) +{ + struct timeval tvStart; + struct runstats stats; + int run; + + stats.totalRuntime = 0; + stats.minRuntime = (unsigned long) 0xffffffffffffffff; + stats.maxRuntime = 0; + stats.numRuns = numRuns; + run = 1; + while(1) { /* loop broken inside */ + if(!bSilent) + printf("starting run %d\n", run); + gettimeofday(&tvStart, NULL); + if(sendMessages() != 0) { + printf("error sending messages (run %d)\n", run); + return 1; + } + endTiming(&tvStart, &stats); + if(run == numRuns) + break; + if(!bSilent) + printf("sleeping %d seconds before next run\n", sleepBetweenRuns); + sleep(sleepBetweenRuns); + ++run; + } + + if(bStatsRecords) { + genStats(&stats); + } return 0; } @@ -369,11 +551,10 @@ int main(int argc, char *argv[]) setvbuf(stdout, buf, _IONBF, 48); - if(!isatty(1)) - bShowProgress = 0; - - while((opt = getopt(argc, argv, "f:F:t:p:c:C:m:i:I:P:d:Dn:M:rB")) != -1) { + while((opt = getopt(argc, argv, "b:ef:F:t:p:c:C:m:i:I:P:d:Dn:M:rsBR:S:T:XW:")) != -1) { switch (opt) { + case 'b': batchsize = atoll(optarg); + break; case 't': targetIP = optarg; break; case 'p': targetPort = atoi(optarg); @@ -413,14 +594,43 @@ int main(int argc, char *argv[]) */ numMsgsToSend = 1000000; break; + case 's': bSilent = 1; + break; case 'B': bBinaryFile = 1; break; + case 'R': numRuns = atoi(optarg); + break; + case 'S': sleepBetweenRuns = atoi(optarg); + break; + case 'X': bStatsRecords = 1; + break; + case 'e': bCSVoutput = 1; + break; + case 'T': if(!strcmp(optarg, "udp")) { + transport = TP_UDP; + } else if(!strcmp(optarg, "tcp")) { + transport = TP_TCP; + } else { + fprintf(stderr, "unkonwn transport '%s'\n", optarg); + exit(1); + } + break; + case 'W': waittime = atoi(optarg); + break; default: printf("invalid option '%c' or value missing - terminating...\n", opt); exit (1); break; } } + if(bStatsRecords && waittime) { + fprintf(stderr, "warning: generating performance stats and useing a waittime " + "is somewhat contradictory!\n"); + } + + if(!isatty(1) || bSilent) + bShowProgress = 0; + if(numConnections > 20) { /* if we use many (whatever this means, 20 is randomly picked) * connections, we need to make sure we have a high enough @@ -447,8 +657,8 @@ int main(int argc, char *argv[]) exit(1); } - if(sendMessages() != 0) { - printf("error sending messages\n"); + if(runTests() != 0) { + printf("error running tests\n"); exit(1); } @@ -457,7 +667,8 @@ int main(int argc, char *argv[]) if(nConnDrops > 0) printf("-D option initiated %ld connection closures\n", nConnDrops); - printf("End of tcpflood Run\n"); + if(!bSilent) + printf("End of tcpflood Run\n"); exit(ret); } |