diff options
Diffstat (limited to 'runtime')
| -rw-r--r-- | runtime/Makefile.am | 17 | ||||
| -rw-r--r-- | runtime/apc.c | 400 | ||||
| -rw-r--r-- | runtime/apc.h | 56 | ||||
| -rw-r--r-- | runtime/cfsysline.c | 6 | ||||
| -rw-r--r-- | runtime/conf.c | 117 | ||||
| -rw-r--r-- | runtime/conf.h | 6 | ||||
| -rw-r--r-- | runtime/linkedlist.c | 2 | ||||
| -rw-r--r-- | runtime/msg.c | 10 | ||||
| -rw-r--r-- | runtime/msg.h | 3 | ||||
| -rw-r--r-- | runtime/obj-types.h | 5 | ||||
| -rw-r--r-- | runtime/obj.c | 107 | ||||
| -rw-r--r-- | runtime/obj.h | 2 | ||||
| -rw-r--r-- | runtime/queue.c | 94 | ||||
| -rw-r--r-- | runtime/queue.h | 2 | ||||
| -rw-r--r-- | runtime/rsyslog.c | 30 | ||||
| -rw-r--r-- | runtime/rsyslog.h | 28 | ||||
| -rw-r--r-- | runtime/rule.c | 450 | ||||
| -rw-r--r-- | runtime/rule.h | 77 | ||||
| -rw-r--r-- | runtime/ruleset.c | 451 | ||||
| -rw-r--r-- | runtime/ruleset.h | 60 | ||||
| -rw-r--r-- | runtime/srUtils.h | 14 | ||||
| -rw-r--r-- | runtime/srutils.c | 27 | ||||
| -rw-r--r-- | runtime/stream.c | 681 | ||||
| -rw-r--r-- | runtime/stream.h | 115 | ||||
| -rw-r--r-- | runtime/sysvar.c | 2 | ||||
| -rw-r--r-- | runtime/unicode-helper.h | 25 | ||||
| -rw-r--r-- | runtime/wti.c | 3 | ||||
| -rw-r--r-- | runtime/wtp.c | 3 | ||||
| -rw-r--r-- | runtime/zlibw.c | 125 | ||||
| -rw-r--r-- | runtime/zlibw.h | 46 |
30 files changed, 2638 insertions, 326 deletions
diff --git a/runtime/Makefile.am b/runtime/Makefile.am index eeb656d6..c2ef7cfa 100644 --- a/runtime/Makefile.am +++ b/runtime/Makefile.am @@ -39,6 +39,8 @@ librsyslog_la_SOURCES = \ obj.h \ modules.c \ modules.h \ + apc.c \ + apc.h \ sync.c \ sync.h \ expr.c \ @@ -67,6 +69,10 @@ librsyslog_la_SOURCES = \ vmop.h \ queue.c \ queue.h \ + ruleset.c \ + ruleset.h \ + rule.c \ + rule.h \ cfsysline.c \ cfsysline.h \ \ @@ -105,6 +111,17 @@ lmregexp_la_LDFLAGS = -module -avoid-version lmregexp_la_LIBADD = endif +# +# zlib support +# +if ENABLE_ZLIB +pkglib_LTLIBRARIES += lmzlibw.la +lmzlibw_la_SOURCES = zlibw.c zlibw.h +lmzlibw_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS) +lmzlibw_la_LDFLAGS = -module -avoid-version +lmzlibw_la_LIBADD = +endif + if ENABLE_INET pkglib_LTLIBRARIES += lmnet.la lmnetstrms.la # diff --git a/runtime/apc.c b/runtime/apc.c new file mode 100644 index 00000000..b0b5f298 --- /dev/null +++ b/runtime/apc.c @@ -0,0 +1,400 @@ +/* apc.c - asynchronous procedure call support + * + * An asynchronous procedure call (APC) is a procedure call (guess what) that is potentially run + * asynchronously to its main thread. It can be scheduled to occur at a caller-provided time. + * As long as the procedure has not been called, the APC entry may be modified by the caller + * or deleted. It is the caller's purpose to make sure proper synchronization is in place. + * The APC object only case about APC's own control structures (which *are* properly + * guarded by synchronization primitives). + * + * Module begun 2009-06-15 by Rainer Gerhards + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ + +#include "config.h" +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <pthread.h> + +#include "rsyslog.h" +#include "obj.h" +#include "apc.h" +#include "srUtils.h" + +/* static data */ +DEFobjStaticHelpers + +/* following is a used to implement a monotonically increasing id for the apcs. That + * ID can be used to cancel an apc request. Note that the ID is generated with modulo + * arithmetic, so at some point, it will wrap. Howerver, this happens at 2^32-1 at + * earliest, so this is not considered a problem. + */ +apc_id_t apcID = 0; + +/* private data structures */ + +/* the apc list and its entries + * This is a doubly-linked list as we need to be able to do inserts + * and deletes right in the middle of the list. It is inspired by the + * Unix callout mechanism. + * Note that we support two generic caller-provided parameters as + * experience shows that at most two are often used. This causes very + * little overhead, but simplifies caller code in cases where exactly + * two parameters are needed. We hope this is a useful optimizaton. + * rgerhards, 2009-06-15 + */ +typedef struct apc_list_s { + struct apc_list_s *pNext; + struct apc_list_s *pPrev; + apc_id_t id; + apc_t *pApc; /* pointer to the APC object to be scheduled */ +} apc_list_t; + +apc_list_t *apcListRoot = NULL; +apc_list_t *apcListTail = NULL; +pthread_mutex_t listMutex; /* needs to be locked for all list operations */ + + +/* destructor for the apc object */ +BEGINobjDestruct(apc) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(apc) +ENDobjDestruct(apc) + + +/* ------------------------------ APC list handling functions ------------------------------ */ + +/* Function that handles changes to the list root. Most importantly, this function + * needs to schedule a new timer. It is OK to call this function with an empty list. + */ +static rsRetVal +listRootChanged(void) +{ + DEFiRet; + + if(apcListRoot == NULL) + FINALIZE; + + // TODO: implement! + +finalize_it: + RETiRet; +} + + +/* insert an apc entry into the APC list. The same entry MUST NOT already be present! + */ +static rsRetVal +insertApc(apc_t *pThis, apc_id_t *pID) +{ + apc_list_t *pCurr; + apc_list_t *pNew; + DEFiRet; + + CHKmalloc(pNew = (apc_list_t*) calloc(1, sizeof(apc_list_t))); + pNew->pApc = pThis; + pNew->id = *pID = apcID++; +dbgprintf("insert apc %p, id %ld\n", pThis, pNew->id); + + /* find right list location */ + if(apcListRoot == NULL) { + /* no need to search, list is empty */ + apcListRoot = pNew; + apcListTail = pNew; + CHKiRet(listRootChanged()); + } else { + for(pCurr = apcListRoot ; pCurr != NULL ; pCurr = pCurr->pNext) { + if(pCurr->pApc->ttExec > pThis->ttExec) + break; + } + + if(pCurr == NULL) { + /* insert at tail */ + pNew->pPrev = apcListTail; + apcListTail->pNext = pNew; + apcListTail = pNew; + } else { + if(pCurr == apcListRoot) { + /* new first entry */ + pCurr->pPrev = pNew; + pNew->pNext = pCurr; + apcListRoot = pNew; + CHKiRet(listRootChanged()); + } else { + /* in the middle of the list */ + pCurr->pPrev = pNew; + pNew->pNext = pCurr; + } + } + } + + +finalize_it: + RETiRet; +} + + +/* Delete an apc entry from the APC list. It is OK if the entry is not found, + * in this case we assume it already has been processed. + */ +static rsRetVal +deleteApc(apc_id_t id) +{ + apc_list_t *pCurr; + DEFiRet; + +dbgprintf("trying to delete apc %ld\n", id); + for(pCurr = apcListRoot ; pCurr != NULL ; pCurr = pCurr->pNext) { + if(pCurr->id == id) { +RUNLOG_STR("apc id found, now deleting!\n"); + if(pCurr == apcListRoot) { + apcListRoot = pCurr->pNext; + CHKiRet(listRootChanged()); + } else { + pCurr->pPrev->pNext = pCurr->pNext; + } + if(pCurr->pNext == NULL) { + apcListTail = pCurr->pPrev; + } else { + pCurr->pNext->pPrev = pCurr->pPrev; + } + free(pCurr); + pCurr = NULL; + break; + } + } + +finalize_it: + RETiRet; +} + + +/* unlist all elements up to the current timestamp. Return this as a seperate list + * to the caller. Returns an empty (NULL ptr) list if there are no such elements. + * The caller must handle that gracefully. The list is returned in the parameter. + */ +static rsRetVal +unlistCurrent(apc_list_t **ppList) +{ + apc_list_t *pCurr; + time_t tCurr; + DEFiRet; + assert(ppList != NULL); + + time(&tCurr); + + if(apcListRoot == NULL || apcListRoot->pApc->ttExec > tCurr) { + *ppList = NULL; + FINALIZE; + } + + *ppList = apcListRoot; + /* now search up to which entry we need to execute */ + for(pCurr = apcListRoot ; pCurr != NULL && pCurr->pApc->ttExec <= tCurr ; pCurr = pCurr->pNext) { + /*JUST SKIP TO LAST ELEMENT*/; + } + + if(pCurr == NULL) { + /* all elements can be unlisted */ + apcListRoot = NULL; + apcListTail = NULL; + } else { + /* need to set a new root */ + pCurr->pPrev->pNext = NULL; /* terminate newly unlisted list */ + pCurr->pPrev = NULL; /* we are the new root */ + apcListRoot = pCurr; + } + +finalize_it: + RETiRet; +} + + +/* ------------------------------ END APC list handling functions ------------------------------ */ + + +/* execute all list elements that are currently scheduled for execution. We do this in two phases. + * In the first phase, we look the list mutex and move everything from the head of the queue to + * the current timestamp to a new to-be-executed list. Then we unlock the mutex and do the actual + * exec (which may take some time). + * Note that the caller is responsible for proper + * caller-level synchronization. The caller may schedule another Apc, this module must + * ensure that (and it does so by not locking the list mutex while we call the Apc). + * Note: this function "consumes" the apc_t, so it is no longer existing after this + * function returns. + */ +// TODO make static and associated with our own pthread-based timer +rsRetVal +execScheduled(void) +{ + apc_list_t *pExecList; + apc_list_t *pCurr; + apc_list_t *pNext; + DEFVARS_mutexProtection_uncond; + DEFiRet; + + BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + iRet = unlistCurrent(&pExecList); + END_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + CHKiRet(iRet); + + DBGPRINTF("running apc scheduler - we have %s to execute\n", + pExecList == NULL ? "nothing" : "something"); + for(pCurr = pExecList ; pCurr != NULL ; pCurr = pNext) { +dbgprintf("executing apc list entry %p, apc %p\n", pCurr, pCurr->pApc); + pNext = pCurr->pNext; + pCurr->pApc->pProc(pCurr->pApc->param1, pCurr->pApc->param2); + apcDestruct(&pCurr->pApc); + free(pCurr); + } + +finalize_it: + RETiRet; +} + + +/* Standard-Constructor + */ +BEGINobjConstruct(apc) /* be sure to specify the object type also in END macro! */ +ENDobjConstruct(apc) + + +/* ConstructionFinalizer + * Note that we use a non-standard calling interface: pID returns the current APC + * id. This is the only way to handle the situation without the need for extra + * locking. + * rgerhards, 2008-01-09 + */ +static rsRetVal +apcConstructFinalize(apc_t *pThis, apc_id_t *pID) +{ + DEFVARS_mutexProtection_uncond; + DEFiRet; + ISOBJ_TYPE_assert(pThis, apc); + assert(pID != NULL); + BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + insertApc(pThis, pID); + END_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); +RUNLOG_STR("apcConstructFinalize post mutex unlock\n"); + RETiRet; +} + + +/* some set methods */ +static rsRetVal +SetProcedure(apc_t *pThis, void (*pProc)(void*, void*)) +{ + ISOBJ_TYPE_assert(pThis, apc); + pThis->pProc = pProc; + return RS_RET_OK; +} +static rsRetVal +SetParam1(apc_t *pThis, void *param1) +{ + ISOBJ_TYPE_assert(pThis, apc); + pThis->param1 = param1; + return RS_RET_OK; +} +static rsRetVal +SetParam2(apc_t *pThis, void *param2) +{ + ISOBJ_TYPE_assert(pThis, apc); + pThis->param1 = param2; + return RS_RET_OK; +} + + +/* cancel an Apc request, ID is provided. It is OK if the ID can not be found, this may + * happen if the Apc was executed in the mean time. So it is safe to call CancelApc() at + * any time. + */ +static rsRetVal +CancelApc(apc_id_t id) +{ + DEFVARS_mutexProtection_uncond; + + BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + deleteApc(id); + END_MTX_PROTECTED_OPERATIONS_UNCOND(&listMutex); + return RS_RET_OK; +} + + +/* debugprint for the apc object */ +BEGINobjDebugPrint(apc) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDebugPrint(apc) + dbgoprint((obj_t*) pThis, "APC module, currently no state info available\n"); +ENDobjDebugPrint(apc) + + +/* queryInterface function + */ +BEGINobjQueryInterface(apc) +CODESTARTobjQueryInterface(apc) + if(pIf->ifVersion != apcCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->Construct = apcConstruct; + pIf->ConstructFinalize = apcConstructFinalize; + pIf->Destruct = apcDestruct; + pIf->DebugPrint = apcDebugPrint; + pIf->CancelApc = CancelApc; + pIf->SetProcedure = SetProcedure; + pIf->SetParam1 = SetParam1; + pIf->SetParam2 = SetParam2; +finalize_it: +ENDobjQueryInterface(apc) + + +/* Exit the apc class. + * rgerhards, 2009-04-06 + */ +BEGINObjClassExit(apc, OBJ_IS_CORE_MODULE) /* class, version */ + //objRelease(apcstk, CORE_COMPONENT); + pthread_mutex_destroy(&listMutex); +ENDObjClassExit(apc) + + +/* Initialize the apc class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-19 + */ +BEGINObjClassInit(apc, 1, OBJ_IS_CORE_MODULE) /* class, version */ + /* request objects we use */ + //CHKiRet(objUse(apcstk, CORE_COMPONENT)); + + /* set our own handlers */ + OBJSetMethodHandler(objMethod_DEBUGPRINT, apcDebugPrint); + OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, apcConstructFinalize); + + /* do other initializations */ + pthread_mutex_init(&listMutex, NULL); +ENDObjClassInit(apc) + +/* vi:set ai: + */ diff --git a/runtime/apc.h b/runtime/apc.h new file mode 100644 index 00000000..7c679b97 --- /dev/null +++ b/runtime/apc.h @@ -0,0 +1,56 @@ +/* The apc object. + * + * See apc.c for more information. + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_APC_H +#define INCLUDED_APC_H + +/* the apc object */ +typedef struct apc_s { + BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ + time_t ttExec; /* when to call procedure (so far seconds...) */ + void (*pProc)(void*, void*); /* which procedure to call */ + void *param1; /* user-supplied parameters */ + void *param2; /* user-supplied parameters */ +} apc_t; + +typedef unsigned long apc_id_t; /* monotonically incrementing apc ID */ + +/* interfaces */ +BEGINinterface(apc) /* name must also be changed in ENDinterface macro! */ + INTERFACEObjDebugPrint(apc); + rsRetVal (*Construct)(apc_t **ppThis); + rsRetVal (*ConstructFinalize)(apc_t *pThis, apc_id_t *); + rsRetVal (*Destruct)(apc_t **ppThis); + rsRetVal (*SetProcedure)(apc_t *pThis, void (*pProc)(void*, void*)); + rsRetVal (*SetParam1)(apc_t *pThis, void *); + rsRetVal (*SetParam2)(apc_t *pThis, void *); + rsRetVal (*CancelApc)(apc_id_t); +ENDinterface(apc) +#define apcCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ + + +/* prototypes */ +PROTOTYPEObj(apc); + +#endif /* #ifndef INCLUDED_APC_H */ diff --git a/runtime/cfsysline.c b/runtime/cfsysline.c index 0fb4247d..184c0d87 100644 --- a/runtime/cfsysline.c +++ b/runtime/cfsysline.c @@ -814,7 +814,7 @@ rsRetVal regCfSysLineHdlr(uchar *pCmdName, int bChainingPermitted, ecslCmdHdrlTy CHKiRet(cslcConstruct(&pThis, bChainingPermitted)); CHKiRet_Hdlr(cslcAddHdlr(pThis, eType, pHdlr, pData, pOwnerCookie)) { cslcDestruct(pThis); - goto finalize_it; + FINALIZE; } /* important: add to list, AFTER everything else is OK. Else * we mess up things in the error case. @@ -825,7 +825,7 @@ rsRetVal regCfSysLineHdlr(uchar *pCmdName, int bChainingPermitted, ecslCmdHdrlTy } CHKiRet_Hdlr(llAppend(&llCmdList, pMyCmdName, (void*) pThis)) { cslcDestruct(pThis); - goto finalize_it; + FINALIZE; } } else { /* command already exists, are we allowed to chain? */ @@ -834,7 +834,7 @@ rsRetVal regCfSysLineHdlr(uchar *pCmdName, int bChainingPermitted, ecslCmdHdrlTy } CHKiRet_Hdlr(cslcAddHdlr(pThis, eType, pHdlr, pData, pOwnerCookie)) { cslcDestruct(pThis); - goto finalize_it; + FINALIZE; } } diff --git a/runtime/conf.c b/runtime/conf.c index 65fc66f1..581254f0 100644 --- a/runtime/conf.c +++ b/runtime/conf.c @@ -69,13 +69,16 @@ #include "expr.h" #include "ctok.h" #include "ctok_token.h" +#include "rule.h" +#include "ruleset.h" +#include "unicode-helper.h" #ifdef OS_SOLARIS # define NAME_MAX MAXNAMELEN #endif /* forward definitions */ -static rsRetVal cfline(uchar *line, selector_t **pfCurr); +static rsRetVal cfline(uchar *line, rule_t **pfCurr); static rsRetVal processConfFile(uchar *pConfFile); @@ -87,6 +90,8 @@ DEFobjCurrIf(ctok_token) DEFobjCurrIf(module) DEFobjCurrIf(errmsg) DEFobjCurrIf(net) +DEFobjCurrIf(rule) +DEFobjCurrIf(ruleset) static int iNbrActions; /* number of actions the running config has. Needs to be init on ReInitConf() */ @@ -392,15 +397,17 @@ finalize_it: static rsRetVal processConfFile(uchar *pConfFile) { - DEFiRet; int iLnNbr = 0; FILE *cf; - selector_t *fCurr = NULL; + rule_t *pCurrRule = NULL; uchar *p; uchar cbuf[CFGLNSIZ]; uchar *cline; int i; int bHadAnError = 0; + uchar *pszOrgLine = NULL; + size_t lenLine; + DEFiRet; ASSERT(pConfFile != NULL); if((cf = fopen((char*)pConfFile, "r")) == NULL) { @@ -413,9 +420,12 @@ processConfFile(uchar *pConfFile) while (fgets((char*)cline, sizeof(cbuf) - (cline - cbuf), cf) != NULL) { ++iLnNbr; /* drop LF - TODO: make it better, replace fgets(), but its clean as it is */ - if(cline[strlen((char*)cline)-1] == '\n') { - cline[strlen((char*)cline) -1] = '\0'; + lenLine = ustrlen(cline); + if(cline[lenLine-1] == '\n') { + cline[lenLine-1] = '\0'; } + free(pszOrgLine); + pszOrgLine = ustrdup(cline); /* save if needed for errmsg, NULL ptr is OK */ /* check for end-of-section, comments, strip off trailing * spaces and newline character. */ @@ -429,7 +439,6 @@ processConfFile(uchar *pConfFile) * TODO: review the code at whole - this is highly suspect (but will go away * once we do the rest of RainerScript). */ - /* was: strcpy((char*)cline, (char*)p); */ for( i = 0 ; p[i] != '\0' ; ++i) { cline[i] = p[i]; } @@ -453,7 +462,7 @@ processConfFile(uchar *pConfFile) /* we now have the complete line, and are positioned at the first non-whitespace * character. So let's process it */ - if(cfline(cbuf, &fCurr) != RS_RET_OK) { + if(cfline(cbuf, &pCurrRule) != RS_RET_OK) { /* we log a message, but otherwise ignore the error. After all, the next * line can be correct. -- rgerhards, 2007-08-02 */ @@ -461,28 +470,32 @@ processConfFile(uchar *pConfFile) dbgprintf("config line NOT successfully processed\n"); snprintf((char*)szErrLoc, sizeof(szErrLoc) / sizeof(uchar), "%s, line %d", pConfFile, iLnNbr); - errmsg.LogError(0, NO_ERRCODE, "the last error occured in %s", (char*)szErrLoc); + errmsg.LogError(0, NO_ERRCODE, "the last error occured in %s:\"%s\"", (char*)szErrLoc, (char*)pszOrgLine); bHadAnError = 1; } } /* we probably have one selector left to be added - so let's do that now */ - CHKiRet(selectorAddList(fCurr)); + if(pCurrRule != NULL) { + CHKiRet(ruleset.AddRule(rule.GetAssRuleset(pCurrRule), &pCurrRule)); + } /* close the configuration file */ - (void) fclose(cf); + fclose(cf); finalize_it: if(iRet != RS_RET_OK) { char errStr[1024]; - if(fCurr != NULL) - selectorDestruct(fCurr); + if(pCurrRule != NULL) + rule.Destruct(&pCurrRule); rs_strerror_r(errno, errStr, sizeof(errStr)); dbgprintf("error %d processing config file '%s'; os error (if any): %s\n", iRet, pConfFile, errStr); } + free(pszOrgLine); + if(bHadAnError && (iRet == RS_RET_OK)) { /* a bit dirty, enhance in future releases */ iRet = RS_RET_NONFATAL_CONFIG_ERR; } @@ -586,7 +599,7 @@ cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pOMSR, int * rgerhards 2005-09-15 */ /* GPLv3 - stems back to sysklogd */ -static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register selector_t *f) +static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register rule_t *pRule) { uchar *p; register uchar *q; @@ -601,17 +614,17 @@ static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register selector_t *f ASSERT(pline != NULL); ASSERT(*pline != NULL); - ASSERT(f != NULL); + ISOBJ_TYPE_assert(pRule, rule); dbgprintf(" - traditional PRI filter\n"); errno = 0; /* keep strerror_r() stuff out of logerror messages */ - f->f_filter_type = FILTER_PRI; + pRule->f_filter_type = FILTER_PRI; /* Note: file structure is pre-initialized to zero because it was * created with calloc()! */ for (i = 0; i <= LOG_NFACILITIES; i++) { - f->f_filterData.f_pmask[i] = TABLE_NOPRI; + pRule->f_filterData.f_pmask[i] = TABLE_NOPRI; } /* scan through the list of selectors */ @@ -666,32 +679,32 @@ static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register selector_t *f for (i = 0; i <= LOG_NFACILITIES; i++) { if ( pri == INTERNAL_NOPRI ) { if ( ignorepri ) - f->f_filterData.f_pmask[i] = TABLE_ALLPRI; + pRule->f_filterData.f_pmask[i] = TABLE_ALLPRI; else - f->f_filterData.f_pmask[i] = TABLE_NOPRI; + pRule->f_filterData.f_pmask[i] = TABLE_NOPRI; } else if ( singlpri ) { if ( ignorepri ) - f->f_filterData.f_pmask[i] &= ~(1<<pri); + pRule->f_filterData.f_pmask[i] &= ~(1<<pri); else - f->f_filterData.f_pmask[i] |= (1<<pri); + pRule->f_filterData.f_pmask[i] |= (1<<pri); } else { if ( pri == TABLE_ALLPRI ) { if ( ignorepri ) - f->f_filterData.f_pmask[i] = TABLE_NOPRI; + pRule->f_filterData.f_pmask[i] = TABLE_NOPRI; else - f->f_filterData.f_pmask[i] = TABLE_ALLPRI; + pRule->f_filterData.f_pmask[i] = TABLE_ALLPRI; } else { if ( ignorepri ) for (i2= 0; i2 <= pri; ++i2) - f->f_filterData.f_pmask[i] &= ~(1<<i2); + pRule->f_filterData.f_pmask[i] &= ~(1<<i2); else for (i2= 0; i2 <= pri; ++i2) - f->f_filterData.f_pmask[i] |= (1<<i2); + pRule->f_filterData.f_pmask[i] |= (1<<i2); } } } @@ -706,27 +719,27 @@ static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register selector_t *f if ( pri == INTERNAL_NOPRI ) { if ( ignorepri ) - f->f_filterData.f_pmask[i >> 3] = TABLE_ALLPRI; + pRule->f_filterData.f_pmask[i >> 3] = TABLE_ALLPRI; else - f->f_filterData.f_pmask[i >> 3] = TABLE_NOPRI; + pRule->f_filterData.f_pmask[i >> 3] = TABLE_NOPRI; } else if ( singlpri ) { if ( ignorepri ) - f->f_filterData.f_pmask[i >> 3] &= ~(1<<pri); + pRule->f_filterData.f_pmask[i >> 3] &= ~(1<<pri); else - f->f_filterData.f_pmask[i >> 3] |= (1<<pri); + pRule->f_filterData.f_pmask[i >> 3] |= (1<<pri); } else { if ( pri == TABLE_ALLPRI ) { if ( ignorepri ) - f->f_filterData.f_pmask[i >> 3] = TABLE_NOPRI; + pRule->f_filterData.f_pmask[i >> 3] = TABLE_NOPRI; else - f->f_filterData.f_pmask[i >> 3] = TABLE_ALLPRI; + pRule->f_filterData.f_pmask[i >> 3] = TABLE_ALLPRI; } else { if ( ignorepri ) for (i2= 0; i2 <= pri; ++i2) - f->f_filterData.f_pmask[i >> 3] &= ~(1<<i2); + pRule->f_filterData.f_pmask[i >> 3] &= ~(1<<i2); else for (i2= 0; i2 <= pri; ++i2) - f->f_filterData.f_pmask[i >> 3] |= (1<<i2); + pRule->f_filterData.f_pmask[i >> 3] |= (1<<i2); } } } @@ -752,7 +765,7 @@ static rsRetVal cflineProcessTradPRIFilter(uchar **pline, register selector_t *f * A pointer to that beginnig is passed back to the caller. * rgerhards 2008-01-19 */ -static rsRetVal cflineProcessIfFilter(uchar **pline, register selector_t *f) +static rsRetVal cflineProcessIfFilter(uchar **pline, register rule_t *f) { DEFiRet; ctok_t *tok; @@ -765,7 +778,6 @@ static rsRetVal cflineProcessIfFilter(uchar **pline, register selector_t *f) dbgprintf(" - general expression-based filter\n"); errno = 0; /* keep strerror_r() stuff out of logerror messages */ -dbgprintf("calling expression parser, pp %p ('%s')\n", *pline, *pline); f->f_filter_type = FILTER_EXPR; /* if we come to over here, pline starts with "if ". We just skip that part. */ @@ -821,7 +833,7 @@ finalize_it: * of the action part. A pointer to that beginnig is passed back to the caller. * rgerhards 2005-09-15 */ -static rsRetVal cflineProcessPropFilter(uchar **pline, register selector_t *f) +static rsRetVal cflineProcessPropFilter(uchar **pline, register rule_t *f) { rsParsObj *pPars; cstr_t *pCSCompOp; @@ -1010,10 +1022,10 @@ finalize_it: /* read the filter part of a configuration line and store the filter - * in the supplied selector_t + * in the supplied rule_t * rgerhards, 2007-08-01 */ -static rsRetVal cflineDoFilter(uchar **pp, selector_t *f) +static rsRetVal cflineDoFilter(uchar **pp, rule_t *f) { DEFiRet; @@ -1106,17 +1118,15 @@ static rsRetVal cflineDoAction(uchar **p, action_t **ppAction) /* Process a configuration file line in traditional "filter selector" format - * or one that builds upon this format. + * or one that builds upon this format. Note that ppRule may be a NULL pointer, + * which is valid and happens if there is no previous line (right at the start + * of the master config file!). */ -static rsRetVal cflineClassic(uchar *p, selector_t **pfCurr) +static rsRetVal +cflineClassic(uchar *p, rule_t **ppRule) { DEFiRet; action_t *pAction; - selector_t *fCurr; - - ASSERT(pfCurr != NULL); - - fCurr = *pfCurr; /* lines starting with '&' have no new filters and just add * new actions to the currently processed selector. @@ -1134,16 +1144,19 @@ static rsRetVal cflineClassic(uchar *p, selector_t **pfCurr) * selector is NULL, which means we do not need to care about it at * all. -- rgerhards, 2007-08-01 */ - CHKiRet(selectorAddList(fCurr)); - CHKiRet(selectorConstruct(&fCurr)); /* create "fresh" selector */ - CHKiRet(cflineDoFilter(&p, fCurr)); /* pull filters */ + if(*ppRule != NULL) { + CHKiRet(ruleset.AddRule(rule.GetAssRuleset(*ppRule), ppRule)); + } + CHKiRet(rule.Construct(ppRule)); /* create "fresh" selector */ + CHKiRet(rule.SetAssRuleset(*ppRule, ruleset.GetCurrent())); /* create "fresh" selector */ + CHKiRet(rule.ConstructFinalize(*ppRule)); /* create "fresh" selector */ + CHKiRet(cflineDoFilter(&p, *ppRule)); /* pull filters */ } CHKiRet(cflineDoAction(&p, &pAction)); - CHKiRet(llAppend(&fCurr->llActList, NULL, (void*) pAction)); + CHKiRet(llAppend(&(*ppRule)->llActList, NULL, (void*) pAction)); finalize_it: - *pfCurr = fCurr; RETiRet; } @@ -1153,7 +1166,7 @@ finalize_it: * rgerhards, 2007-08-01 */ static rsRetVal -cfline(uchar *line, selector_t **pfCurr) +cfline(uchar *line, rule_t **pfCurr) { DEFiRet; @@ -1250,6 +1263,8 @@ CODESTARTObjClassExit(conf) objRelease(module, CORE_COMPONENT); objRelease(errmsg, CORE_COMPONENT); objRelease(net, LM_NET_FILENAME); + objRelease(rule, CORE_COMPONENT); + objRelease(ruleset, CORE_COMPONENT); ENDObjClassExit(conf) @@ -1265,6 +1280,8 @@ BEGINAbstractObjClassInit(conf, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANG CHKiRet(objUse(module, CORE_COMPONENT)); CHKiRet(objUse(errmsg, CORE_COMPONENT)); CHKiRet(objUse(net, LM_NET_FILENAME)); /* TODO: make this dependcy go away! */ + CHKiRet(objUse(rule, CORE_COMPONENT)); + CHKiRet(objUse(ruleset, CORE_COMPONENT)); ENDObjClassInit(conf) /* vi:set ai: diff --git a/runtime/conf.h b/runtime/conf.h index 2494d4dc..25b887be 100644 --- a/runtime/conf.h +++ b/runtime/conf.h @@ -35,7 +35,7 @@ BEGINinterface(conf) /* name must also be changed in ENDinterface macro! */ rsRetVal (*cfsysline)(uchar *p); rsRetVal (*doModLoad)(uchar **pp, __attribute__((unused)) void* pVal); rsRetVal (*doIncludeLine)(uchar **pp, __attribute__((unused)) void* pVal); - rsRetVal (*cfline)(uchar *line, selector_t **pfCurr); + rsRetVal (*cfline)(uchar *line, rule_t **pfCurr); rsRetVal (*processConfFile)(uchar *pConfFile); rsRetVal (*ReInitConf)(void); rsRetVal (*GetNbrActActions)(int *); @@ -51,5 +51,9 @@ PROTOTYPEObj(conf); extern EHostnameCmpMode eDfltHostnameCmpMode; extern cstr_t *pDfltHostnameCmp; extern cstr_t *pDfltProgNameCmp; +/* TODO: the following 2 need to go in conf obj interface... */ +rsRetVal cflineParseTemplateName(uchar** pp, omodStringRequest_t *pOMSR, int iEntry, int iTplOpts, uchar *dfltTplName); +rsRetVal cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pOMSR, int iEntry, int iTplOpts, uchar *pszTpl); + #endif /* #ifndef INCLUDED_CONF_H */ diff --git a/runtime/linkedlist.c b/runtime/linkedlist.c index 8f842e43..cc095f6e 100644 --- a/runtime/linkedlist.c +++ b/runtime/linkedlist.c @@ -398,7 +398,7 @@ rsRetVal llExecFunc(linkedList_t *pThis, rsRetVal (*pFunc)(void*, void*), void* */ llCookie = llCookiePrev; } else if (iRet != RS_RET_OK) { - goto finalize_it; + FINALIZE; } llCookiePrev = llCookie; } diff --git a/runtime/msg.c b/runtime/msg.c index 8122383a..65041a31 100644 --- a/runtime/msg.c +++ b/runtime/msg.c @@ -46,6 +46,7 @@ #include "regexp.h" #include "atomic.h" #include "unicode-helper.h" +#include "ruleset.h" /* static data */ DEFobjStaticHelpers @@ -1165,6 +1166,15 @@ void MsgAssignTAG(msg_t *pMsg, uchar *pBuf) } +/* rgerhards 2009-06-12: set associated ruleset + */ +void MsgSetRuleset(msg_t *pMsg, ruleset_t *pRuleset) +{ + assert(pMsg != NULL); + pMsg->pRuleset = pRuleset; +} + + /* rgerhards 2004-11-16: set TAG in msg object */ void MsgSetTAG(msg_t *pMsg, char* pszTAG) diff --git a/runtime/msg.h b/runtime/msg.h index fe9f87fa..74ff9e60 100644 --- a/runtime/msg.h +++ b/runtime/msg.h @@ -121,8 +121,8 @@ struct msg { char *pszTIMESTAMP_PgSQL;/* TIMESTAMP as PgSQL formatted string (always 21 characters) */ char *pszTIMESTAMP_SecFrac;/* TIMESTAMP fractional seconds (always 6 characters) */ int msgFlags; /* flags associated with this message */ + ruleset_t *pRuleset; /* ruleset to be used for processing this message */ /* now follow fixed-size buffers to safe some time otherwise used for allocs */ - }; @@ -152,6 +152,7 @@ rsRetVal MsgSetPROCID(msg_t *pMsg, char* pszPROCID); rsRetVal MsgSetMSGID(msg_t *pMsg, char* pszMSGID); void MsgAssignTAG(msg_t *pMsg, uchar *pBuf); void MsgSetTAG(msg_t *pMsg, char* pszTAG); +void MsgSetRuleset(msg_t *pMsg, ruleset_t*); rsRetVal MsgSetFlowControlType(msg_t *pMsg, flowControl_t eFlowCtl); rsRetVal MsgSetStructuredData(msg_t *pMsg, char* pszStrucData); void MsgSetRcvFrom(msg_t *pMsg, uchar* pszRcvFrom); diff --git a/runtime/obj-types.h b/runtime/obj-types.h index 78829f94..6c1381ac 100644 --- a/runtime/obj-types.h +++ b/runtime/obj-types.h @@ -105,12 +105,13 @@ struct obj_s { /* the dummy struct that each derived class can be casted to */ # define ISOBJ_TYPE_assert(pObj, objType) \ do { \ ASSERT(pObj != NULL); \ - ASSERT((unsigned) ((obj_t*) (pObj))->iObjCooCKiE == (unsigned) 0xBADEFEE); \ if(strcmp((char*)(((obj_t*)pObj)->pObjInfo->pszID), #objType)) { \ dbgprintf("%s:%d ISOBJ assert failure: invalid object type, expected '%s' " \ - "actual '%s'\n", __FILE__, __LINE__, #objType, (((obj_t*)pObj)->pObjInfo->pszID)); \ + "actual '%s', cookie: %X\n", __FILE__, __LINE__, #objType, \ + (((obj_t*)pObj)->pObjInfo->pszID), ((obj_t*)(pObj))->iObjCooCKiE); \ assert(0); /* trigger assertion, messge we already have */ \ } \ + ASSERT((unsigned) ((obj_t*)(pObj))->iObjCooCKiE == (unsigned) 0xBADEFEE); \ } while(0) #else /* non-debug mode, no checks but much faster */ # define BEGINobjInstance obj_t objData diff --git a/runtime/obj.c b/runtime/obj.c index 355c0f97..f2cb447e 100644 --- a/runtime/obj.c +++ b/runtime/obj.c @@ -87,12 +87,15 @@ #include "modules.h" #include "errmsg.h" #include "cfsysline.h" +#include "unicode-helper.h" +#include "apc.h" /* static data */ DEFobjCurrIf(obj) /* we define our own interface, as this is expected by some macros! */ DEFobjCurrIf(var) DEFobjCurrIf(module) DEFobjCurrIf(errmsg) +DEFobjCurrIf(strm) static objInfo_t *arrObjInfo[OBJ_NUM_IDS]; /* array with object information pointers */ @@ -144,8 +147,8 @@ InfoConstruct(objInfo_t **ppThis, uchar *pszID, int iObjVers, ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); pThis->pszID = pszID; - pThis->lenID = strlen((char*)pszID); - pThis->pszName = (uchar*)strdup((char*)pszID); /* it's OK if we have NULL ptr, GetName() will deal with that! */ + pThis->lenID = ustrlen(pszID); + pThis->pszName = ustrdup(pszID); /* it's OK if we have NULL ptr, GetName() will deal with that! */ pThis->iObjVers = iObjVers; pThis->QueryIF = pQueryIF; pThis->pModInfo = pModInfo; @@ -176,8 +179,7 @@ InfoDestruct(objInfo_t **ppThis) pThis = *ppThis; assert(pThis != NULL); - if(pThis->pszName != NULL) - free(pThis->pszName); + free(pThis->pszName); free(pThis); *ppThis = NULL; @@ -205,9 +207,7 @@ DestructObjSelf(obj_t *pThis) DEFiRet; ISOBJ_assert(pThis); - if(pThis->pszName != NULL) { - free(pThis->pszName); - } + free(pThis->pszName); RETiRet; } @@ -228,20 +228,20 @@ static rsRetVal objSerializeHeader(strm_t *pStrm, obj_t *pObj, uchar *pszRecType assert(!strcmp((char*) pszRecType, "Obj") || !strcmp((char*) pszRecType, "OPB")); /* object cookie and serializer version (so far always 1) */ - CHKiRet(strmWriteChar(pStrm, COOKIE_OBJLINE)); - CHKiRet(strmWrite(pStrm, (uchar*) pszRecType, 3)); /* record types are always 3 octets */ - CHKiRet(strmWriteChar(pStrm, ':')); - CHKiRet(strmWriteChar(pStrm, '1')); + CHKiRet(strm.WriteChar(pStrm, COOKIE_OBJLINE)); + CHKiRet(strm.Write(pStrm, (uchar*) pszRecType, 3)); /* record types are always 3 octets */ + CHKiRet(strm.WriteChar(pStrm, ':')); + CHKiRet(strm.WriteChar(pStrm, '1')); /* object type, version and string length */ - CHKiRet(strmWriteChar(pStrm, ':')); - CHKiRet(strmWrite(pStrm, pObj->pObjInfo->pszID, pObj->pObjInfo->lenID)); - CHKiRet(strmWriteChar(pStrm, ':')); - CHKiRet(strmWriteLong(pStrm, objGetVersion(pObj))); + CHKiRet(strm.WriteChar(pStrm, ':')); + CHKiRet(strm.Write(pStrm, pObj->pObjInfo->pszID, pObj->pObjInfo->lenID)); + CHKiRet(strm.WriteChar(pStrm, ':')); + CHKiRet(strm.WriteLong(pStrm, objGetVersion(pObj))); /* record trailer */ - CHKiRet(strmWriteChar(pStrm, ':')); - CHKiRet(strmWriteChar(pStrm, '\n')); + CHKiRet(strm.WriteChar(pStrm, ':')); + CHKiRet(strm.WriteChar(pStrm, '\n')); finalize_it: RETiRet; @@ -259,7 +259,7 @@ BeginSerialize(strm_t *pStrm, obj_t *pObj) ISOBJ_TYPE_assert(pStrm, strm); ISOBJ_assert(pObj); - CHKiRet(strmRecordBegin(pStrm)); + CHKiRet(strm.RecordBegin(pStrm)); CHKiRet(objSerializeHeader(pStrm, pObj, (uchar*) "Obj")); finalize_it: @@ -284,7 +284,7 @@ BeginSerializePropBag(strm_t *pStrm, obj_t *pObj) ISOBJ_TYPE_assert(pStrm, strm); ISOBJ_assert(pObj); - CHKiRet(strmRecordBegin(pStrm)); + CHKiRet(strm.RecordBegin(pStrm)); CHKiRet(objSerializeHeader(pStrm, pObj, (uchar*) "OPB")); finalize_it: @@ -320,31 +320,31 @@ SerializeProp(strm_t *pStrm, uchar *pszPropName, propType_t propType, void *pUsr switch(propType) { case PROPTYPE_PSZ: pszBuf = (uchar*) pUsr; - lenBuf = strlen((char*) pszBuf); + lenBuf = ustrlen(pszBuf); vType = VARTYPE_STR; break; case PROPTYPE_SHORT: CHKiRet(srUtilItoA((char*) szBuf, sizeof(szBuf), (long) *((short*) pUsr))); pszBuf = szBuf; - lenBuf = strlen((char*) szBuf); + lenBuf = ustrlen(szBuf); vType = VARTYPE_NUMBER; break; case PROPTYPE_INT: CHKiRet(srUtilItoA((char*) szBuf, sizeof(szBuf), (long) *((int*) pUsr))); pszBuf = szBuf; - lenBuf = strlen((char*) szBuf); + lenBuf = ustrlen(szBuf); vType = VARTYPE_NUMBER; break; case PROPTYPE_LONG: CHKiRet(srUtilItoA((char*) szBuf, sizeof(szBuf), *((long*) pUsr))); pszBuf = szBuf; - lenBuf = strlen((char*) szBuf); + lenBuf = ustrlen(szBuf); vType = VARTYPE_NUMBER; break; case PROPTYPE_INT64: CHKiRet(srUtilItoA((char*) szBuf, sizeof(szBuf), *((int64*) pUsr))); pszBuf = szBuf; - lenBuf = strlen((char*) szBuf); + lenBuf = ustrlen(szBuf); vType = VARTYPE_NUMBER; break; case PROPTYPE_CSTR: @@ -377,23 +377,23 @@ SerializeProp(strm_t *pStrm, uchar *pszPropName, propType_t propType, void *pUsr } /* cookie */ - CHKiRet(strmWriteChar(pStrm, COOKIE_PROPLINE)); + CHKiRet(strm.WriteChar(pStrm, COOKIE_PROPLINE)); /* name */ - CHKiRet(strmWrite(pStrm, pszPropName, strlen((char*)pszPropName))); - CHKiRet(strmWriteChar(pStrm, ':')); + CHKiRet(strm.Write(pStrm, pszPropName, ustrlen(pszPropName))); + CHKiRet(strm.WriteChar(pStrm, ':')); /* type */ - CHKiRet(strmWriteLong(pStrm, (int) vType)); - CHKiRet(strmWriteChar(pStrm, ':')); + CHKiRet(strm.WriteLong(pStrm, (int) vType)); + CHKiRet(strm.WriteChar(pStrm, ':')); /* length */ - CHKiRet(strmWriteLong(pStrm, lenBuf)); - CHKiRet(strmWriteChar(pStrm, ':')); + CHKiRet(strm.WriteLong(pStrm, lenBuf)); + CHKiRet(strm.WriteChar(pStrm, ':')); /* data */ - CHKiRet(strmWrite(pStrm, (uchar*) pszBuf, lenBuf)); + CHKiRet(strm.Write(pStrm, (uchar*) pszBuf, lenBuf)); /* trailer */ - CHKiRet(strmWriteChar(pStrm, ':')); - CHKiRet(strmWriteChar(pStrm, '\n')); + CHKiRet(strm.WriteChar(pStrm, ':')); + CHKiRet(strm.WriteChar(pStrm, '\n')); finalize_it: RETiRet; @@ -410,12 +410,12 @@ EndSerialize(strm_t *pStrm) assert(pStrm != NULL); - CHKiRet(strmWriteChar(pStrm, COOKIE_ENDLINE)); - CHKiRet(strmWrite(pStrm, (uchar*) "End\n", sizeof("END\n") - 1)); - CHKiRet(strmWriteChar(pStrm, COOKIE_BLANKLINE)); - CHKiRet(strmWriteChar(pStrm, '\n')); + CHKiRet(strm.WriteChar(pStrm, COOKIE_ENDLINE)); + CHKiRet(strm.Write(pStrm, (uchar*) "End\n", sizeof("END\n") - 1)); + CHKiRet(strm.WriteChar(pStrm, COOKIE_BLANKLINE)); + CHKiRet(strm.WriteChar(pStrm, '\n')); - CHKiRet(strmRecordEnd(pStrm)); + CHKiRet(strm.RecordEnd(pStrm)); finalize_it: RETiRet; @@ -423,7 +423,7 @@ finalize_it: /* define a helper to make code below a bit cleaner (and quicker to write) */ -#define NEXTC CHKiRet(strmReadChar(pStrm, &c))/*;dbgprintf("c: %c\n", c)*/ +#define NEXTC CHKiRet(strm.ReadChar(pStrm, &c))/*;dbgprintf("c: %c\n", c)*/ /* de-serialize an embedded, non-octect-counted string. This is useful @@ -617,7 +617,7 @@ static rsRetVal objDeserializeProperty(var_t *pProp, strm_t *pStrm) NEXTC; if(c != COOKIE_PROPLINE) { /* oops, we've read one char that does not belong to use - unget it first */ - CHKiRet(strmUnreadChar(pStrm, c)); + CHKiRet(strm.UnreadChar(pStrm, c)); ABORT_FINALIZE(RS_RET_NO_PROPLINE); } @@ -718,7 +718,7 @@ static rsRetVal objDeserializeTryRecover(strm_t *pStrm) } } - CHKiRet(strmUnreadChar(pStrm, c)); + CHKiRet(strm.UnreadChar(pStrm, c)); finalize_it: dbgprintf("deserializer has possibly been able to re-sync and recover, state %d\n", iRet); @@ -803,7 +803,7 @@ Deserialize(void *ppObj, uchar *pszTypeExpected, strm_t *pStrm, rsRetVal (*fFixu } } while(iRetLocal != RS_RET_OK); - if(rsCStrSzStrCmp(pstrID, pszTypeExpected, strlen((char*)pszTypeExpected))) /* TODO: optimize strlen() - caller shall provide */ + if(rsCStrSzStrCmp(pstrID, pszTypeExpected, ustrlen(pszTypeExpected))) /* TODO: optimize strlen() - caller shall provide */ ABORT_FINALIZE(RS_RET_INVALID_OID); CHKiRet(FindObjInfo(pstrID, &pObjInfo)); @@ -948,13 +948,8 @@ SetName(obj_t *pThis, uchar *pszName) { DEFiRet; - if(pThis->pszName != NULL) - free(pThis->pszName); - - pThis->pszName = (uchar*) strdup((char*) pszName); - - if(pThis->pszName == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + free(pThis->pszName); + CHKmalloc(pThis->pszName = ustrdup(pszName)); finalize_it: RETiRet; @@ -1057,7 +1052,7 @@ RegisterObj(uchar *pszObjName, objInfo_t *pInfo) i = 0; while(!bFound && i < OBJ_NUM_IDS && arrObjInfo[i] != NULL) { if( arrObjInfo[i] != NULL - && !strcmp((char*)arrObjInfo[i]->pszID, (char*)pszObjName)) { + && !ustrcmp(arrObjInfo[i]->pszID, pszObjName)) { bFound = 1; break; } @@ -1096,7 +1091,7 @@ UnregisterObj(uchar *pszObjName) i = 0; while(!bFound && i < OBJ_NUM_IDS) { if( arrObjInfo[i] != NULL - && !strcmp((char*)arrObjInfo[i]->pszID, (char*)pszObjName)) { + && !ustrcmp(arrObjInfo[i]->pszID, pszObjName)) { bFound = 1; break; } @@ -1278,14 +1273,15 @@ objClassExit(void) { DEFiRet; /* release objects we no longer need */ + objRelease(strm, CORE_COMPONENT); objRelease(var, CORE_COMPONENT); objRelease(module, CORE_COMPONENT); objRelease(errmsg, CORE_COMPONENT); /* TODO: implement the class exits! */ #if 0 - cfsyslineInit(pModInfo); - varClassInit(pModInfo); + cfsyslineExit(pModInfo); + varClassExit(pModInfo); #endif errmsgClassExit(); moduleClassExit(); @@ -1318,13 +1314,16 @@ objClassInit(modInfo_t *pModInfo) CHKiRet(objGetObjInterface(&obj)); /* get ourselves ;) */ /* init classes we use (limit to as few as possible!) */ + CHKiRet(apcClassInit(pModInfo)); CHKiRet(errmsgClassInit(pModInfo)); CHKiRet(cfsyslineInit()); CHKiRet(varClassInit(pModInfo)); CHKiRet(moduleClassInit(pModInfo)); + CHKiRet(strmClassInit(pModInfo)); CHKiRet(objUse(var, CORE_COMPONENT)); CHKiRet(objUse(module, CORE_COMPONENT)); CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(strm, CORE_COMPONENT)); finalize_it: RETiRet; diff --git a/runtime/obj.h b/runtime/obj.h index dc04203b..3973a16e 100644 --- a/runtime/obj.h +++ b/runtime/obj.h @@ -68,7 +68,7 @@ #define objSerializePTR(strm, propName, propType) \ CHKiRet(obj.SerializeProp(strm, (uchar*) #propName, PROPTYPE_##propType, (void*) pThis->propName)); #define DEFobjStaticHelpers \ - static objInfo_t *pObjInfoOBJ = NULL; \ + static objInfo_t __attribute__((unused)) *pObjInfoOBJ = NULL; \ DEFobjCurrIf(obj) diff --git a/runtime/queue.c b/runtime/queue.c index 4e017e84..aa8e6c21 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -59,6 +59,7 @@ /* static data */ DEFobjStaticHelpers DEFobjCurrIf(glbl) +DEFobjCurrIf(strm) /* forward-definitions */ rsRetVal qqueueChkPersist(qqueue_t *pThis); @@ -293,6 +294,7 @@ qqueueStartDA(qqueue_t *pThis) CHKiRet(qqueueSetMaxFileSize(pThis->pqDA, pThis->iMaxFileSize)); CHKiRet(qqueueSetFilePrefix(pThis->pqDA, pThis->pszFilePrefix, pThis->lenFilePrefix)); CHKiRet(qqueueSetiPersistUpdCnt(pThis->pqDA, pThis->iPersistUpdCnt)); + CHKiRet(qqueueSetbSyncQueueFiles(pThis->pqDA, pThis->bSyncQueueFiles)); CHKiRet(qqueueSettoActShutdown(pThis->pqDA, pThis->toActShutdown)); CHKiRet(qqueueSettoEnq(pThis->pqDA, pThis->toEnq)); CHKiRet(qqueueSetEnqOnly(pThis->pqDA, pThis->bDAEnqOnly, MUTEX_ALREADY_LOCKED)); @@ -667,7 +669,7 @@ qqueueLoadPersStrmInfoFixup(strm_t *pStrm, qqueue_t __attribute__((unused)) *pTh DEFiRet; ISOBJ_TYPE_assert(pStrm, strm); ISOBJ_TYPE_assert(pThis, qqueue); - CHKiRet(strmSetDir(pStrm, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetDir(pStrm, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); finalize_it: RETiRet; } @@ -744,11 +746,11 @@ qqueueTryLoadPersistedInfo(qqueue_t *pThis) /* If we reach this point, we have a .qi file */ - CHKiRet(strmConstruct(&psQIF)); - CHKiRet(strmSettOperationsMode(psQIF, STREAMMODE_READ)); - CHKiRet(strmSetsType(psQIF, STREAMTYPE_FILE_SINGLE)); - CHKiRet(strmSetFName(psQIF, pszQIFNam, lenQIFNam)); - CHKiRet(strmConstructFinalize(psQIF)); + CHKiRet(strm.Construct(&psQIF)); + CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_READ)); + CHKiRet(strm.SetsType(psQIF, STREAMTYPE_FILE_SINGLE)); + CHKiRet(strm.SetFName(psQIF, pszQIFNam, lenQIFNam)); + CHKiRet(strm.ConstructFinalize(psQIF)); /* first, we try to read the property bag for ourselfs */ CHKiRet(obj.DeserializePropBag((obj_t*) pThis, psQIF)); @@ -770,8 +772,8 @@ qqueueTryLoadPersistedInfo(qqueue_t *pThis) CHKiRet(obj.Deserialize(&pThis->tVars.disk.pRead, (uchar*) "strm", psQIF, (rsRetVal(*)(obj_t*,void*))qqueueLoadPersStrmInfoFixup, pThis)); - CHKiRet(strmSeekCurrOffs(pThis->tVars.disk.pWrite)); - CHKiRet(strmSeekCurrOffs(pThis->tVars.disk.pRead)); + CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pWrite)); + CHKiRet(strm.SeekCurrOffs(pThis->tVars.disk.pRead)); /* OK, we could successfully read the file, so we now can request that it be * deleted when we are done with the persisted information. @@ -780,7 +782,7 @@ qqueueTryLoadPersistedInfo(qqueue_t *pThis) finalize_it: if(psQIF != NULL) - strmDestruct(&psQIF); + strm.Destruct(&psQIF); if(iRet != RS_RET_OK) { dbgoprint((obj_t*) pThis, "error %d reading .qi file - can not read persisted info (if any)\n", @@ -815,24 +817,26 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) if(bRestarted == 1) { ; } else { - CHKiRet(strmConstruct(&pThis->tVars.disk.pWrite)); - CHKiRet(strmSetDir(pThis->tVars.disk.pWrite, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); - CHKiRet(strmSetiMaxFiles(pThis->tVars.disk.pWrite, 10000000)); - CHKiRet(strmSettOperationsMode(pThis->tVars.disk.pWrite, STREAMMODE_WRITE)); - CHKiRet(strmSetsType(pThis->tVars.disk.pWrite, STREAMTYPE_FILE_CIRCULAR)); - CHKiRet(strmConstructFinalize(pThis->tVars.disk.pWrite)); + CHKiRet(strm.Construct(&pThis->tVars.disk.pWrite)); + CHKiRet(strm.SetbSync(pThis->tVars.disk.pWrite, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetDir(pThis->tVars.disk.pWrite, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + 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)); + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pWrite)); - CHKiRet(strmConstruct(&pThis->tVars.disk.pRead)); - CHKiRet(strmSetbDeleteOnClose(pThis->tVars.disk.pRead, 1)); - CHKiRet(strmSetDir(pThis->tVars.disk.pRead, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); - CHKiRet(strmSetiMaxFiles(pThis->tVars.disk.pRead, 10000000)); - CHKiRet(strmSettOperationsMode(pThis->tVars.disk.pRead, STREAMMODE_READ)); - CHKiRet(strmSetsType(pThis->tVars.disk.pRead, STREAMTYPE_FILE_CIRCULAR)); - CHKiRet(strmConstructFinalize(pThis->tVars.disk.pRead)); + CHKiRet(strm.Construct(&pThis->tVars.disk.pRead)); + CHKiRet(strm.SetbSync(pThis->tVars.disk.pRead, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 1)); + CHKiRet(strm.SetDir(pThis->tVars.disk.pRead, glbl.GetWorkDir(), strlen((char*)glbl.GetWorkDir()))); + CHKiRet(strm.SetiMaxFiles(pThis->tVars.disk.pRead, 10000000)); + CHKiRet(strm.SettOperationsMode(pThis->tVars.disk.pRead, STREAMMODE_READ)); + CHKiRet(strm.SetsType(pThis->tVars.disk.pRead, STREAMTYPE_FILE_CIRCULAR)); + CHKiRet(strm.ConstructFinalize(pThis->tVars.disk.pRead)); - CHKiRet(strmSetFName(pThis->tVars.disk.pWrite, pThis->pszFilePrefix, pThis->lenFilePrefix)); - CHKiRet(strmSetFName(pThis->tVars.disk.pRead, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(strm.SetFName(pThis->tVars.disk.pWrite, pThis->pszFilePrefix, pThis->lenFilePrefix)); + CHKiRet(strm.SetFName(pThis->tVars.disk.pRead, pThis->pszFilePrefix, pThis->lenFilePrefix)); } /* now we set (and overwrite in case of a persisted restart) some parameters which @@ -840,8 +844,8 @@ static rsRetVal qConstructDisk(qqueue_t *pThis) * for example file name generation must not be changed as that would break the * ability to read existing queue files. -- rgerhards, 2008-01-12 */ - CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pWrite, pThis->iMaxFileSize)); - CHKiRet(strmSetiMaxFileSize(pThis->tVars.disk.pRead, pThis->iMaxFileSize)); + CHKiRet(strm.SetiMaxFileSize(pThis->tVars.disk.pWrite, pThis->iMaxFileSize)); + CHKiRet(strm.SetiMaxFileSize(pThis->tVars.disk.pRead, pThis->iMaxFileSize)); finalize_it: RETiRet; @@ -854,8 +858,8 @@ static rsRetVal qDestructDisk(qqueue_t *pThis) ASSERT(pThis != NULL); - strmDestruct(&pThis->tVars.disk.pWrite); - strmDestruct(&pThis->tVars.disk.pRead); + strm.Destruct(&pThis->tVars.disk.pWrite); + strm.Destruct(&pThis->tVars.disk.pRead); RETiRet; } @@ -867,10 +871,10 @@ static rsRetVal qAddDisk(qqueue_t *pThis, void* pUsr) ASSERT(pThis != NULL); - CHKiRet(strmSetWCntr(pThis->tVars.disk.pWrite, &nWriteCount)); + CHKiRet(strm.SetWCntr(pThis->tVars.disk.pWrite, &nWriteCount)); CHKiRet((objSerialize(pUsr))(pUsr, pThis->tVars.disk.pWrite)); - CHKiRet(strmFlush(pThis->tVars.disk.pWrite)); - CHKiRet(strmSetWCntr(pThis->tVars.disk.pWrite, NULL)); /* no more counting for now... */ + CHKiRet(strm.Flush(pThis->tVars.disk.pWrite)); + CHKiRet(strm.SetWCntr(pThis->tVars.disk.pWrite, NULL)); /* no more counting for now... */ pThis->tVars.disk.sizeOnDisk += nWriteCount; @@ -894,9 +898,9 @@ static rsRetVal qDelDisk(qqueue_t *pThis, void **ppUsr) int64 offsIn; int64 offsOut; - CHKiRet(strmGetCurrOffset(pThis->tVars.disk.pRead, &offsIn)); + CHKiRet(strm.GetCurrOffset(pThis->tVars.disk.pRead, &offsIn)); CHKiRet(obj.Deserialize(ppUsr, (uchar*) "msg", pThis->tVars.disk.pRead, NULL, NULL)); - CHKiRet(strmGetCurrOffset(pThis->tVars.disk.pRead, &offsOut)); + CHKiRet(strm.GetCurrOffset(pThis->tVars.disk.pRead, &offsOut)); /* This time it is a bit tricky: we free disk space only upon file deletion. So we need * to keep track of what we have read until we get an out-offset that is lower than the @@ -1917,16 +1921,16 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint) pThis->bNeedDelQIF = 0; } /* indicate spool file needs to be deleted */ - CHKiRet(strmSetbDeleteOnClose(pThis->tVars.disk.pRead, 1)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 1)); FINALIZE; /* nothing left to do, so be happy */ } - CHKiRet(strmConstruct(&psQIF)); - CHKiRet(strmSettOperationsMode(psQIF, STREAMMODE_WRITE)); - CHKiRet(strmSetiAddtlOpenFlags(psQIF, O_TRUNC)); - CHKiRet(strmSetsType(psQIF, STREAMTYPE_FILE_SINGLE)); - CHKiRet(strmSetFName(psQIF, pszQIFNam, lenQIFNam)); - CHKiRet(strmConstructFinalize(psQIF)); + CHKiRet(strm.Construct(&psQIF)); + CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_WRITE_TRUNC)); + CHKiRet(strm.SetbSync(psQIF, pThis->bSyncQueueFiles)); + CHKiRet(strm.SetsType(psQIF, STREAMTYPE_FILE_SINGLE)); + CHKiRet(strm.SetFName(psQIF, pszQIFNam, lenQIFNam)); + CHKiRet(strm.ConstructFinalize(psQIF)); /* first, write the property bag for ourselfs * And, surprisingly enough, we currently need to persist only the size of the @@ -1951,14 +1955,14 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint) } /* now persist the stream info */ - CHKiRet(strmSerialize(pThis->tVars.disk.pWrite, psQIF)); - CHKiRet(strmSerialize(pThis->tVars.disk.pRead, psQIF)); + CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF)); + CHKiRet(strm.Serialize(pThis->tVars.disk.pRead, psQIF)); /* tell the input file object that it must not delete the file on close if the queue * is non-empty - but only if we are not during a simple checkpoint */ if(bIsCheckpoint != QUEUE_CHECKPOINT) { - CHKiRet(strmSetbDeleteOnClose(pThis->tVars.disk.pRead, 0)); + CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 0)); } /* we have persisted the queue object. So whenever it comes to an empty queue, @@ -1968,7 +1972,7 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint) finalize_it: if(psQIF != NULL) - strmDestruct(&psQIF); + strm.Destruct(&psQIF); RETiRet; } @@ -2279,6 +2283,7 @@ finalize_it: /* some simple object access methods */ +DEFpropSetMeth(qqueue, bSyncQueueFiles, int) DEFpropSetMeth(qqueue, iPersistUpdCnt, int) DEFpropSetMeth(qqueue, iDeqtWinFromHr, int) DEFpropSetMeth(qqueue, iDeqtWinToHr, int) @@ -2340,6 +2345,7 @@ rsRetVal qqueueQueryInterface(void) { return RS_RET_NOT_IMPLEMENTED; } BEGINObjClassInit(qqueue, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ CHKiRet(objUse(glbl, CORE_COMPONENT)); + CHKiRet(objUse(strm, CORE_COMPONENT)); /* now set our own handlers */ OBJSetMethodHandler(objMethod_SETPROPERTY, qqueueSetProperty); diff --git a/runtime/queue.h b/runtime/queue.h index a267862d..07f134aa 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -73,6 +73,7 @@ typedef struct queue_s { void *pUsr; /* a global, user-supplied pointer. Is passed back to consumer. */ int iUpdsSincePersist;/* nbr of queue updates since the last persist call */ int iPersistUpdCnt; /* persits queue info after this nbr of updates - 0 -> persist only on shutdown */ + int bSyncQueueFiles;/* if working with files, sync them after each write? */ int iHighWtrMrk; /* high water mark for disk-assisted memory queues */ int iLowWtrMrk; /* low water mark for disk-assisted memory queues */ int iDiscardMrk; /* if the queue is above this mark, low-severity messages are discarded */ @@ -186,6 +187,7 @@ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThread int iMaxQueueSize, rsRetVal (*pConsumer)(void*,void*)); PROTOTYPEObjClassInit(qqueue); PROTOTYPEpropSetMeth(qqueue, iPersistUpdCnt, int); +PROTOTYPEpropSetMeth(qqueue, bSyncQueueFiles, int); PROTOTYPEpropSetMeth(qqueue, iDeqtWinFromHr, int); PROTOTYPEpropSetMeth(qqueue, iDeqtWinToHr, int); PROTOTYPEpropSetMeth(qqueue, toQShutdown, long); diff --git a/runtime/rsyslog.c b/runtime/rsyslog.c index 8df100a1..6f732f0e 100644 --- a/runtime/rsyslog.c +++ b/runtime/rsyslog.c @@ -77,6 +77,8 @@ #include "conf.h" #include "glbl.h" #include "errmsg.h" +#include "rule.h" +#include "ruleset.h" /* forward definitions */ static rsRetVal dfltErrLogger(int, uchar *errMsg); @@ -150,14 +152,10 @@ rsrtInit(char **ppErrObj, obj_if_t *pObjIF) CHKiRet(datetimeClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "msg"; CHKiRet(msgClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "str,"; - CHKiRet(strmClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "wti"; - CHKiRet(wtiClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "wtp"; - CHKiRet(wtpClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "queue"; - CHKiRet(qqueueClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "ctok_token"; + CHKiRet(ctok_tokenClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "ctok"; + CHKiRet(ctokClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "vmstk"; CHKiRet(vmstkClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "sysvar"; @@ -168,12 +166,18 @@ rsrtInit(char **ppErrObj, obj_if_t *pObjIF) CHKiRet(vmopClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "vmprg"; CHKiRet(vmprgClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "ctok_token"; - CHKiRet(ctok_tokenClassInit(NULL)); - if(ppErrObj != NULL) *ppErrObj = "ctok"; - CHKiRet(ctokClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "expr"; CHKiRet(exprClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "rule"; + CHKiRet(ruleClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "ruleset"; + CHKiRet(rulesetClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "wti"; + CHKiRet(wtiClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "wtp"; + CHKiRet(wtpClassInit(NULL)); + if(ppErrObj != NULL) *ppErrObj = "queue"; + CHKiRet(qqueueClassInit(NULL)); if(ppErrObj != NULL) *ppErrObj = "conf"; CHKiRet(confClassInit(NULL)); @@ -206,6 +210,8 @@ rsrtExit(void) /* do actual de-init only if we are the last runtime user */ confClassExit(); glblClassExit(); + rulesetClassExit(); + ruleClassExit(); objClassExit(); /* *THIS* *MUST/SHOULD?* always be the first class initilizer being called (except debug)! */ } diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h index 32177a9f..0fafd700 100644 --- a/runtime/rsyslog.h +++ b/runtime/rsyslog.h @@ -62,7 +62,9 @@ typedef unsigned char uchar;/* get rid of the unhandy "unsigned char" */ typedef struct thrdInfo thrdInfo_t; typedef struct obj_s obj_t; -typedef struct filed selector_t;/* TODO: this so far resides in syslogd.c, think about modularization */ +typedef struct ruleset_s ruleset_t; +typedef struct rule_s rule_t; +//typedef struct filed selector_t;/* TODO: this so far resides in syslogd.c, think about modularization */ typedef struct NetAddr netAddr_t; typedef struct netstrms_s netstrms_t; typedef struct netstrm_s netstrm_t; @@ -97,6 +99,7 @@ typedef struct strmLstnPortList_s strmLstnPortList_t; // TODO: rename? typedef long long int64; typedef long long unsigned uint64; typedef int64 number_t; /* type to use for numbers - TODO: maybe an autoconf option? */ +typedef short bool; #ifdef __hpux typedef unsigned int u_int32_t; /* TODO: is this correct? */ @@ -114,6 +117,20 @@ typedef enum { eFLOWCTL_FULL_DELAY = 2 /**< delay possible for extended period of time */ } flowControl_t; +/* filter operations */ +typedef enum { + FIOP_NOP = 0, /* do not use - No Operation */ + FIOP_CONTAINS = 1, /* contains string? */ + FIOP_ISEQUAL = 2, /* is (exactly) equal? */ + FIOP_STARTSWITH = 3, /* starts with a string? */ + FIOP_REGEX = 4, /* matches a (BRE) regular expression? */ + FIOP_EREREGEX = 5 /* matches a ERE regular expression? */ +} fiop_t; + + +#ifndef _PATH_CONSOLE +#define _PATH_CONSOLE "/dev/console" +#endif /* The error codes below are orginally "borrowed" from * liblogging. As such, we reserve values up to -2999 @@ -279,7 +296,11 @@ enum rsRetVal_ /** return value. All methods return this if not specified oth RS_RET_PREVIOUS_COMMITTED = -2122, /**< output plugin status: previous record was committed (an OK state!) */ RS_RET_ACTION_FAILED = -2123, /**< action failed and is now suspended (consider this permanent for the time being) */ RS_RET_NONFATAL_CONFIG_ERR = -2124, /**< non-fatal error during config processing */ + RS_RET_NON_SIZELIMITCMD = -2125, /**< size limit for file defined, but no size limit command given */ + RS_RET_SIZELIMITCMD_DIDNT_RESOLVE = -2126, /**< size limit command did not resolve situation */ + RS_RET_STREAM_DISABLED = -2127, /**< a file has been disabled (e.g. by size limit restriction) */ RS_RET_FILENAME_INVALID = -2140, /**< filename invalid, not found, no access, ... */ + RS_RET_ZLIB_ERR = -2141, /**< error during zlib call */ /* RainerScript error messages (range 1000.. 1999) */ RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */ @@ -368,6 +389,11 @@ typedef enum rsObjectID rsObjID; # define O_CLOEXEC 0 #endif +/* some constants */ +// TODO: do we really need them - if not, delete -- rgerhards, 2009-06-10 +#define IGNORE_ERROR_CODES 1 +#define ABORT_ON_ERROR 0 + /* The following prototype is convenient, even though it may not be the 100% correct place.. -- rgerhards 2008-01-07 */ void dbgprintf(char *, ...) __attribute__((format(printf, 1, 2))); diff --git a/runtime/rule.c b/runtime/rule.c new file mode 100644 index 00000000..f17c524e --- /dev/null +++ b/runtime/rule.c @@ -0,0 +1,450 @@ +/* rule.c - rsyslog's rule object + * + * See file comment in rule.c for the overall structure of rule processing. + * + * Module begun 2009-06-10 by Rainer Gerhards + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ + +#include "config.h" +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <ctype.h> + +#include "rsyslog.h" +#include "obj.h" +#include "action.h" +#include "rule.h" +#include "errmsg.h" +#include "vm.h" +#include "var.h" +#include "srUtils.h" +#include "dirty.h" /* for getFIOPName */ + +/* static data */ +DEFobjStaticHelpers +DEFobjCurrIf(errmsg) +DEFobjCurrIf(expr) +DEFobjCurrIf(var) +DEFobjCurrIf(vm) + +/* iterate over all actions, this is often needed, for example when HUP processing + * must be done or a shutdown is pending. + */ +static rsRetVal +iterateAllActions(rule_t *pThis, rsRetVal (*pFunc)(void*, void*), void* pParam) +{ + return llExecFunc(&pThis->llActList, pFunc, pParam); +} + + + +/* helper to processMsg(), used to call the configured actions. It is + * executed from within llExecFunc() of the action list. + * rgerhards, 2007-08-02 + */ +typedef struct processMsgDoActions_s { + int bPrevWasSuspended; /* was the previous action suspended? */ + msg_t *pMsg; +} processMsgDoActions_t; +DEFFUNC_llExecFunc(processMsgDoActions) +{ + DEFiRet; + rsRetVal iRetMod; /* return value of module - we do not always pass that back */ + action_t *pAction = (action_t*) pData; + processMsgDoActions_t *pDoActData = (processMsgDoActions_t*) pParam; + + assert(pAction != NULL); + + if((pAction->bExecWhenPrevSusp == 1) && (pDoActData->bPrevWasSuspended == 0)) { + dbgprintf("not calling action because the previous one is not suspended\n"); + ABORT_FINALIZE(RS_RET_OK); + } + + iRetMod = actionCallAction(pAction, pDoActData->pMsg); + if(iRetMod == RS_RET_DISCARDMSG) { + ABORT_FINALIZE(RS_RET_DISCARDMSG); + } else if(iRetMod == RS_RET_SUSPENDED) { + /* indicate suspension for next module to be called */ + pDoActData->bPrevWasSuspended = 1; + } else { + pDoActData->bPrevWasSuspended = 0; + } + +finalize_it: + RETiRet; +} + + +/* This functions looks at the given message and checks if it matches the + * provided filter condition. + */ +static rsRetVal +shouldProcessThisMessage(rule_t *pRule, msg_t *pMsg, int *bProcessMsg) +{ + DEFiRet; + unsigned short pbMustBeFreed; + char *pszPropVal; + int bRet = 0; + vm_t *pVM = NULL; + var_t *pResult = NULL; + + ISOBJ_TYPE_assert(pRule, rule); + assert(pMsg != NULL); + + /* we first have a look at the global, BSD-style block filters (for tag + * and host). Only if they match, we evaluate the actual filter. + * rgerhards, 2005-10-18 + */ + if(pRule->eHostnameCmpMode == HN_NO_COMP) { + /* EMPTY BY INTENSION - we check this value first, because + * it is the one most often used, so this saves us time! + */ + } else if(pRule->eHostnameCmpMode == HN_COMP_MATCH) { + if(rsCStrSzStrCmp(pRule->pCSHostnameComp, (uchar*) getHOSTNAME(pMsg), getHOSTNAMELen(pMsg))) { + /* not equal, so we are already done... */ + dbgprintf("hostname filter '+%s' does not match '%s'\n", + rsCStrGetSzStrNoNULL(pRule->pCSHostnameComp), getHOSTNAME(pMsg)); + FINALIZE; + } + } else { /* must be -hostname */ + if(!rsCStrSzStrCmp(pRule->pCSHostnameComp, (uchar*) getHOSTNAME(pMsg), getHOSTNAMELen(pMsg))) { + /* not equal, so we are already done... */ + dbgprintf("hostname filter '-%s' does not match '%s'\n", + rsCStrGetSzStrNoNULL(pRule->pCSHostnameComp), getHOSTNAME(pMsg)); + FINALIZE; + } + } + + if(pRule->pCSProgNameComp != NULL) { + int bInv = 0, bEqv = 0, offset = 0; + if(*(rsCStrGetSzStrNoNULL(pRule->pCSProgNameComp)) == '-') { + if(*(rsCStrGetSzStrNoNULL(pRule->pCSProgNameComp) + 1) == '-') + offset = 1; + else { + bInv = 1; + offset = 1; + } + } + if(!rsCStrOffsetSzStrCmp(pRule->pCSProgNameComp, offset, (uchar*) getProgramName(pMsg), getProgramNameLen(pMsg))) + bEqv = 1; + + if((!bEqv && !bInv) || (bEqv && bInv)) { + /* not equal or inverted selection, so we are already done... */ + dbgprintf("programname filter '%s' does not match '%s'\n", + rsCStrGetSzStrNoNULL(pRule->pCSProgNameComp), getProgramName(pMsg)); + FINALIZE; + } + } + + /* done with the BSD-style block filters */ + + if(pRule->f_filter_type == FILTER_PRI) { + /* skip messages that are incorrect priority */ + if ( (pRule->f_filterData.f_pmask[pMsg->iFacility] == TABLE_NOPRI) || \ + ((pRule->f_filterData.f_pmask[pMsg->iFacility] & (1<<pMsg->iSeverity)) == 0) ) + bRet = 0; + else + bRet = 1; + } else if(pRule->f_filter_type == FILTER_EXPR) { + CHKiRet(vm.Construct(&pVM)); + CHKiRet(vm.ConstructFinalize(pVM)); + CHKiRet(vm.SetMsg(pVM, pMsg)); + CHKiRet(vm.ExecProg(pVM, pRule->f_filterData.f_expr->pVmprg)); + CHKiRet(vm.PopBoolFromStack(pVM, &pResult)); + dbgprintf("result of expression evaluation: %lld\n", pResult->val.num); + /* VM is destructed on function exit */ + bRet = (pResult->val.num) ? 1 : 0; + } else { + assert(pRule->f_filter_type == FILTER_PROP); /* assert() just in case... */ + pszPropVal = MsgGetProp(pMsg, NULL, pRule->f_filterData.prop.pCSPropName, &pbMustBeFreed); + + /* Now do the compares (short list currently ;)) */ + switch(pRule->f_filterData.prop.operation ) { + case FIOP_CONTAINS: + if(rsCStrLocateInSzStr(pRule->f_filterData.prop.pCSCompValue, (uchar*) pszPropVal) != -1) + bRet = 1; + break; + case FIOP_ISEQUAL: + if(rsCStrSzStrCmp(pRule->f_filterData.prop.pCSCompValue, + (uchar*) pszPropVal, strlen(pszPropVal)) == 0) + bRet = 1; /* process message! */ + break; + case FIOP_STARTSWITH: + if(rsCStrSzStrStartsWithCStr(pRule->f_filterData.prop.pCSCompValue, + (uchar*) pszPropVal, strlen(pszPropVal)) == 0) + bRet = 1; /* process message! */ + break; + case FIOP_REGEX: + if(rsCStrSzStrMatchRegex(pRule->f_filterData.prop.pCSCompValue, + (unsigned char*) pszPropVal, 0, &pRule->f_filterData.prop.regex_cache) == RS_RET_OK) + bRet = 1; + break; + case FIOP_EREREGEX: + if(rsCStrSzStrMatchRegex(pRule->f_filterData.prop.pCSCompValue, + (unsigned char*) pszPropVal, 1, &pRule->f_filterData.prop.regex_cache) == RS_RET_OK) + bRet = 1; + break; + default: + /* here, it handles NOP (for performance reasons) */ + assert(pRule->f_filterData.prop.operation == FIOP_NOP); + bRet = 1; /* as good as any other default ;) */ + break; + } + + /* now check if the value must be negated */ + if(pRule->f_filterData.prop.isNegated) + bRet = (bRet == 1) ? 0 : 1; + + if(Debug) { + dbgprintf("Filter: check for property '%s' (value '%s') ", + rsCStrGetSzStrNoNULL(pRule->f_filterData.prop.pCSPropName), + pszPropVal); + if(pRule->f_filterData.prop.isNegated) + dbgprintf("NOT "); + dbgprintf("%s '%s': %s\n", + getFIOPName(pRule->f_filterData.prop.operation), + rsCStrGetSzStrNoNULL(pRule->f_filterData.prop.pCSCompValue), + bRet ? "TRUE" : "FALSE"); + } + + /* cleanup */ + if(pbMustBeFreed) + free(pszPropVal); + } + +finalize_it: + /* destruct in any case, not just on error, but it makes error handling much easier */ + if(pVM != NULL) + vm.Destruct(&pVM); + + if(pResult != NULL) + var.Destruct(&pResult); + + *bProcessMsg = bRet; + RETiRet; +} + + + +/* Process (consume) a received message. Calls the actions configured. + * rgerhards, 2005-10-13 + */ +static rsRetVal +processMsg(rule_t *pThis, msg_t *pMsg) +{ + int bProcessMsg; + processMsgDoActions_t DoActData; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, rule); + assert(pMsg != NULL); + + /* first check the filters... */ + CHKiRet(shouldProcessThisMessage(pThis, pMsg, &bProcessMsg)); + if(bProcessMsg) { + DoActData.pMsg = pMsg; + DoActData.bPrevWasSuspended = 0; + CHKiRet(llExecFunc(&pThis->llActList, processMsgDoActions, (void*)&DoActData)); + } + +finalize_it: + RETiRet; +} + + +/* Standard-Constructor + */ +BEGINobjConstruct(rule) /* be sure to specify the object type also in END macro! */ +ENDobjConstruct(rule) + + +/* ConstructionFinalizer + * rgerhards, 2008-01-09 + */ +static rsRetVal +ruleConstructFinalize(rule_t *pThis) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, rule); + + /* note: actionDestruct is from action.c API! */ + CHKiRet(llInit(&pThis->llActList, actionDestruct, NULL, NULL)); + +finalize_it: + RETiRet; +} + + +/* destructor for the rule object */ +BEGINobjDestruct(rule) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(rule) + if(pThis->pCSHostnameComp != NULL) + rsCStrDestruct(&pThis->pCSHostnameComp); + if(pThis->pCSProgNameComp != NULL) + rsCStrDestruct(&pThis->pCSProgNameComp); + + if(pThis->f_filter_type == FILTER_PROP) { + if(pThis->f_filterData.prop.pCSPropName != NULL) + rsCStrDestruct(&pThis->f_filterData.prop.pCSPropName); + if(pThis->f_filterData.prop.pCSCompValue != NULL) + rsCStrDestruct(&pThis->f_filterData.prop.pCSCompValue); + if(pThis->f_filterData.prop.regex_cache != NULL) + rsCStrRegexDestruct(&pThis->f_filterData.prop.regex_cache); + } else if(pThis->f_filter_type == FILTER_EXPR) { + if(pThis->f_filterData.f_expr != NULL) + expr.Destruct(&pThis->f_filterData.f_expr); + } + + llDestroy(&pThis->llActList); +ENDobjDestruct(rule) + + +/* set the associated ruleset */ +static rsRetVal +setAssRuleset(rule_t *pThis, ruleset_t *pRuleset) +{ + DEFiRet; + ISOBJ_TYPE_assert(pThis, rule); + ISOBJ_TYPE_assert(pRuleset, ruleset); + pThis->pRuleset = pRuleset; + RETiRet; +} + +/* get the associated ruleset (may be NULL if not set!) */ +static ruleset_t* +getAssRuleset(rule_t *pThis) +{ + ISOBJ_TYPE_assert(pThis, rule); + return pThis->pRuleset; +} + + +/* helper to DebugPrint, to print out all actions via + * the llExecFunc() facility. + */ +DEFFUNC_llExecFunc(dbgPrintInitInfoAction) +{ + DEFiRet; + iRet = actionDbgPrint((action_t*) pData); + dbgprintf("\n"); + RETiRet; +} + + +/* debugprint for the rule object */ +BEGINobjDebugPrint(rule) /* be sure to specify the object type also in END and CODESTART macros! */ + int i; +CODESTARTobjDebugPrint(rule) + dbgoprint((obj_t*) pThis, "rsyslog rule:\n"); + if(pThis->pCSProgNameComp != NULL) + dbgprintf("tag: '%s'\n", rsCStrGetSzStrNoNULL(pThis->pCSProgNameComp)); + if(pThis->eHostnameCmpMode != HN_NO_COMP) + dbgprintf("hostname: %s '%s'\n", + pThis->eHostnameCmpMode == HN_COMP_MATCH ? + "only" : "allbut", + rsCStrGetSzStrNoNULL(pThis->pCSHostnameComp)); + if(pThis->f_filter_type == FILTER_PRI) { + for (i = 0; i <= LOG_NFACILITIES; i++) + if (pThis->f_filterData.f_pmask[i] == TABLE_NOPRI) + dbgprintf(" X "); + else + dbgprintf("%2X ", pThis->f_filterData.f_pmask[i]); + } else if(pThis->f_filter_type == FILTER_EXPR) { + dbgprintf("EXPRESSION-BASED Filter: can currently not be displayed"); + } else { + dbgprintf("PROPERTY-BASED Filter:\n"); + dbgprintf("\tProperty.: '%s'\n", + rsCStrGetSzStrNoNULL(pThis->f_filterData.prop.pCSPropName)); + dbgprintf("\tOperation: "); + if(pThis->f_filterData.prop.isNegated) + dbgprintf("NOT "); + dbgprintf("'%s'\n", getFIOPName(pThis->f_filterData.prop.operation)); + dbgprintf("\tValue....: '%s'\n", + rsCStrGetSzStrNoNULL(pThis->f_filterData.prop.pCSCompValue)); + dbgprintf("\tAction...: "); + } + + dbgprintf("\nActions:\n"); + llExecFunc(&pThis->llActList, dbgPrintInitInfoAction, NULL); /* actions */ + + dbgprintf("\n"); +ENDobjDebugPrint(rule) + + +/* queryInterface function + * rgerhards, 2008-02-21 + */ +BEGINobjQueryInterface(rule) +CODESTARTobjQueryInterface(rule) + if(pIf->ifVersion != ruleCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->Construct = ruleConstruct; + pIf->ConstructFinalize = ruleConstructFinalize; + pIf->Destruct = ruleDestruct; + pIf->DebugPrint = ruleDebugPrint; + + pIf->IterateAllActions = iterateAllActions; + pIf->ProcessMsg = processMsg; + pIf->SetAssRuleset = setAssRuleset; + pIf->GetAssRuleset = getAssRuleset; +finalize_it: +ENDobjQueryInterface(rule) + + +/* Exit the rule class. + * rgerhards, 2009-04-06 + */ +BEGINObjClassExit(rule, OBJ_IS_CORE_MODULE) /* class, version */ + objRelease(errmsg, CORE_COMPONENT); + objRelease(expr, CORE_COMPONENT); + objRelease(var, CORE_COMPONENT); + objRelease(vm, CORE_COMPONENT); +ENDObjClassExit(rule) + + +/* Initialize the rule class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-19 + */ +BEGINObjClassInit(rule, 1, OBJ_IS_CORE_MODULE) /* class, version */ + /* request objects we use */ + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(expr, CORE_COMPONENT)); + CHKiRet(objUse(var, CORE_COMPONENT)); + CHKiRet(objUse(vm, CORE_COMPONENT)); + + /* set our own handlers */ + OBJSetMethodHandler(objMethod_DEBUGPRINT, ruleDebugPrint); + OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, ruleConstructFinalize); +ENDObjClassInit(rule) + +/* vi:set ai: + */ diff --git a/runtime/rule.h b/runtime/rule.h new file mode 100644 index 00000000..38b11c63 --- /dev/null +++ b/runtime/rule.h @@ -0,0 +1,77 @@ +/* The rule object. + * + * This implements rules within rsyslog. + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_RULE_H +#define INCLUDED_RULE_H + +#include "linkedlist.h" +#include "regexp.h" +#include "expr.h" + +/* the rule object */ +struct rule_s { + BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ + /* filter properties */ + enum { + FILTER_PRI = 0, /* traditional PRI based filer */ + FILTER_PROP = 1, /* extended filter, property based */ + FILTER_EXPR = 2 /* extended filter, expression based */ + } f_filter_type; + EHostnameCmpMode eHostnameCmpMode; + cstr_t *pCSHostnameComp; /* hostname to check */ + cstr_t *pCSProgNameComp; /* tag to check or NULL, if not to be checked */ + union { + u_char f_pmask[LOG_NFACILITIES+1]; /* priority mask */ + struct { + cstr_t *pCSPropName; + fiop_t operation; + regex_t *regex_cache; /* cache for compiled REs, if such are used */ + cstr_t *pCSCompValue; /* value to "compare" against */ + char isNegated; /* actually a boolean ;) */ + } prop; + expr_t *f_expr; /* expression object */ + } f_filterData; + + ruleset_t *pRuleset; /* associated ruleset */ + linkedList_t llActList; /* list of configured actions */ +}; + +/* interfaces */ +BEGINinterface(rule) /* name must also be changed in ENDinterface macro! */ + INTERFACEObjDebugPrint(rule); + rsRetVal (*Construct)(rule_t **ppThis); + rsRetVal (*ConstructFinalize)(rule_t __attribute__((unused)) *pThis); + rsRetVal (*Destruct)(rule_t **ppThis); + rsRetVal (*IterateAllActions)(rule_t *pThis, rsRetVal (*pFunc)(void*, void*), void *pParam); + rsRetVal (*ProcessMsg)(rule_t *pThis, msg_t *pMsg); + rsRetVal (*SetAssRuleset)(rule_t *pThis, ruleset_t*); + ruleset_t* (*GetAssRuleset)(rule_t *pThis); +ENDinterface(rule) +#define ruleCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ + + +/* prototypes */ +PROTOTYPEObj(rule); + +#endif /* #ifndef INCLUDED_RULE_H */ diff --git a/runtime/ruleset.c b/runtime/ruleset.c new file mode 100644 index 00000000..93d40e24 --- /dev/null +++ b/runtime/ruleset.c @@ -0,0 +1,451 @@ +/* ruleset.c - rsyslog's ruleset object + * + * We have a two-way structure of linked lists: one global linked list + * (llAllRulesets) hold alls rule sets that we know. Included in each + * list is a list of rules (which contain a list of actions, but that's + * a different story). + * + * Usually, only a single rule set is executed. However, there exist some + * situations where all rules must be iterated over, for example on HUP. Thus, + * we also provide interfaces to do that. + * + * Module begun 2009-06-10 by Rainer Gerhards + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ + +#include "config.h" +#include <stdlib.h> +#include <string.h> +#include <assert.h> +#include <ctype.h> + +#include "rsyslog.h" +#include "obj.h" +#include "msg.h" +#include "ruleset.h" +#include "rule.h" +#include "errmsg.h" +#include "unicode-helper.h" + +static rsRetVal debugPrintAll(void); // TODO: remove! + +/* static data */ +DEFobjStaticHelpers +DEFobjCurrIf(errmsg) +DEFobjCurrIf(rule) + +linkedList_t llRulesets; /* this is NOT a pointer - no typo here ;) */ +ruleset_t *pCurrRuleset = NULL; /* currently "active" ruleset */ +ruleset_t *pDfltRuleset = NULL; /* currentl default ruleset, e.g. for binding to actions which have no other */ + +/* ---------- linked-list key handling functions ---------- */ + +/* destructor for linked list keys. + */ +static rsRetVal keyDestruct(void __attribute__((unused)) *pData) +{ + free(pData); + return RS_RET_OK; +} + + +/* ---------- END linked-list key handling functions ---------- */ + + +/* driver to iterate over all of this ruleset actions */ +typedef struct iterateAllActions_s { + rsRetVal (*pFunc)(void*, void*); + void *pParam; +} iterateAllActions_t; +DEFFUNC_llExecFunc(doIterateRulesetActions) +{ + DEFiRet; + rule_t* pRule = (rule_t*) pData; + iterateAllActions_t *pMyParam = (iterateAllActions_t*) pParam; + iRet = rule.IterateAllActions(pRule, pMyParam->pFunc, pMyParam->pParam); + RETiRet; +} +#if 0 +/* iterate over all actions of THIS rule set. + */ +static rsRetVal +iterateRulesetAllActions(ruleset_t *pThis, rsRetVal (*pFunc)(void*, void*), void* pParam) +{ + iterateAllActions_t params; + DEFiRet; + assert(pFunc != NULL); + + params.pFunc = pFunc; + params.pParam = pParam; + CHKiRet(llExecFunc(&llRulesets, doIterateRulesetActions, ¶ms)); + +finalize_it: + RETiRet; +} + + +/* driver to iterate over all actions */ +DEFFUNC_llExecFunc(doIterateAllActions) +{ + DEFiRet; + ruleset_t* pThis = (ruleset_t*) pData; + iterateAllActions_t *pMyParam = (iterateAllActions_t*) pParam; + iRet = iterateRulesetAllActions(pThis, pMyParam->pFunc, pMyParam->pParam); + RETiRet; +} +#endif +/* iterate over ALL actions present in the WHOLE system. + * this is often needed, for example when HUP processing + * must be done or a shutdown is pending. + */ +static rsRetVal +iterateAllActions(rsRetVal (*pFunc)(void*, void*), void* pParam) +{ + iterateAllActions_t params; + DEFiRet; + assert(pFunc != NULL); + + params.pFunc = pFunc; + params.pParam = pParam; + //CHKiRet(llExecFunc(&llRulesets, doIterateAllActions, ¶ms)); + CHKiRet(llExecFunc(&llRulesets, doIterateRulesetActions, ¶ms)); + +finalize_it: + RETiRet; +} + + + +/* helper to processMsg(), used to call the configured actions. It is + * executed from within llExecFunc() of the action list. + * rgerhards, 2007-08-02 + */ +DEFFUNC_llExecFunc(processMsgDoRules) +{ + ISOBJ_TYPE_assert(pData, rule); + return rule.ProcessMsg((rule_t*) pData, (msg_t*) pParam); +} + + +/* Process (consume) a received message. Calls the actions configured. + * rgerhards, 2005-10-13 + */ +static rsRetVal +processMsg(msg_t *pMsg) +{ + ruleset_t *pThis; + DEFiRet; + assert(pMsg != NULL); + + pThis = (pMsg->pRuleset == NULL) ? pDfltRuleset : pMsg->pRuleset; + ISOBJ_TYPE_assert(pThis, ruleset); + + CHKiRet(llExecFunc(&pThis->llRules, processMsgDoRules, pMsg)); + +finalize_it: + if(iRet == RS_RET_DISCARDMSG) + iRet = RS_RET_OK; + + RETiRet; +} + +/* Add a new rule to the end of the current rule set. We do a number + * of checks and ignore the rule if it does not pass them. + */ +static rsRetVal +addRule(ruleset_t *pThis, rule_t **ppRule) +{ + int iActionCnt; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, ruleset); + ISOBJ_TYPE_assert(*ppRule, rule); + + CHKiRet(llGetNumElts(&(*ppRule)->llActList, &iActionCnt)); + if(iActionCnt == 0) { + errmsg.LogError(0, NO_ERRCODE, "warning: selector line without actions will be discarded"); + rule.Destruct(ppRule); + } else { + CHKiRet(llAppend(&pThis->llRules, NULL, *ppRule)); + dbgprintf("selector line successfully processed\n"); + } + +finalize_it: + RETiRet; +} + + +/* set name for ruleset */ +static rsRetVal setName(ruleset_t *pThis, uchar *pszName) +{ + DEFiRet; + free(pThis->pszName); + CHKmalloc(pThis->pszName = ustrdup(pszName)); + +finalize_it: + RETiRet; +} + + +/* get current ruleset + * We use a non-standard calling interface, as nothing can go wrong and it + * is really much more natural to return the pointer directly. + */ +static ruleset_t* +GetCurrent(void) +{ + return pCurrRuleset; +} + + +/* Find the ruleset with the given name and return a pointer to its object. + */ +static rsRetVal +GetRuleset(ruleset_t **ppRuleset, uchar *pszName) +{ + DEFiRet; + assert(ppRuleset != NULL); + assert(pszName != NULL); + + CHKiRet(llFind(&llRulesets, pszName, (void*) ppRuleset)); + +finalize_it: + RETiRet; +} + + +/* Set a new default rule set. If the default can not be found, no change happens. + */ +static rsRetVal +SetDefaultRuleset(uchar *pszName) +{ + ruleset_t *pRuleset; + DEFiRet; + assert(pszName != NULL); + + CHKiRet(GetRuleset(&pRuleset, pszName)); + pDfltRuleset = pRuleset; + dbgprintf("default rule set changed to %p: '%s'\n", pRuleset, pszName); + +finalize_it: + RETiRet; +} + + +/* Set a new current rule set. If the ruleset can not be found, no change happens. + */ +static rsRetVal +SetCurrRuleset(uchar *pszName) +{ + ruleset_t *pRuleset; + DEFiRet; + assert(pszName != NULL); + + CHKiRet(GetRuleset(&pRuleset, pszName)); + pCurrRuleset = pRuleset; + dbgprintf("current rule set changed to %p: '%s'\n", pRuleset, pszName); + +finalize_it: + RETiRet; +} + + +/* destructor we need to destruct rules inside our linked list contents. + */ +static rsRetVal +doRuleDestruct(void *pData) +{ + rule_t *pRule = (rule_t *) pData; + DEFiRet; + rule.Destruct(&pRule); + RETiRet; +} + + +/* Standard-Constructor + */ +BEGINobjConstruct(ruleset) /* be sure to specify the object type also in END macro! */ + CHKiRet(llInit(&pThis->llRules, doRuleDestruct, NULL, NULL)); +finalize_it: +ENDobjConstruct(ruleset) + + +/* ConstructionFinalizer + * This also adds the rule set to the list of all known rulesets. + */ +static rsRetVal +rulesetConstructFinalize(ruleset_t *pThis) +{ + uchar *keyName; + DEFiRet; + ISOBJ_TYPE_assert(pThis, ruleset); + + /* we must duplicate our name, as the key destructer would also + * free it, resulting in a double-free. It's also cleaner to have + * two separate copies. + */ + CHKmalloc(keyName = ustrdup(pThis->pszName)); + CHKiRet(llAppend(&llRulesets, keyName, pThis)); + + /* this now also is the new current ruleset */ + pCurrRuleset = pThis; + + /* and also the default, if so far none has been set */ + if(pDfltRuleset == NULL) + pDfltRuleset = pThis; + +finalize_it: + RETiRet; +} + + +/* destructor for the ruleset object */ +BEGINobjDestruct(ruleset) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDestruct(ruleset) + dbgprintf("destructing ruleset %p, name %p\n", pThis, pThis->pszName); + llDestroy(&pThis->llRules); + free(pThis->pszName); +ENDobjDestruct(ruleset) + +/* this is a special destructor for the linkedList class. LinkedList does NOT + * provide a pointer to the pointer, but rather the raw pointer itself. So we + * must map this, otherwise the destructor will abort. + */ +static rsRetVal +rulesetDestructForLinkedList(void *pData) +{ + ruleset_t *pThis = (ruleset_t*) pData; + return rulesetDestruct(&pThis); +} + + +/* destruct ALL rule sets that reside in the system. This must + * be callable before unloading this module as the module may + * not be unloaded before unload of the actions is required. This is + * kind of a left-over from previous logic and may be optimized one + * everything runs stable again. -- rgerhards, 2009-06-10 + */ +static rsRetVal +destructAllActions(void) +{ + DEFiRet; + + CHKiRet(llDestroy(&llRulesets)); + CHKiRet(llInit(&llRulesets, rulesetDestructForLinkedList, keyDestruct, strcasecmp)); + +finalize_it: + RETiRet; +} + +/* helper for debugPrint(), initiates rule printing */ +DEFFUNC_llExecFunc(doDebugPrintRule) +{ + return rule.DebugPrint((rule_t*) pData); +} +/* debugprint for the ruleset object */ +BEGINobjDebugPrint(ruleset) /* be sure to specify the object type also in END and CODESTART macros! */ +CODESTARTobjDebugPrint(ruleset) + dbgoprint((obj_t*) pThis, "rsyslog ruleset %s:\n", pThis->pszName); + llExecFunc(&pThis->llRules, doDebugPrintRule, NULL); +ENDobjDebugPrint(ruleset) + + +/* helper for debugPrintAll(), prints a single ruleset */ +DEFFUNC_llExecFunc(doDebugPrintAll) +{ + return rulesetDebugPrint((ruleset_t*) pData); +} +/* debug print all rulesets + */ +static rsRetVal +debugPrintAll(void) +{ + DEFiRet; + dbgprintf("All Rulesets:\n"); + llExecFunc(&llRulesets, doDebugPrintAll, NULL); + dbgprintf("End of Rulesets.\n"); + RETiRet; +} + + +/* queryInterface function + * rgerhards, 2008-02-21 + */ +BEGINobjQueryInterface(ruleset) +CODESTARTobjQueryInterface(ruleset) + if(pIf->ifVersion != rulesetCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->Construct = rulesetConstruct; + pIf->ConstructFinalize = rulesetConstructFinalize; + pIf->Destruct = rulesetDestruct; + pIf->DebugPrint = rulesetDebugPrint; + + pIf->IterateAllActions = iterateAllActions; + pIf->DestructAllActions = destructAllActions; + pIf->AddRule = addRule; + pIf->ProcessMsg = processMsg; + pIf->SetName = setName; + pIf->DebugPrintAll = debugPrintAll; + pIf->GetCurrent = GetCurrent; + pIf->GetRuleset = GetRuleset; + pIf->SetDefaultRuleset = SetDefaultRuleset; + pIf->SetCurrRuleset = SetCurrRuleset; +finalize_it: +ENDobjQueryInterface(ruleset) + + +/* Exit the ruleset class. + * rgerhards, 2009-04-06 + */ +BEGINObjClassExit(ruleset, OBJ_IS_CORE_MODULE) /* class, version */ + llDestroy(&llRulesets); + objRelease(errmsg, CORE_COMPONENT); + objRelease(rule, CORE_COMPONENT); +ENDObjClassExit(ruleset) + + +/* Initialize the ruleset class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-19 + */ +BEGINObjClassInit(ruleset, 1, OBJ_IS_CORE_MODULE) /* class, version */ + /* request objects we use */ + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(objUse(rule, CORE_COMPONENT)); + + /* set our own handlers */ + OBJSetMethodHandler(objMethod_DEBUGPRINT, rulesetDebugPrint); + OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, rulesetConstructFinalize); + + /* prepare global data */ + CHKiRet(llInit(&llRulesets, rulesetDestructForLinkedList, keyDestruct, strcasecmp)); +ENDObjClassInit(ruleset) + +/* vi:set ai: + */ diff --git a/runtime/ruleset.h b/runtime/ruleset.h new file mode 100644 index 00000000..32571687 --- /dev/null +++ b/runtime/ruleset.h @@ -0,0 +1,60 @@ +/* The ruleset object. + * + * This implements rulesets within rsyslog. + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_RULESET_H +#define INCLUDED_RULESET_H + +#include "linkedlist.h" + +/* the ruleset object */ +struct ruleset_s { + BEGINobjInstance; /* Data to implement generic object - MUST be the first data element! */ + linkedList_t llRules; /* this is NOT a pointer - no typo here ;) */ + uchar *pszName; /* name of our ruleset */ +}; + +/* interfaces */ +BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ + INTERFACEObjDebugPrint(ruleset); + rsRetVal (*DebugPrintAll)(void); + rsRetVal (*Construct)(ruleset_t **ppThis); + rsRetVal (*ConstructFinalize)(ruleset_t __attribute__((unused)) *pThis); + rsRetVal (*Destruct)(ruleset_t **ppThis); + rsRetVal (*IterateAllActions)(rsRetVal (*pFunc)(void*, void*), void* pParam); + rsRetVal (*DestructAllActions)(void); + rsRetVal (*AddRule)(ruleset_t *pThis, rule_t **ppRule); + rsRetVal (*SetName)(ruleset_t *pThis, uchar *pszName); + rsRetVal (*ProcessMsg)(msg_t *pMsg); + rsRetVal (*GetRuleset)(ruleset_t **ppThis, uchar*); + rsRetVal (*SetDefaultRuleset)(uchar*); + rsRetVal (*SetCurrRuleset)(uchar*); + ruleset_t* (*GetCurrent)(void); +ENDinterface(ruleset) +#define rulesetCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ + + +/* prototypes */ +PROTOTYPEObj(ruleset); + +#endif /* #ifndef INCLUDED_RULESET_H */ diff --git a/runtime/srUtils.h b/runtime/srUtils.h index bfce4cbb..b37559cf 100644 --- a/runtime/srUtils.h +++ b/runtime/srUtils.h @@ -92,6 +92,7 @@ void srSleep(int iSeconds, int iuSeconds); char *rs_strerror_r(int errnum, char *buf, size_t buflen); int decodeSyslogName(uchar *name, syslogName_t *codetab); int getSubString(uchar **ppSrc, char *pDst, size_t DstSize, char cSep); +rsRetVal getFileSize(uchar *pszName, off_t *pSize); /* mutex operations */ /* some macros to cancel-safe lock a mutex (it will automatically be released @@ -124,4 +125,17 @@ int getSubString(uchar **ppSrc, char *pDst, size_t DstSize, char cSep); d_pthread_mutex_unlock(mut); \ pthread_setcancelstate(iCancelStateSave, NULL); \ } + +/* The unconditional versions of the macro always lock the mutex. They are preferred in + * complex scenarios, where the simple ones might get mixed up by multiple calls. + */ +#define DEFVARS_mutexProtection_uncond\ + int iCancelStateSave +#define BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(mut) \ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &iCancelStateSave); \ + d_pthread_mutex_lock(mut); +#define END_MTX_PROTECTED_OPERATIONS_UNCOND(mut) \ + d_pthread_mutex_unlock(mut); \ + pthread_setcancelstate(iCancelStateSave, NULL); + #endif diff --git a/runtime/srutils.c b/runtime/srutils.c index d01ca20d..5407531f 100644 --- a/runtime/srutils.c +++ b/runtime/srutils.c @@ -553,6 +553,33 @@ int getSubString(uchar **ppSrc, char *pDst, size_t DstSize, char cSep) } +/* get the size of a file or return appropriate error code. If an error is returned, + * *pSize content is undefined. + * rgerhards, 2009-06-12 + */ +rsRetVal +getFileSize(uchar *pszName, off_t *pSize) +{ + int ret; + struct stat statBuf; + DEFiRet; + + ret = stat((char*) pszName, &statBuf); + if(ret == -1) { + switch(errno) { + case EACCES: ABORT_FINALIZE(RS_RET_NO_FILE_ACCESS); + case ENOTDIR: + case ENOENT: ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); + default: ABORT_FINALIZE(RS_RET_FILE_NO_STAT); + } + } + + *pSize = statBuf.st_size; + +finalize_it: + RETiRet; +} + /* vim:set ai: */ diff --git a/runtime/stream.c b/runtime/stream.c index 1cff2da6..f13258b5 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -1,4 +1,3 @@ -//TODO: O_TRUC mode! /* The serial stream class. * * A serial stream provides serial data access. In theory, serial streams @@ -7,8 +6,9 @@ * "driver"). * * File begun on 2008-01-09 by RGerhards + * Large modifications in 2009-06 to support using it with omfile, including zip writer. * - * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -39,24 +39,199 @@ #include <unistd.h> #include <sys/stat.h> /* required for HP UX */ #include <errno.h> +#include <pthread.h> #include "rsyslog.h" #include "stringbuf.h" #include "srUtils.h" #include "obj.h" #include "stream.h" +#include "unicode-helper.h" +#include "module-template.h" +#include "apc.h" /* static data */ DEFobjStaticHelpers +DEFobjCurrIf(zlibw) +DEFobjCurrIf(apc) + +/* forward definitions */ +static rsRetVal strmFlush(strm_t *pThis); +static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); +static rsRetVal strmCloseFile(strm_t *pThis); + /* methods */ -/* first, we define type-specific handlers. The provide a generic functionality, +/* async flush apc handler + */ +static void +flushApc(void *param1, void __attribute__((unused)) *param2) +{ + DEFVARS_mutexProtection_uncond; + strm_t *pThis = (strm_t*) param1; + ISOBJ_TYPE_assert(pThis, strm); + + BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); + strmFlush(pThis); + END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); +} + + +/* Try to resolve a size limit situation. This is used to support custom-file size handlers + * for omfile. It first runs the command, and then checks if we are still above the size + * treshold. Note that this works only with single file names, NOT with circular names. + * Note that pszCurrFName can NOT be taken from pThis, because the stream is closed when + * we are called (and that destroys pszCurrFName, as there is NO CURRENT file name!). So + * we need to receive the name as a parameter. + * initially wirtten 2005-06-21, moved to this class & updates 2009-06-01, both rgerhards + */ +static rsRetVal +resolveFileSizeLimit(strm_t *pThis, uchar *pszCurrFName) +{ + uchar *pParams; + uchar *pCmd; + uchar *p; + off_t actualFileSize; + rsRetVal localRet; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); + assert(pszCurrFName != NULL); + + if(pThis->pszSizeLimitCmd == NULL) { + ABORT_FINALIZE(RS_RET_NON_SIZELIMITCMD); /* nothing we can do in this case... */ + } + + /* we first check if we have command line parameters. We assume this, + * when we have a space in the program name. If we find it, everything after + * the space is treated as a single argument. + */ + CHKmalloc(pCmd = ustrdup(pThis->pszSizeLimitCmd)); + + for(p = pCmd ; *p && *p != ' ' ; ++p) { + /* JUST SKIP */ + } + + if(*p == ' ') { + *p = '\0'; /* pretend string-end */ + pParams = p+1; + } else + pParams = NULL; + + /* the execProg() below is probably not great, but at least is is + * fairly secure now. Once we change the way file size limits are + * handled, we should also revisit how this command is run (and + * with which parameters). rgerhards, 2007-07-20 + */ + execProg(pCmd, 1, pParams); + + free(pCmd); + + localRet = getFileSize(pszCurrFName, &actualFileSize); + + if(localRet == RS_RET_OK && actualFileSize >= pThis->iSizeLimit) { + ABORT_FINALIZE(RS_RET_SIZELIMITCMD_DIDNT_RESOLVE); /* OK, it didn't work out... */ + } else if(localRet != RS_RET_FILE_NOT_FOUND) { + /* file not found is OK, the command may have moved away the file */ + ABORT_FINALIZE(localRet); + } + +finalize_it: + if(iRet != RS_RET_OK) { + if(iRet == RS_RET_SIZELIMITCMD_DIDNT_RESOLVE) + dbgprintf("file size limit cmd for file '%s' did no resolve situation\n", pszCurrFName); + else + dbgprintf("file size limit cmd for file '%s' failed with code %d.\n", pszCurrFName, iRet); + pThis->bDisabled = 1; + } + + RETiRet; +} + + +/* Check if the file has grown beyond the configured omfile iSizeLimit + * and, if so, initiate processing. + */ +static rsRetVal +doSizeLimitProcessing(strm_t *pThis) +{ + uchar *pszCurrFName = NULL; + DEFiRet; + + ISOBJ_TYPE_assert(pThis, strm); + ASSERT(pThis->iSizeLimit != 0); + ASSERT(pThis->fd != -1); + + if(pThis->iCurrOffs >= pThis->iSizeLimit) { + /* strmClosefile() destroys the current file name, so we + * need to preserve it. + */ + CHKmalloc(pszCurrFName = ustrdup(pThis->pszCurrFName)); + CHKiRet(strmCloseFile(pThis)); + CHKiRet(resolveFileSizeLimit(pThis, pszCurrFName)); + } + +finalize_it: + free(pszCurrFName); + RETiRet; +} + + +/* now, we define type-specific handlers. The provide a generic functionality, * but for this specific type of strm. The mapping to these handlers happens during * strm construction. Later on, handlers are called by pointers present in the * strm instance object. */ +/* do the physical open() call on a file. + */ +static rsRetVal +doPhysOpen(strm_t *pThis) +{ + int iFlags; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); + + /* compute which flags we need to provide to open */ + switch(pThis->tOperationsMode) { + case STREAMMODE_READ: + iFlags = O_CLOEXEC | O_NOCTTY | O_RDONLY; + break; + case STREAMMODE_WRITE: /* legacy mode used inside queue engine */ + iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT; + break; + case STREAMMODE_WRITE_TRUNC: + iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT | O_TRUNC; + break; + case STREAMMODE_WRITE_APPEND: + iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT | O_APPEND; + break; + default:assert(0); + break; + } + + pThis->fd = open((char*)pThis->pszCurrFName, iFlags, pThis->tOpenMode); + if(pThis->fd == -1) { + int ierrnoSave = errno; + dbgoprint((obj_t*) pThis, "open error %d, file '%s'\n", errno, pThis->pszCurrFName); + if(ierrnoSave == ENOENT) + ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); + else + ABORT_FINALIZE(RS_RET_IO_ERROR); + } else { + if(!ustrcmp(pThis->pszCurrFName, UCHAR_CONSTANT(_PATH_CONSOLE)) || isatty(pThis->fd)) { + DBGPRINTF("file %d is a tty-type file\n", pThis->fd); + pThis->bIsTTY = 1; + } else { + pThis->bIsTTY = 0; + } + } + +finalize_it: + RETiRet; +} + + /* open a strm file * It is OK to call this function when the stream is already open. In that * case, it returns immediately with RS_RET_OK @@ -64,10 +239,8 @@ DEFobjStaticHelpers static rsRetVal strmOpenFile(strm_t *pThis) { DEFiRet; - int iFlags; ASSERT(pThis != NULL); - ASSERT(pThis->tOperationsMode == STREAMMODE_READ || pThis->tOperationsMode == STREAMMODE_WRITE); if(pThis->fd != -1) ABORT_FINALIZE(RS_RET_OK); @@ -88,28 +261,18 @@ static rsRetVal strmOpenFile(strm_t *pThis) } } - /* compute which flags we need to provide to open */ - if(pThis->tOperationsMode == STREAMMODE_READ) - iFlags = O_RDONLY; - else - iFlags = O_WRONLY | O_CREAT; - - iFlags |= pThis->iAddtlOpenFlags; - - pThis->fd = open((char*)pThis->pszCurrFName, iFlags, pThis->tOpenMode); - if(pThis->fd == -1) { - int ierrnoSave = errno; - dbgoprint((obj_t*) pThis, "open error %d, file '%s'\n", errno, pThis->pszCurrFName); - if(ierrnoSave == ENOENT) - ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); - else - ABORT_FINALIZE(RS_RET_IO_ERROR); - } + CHKiRet(doPhysOpen(pThis)); pThis->iCurrOffs = 0; + if(pThis->tOperationsMode == STREAMMODE_WRITE_APPEND) { + /* we need to obtain the current offset */ + off_t offset; + CHKiRet(getFileSize(pThis->pszCurrFName, &offset)); + pThis->iCurrOffs = offset; + } - dbgoprint((obj_t*) pThis, "opened file '%s' for %s (0x%x) as %d\n", pThis->pszCurrFName, - (pThis->tOperationsMode == STREAMMODE_READ) ? "READ" : "WRITE", iFlags, pThis->fd); + dbgoprint((obj_t*) pThis, "opened file '%s' for %s as %d\n", pThis->pszCurrFName, + (pThis->tOperationsMode == STREAMMODE_READ) ? "READ" : "WRITE", pThis->fd); finalize_it: RETiRet; @@ -128,14 +291,26 @@ static rsRetVal strmCloseFile(strm_t *pThis) ASSERT(pThis->fd != -1); dbgoprint((obj_t*) pThis, "file %d closing\n", pThis->fd); - if(pThis->tOperationsMode == STREAMMODE_WRITE) + if(pThis->tOperationsMode != STREAMMODE_READ) strmFlush(pThis); - close(pThis->fd); // TODO: error check + close(pThis->fd); pThis->fd = -1; + if(pThis->fdDir != -1) { + /* close associated directory handle, if it is open */ + close(pThis->fdDir); + pThis->fdDir = -1; + } + if(pThis->bDeleteOnClose) { - unlink((char*) pThis->pszCurrFName); // TODO: check returncode + if(unlink((char*) pThis->pszCurrFName) == -1) { + char errStr[1024]; + int err = errno; + rs_strerror_r(err, errStr, sizeof(errStr)); + DBGPRINTF("error %d unlinking '%s' - ignored: %s\n", + errno, pThis->pszCurrFName, errStr); + } } pThis->iCurrOffs = 0; /* we are back at begin of file */ @@ -234,10 +409,6 @@ strmHandleEOF(strm_t *pThis) case STREAMTYPE_FILE_CIRCULAR: /* we have multiple files and need to switch to the next one */ /* TODO: think about emulating EOF in this case (not yet needed) */ -#if 0 - if(pThis->iMaxFiles == 0) /* TODO: why do we need this? ;) */ - ABORT_FINALIZE(RS_RET_EOF); -#endif dbgoprint((obj_t*) pThis, "file %d EOF\n", pThis->fd); CHKiRet(strmNextFile(pThis)); break; @@ -295,7 +466,7 @@ finalize_it: * NOTE: needs to be enhanced to support sticking with a strm entry (if not * deleted). */ -rsRetVal strmReadChar(strm_t *pThis, uchar *pC) +static rsRetVal strmReadChar(strm_t *pThis, uchar *pC) { DEFiRet; @@ -329,7 +500,7 @@ finalize_it: * character buffering capability. * rgerhards, 2008-01-07 */ -rsRetVal strmUnreadChar(strm_t *pThis, uchar c) +static rsRetVal strmUnreadChar(strm_t *pThis, uchar c) { ASSERT(pThis != NULL); ASSERT(pThis->iUngetC == -1); @@ -351,7 +522,7 @@ rsRetVal strmUnreadChar(strm_t *pThis, uchar c) * are pthread_killed() upon termination. So if we use their native pointer, they * can cleanup (but only then). */ -rsRetVal +static rsRetVal strmReadLine(strm_t *pThis, cstr_t **ppCStr) { DEFiRet; @@ -383,26 +554,55 @@ finalize_it: BEGINobjConstruct(strm) /* be sure to specify the object type also in END macro! */ pThis->iCurrFNum = 1; pThis->fd = -1; + pThis->fdDir = -1; pThis->iUngetC = -1; pThis->sType = STREAMTYPE_FILE_SINGLE; pThis->sIOBufSize = glblGetIOBufSize(); - pThis->tOpenMode = 0600; /* TODO: make configurable */ + pThis->tOpenMode = 0600; ENDobjConstruct(strm) /* ConstructionFinalizer * rgerhards, 2008-01-09 */ -rsRetVal strmConstructFinalize(strm_t *pThis) +static rsRetVal strmConstructFinalize(strm_t *pThis) { + rsRetVal localRet; DEFiRet; ASSERT(pThis != NULL); - if(pThis->pIOBuf == NULL) { /* allocate our io buffer in case we have not yet */ - if((pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - pThis->iBufPtrMax = 0; /* results in immediate read request */ + CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)); + pThis->iBufPtrMax = 0; /* results in immediate read request */ + if(pThis->iZipLevel) { /* do we need a zip buf? */ + localRet = objUse(zlibw, LM_ZLIBW_FILENAME); + if(localRet != RS_RET_OK) { + pThis->iZipLevel = 0; + DBGPRINTF("stream was requested with zip mode, but zlibw module unavailable (%d) - using " + "without zip\n", localRet); + } else { + /* we use the same size as the original buf, as we would like + * to make sure we can write out everyting with a SINGLE api call! + */ + CHKmalloc(pThis->pZipBuf = (Bytef*) malloc(sizeof(uchar) * pThis->sIOBufSize)); + } + } + + /* if we are aset to sync, we must obtain a file handle to the directory for fsync() purposes */ + if(pThis->bSync && !pThis->bIsTTY) { + pThis->fdDir = open((char*)pThis->pszDir, O_RDONLY | O_CLOEXEC | O_NOCTTY); + if(pThis->fdDir == -1) { + char errStr[1024]; + int err = errno; + rs_strerror_r(err, errStr, sizeof(errStr)); + DBGPRINTF("error %d opening directory file for fsync() use - fsync for directory disabled: %s\n", + errno, errStr); + } + } + + /* if we should call flush apc's, we need a mutex */ + if(pThis->iFlushInterval != 0) { + pthread_mutex_init(&pThis->mut, 0); } finalize_it: @@ -413,21 +613,27 @@ finalize_it: /* destructor for the strm object */ BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */ CODESTARTobjDestruct(strm) - if(pThis->tOperationsMode == STREAMMODE_WRITE) + if(pThis->tOperationsMode != STREAMMODE_READ) strmFlush(pThis); /* ... then free resources */ if(pThis->fd != -1) strmCloseFile(pThis); - if(pThis->pszDir != NULL) - free(pThis->pszDir); - if(pThis->pIOBuf != NULL) - free(pThis->pIOBuf); - if(pThis->pszCurrFName != NULL) - free(pThis->pszCurrFName); - if(pThis->pszFName != NULL) - free(pThis->pszFName); + if(pThis->iZipLevel) { /* do we need a zip buf? */ + objRelease(zlibw, LM_ZLIBW_FILENAME); + } + + if(pThis->iFlushInterval != 0) { + // TODO: check if there is an apc and remove it! + pthread_mutex_destroy(&pThis->mut); + } + + free(pThis->pszDir); + free(pThis->pIOBuf); + free(pThis->pZipBuf); + free(pThis->pszCurrFName); + free(pThis->pszFName); ENDobjDestruct(strm) @@ -453,48 +659,240 @@ finalize_it: RETiRet; } -/* write memory buffer to a stream object. - * To support direct writes of large objects, this method may be called - * with a buffer pointing to some region other than the stream buffer itself. - * However, in that case the stream buffer must be empty (strmFlush() has to - * be called before), because we would otherwise mess up with the sequence - * inside the stream. -- rgerhards, 2008-01-10 + +/* try to recover a tty after a write error. This may have happend + * due to vhangup(), and, if so, we can simply re-open it. */ -static rsRetVal strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf) +#ifdef linux +# define ERR_TTYHUP EIO +#else +# define ERR_TTYHUP EBADF +#endif +static rsRetVal +tryTTYRecover(strm_t *pThis, int err) { DEFiRet; - int iWritten; + ISOBJ_TYPE_assert(pThis, strm); + if(err == ERR_TTYHUP) { + close(pThis->fd); + CHKiRet(doPhysOpen(pThis)); + } - ASSERT(pThis != NULL); - ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0); +finalize_it: + RETiRet; +} +#undef ER_TTYHUP + + +/* issue write() api calls until either the buffer is completely + * written or an error occured (it may happen that multiple writes + * are required, what is perfectly legal. On exit, *pLenBuf contains + * the number of bytes actually written. + * rgerhards, 2009-06-08 + */ +static rsRetVal +doWriteCall(strm_t *pThis, uchar *pBuf, size_t *pLenBuf) +{ + ssize_t lenBuf; + ssize_t iTotalWritten; + ssize_t iWritten; + char *pWriteBuf; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); + + lenBuf = *pLenBuf; + pWriteBuf = (char*) pBuf; + iTotalWritten = 0; + do { + iWritten = write(pThis->fd, pWriteBuf, lenBuf); + if(iWritten < 0) { + char errStr[1024]; + int err = errno; + rs_strerror_r(err, errStr, sizeof(errStr)); + DBGPRINTF("log file (%d) write error %d: %s\n", pThis->fd, err, errStr); + if(err == EINTR) { + /*NO ERROR, just continue */; + } else { + if(pThis->bIsTTY) { + CHKiRet(tryTTYRecover(pThis, err)); + } else { + ABORT_FINALIZE(RS_RET_IO_ERROR); + /* Would it make sense to cover more error cases? So far, I + * do not see good reason to do so. + */ + } + } + } + /* advance buffer to next write position */ + iTotalWritten += iWritten; + lenBuf -= iWritten; + pWriteBuf += iWritten; + } while(lenBuf > 0); /* Warning: do..while()! */ + +finalize_it: + *pLenBuf = iTotalWritten; + RETiRet; +} + + +/* sync the file to disk, so that any unwritten data is persisted. This + * also syncs the directory and thus makes sure that the file survives + * fatal failure. Note that we do NOT return an error status if the + * sync fails. Doing so would probably cause more trouble than it + * is worth (read: data loss may occur where we otherwise might not + * have it). -- rgerhards, 2009-06-08 + */ +static rsRetVal +syncFile(strm_t *pThis) +{ + int ret; + DEFiRet; + + if(pThis->bIsTTY) + FINALIZE; /* TTYs can not be synced */ + + DBGPRINTF("syncing file %d\n", pThis->fd); + ret = fdatasync(pThis->fd); + if(ret != 0) { + char errStr[1024]; + int err = errno; + rs_strerror_r(err, errStr, sizeof(errStr)); + DBGPRINTF("sync failed for file %d with error (%d): %s - ignoring\n", + pThis->fd, err, errStr); + } + + if(pThis->fdDir != -1) { + ret = fsync(pThis->fdDir); + } + +finalize_it: + RETiRet; +} + + +/* physically write to the output file. the provided data is ready for + * writing (e.g. zipped if we are requested to do that). + * Note that if the write() API fails, we do not reset any pointers, but return + * an error code. That means we may redo work in the next iteration. + * rgerhards, 2009-06-04 + */ +static rsRetVal +strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +{ + size_t iWritten; + DEFiRet; + ISOBJ_TYPE_assert(pThis, strm); if(pThis->fd == -1) CHKiRet(strmOpenFile(pThis)); - iWritten = write(pThis->fd, pBuf, lenBuf); - dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, iWritten); - /* TODO: handle error case -- rgerhards, 2008-01-07 */ - - /* Now indicate buffer empty again. We do this in any case, because there - * is no way we could react more intelligently to an error during write. - * This MUST be done BEFORE strCheckNextOutputFile(), otherwise we have an - * endless loop. We reset the buffer pointer also in finalize_it - this is - * necessary if we run into problems. Not resetting it would again cause an - * endless loop. So it is better to loose some data (which also justifies - * duplicating that code, too...) -- rgerhards, 2008-01-10 - */ + iWritten = lenBuf; + CHKiRet(doWriteCall(pThis, pBuf, &iWritten)); + dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, (int) iWritten); + pThis->iBufPtr = 0; pThis->iCurrOffs += iWritten; /* update user counter, if provided */ if(pThis->pUsrWCntr != NULL) *pThis->pUsrWCntr += iWritten; - if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) + if(pThis->bSync) { + CHKiRet(syncFile(pThis)); + } + + if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) { CHKiRet(strmCheckNextOutputFile(pThis)); + } else if(pThis->iSizeLimit != 0) { + CHKiRet(doSizeLimitProcessing(pThis)); + } finalize_it: - pThis->iBufPtr = 0; /* see comment above */ + RETiRet; +} + +/* write the output buffer in zip mode + * This means we compress it first and then do a physical write. + * Note that we always do a full deflateInit ... deflate ... deflateEnd + * sequence. While this is not optimal, we need to do it because we need + * to ensure that the file is readable even when we are aborted. Doing the + * full sequence brings us as far towards this goal as possible (and not + * doing it would be a total failure). It may be worth considering to + * add a config switch so that the user can decide the risk he is ready + * to take, but so far this is not yet implemented (not even requested ;)). + * rgerhards, 2009-06-04 + */ +static rsRetVal +doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +{ + z_stream zstrm; + int zRet; /* zlib return state */ + DEFiRet; + assert(pThis != NULL); + assert(pBuf != NULL); + + /* allocate deflate state */ + zstrm.zalloc = Z_NULL; + zstrm.zfree = Z_NULL; + zstrm.opaque = Z_NULL; + /* see note in file header for the params we use with deflateInit2() */ + zRet = zlibw.DeflateInit2(&zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY); + if(zRet != Z_OK) { + dbgprintf("error %d returned from zlib/deflateInit2()\n", zRet); + ABORT_FINALIZE(RS_RET_ZLIB_ERR); + } + + /* now doing the compression */ + zstrm.avail_in = lenBuf; + zstrm.next_in = (Bytef*) pBuf; + /* run deflate() on input until output buffer not full, finish + compression if all of source has been read in */ + do { + dbgprintf("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in); + zstrm.avail_out = pThis->sIOBufSize; + zstrm.next_out = pThis->pZipBuf; + zRet = zlibw.Deflate(&zstrm, Z_FINISH); /* no bad return value */ + dbgprintf("after deflate, ret %d, avail_out %d\n", zRet, zstrm.avail_out); + assert(zRet != Z_STREAM_ERROR); /* state not clobbered */ + CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, pThis->sIOBufSize - zstrm.avail_out)); + } while (zstrm.avail_out == 0); + assert(zstrm.avail_in == 0); /* all input will be used */ + + + zRet = zlibw.DeflateEnd(&zstrm); + if(zRet != Z_OK) { + dbgprintf("error %d returned from zlib/deflateEnd()\n", zRet); + ABORT_FINALIZE(RS_RET_ZLIB_ERR); + } + +finalize_it: + RETiRet; +} + + +/* write memory buffer to a stream object. + * To support direct writes of large objects, this method may be called + * with a buffer pointing to some region other than the stream buffer itself. + * However, in that case the stream buffer must be empty (strmFlush() has to + * be called before), because we would otherwise mess up with the sequence + * inside the stream. -- rgerhards, 2008-01-10 + */ +static rsRetVal +strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf) +{ + DEFiRet; + + ASSERT(pThis != NULL); + ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0); + + if(pThis->iZipLevel) { + CHKiRet(doZipWrite(pThis, pBuf, lenBuf)); + } else { + /* write without zipping */ + CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf)); + } + +finalize_it: RETiRet; } @@ -503,14 +901,15 @@ finalize_it: * and is automatically called when the output buffer is full. * rgerhards, 2008-01-10 */ -rsRetVal strmFlush(strm_t *pThis) +static rsRetVal +strmFlush(strm_t *pThis) { DEFiRet; ASSERT(pThis != NULL); dbgoprint((obj_t*) pThis, "file %d flush, buflen %ld\n", pThis->fd, (long) pThis->iBufPtr); - if(pThis->tOperationsMode == STREAMMODE_WRITE && pThis->iBufPtr > 0) { + if(pThis->tOperationsMode != STREAMMODE_READ && pThis->iBufPtr > 0) { iRet = strmWriteInternal(pThis, pThis->pIOBuf, pThis->iBufPtr); } @@ -545,7 +944,7 @@ static rsRetVal strmSeek(strm_t *pThis, off_t offs) /* seek to current offset. This is primarily a helper to readjust the OS file * pointer after a strm object has been deserialized. */ -rsRetVal strmSeekCurrOffs(strm_t *pThis) +static rsRetVal strmSeekCurrOffs(strm_t *pThis) { DEFiRet; @@ -558,7 +957,7 @@ rsRetVal strmSeekCurrOffs(strm_t *pThis) /* write a *single* character to a stream object -- rgerhards, 2008-01-10 */ -rsRetVal strmWriteChar(strm_t *pThis, uchar c) +static rsRetVal strmWriteChar(strm_t *pThis, uchar c) { DEFiRet; @@ -578,7 +977,7 @@ finalize_it: /* write an integer value (actually a long) to a stream object */ -rsRetVal strmWriteLong(strm_t *pThis, long i) +static rsRetVal strmWriteLong(strm_t *pThis, long i) { DEFiRet; uchar szBuf[32]; @@ -593,16 +992,48 @@ finalize_it: } +/* schedule an Apc flush request. + * rgerhards, 2009-06-15 + */ +static inline rsRetVal +scheduleFlushRequest(strm_t *pThis) +{ + apc_t *pApc; + DEFiRet; + + CHKiRet(apc.CancelApc(pThis->apcID)); +dbgprintf("XXX: requesting to add apc!\n"); + CHKiRet(apc.Construct(&pApc)); + CHKiRet(apc.SetProcedure(pApc, (void (*)(void*, void*))flushApc)); + CHKiRet(apc.SetParam1(pApc, pThis)); + CHKiRet(apc.ConstructFinalize(pApc, &pThis->apcID)); + +finalize_it: + RETiRet; +} + + /* write memory buffer to a stream object */ -rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) +static rsRetVal +strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) { + DEFVARS_mutexProtection_uncond; DEFiRet; size_t iPartial; ASSERT(pThis != NULL); ASSERT(pBuf != NULL); +dbgprintf("strmWrite(%p, '%65.65s', %ld);, disabled %d, sizelim %ld, size %lld\n", pThis, pBuf,lenBuf, pThis->bDisabled, pThis->iSizeLimit, pThis->iCurrOffs); + if(pThis->bDisabled) + ABORT_FINALIZE(RS_RET_STREAM_DISABLED); + +RUNLOG_VAR("%d", pThis->iFlushInterval); + if(pThis->iFlushInterval != 0) { + BEGIN_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); + } + /* check if the to-be-written data is larger than our buffer size */ if(lenBuf >= pThis->sIOBufSize) { /* it is - so we do a direct write, that is most efficient. @@ -631,7 +1062,17 @@ rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf) } } + /* we ignore the outcome of scheduleFlushRequest(), as we will write the data always at + * termination. For Zip mode, it could be fatal if we write after each record. + */ + if(pThis->iFlushInterval != 0) + scheduleFlushRequest(pThis); + finalize_it: + if(pThis->iFlushInterval != 0) { + END_MTX_PROTECTED_OPERATIONS_UNCOND(&pThis->mut); + } + RETiRet; } @@ -644,34 +1085,27 @@ DEFpropSetMeth(strm, iFileNumDigits, int) DEFpropSetMeth(strm, tOperationsMode, int) DEFpropSetMeth(strm, tOpenMode, mode_t) DEFpropSetMeth(strm, sType, strmType_t) - -rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal) +DEFpropSetMeth(strm, iZipLevel, int) +DEFpropSetMeth(strm, bSync, int) +DEFpropSetMeth(strm, sIOBufSize, size_t) +DEFpropSetMeth(strm, iSizeLimit, off_t) +DEFpropSetMeth(strm, iFlushInterval, int) +DEFpropSetMeth(strm, pszSizeLimitCmd, uchar*) + +static rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal) { pThis->iMaxFiles = iNewVal; pThis->iFileNumDigits = getNumberDigits(iNewVal); return RS_RET_OK; } -rsRetVal strmSetiAddtlOpenFlags(strm_t *pThis, int iNewVal) -{ - DEFiRet; - - if(iNewVal & O_APPEND) - ABORT_FINALIZE(RS_RET_PARAM_ERROR); - - pThis->iAddtlOpenFlags = iNewVal; - -finalize_it: - RETiRet; -} - /* set the stream's file prefix * The passed-in string is duplicated. So if the caller does not need * it any longer, it must free it. * rgerhards, 2008-01-09 */ -rsRetVal +static rsRetVal strmSetFName(strm_t *pThis, uchar *pszName, size_t iLenName) { DEFiRet; @@ -685,7 +1119,7 @@ strmSetFName(strm_t *pThis, uchar *pszName, size_t iLenName) if(pThis->pszFName != NULL) free(pThis->pszFName); - if((pThis->pszFName = malloc(sizeof(uchar) * iLenName + 1)) == NULL) + if((pThis->pszFName = malloc(sizeof(uchar) * (iLenName + 1))) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); memcpy(pThis->pszFName, pszName, iLenName + 1); /* always think about the \0! */ @@ -701,7 +1135,7 @@ finalize_it: * it any longer, it must free it. * rgerhards, 2008-01-09 */ -rsRetVal +static rsRetVal strmSetDir(strm_t *pThis, uchar *pszDir, size_t iLenDir) { DEFiRet; @@ -745,7 +1179,7 @@ finalize_it: * * rgerhards, 2008-01-10 */ -rsRetVal strmRecordBegin(strm_t *pThis) +static rsRetVal strmRecordBegin(strm_t *pThis) { ASSERT(pThis != NULL); ASSERT(pThis->bInRecord == 0); @@ -753,7 +1187,7 @@ rsRetVal strmRecordBegin(strm_t *pThis) return RS_RET_OK; } -rsRetVal strmRecordEnd(strm_t *pThis) +static rsRetVal strmRecordEnd(strm_t *pThis) { DEFiRet; ASSERT(pThis != NULL); @@ -775,7 +1209,7 @@ rsRetVal strmRecordEnd(strm_t *pThis) * We do not serialize the dynamic properties. * rgerhards, 2008-01-10 */ -rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm) +static rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm) { DEFiRet; int i; @@ -821,7 +1255,7 @@ finalize_it: * any new set overwrites the previous one. * rgerhards, 2008-02-27 */ -rsRetVal +static rsRetVal strmSetWCntr(strm_t *pThis, number_t *pWCnt) { DEFiRet; @@ -841,8 +1275,8 @@ strmSetWCntr(strm_t *pThis, number_t *pWCnt) /* This function can be used as a generic way to set properties. * rgerhards, 2008-01-11 */ -#define isProp(name) !rsCStrSzStrCmp(pProp->pcsName, (uchar*) name, sizeof(name) - 1) -rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp) +#define isProp(name) !rsCStrSzStrCmp(pProp->pcsName, UCHAR_CONSTANT(name), sizeof(name) - 1) +static rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp) { DEFiRet; @@ -881,7 +1315,7 @@ finalize_it: * reported on the second call may actually be lower than on the first call. This is due to * file circulation. A caller must deal with that. -- rgerhards, 2008-01-30 */ -rsRetVal +static rsRetVal strmGetCurrOffset(strm_t *pThis, int64 *pOffs) { DEFiRet; @@ -909,8 +1343,38 @@ CODESTARTobjQueryInterface(strm) * work here (if we can support an older interface version - that, * of course, also affects the "if" above). */ - /*xxxpIf->oID = OBJvm; SAMPLE */ - + pIf->Construct = strmConstruct; + pIf->ConstructFinalize = strmConstructFinalize; + pIf->Destruct = strmDestruct; + pIf->ReadChar = strmReadChar; + pIf->UnreadChar = strmUnreadChar; + pIf->ReadLine = strmReadLine; + pIf->SeekCurrOffs = strmSeekCurrOffs; + pIf->Write = strmWrite; + pIf->WriteChar = strmWriteChar; + pIf->WriteLong = strmWriteLong; + pIf->SetFName = strmSetFName; + pIf->SetDir = strmSetDir; + pIf->Flush = strmFlush; + pIf->RecordBegin = strmRecordBegin; + pIf->RecordEnd = strmRecordEnd; + pIf->Serialize = strmSerialize; + pIf->GetCurrOffset = strmGetCurrOffset; + pIf->SetWCntr = strmSetWCntr; + /* set methods */ + pIf->SetbDeleteOnClose = strmSetbDeleteOnClose; + pIf->SetiMaxFileSize = strmSetiMaxFileSize; + pIf->SetiMaxFiles = strmSetiMaxFiles; + pIf->SetiFileNumDigits = strmSetiFileNumDigits; + pIf->SettOperationsMode = strmSettOperationsMode; + pIf->SettOpenMode = strmSettOpenMode; + pIf->SetsType = strmSetsType; + pIf->SetiZipLevel = strmSetiZipLevel; + pIf->SetbSync = strmSetbSync; + pIf->SetsIOBufSize = strmSetsIOBufSize; + pIf->SetiSizeLimit = strmSetiSizeLimit; + pIf->SetiFlushInterval = strmSetiFlushInterval; + pIf->SetpszSizeLimitCmd = strmSetpszSizeLimitCmd; finalize_it: ENDobjQueryInterface(strm) @@ -921,13 +1385,12 @@ ENDobjQueryInterface(strm) */ BEGINObjClassInit(strm, 1, OBJ_IS_CORE_MODULE) /* request objects we use */ + CHKiRet(objUse(apc, CORE_COMPONENT)); OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize); OBJSetMethodHandler(objMethod_SETPROPERTY, strmSetProperty); OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmConstructFinalize); ENDObjClassInit(strm) - -/* - * vi:set ai: +/* vi:set ai: */ diff --git a/runtime/stream.h b/runtime/stream.h index 371358ab..e3ad32b1 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -19,7 +19,29 @@ * can easily be persistet. The bottom line is that it makes much sense to * use this class whereever possible as its features may grow in the future. * - * Copyright 2008 Rainer Gerhards and Adiscon GmbH. + * An important note on writing gzip format via zlib (kept anonymous + * by request): + * + * -------------------------------------------------------------------------- + * We'd like to make sure the output file is in full gzip format + * (compatible with gzip -d/zcat etc). There is a flag in how the output + * is initialized within zlib to properly add the gzip wrappers to the + * output. (gzip is effectively a small metadata wrapper around raw + * zstream output.) + * + * I had written an old bit of code to do this - the documentation on + * deflatInit2() was pretty tricky to nail down on this specific feature: + * + * int deflateInit2 (z_streamp strm, int level, int method, int windowBits, + * int memLevel, int strategy); + * + * I believe "31" would be the value for the "windowBits" field that you'd + * want to try: + * + * deflateInit2(zstrmptr, 6, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY); + * -------------------------------------------------------------------------- + * + * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH. * * This file is part of the rsyslog runtime library. * @@ -47,6 +69,8 @@ #include "obj-types.h" #include "glbl.h" #include "stream.h" +#include "zlibw.h" +#include "apc.h" /* stream types */ typedef enum { @@ -55,10 +79,12 @@ typedef enum { STREAMTYPE_FILE_MONITOR = 2 /**< monitor a (third-party) file */ } strmType_t; -typedef enum { +typedef enum { /* when extending, do NOT change existing modes! */ STREAMMMODE_INVALID = 0, STREAMMODE_READ = 1, - STREAMMODE_WRITE = 2 + STREAMMODE_WRITE = 2, + STREAMMODE_WRITE_TRUNC = 3, + STREAMMODE_WRITE_APPEND = 4 } strmMode_t; /* The strm_t data structure */ @@ -71,61 +97,78 @@ typedef struct strm_s { int lenFName; strmMode_t tOperationsMode; mode_t tOpenMode; - int iAddtlOpenFlags; /* can be used to specifiy additional (compatible!) open flags */ int64 iMaxFileSize;/* maximum size a file may grow to */ int iMaxFiles; /* maximum number of files if a circular mode is in use */ int iFileNumDigits;/* min number of digits to use in file number (only in circular mode) */ - int bDeleteOnClose; /* set to 1 to auto-delete on close -- be careful with that setting! */ + bool bDeleteOnClose; /* set to 1 to auto-delete on close -- be careful with that setting! */ int64 iCurrOffs;/* current offset */ int64 *pUsrWCntr; /* NULL or a user-provided counter that receives the nbr of bytes written since the last CntrSet() */ /* dynamic properties, valid only during file open, not to be persistet */ - size_t sIOBufSize;/* size of IO buffer */ + int bDisabled; /* should file no longer be written to? (currently set only if omfile file size limit fails) */ + int bSync; /* sync this file after every write? */ + size_t sIOBufSize;/* size of IO buffer */ uchar *pszDir; /* Directory */ int lenDir; int fd; /* the file descriptor, -1 if closed */ + int fdDir; /* the directory's descriptor, in case bSync is requested (-1 if closed) */ uchar *pszCurrFName; /* name of current file (if open) */ uchar *pIOBuf; /* io Buffer */ size_t iBufPtrMax; /* current max Ptr in Buffer (if partial read!) */ size_t iBufPtr; /* pointer into current buffer */ int iUngetC; /* char set via UngetChar() call or -1 if none set */ - int bInRecord; /* if 1, indicates that we are currently writing a not-yet complete record */ + bool bInRecord; /* if 1, indicates that we are currently writing a not-yet complete record */ + int iZipLevel; /* zip level (0..9). If 0, zip is completely disabled */ + Bytef *pZipBuf; + /* support for async flush procesing */ + int iFlushInterval; /* flush in which interval - 0, no flushing */ + apc_id_t apcID; /* id of current Apc request (used for cancelling) */ + pthread_mutex_t mut;/* mutex for flush in async mode */ + /* support for omfile size-limiting commands, special counters, NOT persisted! */ + off_t iSizeLimit; /* file size limit, 0 = no limit */ + uchar *pszSizeLimitCmd; /* command to carry out when size limit is reached */ + bool bIsTTY; /* is this a tty file? */ } strm_t; /* interfaces */ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ + rsRetVal (*Construct)(strm_t **ppThis); + rsRetVal (*ConstructFinalize)(strm_t *pThis); + rsRetVal (*Destruct)(strm_t **ppThis); + rsRetVal (*SetMaxFileSize)(strm_t *pThis, int64 iMaxFileSize); + rsRetVal (*SetFileName)(strm_t *pThis, uchar *pszName, size_t iLenName); + rsRetVal (*ReadChar)(strm_t *pThis, uchar *pC); + rsRetVal (*UnreadChar)(strm_t *pThis, uchar c); + rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr); + rsRetVal (*SeekCurrOffs)(strm_t *pThis); + rsRetVal (*Write)(strm_t *pThis, uchar *pBuf, size_t lenBuf); + rsRetVal (*WriteChar)(strm_t *pThis, uchar c); + rsRetVal (*WriteLong)(strm_t *pThis, long i); + rsRetVal (*SetFName)(strm_t *pThis, uchar *pszPrefix, size_t iLenPrefix); + rsRetVal (*SetDir)(strm_t *pThis, uchar *pszDir, size_t iLenDir); + rsRetVal (*Flush)(strm_t *pThis); + rsRetVal (*RecordBegin)(strm_t *pThis); + rsRetVal (*RecordEnd)(strm_t *pThis); + rsRetVal (*Serialize)(strm_t *pThis, strm_t *pStrm); + rsRetVal (*GetCurrOffset)(strm_t *pThis, int64 *pOffs); + rsRetVal (*SetWCntr)(strm_t *pThis, number_t *pWCnt); + INTERFACEpropSetMeth(strm, bDeleteOnClose, int); + INTERFACEpropSetMeth(strm, iMaxFileSize, int); + INTERFACEpropSetMeth(strm, iMaxFiles, int); + INTERFACEpropSetMeth(strm, iFileNumDigits, int); + INTERFACEpropSetMeth(strm, tOperationsMode, int); + INTERFACEpropSetMeth(strm, tOpenMode, mode_t); + INTERFACEpropSetMeth(strm, sType, strmType_t); + INTERFACEpropSetMeth(strm, iZipLevel, int); + INTERFACEpropSetMeth(strm, bSync, int); + INTERFACEpropSetMeth(strm, sIOBufSize, size_t); + INTERFACEpropSetMeth(strm, iSizeLimit, off_t); + INTERFACEpropSetMeth(strm, iFlushInterval, int); + INTERFACEpropSetMeth(strm, pszSizeLimitCmd, uchar*); ENDinterface(strm) -#define strmCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ +#define strmCURR_IF_VERSION 2 /* increment whenever you change the interface structure! */ /* prototypes */ -rsRetVal strmConstruct(strm_t **ppThis); -rsRetVal strmConstructFinalize(strm_t __attribute__((unused)) *pThis); -rsRetVal strmDestruct(strm_t **ppThis); -rsRetVal strmSetMaxFileSize(strm_t *pThis, int64 iMaxFileSize); -rsRetVal strmSetFileName(strm_t *pThis, uchar *pszName, size_t iLenName); -rsRetVal strmReadChar(strm_t *pThis, uchar *pC); -rsRetVal strmUnreadChar(strm_t *pThis, uchar c); -rsRetVal strmReadLine(strm_t *pThis, cstr_t **ppCStr); -rsRetVal strmSeekCurrOffs(strm_t *pThis); -rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); -rsRetVal strmWriteChar(strm_t *pThis, uchar c); -rsRetVal strmWriteLong(strm_t *pThis, long i); -rsRetVal strmSetFName(strm_t *pThis, uchar *pszPrefix, size_t iLenPrefix); -rsRetVal strmSetDir(strm_t *pThis, uchar *pszDir, size_t iLenDir); -rsRetVal strmFlush(strm_t *pThis); -rsRetVal strmRecordBegin(strm_t *pThis); -rsRetVal strmRecordEnd(strm_t *pThis); -rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm); -rsRetVal strmSetiAddtlOpenFlags(strm_t *pThis, int iNewVal); -rsRetVal strmGetCurrOffset(strm_t *pThis, int64 *pOffs); -rsRetVal strmSetWCntr(strm_t *pThis, number_t *pWCnt); PROTOTYPEObjClassInit(strm); -PROTOTYPEpropSetMeth(strm, bDeleteOnClose, int); -PROTOTYPEpropSetMeth(strm, iMaxFileSize, int); -PROTOTYPEpropSetMeth(strm, iMaxFiles, int); -PROTOTYPEpropSetMeth(strm, iFileNumDigits, int); -PROTOTYPEpropSetMeth(strm, tOperationsMode, int); -PROTOTYPEpropSetMeth(strm, tOpenMode, mode_t); -PROTOTYPEpropSetMeth(strm, sType, strmType_t); #endif /* #ifndef STREAM_H_INCLUDED */ diff --git a/runtime/sysvar.c b/runtime/sysvar.c index c102d1f5..4a6ace19 100644 --- a/runtime/sysvar.c +++ b/runtime/sysvar.c @@ -175,8 +175,6 @@ CODESTARTobjQueryInterface(sysvar) * work here (if we can support an older interface version - that, * of course, also affects the "if" above). */ - //xxxpIf->oID = "sysvar";//OBJsysvar; - pIf->Construct = sysvarConstruct; pIf->ConstructFinalize = sysvarConstructFinalize; pIf->Destruct = sysvarDestruct; diff --git a/runtime/unicode-helper.h b/runtime/unicode-helper.h index 36d76a78..7a776f68 100644 --- a/runtime/unicode-helper.h +++ b/runtime/unicode-helper.h @@ -4,6 +4,9 @@ * The following functions are wrappers which hopefully enable us to move * from 8-bit chars to unicode with relative ease when we finally attack this * + * Note: while we prefer inline functions, this leads to invalid references in + * core dumps. So in a debug build, we use macros where appropriate... + * * Begun 2009-05-21 RGerhards * * Copyright (C) 2009 by Rainer Gerhards and Adiscon GmbH @@ -31,6 +34,22 @@ #include <string.h> +#ifdef DEBUG +# define ustrncpy(psz1, psz2, len) strncpy((char*)(psz1), (char*)(psz2), (len)) +# define ustrdup(psz) (uchar*)strdup((char*)(psz)) +#else + static inline uchar* ustrncpy(uchar *psz1, uchar *psz2, size_t len) + { + return (uchar*) strncpy((char*) psz1, (char*) psz2, len); + } + + static inline uchar* ustrdup(uchar *psz) + { + return (uchar*) strdup((char*)psz); + } + +#endif /* #ifdef DEBUG */ + static inline int ustrcmp(uchar *psz1, uchar *psz2) { return strcmp((char*) psz1, (char*) psz2); @@ -41,13 +60,9 @@ static inline int ustrlen(uchar *psz) return strlen((char*) psz); } -static inline uchar* ustrdup(uchar *psz) -{ - return (uchar*) strdup((char*)psz); -} - #define UCHAR_CONSTANT(x) ((uchar*) (x)) +#define CHAR_CONVERT(x) ((char*) (x)) #endif /* multi-include protection */ /* vim:set ai: diff --git a/runtime/wti.c b/runtime/wti.c index 544bffa7..18767ea1 100644 --- a/runtime/wti.c +++ b/runtime/wti.c @@ -201,8 +201,7 @@ CODESTARTobjDestruct(wti) pthread_cond_destroy(&pThis->condExitDone); pthread_mutex_destroy(&pThis->mut); - if(pThis->pszDbgHdr != NULL) - free(pThis->pszDbgHdr); + free(pThis->pszDbgHdr); ENDobjDestruct(wti) diff --git a/runtime/wtp.c b/runtime/wtp.c index 04eb974f..df39daa3 100644 --- a/runtime/wtp.c +++ b/runtime/wtp.c @@ -151,8 +151,7 @@ CODESTARTobjDestruct(wtp) pthread_mutex_destroy(&pThis->mut); pthread_mutex_destroy(&pThis->mutThrdShutdwn); - if(pThis->pszDbgHdr != NULL) - free(pThis->pszDbgHdr); + free(pThis->pszDbgHdr); ENDobjDestruct(wtp) diff --git a/runtime/zlibw.c b/runtime/zlibw.c new file mode 100644 index 00000000..2b386213 --- /dev/null +++ b/runtime/zlibw.c @@ -0,0 +1,125 @@ +/* The zlibwrap object. + * + * This is an rsyslog object wrapper around zlib. + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ + +#include "config.h" +#include <string.h> +#include <assert.h> +#include <zlib.h> + +#include "rsyslog.h" +#include "module-template.h" +#include "obj.h" +#include "zlibw.h" + +MODULE_TYPE_LIB + +/* static data */ +DEFobjStaticHelpers + + +/* ------------------------------ methods ------------------------------ */ + +/* zlib make strong use of macros for its interface functions, so we can not simply + * pass function pointers to them. Instead, we create very small wrappers which call + * the relevant entry points. + */ + +static int myDeflateInit(z_streamp strm, int level) +{ + return deflateInit(strm, level); +} + +static int myDeflateInit2(z_streamp strm, int level, int method, int windowBits, int memLevel, int strategy) +{ + return deflateInit2(strm, level, method, windowBits, memLevel, strategy); +} + +static int myDeflateEnd(z_streamp strm) +{ + return deflateEnd(strm); +} + +static int myDeflate(z_streamp strm, int flush) +{ + return deflate(strm, flush); +} + + +/* queryInterface function + * rgerhards, 2008-03-05 + */ +BEGINobjQueryInterface(zlibw) +CODESTARTobjQueryInterface(zlibw) + if(pIf->ifVersion != zlibwCURR_IF_VERSION) { /* check for current version, increment on each change */ + ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED); + } + + /* ok, we have the right interface, so let's fill it + * Please note that we may also do some backwards-compatibility + * work here (if we can support an older interface version - that, + * of course, also affects the "if" above). + */ + pIf->DeflateInit = myDeflateInit; + pIf->DeflateInit2 = myDeflateInit2; + pIf->Deflate = myDeflate; + pIf->DeflateEnd = myDeflateEnd; +finalize_it: +ENDobjQueryInterface(zlibw) + + +/* Initialize the zlibw class. Must be called as the very first method + * before anything else is called inside this class. + * rgerhards, 2008-02-19 + */ +BEGINAbstractObjClassInit(zlibw, 1, OBJ_IS_LOADABLE_MODULE) /* class, version */ + /* request objects we use */ + + /* set our own handlers */ +ENDObjClassInit(zlibw) + + +/* --------------- here now comes the plumbing that makes as a library module --------------- */ + + +BEGINmodExit +CODESTARTmodExit +ENDmodExit + + +BEGINqueryEtryPt +CODESTARTqueryEtryPt +CODEqueryEtryPt_STD_LIB_QUERIES +ENDqueryEtryPt + + +BEGINmodInit() +CODESTARTmodInit + *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ + + CHKiRet(zlibwClassInit(pModInfo)); /* must be done after tcps_sess, as we use it */ + /* Initialize all classes that are in our module - this includes ourselfs */ +ENDmodInit +/* vi:set ai: + */ diff --git a/runtime/zlibw.h b/runtime/zlibw.h new file mode 100644 index 00000000..63d8f386 --- /dev/null +++ b/runtime/zlibw.h @@ -0,0 +1,46 @@ +/* The zlibw object. It encapsulates the zlib functionality. The primary + * purpose of this wrapper class is to enable rsyslogd core to be build without + * zlib libraries. + * + * Copyright 2009 Rainer Gerhards and Adiscon GmbH. + * + * This file is part of the rsyslog runtime library. + * + * The rsyslog runtime library is free software: you can redistribute it and/or modify + * it under the terms of the GNU Lesser General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * The rsyslog runtime library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution. + */ +#ifndef INCLUDED_ZLIBW_H +#define INCLUDED_ZLIBW_H + +#include <zlib.h> + +/* interfaces */ +BEGINinterface(zlibw) /* name must also be changed in ENDinterface macro! */ + int (*DeflateInit)(z_streamp strm, int); + int (*DeflateInit2)(z_streamp strm, int level, int method, int windowBits, int memLevel, int strategy); + int (*Deflate)(z_streamp strm, int); + int (*DeflateEnd)(z_streamp strm); +ENDinterface(zlibw) +#define zlibwCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */ + + +/* prototypes */ +PROTOTYPEObj(zlibw); + +/* the name of our library binary */ +#define LM_ZLIBW_FILENAME "lmzlibw" + +#endif /* #ifndef INCLUDED_ZLIBW_H */ |