diff options
Diffstat (limited to 'runtime')
45 files changed, 2181 insertions, 1042 deletions
diff --git a/runtime/Makefile.am b/runtime/Makefile.am index 34384bea..230be69e 100644 --- a/runtime/Makefile.am +++ b/runtime/Makefile.am @@ -69,6 +69,8 @@ librsyslog_la_SOURCES = \ prop.h \ ratelimit.c \ ratelimit.h \ + lookup.c \ + lookup.h \ cfsysline.c \ cfsysline.h \ sd-daemon.c \ diff --git a/runtime/batch.h b/runtime/batch.h index 2ec07670..b0aa8574 100644 --- a/runtime/batch.h +++ b/runtime/batch.h @@ -46,17 +46,6 @@ typedef unsigned char batch_state_t; */ struct batch_obj_s { msg_t *pMsg; - /* work variables for action processing; these are reused for each action (or block of - * actions) - */ - sbool bPrevWasSuspended; - /* following are caches to save allocs if not absolutely necessary */ - uchar *staticActStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for strings */ - /* a cache to save malloc(), if not absolutely necessary */ - void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for anything else */ - size_t staticLenStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; - /* and the same for the message length (if used) */ - /* end action work variables */ }; /* the batch @@ -77,11 +66,7 @@ struct batch_s { int maxElem; /* maximum number of elements that this batch supports */ int nElem; /* actual number of element in this entry */ int nElemDeq; /* actual number of elements dequeued (and thus to be deleted) - see comment above! */ - int iDoneUpTo; /* all messages below this index have state other than RDY */ qDeqID deqID; /* ID of dequeue operation that generated this batch */ - int *pbShutdownImmediate;/* end processing of this batch immediately if set to 1 */ - sbool *active; /* which messages are active for processing, NULL=all */ - sbool bSingleRuleset; /* do all msgs of this batch use a single ruleset? */ batch_obj_t *pElem; /* batch elements */ batch_state_t *eltState;/* state (array!) for individual objects. NOTE: we have moved this out of batch_obj_t because we @@ -93,24 +78,6 @@ struct batch_s { }; -/* some inline functions (we may move this off to an object .. or not) */ -static inline void -batchSetSingleRuleset(batch_t *pBatch, sbool val) { - pBatch->bSingleRuleset = val; -} - -/* get the batches ruleset (if we have a single ruleset) */ -static inline ruleset_t* -batchGetRuleset(batch_t *pBatch) { - return (pBatch->nElem > 0) ? pBatch->pElem[0].pMsg->pRuleset : NULL; -} - -/* get the ruleset of a specifc element of the batch (index not verified!) */ -static inline ruleset_t* -batchElemGetRuleset(batch_t *pBatch, int i) { - return pBatch->pElem[i].pMsg->pRuleset; -} - /* get number of msgs for this batch */ static inline int batchNumMsgs(batch_t *pBatch) { @@ -134,8 +101,7 @@ batchSetElemState(batch_t *pBatch, int i, batch_state_t newState) { */ static inline int batchIsValidElem(batch_t *pBatch, int i) { - return( (pBatch->eltState[i] != BATCH_STATE_DISC) - && (pBatch->active == NULL || pBatch->active[i])); + return(pBatch->eltState[i] != BATCH_STATE_DISC); } @@ -152,7 +118,7 @@ batchFree(batch_t *pBatch) { /* staticActParams MUST be freed immediately (if required), * so we do not need to do that! */ - free(pBatch->pElem[i].staticActStrings[j]); + //TODO: do this in wti! free(pBatch->pElem[i].staticActStrings[j]); } } free(pBatch->pElem); @@ -167,11 +133,9 @@ batchFree(batch_t *pBatch) { static inline rsRetVal batchInit(batch_t *pBatch, int maxElem) { DEFiRet; - pBatch->iDoneUpTo = 0; pBatch->maxElem = maxElem; CHKmalloc(pBatch->pElem = calloc((size_t)maxElem, sizeof(batch_obj_t))); CHKmalloc(pBatch->eltState = calloc((size_t)maxElem, sizeof(batch_state_t))); - // TODO: replace calloc by inidividual writes? finalize_it: RETiRet; } diff --git a/runtime/conf.c b/runtime/conf.c index c01715cb..83931bca 100644 --- a/runtime/conf.c +++ b/runtime/conf.c @@ -486,111 +486,6 @@ rsRetVal DecodePRIFilter(uchar *pline, uchar pmask[]) } -/* Helper to cfline(). This function takes the filter part of a property - * based filter and decodes it. It processes the line up to the beginning - * of the action part. A pointer to that beginnig is passed back to the caller. - * rgerhards 2005-09-15 - */ -rsRetVal DecodePropFilter(uchar *pline, struct cnfstmt *stmt) -{ - rsParsObj *pPars; - cstr_t *pCSCompOp; - cstr_t *pCSPropName; - rsRetVal iRet; - int iOffset; /* for compare operations */ - - ASSERT(pline != NULL); - - dbgprintf("Decoding property-based filter '%s'\n", pline); - - /* create parser object starting with line string without leading colon */ - if((iRet = rsParsConstructFromSz(&pPars, pline+1)) != RS_RET_OK) { - errmsg.LogError(0, iRet, "Error %d constructing parser object - ignoring selector", iRet); - return(iRet); - } - - /* read property */ - iRet = parsDelimCStr(pPars, &pCSPropName, ',', 1, 1, 1); - if(iRet != RS_RET_OK) { - errmsg.LogError(0, iRet, "error %d parsing filter property - ignoring selector", iRet); - rsParsDestruct(pPars); - return(iRet); - } - iRet = propNameToID(pCSPropName, &stmt->d.s_propfilt.propID); - if(iRet != RS_RET_OK) { - errmsg.LogError(0, iRet, "error %d parsing filter property - ignoring selector", iRet); - rsParsDestruct(pPars); - return(iRet); - } - if(stmt->d.s_propfilt.propID == PROP_CEE) { - /* in CEE case, we need to preserve the actual property name */ - if((stmt->d.s_propfilt.propName = - es_newStrFromBuf((char*)cstrGetSzStrNoNULL(pCSPropName)+2, cstrLen(pCSPropName)-2)) == NULL) { - cstrDestruct(&pCSPropName); - return(RS_RET_ERR); - } - } - cstrDestruct(&pCSPropName); - - /* read operation */ - iRet = parsDelimCStr(pPars, &pCSCompOp, ',', 1, 1, 1); - if(iRet != RS_RET_OK) { - errmsg.LogError(0, iRet, "error %d compare operation property - ignoring selector", iRet); - rsParsDestruct(pPars); - return(iRet); - } - - /* we now first check if the condition is to be negated. To do so, we first - * must make sure we have at least one char in the param and then check the - * first one. - * rgerhards, 2005-09-26 - */ - if(rsCStrLen(pCSCompOp) > 0) { - if(*rsCStrGetBufBeg(pCSCompOp) == '!') { - stmt->d.s_propfilt.isNegated = 1; - iOffset = 1; /* ignore '!' */ - } else { - stmt->d.s_propfilt.isNegated = 0; - iOffset = 0; - } - } else { - stmt->d.s_propfilt.isNegated = 0; - iOffset = 0; - } - - if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (uchar*) "contains", 8)) { - stmt->d.s_propfilt.operation = FIOP_CONTAINS; - } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (uchar*) "isequal", 7)) { - stmt->d.s_propfilt.operation = FIOP_ISEQUAL; - } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (uchar*) "isempty", 7)) { - stmt->d.s_propfilt.operation = FIOP_ISEMPTY; - } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (uchar*) "startswith", 10)) { - stmt->d.s_propfilt.operation = FIOP_STARTSWITH; - } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (unsigned char*) "regex", 5)) { - stmt->d.s_propfilt.operation = FIOP_REGEX; - } else if(!rsCStrOffsetSzStrCmp(pCSCompOp, iOffset, (unsigned char*) "ereregex", 8)) { - stmt->d.s_propfilt.operation = FIOP_EREREGEX; - } else { - errmsg.LogError(0, NO_ERRCODE, "error: invalid compare operation '%s' - ignoring selector", - (char*) rsCStrGetSzStrNoNULL(pCSCompOp)); - return(RS_RET_ERR); - } - rsCStrDestruct(&pCSCompOp); /* no longer needed */ - - if(stmt->d.s_propfilt.operation != FIOP_ISEMPTY) { - /* read compare value */ - iRet = parsQuotedCStr(pPars, &stmt->d.s_propfilt.pCSCompValue); - if(iRet != RS_RET_OK) { - errmsg.LogError(0, iRet, "error %d compare value property - ignoring selector", iRet); - rsParsDestruct(pPars); - return(iRet); - } - } - - return rsParsDestruct(pPars); -} - - /* process the action part of a selector line * rgerhards, 2007-08-01 */ @@ -623,12 +518,10 @@ rsRetVal cflineDoAction(rsconf_t *conf, uchar **p, action_t **ppAction) bHadWarning = 1; iRet = RS_RET_OK; } - if(iRet == RS_RET_OK || iRet == RS_RET_SUSPENDED) { - if((iRet = addAction(&pAction, pMod, pModData, pOMSR, NULL, NULL, - (iRet == RS_RET_SUSPENDED)? 1 : 0)) == RS_RET_OK) { + if(iRet == RS_RET_OK) { + if((iRet = addAction(&pAction, pMod, pModData, pOMSR, NULL, NULL)) == RS_RET_OK) { /* here check if the module is compatible with select features * (currently, we have no such features!) */ - pAction->eState = ACT_STATE_RDY; /* action is enabled */ conf->actions.nbrActions++; /* one more active action! */ } break; diff --git a/runtime/conf.h b/runtime/conf.h index a1bb51ad..cb956678 100644 --- a/runtime/conf.h +++ b/runtime/conf.h @@ -63,7 +63,6 @@ rsRetVal cflineParseTemplateName(uchar** pp, omodStringRequest_t *pOMSR, int iEn rsRetVal cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pOMSR, int iEntry, int iTplOpts, uchar *pszTpl); rsRetVal DecodePRIFilter(uchar *pline, uchar pmask[]); -rsRetVal DecodePropFilter(uchar *pline, struct cnfstmt *stmt); rsRetVal cflineDoAction(rsconf_t *conf, uchar **p, action_t **ppAction); extern EHostnameCmpMode eDfltHostnameCmpMode; extern cstr_t *pDfltHostnameCmp; diff --git a/runtime/cryprov.h b/runtime/cryprov.h index 005b33f7..2742a4a5 100644 --- a/runtime/cryprov.h +++ b/runtime/cryprov.h @@ -24,14 +24,25 @@ #ifndef INCLUDED_CRYPROV_H #define INCLUDED_CRYPROV_H +/* we unfortunately need to have two different param names depending on the + * context in which parameters are set. Other than (re/over)engineering the core + * interface, we just define some values to keep track of that. + */ +#define CRYPROV_PARAMTYPE_REGULAR 0 +#define CRYPROV_PARAMTYPE_DISK 1 + /* interface */ BEGINinterface(cryprov) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Construct)(void *ppThis); - rsRetVal (*SetCnfParam)(void *ppThis, struct nvlst *lst); + rsRetVal (*SetCnfParam)(void *ppThis, struct nvlst *lst, int paramType); rsRetVal (*Destruct)(void *ppThis); - rsRetVal (*OnFileOpen)(void *pThis, uchar *fn, void *pFileInstData); + rsRetVal (*OnFileOpen)(void *pThis, uchar *fn, void *pFileInstData, char openMode); rsRetVal (*Encrypt)(void *pFileInstData, uchar *buf, size_t *lenBuf); + rsRetVal (*Decrypt)(void *pFileInstData, uchar *buf, size_t *lenBuf); rsRetVal (*OnFileClose)(void *pFileInstData, off64_t offsLogfile); + rsRetVal (*DeleteStateFiles)(uchar *logfn); + rsRetVal (*GetBytesLeftInBlock)(void *pFileInstData, ssize_t *left); + void (*SetDeleteOnClose)(void *pFileInstData, int val); ENDinterface(cryprov) -#define cryprovCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define cryprovCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */ #endif /* #ifndef INCLUDED_CRYPROV_H */ diff --git a/runtime/datetime.c b/runtime/datetime.c index 841ff625..3d50238c 100644 --- a/runtime/datetime.c +++ b/runtime/datetime.c @@ -626,6 +626,15 @@ finalize_it: RETiRet; } +void +applyDfltTZ(struct syslogTime *pTime, char *tz) +{ + pTime->OffsetMode = tz[0]; + pTime->OffsetHour = (tz[1] - '0') * 10 + (tz[2] - '0'); + pTime->OffsetMinute = (tz[4] - '0') * 10 + (tz[5] - '0'); + +} + /******************************************************************* * END CODE-LIBLOGGING * *******************************************************************/ diff --git a/runtime/datetime.h b/runtime/datetime.h index 9f3611e1..72c3a97f 100644 --- a/runtime/datetime.h +++ b/runtime/datetime.h @@ -62,5 +62,6 @@ ENDinterface(datetime) /* prototypes */ PROTOTYPEObj(datetime); +void applyDfltTZ(struct syslogTime *pTime, char *tz); #endif /* #ifndef INCLUDED_DATETIME_H */ diff --git a/runtime/glbl.c b/runtime/glbl.c index 41d56c2c..e1d4e92d 100644 --- a/runtime/glbl.c +++ b/runtime/glbl.c @@ -61,6 +61,7 @@ DEFobjCurrIf(net) * For this object, these variables are obviously what makes the "meat" of the * class... */ +static struct cnfobj *mainqCnfObj = NULL;/* main queue object, to be used later in startup sequence */ static uchar *pszWorkDir = NULL; static int bOptimizeUniProc = 1; /* enable uniprocessor optimizations */ static int bParseHOSTNAMEandTAG = 1; /* parser modification (based on startup params!) */ @@ -91,6 +92,7 @@ static DEF_ATOMIC_HELPER_MUT(mutTerminateInputs); #ifdef USE_UNLIMITED_SELECT static int iFdSetSize = howmany(FD_SETSIZE, __NFDBITS) * sizeof (fd_mask); /* size of select() bitmask in bytes */ #endif +static uchar *SourceIPofLocalClient = NULL; /* [ar] Source IP for local client to be used on multihomed host */ /* tables for interfacing with the v6 config system */ @@ -138,6 +140,7 @@ static dataType Get##nameFunc(void) \ SIMP_PROP(ParseHOSTNAMEandTAG, bParseHOSTNAMEandTAG, int) SIMP_PROP(OptimizeUniProc, bOptimizeUniProc, int) SIMP_PROP(PreserveFQDN, bPreserveFQDN, int) +SIMP_PROP(mainqCnfObj, mainqCnfObj, struct cnfobj *) SIMP_PROP(MaxLine, iMaxLine, int) SIMP_PROP(DefPFFamily, iDefPFFamily, int) /* note that in the future we may check the family argument */ SIMP_PROP(DropMalPTRMsgs, bDropMalPTRMsgs, int) @@ -506,6 +509,23 @@ GetDfltNetstrmDrvrCertFile(void) } +/* [ar] Source IP for local client to be used on multihomed host */ +static rsRetVal +SetSourceIPofLocalClient(uchar *newname) +{ + if(SourceIPofLocalClient != NULL) { + free(SourceIPofLocalClient); } + SourceIPofLocalClient = newname; + return RS_RET_OK; +} + +static uchar* +GetSourceIPofLocalClient(void) +{ + return(SourceIPofLocalClient); +} + + /* queryInterface function * rgerhards, 2008-02-21 */ @@ -526,6 +546,8 @@ CODESTARTobjQueryInterface(glbl) pIf->GetLocalHostIP = GetLocalHostIP; pIf->SetGlobalInputTermination = SetGlobalInputTermination; pIf->GetGlobalInputTermState = GetGlobalInputTermState; + pIf->GetSourceIPofLocalClient = GetSourceIPofLocalClient; /* [ar] */ + pIf->SetSourceIPofLocalClient = SetSourceIPofLocalClient; /* [ar] */ #define SIMP_PROP(name) \ pIf->Get##name = Get##name; \ pIf->Set##name = Set##name; @@ -537,6 +559,7 @@ CODESTARTobjQueryInterface(glbl) SIMP_PROP(DropMalPTRMsgs); SIMP_PROP(Option_DisallowWarning); SIMP_PROP(DisableDNS); + SIMP_PROP(mainqCnfObj); SIMP_PROP(LocalFQDNName) SIMP_PROP(LocalHostName) SIMP_PROP(LocalDomain) @@ -587,6 +610,8 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a void glblPrepCnf(void) { + free(mainqCnfObj); + mainqCnfObj = NULL; free(cnfparamvals); cnfparamvals = NULL; } @@ -604,6 +629,34 @@ glblProcessCnf(struct cnfobj *o) cnfparamsPrint(¶mblk, cnfparamvals); } +/* Set mainq parameters. Note that when this is not called, we'll use the + * legacy parameter config. mainq parameters can only be set once. + */ +void +glblProcessMainQCnf(struct cnfobj *o) +{ + if(mainqCnfObj == NULL) { + mainqCnfObj = o; + } else { + errmsg.LogError(0, RS_RET_ERR, "main_queue() object can only be specified " + "once - all but first ignored\n"); + } +} + +/* destruct the main q cnf object after it is no longer needed. This is + * also used to do some final checks. + */ +void +glblDestructMainqCnfObj() +{ + /* Only destruct if not NULL! */ + if (mainqCnfObj != NULL) { + nvlstChkUnused(mainqCnfObj->nvlst); + } + cnfobjDestruct(mainqCnfObj); + mainqCnfObj = NULL; +} + void glblDoneLoadCnf(void) { diff --git a/runtime/glbl.h b/runtime/glbl.h index e95e48f7..4b1cd9f8 100644 --- a/runtime/glbl.h +++ b/runtime/glbl.h @@ -52,6 +52,7 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */ SIMP_PROP(Option_DisallowWarning, int) SIMP_PROP(DisableDNS, int) SIMP_PROP(LocalFQDNName, uchar*) + SIMP_PROP(mainqCnfObj, struct cnfobj*) SIMP_PROP(LocalHostName, uchar*) SIMP_PROP(LocalDomain, uchar*) SIMP_PROP(StripDomains, char**) @@ -81,6 +82,8 @@ BEGINinterface(glbl) /* name must also be changed in ENDinterface macro! */ /* next change is v9! */ /* v8 - 2012-03-21 */ prop_t* (*GetLocalHostIP)(void); + uchar* (*GetSourceIPofLocalClient)(void); /* [ar] */ + rsRetVal (*SetSourceIPofLocalClient)(uchar*); /* [ar] */ #undef SIMP_PROP ENDinterface(glbl) #define glblCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */ @@ -94,6 +97,8 @@ static inline void glblSetOurPid(pid_t pid) { glbl_ourpid = pid; } void glblPrepCnf(void); void glblProcessCnf(struct cnfobj *o); +void glblProcessMainQCnf(struct cnfobj *o); +void glblDestructMainqCnfObj(); void glblDoneLoadCnf(void); #endif /* #ifndef GLBL_H_INCLUDED */ diff --git a/runtime/libgcry.c b/runtime/libgcry.c index 51c10af4..4772cf47 100644 --- a/runtime/libgcry.c +++ b/runtime/libgcry.c @@ -49,8 +49,12 @@ #include <errno.h> #include "rsyslog.h" +#include "srUtils.h" #include "libgcry.h" +#define READBUF_SIZE 4096 /* size of the read buffer */ + +static rsRetVal rsgcryBlkBegin(gcryfile gf); static rsRetVal eiWriteRec(gcryfile gf, char *recHdr, size_t lenRecHdr, char *buf, size_t lenBuf) @@ -90,19 +94,66 @@ finalize_it: RETiRet; } +static rsRetVal +eiRead(gcryfile gf) +{ + ssize_t nRead; + DEFiRet; + + if(gf->readBuf == NULL) { + CHKmalloc(gf->readBuf = malloc(READBUF_SIZE)); + } + + nRead = read(gf->fd, gf->readBuf, READBUF_SIZE); + if(nRead <= 0) { /* TODO: provide specific EOF case? */ + ABORT_FINALIZE(RS_RET_ERR); + } + gf->readBufMaxIdx = (int16_t) nRead; + gf->readBufIdx = 0; + +finalize_it: + RETiRet; +} + + +/* returns EOF on any kind of error */ +static int +eiReadChar(gcryfile gf) +{ + int c; + + if(gf->readBufIdx >= gf->readBufMaxIdx) { + if(eiRead(gf) != RS_RET_OK) { + c = EOF; + goto finalize_it; + } + } + c = gf->readBuf[gf->readBufIdx++]; +finalize_it: + return c; +} + static rsRetVal eiCheckFiletype(gcryfile gf) { char hdrBuf[128]; size_t toRead, didRead; + sbool bNeedClose = 0; DEFiRet; - CHKiRet(eiOpenRead(gf)); + if(gf->fd == -1) { + bNeedClose = 1; + CHKiRet(eiOpenRead(gf)); + } + if(Debug) memset(hdrBuf, 0, sizeof(hdrBuf)); /* for dbgprintf below! */ toRead = sizeof("FILETYPE:")-1 + sizeof(RSGCRY_FILETYPE_NAME)-1 + 1; didRead = read(gf->fd, hdrBuf, toRead); - close(gf->fd); + if(bNeedClose) { + close(gf->fd); + gf->fd = -1; + } DBGPRINTF("eiCheckFiletype read %d bytes: '%s'\n", didRead, hdrBuf); if( didRead != toRead || strncmp(hdrBuf, "FILETYPE:" RSGCRY_FILETYPE_NAME "\n", toRead)) @@ -111,6 +162,98 @@ finalize_it: RETiRet; } +/* rectype/value must be EIF_MAX_*_LEN+1 long! + * returns 0 on success or something else on error/EOF + */ +static rsRetVal +eiGetRecord(gcryfile gf, char *rectype, char *value) +{ + unsigned short i, j; + int c; + DEFiRet; + + c = eiReadChar(gf); + if(c == EOF) { ABORT_FINALIZE(RS_RET_NO_DATA); } + for(i = 0 ; i < EIF_MAX_RECTYPE_LEN ; ++i) { + if(c == ':' || c == EOF) + break; + rectype[i] = c; + c = eiReadChar(gf); + } + if(c != ':') { ABORT_FINALIZE(RS_RET_ERR); } + rectype[i] = '\0'; + j = 0; + for(++i ; i < EIF_MAX_VALUE_LEN ; ++i, ++j) { + c = eiReadChar(gf); + if(c == '\n' || c == EOF) + break; + value[j] = c; + } + if(c != '\n') { ABORT_FINALIZE(RS_RET_ERR); } + value[j] = '\0'; +finalize_it: + RETiRet; +} + +static rsRetVal +eiGetIV(gcryfile gf, uchar *iv, size_t leniv) +{ + char rectype[EIF_MAX_RECTYPE_LEN+1]; + char value[EIF_MAX_VALUE_LEN+1]; + size_t valueLen; + unsigned short i, j; + unsigned char nibble; + DEFiRet; + + CHKiRet(eiGetRecord(gf, rectype, value)); + if(strcmp(rectype, "IV")) { + DBGPRINTF("no IV record found when expected, record type " + "seen is '%s'\n", rectype); + ABORT_FINALIZE(RS_RET_ERR); + } + valueLen = strlen(value); + if(valueLen/2 != leniv) { + DBGPRINTF("length of IV is %d, expected %d\n", + valueLen/2, leniv); + ABORT_FINALIZE(RS_RET_ERR); + } + + for(i = j = 0 ; i < valueLen ; ++i) { + if(value[i] >= '0' && value[i] <= '9') + nibble = value[i] - '0'; + else if(value[i] >= 'a' && value[i] <= 'f') + nibble = value[i] - 'a' + 10; + else { + DBGPRINTF("invalid IV '%s'\n", value); + ABORT_FINALIZE(RS_RET_ERR); + } + if(i % 2 == 0) + iv[j] = nibble << 4; + else + iv[j++] |= nibble; + } +finalize_it: + RETiRet; +} + +static rsRetVal +eiGetEND(gcryfile gf, off64_t *offs) +{ + char rectype[EIF_MAX_RECTYPE_LEN+1]; + char value[EIF_MAX_VALUE_LEN+1]; + DEFiRet; + + CHKiRet(eiGetRecord(gf, rectype, value)); + if(strcmp(rectype, "END")) { + DBGPRINTF("no END record found when expected, record type " + "seen is '%s'\n", rectype); + ABORT_FINALIZE(RS_RET_ERR); + } + *offs = atoll(value); +finalize_it: + RETiRet; +} + static rsRetVal eiOpenAppend(gcryfile gf) { @@ -177,13 +320,55 @@ eiClose(gcryfile gf, off64_t offsLogfile) size_t len; if(gf->fd == -1) return; - /* 2^64 is 20 digits, so the snprintf buffer is large enough */ - len = snprintf(offs, sizeof(offs), "%lld", offsLogfile); - eiWriteRec(gf, "END:", 4, offs, len); + if(gf->openMode == 'w') { + /* 2^64 is 20 digits, so the snprintf buffer is large enough */ + len = snprintf(offs, sizeof(offs), "%lld", offsLogfile); + eiWriteRec(gf, "END:", 4, offs, len); + } + gcry_cipher_close(gf->chd); + free(gf->readBuf); close(gf->fd); + gf->fd = -1; DBGPRINTF("encryption info file %s: closed\n", gf->eiName); } +/* this returns the number of bytes left inside the block or -1, if the block + * size is unbounded. The function automatically handles end-of-block and begins + * to read the next block in this case. + */ +rsRetVal +gcryfileGetBytesLeftInBlock(gcryfile gf, ssize_t *left) +{ + DEFiRet; + if(gf->bytesToBlkEnd == 0) { + DBGPRINTF("libgcry: end of current crypto block\n"); + gcry_cipher_close(gf->chd); + CHKiRet(rsgcryBlkBegin(gf)); + } + *left = gf->bytesToBlkEnd; +finalize_it: + // TODO: remove once this code is sufficiently well-proven + DBGPRINTF("gcryfileGetBytesLeftInBlock returns %lld, iRet %d\n", (long long) *left, iRet); + RETiRet; +} + +/* this is a special functon for use by the rsyslog disk queue subsystem. It + * needs to have the capability to delete state when a queue file is rolled + * over. This simply generates the file name and deletes it. It must take care + * of "all" state files, which currently happens to be a single one. + */ +rsRetVal +gcryfileDeleteState(uchar *logfn) +{ + char fn[MAXFNAME+1]; + DEFiRet; + snprintf(fn, sizeof(fn), "%s%s", logfn, ENCINFO_SUFFIX); + fn[MAXFNAME] = '\0'; /* be on save side */ + DBGPRINTF("crypto provider deletes state file '%s' on request\n", fn); + unlink(fn); + RETiRet; +} + static rsRetVal gcryfileConstruct(gcryctx ctx, gcryfile *pgf, uchar *logfn) { @@ -193,6 +378,7 @@ gcryfileConstruct(gcryctx ctx, gcryfile *pgf, uchar *logfn) CHKmalloc(gf = calloc(1, sizeof(struct gcryfile_s))); gf->ctx = ctx; + gf->fd = -1; snprintf(fn, sizeof(fn), "%s%s", logfn, ENCINFO_SUFFIX); fn[MAXFNAME] = '\0'; /* be on save side */ gf->eiName = (uchar*) strdup(fn); @@ -219,7 +405,12 @@ gcryfileDestruct(gcryfile gf, off64_t offsLogfile) if(gf == NULL) goto done; + DBGPRINTF("libgcry: close file %s\n", gf->eiName); eiClose(gf, offsLogfile); + if(gf->bDeleteOnClose) { + DBGPRINTF("unlink file '%s' due to bDeleteOnClose set\n", gf->eiName); + unlink((char*)gf->eiName); + } free(gf->eiName); free(gf); done: return r; @@ -246,13 +437,13 @@ addPadding(gcryfile pF, uchar *buf, size_t *plen) } static inline void -removePadding(char *buf, size_t *plen) +removePadding(uchar *buf, size_t *plen) { unsigned len = (unsigned) *plen; unsigned iSrc, iDst; - char *frstNUL; + uchar *frstNUL; - frstNUL = strchr(buf, 0x00); + frstNUL = (uchar*)strchr((char*)buf, 0x00); if(frstNUL == NULL) goto done; iDst = iSrc = frstNUL - buf; @@ -343,53 +534,123 @@ seedIV(gcryfile gf, uchar **iv) } } -rsRetVal -rsgcryInitCrypt(gcryctx ctx, gcryfile *pgf, uchar *fname) +static inline rsRetVal +readIV(gcryfile gf, uchar **iv) { - gcry_error_t gcryError; - gcryfile gf = NULL; - uchar *iv = NULL; + rsRetVal localRet; DEFiRet; - CHKiRet(gcryfileConstruct(ctx, &gf, fname)); + if(gf->fd == -1) { + while(gf->fd == -1) { + localRet = eiOpenRead(gf); + if(localRet == RS_RET_EI_NO_EXISTS) { + /* wait until it is created */ + srSleep(0, 10000); + } else { + CHKiRet(localRet); + } + } + CHKiRet(eiCheckFiletype(gf)); + } + *iv = malloc(gf->blkLength); /* do NOT zero-out! */ + CHKiRet(eiGetIV(gf, *iv, (size_t) gf->blkLength)); +finalize_it: + RETiRet; +} + +/* this tries to read the END record. HOWEVER, no such record may be + * present, which is the case if we handle a currently-written to queue + * file. On the other hand, the queue file may contain multiple blocks. So + * what we do is try to see if there is a block end or not - and set the + * status accordingly. Note that once we found no end-of-block, we will never + * retry. This is because that case can never happen under current queue + * implementations. -- gerhards, 2013-05-16 + */ +static inline rsRetVal +readBlkEnd(gcryfile gf) +{ + off64_t blkEnd; + DEFiRet; + + iRet = eiGetEND(gf, &blkEnd); + if(iRet == RS_RET_OK) { + gf->bytesToBlkEnd = (ssize_t) blkEnd; + } else if(iRet == RS_RET_NO_DATA) { + gf->bytesToBlkEnd = -1; + } else { + FINALIZE; + } + +finalize_it: + RETiRet; +} - gf->blkLength = gcry_cipher_get_algo_blklen(ctx->algo); - gcryError = gcry_cipher_open(&gf->chd, ctx->algo, ctx->mode, 0); +/* Read the block begin metadata and set our state variables accordingly. Can also + * be used to init the first block in write case. + */ +static rsRetVal +rsgcryBlkBegin(gcryfile gf) +{ + gcry_error_t gcryError; + uchar *iv = NULL; + DEFiRet; + + gcryError = gcry_cipher_open(&gf->chd, gf->ctx->algo, gf->ctx->mode, 0); if (gcryError) { - dbgprintf("gcry_cipher_open failed: %s/%s\n", - gcry_strsource(gcryError), - gcry_strerror(gcryError)); + DBGPRINTF("gcry_cipher_open failed: %s/%s\n", + gcry_strsource(gcryError), gcry_strerror(gcryError)); ABORT_FINALIZE(RS_RET_ERR); } gcryError = gcry_cipher_setkey(gf->chd, gf->ctx->key, gf->ctx->keyLen); if (gcryError) { - dbgprintf("gcry_cipher_setkey failed: %s/%s\n", - gcry_strsource(gcryError), - gcry_strerror(gcryError)); + DBGPRINTF("gcry_cipher_setkey failed: %s/%s\n", + gcry_strsource(gcryError), gcry_strerror(gcryError)); ABORT_FINALIZE(RS_RET_ERR); } - seedIV(gf, &iv); + if(gf->openMode == 'r') { + readIV(gf, &iv); + readBlkEnd(gf); + } else { + seedIV(gf, &iv); + } + gcryError = gcry_cipher_setiv(gf->chd, iv, gf->blkLength); if (gcryError) { - dbgprintf("gcry_cipher_setiv failed: %s/%s\n", - gcry_strsource(gcryError), - gcry_strerror(gcryError)); + DBGPRINTF("gcry_cipher_setiv failed: %s/%s\n", + gcry_strsource(gcryError), gcry_strerror(gcryError)); ABORT_FINALIZE(RS_RET_ERR); } - CHKiRet(eiOpenAppend(gf)); - CHKiRet(eiWriteIV(gf, iv)); - *pgf = gf; + + if(gf->openMode == 'w') { + CHKiRet(eiOpenAppend(gf)); + CHKiRet(eiWriteIV(gf, iv)); + } finalize_it: free(iv); + RETiRet; +} + +rsRetVal +rsgcryInitCrypt(gcryctx ctx, gcryfile *pgf, uchar *fname, char openMode) +{ + gcryfile gf = NULL; + DEFiRet; + + CHKiRet(gcryfileConstruct(ctx, &gf, fname)); + gf->openMode = openMode; + gf->blkLength = gcry_cipher_get_algo_blklen(ctx->algo); + CHKiRet(rsgcryBlkBegin(gf)); + *pgf = gf; +finalize_it: if(iRet != RS_RET_OK && gf != NULL) gcryfileDestruct(gf, -1); RETiRet; } -int +rsRetVal rsgcryEncrypt(gcryfile pF, uchar *buf, size_t *len) { int gcryError; @@ -410,6 +671,35 @@ finalize_it: RETiRet; } +/* TODO: handle multiple blocks + * test-read END record; if present, store offset, else unbounded (current active block) + * when decrypting, check if bound is reached. If yes, split into two blocks, get new IV for + * second one. + */ +rsRetVal +rsgcryDecrypt(gcryfile pF, uchar *buf, size_t *len) +{ + gcry_error_t gcryError; + DEFiRet; + + if(pF->bytesToBlkEnd != -1) + pF->bytesToBlkEnd -= *len; + gcryError = gcry_cipher_decrypt(pF->chd, buf, *len, NULL, 0); + if(gcryError) { + DBGPRINTF("gcry_cipher_decrypt failed: %s/%s\n", + gcry_strsource(gcryError), + gcry_strerror(gcryError)); + ABORT_FINALIZE(RS_RET_ERR); + } + removePadding(buf, len); + // TODO: remove dbgprintf once things are sufficently stable -- rgerhards, 2013-05-16 + dbgprintf("libgcry: decrypted, bytesToBlkEnd %lld, buffer is now '%50.50s'\n", (long long) pF->bytesToBlkEnd, buf); + +finalize_it: + RETiRet; +} + + /* module-init dummy for potential later use */ int diff --git a/runtime/libgcry.h b/runtime/libgcry.h index 692ce408..ae5a6735 100644 --- a/runtime/libgcry.h +++ b/runtime/libgcry.h @@ -38,7 +38,15 @@ struct gcryfile_s { size_t blkLength; /* size of low-level crypto block */ uchar *eiName; /* name of .encinfo file */ int fd; /* descriptor of .encinfo file (-1 if not open) */ + char openMode; /* 'r': read, 'w': write */ gcryctx ctx; + uchar *readBuf; + int16_t readBufIdx; + int16_t readBufMaxIdx; + int8_t bDeleteOnClose; /* for queue support, similar to stream subsys */ + ssize_t bytesToBlkEnd; /* number of bytes remaining in current crypto block + -1 means -> no end (still being writen to, queue files), + 0 means -> end of block, new one must be started. */ }; int gcryGetKeyFromFile(char *fn, char **key, unsigned *keylen); @@ -50,9 +58,12 @@ rsRetVal rsgcrySetAlgo(gcryctx ctx, uchar *modename); gcryctx gcryCtxNew(void); void rsgcryCtxDel(gcryctx ctx); int gcryfileDestruct(gcryfile gf, off64_t offsLogfile); -rsRetVal rsgcryInitCrypt(gcryctx ctx, gcryfile *pgf, uchar *fname); -int rsgcryEncrypt(gcryfile pF, uchar *buf, size_t *len); +rsRetVal rsgcryInitCrypt(gcryctx ctx, gcryfile *pgf, uchar *fname, char openMode); +rsRetVal rsgcryEncrypt(gcryfile pF, uchar *buf, size_t *len); +rsRetVal rsgcryDecrypt(gcryfile pF, uchar *buf, size_t *len); int gcryGetKeyFromProg(char *cmd, char **key, unsigned *keylen); +rsRetVal gcryfileDeleteState(uchar *fn); +rsRetVal gcryfileGetBytesLeftInBlock(gcryfile gf, ssize_t *left); /* error states */ #define RSGCRYE_EI_OPEN 1 /* error opening .encinfo file */ @@ -63,6 +74,14 @@ int gcryGetKeyFromProg(char *cmd, char **key, unsigned *keylen); #define RSGCRY_FILETYPE_NAME "rsyslog-enrcyption-info" #define ENCINFO_SUFFIX ".encinfo" +/* Note: gf may validly be NULL, e.g. if file has not yet been opened! */ +static inline void +gcryfileSetDeleteOnClose(gcryfile gf, int val) +{ + if(gf != NULL) + gf->bDeleteOnClose = val; +} + static inline int rsgcryAlgoname2Algo(char *algoname) { if(!strcmp((char*)algoname, "3DES")) return GCRY_CIPHER_3DES; diff --git a/runtime/lmcry_gcry.c b/runtime/lmcry_gcry.c index 0a9b94bc..9a0c0072 100644 --- a/runtime/lmcry_gcry.c +++ b/runtime/lmcry_gcry.c @@ -43,17 +43,30 @@ DEFobjCurrIf(errmsg) DEFobjCurrIf(glbl) /* tables for interfacing with the v6 config system */ -static struct cnfparamdescr cnfpdescr[] = { +static struct cnfparamdescr cnfpdescrRegular[] = { { "cry.key", eCmdHdlrGetWord, 0 }, { "cry.keyfile", eCmdHdlrGetWord, 0 }, { "cry.keyprogram", eCmdHdlrGetWord, 0 }, { "cry.mode", eCmdHdlrGetWord, 0 }, /* CBC, ECB, etc */ { "cry.algo", eCmdHdlrGetWord, 0 } }; -static struct cnfparamblk pblk = +static struct cnfparamblk pblkRegular = { CNFPARAMBLK_VERSION, - sizeof(cnfpdescr)/sizeof(struct cnfparamdescr), - cnfpdescr + sizeof(cnfpdescrRegular)/sizeof(struct cnfparamdescr), + cnfpdescrRegular + }; + +static struct cnfparamdescr cnfpdescrQueue[] = { + { "queue.cry.key", eCmdHdlrGetWord, 0 }, + { "queue.cry.keyfile", eCmdHdlrGetWord, 0 }, + { "queue.cry.keyprogram", eCmdHdlrGetWord, 0 }, + { "queue.cry.mode", eCmdHdlrGetWord, 0 }, /* CBC, ECB, etc */ + { "queue.cry.algo", eCmdHdlrGetWord, 0 } +}; +static struct cnfparamblk pblkQueue = + { CNFPARAMBLK_VERSION, + sizeof(cnfpdescrQueue)/sizeof(struct cnfparamdescr), + cnfpdescrQueue }; @@ -85,7 +98,7 @@ ENDobjDestruct(lmcry_gcry) * Defaults are expected to have been set during construction. */ static rsRetVal -SetCnfParam(void *pT, struct nvlst *lst) +SetCnfParam(void *pT, struct nvlst *lst, int paramType) { lmcry_gcry_t *pThis = (lmcry_gcry_t*) pT; int i, r; @@ -97,34 +110,41 @@ SetCnfParam(void *pT, struct nvlst *lst) uchar *mode = NULL; int nKeys; /* number of keys (actually methods) specified */ struct cnfparamvals *pvals; + struct cnfparamblk *pblk; DEFiRet; + pblk = (paramType == CRYPROV_PARAMTYPE_REGULAR ) ? &pblkRegular : &pblkQueue; nKeys = 0; - pvals = nvlstGetParams(lst, &pblk, NULL); + pvals = nvlstGetParams(lst, pblk, NULL); if(Debug) { dbgprintf("param blk in lmcry_gcry:\n"); - cnfparamsPrint(&pblk, pvals); + cnfparamsPrint(pblk, pvals); } - for(i = 0 ; i < pblk.nParams ; ++i) { + for(i = 0 ; i < pblk->nParams ; ++i) { if(!pvals[i].bUsed) continue; - if(!strcmp(pblk.descr[i].name, "cry.key")) { + if(!strcmp(pblk->descr[i].name, "cry.key") || + !strcmp(pblk->descr[i].name, "queue.cry.key")) { key = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); ++nKeys; - } else if(!strcmp(pblk.descr[i].name, "cry.keyfile")) { + } else if(!strcmp(pblk->descr[i].name, "cry.keyfile") || + !strcmp(pblk->descr[i].name, "queue.cry.keyfile")) { keyfile = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); ++nKeys; - } else if(!strcmp(pblk.descr[i].name, "cry.keyprogram")) { + } else if(!strcmp(pblk->descr[i].name, "cry.keyprogram") || + !strcmp(pblk->descr[i].name, "queue.cry.keyprogram")) { keyprogram = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); ++nKeys; - } else if(!strcmp(pblk.descr[i].name, "cry.mode")) { + } else if(!strcmp(pblk->descr[i].name, "cry.mode") || + !strcmp(pblk->descr[i].name, "queue.cry.mode")) { mode = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); - } else if(!strcmp(pblk.descr[i].name, "cry.algo")) { + } else if(!strcmp(pblk->descr[i].name, "cry.algo") || + !strcmp(pblk->descr[i].name, "queue.cry.algo")) { algo = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else { DBGPRINTF("lmcry_gcry: program error, non-handled " - "param '%s'\n", pblk.descr[i].name); + "param '%s'\n", pblk->descr[i].name); } } if(algo != NULL) { @@ -177,7 +197,7 @@ SetCnfParam(void *pT, struct nvlst *lst) ABORT_FINALIZE(RS_RET_INVALID_PARAMS); } - cnfparamvalsDestruct(pvals, &pblk); + cnfparamvalsDestruct(pvals, pblk); if(key != NULL) { memset(key, 0, strlen((char*)key)); free(key); @@ -189,15 +209,33 @@ finalize_it: RETiRet; } +static void +SetDeleteOnClose(void *pF, int val) +{ + gcryfileSetDeleteOnClose(pF, val); +} + +static rsRetVal +GetBytesLeftInBlock(void *pF, ssize_t *left) +{ + return gcryfileGetBytesLeftInBlock((gcryfile) pF, left); +} static rsRetVal -OnFileOpen(void *pT, uchar *fn, void *pGF) +DeleteStateFiles(uchar *logfn) +{ + return gcryfileDeleteState(logfn); +} + +static rsRetVal +OnFileOpen(void *pT, uchar *fn, void *pGF, char openMode) { lmcry_gcry_t *pThis = (lmcry_gcry_t*) pT; gcryfile *pgf = (gcryfile*) pGF; DEFiRet; + DBGPRINTF("lmcry_gcry: open file '%s', mode '%c'\n", fn, openMode); - CHKiRet(rsgcryInitCrypt(pThis->ctx, pgf, fn)); + CHKiRet(rsgcryInitCrypt(pThis->ctx, pgf, fn, openMode)); finalize_it: /* TODO: enable this error message (need to cleanup loop first ;)) errmsg.LogError(0, iRet, "Encryption Provider" @@ -207,6 +245,16 @@ finalize_it: } static rsRetVal +Decrypt(void *pF, uchar *rec, size_t *lenRec) +{ + DEFiRet; + iRet = rsgcryDecrypt(pF, rec, lenRec); + + RETiRet; +} + + +static rsRetVal Encrypt(void *pF, uchar *rec, size_t *lenRec) { DEFiRet; @@ -231,10 +279,14 @@ CODESTARTobjQueryInterface(lmcry_gcry) } pIf->Construct = (rsRetVal(*)(void*)) lmcry_gcryConstruct; pIf->SetCnfParam = SetCnfParam; + pIf->SetDeleteOnClose = SetDeleteOnClose; pIf->Destruct = (rsRetVal(*)(void*)) lmcry_gcryDestruct; pIf->OnFileOpen = OnFileOpen; pIf->Encrypt = Encrypt; + pIf->Decrypt = Decrypt; pIf->OnFileClose = OnFileClose; + pIf->DeleteStateFiles = DeleteStateFiles; + pIf->GetBytesLeftInBlock = GetBytesLeftInBlock; finalize_it: ENDobjQueryInterface(lmcry_gcry) diff --git a/runtime/lookup.c b/runtime/lookup.c new file mode 100644 index 00000000..bc3b1a94 --- /dev/null +++ b/runtime/lookup.c @@ -0,0 +1,375 @@ +/* lookup.c + * Support for lookup tables in RainerScript. + * + * Copyright 2013 Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * -or- + * see COPYING.ASL20 in the source distribution + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "config.h" +#include <stdlib.h> +#include <string.h> +#include <errno.h> +#include <sys/types.h> +#include <sys/stat.h> +#include <fcntl.h> +#include <json/json.h> +#include <json/json.h> +#include <assert.h> + +#include "rsyslog.h" +#include "srUtils.h" +#include "errmsg.h" +#include "lookup.h" +#include "msg.h" +#include "rsconf.h" +#include "dirty.h" +#include "unicode-helper.h" + +/* definitions for objects we access */ +DEFobjStaticHelpers +DEFobjCurrIf(errmsg) +DEFobjCurrIf(glbl) + +/* forward definitions */ +static rsRetVal lookupReadFile(lookup_t *pThis); + +/* static data */ +/* tables for interfacing with the v6 config system (as far as we need to) */ +static struct cnfparamdescr modpdescr[] = { + { "name", eCmdHdlrString, CNFPARAM_REQUIRED }, + { "file", eCmdHdlrString, CNFPARAM_REQUIRED } +}; +static struct cnfparamblk modpblk = + { CNFPARAMBLK_VERSION, + sizeof(modpdescr)/sizeof(struct cnfparamdescr), + modpdescr + }; + + +/* create a new lookup table object AND include it in our list of + * lookup tables. + */ +rsRetVal +lookupNew(lookup_t **ppThis) +{ + lookup_t *pThis = NULL; + DEFiRet; + + CHKmalloc(pThis = malloc(sizeof(lookup_t))); + pthread_rwlock_init(&pThis->rwlock, NULL); + pThis->name = NULL; + + if(loadConf->lu_tabs.root == NULL) { + loadConf->lu_tabs.root = pThis; + pThis->next = NULL; + } else { + pThis->next = loadConf->lu_tabs.last; + } + loadConf->lu_tabs.last = pThis; + + *ppThis = pThis; +finalize_it: + if(iRet != RS_RET_OK) { + free(pThis); + } + RETiRet; +} +void +lookupDestruct(lookup_t *pThis) +{ + pthread_rwlock_destroy(&pThis->rwlock); + free(pThis->name); + free(pThis); +} + +void +lookupInitCnf(lookup_tables_t *lu_tabs) +{ + lu_tabs->root = NULL; + lu_tabs->last = NULL; +} + + +/* comparison function for qsort() and string array compare + * this is for the string lookup table type + */ +static int +qs_arrcmp_strtab(const void *s1, const void *s2) +{ + return ustrcmp(((lookup_string_tab_etry_t*)s1)->key, ((lookup_string_tab_etry_t*)s2)->key); +} +/* comparison function for bsearch() and string array compare + * this is for the string lookup table type + */ +static int +bs_arrcmp_strtab(const void *s1, const void *s2) +{ + return strcmp((char*)s1, (char*)((lookup_string_tab_etry_t*)s2)->key); +} + +rsRetVal +lookupBuildTable(lookup_t *pThis, struct json_object *jroot) +{ + struct json_object *jversion, *jnomatch, *jtype, *jtab; + struct json_object *jrow, *jindex, *jvalue; + uint32_t i; + uint32_t maxStrSize; + DEFiRet; + + jversion = json_object_object_get(jroot, "version"); + jnomatch = json_object_object_get(jroot, "nomatch"); + jtype = json_object_object_get(jroot, "type"); + jtab = json_object_object_get(jroot, "table"); + pThis->nmemb = json_object_array_length(jtab); + CHKmalloc(pThis->d.strtab = malloc(pThis->nmemb * sizeof(lookup_string_tab_etry_t))); + + maxStrSize = 0; + for(i = 0 ; i < pThis->nmemb ; ++i) { + jrow = json_object_array_get_idx(jtab, i); + jindex = json_object_object_get(jrow, "index"); + jvalue = json_object_object_get(jrow, "value"); + CHKmalloc(pThis->d.strtab[i].key = (uchar*) strdup(json_object_get_string(jindex))); + CHKmalloc(pThis->d.strtab[i].val = (uchar*) strdup(json_object_get_string(jvalue))); + maxStrSize += ustrlen(pThis->d.strtab[i].val); + } + + qsort(pThis->d.strtab, pThis->nmemb, sizeof(lookup_string_tab_etry_t), qs_arrcmp_strtab); +dbgprintf("DDDD: table loaded (max size %u):\n", maxStrSize); +for(i = 0 ; i < pThis->nmemb ; ++i) + dbgprintf("key: '%s', val: '%s'\n", pThis->d.strtab[i].key, pThis->d.strtab[i].val); + +finalize_it: + RETiRet; +} + + +/* find a lookup table. This is a naive O(n) algo, but this really + * doesn't matter as it is called only a few times during config + * load. The function returns either a pointer to the requested + * table or NULL, if not found. + */ +lookup_t * +lookupFindTable(uchar *name) +{ + lookup_t *curr; + + for(curr = loadConf->lu_tabs.root ; curr != NULL ; curr = curr->next) { + if(!ustrcmp(curr->name, name)) + break; + } + return curr; +} + + +/* this reloads a lookup table. This is done while the engine is running, + * as such the function must ensure proper locking and proper order of + * operations (so that nothing can interfere). If the table cannot be loaded, + * the old table is continued to be used. + */ +static rsRetVal +lookupReload(lookup_t *pThis) +{ + uint32_t i; + lookup_t newlu; /* dummy to be able to use support functions without + affecting current settings. */ + DEFiRet; + + DBGPRINTF("reload requested for lookup table '%s'\n", pThis->name); + memset(&newlu, 0, sizeof(newlu)); + CHKmalloc(newlu.name = ustrdup(pThis->name)); + CHKmalloc(newlu.filename = ustrdup(pThis->filename)); + CHKiRet(lookupReadFile(&newlu)); + /* all went well, copy over data members */ + pthread_rwlock_wrlock(&pThis->rwlock); + for(i = 0 ; i < pThis->nmemb ; ++i) { + free(pThis->d.strtab[i].key), /* we don't care about exec order of frees */ + free(pThis->d.strtab[i].val); + } + free(pThis->d.strtab); + pThis->d.strtab = newlu.d.strtab; /* hand table AND ALL STRINGS over! */ + pthread_rwlock_unlock(&pThis->rwlock); + errmsg.LogError(0, RS_RET_OK, "lookup table '%s' reloaded from file '%s'", + pThis->name, pThis->filename); +finalize_it: + free(newlu.name); + free(newlu.filename); + RETiRet; +} + + +/* reload all lookup tables on HUP */ +void +lookupDoHUP() +{ + lookup_t *lu; + for(lu = loadConf->lu_tabs.root ; lu != NULL ; lu = lu->next) { + lookupReload(lu); + } +} + + +/* returns either a pointer to the value (read only!) or NULL + * if either the key could not be found or an error occured. + * Note that an estr_t object is returned. The caller is + * responsible for freeing it. + */ +es_str_t * +lookupKey_estr(lookup_t *pThis, uchar *key) +{ + lookup_string_tab_etry_t *etry; + char *r; + es_str_t *estr; + + pthread_rwlock_rdlock(&pThis->rwlock); + etry = bsearch(key, pThis->d.strtab, pThis->nmemb, sizeof(lookup_string_tab_etry_t), bs_arrcmp_strtab); + if(etry == NULL) { + r = ""; // TODO: use set default + } else { + r = (char*)etry->val; + } + estr = es_newStrFromCStr(r, strlen(r)); + pthread_rwlock_unlock(&pThis->rwlock); + return estr; +} + + +/* note: widely-deployed json_c 0.9 does NOT support incremental + * parsing. In order to keep compatible with e.g. Ubuntu 12.04LTS, + * we read the file into one big memory buffer and parse it at once. + * While this is not very elegant, it will not pose any real issue + * for "reasonable" lookup tables (and "unreasonably" large ones + * will probably have other issues as well...). + */ +static rsRetVal +lookupReadFile(lookup_t *pThis) +{ + struct json_tokener *tokener = NULL; + struct json_object *json = NULL; + int eno = errno; + char errStr[1024]; + char *iobuf = NULL; + int fd; + ssize_t nread; + struct stat sb; + DEFiRet; + + + if(stat((char*)pThis->filename, &sb) == -1) { + eno = errno; + errmsg.LogError(0, RS_RET_FILE_NOT_FOUND, + "lookup table file '%s' stat failed: %s", + pThis->filename, rs_strerror_r(eno, errStr, sizeof(errStr))); + ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); + } + + CHKmalloc(iobuf = malloc(sb.st_size)); + + if((fd = open((const char*) pThis->filename, O_RDONLY)) == -1) { + eno = errno; + errmsg.LogError(0, RS_RET_FILE_NOT_FOUND, + "lookup table file '%s' could not be opened: %s", + pThis->filename, rs_strerror_r(eno, errStr, sizeof(errStr))); + ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); + } + + tokener = json_tokener_new(); + nread = read(fd, iobuf, sb.st_size); + if(nread != (ssize_t) sb.st_size) { + eno = errno; + errmsg.LogError(0, RS_RET_READ_ERR, + "lookup table file '%s' read error: %s", + pThis->filename, rs_strerror_r(eno, errStr, sizeof(errStr))); + ABORT_FINALIZE(RS_RET_READ_ERR); + } + + json = json_tokener_parse_ex(tokener, iobuf, sb.st_size); + if(json == NULL) { + errmsg.LogError(0, RS_RET_JSON_PARSE_ERR, + "lookup table file '%s' json parsing error", + pThis->filename); + ABORT_FINALIZE(RS_RET_JSON_PARSE_ERR); + } + free(iobuf); /* early free to sever resources*/ + iobuf = NULL; /* make sure no double-free */ + + /* got json object, now populate our own in-memory structure */ + CHKiRet(lookupBuildTable(pThis, json)); + +finalize_it: + free(iobuf); + if(tokener != NULL) + json_tokener_free(tokener); + if(json != NULL) + json_object_put(json); + RETiRet; +} + + +rsRetVal +lookupProcessCnf(struct cnfobj *o) +{ + struct cnfparamvals *pvals; + lookup_t *lu; + short i; + DEFiRet; + + pvals = nvlstGetParams(o->nvlst, &modpblk, NULL); + if(pvals == NULL) { + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + DBGPRINTF("lookupProcessCnf params:\n"); + cnfparamsPrint(&modpblk, pvals); + + CHKiRet(lookupNew(&lu)); + + for(i = 0 ; i < modpblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(modpblk.descr[i].name, "file")) { + CHKmalloc(lu->filename = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL)); + } else if(!strcmp(modpblk.descr[i].name, "name")) { + CHKmalloc(lu->name = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL)); + } else { + dbgprintf("lookup_table: program error, non-handled " + "param '%s'\n", modpblk.descr[i].name); + } + } + CHKiRet(lookupReadFile(lu)); + DBGPRINTF("lookup table '%s' loaded from file '%s'\n", lu->name, lu->filename); + +finalize_it: + cnfparamvalsDestruct(pvals, &modpblk); + RETiRet; +} + +void +lookupClassExit(void) +{ + objRelease(glbl, CORE_COMPONENT); + objRelease(errmsg, CORE_COMPONENT); +} + +rsRetVal +lookupClassInit(void) +{ + DEFiRet; + CHKiRet(objGetObjInterface(&obj)); + CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(errmsg, CORE_COMPONENT)); +finalize_it: + RETiRet; +} diff --git a/runtime/lookup.h b/runtime/lookup.h new file mode 100644 index 00000000..c478d679 --- /dev/null +++ b/runtime/lookup.h @@ -0,0 +1,57 @@ +/* header for lookup.c + * + * Copyright 2013 Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * -or- + * see COPYING.ASL20 in the source distribution + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef INCLUDED_LOOKUP_H +#define INCLUDED_LOOKUP_H +#include <libestr.h> + +struct lookup_tables_s { + lookup_t *root; /* the root of the template list */ + lookup_t *last; /* points to the last element of the template list */ +}; + +struct lookup_string_tab_etry_s { + uchar *key; + uchar *val; +}; + +/* a single lookup table */ +struct lookup_s { + pthread_rwlock_t rwlock; /* protect us in case of dynamic reloads */ + uchar *name; + uchar *filename; + uint32_t nmemb; + union { + lookup_string_tab_etry_t *strtab; + } d; + lookup_t *next; +}; + +/* prototypes */ +void lookupInitCnf(lookup_tables_t *lu_tabs); +rsRetVal lookupProcessCnf(struct cnfobj *o); +lookup_t *lookupFindTable(uchar *name); +es_str_t * lookupKey_estr(lookup_t *pThis, uchar *key); +void lookupDestruct(lookup_t *pThis); +void lookupClassExit(void); +void lookupDoHUP(); +rsRetVal lookupClassInit(void); + +#endif /* #ifndef INCLUDED_LOOKUP_H */ diff --git a/runtime/module-template.h b/runtime/module-template.h index 8a958f90..f65aaf0f 100644 --- a/runtime/module-template.h +++ b/runtime/module-template.h @@ -175,6 +175,44 @@ static rsRetVal freeInstance(void* pModData)\ RETiRet;\ } +/* createWrkrInstance() + */ +#define BEGINcreateWrkrInstance \ +static rsRetVal createWrkrInstance(wrkrInstanceData_t **ppWrkrData, instanceData *pData)\ + {\ + DEFiRet; /* store error code here */\ + wrkrInstanceData_t *pWrkrData; /* use this to point to data elements */ + +#define CODESTARTcreateWrkrInstance \ + if((pWrkrData = calloc(1, sizeof(wrkrInstanceData_t))) == NULL) {\ + *ppWrkrData = NULL;\ + ENDfunc \ + return RS_RET_OUT_OF_MEMORY;\ + } \ + pWrkrData->pData = pData; + +#define ENDcreateWrkrInstance \ + *ppWrkrData = pWrkrData;\ + RETiRet;\ +} + +/* freeWrkrInstance */ +#define BEGINfreeWrkrInstance \ +static rsRetVal freeWrkrInstance(void* pd)\ +{\ + DEFiRet;\ + wrkrInstanceData_t *pWrkrData; + +#define CODESTARTfreeWrkrInstance \ + pWrkrData = (wrkrInstanceData_t*) pd; + +#define ENDfreeWrkrInstance \ + if(pWrkrData != NULL)\ + free(pWrkrData); /* we need to free this in any case */\ + RETiRet;\ +} + + /* isCompatibleWithFeature() */ #define BEGINisCompatibleWithFeature \ @@ -194,7 +232,7 @@ static rsRetVal isCompatibleWithFeature(syslogFeature __attribute__((unused)) eF * introduced in v4.3.3 -- rgerhards, 2009-04-27 */ #define BEGINbeginTransaction \ -static rsRetVal beginTransaction(instanceData __attribute__((unused)) *pData)\ +static rsRetVal beginTransaction(wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ DEFiRet; @@ -209,7 +247,7 @@ static rsRetVal beginTransaction(instanceData __attribute__((unused)) *pData)\ * introduced in v4.3.3 -- rgerhards, 2009-04-27 */ #define BEGINendTransaction \ -static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\ +static rsRetVal endTransaction(wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ DEFiRet; @@ -223,7 +261,7 @@ static rsRetVal endTransaction(instanceData __attribute__((unused)) *pData)\ /* doAction() */ #define BEGINdoAction \ -static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, instanceData __attribute__((unused)) *pData)\ +static rsRetVal doAction(uchar __attribute__((unused)) **ppString, unsigned __attribute__((unused)) iMsgOpts, wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ DEFiRet; @@ -382,12 +420,12 @@ static rsRetVal newInpInst(struct nvlst *lst)\ * rgerhard, 2007-08-02 */ #define BEGINtryResume \ -static rsRetVal tryResume(instanceData __attribute__((unused)) *pData)\ +static rsRetVal tryResume(wrkrInstanceData_t __attribute__((unused)) *pWrkrData)\ {\ DEFiRet; #define CODESTARTtryResume \ - assert(pData != NULL); + assert(pWrkrData != NULL); #define ENDtryResume \ RETiRet;\ @@ -467,6 +505,13 @@ static rsRetVal queryEtryPt(uchar *name, rsRetVal (**pEtryPoint)())\ *pEtryPoint = tryResume;\ } +/* standard queries for output module interface in rsyslog v8+ */ +#define CODEqueryEtryPt_STD_OMOD8_QUERIES \ + else if(!strcmp((char*) name, "createWrkrInstance")) {\ + *pEtryPoint = createWrkrInstance;\ + } else if(!strcmp((char*) name, "freeWrkrInstance")) {\ + *pEtryPoint = freeWrkrInstance;\ + } /* the following definition is queryEtryPt block that must be added * if an output module supports the transactional interface. diff --git a/runtime/modules.c b/runtime/modules.c index 56606306..52096082 100644 --- a/runtime/modules.c +++ b/runtime/modules.c @@ -652,6 +652,9 @@ doModInit(rsRetVal (*modInit)(int, int*, rsRetVal(**)(), rsRetVal(*)(), modInfo_ CHKiRet((*pNew->modQueryEtryPt)((uchar*)"doAction", &pNew->mod.om.doAction)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"parseSelectorAct", &pNew->mod.om.parseSelectorAct)); CHKiRet((*pNew->modQueryEtryPt)((uchar*)"tryResume", &pNew->tryResume)); + CHKiRet((*pNew->modQueryEtryPt)((uchar*)"createWrkrInstance", &pNew->mod.om.createWrkrInstance)); + CHKiRet((*pNew->modQueryEtryPt)((uchar*)"freeWrkrInstance", &pNew->mod.om.freeWrkrInstance)); + /* try load optional interfaces */ localRet = (*pNew->modQueryEtryPt)((uchar*)"doHUP", &pNew->doHUP); if(localRet != RS_RET_OK && localRet != RS_RET_MODULE_ENTRY_POINT_NOT_FOUND) diff --git a/runtime/modules.h b/runtime/modules.h index 64644be2..a0637ac9 100644 --- a/runtime/modules.h +++ b/runtime/modules.h @@ -122,10 +122,6 @@ struct modInfo_s { rsRetVal (*activateCnf)(void*Cnf); /* make provided config the running conf */ rsRetVal (*freeCnf)(void*Cnf); /* end v2 config system specific */ - /* below: create an instance of this module. Most importantly the module - * can allocate instance memory in this call. - */ - rsRetVal (*createInstance)(); union { struct {/* data for input modules */ /* TODO: remove? */rsRetVal (*willRun)(void); /* check if the current config will be able to run*/ @@ -143,6 +139,8 @@ struct modInfo_s { rsRetVal (*parseSelectorAct)(uchar**, void**,omodStringRequest_t**); rsRetVal (*newActInst)(uchar *modName, struct nvlst *lst, void **, omodStringRequest_t **); rsRetVal (*SetShutdownImmdtPtr)(void *pData, void *pPtr); + rsRetVal (*createWrkrInstance)(void*ppWrkrData, void*pData); + rsRetVal (*freeWrkrInstance)(void*pWrkrData); } om; struct { /* data for library modules */ char dummy; diff --git a/runtime/msg.c b/runtime/msg.c index 10ecf48a..5df4e031 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -65,6 +65,13 @@ #include "net.h" #include "var.h" #include "rsconf.h" +#include "parserif.h" + +/* TODO: move the global variable root to the config object - had no time to to it + * right now before vacation -- rgerhards, 2013-07-22 + */ +static pthread_rwlock_t glblVars_rwlock; +struct json_object *global_var_root = NULL; /* static data */ DEFobjStaticHelpers @@ -317,7 +324,7 @@ static pthread_mutex_t mutTrimCtr; /* mutex to handle malloc trim */ /* some forward declarations */ static int getAPPNAMELen(msg_t *pM, sbool bLockMutex); -static rsRetVal jsonPathFindParent(msg_t *pM, uchar *name, uchar *leaf, struct json_object **parent, int bCreate); +static rsRetVal jsonPathFindParent(struct json_object *jroot, uchar *name, uchar *leaf, struct json_object **parent, int bCreate); static uchar * jsonPathGetLeaf(uchar *name, int lenName); static struct json_object *jsonDeepCopy(struct json_object *src); @@ -377,13 +384,6 @@ MsgSetRulesetByName(msg_t *pMsg, cstr_t *rulesetName) rulesetGetRuleset(runConf, &(pMsg->pRuleset), rsCStrGetSzStrNoNULL(rulesetName)); } - -static inline int getProtocolVersion(msg_t *pM) -{ - return(pM->iProtocolVersion); -} - - /* do a DNS reverse resolution, if not already done, reflect status * rgerhards, 2009-11-16 */ @@ -452,14 +452,12 @@ getRcvFromIP(msg_t *pM) } -/* map a property name (C string) to a property ID */ +/* map a property name (string) to a property ID */ rsRetVal -propNameStrToID(uchar *pName, propid_t *pPropID) +propNameToID(uchar *pName, propid_t *pPropID) { DEFiRet; - assert(pName != NULL); - /* sometimes there are aliases to the original MonitoWare * property names. These come after || in the ifs below. */ if(!strcmp((char*) pName, "msg")) { @@ -534,13 +532,18 @@ propNameStrToID(uchar *pName, propid_t *pPropID) *pPropID = PROP_SYS_MYHOSTNAME; } else if(!strcmp((char*) pName, "$!all-json")) { *pPropID = PROP_CEE_ALL_JSON; - } else if(!strncmp((char*) pName, "$!", 2)) { - *pPropID = PROP_CEE; } else if(!strcmp((char*) pName, "$bom")) { *pPropID = PROP_SYS_BOM; } else if(!strcmp((char*) pName, "$uptime")) { *pPropID = PROP_SYS_UPTIME; + } else if(!strncmp((char*) pName, "$!", 2) || pName[0] == '!') { + *pPropID = PROP_CEE; + } else if(!strncmp((char*) pName, "$.", 2) || pName[0] == '.') { + *pPropID = PROP_LOCAL_VAR; + } else if(!strncmp((char*) pName, "$/", 2) || pName[0] == '/') { + *pPropID = PROP_GLOBAL_VAR; } else { + DBGPRINTF("PROP_INVALID for name '%s'\n", pName); *pPropID = PROP_INVALID; iRet = RS_RET_VAR_NOT_FOUND; } @@ -549,21 +552,6 @@ propNameStrToID(uchar *pName, propid_t *pPropID) } -/* map a property name (string) to a property ID */ -rsRetVal -propNameToID(cstr_t *pCSPropName, propid_t *pPropID) -{ - uchar *pName; - DEFiRet; - - assert(pCSPropName != NULL); - assert(pPropID != NULL); - pName = rsCStrGetSzStrNoNULL(pCSPropName); - iRet = propNameStrToID(pName, pPropID); - RETiRet; -} - - /* map a property ID to a name string (useful for displaying) */ uchar *propIDToName(propid_t propID) { @@ -634,6 +622,8 @@ uchar *propIDToName(propid_t propID) return UCHAR_CONSTANT("$MYHOSTNAME"); case PROP_CEE: return UCHAR_CONSTANT("*CEE-based property*"); + case PROP_LOCAL_VAR: + return UCHAR_CONSTANT("*LOCAL_VARIABLE*"); case PROP_CEE_ALL_JSON: return UCHAR_CONSTANT("$!all-json"); case PROP_SYS_BOM: @@ -699,7 +689,7 @@ static inline rsRetVal msgBaseConstruct(msg_t **ppThis) pM->pszTIMESTAMP3339 = NULL; pM->pszTIMESTAMP_MySQL = NULL; pM->pszTIMESTAMP_PgSQL = NULL; - pM->pCSStrucData = NULL; + pM->pszStrucData = NULL; pM->pCSAPPNAME = NULL; pM->pCSPROCID = NULL; pM->pCSMSGID = NULL; @@ -708,6 +698,8 @@ static inline rsRetVal msgBaseConstruct(msg_t **ppThis) pM->rcvFrom.pRcvFrom = NULL; pM->pRuleset = NULL; pM->json = NULL; + pM->localvars = NULL; + pM->dfltTZ[0] = '\0'; memset(&pM->tRcvdAt, 0, sizeof(pM->tRcvdAt)); memset(&pM->tTIMESTAMP, 0, sizeof(pM->tTIMESTAMP)); pM->TAG.pszTAG = NULL; @@ -841,10 +833,9 @@ CODESTARTobjDestruct(msg) free(pThis->pszRcvdAt_PgSQL); free(pThis->pszTIMESTAMP_MySQL); free(pThis->pszTIMESTAMP_PgSQL); + free(pThis->pszStrucData); if(pThis->iLenPROGNAME >= CONF_PROGNAME_BUFSIZE) free(pThis->PROGNAME.ptr); - if(pThis->pCSStrucData != NULL) - rsCStrDestruct(&pThis->pCSStrucData); if(pThis->pCSAPPNAME != NULL) rsCStrDestruct(&pThis->pCSAPPNAME); if(pThis->pCSPROCID != NULL) @@ -853,6 +844,8 @@ CODESTARTobjDestruct(msg) rsCStrDestruct(&pThis->pCSMSGID); if(pThis->json != NULL) json_object_put(pThis->json); + if(pThis->localvars != NULL) + json_object_put(pThis->localvars); if(pThis->pszUUID != NULL) free(pThis->pszUUID); # ifndef HAVE_ATOMIC_BUILTINS @@ -994,14 +987,21 @@ msg_t* MsgDup(msg_t* pOld) tmpCOPYSZ(HOSTNAME); } } + if(pOld->pszStrucData == NULL) { + pNew->pszStrucData = NULL; + } else { + pNew->pszStrucData = (uchar*)strdup((char*)pOld->pszStrucData); + pNew->lenStrucData = pOld->lenStrucData; + } - tmpCOPYCSTR(StrucData); tmpCOPYCSTR(APPNAME); tmpCOPYCSTR(PROCID); tmpCOPYCSTR(MSGID); if(pOld->json != NULL) pNew->json = jsonDeepCopy(pOld->json); + if(pOld->localvars != NULL) + pNew->localvars = jsonDeepCopy(pOld->localvars); /* we do not copy all other cache properties, as we do not even know * if they are needed once again. So we let them re-create if needed. @@ -1056,12 +1056,17 @@ static rsRetVal MsgSerialize(msg_t *pThis, strm_t *pStrm) CHKiRet(obj.SerializeProp(pStrm, UCHAR_CONSTANT("pszRcvFrom"), PROPTYPE_PSZ, (void*) psz)); psz = getRcvFromIP(pThis); CHKiRet(obj.SerializeProp(pStrm, UCHAR_CONSTANT("pszRcvFromIP"), PROPTYPE_PSZ, (void*) psz)); + psz = pThis->pszStrucData; + CHKiRet(obj.SerializeProp(pStrm, UCHAR_CONSTANT("pszRcvStrucData"), PROPTYPE_PSZ, (void*) psz)); if(pThis->json != NULL) { psz = (uchar*) json_object_get_string(pThis->json); CHKiRet(obj.SerializeProp(pStrm, UCHAR_CONSTANT("json"), PROPTYPE_PSZ, (void*) psz)); } + if(pThis->localvars != NULL) { + psz = (uchar*) json_object_get_string(pThis->localvars); + CHKiRet(obj.SerializeProp(pStrm, UCHAR_CONSTANT("localvars"), PROPTYPE_PSZ, (void*) psz)); + } - objSerializePTR(pStrm, pCSStrucData, CSTR); objSerializePTR(pStrm, pCSAPPNAME, CSTR); objSerializePTR(pStrm, pCSPROCID, CSTR); objSerializePTR(pStrm, pCSMSGID, CSTR); @@ -1202,7 +1207,15 @@ MsgDeserialize(msg_t *pMsg, strm_t *pStrm) reinitVar(pVar); CHKiRet(objDeserializeProperty(pVar, pStrm)); } - if(isProp("pCSStrucData")) { + if(isProp("localvars")) { + tokener = json_tokener_new(); + pMsg->localvars = json_tokener_parse_ex(tokener, (char*)rsCStrGetSzStrNoNULL(pVar->val.pStr), + cstrLen(pVar->val.pStr)); + json_tokener_free(tokener); + reinitVar(pVar); + CHKiRet(objDeserializeProperty(pVar, pStrm)); + } + if(isProp("pszStrucData")) { MsgSetStructuredData(pMsg, (char*) rsCStrGetSzStrNoNULL(pVar->val.pStr)); reinitVar(pVar); CHKiRet(objDeserializeProperty(pVar, pStrm)); @@ -1292,7 +1305,7 @@ static rsRetVal aquirePROCIDFromTAG(msg_t *pM) if(pM->pCSPROCID != NULL) return RS_RET_OK; /* we are already done ;) */ - if(getProtocolVersion(pM) != 0) + if(msgGetProtocolVersion(pM) != 0) return RS_RET_OK; /* we can only emulate if we have legacy format */ pszTag = (uchar*) ((pM->iLenTAG < CONF_TAG_BUFSIZE) ? pM->TAG.szBuf : pM->TAG.pszTAG); @@ -1975,7 +1988,7 @@ static inline void tryEmulateTAG(msg_t *pM, sbool bLockMutex) return; /* done, no need to emulate */ } - if(getProtocolVersion(pM) == 1) { + if(msgGetProtocolVersion(pM) == 1) { if(!strcmp(getPROCID(pM, MUTEX_ALREADY_LOCKED), "-")) { /* no process ID, use APP-NAME only */ MsgSetTAG(pM, (uchar*) getAPPNAME(pM, MUTEX_ALREADY_LOCKED), getAPPNAMELen(pM, MUTEX_ALREADY_LOCKED)); @@ -2075,42 +2088,27 @@ rsRetVal MsgSetStructuredData(msg_t *pMsg, char* pszStrucData) { DEFiRet; ISOBJ_TYPE_assert(pMsg, msg); - if(pMsg->pCSStrucData == NULL) { - /* we need to obtain the object first */ - CHKiRet(rsCStrConstruct(&pMsg->pCSStrucData)); - } - /* if we reach this point, we have the object */ - iRet = rsCStrSetSzStr(pMsg->pCSStrucData, (uchar*) pszStrucData); - + free(pMsg->pszStrucData); + CHKmalloc(pMsg->pszStrucData = (uchar*)strdup(pszStrucData)); + pMsg->lenStrucData = strlen(pszStrucData); finalize_it: RETiRet; } -/* get the length of the "STRUCTURED-DATA" sz string - * rgerhards, 2005-11-24 - */ -#if 0 /* This method is currently not called, be we like to preserve it */ -static int getStructuredDataLen(msg_t *pM) -{ - return (pM->pCSStrucData == NULL) ? 1 : rsCStrLen(pM->pCSStrucData); -} -#endif - -/* get the "STRUCTURED-DATA" as sz string - * rgerhards, 2005-11-24 - */ -static inline char *getStructuredData(msg_t *pM) +/* get the "STRUCTURED-DATA" as sz string, including length */ +void +MsgGetStructuredData(msg_t *pM, uchar **pBuf, rs_size_t *len) { - uchar *pszRet; - MsgLock(pM); - if(pM->pCSStrucData == NULL) - pszRet = UCHAR_CONSTANT("-"); - else - pszRet = rsCStrGetSzStrNoNULL(pM->pCSStrucData); + if(pM->pszStrucData == NULL) { + *pBuf = UCHAR_CONSTANT("-"), + *len = 1; + } else { + *pBuf = pM->pszStrucData, + *len = pM->lenStrucData; + } MsgUnlock(pM); - return (char*) pszRet; } /* get the "programname" as sz string @@ -2145,7 +2143,7 @@ static void tryEmulateAPPNAME(msg_t *pM) if(pM->pCSAPPNAME != NULL) return; /* we are already done */ - if(getProtocolVersion(pM) == 0) { + if(msgGetProtocolVersion(pM) == 0) { /* only then it makes sense to emulate */ MsgSetAPPNAME(pM, (char*)getProgramName(pM, MUTEX_ALREADY_LOCKED)); } @@ -2215,6 +2213,15 @@ void MsgSetInputName(msg_t *pThis, prop_t *inputName) pThis->pInputName = inputName; } +/* Set default TZ. Note that at most 7 chars are set, as we would + * otherwise overrun our buffer! + */ +void MsgSetDfltTZ(msg_t *pThis, char *tz) +{ + strncpy(pThis->dfltTZ, tz, 7); + pThis->dfltTZ[7] = '\0'; /* ensure 0-Term in case of overflow! */ +} + /* Set the pfrominet socket store, so that we can obtain the peer at some * later time. Note that we do not check if pRcvFrom is already set, so this @@ -2237,7 +2244,6 @@ finalize_it: RETiRet; } - /* rgerhards 2008-09-10: set RcvFrom name in msg object. This calls AddRef() * on the property, because this must be done in all current cases and there * is no case expected where this may not be necessary. @@ -2463,11 +2469,17 @@ typedef enum ENOWType { NOW_NOW, NOW_YEAR, NOW_MONTH, NOW_DAY, NOW_HOUR, NOW_HHO static uchar *getNOW(eNOWType eNow, struct syslogTime *t) { uchar *pBuf; + struct syslogTime tt; if((pBuf = (uchar*) MALLOC(sizeof(uchar) * tmpBUFSIZE)) == NULL) { return NULL; } + if(t == NULL) { /* can happen if called via script engine */ + datetime.getCurrTime(&tt, NULL); + t = &tt; + } + if(t->year == 0) { /* not yet set! */ datetime.getCurrTime(t, NULL); } @@ -2510,12 +2522,12 @@ static uchar *getNOW(eNOWType eNow, struct syslogTime *t) #undef tmpBUFSIZE /* clean up */ -/* Get a CEE-Property as string value*/ +/* Get a JSON-Property as string value (used for various types of JSON-based vars) */ rsRetVal -getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) +getJSONPropVal(msg_t *pMsg, msgPropDescr_t *pProp, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed) { - uchar *name = NULL; uchar *leaf; + struct json_object *jroot; struct json_object *parent; struct json_object *field; DEFiRet; @@ -2523,15 +2535,26 @@ getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, un if(*pbMustBeFreed) free(*pRes); *pRes = NULL; - // TODO: mutex? - if(pM->json == NULL) goto finalize_it; - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { - field = pM->json; + if(pProp->id == PROP_CEE) { + jroot = pMsg->json; + } else if(pProp->id == PROP_LOCAL_VAR) { + jroot = pMsg->localvars; + } else if(pProp->id == PROP_GLOBAL_VAR) { + pthread_rwlock_rdlock(&glblVars_rwlock); + jroot = global_var_root; } else { - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 1)); + DBGPRINTF("msgGetJSONPropVal; invalid property id %d\n", + pProp->id); + ABORT_FINALIZE(RS_RET_NOT_FOUND); + } + if(jroot == NULL) goto finalize_it; + + if(!strcmp((char*)pProp->name, "!")) { + field = jroot; + } else { + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 1)); field = json_object_object_get(parent, (char*)leaf); } if(field != NULL) { @@ -2541,7 +2564,8 @@ getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, un } finalize_it: - free(name); + if(pProp->id == PROP_GLOBAL_VAR) + pthread_rwlock_unlock(&glblVars_rwlock); if(*pRes == NULL) { /* could not find any value, so set it to empty */ *pRes = (unsigned char*)""; @@ -2551,35 +2575,47 @@ finalize_it: } -/* Get a CEE-Property as native json object - */ +/* Get a JSON-based-variable as native json object */ rsRetVal -msgGetCEEPropJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson) +msgGetJSONPropJSON(msg_t *pMsg, msgPropDescr_t *pProp, struct json_object **pjson) { - uchar *name = NULL; + struct json_object *jroot; uchar *leaf; struct json_object *parent; DEFiRet; - // TODO: mutex? - if(pM->json == NULL) { + if(pProp->id == PROP_CEE) { + jroot = pMsg->json; + } else if(pProp->id == PROP_LOCAL_VAR) { + jroot = pMsg->localvars; + } else if(pProp->id == PROP_GLOBAL_VAR) { + pthread_rwlock_rdlock(&glblVars_rwlock); + jroot = global_var_root; + } else { + DBGPRINTF("msgGetJSONPropJSON; invalid property id %d\n", + pProp->id); + ABORT_FINALIZE(RS_RET_NOT_FOUND); + } + if(jroot == NULL) { + DBGPRINTF("msgGetJSONPropJSON; jroot empty for property %s\n", + pProp->name); ABORT_FINALIZE(RS_RET_NOT_FOUND); } - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { - *pjson = pM->json; + if(!strcmp((char*)pProp->name, "!")) { + *pjson = jroot; FINALIZE; } - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 1)); + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 1)); *pjson = json_object_object_get(parent, (char*)leaf); if(*pjson == NULL) { ABORT_FINALIZE(RS_RET_NOT_FOUND); } finalize_it: - free(name); + if(pProp->id == PROP_GLOBAL_VAR) + pthread_rwlock_unlock(&glblVars_rwlock); RETiRet; } @@ -2784,7 +2820,7 @@ finalize_it: *pPropLen = sizeof("**OUT OF MEMORY**") - 1; \ return(UCHAR_CONSTANT("**OUT OF MEMORY**"));} uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, - propid_t propid, es_str_t *propName, rs_size_t *pPropLen, + msgPropDescr_t *pProp, rs_size_t *pPropLen, unsigned short *pbMustBeFreed, struct syslogTime *ttNow) { uchar *pRes; /* result pointer */ @@ -2807,7 +2843,7 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, *pbMustBeFreed = 0; - switch(propid) { + switch(pProp->id) { case PROP_MSG: pRes = getMSG(pMsg); bufLen = getMSGLen(pMsg); @@ -2880,7 +2916,7 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, pRes = (uchar*)getProtocolVersionString(pMsg); break; case PROP_STRUCTURED_DATA: - pRes = (uchar*)getStructuredData(pMsg); + MsgGetStructuredData(pMsg, &pRes, &bufLen); break; case PROP_APP_NAME: pRes = (uchar*)getAPPNAME(pMsg, LOCK_MUTEX); @@ -2979,7 +3015,9 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } break; case PROP_CEE: - getCEEPropVal(pMsg, propName, &pRes, &bufLen, pbMustBeFreed); + case PROP_LOCAL_VAR: + case PROP_GLOBAL_VAR: + getJSONPropVal(pMsg, pProp, &pRes, &bufLen, pbMustBeFreed); break; case PROP_SYS_BOM: if(*pbMustBeFreed == 1) @@ -3040,7 +3078,7 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, /* there is no point in continuing, we may even otherwise render the * error message unreadable. rgerhards, 2007-07-10 */ - dbgprintf("invalid property id: '%d'\n", propid); + dbgprintf("invalid property id: '%d'\n", pProp->id); *pbMustBeFreed = 0; *pPropLen = sizeof("**INVALID PROPERTY NAME**") - 1; return UCHAR_CONSTANT("**INVALID PROPERTY NAME**"); @@ -3648,66 +3686,6 @@ uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, } -/* The function returns a cee variable suitable for use with RainerScript. - * Note: caller must free the returned string. - * Note that we need to do a lot of conversions between es_str_t and cstr -- this will go away once - * we have moved larger parts of rsyslog to es_str_t. Acceptable for the moment, especially as we intend - * to rewrite the script engine as well! - * rgerhards, 2010-12-03 - */ -es_str_t* -msgGetCEEVarNew(msg_t *pMsg, char *name) -{ - uchar *leaf; - char *val; - es_str_t *estr = NULL; - struct json_object *json, *parent; - - ISOBJ_TYPE_assert(pMsg, msg); - - if(pMsg->json == NULL) { - estr = es_newStr(1); - goto done; - } - leaf = jsonPathGetLeaf((uchar*)name, strlen(name)); - if(jsonPathFindParent(pMsg, (uchar*)name, leaf, &parent, 1) != RS_RET_OK) { - estr = es_newStr(1); - goto done; - } - json = json_object_object_get(parent, (char*)leaf); - val = (char*)json_object_get_string(json); - estr = es_newStrFromCStr(val, strlen(val)); -done: - return estr; -} - - -/* Return an es_str_t for given message property. - */ -es_str_t* -msgGetMsgVarNew(msg_t *pThis, uchar *name) -{ - rs_size_t propLen; - uchar *pszProp = NULL; - propid_t propid; - unsigned short bMustBeFreed = 0; - es_str_t *estr; - - ISOBJ_TYPE_assert(pThis, msg); - - /* always call MsgGetProp() without a template specifier */ - /* TODO: optimize propNameToID() call -- rgerhards, 2009-06-26 */ - propNameStrToID(name, &propid); - pszProp = (uchar*) MsgGetProp(pThis, NULL, propid, NULL, &propLen, &bMustBeFreed, NULL); - - estr = es_newStrFromCStr((char*)pszProp, propLen); - if(bMustBeFreed) - free(pszProp); - - return estr; -} - - /* This function can be used as a generic way to set properties. * We have to handle a lot of legacy, so our return value is not always * 100% correct (called functions do not always provide one, should @@ -3758,7 +3736,7 @@ rsRetVal MsgSetProperty(msg_t *pThis, var_t *pProp) prop.Destruct(&propRcvFrom); } else if(isProp("pszHOSTNAME")) { MsgSetHOSTNAME(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr), rsCStrLen(pProp->val.pStr)); - } else if(isProp("pCSStrucData")) { + } else if(isProp("pszStrucData")) { MsgSetStructuredData(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); } else if(isProp("pCSAPPNAME")) { MsgSetAPPNAME(pThis, (char*) rsCStrGetSzStrNoNULL(pProp->val.pStr)); @@ -3809,16 +3787,22 @@ static uchar * jsonPathGetLeaf(uchar *name, int lenName) { int i; - for(i = lenName ; name[i] != '!' && i >= 0 ; --i) - /* just skip */; - if(name[i] == '!') + for(i = lenName ; i >= 0 ; --i) + if(i == 0) { + if(name[0] == '!' || name[0] == '.' || name[0] == '/') + break; + } else { + if(name[i] == '!') + break; + } + if(name[i] == '!' || name[i] == '.' || name[i] == '/') ++i; return name + i; } static rsRetVal -jsonPathFindNext(struct json_object *root, uchar **name, uchar *leaf, +jsonPathFindNext(struct json_object *root, uchar *namestart, uchar **name, uchar *leaf, struct json_object **found, int bCreate) { uchar namebuf[1024]; @@ -3827,9 +3811,9 @@ jsonPathFindNext(struct json_object *root, uchar **name, uchar *leaf, uchar *p = *name; DEFiRet; - if(*p == '!') + if(*p == '!' || (*name == namestart && (*p == '.' || *p == '/'))) ++p; - for(i = 0 ; *p && *p != '!' && p != leaf && i < sizeof(namebuf)-1 ; ++i, ++p) + for(i = 0 ; *p && !(p == namestart && (*p == '.' || *p == '/')) && *p != '!' && p != leaf && i < sizeof(namebuf)-1 ; ++i, ++p) namebuf[i] = *p; if(i > 0) { namebuf[i] = '\0'; @@ -3853,12 +3837,14 @@ finalize_it: } static rsRetVal -jsonPathFindParent(msg_t *pM, uchar *name, uchar *leaf, struct json_object **parent, int bCreate) +jsonPathFindParent(struct json_object *jroot, uchar *name, uchar *leaf, struct json_object **parent, int bCreate) { + uchar *namestart; DEFiRet; - *parent = pM->json; + namestart = name; + *parent = jroot; while(name < leaf-1) { - jsonPathFindNext(*parent, &name, leaf, parent, bCreate); + jsonPathFindNext(*parent, namestart, &name, leaf, parent, bCreate); } RETiRet; } @@ -3885,31 +3871,28 @@ DBGPRINTF("AAAA jsonMerge adds '%s'\n", it.key); /* find a JSON structure element (field or container doesn't matter). */ rsRetVal -jsonFind(msg_t *pM, es_str_t *propName, struct json_object **jsonres) +jsonFind(struct json_object *jroot, msgPropDescr_t *pProp, struct json_object **jsonres) { - uchar *name = NULL; uchar *leaf; struct json_object *parent; struct json_object *field; DEFiRet; - if(pM->json == NULL) { + if(jroot == NULL) { field = NULL; goto finalize_it; } - if(!es_strbufcmp(propName, (uchar*)"!", 1)) { - field = pM->json; + if(!strcmp((char*)pProp->name, "!")) { + field = jroot; } else { - name = (uchar*)es_str2cstr(propName, NULL); - leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 0)); + leaf = jsonPathGetLeaf(pProp->name, pProp->nameLen); + CHKiRet(jsonPathFindParent(jroot, pProp->name, leaf, &parent, 0)); field = json_object_object_get(parent, (char*)leaf); } *jsonres = field; finalize_it: - free(name); RETiRet; } @@ -3917,23 +3900,33 @@ rsRetVal msgAddJSON(msg_t *pM, uchar *name, struct json_object *json) { /* TODO: error checks! This is a quick&dirty PoC! */ + struct json_object **pjroot; struct json_object *parent, *leafnode; uchar *leaf; DEFiRet; MsgLock(pM); - if(name[0] == '!' && name[1] == '\0') { - if(pM->json == NULL) - pM->json = json; + if(name[0] == '!') { + pjroot = &pM->json; + } else if(name[0] == '.') { + pjroot = &pM->localvars; + } else { /* globl var */ + pthread_rwlock_wrlock(&glblVars_rwlock); + pjroot = &global_var_root; + } + + if(name[1] == '\0') { /* full tree? */ + if(*pjroot == NULL) + *pjroot = json; else - CHKiRet(jsonMerge(pM->json, json)); + CHKiRet(jsonMerge(*pjroot, json)); } else { - if(pM->json == NULL) { + if(*pjroot == NULL) { /* now we need a root obj */ - pM->json = json_object_new_object(); + *pjroot = json_object_new_object(); } leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 1)); + CHKiRet(jsonPathFindParent(*pjroot, name, leaf, &parent, 1)); if (json_object_get_type(parent) != json_type_object) { DBGPRINTF("msgAddJSON: not a container in json path," "name is '%s'\n", name); @@ -3945,7 +3938,7 @@ msgAddJSON(msg_t *pM, uchar *name, struct json_object *json) json_object_object_add(parent, (char*)leaf, json); } else { if(json_object_get_type(json) == json_type_object) { - CHKiRet(jsonMerge(pM->json, json)); + CHKiRet(jsonMerge(*pjroot, json)); } else { //dbgprintf("AAAA: leafnode already exists, type is %d, update with %d\n", (int)json_object_get_type(leafnode), (int)json_object_get_type(json)); /* TODO: improve the code below, however, the current @@ -3971,33 +3964,52 @@ msgAddJSON(msg_t *pM, uchar *name, struct json_object *json) } finalize_it: + if(name[0] == '/') + pthread_rwlock_unlock(&glblVars_rwlock); MsgUnlock(pM); RETiRet; } + rsRetVal msgDelJSON(msg_t *pM, uchar *name) { + struct json_object **jroot; struct json_object *parent, *leafnode; uchar *leaf; DEFiRet; dbgprintf("AAAA: unset variable '%s'\n", name); MsgLock(pM); - if(name[0] == '!' && name[1] == '\0') { - /* strange, but I think we should permit this. After all, + + if(name[0] == '!') { + jroot = &pM->json; + } else if(name[0] == '.') { + jroot = &pM->localvars; + } else { /* globl var */ + pthread_rwlock_wrlock(&glblVars_rwlock); + jroot = &global_var_root; + } + if(jroot == NULL) { + DBGPRINTF("msgDelJSONVar; jroot empty in unset for property %s\n", + name); + FINALIZE; + } + + if(name[1] == '\0') { + /* full tree! Strange, but I think we should permit this. After all, * we trust rsyslog.conf to be written by the admin. */ DBGPRINTF("unsetting JSON root object\n"); - json_object_put(pM->json); - pM->json = NULL; + json_object_put(*jroot); + *jroot = NULL; } else { - if(pM->json == NULL) { + if(*jroot == NULL) { /* now we need a root obj */ - pM->json = json_object_new_object(); + *jroot = json_object_new_object(); } leaf = jsonPathGetLeaf(name, ustrlen(name)); - CHKiRet(jsonPathFindParent(pM, name, leaf, &parent, 1)); + CHKiRet(jsonPathFindParent(*jroot, name, leaf, &parent, 1)); leafnode = json_object_object_get(parent, (char*)leaf); DBGPRINTF("AAAA: unset found JSON value path '%s', " "leaf '%s', leafnode %p\n", name, leaf, leafnode); if(leafnode == NULL) { @@ -4012,6 +4024,8 @@ DBGPRINTF("AAAA: unset found JSON value path '%s', " "leaf '%s', leafnode %p\n", } finalize_it: + if(name[0] == '/') + pthread_rwlock_unlock(&glblVars_rwlock); MsgUnlock(pM); RETiRet; } @@ -4033,7 +4047,11 @@ jsonDeepCopy(struct json_object *src) dst = json_object_new_double(json_object_get_double(src)); break; case json_type_int: +#ifdef HAVE_JSON_OBJECT_NEW_INT64 + dst = json_object_new_int64(json_object_get_int64(src)); +#else /* HAVE_JSON_OBJECT_NEW_INT64 */ dst = json_object_new_int(json_object_get_int(src)); +#endif /* HAVE_JSON_OBJECT_NEW_INT64 */ break; case json_type_string: dst = json_object_new_string(json_object_get_string(src)); @@ -4076,7 +4094,11 @@ msgSetJSONFromVar(msg_t *pMsg, uchar *varname, struct var *v) free(cstr); break; case 'N':/* number (integer) */ +#ifdef HAVE_JSON_OBJECT_NEW_INT64 + json = json_object_new_int64(v->d.n); +#else /* HAVE_JSON_OBJECT_NEW_INT64 */ json = json_object_new_int((int) v->d.n); +#endif /* HAVE_JSON_OBJECT_NEW_INT64 */ break; case 'J':/* native JSON */ json = jsonDeepCopy(v->d.json); @@ -4085,11 +4107,76 @@ msgSetJSONFromVar(msg_t *pMsg, uchar *varname, struct var *v) v->datatype); ABORT_FINALIZE(RS_RET_ERR); } - msgAddJSON(pMsg, varname+1, json); + + msgAddJSON(pMsg, varname, json); +finalize_it: + RETiRet; +} + +rsRetVal +MsgAddToStructuredData(msg_t *pMsg, uchar *toadd, rs_size_t len) +{ + uchar *newptr; + rs_size_t newlen; + DEFiRet; + newlen = (pMsg->pszStrucData[0] == '-') ? len : pMsg->lenStrucData + len; + CHKmalloc(newptr = (uchar*) realloc(pMsg->pszStrucData, newlen+1)); + pMsg->pszStrucData = newptr; + if(pMsg->pszStrucData[0] == '-') { /* empty? */ + memcpy(pMsg->pszStrucData, toadd, len); + } else { + memcpy(pMsg->pszStrucData+pMsg->lenStrucData, toadd, len); + } + pMsg->pszStrucData[newlen] = '\0'; + pMsg->lenStrucData = newlen; finalize_it: RETiRet; } + +/* Fill a message propert description. Space must already be alloced + * by the caller. This is for efficiency, as we expect this to happen + * as part of a larger structure alloc. + * Note that CEE/LOCAL_VAR properties can come in either as + * "$!xx"/"$.xx" or "!xx"/".xx" - we will unify them here. + */ +rsRetVal +msgPropDescrFill(msgPropDescr_t *pProp, uchar *name, int nameLen) +{ + propid_t id; + int offs; + DEFiRet; + if(propNameToID(name, &id) != RS_RET_OK) { + parser_errmsg("invalid property '%s'", name); + ABORT_FINALIZE(RS_RET_INVLD_PROP); + } + if(id == PROP_CEE || id == PROP_LOCAL_VAR || id == PROP_GLOBAL_VAR) { + /* in these cases, we need the field name for later processing */ + /* normalize name: remove $ if present */ + offs = (name[0] == '$') ? 1 : 0; + pProp->name = ustrdup(name + offs); + pProp->nameLen = nameLen - offs; + /* we patch the root name, so that support functions do not need to + * check for different root chars. */ + pProp->name[0] = '!'; + } + pProp->id = id; +finalize_it: + RETiRet; +} + +void +msgPropDescrDestruct(msgPropDescr_t *pProp) +{ + if(pProp != NULL) { + if(pProp->id == PROP_CEE || + pProp->id == PROP_LOCAL_VAR || + pProp->id == PROP_GLOBAL_VAR) + free(pProp->name); + } +} + + /* dummy */ rsRetVal msgQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } @@ -4098,6 +4185,8 @@ rsRetVal msgQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } * rgerhards, 2008-01-04 */ BEGINObjClassInit(msg, 1, OBJ_IS_CORE_MODULE) + pthread_rwlock_init(&glblVars_rwlock, NULL); + /* request objects we use */ CHKiRet(objUse(datetime, CORE_COMPONENT)); CHKiRet(objUse(glbl, CORE_COMPONENT)); diff --git a/runtime/msg.h b/runtime/msg.h index e7babdbb..a2392a20 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -3,7 +3,7 @@ * * File begun on 2007-07-13 by RGerhards (extracted from syslogd.c) * - * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -30,6 +30,7 @@ #include <pthread.h> #include <libestr.h> +#include <stdint.h> #include <json.h> #include "obj.h" #include "syslogd-types.h" @@ -85,7 +86,8 @@ struct msg { char *pszTIMESTAMP3339; /* TIMESTAMP as RFC3339 formatted string (32 charcters at most) */ char *pszTIMESTAMP_MySQL;/* TIMESTAMP as MySQL formatted string (always 14 charcters) */ char *pszTIMESTAMP_PgSQL;/* TIMESTAMP as PgSQL formatted string (always 21 characters) */ - cstr_t *pCSStrucData; /* STRUCTURED-DATA */ + uchar *pszStrucData; /* STRUCTURED-DATA */ + uint16_t lenStrucData; /* (cached) length of STRUCTURED-DATA */ cstr_t *pCSAPPNAME; /* APP-NAME */ cstr_t *pCSPROCID; /* PROCID */ cstr_t *pCSMSGID; /* MSGID */ @@ -107,6 +109,7 @@ struct msg { struct syslogTime tRcvdAt;/* time the message entered this program */ struct syslogTime tTIMESTAMP;/* (parsed) value of the timestamp */ struct json_object *json; + struct json_object *localvars; /* some fixed-size buffers to save malloc()/free() for frequently used fields (from the default templates) */ uchar szRawMsg[CONF_RAWMSG_BUFSIZE]; /* most messages are small, and these are stored here (without malloc/free!) */ uchar szHOSTNAME[CONF_HOSTNAME_BUFSIZE]; @@ -124,7 +127,8 @@ struct msg { char pszRcvdAt_SecFrac[7]; /* same as above. Both are fractional seconds for their respective timestamp */ char pszTIMESTAMP_Unix[12]; /* almost as small as a pointer! */ char pszRcvdAt_Unix[12]; - uchar *pszUUID; /* The message's UUID */ + char dfltTZ[8]; /* 7 chars max, less overhead than ptr! */ + uchar *pszUUID; /* The message's UUID */ }; @@ -141,6 +145,9 @@ struct msg { #define NEEDS_ACLCHK_U 0x080 /* check UDP ACLs after DNS resolution has been done in main queue consumer */ #define NO_PRI_IN_RAW 0x100 /* rawmsg does not include a PRI (Solaris!), but PRI is already set correctly in the msg object */ +/* (syslog) protocol types */ +#define MSG_LEGACY_PROTOCOL 0 +#define MSG_RFC5424_PROTOCOL 1 /* function prototypes */ @@ -154,6 +161,7 @@ msg_t* MsgDup(msg_t* pOld); msg_t *MsgAddRef(msg_t *pM); void setProtocolVersion(msg_t *pM, int iNewVersion); void MsgSetInputName(msg_t *pMsg, prop_t*); +void MsgSetDfltTZ(msg_t *pThis, char *tz); rsRetVal MsgSetAPPNAME(msg_t *pMsg, char* pszAPPNAME); rsRetVal MsgSetPROCID(msg_t *pMsg, char* pszPROCID); rsRetVal MsgSetMSGID(msg_t *pMsg, char* pszMSGID); @@ -162,6 +170,8 @@ void MsgSetTAG(msg_t *pMsg, uchar* pszBuf, size_t lenBuf); void MsgSetRuleset(msg_t *pMsg, ruleset_t*); rsRetVal MsgSetFlowControlType(msg_t *pMsg, flowControl_t eFlowCtl); rsRetVal MsgSetStructuredData(msg_t *pMsg, char* pszStrucData); +rsRetVal MsgAddToStructuredData(msg_t *pMsg, uchar *toadd, rs_size_t len); +void MsgGetStructuredData(msg_t *pM, uchar **pBuf, rs_size_t *len); rsRetVal msgSetFromSockinfo(msg_t *pThis, struct sockaddr_storage *sa); void MsgSetRcvFrom(msg_t *pMsg, prop_t*); void MsgSetRcvFromStr(msg_t *pMsg, uchar* pszRcvFrom, int, prop_t **); @@ -173,20 +183,14 @@ void MsgSetMSGoffs(msg_t *pMsg, short offs); void MsgSetRawMsgWOSize(msg_t *pMsg, char* pszRawMsg); void MsgSetRawMsg(msg_t *pMsg, char* pszRawMsg, size_t lenMsg); rsRetVal MsgReplaceMSG(msg_t *pThis, uchar* pszMSG, int lenMSG); -uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, - propid_t propid, es_str_t *propName, +uchar *MsgGetProp(msg_t *pMsg, struct templateEntry *pTpe, msgPropDescr_t *pProp, rs_size_t *pPropLen, unsigned short *pbMustBeFreed, struct syslogTime *ttNow); -rsRetVal msgGetMsgVar(msg_t *pThis, cstr_t *pstrPropName, var_t **ppVar); -es_str_t* msgGetMsgVarNew(msg_t *pThis, uchar *name); uchar *getRcvFrom(msg_t *pM); void getTAG(msg_t *pM, uchar **ppBuf, int *piLen); char *getTimeReported(msg_t *pM, enum tplFormatTypes eFmt); char *getPRI(msg_t *pMsg); void getRawMsg(msg_t *pM, uchar **pBuf, int *piLen); -rsRetVal msgGetCEEVar(msg_t *pThis, cstr_t *propName, var_t **ppVar); -es_str_t* msgGetCEEVarNew(msg_t *pMsg, char *name); rsRetVal msgAddJSON(msg_t *pM, uchar *name, struct json_object *json); -rsRetVal getCEEPropVal(msg_t *pM, es_str_t *propName, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed); rsRetVal MsgGetSeverity(msg_t *pThis, int *piSeverity); rsRetVal MsgDeserialize(msg_t *pMsg, strm_t *pStrm); @@ -202,18 +206,29 @@ char *getHOSTNAME(msg_t *pM); int getHOSTNAMELen(msg_t *pM); uchar *getProgramName(msg_t *pM, sbool bLockMutex); uchar *getRcvFrom(msg_t *pM); -rsRetVal propNameToID(cstr_t *pCSPropName, propid_t *pPropID); +rsRetVal propNameToID(uchar *pName, propid_t *pPropID); uchar *propIDToName(propid_t propID); -rsRetVal msgGetCEEPropJSON(msg_t *pM, es_str_t *propName, struct json_object **pjson); +rsRetVal msgGetJSONPropJSON(msg_t *pMsg, msgPropDescr_t *pProp, struct json_object **pjson); +rsRetVal getJSONPropVal(msg_t *pMsg, msgPropDescr_t *pProp, uchar **pRes, rs_size_t *buflen, unsigned short *pbMustBeFreed); rsRetVal msgSetJSONFromVar(msg_t *pMsg, uchar *varname, struct var *var); rsRetVal msgDelJSON(msg_t *pMsg, uchar *varname); -rsRetVal jsonFind(msg_t *pM, es_str_t *propName, struct json_object **jsonres); +rsRetVal jsonFind(struct json_object *jroot, msgPropDescr_t *pProp, struct json_object **jsonres); -static inline rsRetVal -msgUnsetJSON(msg_t *pMsg, uchar *varname) { - return msgDelJSON(pMsg, varname+1); +rsRetVal msgPropDescrFill(msgPropDescr_t *pProp, uchar *name, int nameLen); +void msgPropDescrDestruct(msgPropDescr_t *pProp); + +static inline int +msgGetProtocolVersion(msg_t *pM) +{ + return(pM->iProtocolVersion); } +/* returns non-zero if the message has structured data, 0 otherwise */ +static inline sbool +MsgHasStructuredData(msg_t *pM) +{ + return (pM->pszStrucData == NULL) ? 0 : 1; +} /* ------------------------------ some inline functions ------------------------------ */ diff --git a/runtime/net.c b/runtime/net.c index 13391cc0..7c180b19 100644 --- a/runtime/net.c +++ b/runtime/net.c @@ -1189,12 +1189,16 @@ void closeUDPListenSockets(int *pSockArr) * hostname and/or pszPort may be NULL, but not both! * bIsServer indicates if a server socket should be created * 1 - server, 0 - client + * param rcvbuf indicates desired rcvbuf size; 0 means OS default */ -int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer) +int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer, int rcvbuf) { struct addrinfo hints, *res, *r; int error, maxs, *s, *socks, on = 1; int sockflags; + int actrcvbuf; + socklen_t optlen; + char errStr[1024]; assert(!((pszPort == NULL) && (hostname == NULL))); memset(&hints, 0, sizeof(hints)); @@ -1297,6 +1301,35 @@ int *create_udp_socket(uchar *hostname, uchar *pszPort, int bIsServer) continue; } + if(rcvbuf != 0) { +# if defined(SO_RCVBUFFORCE) + if(setsockopt(*s, SOL_SOCKET, SO_RCVBUFFORCE, &rcvbuf, sizeof(rcvbuf)) < 0) +# endif + { + /* if we fail, try to do it the regular way. Experiments show that at + * least some platforms do not return an error here, but silently set + * it to the max permitted value. So we do our error check a bit + * differently by querying the size below. + */ + setsockopt(*s, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf)); + } + } + + if(Debug || rcvbuf != 0) { + optlen = sizeof(actrcvbuf); + if(getsockopt(*s, SOL_SOCKET, SO_RCVBUF, &actrcvbuf, &optlen) == 0) { + dbgprintf("socket %d, actual rcvbuf size %d\n", *s, actrcvbuf); + if(rcvbuf != 0 && actrcvbuf/2 != rcvbuf) { + errmsg.LogError(errno, NO_ERRCODE, + "cannot set rcvbuf size %d for socket %d, value now is %d", + rcvbuf, *s, actrcvbuf/2); + } + } else { + dbgprintf("could not obtain rcvbuf size for socket %d: %s\n", + *s, rs_strerror_r(errno, errStr, sizeof(errStr))); + } + } + if(bIsServer) { /* rgerhards, 2007-06-22: if we run on a kernel that does not support * the IPV6_V6ONLY socket option, we need to use a work-around. On such diff --git a/runtime/net.h b/runtime/net.h index b196116b..d7a7b512 100644 --- a/runtime/net.h +++ b/runtime/net.h @@ -137,7 +137,7 @@ BEGINinterface(net) /* name must also be changed in ENDinterface macro! */ void (*PrintAllowedSenders)(int iListToPrint); void (*clearAllowedSenders)(uchar*); void (*debugListenInfo)(int fd, char *type); - int *(*create_udp_socket)(uchar *hostname, uchar *LogPort, int bIsServer); + int *(*create_udp_socket)(uchar *hostname, uchar *LogPort, int bIsServer, int rcvbuf); void (*closeUDPListenSockets)(int *finet); int (*isAllowedSender)(uchar *pszType, struct sockaddr *pFrom, const char *pszFromHost); /* deprecated! */ rsRetVal (*getLocalHostname)(uchar**); diff --git a/runtime/nspoll.c b/runtime/nspoll.c index a936b255..43631f4e 100644 --- a/runtime/nspoll.c +++ b/runtime/nspoll.c @@ -66,7 +66,6 @@ loadDrvr(nspoll_t *pThis) uchar szDrvrName[48]; /* 48 shall be large enough */ pBaseDrvrName = pThis->pBaseDrvrName; - if(pBaseDrvrName == NULL) /* if no drvr name is set, use system default */ pBaseDrvrName = glbl.GetDfltNetstrmDrvr(); if(snprintf((char*)szDrvrName, sizeof(szDrvrName), "lmnsdpoll_%s", pBaseDrvrName) == sizeof(szDrvrName)) ABORT_FINALIZE(RS_RET_DRVRNAME_TOO_LONG); @@ -138,6 +137,29 @@ Wait(nspoll_t *pThis, int timeout, int *numEntries, nsd_epworkset_t workset[]) { } +/* set the base driver name. If the driver name + * is set to NULL, the previously set name is deleted but + * no name set again (which results in the system default being + * used)-- rgerhards, 2008-05-05 + */ +static rsRetVal +SetDrvrName(nspoll_t *pThis, uchar *pszName) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, netstrms); + if(pThis->pBaseDrvrName != NULL) { + free(pThis->pBaseDrvrName); + pThis->pBaseDrvrName = NULL; + } + + if(pszName != NULL) { + CHKmalloc(pThis->pBaseDrvrName = (uchar*) strdup((char*) pszName)); + } +finalize_it: + RETiRet; +} + + /* semantics like the epoll_ctl() function, does the same thing. * rgerhards, 2009-11-18 */ @@ -164,6 +186,7 @@ CODESTARTobjQueryInterface(nspoll) */ pIf->Construct = nspollConstruct; pIf->ConstructFinalize = ConstructFinalize; + pIf->SetDrvrName = SetDrvrName; pIf->Destruct = nspollDestruct; pIf->Wait = Wait; pIf->Ctl = Ctl; diff --git a/runtime/nspoll.h b/runtime/nspoll.h index 037f6c38..3a6e060c 100644 --- a/runtime/nspoll.h +++ b/runtime/nspoll.h @@ -53,8 +53,10 @@ BEGINinterface(nspoll) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Wait)(nspoll_t *pNsdpoll, int timeout, int *numEntries, nsd_epworkset_t workset[]); rsRetVal (*Ctl)(nspoll_t *pNsdpoll, netstrm_t *pStrm, int id, void *pUsr, int mode, int op); rsRetVal (*IsEPollSupported)(void); /* static method */ + /* v3 - 2013-09-17 by rgerhards */ + rsRetVal (*SetDrvrName)(nspoll_t *pThis, uchar *name); ENDinterface(nspoll) -#define nspollCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ +#define nspollCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */ /* interface change in v2 is that wait supports multiple return objects */ /* prototypes */ diff --git a/runtime/nssel.c b/runtime/nssel.c index 751dae9b..6ca0f262 100644 --- a/runtime/nssel.c +++ b/runtime/nssel.c @@ -127,6 +127,29 @@ finalize_it: } +/* set the base driver name. If the driver name + * is set to NULL, the previously set name is deleted but + * no name set again (which results in the system default being + * used)-- rgerhards, 2008-05-05 + */ +static rsRetVal +SetDrvrName(nssel_t *pThis, uchar *pszName) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, netstrms); + if(pThis->pBaseDrvrName != NULL) { + free(pThis->pBaseDrvrName); + pThis->pBaseDrvrName = NULL; + } + + if(pszName != NULL) { + CHKmalloc(pThis->pBaseDrvrName = (uchar*) strdup((char*) pszName)); + } +finalize_it: + RETiRet; +} + + /* Add a stream object to the current select() set. * Note that a single stream may have multiple "sockets" if * it is a listener. If so, all of them are begin added. @@ -195,6 +218,7 @@ CODESTARTobjQueryInterface(nssel) pIf->Construct = nsselConstruct; pIf->ConstructFinalize = ConstructFinalize; pIf->Destruct = nsselDestruct; + pIf->SetDrvrName = SetDrvrName; pIf->Add = Add; pIf->Wait = Wait; pIf->IsReady = IsReady; diff --git a/runtime/nssel.h b/runtime/nssel.h index d7f4fcd3..6131d9b4 100644 --- a/runtime/nssel.h +++ b/runtime/nssel.h @@ -42,8 +42,10 @@ BEGINinterface(nssel) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Add)(nssel_t *pThis, netstrm_t *pStrm, nsdsel_waitOp_t waitOp); rsRetVal (*Wait)(nssel_t *pThis, int *pNumReady); rsRetVal (*IsReady)(nssel_t *pThis, netstrm_t *pStrm, nsdsel_waitOp_t waitOp, int *pbIsReady, int *piNumReady); + /* v2 - 2013-09-17 by rgerhards */ + rsRetVal (*SetDrvrName)(nssel_t *pThis, uchar *name); ENDinterface(nssel) -#define nsselCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define nsselCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ /* prototypes */ PROTOTYPEObj(nssel); diff --git a/runtime/queue.c b/runtime/queue.c index 073c6823..8fb734e7 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -12,7 +12,7 @@ * function names - this makes it really hard to read and does not provide much * benefit, at least I (now) think so... * - * Copyright 2008-2011 Rainer Gerhards and Adiscon GmbH. + * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -85,7 +85,6 @@ static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiS static rsRetVal qAddDirect(qqueue_t *pThis, msg_t *pMsg); static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qConstructDirect(qqueue_t __attribute__((unused)) *pThis); -static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis); static rsRetVal qDestructDisk(qqueue_t *pThis); /* some constants for queuePersist () */ @@ -118,6 +117,7 @@ static struct cnfparamdescr cnfpdescr[] = { { "queue.dequeueslowdown", eCmdHdlrInt, 0 }, { "queue.dequeuetimebegin", eCmdHdlrInt, 0 }, { "queue.dequeuetimeend", eCmdHdlrInt, 0 }, + { "queue.cry.provider", eCmdHdlrGetWord, 0 } }; static struct cnfparamblk pblk = { CNFPARAMBLK_VERSION, @@ -350,16 +350,15 @@ qqueueAdviseMaxWorkers(qqueue_t *pThis) if(pThis->bIsDA && getLogicalQueueSize(pThis) >= pThis->iHighWtrMrk) { DBGOPRINT((obj_t*) pThis, "(re)activating DA worker\n"); wtpAdviseMaxWorkers(pThis->pWtpDA, 1); /* disk queues have always one worker */ + } + if(getLogicalQueueSize(pThis) == 0) { + iMaxWorkers = 0; + } else if(pThis->qType == QUEUETYPE_DISK || pThis->iMinMsgsPerWrkr == 0) { + iMaxWorkers = 1; } else { - if(getLogicalQueueSize(pThis) == 0) { - iMaxWorkers = 0; - } else if(pThis->qType == QUEUETYPE_DISK || pThis->iMinMsgsPerWrkr == 0) { - iMaxWorkers = 1; - } else { - iMaxWorkers = getLogicalQueueSize(pThis) / pThis->iMinMsgsPerWrkr + 1; - } - wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers); + iMaxWorkers = getLogicalQueueSize(pThis) / pThis->iMinMsgsPerWrkr + 1; } + wtpAdviseMaxWorkers(pThis->pWtpReg, iMaxWorkers); } RETiRet; @@ -483,7 +482,6 @@ InitDA(qqueue_t *pThis, int bLockMutex) CHKiRet(wtpSetpfDoWork (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, void *pWti)) ConsumerDA)); CHKiRet(wtpSetpfObjProcessed (pThis->pWtpDA, (rsRetVal (*)(void *pUsr, wti_t *pWti)) batchProcessed)); CHKiRet(wtpSetpmutUsr (pThis->pWtpDA, pThis->mut)); - CHKiRet(wtpSetpcondBusy (pThis->pWtpDA, &pThis->notEmpty)); CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpDA, 1)); CHKiRet(wtpSettoWrkShutdown (pThis->pWtpDA, pThis->toWrkShutdown)); CHKiRet(wtpSetpUsr (pThis->pWtpDA, pThis)); @@ -706,9 +704,13 @@ queueSwitchToEmergencyMode(qqueue_t *pThis, rsRetVal initiatingError) pThis->qType = QUEUETYPE_DIRECT; pThis->qConstruct = qConstructDirect; pThis->qDestruct = qDestructDirect; + /* these entry points shall not be used in direct mode + * To catch program errors, make us abort if that happens! + * rgerhards, 2013-11-05 + */ pThis->qAdd = qAddDirect; - pThis->qDel = qDelDirect; pThis->MultiEnq = qqueueMultiEnqObjDirect; + pThis->qDel = NULL; if(pThis->pqParent != NULL) { DBGOPRINT((obj_t*) pThis, "DA queue is in emergency mode, disabling DA in parent\n"); pThis->pqParent->bIsDA = 0; @@ -776,11 +778,19 @@ qqueueTryLoadPersistedInfo(qqueue_t *pThis) (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); CHKiRet(obj.Deserialize(&pThis->tVars.disk.pReadDel, (uchar*) "strm", psQIF, (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); - /* create a duplicate for the read "pointer". */ CHKiRet(strm.Dup(pThis->tVars.disk.pReadDel, &pThis->tVars.disk.pReadDeq)); CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDeq, 0)); /* deq must NOT delete the files! */ CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDeq)); + /* if we use a crypto provider, we need to amend the objects with it's info */ + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pWrite, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pWrite, pThis->cryprovData)); + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDeq, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDeq, pThis->cryprovData)); + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDel, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDel, pThis->cryprovData)); + } CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pWrite)); CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pReadDel)); @@ -834,6 +844,10 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pWrite, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pWrite, STREAMMODE_WRITE)); CHKiRet(strm.SetsType(pThis->tVars.disk.pWrite, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pWrite, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pWrite, pThis->cryprovData)); + } CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pWrite)); CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDeq)); @@ -842,6 +856,10 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDeq, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDeq, STREAMMODE_READ)); CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDeq, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDeq, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDeq, pThis->cryprovData)); + } CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDeq)); CHKiRet(strm.Construct(&pThis->tVars.disk.pReadDel)); @@ -851,6 +869,10 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pReadDel, 10000000)); CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pReadDel, STREAMMODE_READ)); CHKiRet(strm.SetsType(pThis->tVars.disk.pReadDel, STREAMTYPE_FILE_CIRCULAR)); + if(pThis->useCryprov) { + CHKiRet(strm.Setcryprov(pThis->tVars.disk.pReadDel, &pThis->cryprov)); + CHKiRet(strm.SetcryprovData(pThis->tVars.disk.pReadDel, pThis->cryprovData)); + } CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pReadDel)); CHKiRet(strm.SetFName(pThis->tVars.disk.pWrite, pThis->pszFilePrefix, pThis->lenFilePrefix)); @@ -938,13 +960,11 @@ static rsRetVal qDestructDirect(qqueue_t __attribute__((unused)) *pThis) return RS_RET_OK; } -static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) +static rsRetVal qAddDirectWithWti(qqueue_t *pThis, msg_t* pMsg, wti_t *pWti) { batch_t singleBatch; batch_obj_t batchObj; batch_state_t batchState = BATCH_STATE_RDY; - sbool active = 1; - int i; DEFiRet; //TODO: init batchObj (states _OK and new fields -- CHECK) @@ -964,46 +984,29 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) singleBatch.nElem = 1; /* there always is only one in direct mode */ singleBatch.pElem = &batchObj; singleBatch.eltState = &batchState; - singleBatch.active = &active; - iRet = pThis->pConsumer(pThis->pAction, &singleBatch, &pThis->bShutdownImmediate); - /* delete the batch string params: TODO: create its own "class" for this */ - for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) { - free(batchObj.staticActStrings[i]); - } + iRet = pThis->pConsumer(pThis->pAction, &singleBatch, pWti); msgDestruct(&pMsg); RETiRet; } -/* "enqueue" a batch in direct mode. This is a shortcut which saves all the overhead - * otherwise incured. -- rgerhards, ~2010-06-23 +/* this is called if we do not have a pWti. This currently only happens + * when we are called from a main queue in direct mode. If so, we need + * to obtain a dummy pWti. */ -rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch) +static rsRetVal +qAddDirect(qqueue_t *pThis, msg_t* pMsg) { + wti_t *pWti; DEFiRet; - ASSERT(pThis != NULL); - - /* calling the consumer is quite different here than it is from a worker thread */ - /* we need to provide the consumer's return value back to the caller because in direct - * mode the consumer probably has a lot to convey (which get's lost in the other modes - * because they are asynchronous. But direct mode is deliberately synchronous. - * rgerhards, 2008-02-12 - * We use our knowledge about the batch_t structure below, but without that, we - * pay a too-large performance toll... -- rgerhards, 2009-04-22 - */ - iRet = pThis->pConsumer(pThis->pAction, pBatch, NULL); - + pWti = wtiGetDummy(); + pWti->pbShutdownImmediate = &pThis->bShutdownImmediate; + iRet = qAddDirectWithWti(pThis, pMsg, pWti); RETiRet; } -static rsRetVal qDelDirect(qqueue_t __attribute__((unused)) *pThis) -{ - return RS_RET_OK; -} - - /* --------------- end type-specific handlers -------------------- */ @@ -1298,7 +1301,7 @@ finalize_it: * to modify some parameters before the queue is actually started. */ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*,int*)) + int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*)) { DEFiRet; qqueue_t *pThis; @@ -1321,6 +1324,7 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread pThis->iMaxFileSize = 1024 * 1024; /* default is 1 MiB */ pThis->iQueueSize = 0; pThis->nLogDeq = 0; + pThis->useCryprov = 0; pThis->iMaxQueueSize = iMaxQueueSize; pThis->pConsumer = pConsumer; pThis->iNumWorkerThreads = iWorkerThreads; @@ -1856,7 +1860,8 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave); - CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, &pThis->bShutdownImmediate)); + pWti->pbShutdownImmediate = &pThis->bShutdownImmediate; + CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti)); /* we now need to check if we should deliberately delay processing a bit * and, if so, do that. -- rgerhards, 2008-01-30 @@ -2078,9 +2083,13 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ case QUEUETYPE_DIRECT: pThis->qConstruct = qConstructDirect; pThis->qDestruct = qDestructDirect; + /* these entry points shall not be used in direct mode + * To catch program errors, make us abort if that happens! + * rgerhards, 2013-11-05 + */ pThis->qAdd = qAddDirect; - pThis->qDel = qDelDirect; pThis->MultiEnq = qqueueMultiEnqObjDirect; + pThis->qDel = NULL; break; } @@ -2118,7 +2127,6 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ pthread_mutex_init(&pThis->mutThrdMgmt, NULL); pthread_cond_init (&pThis->notFull, NULL); - pthread_cond_init (&pThis->notEmpty, NULL); pthread_cond_init (&pThis->belowFullDlyWtrMrk, NULL); pthread_cond_init (&pThis->belowLightDlyWtrMrk, NULL); @@ -2136,11 +2144,13 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ } DBGOPRINT((obj_t*) pThis, "type %d, enq-only %d, disk assisted %d, maxFileSz %lld, maxQSize %d, lqsize %d, pqsize %d, child %d, " - "full delay %d, light delay %d, deq batch size %d starting, high wtrrmrk %d, low wtrmrk %d\n", + "full delay %d, light delay %d, deq batch size %d starting, high wtrmrk %d, low wtrmrk %d\n" + "max wrkr %d, min msgs f. wrkr %d\n", pThis->qType, pThis->bEnqOnly, pThis->bIsDA, pThis->iMaxFileSize, pThis->iMaxQueueSize, getLogicalQueueSize(pThis), getPhysicalQueueSize(pThis), pThis->pqParent == NULL ? 0 : 1, pThis->iFullDlyMrk, pThis->iLightDlyMrk, - pThis->iDeqBatchSize, pThis->iHighWtrMrk, pThis->iLowWtrMrk); + pThis->iDeqBatchSize, pThis->iHighWtrMrk, pThis->iLowWtrMrk, + pThis->iNumWorkerThreads, pThis->iMinMsgsPerWrkr); pThis->bQueueStarted = 1; if(pThis->qType == QUEUETYPE_DIRECT) @@ -2157,7 +2167,6 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ CHKiRet(wtpSetpfDoWork (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, void *pWti)) ConsumerReg)); CHKiRet(wtpSetpfObjProcessed (pThis->pWtpReg, (rsRetVal (*)(void *pUsr, wti_t *pWti)) batchProcessed)); CHKiRet(wtpSetpmutUsr (pThis->pWtpReg, pThis->mut)); - CHKiRet(wtpSetpcondBusy (pThis->pWtpReg, &pThis->notEmpty)); CHKiRet(wtpSetiNumWorkerThreads (pThis->pWtpReg, pThis->iNumWorkerThreads)); CHKiRet(wtpSettoWrkShutdown (pThis->pWtpReg, pThis->toWrkShutdown)); CHKiRet(wtpSetpUsr (pThis->pWtpReg, pThis)); @@ -2181,26 +2190,26 @@ qqueueStart(qqueue_t *pThis) /* this is the ConstructionFinalizer */ /* we need to save the queue size, as the stats module initializes it to 0! */ /* iQueueSize is a dual-use counter: no init, no mutex! */ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("size"), - ctrType_Int, &pThis->iQueueSize)); + ctrType_Int, CTR_FLAG_NONE, &pThis->iQueueSize)); STATSCOUNTER_INIT(pThis->ctrEnqueued, pThis->mutCtrEnqueued); CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("enqueued"), - ctrType_IntCtr, &pThis->ctrEnqueued)); + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrEnqueued)); STATSCOUNTER_INIT(pThis->ctrFull, pThis->mutCtrFull); CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("full"), - ctrType_IntCtr, &pThis->ctrFull)); + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrFull)); STATSCOUNTER_INIT(pThis->ctrFDscrd, pThis->mutCtrFDscrd); CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.full"), - ctrType_IntCtr, &pThis->ctrFDscrd)); + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrFDscrd)); STATSCOUNTER_INIT(pThis->ctrNFDscrd, pThis->mutCtrNFDscrd); CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("discarded.nf"), - ctrType_IntCtr, &pThis->ctrNFDscrd)); + ctrType_IntCtr, CTR_FLAG_RESETTABLE, &pThis->ctrNFDscrd)); pThis->ctrMaxqsize = 0; /* no mutex needed, thus no init call */ CHKiRet(statsobj.AddCounter(pThis->statsobj, UCHAR_CONSTANT("maxqsize"), - ctrType_Int, &pThis->ctrMaxqsize)); + ctrType_Int, CTR_FLAG_NONE, &pThis->ctrMaxqsize)); CHKiRet(statsobj.ConstructFinalize(pThis->statsobj)); @@ -2422,7 +2431,6 @@ CODESTARTobjDestruct(qqueue) } pthread_mutex_destroy(&pThis->mutThrdMgmt); pthread_cond_destroy(&pThis->notFull); - pthread_cond_destroy(&pThis->notEmpty); pthread_cond_destroy(&pThis->belowFullDlyWtrMrk); pthread_cond_destroy(&pThis->belowLightDlyWtrMrk); @@ -2435,6 +2443,13 @@ CODESTARTobjDestruct(qqueue) free(pThis->pszFilePrefix); free(pThis->pszSpoolDir); + if(pThis->useCryprov) { + pThis->cryprov.Destruct(&pThis->cryprovData); + obj.ReleaseObj(__FILE__, pThis->cryprovNameFull+2, pThis->cryprovNameFull, + (void*) &pThis->cryprov); + free(pThis->cryprovName); + free(pThis->cryprovNameFull); + } /* some queues do not provide stats and thus have no statsobj! */ if(pThis->statsobj != NULL) @@ -2656,13 +2671,14 @@ static rsRetVal qqueueMultiEnqObjDirect(qqueue_t *pThis, multi_submit_t *pMultiSub) { int i; + wti_t *pWti; DEFiRet; - ISOBJ_TYPE_assert(pThis, qqueue); - assert(pMultiSub != NULL); + pWti = wtiGetDummy(); + pWti->pbShutdownImmediate = &pThis->bShutdownImmediate; for(i = 0 ; i < pMultiSub->nElem ; ++i) { - CHKiRet(qAddDirect(pThis, (void*)pMultiSub->ppMsgs[i])); + CHKiRet(qAddDirectWithWti(pThis, (void*)pMultiSub->ppMsgs[i], pWti)); } finalize_it: @@ -2671,22 +2687,7 @@ finalize_it: /* ------------------------------ END multi-enqueue functions ------------------------------ */ -/* enqueue a new user data element in direct mode - * NOTE/TODO: This is a TESTER/EXPERIEMENTAL, to be changed to better - * code later on (like multi submit!) 2010-06-10 - * Enqueues the new element and awakes worker thread. - */ -rsRetVal -qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg) -{ - DEFiRet; - ISOBJ_TYPE_assert(pThis, qqueue); - iRet = qAddDirect(pThis, pMsg); - RETiRet; -} - - -/* enqueue a new user data element +/* enqueue a new user data element * Enqueues the new element and awakes worker thread. */ rsRetVal @@ -2720,27 +2721,67 @@ finalize_it: } -/* take v6 config list and extract the queue params out of it. Hand the - * param values back to the caller. Caller is responsible for destructing - * them when no longer needed. Caller can use this param block to configure - * all parameters for a newly created queue with one call to qqueueSetParams(). - * rgerhards, 2011-07-22 +/* are any queue params set at all? 1 - yes, 0 - no + * We need to evaluate the param block for this function, which is somewhat + * inefficient. HOWEVER, this is only done during config load, so we really + * don't care... -- rgerhards, 2013-05-10 */ -rsRetVal -qqueueDoCnfParams(struct nvlst *lst, struct cnfparamvals **ppvals) +int +queueCnfParamsSet(struct nvlst *lst) { - *ppvals = nvlstGetParams(lst, &pblk, NULL); - return RS_RET_OK; + int r; + struct cnfparamvals *pvals; + + pvals = nvlstGetParams(lst, &pblk, NULL); + r = cnfparamvalsIsSet(&pblk, pvals); + cnfparamvalsDestruct(pvals, &pblk); + return r; } -/* are any queue params set at all? 1 - yes, 0 - no */ -int -queueCnfParamsSet(struct cnfparamvals *pvals) +static inline rsRetVal +initCryprov(qqueue_t *pThis, struct nvlst *lst) { - return cnfparamvalsIsSet(&pblk, pvals); -} + uchar szDrvrName[1024]; + DEFiRet; + if(snprintf((char*)szDrvrName, sizeof(szDrvrName), "lmcry_%s", pThis->cryprovName) + == sizeof(szDrvrName)) { + errmsg.LogError(0, RS_RET_ERR, "queue: crypto provider " + "name is too long: '%s' - encryption disabled", + pThis->cryprovName); + ABORT_FINALIZE(RS_RET_ERR); + } + pThis->cryprovNameFull = ustrdup(szDrvrName); + + pThis->cryprov.ifVersion = cryprovCURR_IF_VERSION; + /* The pDrvrName+2 below is a hack to obtain the object name. It + * safes us to have yet another variable with the name without "lm" in + * front of it. If we change the module load interface, we may re-think + * about this hack, but for the time being it is efficient and clean enough. + */ + if(obj.UseObj(__FILE__, szDrvrName, szDrvrName, (void*) &pThis->cryprov) + != RS_RET_OK) { + errmsg.LogError(0, RS_RET_LOAD_ERROR, "queue: could not load " + "crypto provider '%s' - encryption disabled", + szDrvrName); + ABORT_FINALIZE(RS_RET_CRYPROV_ERR); + } + + if(pThis->cryprov.Construct(&pThis->cryprovData) != RS_RET_OK) { + errmsg.LogError(0, RS_RET_CRYPROV_ERR, "queue: error constructing " + "crypto provider %s dataset - encryption disabled", + szDrvrName); + ABORT_FINALIZE(RS_RET_CRYPROV_ERR); + } + CHKiRet(pThis->cryprov.SetCnfParam(pThis->cryprovData, lst, CRYPROV_PARAMTYPE_DISK)); + + dbgprintf("loaded crypto provider %s, data instance at %p\n", + szDrvrName, pThis->cryprovData); + pThis->useCryprov = 1; +finalize_it: + RETiRet; +} /* apply all params from param block to queue. Must be called before * finalizing. This supports the v6 config system. Defaults were already @@ -2748,15 +2789,24 @@ queueCnfParamsSet(struct cnfparamvals *pvals) * function. */ rsRetVal -qqueueApplyCnfParam(qqueue_t *pThis, struct cnfparamvals *pvals) +qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst) { int i; + struct cnfparamvals *pvals; + + pvals = nvlstGetParams(lst, &pblk, NULL); + if(Debug) { + dbgprintf("queue param blk:\n"); + cnfparamsPrint(&pblk, pvals); + } for(i = 0 ; i < pblk.nParams ; ++i) { if(!pvals[i].bUsed) continue; if(!strcmp(pblk.descr[i].name, "queue.filename")) { pThis->pszFilePrefix = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); pThis->lenFilePrefix = es_strlen(pvals[i].val.d.estr); + } else if(!strcmp(pblk.descr[i].name, "queue.cry.provider")) { + pThis->cryprovName = (uchar*) es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(pblk.descr[i].name, "queue.size")) { pThis->iMaxQueueSize = pvals[i].val.d.n; } else if(!strcmp(pblk.descr[i].name, "queue.dequeuebatchsize")) { @@ -2808,12 +2858,27 @@ qqueueApplyCnfParam(qqueue_t *pThis, struct cnfparamvals *pvals) "param '%s'\n", pblk.descr[i].name); } } - if(pThis->qType == QUEUETYPE_DISK && pThis->pszFilePrefix == NULL) { - errmsg.LogError(0, RS_RET_QUEUE_DISK_NO_FN, "error on queue '%s', disk mode selected, but " - "no queue file name given; queue type changed to 'linkedList'", + if(pThis->qType == QUEUETYPE_DISK) { + if(pThis->pszFilePrefix == NULL) { + errmsg.LogError(0, RS_RET_QUEUE_DISK_NO_FN, "error on queue '%s', disk mode selected, but " + "no queue file name given; queue type changed to 'linkedList'", + obj.GetName((obj_t*) pThis)); + pThis->qType = QUEUETYPE_LINKEDLIST; + } + } + + if(pThis->pszFilePrefix == NULL && pThis->cryprovName != NULL) { + errmsg.LogError(0, RS_RET_QUEUE_CRY_DISK_ONLY, "error on queue '%s', crypto provider can " + "only be set for disk or disk assisted queue - ignored", obj.GetName((obj_t*) pThis)); - pThis->qType = QUEUETYPE_LINKEDLIST; + free(pThis->cryprovName); + pThis->cryprovName = NULL; + } + + if(pThis->cryprovName != NULL) { + initCryprov(pThis, lst); } + cnfparamvalsDestruct(pvals, &pblk); return RS_RET_OK; } diff --git a/runtime/queue.h b/runtime/queue.h index 886fac8d..86107d24 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -30,6 +30,7 @@ #include "batch.h" #include "stream.h" #include "statsobj.h" +#include "cryprov.h" /* support for the toDelete list */ typedef struct toDeleteLst_s toDeleteLst_t; @@ -102,11 +103,10 @@ struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,batch_t*,int*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,batch_t*, wti_t*); /* user-supplied consumer function for dequeued messages */ /* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the * user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 - * is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero - * during normal operations and one if the consumer must urgently shut down. + * is pointer to an array of message message pointers) */ /* type-specific handlers (set during construction) */ rsRetVal (*qConstruct)(struct queue_s *pThis); @@ -121,7 +121,7 @@ struct queue_s { /* synchronization variables */ pthread_mutex_t mutThrdMgmt; /* mutex for the queue's thread management */ pthread_mutex_t *mut; /* mutex for enqueing and dequeueing messages */ - pthread_cond_t notFull, notEmpty; + pthread_cond_t notFull; pthread_cond_t belowFullDlyWtrMrk; /* below eFLOWCTL_FULL_DELAY watermark */ pthread_cond_t belowLightDlyWtrMrk; /* below eFLOWCTL_FULL_DELAY watermark */ int bThrdStateChanged; /* at least one thread state has changed if 1 */ @@ -168,6 +168,11 @@ struct queue_s { strm_t *pReadDel; /* current file for deleting */ } disk; } tVars; + sbool useCryprov; /* quicker than checkig ptr (1 vs 8 bytes!) */ + uchar *cryprovName; /* crypto provider to use */ + cryprov_if_t cryprov; /* ptr to crypto provider interface */ + void *cryprovData; /* opaque data ptr for provider use */ + uchar *cryprovNameFull;/* full internal crypto provider name */ DEF_ATOMIC_HELPER_MUT(mutQueueSize); DEF_ATOMIC_HELPER_MUT(mutLogDeq); /* for statistics subsystem */ @@ -189,17 +194,14 @@ struct queue_s { /* prototypes */ rsRetVal qqueueDestruct(qqueue_t **ppThis); -rsRetVal qqueueEnqMsgDirect(qqueue_t *pThis, msg_t *pMsg); rsRetVal qqueueEnqMsg(qqueue_t *pThis, flowControl_t flwCtlType, msg_t *pMsg); rsRetVal qqueueStart(qqueue_t *pThis); rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize); rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*)); -rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch); -rsRetVal qqueueDoCnfParams(struct nvlst *lst, struct cnfparamvals **ppvals); -int queueCnfParamsSet(struct cnfparamvals *pvals); -rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct cnfparamvals *pvals); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *)); +int queueCnfParamsSet(struct nvlst *lst); +rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst); void qqueueSetDefaultsRulesetQueue(qqueue_t *pThis); void qqueueSetDefaultsActionQueue(qqueue_t *pThis); void qqueueDbgPrint(qqueue_t *pThis); diff --git a/runtime/rsconf.c b/runtime/rsconf.c index d8b81f1b..f2feb989 100644 --- a/runtime/rsconf.c +++ b/runtime/rsconf.c @@ -2,7 +2,7 @@ * * Module begun 2011-04-19 by Rainer Gerhards * - * Copyright 2011-2012 Adiscon GmbH. + * Copyright 2011-2013 Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -83,7 +83,7 @@ rsconf_t *runConf = NULL;/* the currently running config */ rsconf_t *loadConf = NULL;/* the config currently being loaded (no concurrent config load supported!) */ /* 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%'\ninputname: %inputname% rawmsg: '%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$.:%$.%\n$/:%$/%\n\n\""; static uchar template_SyslogProtocol23Format[] = "\"<%PRI%>1 %TIMESTAMP:::date-rfc3339% %HOSTNAME% %APP-NAME% %PROCID% %MSGID% %STRUCTURED-DATA% %msg%\n\""; static uchar template_TraditionalFileFormat[] = "=RSYSLOG_TraditionalFileFormat"; static uchar template_FileFormat[] = "=RSYSLOG_FileFormat"; @@ -124,15 +124,16 @@ BEGINobjConstruct(rsconf) /* be sure to specify the object type also in END macr pThis->templates.last = NULL; pThis->templates.lastStatic = NULL; pThis->actions.nbrActions = 0; + lookupInitCnf(&pThis->lu_tabs); CHKiRet(llInit(&pThis->rulesets.llRulesets, rulesetDestructForLinkedList, rulesetKeyDestruct, strcasecmp)); /* queue params */ - pThis->globals.mainQ.iMainMsgQueueSize = 10000; - pThis->globals.mainQ.iMainMsgQHighWtrMark = 8000; - pThis->globals.mainQ.iMainMsgQLowWtrMark = 2000; - pThis->globals.mainQ.iMainMsgQDiscardMark = 9800; + pThis->globals.mainQ.iMainMsgQueueSize = 100000; + pThis->globals.mainQ.iMainMsgQHighWtrMark = 80000; + pThis->globals.mainQ.iMainMsgQLowWtrMark = 20000; + pThis->globals.mainQ.iMainMsgQDiscardMark = 98000; pThis->globals.mainQ.iMainMsgQDiscardSeverity = 8; - pThis->globals.mainQ.iMainMsgQueueNumWorkers = 1; + pThis->globals.mainQ.iMainMsgQueueNumWorkers = 2; pThis->globals.mainQ.MainMsgQueType = QUEUETYPE_FIXED_ARRAY; pThis->globals.mainQ.pszMainMsgQFName = NULL; pThis->globals.mainQ.iMainMsgQueMaxFileSize = 1024*1024; @@ -142,10 +143,10 @@ BEGINobjConstruct(rsconf) /* be sure to specify the object type also in END macr pThis->globals.mainQ.iMainMsgQtoActShutdown = 1000; pThis->globals.mainQ.iMainMsgQtoEnq = 2000; pThis->globals.mainQ.iMainMsgQtoWrkShutdown = 60000; - pThis->globals.mainQ.iMainMsgQWrkMinMsgs = 100; + pThis->globals.mainQ.iMainMsgQWrkMinMsgs = 40000; pThis->globals.mainQ.iMainMsgQDeqSlowdown = 0; pThis->globals.mainQ.iMainMsgQueMaxDiskSpace = 0; - pThis->globals.mainQ.iMainMsgQueDeqBatchSize = 32; + pThis->globals.mainQ.iMainMsgQueDeqBatchSize = 256; pThis->globals.mainQ.bMainMsgQSaveOnShutdown = 1; pThis->globals.mainQ.iMainMsgQueueDeqtWinFromHr = 0; pThis->globals.mainQ.iMainMsgQueueDeqtWinToHr = 25; @@ -253,92 +254,6 @@ CODESTARTobjDebugPrint(rsconf) ENDobjDebugPrint(rsconf) -/* This function returns the current date in different - * variants. It is used to construct the $NOW series of - * system properties. The returned buffer must be freed - * by the caller when no longer needed. If the function - * can not allocate memory, it returns a NULL pointer. - * TODO: this was taken from msg.c and we should consolidate it with the code - * there. This is especially important when we increase the number of system - * variables (what we definitely want to do). - */ -typedef enum ENOWType { NOW_NOW, NOW_YEAR, NOW_MONTH, NOW_DAY, NOW_HOUR, NOW_MINUTE } eNOWType; -static rsRetVal -getNOW(eNOWType eNow, es_str_t **estr) -{ - DEFiRet; - uchar szBuf[16]; - struct syslogTime t; - es_size_t len; - - datetime.getCurrTime(&t, NULL); - switch(eNow) { - case NOW_NOW: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), - "%4.4d-%2.2d-%2.2d", t.year, t.month, t.day); - break; - case NOW_YEAR: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%4.4d", t.year); - break; - case NOW_MONTH: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.month); - break; - case NOW_DAY: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.day); - break; - case NOW_HOUR: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.hour); - break; - case NOW_MINUTE: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "%2.2d", t.minute); - break; - default: - len = snprintf((char*) szBuf, sizeof(szBuf)/sizeof(uchar), "*invld eNow*"); - break; - } - - /* now create a string object out of it and hand that over to the var */ - *estr = es_newStrFromCStr((char*)szBuf, len); - - RETiRet; -} - - - -static inline es_str_t * -getSysVar(char *name) -{ - es_str_t *estr = NULL; - rsRetVal iRet = RS_RET_OK; - - if(!strcmp(name, "now")) { - CHKiRet(getNOW(NOW_NOW, &estr)); - } else if(!strcmp(name, "year")) { - CHKiRet(getNOW(NOW_YEAR, &estr)); - } else if(!strcmp(name, "month")) { - CHKiRet(getNOW(NOW_MONTH, &estr)); - } else if(!strcmp(name, "day")) { - CHKiRet(getNOW(NOW_DAY, &estr)); - } else if(!strcmp(name, "hour")) { - CHKiRet(getNOW(NOW_HOUR, &estr)); - } else if(!strcmp(name, "minute")) { - CHKiRet(getNOW(NOW_MINUTE, &estr)); - } else if(!strcmp(name, "myhostname")) { - char *hn = (char*)glbl.GetLocalHostName(); - estr = es_newStrFromCStr(hn, strlen(hn)); - } else { - ABORT_FINALIZE(RS_RET_SYSVAR_NOT_FOUND); - } -finalize_it: - if(iRet != RS_RET_OK) { - dbgprintf("getSysVar error iRet %d\n", iRet); - if(estr == NULL) - estr = es_newStrFromCStr("*ERROR*", sizeof("*ERROR*") - 1); - } - return estr; -} - - /* Process input() objects */ rsRetVal inputProcessCnf(struct cnfobj *o) @@ -377,6 +292,21 @@ finalize_it: extern int yylineno; void +parser_warnmsg(char *fmt, ...) +{ + va_list ap; + char errBuf[1024]; + + va_start(ap, fmt); + if(vsnprintf(errBuf, sizeof(errBuf), fmt, ap) == sizeof(errBuf)) + errBuf[sizeof(errBuf)-1] = '\0'; + errmsg.LogError(0, RS_RET_CONF_PARSE_WARNING, + "warning during parsing file %s, on or before line %d: %s", + cnfcurrfn, yylineno, errBuf); + va_end(ap); +} + +void parser_errmsg(char *fmt, ...) { va_list ap; @@ -399,6 +329,7 @@ yyerror(char *s) } void cnfDoObj(struct cnfobj *o) { + int bDestructObj = 1; int bChkUnuse = 1; dbgprintf("cnf:global:obj: "); @@ -407,12 +338,19 @@ void cnfDoObj(struct cnfobj *o) case CNFOBJ_GLOBAL: glblProcessCnf(o); break; + case CNFOBJ_MAINQ: + glblProcessMainQCnf(o); + bDestructObj = 0; + break; case CNFOBJ_MODULE: modulesProcessCnf(o); break; case CNFOBJ_INPUT: inputProcessCnf(o); break; + case CNFOBJ_LOOKUP_TABLE: + lookupProcessCnf(o); + break; case CNFOBJ_TPL: if(tplProcessCnf(o) != RS_RET_OK) parser_errmsg("error processing template object"); @@ -430,9 +368,11 @@ void cnfDoObj(struct cnfobj *o) o->objType); break; } - if(bChkUnuse) - nvlstChkUnused(o->nvlst); - cnfobjDestruct(o); + if(bDestructObj) { + if(bChkUnuse) + nvlstChkUnused(o->nvlst); + cnfobjDestruct(o); + } } void cnfDoScript(struct cnfstmt *script) @@ -468,30 +408,6 @@ void cnfDoBSDHost(char *ln) "solution (Block '%s')", ln); free(ln); } - -es_str_t* -cnfGetVar(char *name, void *usrptr) -{ - es_str_t *estr; - if(name[0] == '$') { - if(name[1] == '$') - estr = getSysVar(name+2); - else if(name[1] == '!') - estr = msgGetCEEVarNew((msg_t*) usrptr, name+2); - else - estr = msgGetMsgVarNew((msg_t*) usrptr, (uchar*)name+1); - } else { /* if this happens, we have a program logic error */ - estr = es_newStrFromCStr("err: var must start with $", - strlen("err: var must start with $")); - } - if(Debug) { - char *s; - s = es_str2cstr(estr, NULL); - dbgprintf("rainerscript: var '%s': '%s'\n", name, s); - free(s); - } - return estr; -} /*------------------------------ end interface to flex/bison parser ------------------------------*/ @@ -757,9 +673,18 @@ startInputModules(void) static inline rsRetVal activateMainQueue() { + struct cnfobj *mainqCnfObj; DEFiRet; + + mainqCnfObj = glbl.GetmainqCnfObj(); + DBGPRINTF("activateMainQueue: mainq cnf obj ptr is %p\n", mainqCnfObj); /* create message queue */ - CHKiRet_Hdlr(createMainQueue(&pMsgQueue, UCHAR_CONSTANT("main Q"), NULL)) { + iRet = createMainQueue(&pMsgQueue, UCHAR_CONSTANT("main Q"), + (mainqCnfObj == NULL) ? NULL : mainqCnfObj->nvlst); + if(iRet == RS_RET_OK) { + iRet = startMainQueue(pMsgQueue); + } + if(iRet != RS_RET_OK) { /* no queue is fatal, we need to give up in that case... */ fprintf(stderr, "fatal error %d: could not create message queue - rsyslogd can not run!\n", iRet); FINALIZE; @@ -768,6 +693,7 @@ activateMainQueue() bHaveMainQueue = (ourConf->globals.mainQ.MainMsgQueType == QUEUETYPE_DIRECT) ? 0 : 1; DBGPRINTF("Main processing queue is initialized and running\n"); finalize_it: + glblDestructMainqCnfObj(); RETiRet; } @@ -821,6 +747,7 @@ activate(rsconf_t *cnf) tellModulesActivateConfig(); startInputModules(); CHKiRet(activateActions()); + CHKiRet(activateRulesetQueues()); CHKiRet(activateMainQueue()); /* finally let the inputs run... */ runInputModules(); @@ -1298,6 +1225,7 @@ ourConf = loadConf; // TODO: remove, once ourConf is gone! ABORT_FINALIZE(RS_RET_NO_ACTIONS); } tellLexEndParsing(); + DBGPRINTF("Number of actions in this configuration: %d\n", iActionNbr); rulesetOptimizeAll(loadConf); tellCoreConfigLoadDone(); diff --git a/runtime/rsconf.h b/runtime/rsconf.h index 484fec8c..894c0d12 100644 --- a/runtime/rsconf.h +++ b/runtime/rsconf.h @@ -25,6 +25,7 @@ #include "linkedlist.h" #include "queue.h" +#include "lookup.h" /* --- configuration objects (the plan is to have ALL upper layers in this file) --- */ @@ -143,6 +144,7 @@ struct rsconf_s { globals_t globals; defaults_t defaults; templates_t templates; + lookup_tables_t lu_tabs; outchannels_t och; actions_t actions; rulesets_t rulesets; diff --git a/runtime/rsyslog.c b/runtime/rsyslog.c index 047dfa9b..53c7855a 100644 --- a/runtime/rsyslog.c +++ b/runtime/rsyslog.c @@ -74,6 +74,7 @@ #include "prop.h" #include "ruleset.h" #include "parser.h" +#include "lookup.h" #include "strgen.h" #include "statsobj.h" #include "atomic.h" @@ -186,6 +187,8 @@ rsrtInit(char **ppErrObj, obj_if_t *pObjIF) CHKiRet(strgenClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "rsconf"; CHKiRet(rsconfClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "lookup"; + CHKiRet(lookupClassInit()); /* dummy "classes" */ if(ppErrObj != NULL) *ppErrObj = "str"; diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index e62ba867..e0331da7 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -49,6 +49,7 @@ #define CONF_PROGNAME_BUFSIZE 16 #define CONF_HOSTNAME_BUFSIZE 32 #define CONF_PROP_BUFSIZE 16 /* should be close to sizeof(ptr) or lighly above it */ +#define CONF_IPARAMS_BUFSIZE 16 /* initial size of iparams array in wti (is automatically extended) */ #define CONF_MIN_SIZE_FOR_COMPRESS 60 /* config param: minimum message size to try compression. The smaller * the message, the less likely is any compression gain. We check for * gain before we submit the message. But to do so we still need to @@ -102,50 +103,6 @@ #define _PATH_CONSOLE "/dev/console" #endif -/* properties are now encoded as (tiny) integers. I do not use an enum as I would like - * to keep the memory footprint small (and thus cache hits high). - * rgerhards, 2009-06-26 - */ -typedef uintTiny propid_t; -#define PROP_INVALID 0 -#define PROP_MSG 1 -#define PROP_TIMESTAMP 2 -#define PROP_HOSTNAME 3 -#define PROP_SYSLOGTAG 4 -#define PROP_RAWMSG 5 -#define PROP_INPUTNAME 6 -#define PROP_FROMHOST 7 -#define PROP_FROMHOST_IP 8 -#define PROP_PRI 9 -#define PROP_PRI_TEXT 10 -#define PROP_IUT 11 -#define PROP_SYSLOGFACILITY 12 -#define PROP_SYSLOGFACILITY_TEXT 13 -#define PROP_SYSLOGSEVERITY 14 -#define PROP_SYSLOGSEVERITY_TEXT 15 -#define PROP_TIMEGENERATED 16 -#define PROP_PROGRAMNAME 17 -#define PROP_PROTOCOL_VERSION 18 -#define PROP_STRUCTURED_DATA 19 -#define PROP_APP_NAME 20 -#define PROP_PROCID 21 -#define PROP_MSGID 22 -#define PROP_PARSESUCCESS 23 -#define PROP_SYS_NOW 150 -#define PROP_SYS_YEAR 151 -#define PROP_SYS_MONTH 152 -#define PROP_SYS_DAY 153 -#define PROP_SYS_HOUR 154 -#define PROP_SYS_HHOUR 155 -#define PROP_SYS_QHOUR 156 -#define PROP_SYS_MINUTE 157 -#define PROP_SYS_MYHOSTNAME 158 -#define PROP_CEE 200 -#define PROP_CEE_ALL_JSON 201 -#define PROP_SYS_BOM 159 -#define PROP_SYS_UPTIME 160 -#define PROP_UUID 161 - /* The error codes below are orginally "borrowed" from * liblogging. As such, we reserve values up to -2999 @@ -399,7 +356,7 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_RULESET_EXISTS = -2306,/**< ruleset already exists */ RS_RET_DEPRECATED = -2307,/**< deprecated functionality is used */ RS_RET_DS_PROP_SEQ_ERR = -2308,/**< property sequence error deserializing object */ - RS_RET_TPL_INVLD_PROP = -2309,/**< property name error in template (unknown name) */ + RS_RET_INVLD_PROP = -2309,/**< property name error (unknown name) */ RS_RET_NO_RULEBASE = -2310,/**< mmnormalize: rulebase can not be found or otherwise invalid */ RS_RET_INVLD_MODE = -2311,/**< invalid mode specified in configuration */ RS_RET_INVLD_ANON_BITS = -2312,/**< mmanon: invalid number of bits to anonymize specified */ @@ -416,6 +373,16 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_CA_CERT_MISSING = -2329,/**< a CA cert is missing where one is required (e.g. TLS) */ RS_RET_CERT_MISSING = -2330,/**< a cert is missing where one is required (e.g. TLS) */ RS_RET_CERTKEY_MISSING = -2331,/**< a cert (private) key is missing where one is required (e.g. TLS) */ + RS_RET_STRUC_DATA_INVLD = -2349,/**< structured data is malformed */ + + /* up to 2350 reserved for 7.4 */ + RS_RET_QUEUE_CRY_DISK_ONLY = -2351,/**< crypto provider only supported for disk-associated queues */ + RS_RET_NO_DATA = -2352,/**< file has no data; more a state than a real error */ + RS_RET_RELP_AUTH_FAIL = -2353,/**< RELP peer authentication failed */ + RS_RET_ERR_UDPSEND = -2354,/**< sending msg via UDP failed */ + RS_RET_LAST_ERRREPORT = -2355,/**< module does not emit more error messages as limit is reached */ + RS_RET_READ_ERR = -2356,/**< read error occured (file i/o) */ + RS_RET_CONF_PARSE_WARNING = -2357,/**< warning parsing config file */ /* RainerScript error messages (range 1000.. 1999) */ RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */ @@ -434,7 +401,12 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth * Be sure to call the to-be-returned variable always "iRet" and * the function finalizer always "finalize_it". */ -#define CHKiRet(code) if((iRet = code) != RS_RET_OK) goto finalize_it +#if HAVE_BUILTIN_EXCEPT +# define CHKiRet(code) if(__builtin_expect(((iRet = code) != RS_RET_OK), 0)) goto finalize_it +#else +# define CHKiRet(code) if((iRet = code) != RS_RET_OK) goto finalize_it +#endif + /* macro below is to be used if we need our own handling, eg for cleanup */ #define CHKiRet_Hdlr(code) if((iRet = code) != RS_RET_OK) /* macro below is to handle failing malloc/calloc/strdup... which we almost always handle in the same way... */ diff --git a/runtime/ruleset.c b/runtime/ruleset.c index 1afb4039..e217b3e0 100644 --- a/runtime/ruleset.c +++ b/runtime/ruleset.c @@ -48,6 +48,7 @@ #include "rainerscript.h" #include "srUtils.h" #include "modules.h" +#include "wti.h" #include "dirty.h" /* for main ruleset queue creation */ /* static data */ @@ -67,8 +68,8 @@ static struct cnfparamblk rspblk = }; /* forward definitions */ -static rsRetVal processBatch(batch_t *pBatch); -static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active); +static rsRetVal processBatch(batch_t *pBatch, wti_t *pWti); +static void scriptExec(struct cnfstmt *root, msg_t *pMsg, wti_t *pWti); /* ---------- linked-list key handling functions (ruleset) ---------- */ @@ -160,250 +161,112 @@ finalize_it: RETiRet; } - -/* This function is similar to processBatch(), but works on a batch that - * contains rules from multiple rulesets. In this case, we can not push - * the whole batch through the ruleset. Instead, we examine it and - * partition it into sub-rulesets which we then push through the system. - * rgerhards, 2010-06-15 - */ -static inline rsRetVal -processBatchMultiRuleset(batch_t *pBatch) +/* driver to iterate over all rulesets */ +DEFFUNC_llExecFunc(doActivateRulesetQueues) { - ruleset_t *currRuleset; - batch_t snglRuleBatch; - int i; - int iStart; /* start index of partial batch */ - int iNew; /* index for new (temporary) batch */ - int bHaveUnprocessed; /* do we (still) have unprocessed entries? (loop term predicate) */ DEFiRet; - - do { - bHaveUnprocessed = 0; - /* search for first unprocessed element */ - for(iStart = 0 ; iStart < pBatch->nElem && pBatch->eltState[iStart] == BATCH_STATE_DISC ; ++iStart) - /* just search, no action */; - if(iStart == pBatch->nElem) - break; /* everything processed */ - - /* prepare temporary batch */ - CHKiRet(batchInit(&snglRuleBatch, pBatch->nElem)); - snglRuleBatch.pbShutdownImmediate = pBatch->pbShutdownImmediate; - currRuleset = batchElemGetRuleset(pBatch, iStart); - iNew = 0; - for(i = iStart ; i < pBatch->nElem ; ++i) { - if(batchElemGetRuleset(pBatch, i) == currRuleset) { - /* for performance reasons, we copy only those members that we actually need */ - snglRuleBatch.pElem[iNew].pMsg = pBatch->pElem[i].pMsg; - snglRuleBatch.eltState[iNew] = pBatch->eltState[i]; - ++iNew; - /* We indicate the element also as done, so it will not be processed again */ - pBatch->eltState[i] = BATCH_STATE_DISC; - } else { - bHaveUnprocessed = 1; - } - } - snglRuleBatch.nElem = iNew; /* was left just right by the for loop */ - batchSetSingleRuleset(&snglRuleBatch, 1); - /* process temp batch */ - processBatch(&snglRuleBatch); - batchFree(&snglRuleBatch); - } while(bHaveUnprocessed == 1); - -finalize_it: + ruleset_t* pThis = (ruleset_t*) pData; + dbgprintf("Activating Ruleset Queue[%p] for Ruleset %s\n", + pThis->pQueue, pThis->pszName); + if(pThis->pQueue != NULL) + startMainQueue(pThis->pQueue); RETiRet; } - -/* return a new "active" structure for the batch. Free with freeActive(). */ -static inline sbool *newActive(batch_t *pBatch) +/* activate all ruleset queues */ +rsRetVal +activateRulesetQueues() { - return malloc(sizeof(sbool) * batchNumMsgs(pBatch)); - -} -static inline void freeActive(sbool *active) { free(active); } + DEFiRet; + llExecFunc(&(runConf->rulesets.llRulesets), doActivateRulesetQueues, NULL); -/* for details, see scriptExec() header comment! */ -/* call action for all messages with filter on */ -static rsRetVal -execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) -{ - DEFiRet; -dbgprintf("RRRR: execAct [%s]: batch of %d elements, active %p\n", modGetName(stmt->d.act->pMod), batchNumMsgs(pBatch), active); - pBatch->active = active; - stmt->d.act->submitToActQ(stmt->d.act, pBatch); RETiRet; } -static rsRetVal -execSet(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) + +static void +execAct(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti) { - int i; - struct var result; - DEFiRet; - for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - if( pBatch->eltState[i] != BATCH_STATE_DISC - && (active == NULL || active[i])) { - cnfexprEval(stmt->d.s_set.expr, &result, pBatch->pElem[i].pMsg); - msgSetJSONFromVar(pBatch->pElem[i].pMsg, stmt->d.s_set.varname, - &result); - varDelete(&result); - } + if(stmt->d.act->bDisabled) { + DBGPRINTF("action %d died, do NOT execute\n", stmt->d.act->iActionNbr); + goto done; } - RETiRet; + + DBGPRINTF("executing action %d\n", stmt->d.act->iActionNbr); + stmt->d.act->submitToActQ(stmt->d.act, pWti, pMsg); +done: return; } -static rsRetVal -execUnset(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +static void +execSet(struct cnfstmt *stmt, msg_t *pMsg) { - int i; - DEFiRet; - for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - if( pBatch->eltState[i] != BATCH_STATE_DISC - && (active == NULL || active[i])) { - msgUnsetJSON(pBatch->pElem[i].pMsg, stmt->d.s_unset.varname); - } - } - RETiRet; + struct var result; + cnfexprEval(stmt->d.s_set.expr, &result, pMsg); + msgSetJSONFromVar(pMsg, stmt->d.s_set.varname, &result); + varDelete(&result); } -/* for details, see scriptExec() header comment! */ -/* "stop" simply discards the filtered items - it's just a (hopefully more intuitive - * shortcut for users. - */ -static rsRetVal -execStop(batch_t *pBatch, sbool *active) +static void +execUnset(struct cnfstmt *stmt, msg_t *pMsg) { - int i; - DEFiRet; - for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pBatch->pbShutdownImmediate) ; ++i) { - if( pBatch->eltState[i] != BATCH_STATE_DISC - && (active == NULL || active[i])) { - pBatch->eltState[i] = BATCH_STATE_DISC; - } - } - RETiRet; + msgDelJSON(pMsg, stmt->d.s_unset.varname); } + static rsRetVal -execCall(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execCall(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti) { - msg_t *pMsg; - int i; DEFiRet; if(stmt->d.s_call.ruleset == NULL) { - scriptExec(stmt->d.s_call.stmt, pBatch, active); + scriptExec(stmt->d.s_call.stmt, pMsg, pWti); } else { - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - CHKmalloc(pMsg = MsgDup((msg_t*) pBatch->pElem[i].pMsg)); - DBGPRINTF("CALL: forwarding message %d to async ruleset %p\n", - i, stmt->d.s_call.ruleset->pQueue); - MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY); - MsgSetRuleset(pMsg, stmt->d.s_call.ruleset); - /* Note: we intentionally use submitMsg2() here, as we process messages - * that were already run through the rate-limiter. - */ - submitMsg2(pMsg); - } + CHKmalloc(pMsg = MsgDup((msg_t*) pMsg)); + DBGPRINTF("CALL: forwarding message to async ruleset %p\n", + stmt->d.s_call.ruleset->pQueue); + MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY); + MsgSetRuleset(pMsg, stmt->d.s_call.ruleset); + /* Note: we intentionally use submitMsg2() here, as we process messages + * that were already run through the rate-limiter. + */ + submitMsg2(pMsg); } finalize_it: RETiRet; } -/* for details, see scriptExec() header comment! */ -// save current filter, evaluate new one -// perform then (if any message) -// if ELSE given: -// set new filter, inverted -// perform else (if any messages) -static rsRetVal -execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +static void +execIf(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti) { - sbool *newAct; - int i; sbool bRet; - sbool allInactive = 1; - DEFiRet; - newAct = newActive(pBatch); - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(*(pBatch->pbShutdownImmediate)) - FINALIZE; - if(pBatch->eltState[i] == BATCH_STATE_DISC) - continue; /* will be ignored in any case */ - if(active == NULL || active[i]) { - bRet = cnfexprEvalBool(stmt->d.s_if.expr, pBatch->pElem[i].pMsg); - allInactive = 0; - } else - bRet = 0; - newAct[i] = bRet; - DBGPRINTF("batch: item %d: expr eval: %d\n", i, bRet); - } - - if(allInactive) { - DBGPRINTF("execIf: all batch elements are inactive, holding execution\n"); - freeActive(newAct); - FINALIZE; - } - - if(stmt->d.s_if.t_then != NULL) { - scriptExec(stmt->d.s_if.t_then, pBatch, newAct); - } - if(stmt->d.s_if.t_else != NULL) { - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(*(pBatch->pbShutdownImmediate)) - FINALIZE; - if(pBatch->eltState[i] != BATCH_STATE_DISC - && (active == NULL || active[i])) - newAct[i] = !newAct[i]; - } - scriptExec(stmt->d.s_if.t_else, pBatch, newAct); + bRet = cnfexprEvalBool(stmt->d.s_if.expr, pMsg); + DBGPRINTF("if condition result is %d\n", bRet); + if(bRet) { + if(stmt->d.s_if.t_then != NULL) + scriptExec(stmt->d.s_if.t_then, pMsg, pWti); + } else { + if(stmt->d.s_if.t_else != NULL) + scriptExec(stmt->d.s_if.t_else, pMsg, pWti); } - freeActive(newAct); -finalize_it: - RETiRet; } -/* for details, see scriptExec() header comment! */ static void -execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPRIFILT(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti) { - sbool *newAct; - msg_t *pMsg; int bRet; - int i; - newAct = newActive(pBatch); - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(*(pBatch->pbShutdownImmediate)) - return; - if(pBatch->eltState[i] == BATCH_STATE_DISC) - continue; /* will be ignored in any case */ - pMsg = pBatch->pElem[i].pMsg; - if(active == NULL || active[i]) { - if( (stmt->d.s_prifilt.pmask[pMsg->iFacility] == TABLE_NOPRI) || - ((stmt->d.s_prifilt.pmask[pMsg->iFacility] - & (1<<pMsg->iSeverity)) == 0) ) - bRet = 0; - else - bRet = 1; - } else - bRet = 0; - newAct[i] = bRet; - DBGPRINTF("batch: item %d PRIFILT %d\n", i, newAct[i]); - } - - if(stmt->d.s_prifilt.t_then != NULL) { - scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct); - } - if(stmt->d.s_prifilt.t_else != NULL) { - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(*(pBatch->pbShutdownImmediate)) - return; - if(pBatch->eltState[i] != BATCH_STATE_DISC - && (active == NULL || active[i])) - newAct[i] = !newAct[i]; - } - scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct); + if( (stmt->d.s_prifilt.pmask[pMsg->iFacility] == TABLE_NOPRI) || + ((stmt->d.s_prifilt.pmask[pMsg->iFacility] + & (1<<pMsg->iSeverity)) == 0) ) + bRet = 0; + else + bRet = 1; + + DBGPRINTF("PRIFILT condition result is %d\n", bRet); + if(bRet) { + if(stmt->d.s_prifilt.t_then != NULL) + scriptExec(stmt->d.s_prifilt.t_then, pMsg, pWti); + } else { + if(stmt->d.s_prifilt.t_else != NULL) + scriptExec(stmt->d.s_prifilt.t_else, pMsg, pWti); } - freeActive(newAct); } @@ -416,12 +279,11 @@ evalPROPFILT(struct cnfstmt *stmt, msg_t *pMsg) int bRet = 0; rs_size_t propLen; - if(stmt->d.s_propfilt.propID == PROP_INVALID) + if(stmt->d.s_propfilt.prop.id == PROP_INVALID) goto done; - pszPropVal = MsgGetProp(pMsg, NULL, stmt->d.s_propfilt.propID, - stmt->d.s_propfilt.propName, &propLen, - &pbMustBeFreed, NULL); + pszPropVal = MsgGetProp(pMsg, NULL, &stmt->d.s_propfilt.prop, + &propLen, &pbMustBeFreed, NULL); /* Now do the compares (short list currently ;)) */ switch(stmt->d.s_propfilt.operation ) { @@ -465,15 +327,18 @@ evalPROPFILT(struct cnfstmt *stmt, msg_t *pMsg) bRet = (bRet == 1) ? 0 : 1; if(Debug) { - char *cstr; - if(stmt->d.s_propfilt.propID == PROP_CEE) { - cstr = es_str2cstr(stmt->d.s_propfilt.propName, NULL); + if(stmt->d.s_propfilt.prop.id == PROP_CEE) { DBGPRINTF("Filter: check for CEE property '%s' (value '%s') ", - cstr, pszPropVal); - free(cstr); + stmt->d.s_propfilt.prop.name, pszPropVal); + } else if(stmt->d.s_propfilt.prop.id == PROP_LOCAL_VAR) { + DBGPRINTF("Filter: check for local var '%s' (value '%s') ", + stmt->d.s_propfilt.prop.name, pszPropVal); + } else if(stmt->d.s_propfilt.prop.id == PROP_GLOBAL_VAR) { + DBGPRINTF("Filter: check for global var '%s' (value '%s') ", + stmt->d.s_propfilt.prop.name, pszPropVal); } else { DBGPRINTF("Filter: check for property '%s' (value '%s') ", - propIDToName(stmt->d.s_propfilt.propID), pszPropVal); + propIDToName(stmt->d.s_propfilt.prop.id), pszPropVal); } if(stmt->d.s_propfilt.isNegated) DBGPRINTF("NOT "); @@ -496,79 +361,63 @@ done: return bRet; } -/* for details, see scriptExec() header comment! */ static void -execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPROPFILT(struct cnfstmt *stmt, msg_t *pMsg, wti_t *pWti) { - sbool *thenAct; sbool bRet; - int i; - thenAct = newActive(pBatch); - for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { - if(*(pBatch->pbShutdownImmediate)) - return; - if(pBatch->eltState[i] == BATCH_STATE_DISC) - continue; /* will be ignored in any case */ - if(active == NULL || active[i]) { - bRet = evalPROPFILT(stmt, pBatch->pElem[i].pMsg); - } else - bRet = 0; - thenAct[i] = bRet; - DBGPRINTF("batch: item %d PROPFILT %d\n", i, thenAct[i]); - } - scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct); - freeActive(thenAct); + bRet = evalPROPFILT(stmt, pMsg); + DBGPRINTF("PROPFILT condition result is %d\n", bRet); + if(bRet) + scriptExec(stmt->d.s_propfilt.t_then, pMsg, pWti); } /* The rainerscript execution engine. It is debatable if that would be better * contained in grammer/rainerscript.c, HOWEVER, that file focusses primarily * on the parsing and object creation part. So as an actual executor, it is * better suited here. - * param active: if NULL, all messages are active (to be processed), if non-null - * this is an array of the same size as the batch. If 1, the message - * is to be processed, otherwise not. - * NOTE: this function must receive batches which contain a single ruleset ONLY! * rgerhards, 2012-09-04 */ -static rsRetVal -scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) +static void +scriptExec(struct cnfstmt *root, msg_t *pMsg, wti_t *pWti) { - DEFiRet; struct cnfstmt *stmt; for(stmt = root ; stmt != NULL ; stmt = stmt->next) { + if(*pWti->pbShutdownImmediate) { + DBGPRINTF("scriptExec: ShutdownImmediate set, " + "force terminating\n"); + goto done; + } if(Debug) { - dbgprintf("scriptExec: batch of %d elements, active %p, active[0]:%d\n", - batchNumMsgs(pBatch), active, (active == NULL ? 1 : active[0])); cnfstmtPrintOnly(stmt, 2, 0); } switch(stmt->nodetype) { case S_NOP: break; case S_STOP: - execStop(pBatch, active); + goto done; break; case S_ACT: - execAct(stmt, pBatch, active); + execAct(stmt, pMsg, pWti); break; case S_SET: - execSet(stmt, pBatch, active); + execSet(stmt, pMsg); break; case S_UNSET: - execUnset(stmt, pBatch, active); + execUnset(stmt, pMsg); break; case S_CALL: - execCall(stmt, pBatch, active); + execCall(stmt, pMsg, pWti); break; case S_IF: - execIf(stmt, pBatch, active); + execIf(stmt, pMsg, pWti); break; case S_PRIFILT: - execPRIFILT(stmt, pBatch, active); + execPRIFILT(stmt, pMsg, pWti); break; case S_PROPFILT: - execPROPFILT(stmt, pBatch, active); + execPROPFILT(stmt, pMsg, pWti); break; default: dbgprintf("error: unknown stmt type %u during exec\n", @@ -576,36 +425,42 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) break; } } - RETiRet; +done: return; } /* Process (consume) a batch of messages. Calls the actions configured. - * If the whole batch uses a singel ruleset, we can process the batch as - * a whole. Otherwise, we need to process it slower, on a message-by-message - * basis (what can be optimized to a per-ruleset basis) - * rgerhards, 2005-10-13 + * This is called by MAIN queues. */ static rsRetVal -processBatch(batch_t *pBatch) +processBatch(batch_t *pBatch, wti_t *pWti) { - ruleset_t *pThis; + int i; + msg_t *pMsg; + ruleset_t *pRuleset; DEFiRet; - assert(pBatch != NULL); - - DBGPRINTF("processBatch: batch of %d elements must be processed\n", pBatch->nElem); - if(pBatch->bSingleRuleset) { - pThis = batchGetRuleset(pBatch); - if(pThis == NULL) - pThis = ourConf->rulesets.pDflt; - ISOBJ_TYPE_assert(pThis, ruleset); - CHKiRet(scriptExec(pThis->root, pBatch, NULL)); - } else { - CHKiRet(processBatchMultiRuleset(pBatch)); + + DBGPRINTF("processBATCH: batch of %d elements must be processed\n", pBatch->nElem); + + wtiResetExecState(pWti, pBatch); + + /* execution phase */ + for(i = 0 ; i < batchNumMsgs(pBatch) && !*(pWti->pbShutdownImmediate) ; ++i) { + pMsg = pBatch->pElem[i].pMsg; + DBGPRINTF("processBATCH: next msg %d: %.128s\n", i, pMsg->pszRawMsg); + pRuleset = (pMsg->pRuleset == NULL) ? ourConf->rulesets.pDflt : pMsg->pRuleset; + scriptExec(pRuleset->root, pMsg, pWti); + // TODO: think if we need a return state of scriptExec - most probably + // the answer is "no", as we need to process the batch in any case! + // TODO: we must refactor this! flag messages as committed + batchSetElemState(pBatch, i, BATCH_STATE_COMM); } -finalize_it: - DBGPRINTF("ruleset.ProcessMsg() returns %d\n", iRet); + /* commit phase */ + dbgprintf("END batch execution phase, entering to commit phase\n"); + actionCommitAllDirect(pWti); + + DBGPRINTF("processBATCH: batch of %d elements has been processed\n", pBatch->nElem); RETiRet; } @@ -952,7 +807,6 @@ rsRetVal rulesetProcessCnf(struct cnfobj *o) { struct cnfparamvals *pvals; - struct cnfparamvals *queueParams; rsRetVal localRet; uchar *rsName = NULL; uchar *parserName; @@ -998,11 +852,10 @@ rulesetProcessCnf(struct cnfobj *o) } /* pick up ruleset queue parameters */ - qqueueDoCnfParams(o->nvlst, &queueParams); - if(queueCnfParamsSet(queueParams)) { + if(queueCnfParamsSet(o->nvlst)) { rsname = (pRuleset->pszName == NULL) ? (uchar*) "[ruleset]" : pRuleset->pszName; DBGPRINTF("adding a ruleset-specific \"main\" queue for ruleset '%s'\n", rsname); - CHKiRet(createMainQueue(&pRuleset->pQueue, rsname, queueParams)); + CHKiRet(createMainQueue(&pRuleset->pQueue, rsname, o->nvlst)); } finalize_it: diff --git a/runtime/ruleset.h b/runtime/ruleset.h index 64fe92ff..d3dfd664 100644 --- a/runtime/ruleset.h +++ b/runtime/ruleset.h @@ -2,7 +2,7 @@ * * This implements rulesets within rsyslog. * - * Copyright 2009-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2009-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -46,7 +46,7 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Destruct)(ruleset_t **ppThis); rsRetVal (*DestructAllActions)(rsconf_t *conf); rsRetVal (*SetName)(ruleset_t *pThis, uchar *pszName); - rsRetVal (*ProcessBatch)(batch_t*); + rsRetVal (*ProcessBatch)(batch_t*, wti_t *); rsRetVal (*GetRuleset)(rsconf_t *conf, ruleset_t **ppThis, uchar*); rsRetVal (*SetDefaultRuleset)(rsconf_t *conf, uchar*); rsRetVal (*SetCurrRuleset)(rsconf_t *conf, uchar*); @@ -64,8 +64,9 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ /* AddRule() removed */ /*TODO:REMOVE*/rsRetVal (*IterateAllActions)(rsconf_t *conf, rsRetVal (*pFunc)(void*, void*), void* pParam); void (*AddScript)(ruleset_t *pThis, struct cnfstmt *script); + /* v8: changed processBatch interface */ ENDinterface(ruleset) -#define rulesetCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */ +#define rulesetCURR_IF_VERSION 8 /* increment whenever you change the interface structure! */ /* prototypes */ @@ -104,6 +105,7 @@ rulesetHasQueue(ruleset_t *pRuleset) rsRetVal rulesetGetRuleset(rsconf_t *conf, ruleset_t **ppRuleset, uchar *pszName); rsRetVal rulesetOptimizeAll(rsconf_t *conf); rsRetVal rulesetProcessCnf(struct cnfobj *o); +rsRetVal activateRulesetQueues(void); /* Set a current rule set to already-known pointer */ static inline void diff --git a/runtime/srutils.c b/runtime/srutils.c index 6a509b4a..8eb2459c 100644 --- a/runtime/srutils.c +++ b/runtime/srutils.c @@ -86,6 +86,7 @@ syslogName_t syslogFacNames[] = { {"mark", LOG_MARK}, /* INTERNAL */ {"news", LOG_NEWS}, {"security", LOG_AUTH}, /* DEPRECATED */ + {"bsd_security", (13<<3) }, /* BSD-specific, unfortunatly with duplicate name... */ {"syslog", LOG_SYSLOG}, {"user", LOG_USER}, {"uucp", LOG_UUCP}, @@ -95,6 +96,7 @@ syslogName_t syslogFacNames[] = { #if defined(LOG_AUDIT) {"audit", LOG_AUDIT}, #endif + {"console", (14 << 3)}, /* BSD-specific priority */ {"local0", LOG_LOCAL0}, {"local1", LOG_LOCAL1}, {"local2", LOG_LOCAL2}, diff --git a/runtime/statsobj.c b/runtime/statsobj.c index 25275616..edac7d48 100644 --- a/runtime/statsobj.c +++ b/runtime/statsobj.c @@ -142,7 +142,7 @@ finalize_it: * is called. */ static rsRetVal -addCounter(statsobj_t *pThis, uchar *ctrName, statsCtrType_t ctrType, void *pCtr) +addCounter(statsobj_t *pThis, uchar *ctrName, statsCtrType_t ctrType, int8_t flags, void *pCtr) { ctr_t *ctr; DEFiRet; @@ -151,6 +151,7 @@ addCounter(statsobj_t *pThis, uchar *ctrName, statsCtrType_t ctrType, void *pCtr ctr->next = NULL; ctr->prev = NULL; CHKmalloc(ctr->name = ustrdup(ctrName)); + ctr->flags = flags; ctr->ctrType = ctrType; switch(ctrType) { case ctrType_IntCtr: @@ -166,9 +167,24 @@ finalize_it: RETiRet; } +static inline void +resetResettableCtr(ctr_t *pCtr, int8_t bResetCtrs) +{ + if(bResetCtrs && (pCtr->flags & CTR_FLAG_RESETTABLE)) { + switch(pCtr->ctrType) { + case ctrType_IntCtr: + *(pCtr->val.pIntCtr) = 0; + break; + case ctrType_Int: + *(pCtr->val.pInt) = 0; + break; + } + } +} + /* get all the object's countes together as CEE. */ static rsRetVal -getStatsLineCEE(statsobj_t *pThis, cstr_t **ppcstr, int cee_cookie) +getStatsLineCEE(statsobj_t *pThis, cstr_t **ppcstr, int cee_cookie, int8_t bResetCtrs) { cstr_t *pcstr; ctr_t *pCtr; @@ -209,7 +225,7 @@ getStatsLineCEE(statsobj_t *pThis, cstr_t **ppcstr, int cee_cookie) } else { cstrAppendChar(pcstr, '}'); } - + resetResettableCtr(pCtr, bResetCtrs); } pthread_mutex_unlock(&pThis->mutCtr); @@ -223,7 +239,7 @@ finalize_it: /* get all the object's countes together with object name as one line. */ static rsRetVal -getStatsLine(statsobj_t *pThis, cstr_t **ppcstr) +getStatsLine(statsobj_t *pThis, cstr_t **ppcstr, int8_t bResetCtrs) { cstr_t *pcstr; ctr_t *pCtr; @@ -247,6 +263,7 @@ getStatsLine(statsobj_t *pThis, cstr_t **ppcstr) break; } cstrAppendChar(pcstr, ' '); + resetResettableCtr(pCtr, bResetCtrs); } pthread_mutex_unlock(&pThis->mutCtr); @@ -265,7 +282,7 @@ finalize_it: * line. If the callback reports an error, processing is stopped. */ static rsRetVal -getAllStatsLines(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt) +getAllStatsLines(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt, int8_t bResetCtrs) { statsobj_t *o; cstr_t *cstr; @@ -274,13 +291,13 @@ getAllStatsLines(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt for(o = objRoot ; o != NULL ; o = o->next) { switch(fmt) { case statsFmt_Legacy: - CHKiRet(getStatsLine(o, &cstr)); + CHKiRet(getStatsLine(o, &cstr, bResetCtrs)); break; case statsFmt_CEE: - CHKiRet(getStatsLineCEE(o, &cstr, 1)); + CHKiRet(getStatsLineCEE(o, &cstr, 1, bResetCtrs)); break; case statsFmt_JSON: - CHKiRet(getStatsLineCEE(o, &cstr, 0)); + CHKiRet(getStatsLineCEE(o, &cstr, 0, bResetCtrs)); break; } CHKiRet(cb(usrptr, cstr)); @@ -348,7 +365,7 @@ CODESTARTobjQueryInterface(statsobj) pIf->Destruct = statsobjDestruct; pIf->DebugPrint = statsobjDebugPrint; pIf->SetName = setName; - pIf->GetStatsLine = getStatsLine; + //pIf->GetStatsLine = getStatsLine; pIf->GetAllStatsLines = getAllStatsLines; pIf->AddCounter = addCounter; pIf->EnableStats = enableStats; diff --git a/runtime/statsobj.h b/runtime/statsobj.h index 14b33215..d56485de 100644 --- a/runtime/statsobj.h +++ b/runtime/statsobj.h @@ -50,6 +50,9 @@ typedef enum statsFmtType_e { statsFmt_CEE } statsFmtType_t; +/* counter flags */ +#define CTR_FLAG_NONE 0 +#define CTR_FLAG_RESETTABLE 1 /* helper entity, the counter */ typedef struct ctr_s { @@ -59,6 +62,7 @@ typedef struct ctr_s { intctr_t *pIntCtr; int *pInt; } val; + int8_t flags; struct ctr_s *next, *prev; } ctr_t; @@ -82,15 +86,17 @@ BEGINinterface(statsobj) /* name must also be changed in ENDinterface macro! */ rsRetVal (*ConstructFinalize)(statsobj_t *pThis); rsRetVal (*Destruct)(statsobj_t **ppThis); rsRetVal (*SetName)(statsobj_t *pThis, uchar *name); - rsRetVal (*GetStatsLine)(statsobj_t *pThis, cstr_t **ppcstr); - rsRetVal (*GetAllStatsLines)(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt); - rsRetVal (*AddCounter)(statsobj_t *pThis, uchar *ctrName, statsCtrType_t ctrType, void *pCtr); + //rsRetVal (*GetStatsLine)(statsobj_t *pThis, cstr_t **ppcstr); + rsRetVal (*GetAllStatsLines)(rsRetVal(*cb)(void*, cstr_t*), void *usrptr, statsFmtType_t fmt, int8_t bResetCtr); + rsRetVal (*AddCounter)(statsobj_t *pThis, uchar *ctrName, statsCtrType_t ctrType, int8_t flags, void *pCtr); rsRetVal (*EnableStats)(void); ENDinterface(statsobj) -#define statsobjCURR_IF_VERSION 10 /* increment whenever you change the interface structure! */ +#define statsobjCURR_IF_VERSION 11 /* increment whenever you change the interface structure! */ /* Changes * v2-v9 rserved for future use in "older" version branches * v10, 2012-04-01: GetAllStatsLines got fmt parameter + * v11, 2013-09-07: - add "flags" to AddCounter API + * - GetAllStatsLines got parameter telling if ctrs shall be reset */ diff --git a/runtime/stream.c b/runtime/stream.c index 94fc0ca7..b35d6a11 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -256,7 +256,9 @@ doPhysOpen(strm_t *pThis) if(pThis->cryprov != NULL) { CHKiRet(pThis->cryprov->OnFileOpen(pThis->cryprovData, - pThis->pszCurrFName, &pThis->cryprovFileData)); + pThis->pszCurrFName, &pThis->cryprovFileData, + (pThis->tOperationsMode == STREAMMODE_READ) ? 'r' : 'w')); + pThis->cryprov->SetDeleteOnClose(pThis->cryprovFileData, pThis->bDeleteOnClose); } finalize_it: RETiRet; @@ -404,6 +406,12 @@ static rsRetVal strmCloseFile(strm_t *pThis) } } + /* if we have a signature provider, we must make sure that the crypto + * state files are opened and proper close processing happens. */ + if(pThis->cryprov != NULL && pThis->fd == -1) { + strmOpenFile(pThis); + } + /* the file may already be closed (or never have opened), so guard * against this. -- rgerhards, 2010-03-19 */ @@ -550,11 +558,14 @@ finalize_it: * rgerhards, 2008-02-13 */ static rsRetVal -strmReadBuf(strm_t *pThis) +strmReadBuf(strm_t *pThis, int *padBytes) { DEFiRet; int bRun; long iLenRead; + size_t actualDataLen; + size_t toRead; + ssize_t bytesLeft; ISOBJ_TYPE_assert(pThis, strm); /* We need to try read at least twice because we may run into EOF and need to switch files. */ @@ -565,13 +576,35 @@ strmReadBuf(strm_t *pThis) * rgerhards, 2008-02-13 */ CHKiRet(strmOpenFile(pThis)); - iLenRead = read(pThis->fd, pThis->pIOBuf, pThis->sIOBufSize); + if(pThis->cryprov == NULL) { + toRead = pThis->sIOBufSize; + } else { + CHKiRet(pThis->cryprov->GetBytesLeftInBlock(pThis->cryprovFileData, &bytesLeft)); + if(bytesLeft == -1 || bytesLeft > (ssize_t) pThis->sIOBufSize) { + toRead = pThis->sIOBufSize; + } else { + toRead = (size_t) bytesLeft; + } + } + iLenRead = read(pThis->fd, pThis->pIOBuf, toRead); DBGOPRINT((obj_t*) pThis, "file %d read %ld bytes\n", pThis->fd, iLenRead); + /* end crypto */ if(iLenRead == 0) { CHKiRet(strmHandleEOF(pThis)); } else if(iLenRead < 0) ABORT_FINALIZE(RS_RET_IO_ERROR); else { /* good read */ + /* here we place our crypto interface */ + if(pThis->cryprov != NULL) { + actualDataLen = iLenRead; + pThis->cryprov->Decrypt(pThis->cryprovFileData, pThis->pIOBuf, &actualDataLen); + *padBytes = iLenRead - actualDataLen; + iLenRead = actualDataLen; + DBGOPRINT((obj_t*) pThis, "encrypted file %d pad bytes %d, actual " + "data %ld\n", pThis->fd, *padBytes, iLenRead); + } else { + *padBytes = 0; + } pThis->iBufPtrMax = iLenRead; bRun = 0; /* exit loop */ } @@ -593,6 +626,7 @@ finalize_it: */ static rsRetVal strmReadChar(strm_t *pThis, uchar *pC) { + int padBytes = 0; /* in crypto mode, we may have some padding (non-data) bytes */ DEFiRet; ASSERT(pThis != NULL); @@ -608,8 +642,9 @@ static rsRetVal strmReadChar(strm_t *pThis, uchar *pC) /* do we need to obtain a new buffer? */ if(pThis->iBufPtr >= pThis->iBufPtrMax) { - CHKiRet(strmReadBuf(pThis)); + CHKiRet(strmReadBuf(pThis, &padBytes)); } + pThis->iCurrOffs += padBytes; /* if we reach this point, we have data available in the buffer */ @@ -645,7 +680,7 @@ static rsRetVal strmUnreadChar(strm_t *pThis, uchar c) * destruction of the returned CStr object! -- dlang 2010-12-13 */ static rsRetVal -strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) +strmReadLine(strm_t *pThis, cstr_t **ppCStr, uint8_t mode, sbool bEscapeLF) { /* mode = 0 single line mode (equivalent to ReadLine) * mode = 1 LFLF mode (paragraph, blank line between entries) @@ -655,6 +690,7 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) uchar c; uchar finished; rsRetVal readCharRet; + sbool bPrevWasNL; DEFiRet; ASSERT(pThis != NULL); @@ -680,18 +716,25 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) CHKiRet(cstrFinalize(*ppCStr)); } else if(mode == 1) { finished=0; + bPrevWasNL = 0; while(finished == 0){ if(c != '\n') { CHKiRet(cstrAppendChar(*ppCStr, c)); CHKiRet(strmReadChar(pThis, &c)); + bPrevWasNL = 0; } else { if ((((*ppCStr)->iStrLen) > 0) ){ - if ((*ppCStr)->pBuf[(*ppCStr)->iStrLen -1 ] == '\n'){ - rsCStrTruncate(*ppCStr,1); /* remove the prior newline */ + if(bPrevWasNL) { + rsCStrTruncate(*ppCStr, (bEscapeLF) ? 4 : 1); /* remove the prior newline */ finished=1; } else { - CHKiRet(cstrAppendChar(*ppCStr, c)); + if(bEscapeLF) { + CHKiRet(rsCStrAppendStrWithLen(*ppCStr, (uchar*)"#012", sizeof("#012")-1)); + } else { + CHKiRet(cstrAppendChar(*ppCStr, c)); + } CHKiRet(strmReadChar(pThis, &c)); + bPrevWasNL = 1; } } else { finished=1; /* this is a blank line, a \n with nothing since the last complete record */ @@ -702,6 +745,7 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) } else if(mode == 2) { /* indented follow-up lines */ finished=0; + bPrevWasNL = 0; while(finished == 0){ if ((*ppCStr)->iStrLen == 0){ if(c != '\n') { @@ -712,22 +756,31 @@ strmReadLine(strm_t *pThis, cstr_t **ppCStr, int mode) finished=1; /* this is a blank line, a \n with nothing since the last complete record */ } } else { - if ((*ppCStr)->pBuf[(*ppCStr)->iStrLen -1 ] != '\n'){ - /* not the first character after a newline, add it to the buffer */ - CHKiRet(cstrAppendChar(*ppCStr, c)); - CHKiRet(strmReadChar(pThis, &c)); - } else { + if(bPrevWasNL) { if ((c == ' ') || (c == '\t')){ CHKiRet(cstrAppendChar(*ppCStr, c)); CHKiRet(strmReadChar(pThis, &c)); + bPrevWasNL = 0; } else { /* clean things up by putting the character we just read back into * the input buffer and removing the LF character that is currently at the * end of the output string */ CHKiRet(strmUnreadChar(pThis, c)); - rsCStrTruncate(*ppCStr,1); + rsCStrTruncate(*ppCStr, (bEscapeLF) ? 4 : 1); finished=1; } + } else { /* not the first character after a newline, add it to the buffer */ + if(c == '\n') { + bPrevWasNL = 1; + if(bEscapeLF) { + CHKiRet(rsCStrAppendStrWithLen(*ppCStr, (uchar*)"#012", sizeof("#012")-1)); + } else { + CHKiRet(cstrAppendChar(*ppCStr, c)); + } + } else { + CHKiRet(cstrAppendChar(*ppCStr, c)); + } + CHKiRet(strmReadChar(pThis, &c)); } } } @@ -1454,6 +1507,8 @@ strmMultiFileSeek(strm_t *pThis, int FNum, off64_t offs, off64_t *bytesDel) "deleting '%s' (%lld bytes)\n", pThis->iCurrFNum, FNum, pThis->pszCurrFName, (long long) *bytesDel); unlink((char*)pThis->pszCurrFName); + if(pThis->cryprov != NULL) + pThis->cryprov->DeleteStateFiles(pThis->pszCurrFName); free(pThis->pszCurrFName); pThis->pszCurrFName = NULL; pThis->iCurrFNum = FNum; @@ -1467,17 +1522,31 @@ finalize_it: } - /* seek to current offset. This is primarily a helper to readjust the OS file * pointer after a strm object has been deserialized. */ static rsRetVal strmSeekCurrOffs(strm_t *pThis) { + off64_t targetOffs; + uchar c; DEFiRet; ISOBJ_TYPE_assert(pThis, strm); - iRet = strmSeek(pThis, pThis->iCurrOffs); + if(pThis->cryprov == NULL || pThis->tOperationsMode != STREAMMODE_READ) { + iRet = strmSeek(pThis, pThis->iCurrOffs); + FINALIZE; + } + + /* As the cryprov may use CBC or similiar things, we need to read skip data */ + targetOffs = pThis->iCurrOffs; + pThis->iCurrOffs = 0; + DBGOPRINT((obj_t*) pThis, "encrypted, doing skip read of %lld bytes\n", + (long long) targetOffs); + while(targetOffs != pThis->iCurrOffs) { + CHKiRet(strmReadChar(pThis, &c)); + } +finalize_it: RETiRet; } @@ -1604,7 +1673,6 @@ finalize_it: /* property set methods */ /* simple ones first */ -DEFpropSetMeth(strm, bDeleteOnClose, int) DEFpropSetMeth(strm, iMaxFileSize, int64) DEFpropSetMeth(strm, iFileNumDigits, int) DEFpropSetMeth(strm, tOperationsMode, int) @@ -1620,6 +1688,15 @@ DEFpropSetMeth(strm, pszSizeLimitCmd, uchar*) DEFpropSetMeth(strm, cryprov, cryprov_if_t*) DEFpropSetMeth(strm, cryprovData, void*) +static rsRetVal strmSetbDeleteOnClose(strm_t *pThis, int val) +{ + pThis->bDeleteOnClose = val; + if(pThis->cryprov != NULL) { + pThis->cryprov->SetDeleteOnClose(pThis->cryprovFileData, pThis->bDeleteOnClose); + } + return RS_RET_OK; +} + static rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal) { pThis->iMaxFiles = iNewVal; diff --git a/runtime/stream.h b/runtime/stream.h index 4f4a4301..092d3226 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -66,6 +66,7 @@ #define STREAM_H_INCLUDED #include <pthread.h> +#include <stdint.h> #include "obj-types.h" #include "glbl.h" #include "stream.h" @@ -188,7 +189,7 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ INTERFACEpropSetMeth(strm, iFlushInterval, int); INTERFACEpropSetMeth(strm, pszSizeLimitCmd, uchar*); /* v6 added */ - rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr, int mode); + rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr, uint8_t mode, sbool bEscapeLF); /* v7 added 2012-09-14 */ INTERFACEpropSetMeth(strm, bVeryReliableZip, int); /* v8 added 2013-03-21 */ @@ -197,7 +198,8 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ INTERFACEpropSetMeth(strm, cryprov, cryprov_if_t*); INTERFACEpropSetMeth(strm, cryprovData, void*); ENDinterface(strm) -#define strmCURR_IF_VERSION 9 /* increment whenever you change the interface structure! */ +#define strmCURR_IF_VERSION 10 /* increment whenever you change the interface structure! */ +/* V10, 2013-09-10: added new parameter bEscapeLF, changed mode to uint8_t (rgerhards) */ static inline int strmGetCurrFileNum(strm_t *pStrm) { diff --git a/runtime/stringbuf.c b/runtime/stringbuf.c index 13f38710..97bfed40 100644 --- a/runtime/stringbuf.c +++ b/runtime/stringbuf.c @@ -528,26 +528,6 @@ rsRetVal rsCStrTruncate(cstr_t *pThis, size_t nTrunc) /* Trim trailing whitespace from a given string */ -rsRetVal rsCStrTrimTrailingWhiteSpace(cstr_t *pThis) -{ - register int i; - register uchar *pC; - rsCHECKVALIDOBJECT(pThis, OIDrsCStr); - - i = pThis->iStrLen; - pC = pThis->pBuf + i - 1; - while(i > 0 && isspace((int)*pC)) { - --pC; - --i; - } - /* i now is the new string length! */ - pThis->iStrLen = i; - - return RS_RET_OK; -} - -/* Trim trailing whitespace from a given string - */ rsRetVal cstrTrimTrailingWhiteSpace(cstr_t *pThis) { register int i; @@ -563,8 +543,10 @@ rsRetVal cstrTrimTrailingWhiteSpace(cstr_t *pThis) --i; } /* i now is the new string length! */ - pThis->iStrLen = i; - pThis->pBuf[pThis->iStrLen] = '\0'; /* we always have this space */ + if(i != (int) pThis->iStrLen) { + pThis->iStrLen = i; + pThis->pBuf[pThis->iStrLen] = '\0'; /* we always have this space */ + } done: return RS_RET_OK; } diff --git a/runtime/stringbuf.h b/runtime/stringbuf.h index d0502a5b..2df48ab4 100644 --- a/runtime/stringbuf.h +++ b/runtime/stringbuf.h @@ -155,7 +155,6 @@ static inline uchar* cstrGetSzStrNoNULL(cstr_t *pThis) */ rsRetVal rsCStrTruncate(cstr_t *pThis, size_t nTrunc); -rsRetVal rsCStrTrimTrailingWhiteSpace(cstr_t *pThis); rsRetVal cstrTrimTrailingWhiteSpace(cstr_t *pThis); /** diff --git a/runtime/typedefs.h b/runtime/typedefs.h index d3f68b4a..2efd7d65 100644 --- a/runtime/typedefs.h +++ b/runtime/typedefs.h @@ -25,6 +25,7 @@ */ #ifndef INCLUDED_TYPEDEFS_H #define INCLUDED_TYPEDEFS_H +#include <stdint.h> #if defined(__FreeBSD__) #include <sys/types.h> #endif @@ -63,6 +64,7 @@ typedef struct nsdsel_ptcp_s nsdsel_ptcp_t; typedef struct nsdsel_gtls_s nsdsel_gtls_t; typedef struct nsdpoll_ptcp_s nsdpoll_ptcp_t; typedef struct wti_s wti_t; +typedef struct msgPropDescr_s msgPropDescr_t; typedef struct msg msg_t; typedef struct queue_s qqueue_t; typedef struct prop_s prop_t; @@ -100,6 +102,9 @@ typedef struct outchannels_s outchannels_t; typedef struct modConfData_s modConfData_t; typedef struct instanceConf_s instanceConf_t; typedef struct ratelimit_s ratelimit_t; +typedef struct lookup_string_tab_etry_s lookup_string_tab_etry_t; +typedef struct lookup_tables_s lookup_tables_t; +typedef struct lookup_s lookup_t; typedef struct action_s action_t; typedef int rs_size_t; /* we do never need more than 2Gig strings, signed permits to * use -1 as a special flag. */ @@ -159,6 +164,53 @@ typedef enum { typedef off_t off64_t; #endif + +/* properties are now encoded as (tiny) integers. I do not use an enum as I would like + * to keep the memory footprint small (and thus cache hits high). + * rgerhards, 2009-06-26 + */ +typedef uintTiny propid_t; +#define PROP_INVALID 0 +#define PROP_MSG 1 +#define PROP_TIMESTAMP 2 +#define PROP_HOSTNAME 3 +#define PROP_SYSLOGTAG 4 +#define PROP_RAWMSG 5 +#define PROP_INPUTNAME 6 +#define PROP_FROMHOST 7 +#define PROP_FROMHOST_IP 8 +#define PROP_PRI 9 +#define PROP_PRI_TEXT 10 +#define PROP_IUT 11 +#define PROP_SYSLOGFACILITY 12 +#define PROP_SYSLOGFACILITY_TEXT 13 +#define PROP_SYSLOGSEVERITY 14 +#define PROP_SYSLOGSEVERITY_TEXT 15 +#define PROP_TIMEGENERATED 16 +#define PROP_PROGRAMNAME 17 +#define PROP_PROTOCOL_VERSION 18 +#define PROP_STRUCTURED_DATA 19 +#define PROP_APP_NAME 20 +#define PROP_PROCID 21 +#define PROP_MSGID 22 +#define PROP_PARSESUCCESS 23 +#define PROP_SYS_NOW 150 +#define PROP_SYS_YEAR 151 +#define PROP_SYS_MONTH 152 +#define PROP_SYS_DAY 153 +#define PROP_SYS_HOUR 154 +#define PROP_SYS_HHOUR 155 +#define PROP_SYS_QHOUR 156 +#define PROP_SYS_MINUTE 157 +#define PROP_SYS_MYHOSTNAME 158 +#define PROP_SYS_BOM 159 +#define PROP_SYS_UPTIME 160 +#define PROP_UUID 161 +#define PROP_CEE 200 +#define PROP_CEE_ALL_JSON 201 +#define PROP_LOCAL_VAR 202 +#define PROP_GLOBAL_VAR 203 + /* types of configuration handlers */ typedef enum cslCmdHdlrType { @@ -209,6 +261,13 @@ struct multi_submit_s { msg_t **ppMsgs; }; +/* the following structure is a helper to describe a message property */ +struct msgPropDescr_s { + propid_t id; + uchar *name; /* name and lenName are only set for dynamic */ + int nameLen; /* properties (JSON) */ +}; + #endif /* multi-include protection */ /* vim:set ai: */ diff --git a/runtime/wti.c b/runtime/wti.c index f91fb5a9..a9154670 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -44,12 +44,15 @@ #include "wti.h" #include "obj.h" #include "glbl.h" +#include "action.h" #include "atomic.h" /* static data */ DEFobjStaticHelpers DEFobjCurrIf(glbl) +pthread_key_t thrd_wti_key; + /* forward-definitions */ /* methods */ @@ -171,6 +174,8 @@ BEGINobjDestruct(wti) /* be sure to specify the object type also in END and CODE CODESTARTobjDestruct(wti) /* actual destruction */ batchFree(&pThis->batch); + free(pThis->actWrkrInfo); + pthread_cond_destroy(&pThis->pcondBusy); DESTROY_ATOMIC_HELPER_MUT(pThis->mutIsRunning); free(pThis->pszDbgHdr); @@ -181,6 +186,7 @@ ENDobjDestruct(wti) */ BEGINobjConstruct(wti) /* be sure to specify the object type also in END macro! */ INIT_ATOMIC_HELPER_MUT(pThis->mutIsRunning); + pthread_cond_init(&pThis->pcondBusy, NULL); ENDobjConstruct(wti) @@ -195,11 +201,20 @@ wtiConstructFinalize(wti_t *pThis) ISOBJ_TYPE_assert(pThis, wti); - DBGPRINTF("%s: finalizing construction of worker instance data\n", wtiGetDbgHdr(pThis)); + DBGPRINTF("%s: finalizing construction of worker instance data (for %d actions)\n", + wtiGetDbgHdr(pThis), iActionNbr); /* initialize our thread instance descriptor (no concurrency here) */ pThis->bIsRunning = RSFALSE; + /* must use calloc as we need zero-init */ + CHKmalloc(pThis->actWrkrInfo = calloc(iActionNbr, sizeof(actWrkrInfo_t))); + + if(pThis->pWtp == NULL) { + dbgprintf("wtiConstructFinalize: pWtp not set, this may be intentional\n"); + FINALIZE; + } + /* we now alloc the array for user pointers. We obtain the max from the queue itself. */ CHKiRet(pThis->pWtp->pfGetDeqBatchSize(pThis->pWtp->pUsr, &iDeqBatchSize)); CHKiRet(batchInit(&pThis->batch, iDeqBatchSize)); @@ -249,10 +264,10 @@ doIdleProcessing(wti_t *pThis, wtp_t *pWtp, int *pbInactivityTOOccured) if(pThis->bAlwaysRunning) { /* never shut down any started worker */ - d_pthread_cond_wait(pWtp->pcondBusy, pWtp->pmutUsr); + d_pthread_cond_wait(&pThis->pcondBusy, pWtp->pmutUsr); } else { timeoutComp(&t, pWtp->toWrkShutdown);/* get absolute timeout */ - if(d_pthread_cond_timedwait(pWtp->pcondBusy, pWtp->pmutUsr, &t) != 0) { + if(d_pthread_cond_timedwait(&pThis->pcondBusy, pWtp->pmutUsr, &t) != 0) { DBGPRINTF("%s: inactivity timeout, worker terminating...\n", wtiGetDbgHdr(pThis)); *pbInactivityTOOccured = 1; /* indicate we had a timeout */ } @@ -277,6 +292,7 @@ wtiWorker(wti_t *pThis) rsRetVal localRet; rsRetVal terminateRet; int iCancelStateSave; + int i; DEFiRet; ISOBJ_TYPE_assert(pThis, wti); @@ -286,6 +302,7 @@ wtiWorker(wti_t *pThis) dbgSetThrdName(pThis->pszDbgHdr); pthread_cleanup_push(wtiWorkerCancelCleanup, pThis); pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); +dbgprintf("DDDD: wti %p: worker starting\n", pThis); /* now we have our identity, on to real processing */ while(1) { /* loop will be broken below - need to do mutex locks */ @@ -332,9 +349,20 @@ wtiWorker(wti_t *pThis) bInactivityTOOccured = 0; /* reset for next run */ } + DBGPRINTF("DDDD: wti %p: worker cleanup up action instances\n", pThis); + for(i = 0 ; i < iActionNbr ; ++i) { + dbgprintf("wti %p, action %d, ptr %p\n", pThis, i, pThis->actWrkrInfo[i].actWrkrData); + if(pThis->actWrkrInfo[i].actWrkrData != NULL) { + dbgprintf("DDDD: calling freeWrkrData!\n"); + pThis->actWrkrInfo[i].pAction->pMod->mod.om.freeWrkrInstance(pThis->actWrkrInfo[i].actWrkrData); + pThis->actWrkrInfo[i].actWrkrData = NULL; /* re-init for next activation */ + } + } + /* indicate termination */ pthread_cleanup_pop(0); /* remove cleanup handler */ pthread_setcancelstate(iCancelStateSave, NULL); +dbgprintf("DDDD: wti %p: worker exiting\n", pThis); RETiRet; } @@ -374,6 +402,33 @@ finalize_it: } +/* This function returns (and creates if necessary) a dummy wti suitable + * for use by the rule engine. It is intended to be used for direct-mode + * main queues (folks, don't do that!). Once created, data is stored in + * thread-specific storage. + * Note: we do NOT do error checking -- if this functions fails, all the + * rest will fail as well... (also, it will only fail under OOM, so...). + * Memleak: we leak pWti's when run in direct mode. However, this is only + * a cosmetic leak, as we need them until all inputs are terminated, + * what means essentially until rsyslog itself is terminated. So we + * don't care -- it's just not nice in valgrind, but that's it. + */ +wti_t * +wtiGetDummy(void) +{ + wti_t *pWti; + + pWti = (wti_t*) pthread_getspecific(thrd_wti_key); + if(pWti == NULL) { + wtiConstruct(&pWti); + wtiConstructFinalize(pWti); + if(pthread_setspecific(thrd_wti_key, pWti) != 0) { + DBGPRINTF("wtiGetDummy: error setspecific thrd_wti_key\n"); + } + } + return pWti; +} + /* dummy */ rsRetVal wtiQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } @@ -383,6 +438,7 @@ BEGINObjClassExit(wti, OBJ_IS_CORE_MODULE) /* CHANGE class also in END MACRO! */ CODESTARTObjClassExit(nsdsel_gtls) /* release objects we no longer need */ objRelease(glbl, CORE_COMPONENT); + pthread_key_delete(thrd_wti_key); ENDObjClassExit(wti) @@ -391,8 +447,14 @@ ENDObjClassExit(wti) * rgerhards, 2008-01-09 */ BEGINObjClassInit(wti, 1, OBJ_IS_CORE_MODULE) /* one is the object version (most important for persisting) */ + int r; /* request objects we use */ CHKiRet(objUse(glbl, CORE_COMPONENT)); + r = pthread_key_create(&thrd_wti_key, NULL); + if(r != 0) { + dbgprintf("wti.c: pthread_key_create failed\n"); + iRet = RS_RET_ERR; + } ENDObjClassInit(wti) /* vi:set ai: diff --git a/runtime/wti.h b/runtime/wti.h index 014251f0..adc7897c 100644 --- a/runtime/wti.h +++ b/runtime/wti.h @@ -1,6 +1,6 @@ /* Definition of the worker thread instance (wti) class. * - * Copyright 2008-2012 Adiscon GmbH. + * Copyright 2008-2013 Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -26,18 +26,67 @@ #include "wtp.h" #include "obj.h" #include "batch.h" +#include "action.h" +#define ACT_STATE_RDY 0 /* action ready, waiting for new transaction */ +#define ACT_STATE_ITX 1 /* transaction active, waiting for new data or commit */ +#define ACT_STATE_COMM 2 /* transaction finished (a transient state) */ +#define ACT_STATE_RTRY 3 /* failure occured, trying to restablish ready state */ +#define ACT_STATE_SUSP 4 /* suspended due to failure (return fail until timeout expired) */ +/* note: 3 bit bit field --> highest value is 7! */ + +/* The following structure defines immutable parameters which need to + * be passed as action parameters. Note that the current implementation + * does NOT focus on performance, but on a simple PoC in order to get + * things going. TODO: Once it works, revisit this code and think about + * an array implementation. We also need to support other passing modes + * as well. -- gerhards, 2013-11-04 + */ +typedef struct actWrkrIParams { + int msgFlags; + /* following are caches to save allocs if not absolutely necessary */ + uchar *staticActStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /**< for strings */ + /* a cache to save malloc(), if not absolutely necessary */ + unsigned staticLenStrings[CONF_OMOD_NUMSTRINGS_MAXSIZE]; + /* and the same for the message length (if used) */ + void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE]; +} actWrkrIParams_t; + +typedef struct actWrkrInfo { + action_t *pAction; + void *actWrkrData; + uint16_t uResumeOKinRow;/* number of times in a row that resume said OK with an immediate failure following */ + int iNbrResRtry; /* number of retries since last suspend */ + struct { + unsigned actState : 3; + } flags; + actWrkrIParams_t *iparams; + int currIParam; + int maxIParams; /* current max */ + void *staticActParams[CONF_OMOD_NUMSTRINGS_MAXSIZE]; /* for non-strings */ +} actWrkrInfo_t; + /* the worker thread instance class */ struct wti_s { BEGINobjInstance; pthread_t thrdID; /* thread ID */ int bIsRunning; /* is this thread currently running? (must be int for atomic op!) */ sbool bAlwaysRunning; /* should this thread always run? */ + int *pbShutdownImmediate;/* end processing of this batch immediately if set to 1 */ wtp_t *pWtp; /* my worker thread pool (important if only the work thread instance is passed! */ batch_t batch; /* pointer to an object array meaningful for current user pointer (e.g. queue pUsr data elemt) */ uchar *pszDbgHdr; /* header string for debug messages */ + actWrkrInfo_t *actWrkrInfo; /* *array* of action wrkr infos for all actions (sized for max nbr of actions in config!) */ + pthread_cond_t pcondBusy; /* condition to wake up the worker, protected by pmutUsr in wtp */ DEF_ATOMIC_HELPER_MUT(mutIsRunning); + struct { + uint8_t bPrevWasSuspended; + uint8_t bDoAutoCommit; /* do a commit after each message + * this is usually set for batches with 0 element, but may + * also be added as a user-selectable option (not implemented yet) + */ + } execState; /* state for the execution engine */ }; @@ -52,8 +101,96 @@ rsRetVal wtiSetAlwaysRunning(wti_t *pThis); rsRetVal wtiSetState(wti_t *pThis, sbool bNew); rsRetVal wtiWakeupThrd(wti_t *pThis); sbool wtiGetState(wti_t *pThis); +wti_t *wtiGetDummy(void); PROTOTYPEObjClassInit(wti); PROTOTYPEpropSetMeth(wti, pszDbgHdr, uchar*); PROTOTYPEpropSetMeth(wti, pWtp, wtp_t*); +static inline uint8_t +getActionStateByNbr(wti_t *pWti, int iActNbr) +{ + return((uint8_t) pWti->actWrkrInfo[iActNbr].flags.actState); +} + +static inline uint8_t +getActionState(wti_t *pWti, action_t *pAction) +{ + return((uint8_t) pWti->actWrkrInfo[pAction->iActionNbr].flags.actState); +} + +static inline void +setActionState(wti_t *pWti, action_t *pAction, uint8_t newState) +{ + pWti->actWrkrInfo[pAction->iActionNbr].flags.actState = newState; +} + +static inline uint16_t +getActionResumeInRow(wti_t *pWti, action_t *pAction) +{ + return(pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow); +} + +static inline void +setActionResumeInRow(wti_t *pWti, action_t *pAction, uint16_t val) +{ + pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow = val; +} + +static inline void +incActionResumeInRow(wti_t *pWti, action_t *pAction) +{ + pWti->actWrkrInfo[pAction->iActionNbr].uResumeOKinRow++; +} + +static inline int +getActionNbrResRtry(wti_t *pWti, action_t *pAction) +{ + return(pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry); +} + +static inline void +setActionNbrResRtry(wti_t *pWti, action_t *pAction, uint16_t val) +{ + pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry = val; +} + +static inline void +incActionNbrResRtry(wti_t *pWti, action_t *pAction) +{ + pWti->actWrkrInfo[pAction->iActionNbr].iNbrResRtry++; +} + +static inline rsRetVal +wtiNewIParam(wti_t *pWti, action_t *pAction, actWrkrIParams_t **piparams) +{ + actWrkrInfo_t *wrkrInfo; + actWrkrIParams_t *iparams; + int newMax; + DEFiRet; + + wrkrInfo = &(pWti->actWrkrInfo[pAction->iActionNbr]); + if(wrkrInfo->currIParam == wrkrInfo->maxIParams) { + /* we need to extend */ + dbgprintf("DDDD: extending iparams, max %d\n", wrkrInfo->maxIParams); + newMax = (wrkrInfo->maxIParams == 0) ? CONF_IPARAMS_BUFSIZE : 2 * wrkrInfo->maxIParams; + CHKmalloc(iparams = realloc(wrkrInfo->iparams, sizeof(actWrkrIParams_t) * newMax)); + wrkrInfo->iparams = iparams; + wrkrInfo->maxIParams = newMax; + } +dbgprintf("DDDD: adding param %d for action %d\n", wrkrInfo->currIParam, pAction->iActionNbr); + iparams = wrkrInfo->iparams + wrkrInfo->currIParam; + memset(iparams, 0, sizeof(actWrkrIParams_t)); + *piparams = iparams; + ++wrkrInfo->currIParam; + +finalize_it: + RETiRet; +} + +static inline void +wtiResetExecState(wti_t *pWti, batch_t *pBatch) +{ + pWti->execState.bPrevWasSuspended = 0; + pWti->execState.bDoAutoCommit = (batchNumMsgs(pBatch) == 1); +} #endif /* #ifndef WTI_H_INCLUDED */ diff --git a/runtime/wtp.c b/runtime/wtp.c index 19151e7c..66942e6b 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -8,7 +8,7 @@ * (and in the web doc set on http://www.rsyslog.com/doc). Be sure to read it * if you are getting aquainted to the object. * - * Copyright 2008,2009 Rainer Gerhards and Adiscon GmbH. + * Copyright 2008-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -91,6 +91,7 @@ BEGINobjConstruct(wtp) /* be sure to specify the object type also in END macro! pthread_cond_init(&pThis->condThrdTrm, NULL); pthread_attr_init(&pThis->attrThrd); /* Set thread scheduling policy to default */ +#warning do we need this any longer? I think it was a cure for an already fixed bug.. #ifdef HAVE_PTHREAD_SETSCHEDPARAM pthread_attr_setschedpolicy(&pThis->attrThrd, default_thr_sched_policy); pthread_attr_setschedparam(&pThis->attrThrd, &default_sched_param); @@ -121,7 +122,8 @@ wtpConstructFinalize(wtp_t *pThis) ISOBJ_TYPE_assert(pThis, wtp); - DBGPRINTF("%s: finalizing construction of worker thread pool\n", wtpGetDbgHdr(pThis)); + DBGPRINTF("%s: finalizing construction of worker thread pool (numworkerThreads %d)\n", + wtpGetDbgHdr(pThis), pThis->iNumWorkerThreads); /* alloc and construct workers - this can only be done in finalizer as we previously do * not know the max number of workers */ @@ -233,9 +235,9 @@ wtpShutdownAll(wtp_t *pThis, wtpState_t tShutdownCmd, struct timespec *ptTimeout /* lock mutex to prevent races (may otherwise happen during idle processing and such...) */ d_pthread_mutex_lock(pThis->pmutUsr); wtpSetState(pThis, tShutdownCmd); - pthread_cond_broadcast(pThis->pcondBusy); /* wake up all workers */ /* awake workers in retry loop */ for(i = 0 ; i < pThis->iNumWorkerThreads ; ++i) { + pthread_cond_signal(&pThis->pWrkr[i]->pcondBusy); wtiWakeupThrd(pThis->pWrkr[i]); } d_pthread_mutex_unlock(pThis->pmutUsr); @@ -455,7 +457,7 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr) { DEFiRet; int nMissing; /* number workers missing to run */ - int i; + int i, nRunning; ISOBJ_TYPE_assert(pThis, wtp); @@ -475,7 +477,13 @@ wtpAdviseMaxWorkers(wtp_t *pThis, int nMaxWrkr) CHKiRet(wtpStartWrkr(pThis)); } } else { - pthread_cond_signal(pThis->pcondBusy); + /* we have needed number of workers, but they may be sleeping */ + for(i = 0, nRunning = 0; i < pThis->iNumWorkerThreads && nRunning < nMaxWrkr; ++i) { + if (wtiGetState(pThis->pWrkr[i]) != WRKTHRD_STOPPED) { + pthread_cond_signal(&pThis->pWrkr[i]->pcondBusy); + nRunning++; + } + } } @@ -490,7 +498,6 @@ DEFpropSetMeth(wtp, wtpState, wtpState_t) DEFpropSetMeth(wtp, iNumWorkerThreads, int) DEFpropSetMeth(wtp, pUsr, void*) DEFpropSetMethPTR(wtp, pmutUsr, pthread_mutex_t) -DEFpropSetMethPTR(wtp, pcondBusy, pthread_cond_t) DEFpropSetMethFP(wtp, pfChkStopWrkr, rsRetVal(*pVal)(void*, int)) DEFpropSetMethFP(wtp, pfRateLimiter, rsRetVal(*pVal)(void*)) DEFpropSetMethFP(wtp, pfGetDeqBatchSize, rsRetVal(*pVal)(void*, int*)) diff --git a/runtime/wtp.h b/runtime/wtp.h index 25992f7f..4bc284cb 100644 --- a/runtime/wtp.h +++ b/runtime/wtp.h @@ -56,7 +56,6 @@ struct wtp_s { void *pUsr; /* pointer to user object (in this case, the queue the wtp belongs to) */ pthread_attr_t attrThrd;/* attribute for new threads (created just once and cached here) */ pthread_mutex_t *pmutUsr; - pthread_cond_t *pcondBusy; /* condition the user will signal "busy again, keep runing" on (awakes worker) */ rsRetVal (*pfChkStopWrkr)(void *pUsr, int); rsRetVal (*pfGetDeqBatchSize)(void *pUsr, int*); /* obtains max dequeue count from queue config */ rsRetVal (*pfObjProcessed)(void *pUsr, wti_t *pWti); /* indicate user object is processed */ @@ -95,6 +94,5 @@ PROTOTYPEpropSetMeth(wtp, iMaxWorkerThreads, int); PROTOTYPEpropSetMeth(wtp, pUsr, void*); PROTOTYPEpropSetMeth(wtp, iNumWorkerThreads, int); PROTOTYPEpropSetMethPTR(wtp, pmutUsr, pthread_mutex_t); -PROTOTYPEpropSetMethPTR(wtp, pcondBusy, pthread_cond_t); #endif /* #ifndef WTP_H_INCLUDED */ |