diff options
Diffstat (limited to 'plugins')
29 files changed, 1955 insertions, 283 deletions
diff --git a/plugins/imdiag/imdiag.c b/plugins/imdiag/imdiag.c index 640c9e1b..5fdc6ef1 100644 --- a/plugins/imdiag/imdiag.c +++ b/plugins/imdiag/imdiag.c @@ -53,6 +53,7 @@ #include "srUtils.h" #include "msg.h" #include "datetime.h" +#include "ratelimit.h" #include "net.h" /* for permittedPeers, may be removed when this is removed */ MODULE_TYPE_INPUT @@ -199,7 +200,7 @@ finalize_it: /* actually submit a message to the rsyslog core */ static rsRetVal -doInjectMsg(int iNum) +doInjectMsg(int iNum, ratelimit_t *ratelimiter) { uchar szMsg[1024]; msg_t *pMsg; @@ -219,7 +220,7 @@ doInjectMsg(int iNum) pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME; MsgSetRcvFrom(pMsg, pRcvDummy); CHKiRet(MsgSetRcvFromIP(pMsg, pRcvIPDummy)); - CHKiRet(submitMsg(pMsg)); + CHKiRet(ratelimitAddMsg(ratelimiter, NULL, pMsg)); finalize_it: RETiRet; @@ -237,6 +238,7 @@ injectMsg(uchar *pszCmd, tcps_sess_t *pSess) int iFrom; int nMsgs; int i; + ratelimit_t *ratelimit; DEFiRet; /* we do not check errors here! */ @@ -244,13 +246,15 @@ injectMsg(uchar *pszCmd, tcps_sess_t *pSess) iFrom = atoi((char*)wordBuf); getFirstWord(&pszCmd, wordBuf, sizeof(wordBuf)/sizeof(uchar), TO_LOWERCASE); nMsgs = atoi((char*)wordBuf); + ratelimitNew(&ratelimit, "imdiag", "injectmsg"); for(i = 0 ; i < nMsgs ; ++i) { - doInjectMsg(i + iFrom); + doInjectMsg(i + iFrom, ratelimit); } CHKiRet(sendResponse(pSess, "%d messages injected\n", nMsgs)); DBGPRINTF("imdiag: %d messages injected\n", nMsgs); + ratelimitDestruct(ratelimit); finalize_it: RETiRet; diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c index 188d692b..0f155c10 100644 --- a/plugins/imfile/imfile.c +++ b/plugins/imfile/imfile.c @@ -48,6 +48,7 @@ #include "prop.h" #include "stringbuf.h" #include "ruleset.h" +#include "ratelimit.h" MODULE_TYPE_INPUT /* must be present for input modules, do not remove */ MODULE_TYPE_NOKEEP @@ -82,6 +83,7 @@ typedef struct fileInfo_s { strm_t *pStrm; /* its stream (NULL if not assigned) */ int readMode; /* which mode to use in ReadMulteLine call? */ ruleset_t *pRuleset; /* ruleset to bind listener to (use system default if unspecified) */ + ratelimit_t *ratelimiter; multi_submit_t multiSub; } fileInfo_t; @@ -189,9 +191,7 @@ static rsRetVal enqLine(fileInfo_t *pInfo, cstr_t *cstrLine) pMsg->iFacility = LOG_FAC(pInfo->iFacility); pMsg->iSeverity = LOG_PRI(pInfo->iSeverity); MsgSetRuleset(pMsg, pInfo->pRuleset); - pInfo->multiSub.ppMsgs[pInfo->multiSub.nElem++] = pMsg; - if(pInfo->multiSub.nElem == pInfo->multiSub.maxElem) - CHKiRet(multiSubmitMsg(&pInfo->multiSub)); + ratelimitAddMsg(pInfo->ratelimiter, &pInfo->multiSub, pMsg); finalize_it: RETiRet; } @@ -246,6 +246,8 @@ finalize_it: strm.Destruct(&psSF); if(iRet != RS_RET_OK) { + if(pThis->pStrm != NULL) + strm.Destruct(&pThis->pStrm); CHKiRet(strm.Construct(&pThis->pStrm)); CHKiRet(strm.SettOperationsMode(pThis->pStrm, STREAMMODE_READ)); CHKiRet(strm.SetsType(pThis->pStrm, STREAMTYPE_FILE_MONITOR)); @@ -304,10 +306,7 @@ static rsRetVal pollFile(fileInfo_t *pThis, int *pbHadFileData) } finalize_it: - if(pThis->multiSub.nElem > 0) { - /* submit everything that was not yet submitted */ - CHKiRet(multiSubmitMsg(&pThis->multiSub)); - } + multiSubmitFlush(&pThis->multiSub); pthread_cleanup_pop(0); if(pCStr != NULL) { @@ -415,6 +414,7 @@ addListner(instanceConf_t *inst) pThis->lenTag = ustrlen(pThis->pszTag); pThis->pszStateFile = (uchar*) strdup((char*) inst->pszStateFile); + CHKiRet(ratelimitNew(&pThis->ratelimiter, "imfile", (char*)inst->pszFileName)); CHKmalloc(pThis->multiSub.ppMsgs = MALLOC(inst->nMultiSub * sizeof(msg_t*))); pThis->multiSub.maxElem = inst->nMultiSub; pThis->multiSub.nElem = 0; @@ -765,6 +765,8 @@ CODESTARTafterRun persistStrmState(&files[i]); strm.Destruct(&(files[i].pStrm)); } + ratelimitDestruct(files[i].ratelimiter); + free(files[i].multiSub.ppMsgs); free(files[i].pszFileName); free(files[i].pszTag); free(files[i].pszStateFile); diff --git a/plugins/imklog/bsd.c b/plugins/imklog/bsd.c index cddc6737..0fbee491 100644 --- a/plugins/imklog/bsd.c +++ b/plugins/imklog/bsd.c @@ -58,9 +58,6 @@ static int fklog = -1; /* kernel log fd */ #ifdef OS_LINUX /* submit a message to imklog Syslog() API. In this function, we check if * a kernel timestamp is present and, if so, extract and strip it. - * Note: this is an extra processing step. We should revisit the whole - * idea in v6 and remove all that old stuff that we do not longer need - * (like symbol resolution). <-- TODO * Note that this is heavily Linux specific and thus is not compiled or * used for BSD. * Special thanks to Lennart Poettering for suggesting on how to convert diff --git a/plugins/imklog/imklog.c b/plugins/imklog/imklog.c index a24fc63b..810ac264 100644 --- a/plugins/imklog/imklog.c +++ b/plugins/imklog/imklog.c @@ -95,7 +95,7 @@ static struct cnfparamdescr modpdescr[] = { { "permitnonkernelfacility", eCmdHdlrBinary, 0 }, { "consoleloglevel", eCmdHdlrInt, 0 }, { "parsekerneltimestamp", eCmdHdlrBinary, 0 }, - { "keepkerneltimestamp", eCmdHdlrBinary, 0 }, + { "keepkerneltimestamp", eCmdHdlrBinary, 0 }, { "internalmsgfacility", eCmdHdlrFacility, 0 } }; static struct cnfparamblk modpblk = @@ -105,7 +105,7 @@ static struct cnfparamblk modpblk = }; static prop_t *pInputName = NULL; /* there is only one global inputName for all messages generated by this module */ -static prop_t *pLocalHostIP = NULL; /* a pseudo-constant propterty for 127.0.0.1 */ +static prop_t *pLocalHostIP = NULL; static inline void initConfigSettings(void) @@ -150,7 +150,8 @@ enqMsg(uchar *msg, uchar* pszTag, int iFacility, int iSeverity, struct timeval * MsgSetTAG(pMsg, pszTag, ustrlen(pszTag)); pMsg->iFacility = iFacility; pMsg->iSeverity = iSeverity; - CHKiRet(submitMsg(pMsg)); + /* note: we do NOT use rate-limiting, as the kernel itself does rate-limiting */ + CHKiRet(submitMsg2(pMsg)); finalize_it: RETiRet; @@ -294,6 +295,7 @@ CODESTARTbeginCnfLoad pModConf->bParseKernelStamp = 0; pModConf->bKeepKernelStamp = 0; pModConf->console_log_level = -1; + pModConf->bKeepKernelStamp = 0; pModConf->iFacilIntMsg = klogFacilIntMsg(); loadModConf->configSetViaV2Method = 0; bLegacyCnfModGlobalsPermitted = 1; diff --git a/plugins/imklog/imklog.h b/plugins/imklog/imklog.h index fa517ccc..1cf9b05a 100644 --- a/plugins/imklog/imklog.h +++ b/plugins/imklog/imklog.h @@ -35,9 +35,9 @@ struct modConfData_s { int iFacilIntMsg; uchar *pszPath; int console_log_level; - sbool bPermitNonKernel; sbool bParseKernelStamp; sbool bKeepKernelStamp; + sbool bPermitNonKernel; sbool configSetViaV2Method; }; diff --git a/plugins/impstats/impstats.c b/plugins/impstats/impstats.c index 571d734e..cdd205fd 100644 --- a/plugins/impstats/impstats.c +++ b/plugins/impstats/impstats.c @@ -145,7 +145,8 @@ doSubmitMsg(uchar *line) pMsg->iSeverity = runModConf->iSeverity; pMsg->msgFlags = 0; - submitMsg(pMsg); + /* we do not use rate-limiting, as the stats message always need to be emitted */ + submitMsg2(pMsg); DBGPRINTF("impstats: submit [%d,%d] msg '%s'\n", runModConf->iFacility, runModConf->iSeverity, line); diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c index 9888086f..5c8bb67a 100644 --- a/plugins/imptcp/imptcp.c +++ b/plugins/imptcp/imptcp.c @@ -67,6 +67,7 @@ #include "ruleset.h" #include "msg.h" #include "statsobj.h" +#include "ratelimit.h" #include "net.h" /* for permittedPeers, may be removed when this is removed */ /* the define is from tcpsrv.h, we need to find a new (but easier!!!) abstraction layer some time ... */ @@ -121,6 +122,8 @@ struct instanceConf_s { uchar *pszBindRuleset; /* name of ruleset to bind to */ uchar *pszInputName; /* value for inputname property, NULL is OK and handled by core engine */ ruleset_t *pBindRuleset; /* ruleset to bind listener to (use system default if unspecified) */ + int ratelimitInterval; + int ratelimitBurst; struct instanceConf_s *next; }; @@ -158,6 +161,8 @@ static struct cnfparamdescr inppdescr[] = { { "keepalive.time", eCmdHdlrInt, 0 }, { "keepalive.interval", eCmdHdlrInt, 0 }, { "addtlframedelimiter", eCmdHdlrInt, 0 }, + { "ratelimit.interval", eCmdHdlrInt, 0 }, + { "ratelimit.burst", eCmdHdlrInt, 0 } }; static struct cnfparamblk inppblk = { CNFPARAMBLK_VERSION, @@ -195,6 +200,7 @@ struct ptcpsrv_s { sbool bKeepAlive; /* support keep-alive packets */ sbool bEmitMsgOnClose; sbool bSuppOctetFram; + ratelimit_t *ratelimiter; }; /* the ptcp session object. Describes a single active session. @@ -295,6 +301,7 @@ destructSess(ptcpsess_t *pSess) static void destructSrv(ptcpsrv_t *pSrv) { + ratelimitDestruct(pSrv->ratelimiter); prop.Destruct(&pSrv->pInputName); pthread_mutex_destroy(&pSrv->mutSessLst); free(pSrv->pszInputName); @@ -679,14 +686,7 @@ doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, mult MsgSetRuleset(pMsg, pSrv->pRuleset); STATSCOUNTER_INC(pThis->pLstn->ctrSubmit, pThis->pLstn->mutCtrSubmit); - if(pMultiSub == NULL) { - CHKiRet(submitMsg(pMsg)); - } else { - pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg; - if(pMultiSub->nElem == pMultiSub->maxElem) - CHKiRet(multiSubmitMsg(pMultiSub)); - } - + ratelimitAddMsg(pSrv->ratelimiter, pMultiSub, pMsg); finalize_it: /* reset status variables */ @@ -805,12 +805,11 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG * we have just received a bunch of data! -- rgerhards, 2009-06-16 * EXTRACT from tcps_sess.c */ -#define NUM_MULTISUB 1024 static rsRetVal DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) { multi_submit_t multiSub; - msg_t *pMsgs[NUM_MULTISUB]; + msg_t *pMsgs[CONF_NUM_MULTISUB]; struct syslogTime stTime; time_t ttGenTime; char *pEnd; @@ -821,7 +820,7 @@ DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) datetime.getCurrTime(&stTime, &ttGenTime); multiSub.ppMsgs = pMsgs; - multiSub.maxElem = NUM_MULTISUB; + multiSub.maxElem = CONF_NUM_MULTISUB; multiSub.nElem = 0; /* We now copy the message to the session buffer. */ @@ -831,15 +830,11 @@ DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub)); } - if(multiSub.nElem > 0) { - /* submit anything that was not yet submitted */ - CHKiRet(multiSubmitMsg(&multiSub)); - } + iRet = multiSubmitFlush(&multiSub); finalize_it: RETiRet; } -#undef NUM_MULTISUB /****************************************** --END-- TCP SUPPORT FUNCTIONS ***********************************/ @@ -1051,6 +1046,8 @@ createInstance(instanceConf_t **pinst) inst->bEmitMsgOnClose = 0; inst->iAddtlFrameDelim = TCPSRV_NO_ADDTL_DELIMITER; inst->pBindRuleset = NULL; + inst->ratelimitBurst = 10000; /* arbitrary high limit */ + inst->ratelimitInterval = 0; /* off */ /* node created, let's add to config */ if(loadModConf->tail == NULL) { @@ -1130,6 +1127,9 @@ addListner(modConfData_t __attribute__((unused)) *modConf, instanceConf_t *inst) pSrv->iKeepAliveProbes = inst->iKeepAliveProbes; pSrv->iKeepAliveTime = inst->iKeepAliveTime; pSrv->bEmitMsgOnClose = inst->bEmitMsgOnClose; + CHKiRet(ratelimitNew(&pSrv->ratelimiter, "imtcp", (char*)inst->pszBindPort)); + ratelimitSetLinuxLike(pSrv->ratelimiter, inst->ratelimitInterval, inst->ratelimitBurst); + ratelimitSetThreadSafe(pSrv->ratelimiter); CHKmalloc(pSrv->port = ustrdup(inst->pszBindPort)); pSrv->iAddtlFrameDelim = inst->iAddtlFrameDelim; if(inst->pszBindAddr == NULL) @@ -1458,6 +1458,10 @@ CODESTARTnewInpInst inst->iAddtlFrameDelim = (int) pvals[i].val.d.n; } else if(!strcmp(inppblk.descr[i].name, "notifyonconnectionclose")) { inst->bEmitMsgOnClose = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.burst")) { + inst->ratelimitBurst = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.interval")) { + inst->ratelimitInterval = (int) pvals[i].val.d.n; } else { dbgprintf("imptcp: program error, non-handled " "param '%s'\n", inppblk.descr[i].name); diff --git a/plugins/imrelp/imrelp.c b/plugins/imrelp/imrelp.c index 5ee3b4b9..dc67f4fe 100644 --- a/plugins/imrelp/imrelp.c +++ b/plugins/imrelp/imrelp.c @@ -113,11 +113,29 @@ static struct cnfparamblk inppblk = * we will only see the hostname (twice). -- rgerhards, 2009-10-14 */ static relpRetVal -onSyslogRcv(uchar *pHostname, uchar *pIP, uchar *pMsg, size_t lenMsg) +onSyslogRcv(uchar *pHostname, uchar *pIP, uchar *msg, size_t lenMsg) { + prop_t *pProp = NULL; + msg_t *pMsg; DEFiRet; - parseAndSubmitMessage(pHostname, pIP, pMsg, lenMsg, PARSE_HOSTNAME, - eFLOWCTL_LIGHT_DELAY, pInputName, NULL, 0, runModConf->pBindRuleset); + + CHKiRet(msgConstruct(&pMsg)); + MsgSetInputName(pMsg, pInputName); + MsgSetRawMsg(pMsg, (char*)msg, lenMsg); + MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY); + MsgSetRuleset(pMsg, runModConf->pBindRuleset); + pMsg->msgFlags = PARSE_HOSTNAME | NEEDS_PARSING; + + /* TODO: optimize this, we can store it inside the session, requires + * changes to librelp --> next librelp iteration?. rgerhards, 2012-10-29 + */ + MsgSetRcvFromStr(pMsg, pHostname, ustrlen(pHostname), &pProp); + CHKiRet(prop.Destruct(&pProp)); + CHKiRet(MsgSetRcvFromIPStr(pMsg, pIP, ustrlen(pIP), &pProp)); + CHKiRet(prop.Destruct(&pProp)); + CHKiRet(submitMsg2(pMsg)); + +finalize_it: RETiRet; } diff --git a/plugins/imtcp/imtcp.c b/plugins/imtcp/imtcp.c index 0cecb704..fc22d452 100644 --- a/plugins/imtcp/imtcp.c +++ b/plugins/imtcp/imtcp.c @@ -105,6 +105,8 @@ struct instanceConf_s { uchar *pszBindRuleset; /* name of ruleset to bind to */ ruleset_t *pBindRuleset; /* ruleset to bind listener to (use system default if unspecified) */ uchar *pszInputName; /* value for inputname property, NULL is OK and handled by core engine */ + int ratelimitInterval; + int ratelimitBurst; int bSuppOctetFram; struct instanceConf_s *next; }; @@ -155,7 +157,9 @@ static struct cnfparamdescr inppdescr[] = { { "port", eCmdHdlrString, CNFPARAM_REQUIRED }, /* legacy: InputTCPServerRun */ { "name", eCmdHdlrString, 0 }, { "ruleset", eCmdHdlrString, 0 }, - { "supportOctetCountedFraming", eCmdHdlrBinary, 0 } + { "supportOctetCountedFraming", eCmdHdlrBinary, 0 }, + { "ratelimit.interval", eCmdHdlrInt, 0 }, + { "ratelimit.burst", eCmdHdlrInt, 0 } }; static struct cnfparamblk inppblk = { CNFPARAMBLK_VERSION, @@ -251,6 +255,8 @@ createInstance(instanceConf_t **pinst) inst->pszBindRuleset = NULL; inst->pszInputName = NULL; inst->bSuppOctetFram = 1; + inst->ratelimitInterval = 0; + inst->ratelimitBurst = 10000; /* node created, let's add to config */ if(loadModConf->tail == NULL) { @@ -334,6 +340,7 @@ addListner(modConfData_t *modConf, instanceConf_t *inst) CHKiRet(tcpsrv.SetRuleset(pOurTcpsrv, inst->pBindRuleset)); CHKiRet(tcpsrv.SetInputName(pOurTcpsrv, inst->pszInputName == NULL ? UCHAR_CONSTANT("imtcp") : inst->pszInputName)); + CHKiRet(tcpsrv.SetLinuxLikeRatelimiters(pOurTcpsrv, inst->ratelimitInterval, inst->ratelimitBurst)); tcpsrv.configureTCPListen(pOurTcpsrv, inst->pszBindPort, inst->bSuppOctetFram); finalize_it: @@ -376,6 +383,10 @@ CODESTARTnewInpInst inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(inppblk.descr[i].name, "supportOctetCountedFraming")) { inst->bSuppOctetFram = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.burst")) { + inst->ratelimitBurst = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.interval")) { + inst->ratelimitInterval = (int) pvals[i].val.d.n; } else { dbgprintf("imtcp: program error, non-handled " "param '%s'\n", inppblk.descr[i].name); @@ -487,10 +498,10 @@ CODESTARTendCnfLoad loadModConf->pszStrmDrvrAuthMode = NULL; } else { loadModConf->pszStrmDrvrAuthMode = cs.pszStrmDrvrAuthMode; + cs.pszStrmDrvrAuthMode = NULL; } } - if((cs.pszStrmDrvrAuthMode == NULL) || (cs.pszStrmDrvrAuthMode[0] == '\0')) - free(cs.pszStrmDrvrAuthMode); + free(cs.pszStrmDrvrAuthMode); cs.pszStrmDrvrAuthMode = NULL; loadModConf = NULL; /* done loading */ @@ -550,6 +561,7 @@ ENDactivateCnf BEGINfreeCnf instanceConf_t *inst, *del; CODESTARTfreeCnf + free(pModConf->pszStrmDrvrAuthMode); if(pModConf->permittedPeers != NULL) { cnfarrayContentDestruct(pModConf->permittedPeers); free(pModConf->permittedPeers); @@ -580,7 +592,9 @@ ENDwillRun BEGINafterRun CODESTARTafterRun - /* do cleanup here */ + if(pOurTcpsrv != NULL) + iRet = tcpsrv.Destruct(&pOurTcpsrv); + net.clearAllowedSenders(UCHAR_CONSTANT("TCP")); ENDafterRun @@ -594,9 +608,6 @@ ENDisCompatibleWithFeature BEGINmodExit CODESTARTmodExit - if(pOurTcpsrv != NULL) - iRet = tcpsrv.Destruct(&pOurTcpsrv); - if(pPermPeersRoot != NULL) { net.DestructPermittedPeers(&pPermPeersRoot); } diff --git a/plugins/imudp/imudp.c b/plugins/imudp/imudp.c index 9ec1dbbc..dde8f105 100644 --- a/plugins/imudp/imudp.c +++ b/plugins/imudp/imudp.c @@ -4,8 +4,6 @@ * NOTE: read comments in module-template.h to understand how this file * works! * - * File begun on 2007-12-21 by RGerhards (extracted from syslogd.c) - * * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH. * * This file is part of rsyslog. @@ -53,6 +51,7 @@ #include "prop.h" #include "ruleset.h" #include "statsobj.h" +#include "ratelimit.h" #include "unicode-helper.h" MODULE_TYPE_INPUT @@ -77,6 +76,7 @@ static struct lstn_s { int sock; /* socket */ ruleset_t *pRuleset; /* bound ruleset */ statsobj_t *stats; /* listener stats */ + ratelimit_t *ratelimiter; STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit) } *lcnfRoot = NULL, *lcnfLast = NULL; @@ -109,6 +109,8 @@ struct instanceConf_s { uchar *pszBindPort; /* Port to bind socket to */ uchar *pszBindRuleset; /* name of ruleset to bind to */ ruleset_t *pBindRuleset; /* ruleset to bind listener to (use system default if unspecified) */ + int ratelimitInterval; + int ratelimitBurst; struct instanceConf_s *next; }; @@ -140,7 +142,9 @@ static struct cnfparamblk modpblk = static struct cnfparamdescr inppdescr[] = { { "port", eCmdHdlrArray, CNFPARAM_REQUIRED }, /* legacy: InputTCPServerRun */ { "address", eCmdHdlrString, 0 }, - { "ruleset", eCmdHdlrString, 0 } + { "ruleset", eCmdHdlrString, 0 }, + { "ratelimit.interval", eCmdHdlrInt, 0 }, + { "ratelimit.burst", eCmdHdlrInt, 0 } }; static struct cnfparamblk inppblk = { CNFPARAMBLK_VERSION, @@ -165,6 +169,8 @@ createInstance(instanceConf_t **pinst) inst->pszBindPort = NULL; inst->pszBindAddr = NULL; inst->pszBindRuleset = NULL; + inst->ratelimitBurst = 10000; /* arbitrary high limit */ + inst->ratelimitInterval = 0; /* off */ /* node created, let's add to config */ if(loadModConf->tail == NULL) { @@ -223,7 +229,7 @@ addListner(instanceConf_t *inst) struct lstn_s *newlcnfinfo; uchar *bindName; uchar *port; - uchar statname[64]; + uchar dispname[64]; /* check which address to bind to. We could do this more compact, but have not * done so in order to make the code more readable. -- rgerhards, 2007-12-27 @@ -248,11 +254,14 @@ addListner(instanceConf_t *inst) newlcnfinfo->next = NULL; newlcnfinfo->sock = newSocks[iSrc]; newlcnfinfo->pRuleset = inst->pBindRuleset; + snprintf((char*)dispname, sizeof(dispname), "imudp(%s:%s)", bindName, port); + dispname[sizeof(dispname)-1] = '\0'; /* just to be on the save side... */ + CHKiRet(ratelimitNew(&newlcnfinfo->ratelimiter, (char*)dispname, NULL)); + ratelimitSetLinuxLike(newlcnfinfo->ratelimiter, inst->ratelimitInterval, + inst->ratelimitBurst); /* support statistics gathering */ CHKiRet(statsobj.Construct(&(newlcnfinfo->stats))); - snprintf((char*)statname, sizeof(statname), "imudp(%s:%s)", bindName, port); - statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */ - CHKiRet(statsobj.SetName(newlcnfinfo->stats, statname)); + CHKiRet(statsobj.SetName(newlcnfinfo->stats, dispname)); STATSCOUNTER_INIT(newlcnfinfo->ctrSubmit, newlcnfinfo->mutCtrSubmit); CHKiRet(statsobj.AddCounter(newlcnfinfo->stats, UCHAR_CONSTANT("submitted"), ctrType_IntCtr, &(newlcnfinfo->ctrSubmit))); @@ -304,7 +313,6 @@ std_checkRuleset_genErrMsg(__attribute__((unused)) modConfData_t *modConf, insta static inline rsRetVal processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *frominetPrev, int *pbIsPermitted) { - DEFiRet; int iNbrTimeUsed; time_t ttGenTime; struct syslogTime stTime; @@ -314,9 +322,15 @@ processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *f msg_t *pMsg; prop_t *propFromHost = NULL; prop_t *propFromHostIP = NULL; + multi_submit_t multiSub; + msg_t *pMsgs[CONF_NUM_MULTISUB]; char errStr[1024]; + DEFiRet; assert(pThrd != NULL); + multiSub.ppMsgs = pMsgs; + multiSub.maxElem = CONF_NUM_MULTISUB; + multiSub.nElem = 0; iNbrTimeUsed = 0; while(1) { /* loop is terminated if we have a bad receive, done below in the body */ if(pThrd->bShallStop == RSTRUE) @@ -383,12 +397,15 @@ processSocket(thrdInfo_t *pThrd, struct lstn_s *lstn, struct sockaddr_storage *f if(*pbIsPermitted == 2) pMsg->msgFlags |= NEEDS_ACLCHK_U; /* request ACL check after resolution */ CHKiRet(msgSetFromSockinfo(pMsg, &frominet)); - CHKiRet(submitMsg(pMsg)); + CHKiRet(ratelimitAddMsg(lstn->ratelimiter, &multiSub, pMsg)); STATSCOUNTER_INC(lstn->ctrSubmit, lstn->mutCtrSubmit); } } + finalize_it: + multiSubmitFlush(&multiSub); + if(propFromHost != NULL) prop.Destruct(&propFromHost); if(propFromHostIP != NULL) @@ -682,6 +699,10 @@ createListner(es_str_t *port, struct cnfparamvals *pvals) inst->pszBindAddr = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(inppblk.descr[i].name, "ruleset")) { inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.burst")) { + inst->ratelimitBurst = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "ratelimit.interval")) { + inst->ratelimitInterval = (int) pvals[i].val.d.n; } else { dbgprintf("imudp: program error, non-handled " "param '%s'\n", inppblk.descr[i].name); @@ -884,6 +905,7 @@ CODESTARTafterRun net.clearAllowedSenders((uchar*)"UDP"); for(lstn = lcnfRoot ; lstn != NULL ; ) { statsobj.Destruct(&(lstn->stats)); + ratelimitDestruct(lstn->ratelimiter); close(lstn->sock); lstnDel = lstn; lstn = lstn->next; diff --git a/plugins/imuxsock/imuxsock.c b/plugins/imuxsock/imuxsock.c index 871a1fa5..66a1c648 100644 --- a/plugins/imuxsock/imuxsock.c +++ b/plugins/imuxsock/imuxsock.c @@ -55,6 +55,7 @@ #include "statsobj.h" #include "datetime.h" #include "hashtable.h" +#include "ratelimit.h" MODULE_TYPE_INPUT MODULE_TYPE_NOKEEP @@ -105,15 +106,6 @@ STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit) STATSCOUNTER_DEF(ctrLostRatelimit, mutCtrLostRatelimit) STATSCOUNTER_DEF(ctrNumRatelimiters, mutCtrNumRatelimiters) -struct rs_ratelimit_state { - unsigned short interval; - unsigned short burst; - unsigned done; - unsigned missed; - time_t begin; -}; -typedef struct rs_ratelimit_state rs_ratelimit_state_t; - /* a very simple "hash function" for process IDs - we simply use the * pid itself: it is quite expected that all pids may log some time, but @@ -143,6 +135,7 @@ typedef struct lstn_s { int flowCtl; /* flow control settings for this socket */ int ratelimitInterval; int ratelimitBurst; + ratelimit_t *dflt_ratelimiter;/*ratelimiter to apply if none else is to be used */ intTiny ratelimitSev; /* severity level (and below) for which rate-limiting shall apply */ struct hashtable *ht; /* our hashtable for rate-limiting */ sbool bParseHost; /* should parser parse host name? read-only after startup */ @@ -271,74 +264,9 @@ static struct cnfparamblk inppblk = /* we do not use this, because we do not bind to a ruleset so far * enable when this is changed: #include "im-helper.h" */ /* must be included AFTER the type definitions! */ -static void -initRatelimitState(struct rs_ratelimit_state *rs, unsigned short interval, unsigned short burst) -{ - rs->interval = interval; - rs->burst = burst; - rs->done = 0; - rs->missed = 0; - rs->begin = 0; -} - static int bLegacyCnfModGlobalsPermitted;/* are legacy module-global config parameters permitted? */ -/* ratelimiting support, modelled after the linux kernel - * returns 1 if message is within rate limit and shall be - * processed, 0 otherwise. - * This implementation is NOT THREAD-SAFE and must not - * be called concurrently. - */ -static inline int -withinRatelimit(struct rs_ratelimit_state *rs, time_t tt, pid_t pid) -{ - int ret; - uchar msgbuf[1024]; - - if(rs->interval == 0) { - ret = 1; - goto finalize_it; - } - - assert(rs->burst != 0); - - if(rs->begin == 0) - rs->begin = tt; - - /* resume if we go out of out time window */ - if(tt > rs->begin + rs->interval) { - if(rs->missed) { - snprintf((char*)msgbuf, sizeof(msgbuf), - "imuxsock lost %u messages from pid %lu due to rate-limiting", - rs->missed, (unsigned long) pid); - logmsgInternal(RS_RET_RATE_LIMITED, LOG_SYSLOG|LOG_INFO, msgbuf, 0); - rs->missed = 0; - } - rs->begin = 0; - rs->done = 0; - } - - /* do actual limit check */ - if(rs->burst > rs->done) { - rs->done++; - ret = 1; - } else { - if(rs->missed == 0) { - snprintf((char*)msgbuf, sizeof(msgbuf), - "imuxsock begins to drop messages from pid %lu due to rate-limiting", - (unsigned long) pid); - logmsgInternal(RS_RET_RATE_LIMITED, LOG_SYSLOG|LOG_INFO, msgbuf, 0); - } - rs->missed++; - ret = 0; - } - -finalize_it: - return ret; -} - - /* create input instance, set default paramters, and * add it to the list of instances. */ @@ -445,7 +373,8 @@ addListner(instanceConf_t *inst) CHKiRet(prop.ConstructFinalize(listeners[nfd].hostName)); } if(inst->ratelimitInterval > 0) { - if((listeners[nfd].ht = create_hashtable(100, hash_from_key_fn, key_equals_fn, NULL)) == NULL) { + if((listeners[nfd].ht = create_hashtable(100, hash_from_key_fn, key_equals_fn, + (void(*)(void*))ratelimitDestruct)) == NULL) { /* in this case, we simply turn off rate-limiting */ DBGPRINTF("imuxsock: turning off rate limiting because we could not " "create hash table\n"); @@ -464,6 +393,10 @@ addListner(instanceConf_t *inst) listeners[nfd].bParseTrusted = inst->bParseTrusted; listeners[nfd].bWritePid = inst->bWritePid; listeners[nfd].bUseSysTimeStamp = inst->bUseSysTimeStamp; + CHKiRet(ratelimitNew(&listeners[nfd].dflt_ratelimiter, "imuxsock", NULL)); + ratelimitSetLinuxLike(listeners[nfd].dflt_ratelimiter, + listeners[nfd].ratelimitInterval, + listeners[nfd].ratelimitBurst); nfd++; } else { errmsg.LogError(0, NO_ERRCODE, "Out of unix socket name descriptors, ignoring %s\n", @@ -475,7 +408,7 @@ finalize_it: } -/* discard all log sockets except for "socket" 0. Data for it comes from +/* discard/Destruct all log sockets except for "socket" 0. Data for it comes from * the constant memory pool - and if not, it is freeed via some other pointer. */ static rsRetVal discardLogSockets(void) @@ -493,6 +426,7 @@ static rsRetVal discardLogSockets(void) if(listeners[i].ht != NULL) { hashtable_destroy(listeners[i].ht, 1); /* 1 => free all values automatically */ } + ratelimitDestruct(listeners[i].dflt_ratelimiter); } return RS_RET_OK; @@ -604,19 +538,22 @@ finalize_it: * listener (the latter being a performance enhancement). */ static inline rsRetVal -findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl) +findRatelimiter(lstn_t *pLstn, struct ucred *cred, ratelimit_t **prl) { - rs_ratelimit_state_t *rl; + ratelimit_t *rl; int r; pid_t *keybuf; + char pidbuf[256]; DEFiRet; if(cred == NULL) FINALIZE; +#if 0 // TODO: check deactivated? if(pLstn->ratelimitInterval == 0) { *prl = NULL; FINALIZE; } +#endif rl = hashtable_search(pLstn->ht, &cred->pid); if(rl == NULL) { @@ -624,10 +561,13 @@ findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl) DBGPRINTF("imuxsock: no ratelimiter for pid %lu, creating one\n", (unsigned long) cred->pid); STATSCOUNTER_INC(ctrNumRatelimiters, mutCtrNumRatelimiters); - CHKmalloc(rl = malloc(sizeof(rs_ratelimit_state_t))); + snprintf(pidbuf, sizeof(pidbuf), "pid %lu", + (unsigned long) cred->pid); + pidbuf[sizeof(pidbuf)-1] = '\0'; /* to be on safe side */ + CHKiRet(ratelimitNew(&rl, "imuxsock", pidbuf)); + ratelimitSetLinuxLike(rl, pLstn->ratelimitInterval, pLstn->ratelimitBurst); CHKmalloc(keybuf = malloc(sizeof(pid_t))); *keybuf = cred->pid; - initRatelimitState(rl, pLstn->ratelimitInterval, pLstn->ratelimitBurst); r = hashtable_insert(pLstn->ht, keybuf, rl); if(r == 0) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); @@ -636,6 +576,8 @@ findRatelimiter(lstn_t *pLstn, struct ucred *cred, rs_ratelimit_state_t **prl) *prl = rl; finalize_it: + if(*prl == NULL) + *prl = pLstn->dflt_ratelimiter; RETiRet; } @@ -762,28 +704,6 @@ copyescaped(uchar *dstbuf, uchar *inbuf, int inlen) } -#if 0 -/* Creates new field to be added to event - * used for SystemLogParseTrusted parsing - */ -struct ee_field * -createNewField(char *fieldname, char *value, int lenValue) { - es_str_t *newStr; - struct ee_value *newVal; - struct ee_field *newField; - - newStr = es_newStrFromBuf(value, (es_size_t) lenValue); - - newVal = ee_newValue(ctxee); - ee_setStrValue(newVal, newStr); - - newField = ee_newFieldFromNV(ctxee, fieldname, newVal); - - return newField; -} -#endif - - /* submit received message to the queue engine * We now parse the message according to expected format so that we * can also mangle it if necessary. @@ -802,8 +722,8 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim uchar bufParseTAG[CONF_TAG_MAXSIZE]; struct syslogTime st; time_t tt; - rs_ratelimit_state_t *ratelimiter = NULL; int lenProp; + ratelimit_t *ratelimiter = NULL; uchar propBuf[1024]; uchar msgbuf[8192]; uchar *pmsgbuf; @@ -842,10 +762,12 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim tt = ts->tv_sec; } +#if 0 // TODO: think about stats counters (or wait for request...?) if(ratelimiter != NULL && !withinRatelimit(ratelimiter, tt, cred->pid)) { STATSCOUNTER_INC(ctrLostRatelimit, mutCtrLostRatelimit); FINALIZE; } +#endif /* created trusted properties */ if(cred != NULL && pLstn->bAnnotate) { @@ -980,8 +902,7 @@ SubmitMsg(uchar *pRcv, int lenRcv, lstn_t *pLstn, struct ucred *cred, struct tim MsgSetRcvFrom(pMsg, pLstn->hostName == NULL ? glbl.GetLocalHostNameProp() : pLstn->hostName); CHKiRet(MsgSetRcvFromIP(pMsg, pLocalHostIP)); - CHKiRet(submitMsg(pMsg)); - + ratelimitAddMsg(ratelimiter, NULL, pMsg); STATSCOUNTER_INC(ctrSubmit, mutCtrSubmit); finalize_it: RETiRet; @@ -1123,6 +1044,10 @@ activateListeners() listeners[0].bUseSysTimeStamp = runModConf->bUseSysTimeStamp; listeners[0].flags = runModConf->bIgnoreTimestamp ? IGNDATE : NOFLAG; listeners[0].flowCtl = runModConf->bUseFlowCtl ? eFLOWCTL_LIGHT_DELAY : eFLOWCTL_NO_DELAY; + CHKiRet(ratelimitNew(&listeners[0].dflt_ratelimiter, "imuxsock", NULL)); + ratelimitSetLinuxLike(listeners[0].dflt_ratelimiter, + listeners[0].ratelimitInterval, + listeners[0].ratelimitBurst); sd_fds = sd_listen_fds(0); if(sd_fds < 0) { diff --git a/plugins/mmjsonparse/mmjsonparse.c b/plugins/mmjsonparse/mmjsonparse.c index a13962ab..ce32b7b7 100644 --- a/plugins/mmjsonparse/mmjsonparse.c +++ b/plugins/mmjsonparse/mmjsonparse.c @@ -62,11 +62,35 @@ typedef struct _instanceData { struct json_tokener *tokener; } instanceData; +struct modConfData_s { + rsconf_t *pConf; /* our overall config object */ +}; +static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */ +static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */ -BEGINinitConfVars /* (re)set config variables to default values */ -CODESTARTinitConfVars - resetConfigVariables(NULL, NULL); -ENDinitConfVars + +BEGINbeginCnfLoad +CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; +ENDbeginCnfLoad + +BEGINendCnfLoad +CODESTARTendCnfLoad +ENDendCnfLoad + +BEGINcheckCnf +CODESTARTcheckCnf +ENDcheckCnf + +BEGINactivateCnf +CODESTARTactivateCnf + runModConf = pModConf; +ENDactivateCnf + +BEGINfreeCnf +CODESTARTfreeCnf +ENDfreeCnf BEGINcreateInstance @@ -177,6 +201,22 @@ finalize_it: MsgSetParseSuccess(pMsg, bSuccess); ENDdoAction +BEGINnewActInst +CODESTARTnewActInst + /* Note: we currently do not have any parameters, so we do not need + * the lst ptr. However, we will most probably need params in the + * future. + */ + DBGPRINTF("newActInst (mmjsonparse)\n"); + + CODE_STD_STRING_REQUESTnewActInst(1) + CHKiRet(OMSRsetEntry(*ppOMSR, 0, NULL, OMSR_TPL_AS_MSG)); + CHKiRet(createInstance(&pData)); + /*setInstParamDefaults(pData);*/ + +CODE_STD_FINALIZERnewActInst +/* cnfparamvalsDestruct(pvals, &actpblk);*/ +ENDnewActInst BEGINparseSelectorAct CODESTARTparseSelectorAct @@ -218,7 +258,8 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES -CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES +CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES +CODEqueryEtryPt_STD_CONF2_QUERIES ENDqueryEtryPt @@ -238,7 +279,6 @@ BEGINmodInit() unsigned long opts; int bMsgPassingSupported; CODESTARTmodInit -INITLegCnfVars *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ CODEmodInit_QueryRegCFSLineHdlr diff --git a/plugins/mmnormalize/mmnormalize.c b/plugins/mmnormalize/mmnormalize.c index d3fba39b..fd2004a3 100644 --- a/plugins/mmnormalize/mmnormalize.c +++ b/plugins/mmnormalize/mmnormalize.c @@ -65,6 +65,7 @@ DEF_OMOD_STATIC_DATA typedef struct _instanceData { sbool bUseRawMsg; /**< use %rawmsg% instead of %msg% */ + uchar *rulebase; /**< name of rulebase to use */ ln_ctx ctxln; /**< context to be used for liblognorm */ ee_ctx ctxee; /**< context to be used for libee */ } instanceData; @@ -75,6 +76,58 @@ typedef struct configSettings_s { } configSettings_t; static configSettings_t cs; +/* tables for interfacing with the v6 config system */ +/* action (instance) parameters */ +static struct cnfparamdescr actpdescr[] = { + { "rulebase", eCmdHdlrGetWord, 1 }, + { "userawmsg", eCmdHdlrBinary, 0 } +}; +static struct cnfparamblk actpblk = + { CNFPARAMBLK_VERSION, + sizeof(actpdescr)/sizeof(struct cnfparamdescr), + actpdescr + }; + +struct modConfData_s { + rsconf_t *pConf; /* our overall config object */ +}; + +static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */ +static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */ + + +/* to be called to build the libee part of the instance ONCE ALL PARAMETERS ARE CORRECT + * (and set within pData!). + */ +static rsRetVal +buildInstance(instanceData *pData) +{ + DEFiRet; + if((pData->ctxee = ee_initCtx()) == NULL) { + errmsg.LogError(0, RS_RET_ERR_LIBEE_INIT, "error: could not initialize libee " + "ctx, cannot activate action"); + ABORT_FINALIZE(RS_RET_ERR_LIBEE_INIT); + } + + if((pData->ctxln = ln_initCtx()) == NULL) { + errmsg.LogError(0, RS_RET_ERR_LIBLOGNORM_INIT, "error: could not initialize " + "liblognorm ctx, cannot activate action"); + ee_exitCtx(pData->ctxee); + ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_INIT); + } + ln_setEECtx(pData->ctxln, pData->ctxee); + if(ln_loadSamples(pData->ctxln, (char*) pData->rulebase) != 0) { + errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: normalization rulebase '%s' " + "could not be loaded cannot activate action", cs.rulebase); + ee_exitCtx(pData->ctxee); + ln_exitCtx(pData->ctxln); + ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD); + } +finalize_it: + RETiRet; +} + + BEGINinitConfVars /* (re)set config variables to default values */ CODESTARTinitConfVars resetConfigVariables(NULL, NULL); @@ -86,6 +139,35 @@ CODESTARTcreateInstance ENDcreateInstance +BEGINbeginCnfLoad +CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; +ENDbeginCnfLoad + + +BEGINendCnfLoad +CODESTARTendCnfLoad + loadModConf = NULL; /* done loading */ + /* free legacy config vars */ + free(cs.rulebase); + cs.rulebase = NULL; +ENDendCnfLoad + +BEGINcheckCnf +CODESTARTcheckCnf +ENDcheckCnf + +BEGINactivateCnf +CODESTARTactivateCnf + runModConf = pModConf; +ENDactivateCnf + +BEGINfreeCnf +CODESTARTfreeCnf +ENDfreeCnf + + BEGINisCompatibleWithFeature CODESTARTisCompatibleWithFeature ENDisCompatibleWithFeature @@ -93,6 +175,7 @@ ENDisCompatibleWithFeature BEGINfreeInstance CODESTARTfreeInstance + free(pData->rulebase); ee_exitCtx(pData->ctxee); ln_exitCtx(pData->ctxln); ENDfreeInstance @@ -141,7 +224,7 @@ CODESTARTdoAction es_deleteStr(str); /* reformat to our json data struct */ - // TODO: this is all extremly ineffcient! + /* TODO: this is all extremly ineffcient! */ ee_fmtEventToJSON(event, &str); cstrJSON = es_str2cstr(str, NULL); dbgprintf("mmnormalize generated: %s\n", cstrJSON); @@ -156,6 +239,59 @@ CODESTARTdoAction ENDdoAction +static inline void +setInstParamDefaults(instanceData *pData) +{ + pData->rulebase = NULL; + pData->bUseRawMsg = 0; +} + +BEGINnewActInst + struct cnfparamvals *pvals; + int i; + int bDestructPValsOnExit; +CODESTARTnewActInst + DBGPRINTF("newActInst (mmnormalize)\n"); + + bDestructPValsOnExit = 0; + pvals = nvlstGetParams(lst, &actpblk, NULL); + if(pvals == NULL) { + errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "mmnormalize: error reading " + "config parameters"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + bDestructPValsOnExit = 1; + + if(Debug) { + dbgprintf("action param blk in mmnormalize:\n"); + cnfparamsPrint(&actpblk, pvals); + } + + CHKiRet(createInstance(&pData)); + setInstParamDefaults(pData); + + for(i = 0 ; i < actpblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(actpblk.descr[i].name, "rulebase")) { + pData->rulebase = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(actpblk.descr[i].name, "userawmsg")) { + pData->bUseRawMsg = (int) pvals[i].val.d.n; + } else { + DBGPRINTF("mmnormalize: program error, non-handled " + "param '%s'\n", actpblk.descr[i].name); + } + } + CODE_STD_STRING_REQUESTnewActInst(1) + CHKiRet(OMSRsetEntry(*ppOMSR, 0, NULL, OMSR_TPL_AS_MSG)); + + iRet = buildInstance(pData); +CODE_STD_FINALIZERnewActInst + if(bDestructPValsOnExit) + cnfparamvalsDestruct(pvals, &actpblk); +ENDnewActInst + + BEGINparseSelectorAct CODESTARTparseSelectorAct CODE_STD_STRING_REQUESTparseSelectorAct(1) @@ -165,15 +301,21 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) } if(cs.rulebase == NULL) { - errmsg.LogError(0, RS_RET_NO_RULESET, "error: no normalization rulebase was specified, use " + errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: no normalization rulebase was specified, use " "$MMNormalizeSampleDB directive first!"); - ABORT_FINALIZE(RS_RET_NO_RULESET); + ABORT_FINALIZE(RS_RET_NO_RULEBASE); } /* ok, if we reach this point, we have something for us */ p += sizeof(":mmnormalize:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ CHKiRet(createInstance(&pData)); + pData->rulebase = cs.rulebase; + pData->bUseRawMsg = cs.bUseRawMsg; + /* all config vars auto-reset! */ + cs.bUseRawMsg = 0; + cs.rulebase = NULL; /* we used it up! */ + /* check if a non-standard template is to be applied */ if(*(p-1) == ';') --p; @@ -181,34 +323,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) * the format specified (if any) is always ignored. */ CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_TPL_AS_MSG, (uchar*) "RSYSLOG_FileFormat")); - - /* finally build the instance */ - if((pData->ctxee = ee_initCtx()) == NULL) { - errmsg.LogError(0, RS_RET_NO_RULESET, "error: could not initialize libee ctx, cannot " - "activate action"); - ABORT_FINALIZE(RS_RET_ERR_LIBEE_INIT); - } - - if((pData->ctxln = ln_initCtx()) == NULL) { - errmsg.LogError(0, RS_RET_NO_RULESET, "error: could not initialize liblognorm ctx, cannot " - "activate action"); - ee_exitCtx(pData->ctxee); - ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_INIT); - } - ln_setEECtx(pData->ctxln, pData->ctxee); - if(ln_loadSamples(pData->ctxln, (char*) cs.rulebase) != 0) { - errmsg.LogError(0, RS_RET_NO_RULESET, "error: normalization rulebase '%s' could not be loaded " - "cannot activate action", cs.rulebase); - ee_exitCtx(pData->ctxee); - ln_exitCtx(pData->ctxln); - ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD); - } - pData->bUseRawMsg = cs.bUseRawMsg; - - /* all config vars auto-reset! */ - cs.bUseRawMsg = 0; - free(cs.rulebase); - cs.rulebase = NULL; + CHKiRet(buildInstance(pData)); CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct @@ -222,7 +337,8 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES -CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES +CODEqueryEtryPt_STD_CONF2_QUERIES +CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES ENDqueryEtryPt diff --git a/plugins/omelasticsearch/Makefile.am b/plugins/omelasticsearch/Makefile.am index 2fadb74d..ba85a896 100644 --- a/plugins/omelasticsearch/Makefile.am +++ b/plugins/omelasticsearch/Makefile.am @@ -1,6 +1,7 @@ pkglib_LTLIBRARIES = omelasticsearch.la -omelasticsearch_la_SOURCES = omelasticsearch.c +# TODO: replace cJSON +omelasticsearch_la_SOURCES = omelasticsearch.c cJSON/cjson.c cJSON/cjson.h omelasticsearch_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) omelasticsearch_la_LDFLAGS = -module -avoid-version omelasticsearch_la_LIBADD = $(CURL_LIBS) $(LIBM) diff --git a/plugins/omelasticsearch/README b/plugins/omelasticsearch/README new file mode 100644 index 00000000..9021bc0e --- /dev/null +++ b/plugins/omelasticsearch/README @@ -0,0 +1,17 @@ +How to produce an error: +======================== +It's quite easy to get 400, if you put a wrong mapping to your +index. That would be easy to reproduce in "normal" omelasticsearch usage +conditions, by only altering the ES configuration: + +1. Make your index first. Let's call it "testindex": +$ curl -XPUT localhost:9200/testindex/ + +2. Put your mapping for a search type called "mytype", where you specify +that date property should be an integer: +$ curl -XPUT localhost:9200/testindex/mytype/_mapping -d '{"mytype":{"properties": {"timegenerated":{"type":"integer"}}}}' + +3. Now try to insert something where date is not an integer: +$ curl -XPOST localhost:9200/testindex/mytype/ -d '{"timegenerated":"bla"}' +{"error":"MapperParsingException[Failed to parse [date]]; nested: NumberFormatException[For input string: \"bla\"]; ","status":400} + diff --git a/plugins/omelasticsearch/cJSON/README b/plugins/omelasticsearch/cJSON/README new file mode 100644 index 00000000..7531c049 --- /dev/null +++ b/plugins/omelasticsearch/cJSON/README @@ -0,0 +1,247 @@ +/* + Copyright (c) 2009 Dave Gamble + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. +*/ + +Welcome to cJSON. + +cJSON aims to be the dumbest possible parser that you can get your job done with. +It's a single file of C, and a single header file. + +JSON is described best here: http://www.json.org/ +It's like XML, but fat-free. You use it to move data around, store things, or just +generally represent your program's state. + + +First up, how do I build? +Add cJSON.c to your project, and put cJSON.h somewhere in the header search path. +For example, to build the test app: + +gcc cJSON.c test.c -o test -lm +./test + + +As a library, cJSON exists to take away as much legwork as it can, but not get in your way. +As a point of pragmatism (i.e. ignoring the truth), I'm going to say that you can use it +in one of two modes: Auto and Manual. Let's have a quick run-through. + + +I lifted some JSON from this page: http://www.json.org/fatfree.html +That page inspired me to write cJSON, which is a parser that tries to share the same +philosophy as JSON itself. Simple, dumb, out of the way. + +Some JSON: +{ + "name": "Jack (\"Bee\") Nimble", + "format": { + "type": "rect", + "width": 1920, + "height": 1080, + "interlace": false, + "frame rate": 24 + } +} + +Assume that you got this from a file, a webserver, or magic JSON elves, whatever, +you have a char * to it. Everything is a cJSON struct. +Get it parsed: + cJSON *root = cJSON_Parse(my_json_string); + +This is an object. We're in C. We don't have objects. But we do have structs. +What's the framerate? + + cJSON *format = cJSON_GetObjectItem(root,"format"); + int framerate = cJSON_GetObjectItem(format,"frame rate")->valueint; + + +Want to change the framerate? + cJSON_GetObjectItem(format,"frame rate")->valueint=25; + +Back to disk? + char *rendered=cJSON_Print(root); + +Finished? Delete the root (this takes care of everything else). + cJSON_Delete(root); + +That's AUTO mode. If you're going to use Auto mode, you really ought to check pointers +before you dereference them. If you want to see how you'd build this struct in code? + cJSON *root,*fmt; + root=cJSON_CreateObject(); + cJSON_AddItemToObject(root, "name", cJSON_CreateString("Jack (\"Bee\") Nimble")); + cJSON_AddItemToObject(root, "format", fmt=cJSON_CreateObject()); + cJSON_AddStringToObject(fmt,"type", "rect"); + cJSON_AddNumberToObject(fmt,"width", 1920); + cJSON_AddNumberToObject(fmt,"height", 1080); + cJSON_AddFalseToObject (fmt,"interlace"); + cJSON_AddNumberToObject(fmt,"frame rate", 24); + +Hopefully we can agree that's not a lot of code? There's no overhead, no unnecessary setup. +Look at test.c for a bunch of nice examples, mostly all ripped off the json.org site, and +a few from elsewhere. + +What about manual mode? First up you need some detail. +Let's cover how the cJSON objects represent the JSON data. +cJSON doesn't distinguish arrays from objects in handling; just type. +Each cJSON has, potentially, a child, siblings, value, a name. + +The root object has: Object Type and a Child +The Child has name "name", with value "Jack ("Bee") Nimble", and a sibling: +Sibling has type Object, name "format", and a child. +That child has type String, name "type", value "rect", and a sibling: +Sibling has type Number, name "width", value 1920, and a sibling: +Sibling has type Number, name "height", value 1080, and a sibling: +Sibling hs type False, name "interlace", and a sibling: +Sibling has type Number, name "frame rate", value 24 + +Here's the structure: +typedef struct cJSON { + struct cJSON *next,*prev; + struct cJSON *child; + + int type; + + char *valuestring; + int valueint; + double valuedouble; + + char *string; +} cJSON; + +By default all values are 0 unless set by virtue of being meaningful. + +next/prev is a doubly linked list of siblings. next takes you to your sibling, +prev takes you back from your sibling to you. +Only objects and arrays have a "child", and it's the head of the doubly linked list. +A "child" entry will have prev==0, but next potentially points on. The last sibling has next=0. +The type expresses Null/True/False/Number/String/Array/Object, all of which are #defined in +cJSON.h + +A Number has valueint and valuedouble. If you're expecting an int, read valueint, if not read +valuedouble. + +Any entry which is in the linked list which is the child of an object will have a "string" +which is the "name" of the entry. When I said "name" in the above example, that's "string". +"string" is the JSON name for the 'variable name' if you will. + +Now you can trivially walk the lists, recursively, and parse as you please. +You can invoke cJSON_Parse to get cJSON to parse for you, and then you can take +the root object, and traverse the structure (which is, formally, an N-tree), +and tokenise as you please. If you wanted to build a callback style parser, this is how +you'd do it (just an example, since these things are very specific): + +void parse_and_callback(cJSON *item,const char *prefix) +{ + while (item) + { + char *newprefix=malloc(strlen(prefix)+strlen(item->name)+2); + sprintf(newprefix,"%s/%s",prefix,item->name); + int dorecurse=callback(newprefix, item->type, item); + if (item->child && dorecurse) parse_and_callback(item->child,newprefix); + item=item->next; + free(newprefix); + } +} + +The prefix process will build you a separated list, to simplify your callback handling. +The 'dorecurse' flag would let the callback decide to handle sub-arrays on it's own, or +let you invoke it per-item. For the item above, your callback might look like this: + +int callback(const char *name,int type,cJSON *item) +{ + if (!strcmp(name,"name")) { /* populate name */ } + else if (!strcmp(name,"format/type") { /* handle "rect" */ } + else if (!strcmp(name,"format/width") { /* 800 */ } + else if (!strcmp(name,"format/height") { /* 600 */ } + else if (!strcmp(name,"format/interlace") { /* false */ } + else if (!strcmp(name,"format/frame rate") { /* 24 */ } + return 1; +} + +Alternatively, you might like to parse iteratively. +You'd use: + +void parse_object(cJSON *item) +{ + int i; for (i=0;i<cJSON_GetArraySize(item);i++) + { + cJSON *subitem=cJSON_GetArrayItem(item,i); + // handle subitem. + } +} + +Or, for PROPER manual mode: + +void parse_object(cJSON *item) +{ + cJSON *subitem=item->child; + while (subitem) + { + // handle subitem + if (subitem->child) parse_object(subitem->child); + + subitem=subitem->next; + } +} + +Of course, this should look familiar, since this is just a stripped-down version +of the callback-parser. + +This should cover most uses you'll find for parsing. The rest should be possible +to infer.. and if in doubt, read the source! There's not a lot of it! ;) + + +In terms of constructing JSON data, the example code above is the right way to do it. +You can, of course, hand your sub-objects to other functions to populate. +Also, if you find a use for it, you can manually build the objects. +For instance, suppose you wanted to build an array of objects? + +cJSON *objects[24]; + +cJSON *Create_array_of_anything(cJSON **items,int num) +{ + int i;cJSON *prev, *root=cJSON_CreateArray(); + for (i=0;i<24;i++) + { + if (!i) root->child=objects[i]; + else prev->next=objects[i], objects[i]->prev=prev; + prev=objects[i]; + } + return root; +} + +and simply: Create_array_of_anything(objects,24); + +cJSON doesn't make any assumptions about what order you create things in. +You can attach the objects, as above, and later add children to each +of those objects. + +As soon as you call cJSON_Print, it renders the structure to text. + + + +The test.c code shows how to handle a bunch of typical cases. If you uncomment +the code, it'll load, parse and print a bunch of test files, also from json.org, +which are more complex than I'd care to try and stash into a const char array[]. + + +Enjoy cJSON! + + +- Dave Gamble, Aug 2009 diff --git a/plugins/omelasticsearch/cJSON/cjson.c b/plugins/omelasticsearch/cJSON/cjson.c new file mode 100644 index 00000000..99a831e9 --- /dev/null +++ b/plugins/omelasticsearch/cJSON/cjson.c @@ -0,0 +1,514 @@ +/* + Copyright (c) 2009 Dave Gamble + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. +*/ + +/* cJSON */ +/* JSON parser in C. */ + +#include <string.h> +#include <stdio.h> +#include <math.h> +#include <stdlib.h> +#include <float.h> +#include <limits.h> +#include <ctype.h> +#include "cjson.h" + +static const char *ep; + +const char *cJSON_GetErrorPtr() {return ep;} + +static int cJSON_strcasecmp(const char *s1,const char *s2) +{ + if (!s1) return (s1==s2)?0:1;if (!s2) return 1; + for(; tolower(*s1) == tolower(*s2); ++s1, ++s2) if(*s1 == 0) return 0; + return tolower(*(const unsigned char *)s1) - tolower(*(const unsigned char *)s2); +} + +static void *(*cJSON_malloc)(size_t sz) = malloc; +static void (*cJSON_free)(void *ptr) = free; + +static char* cJSON_strdup(const char* str) +{ + size_t len; + char* copy; + + len = strlen(str) + 1; + if (!(copy = (char*)cJSON_malloc(len))) return 0; + memcpy(copy,str,len); + return copy; +} + +void cJSON_InitHooks(cJSON_Hooks* hooks) +{ + if (!hooks) { /* Reset hooks */ + cJSON_malloc = malloc; + cJSON_free = free; + return; + } + + cJSON_malloc = (hooks->malloc_fn)?hooks->malloc_fn:malloc; + cJSON_free = (hooks->free_fn)?hooks->free_fn:free; +} + +/* Internal constructor. */ +static cJSON *cJSON_New_Item() +{ + cJSON* node = (cJSON*)cJSON_malloc(sizeof(cJSON)); + if (node) memset(node,0,sizeof(cJSON)); + return node; +} + +/* Delete a cJSON structure. */ +void cJSON_Delete(cJSON *c) +{ + cJSON *next; + while (c) + { + next=c->next; + if (!(c->type&cJSON_IsReference) && c->child) cJSON_Delete(c->child); + if (!(c->type&cJSON_IsReference) && c->valuestring) cJSON_free(c->valuestring); + if (c->string) cJSON_free(c->string); + cJSON_free(c); + c=next; + } +} + +/* Parse the input text to generate a number, and populate the result into item. */ +static const char *parse_number(cJSON *item,const char *num) +{ + double n=0,sign=1,scale=0;int subscale=0,signsubscale=1; + + /* Could use sscanf for this? */ + if (*num=='-') sign=-1,num++; /* Has sign? */ + if (*num=='0') num++; /* is zero */ + if (*num>='1' && *num<='9') do n=(n*10.0)+(*num++ -'0'); while (*num>='0' && *num<='9'); /* Number? */ + if (*num=='.' && num[1]>='0' && num[1]<='9') {num++; do n=(n*10.0)+(*num++ -'0'),scale--; while (*num>='0' && *num<='9');} /* Fractional part? */ + if (*num=='e' || *num=='E') /* Exponent? */ + { num++;if (*num=='+') num++; else if (*num=='-') signsubscale=-1,num++; /* With sign? */ + while (*num>='0' && *num<='9') subscale=(subscale*10)+(*num++ - '0'); /* Number? */ + } + + n=sign*n*pow(10.0,(scale+subscale*signsubscale)); /* number = +/- number.fraction * 10^+/- exponent */ + + item->valuedouble=n; + item->valueint=(int)n; + item->type=cJSON_Number; + return num; +} + +/* Render the number nicely from the given item into a string. */ +char *cJSON_print_number(cJSON *item) +{ + char *str; + double d=item->valuedouble; + if (fabs(((double)item->valueint)-d)<=DBL_EPSILON && d<=INT_MAX && d>=INT_MIN) + { + str=(char*)cJSON_malloc(21); /* 2^64+1 can be represented in 21 chars. */ + if (str) sprintf(str,"%d",item->valueint); + } + else + { + str=(char*)cJSON_malloc(64); /* This is a nice tradeoff. */ + if (str) + { + if (fabs(floor(d)-d)<=DBL_EPSILON) sprintf(str,"%.0f",d); + else if (fabs(d)<1.0e-6 || fabs(d)>1.0e9) sprintf(str,"%e",d); + else sprintf(str,"%f",d); + } + } + return str; +} + +/* Parse the input text into an unescaped cstring, and populate item. */ +static const unsigned char firstByteMark[7] = { 0x00, 0x00, 0xC0, 0xE0, 0xF0, 0xF8, 0xFC }; +static const char *parse_string(cJSON *item,const char *str) +{ + const char *ptr=str+1;char *ptr2;char *out;int len=0;unsigned uc,uc2; + if (*str!='\"') {ep=str;return 0;} /* not a string! */ + + while (*ptr!='\"' && *ptr && ++len) if (*ptr++ == '\\') ptr++; /* Skip escaped quotes. */ + + out=(char*)cJSON_malloc(len+1); /* This is how long we need for the string, roughly. */ + if (!out) return 0; + + ptr=str+1;ptr2=out; + while (*ptr!='\"' && *ptr) + { + if (*ptr!='\\') *ptr2++=*ptr++; + else + { + ptr++; + switch (*ptr) + { + case 'b': *ptr2++='\b'; break; + case 'f': *ptr2++='\f'; break; + case 'n': *ptr2++='\n'; break; + case 'r': *ptr2++='\r'; break; + case 't': *ptr2++='\t'; break; + case 'u': /* transcode utf16 to utf8. */ + sscanf(ptr+1,"%4x",&uc);ptr+=4; /* get the unicode char. */ + + if ((uc>=0xDC00 && uc<=0xDFFF) || uc==0) break; // check for invalid. + + if (uc>=0xD800 && uc<=0xDBFF) // UTF16 surrogate pairs. + { + if (ptr[1]!='\\' || ptr[2]!='u') break; // missing second-half of surrogate. + sscanf(ptr+3,"%4x",&uc2);ptr+=6; + if (uc2<0xDC00 || uc2>0xDFFF) break; // invalid second-half of surrogate. + uc=0x10000 | ((uc&0x3FF)<<10) | (uc2&0x3FF); + } + + len=4;if (uc<0x80) len=1;else if (uc<0x800) len=2;else if (uc<0x10000) len=3; ptr2+=len; + + switch (len) { + case 4: *--ptr2 =((uc | 0x80) & 0xBF); uc >>= 6; + case 3: *--ptr2 =((uc | 0x80) & 0xBF); uc >>= 6; + case 2: *--ptr2 =((uc | 0x80) & 0xBF); uc >>= 6; + case 1: *--ptr2 =(uc | firstByteMark[len]); + } + ptr2+=len; + break; + default: *ptr2++=*ptr; break; + } + ptr++; + } + } + *ptr2=0; + if (*ptr=='\"') ptr++; + item->valuestring=out; + item->type=cJSON_String; + return ptr; +} + +/* Render the cstring provided to an escaped version that can be printed. */ +static char *print_string_ptr(const char *str) +{ + const char *ptr;char *ptr2,*out;int len=0;unsigned char token; + + if (!str) return cJSON_strdup(""); + ptr=str;while ((token=*ptr) && ++len) {if (strchr("\"\\\b\f\n\r\t",token)) len++; else if (token<32) len+=5;ptr++;} + + out=(char*)cJSON_malloc(len+3); + if (!out) return 0; + + ptr2=out;ptr=str; + *ptr2++='\"'; + while (*ptr) + { + if ((unsigned char)*ptr>31 && *ptr!='\"' && *ptr!='\\') *ptr2++=*ptr++; + else + { + *ptr2++='\\'; + switch (token=*ptr++) + { + case '\\': *ptr2++='\\'; break; + case '\"': *ptr2++='\"'; break; + case '\b': *ptr2++='b'; break; + case '\f': *ptr2++='f'; break; + case '\n': *ptr2++='n'; break; + case '\r': *ptr2++='r'; break; + case '\t': *ptr2++='t'; break; + default: sprintf(ptr2,"u%04x",token);ptr2+=5; break; /* escape and print */ + } + } + } + *ptr2++='\"';*ptr2++=0; + return out; +} +/* Invote print_string_ptr (which is useful) on an item. */ +static char *print_string(cJSON *item) {return print_string_ptr(item->valuestring);} + +/* Predeclare these prototypes. */ +static const char *parse_value(cJSON *item,const char *value); +static char *print_value(cJSON *item,int depth,int fmt); +static const char *parse_array(cJSON *item,const char *value); +static char *print_array(cJSON *item,int depth,int fmt); +static const char *parse_object(cJSON *item,const char *value); +static char *print_object(cJSON *item,int depth,int fmt); + +/* Utility to jump whitespace and cr/lf */ +static const char *skip(const char *in) {while (in && *in && (unsigned char)*in<=32) in++; return in;} + +/* Parse an object - create a new root, and populate. */ +cJSON *cJSON_Parse(const char *value) +{ + cJSON *c=cJSON_New_Item(); + ep=0; + if (!c) return 0; /* memory fail */ + + if (!parse_value(c,skip(value))) {cJSON_Delete(c);return 0;} + return c; +} + +/* Render a cJSON item/entity/structure to text. */ +char *cJSON_Print(cJSON *item) {return print_value(item,0,1);} +char *cJSON_PrintUnformatted(cJSON *item) {return print_value(item,0,0);} + +/* Parser core - when encountering text, process appropriately. */ +static const char *parse_value(cJSON *item,const char *value) +{ + if (!value) return 0; /* Fail on null. */ + if (!strncmp(value,"null",4)) { item->type=cJSON_NULL; return value+4; } + if (!strncmp(value,"false",5)) { item->type=cJSON_False; return value+5; } + if (!strncmp(value,"true",4)) { item->type=cJSON_True; item->valueint=1; return value+4; } + if (*value=='\"') { return parse_string(item,value); } + if (*value=='-' || (*value>='0' && *value<='9')) { return parse_number(item,value); } + if (*value=='[') { return parse_array(item,value); } + if (*value=='{') { return parse_object(item,value); } + + ep=value;return 0; /* failure. */ +} + +/* Render a value to text. */ +static char *print_value(cJSON *item,int depth,int fmt) +{ + char *out=0; + if (!item) return 0; + switch ((item->type)&255) + { + case cJSON_NULL: out=cJSON_strdup("null"); break; + case cJSON_False: out=cJSON_strdup("false");break; + case cJSON_True: out=cJSON_strdup("true"); break; + case cJSON_Number: out=cJSON_print_number(item);break; + case cJSON_String: out=print_string(item);break; + case cJSON_Array: out=print_array(item,depth,fmt);break; + case cJSON_Object: out=print_object(item,depth,fmt);break; + } + return out; +} + +/* Build an array from input text. */ +static const char *parse_array(cJSON *item,const char *value) +{ + cJSON *child; + if (*value!='[') {ep=value;return 0;} /* not an array! */ + + item->type=cJSON_Array; + value=skip(value+1); + if (*value==']') return value+1; /* empty array. */ + + item->child=child=cJSON_New_Item(); + if (!item->child) return 0; /* memory fail */ + value=skip(parse_value(child,skip(value))); /* skip any spacing, get the value. */ + if (!value) return 0; + + while (*value==',') + { + cJSON *new_item; + if (!(new_item=cJSON_New_Item())) return 0; /* memory fail */ + child->next=new_item;new_item->prev=child;child=new_item; + value=skip(parse_value(child,skip(value+1))); + if (!value) return 0; /* memory fail */ + } + + if (*value==']') return value+1; /* end of array */ + ep=value;return 0; /* malformed. */ +} + +/* Render an array to text */ +static char *print_array(cJSON *item,int depth,int fmt) +{ + char **entries; + char *out=0,*ptr,*ret;int len=5; + cJSON *child=item->child; + int numentries=0,i=0,fail=0; + + /* How many entries in the array? */ + while (child) numentries++,child=child->next; + /* Allocate an array to hold the values for each */ + entries=(char**)cJSON_malloc(numentries*sizeof(char*)); + if (!entries) return 0; + memset(entries,0,numentries*sizeof(char*)); + /* Retrieve all the results: */ + child=item->child; + while (child && !fail) + { + ret=print_value(child,depth+1,fmt); + entries[i++]=ret; + if (ret) len+=strlen(ret)+2+(fmt?1:0); else fail=1; + child=child->next; + } + + /* If we didn't fail, try to malloc the output string */ + if (!fail) out=(char*)cJSON_malloc(len); + /* If that fails, we fail. */ + if (!out) fail=1; + + /* Handle failure. */ + if (fail) + { + for (i=0;i<numentries;i++) if (entries[i]) cJSON_free(entries[i]); + cJSON_free(entries); + return 0; + } + + /* Compose the output array. */ + *out='['; + ptr=out+1;*ptr=0; + for (i=0;i<numentries;i++) + { + strcpy(ptr,entries[i]);ptr+=strlen(entries[i]); + if (i!=numentries-1) {*ptr++=',';if(fmt)*ptr++=' ';*ptr=0;} + cJSON_free(entries[i]); + } + cJSON_free(entries); + *ptr++=']';*ptr++=0; + return out; +} + +/* Build an object from the text. */ +static const char *parse_object(cJSON *item,const char *value) +{ + cJSON *child; + if (*value!='{') {ep=value;return 0;} /* not an object! */ + + item->type=cJSON_Object; + value=skip(value+1); + if (*value=='}') return value+1; /* empty array. */ + + item->child=child=cJSON_New_Item(); + if (!item->child) return 0; + value=skip(parse_string(child,skip(value))); + if (!value) return 0; + child->string=child->valuestring;child->valuestring=0; + if (*value!=':') {ep=value;return 0;} /* fail! */ + value=skip(parse_value(child,skip(value+1))); /* skip any spacing, get the value. */ + if (!value) return 0; + + while (*value==',') + { + cJSON *new_item; + if (!(new_item=cJSON_New_Item())) return 0; /* memory fail */ + child->next=new_item;new_item->prev=child;child=new_item; + value=skip(parse_string(child,skip(value+1))); + if (!value) return 0; + child->string=child->valuestring;child->valuestring=0; + if (*value!=':') {ep=value;return 0;} /* fail! */ + value=skip(parse_value(child,skip(value+1))); /* skip any spacing, get the value. */ + if (!value) return 0; + } + + if (*value=='}') return value+1; /* end of array */ + ep=value;return 0; /* malformed. */ +} + +/* Render an object to text. */ +static char *print_object(cJSON *item,int depth,int fmt) +{ + char **entries=0,**names=0; + char *out=0,*ptr,*ret,*str;int len=7,i=0,j; + cJSON *child=item->child; + int numentries=0,fail=0; + /* Count the number of entries. */ + while (child) numentries++,child=child->next; + /* Allocate space for the names and the objects */ + entries=(char**)cJSON_malloc(numentries*sizeof(char*)); + if (!entries) return 0; + names=(char**)cJSON_malloc(numentries*sizeof(char*)); + if (!names) {cJSON_free(entries);return 0;} + memset(entries,0,sizeof(char*)*numentries); + memset(names,0,sizeof(char*)*numentries); + + /* Collect all the results into our arrays: */ + child=item->child;depth++;if (fmt) len+=depth; + while (child) + { + names[i]=str=print_string_ptr(child->string); + entries[i++]=ret=print_value(child,depth,fmt); + if (str && ret) len+=strlen(ret)+strlen(str)+2+(fmt?2+depth:0); else fail=1; + child=child->next; + } + + /* Try to allocate the output string */ + if (!fail) out=(char*)cJSON_malloc(len); + if (!out) fail=1; + + /* Handle failure */ + if (fail) + { + for (i=0;i<numentries;i++) {if (names[i]) cJSON_free(names[i]);if (entries[i]) cJSON_free(entries[i]);} + cJSON_free(names);cJSON_free(entries); + return 0; + } + + /* Compose the output: */ + *out='{';ptr=out+1;if (fmt)*ptr++='\n';*ptr=0; + for (i=0;i<numentries;i++) + { + if (fmt) for (j=0;j<depth;j++) *ptr++='\t'; + strcpy(ptr,names[i]);ptr+=strlen(names[i]); + *ptr++=':';if (fmt) *ptr++='\t'; + strcpy(ptr,entries[i]);ptr+=strlen(entries[i]); + if (i!=numentries-1) *ptr++=','; + if (fmt) *ptr++='\n';*ptr=0; + cJSON_free(names[i]);cJSON_free(entries[i]); + } + + cJSON_free(names);cJSON_free(entries); + if (fmt) for (i=0;i<depth-1;i++) *ptr++='\t'; + *ptr++='}';*ptr++=0; + return out; +} + +/* Get Array size/item / object item. */ +int cJSON_GetArraySize(cJSON *array) {cJSON *c=array->child;int i=0;while(c)i++,c=c->next;return i;} +cJSON *cJSON_GetArrayItem(cJSON *array,int item) {cJSON *c=array->child; while (c && item>0) item--,c=c->next; return c;} +cJSON *cJSON_GetObjectItem(cJSON *object,const char *string) {cJSON *c=object->child; while (c && cJSON_strcasecmp(c->string,string)) c=c->next; return c;} + +/* Utility for array list handling. */ +static void suffix_object(cJSON *prev,cJSON *item) {prev->next=item;item->prev=prev;} +/* Utility for handling references. */ +static cJSON *create_reference(cJSON *item) {cJSON *ref=cJSON_New_Item();if (!ref) return 0;memcpy(ref,item,sizeof(cJSON));ref->string=0;ref->type|=cJSON_IsReference;ref->next=ref->prev=0;return ref;} + +/* Add item to array/object. */ +void cJSON_AddItemToArray(cJSON *array, cJSON *item) {cJSON *c=array->child;if (!item) return; if (!c) {array->child=item;} else {while (c && c->next) c=c->next; suffix_object(c,item);}} +void cJSON_AddItemToObject(cJSON *object,const char *string,cJSON *item) {if (!item) return; if (item->string) cJSON_free(item->string);item->string=cJSON_strdup(string);cJSON_AddItemToArray(object,item);} +void cJSON_AddItemReferenceToArray(cJSON *array, cJSON *item) {cJSON_AddItemToArray(array,create_reference(item));} +void cJSON_AddItemReferenceToObject(cJSON *object,const char *string,cJSON *item) {cJSON_AddItemToObject(object,string,create_reference(item));} + +cJSON *cJSON_DetachItemFromArray(cJSON *array,int which) {cJSON *c=array->child;while (c && which>0) c=c->next,which--;if (!c) return 0; + if (c->prev) c->prev->next=c->next;if (c->next) c->next->prev=c->prev;if (c==array->child) array->child=c->next;c->prev=c->next=0;return c;} +void cJSON_DeleteItemFromArray(cJSON *array,int which) {cJSON_Delete(cJSON_DetachItemFromArray(array,which));} +cJSON *cJSON_DetachItemFromObject(cJSON *object,const char *string) {int i=0;cJSON *c=object->child;while (c && cJSON_strcasecmp(c->string,string)) i++,c=c->next;if (c) return cJSON_DetachItemFromArray(object,i);return 0;} +void cJSON_DeleteItemFromObject(cJSON *object,const char *string) {cJSON_Delete(cJSON_DetachItemFromObject(object,string));} + +/* Replace array/object items with new ones. */ +void cJSON_ReplaceItemInArray(cJSON *array,int which,cJSON *newitem) {cJSON *c=array->child;while (c && which>0) c=c->next,which--;if (!c) return; + newitem->next=c->next;newitem->prev=c->prev;if (newitem->next) newitem->next->prev=newitem; + if (c==array->child) array->child=newitem; else newitem->prev->next=newitem;c->next=c->prev=0;cJSON_Delete(c);} +void cJSON_ReplaceItemInObject(cJSON *object,const char *string,cJSON *newitem){int i=0;cJSON *c=object->child;while(c && cJSON_strcasecmp(c->string,string))i++,c=c->next;if(c){newitem->string=cJSON_strdup(string);cJSON_ReplaceItemInArray(object,i,newitem);}} + +/* Create basic types: */ +cJSON *cJSON_CreateNull() {cJSON *item=cJSON_New_Item();if(item)item->type=cJSON_NULL;return item;} +cJSON *cJSON_CreateTrue() {cJSON *item=cJSON_New_Item();if(item)item->type=cJSON_True;return item;} +cJSON *cJSON_CreateFalse() {cJSON *item=cJSON_New_Item();if(item)item->type=cJSON_False;return item;} +cJSON *cJSON_CreateBool(int b) {cJSON *item=cJSON_New_Item();if(item)item->type=b?cJSON_True:cJSON_False;return item;} +cJSON *cJSON_CreateNumber(double num) {cJSON *item=cJSON_New_Item();if(item){item->type=cJSON_Number;item->valuedouble=num;item->valueint=(int)num;}return item;} +cJSON *cJSON_CreateString(const char *string) {cJSON *item=cJSON_New_Item();if(item){item->type=cJSON_String;item->valuestring=cJSON_strdup(string);}return item;} +cJSON *cJSON_CreateArray() {cJSON *item=cJSON_New_Item();if(item)item->type=cJSON_Array;return item;} +cJSON *cJSON_CreateObject() {cJSON *item=cJSON_New_Item();if(item)item->type=cJSON_Object;return item;} + +/* Create Arrays: */ +cJSON *cJSON_CreateIntArray(int *numbers,int count) {int i;cJSON *n=0,*p=0,*a=cJSON_CreateArray();for(i=0;a && i<count;i++){n=cJSON_CreateNumber(numbers[i]);if(!i)a->child=n;else suffix_object(p,n);p=n;}return a;} +cJSON *cJSON_CreateFloatArray(float *numbers,int count) {int i;cJSON *n=0,*p=0,*a=cJSON_CreateArray();for(i=0;a && i<count;i++){n=cJSON_CreateNumber(numbers[i]);if(!i)a->child=n;else suffix_object(p,n);p=n;}return a;} +cJSON *cJSON_CreateDoubleArray(double *numbers,int count) {int i;cJSON *n=0,*p=0,*a=cJSON_CreateArray();for(i=0;a && i<count;i++){n=cJSON_CreateNumber(numbers[i]);if(!i)a->child=n;else suffix_object(p,n);p=n;}return a;} +cJSON *cJSON_CreateStringArray(const char **strings,int count) {int i;cJSON *n=0,*p=0,*a=cJSON_CreateArray();for(i=0;a && i<count;i++){n=cJSON_CreateString(strings[i]);if(!i)a->child=n;else suffix_object(p,n);p=n;}return a;} diff --git a/plugins/omelasticsearch/cJSON/cjson.h b/plugins/omelasticsearch/cJSON/cjson.h new file mode 100644 index 00000000..a621720c --- /dev/null +++ b/plugins/omelasticsearch/cJSON/cjson.h @@ -0,0 +1,130 @@ +/* + Copyright (c) 2009 Dave Gamble + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. +*/ + +#ifndef cJSON__h +#define cJSON__h + +#ifdef __cplusplus +extern "C" +{ +#endif + +/* cJSON Types: */ +#define cJSON_False 0 +#define cJSON_True 1 +#define cJSON_NULL 2 +#define cJSON_Number 3 +#define cJSON_String 4 +#define cJSON_Array 5 +#define cJSON_Object 6 + +#define cJSON_IsReference 256 + +/* The cJSON structure: */ +typedef struct cJSON { + struct cJSON *next,*prev; /* next/prev allow you to walk array/object chains. Alternatively, use GetArraySize/GetArrayItem/GetObjectItem */ + struct cJSON *child; /* An array or object item will have a child pointer pointing to a chain of the items in the array/object. */ + + int type; /* The type of the item, as above. */ + + char *valuestring; /* The item's string, if type==cJSON_String */ + int valueint; /* The item's number, if type==cJSON_Number */ + double valuedouble; /* The item's number, if type==cJSON_Number */ + + char *string; /* The item's name string, if this item is the child of, or is in the list of subitems of an object. */ +} cJSON; + +typedef struct cJSON_Hooks { + void *(*malloc_fn)(size_t sz); + void (*free_fn)(void *ptr); +} cJSON_Hooks; + +/* Supply malloc, realloc and free functions to cJSON */ +extern void cJSON_InitHooks(cJSON_Hooks* hooks); + + +/* Supply a block of JSON, and this returns a cJSON object you can interrogate. Call cJSON_Delete when finished. */ +extern cJSON *cJSON_Parse(const char *value); +/* Render a cJSON entity to text for transfer/storage. Free the char* when finished. */ +extern char *cJSON_Print(cJSON *item); +/* Render a cJSON entity to text for transfer/storage without any formatting. Free the char* when finished. */ +extern char *cJSON_PrintUnformatted(cJSON *item); +/* Delete a cJSON entity and all subentities. */ +extern void cJSON_Delete(cJSON *c); + +/* Returns the number of items in an array (or object). */ +extern int cJSON_GetArraySize(cJSON *array); +/* Retrieve item number "item" from array "array". Returns NULL if unsuccessful. */ +extern cJSON *cJSON_GetArrayItem(cJSON *array,int item); +/* Get item "string" from object. Case insensitive. */ +extern cJSON *cJSON_GetObjectItem(cJSON *object,const char *string); + +/* For analysing failed parses. This returns a pointer to the parse error. You'll probably need to look a few chars back to make sense of it. Defined when cJSON_Parse() returns 0. 0 when cJSON_Parse() succeeds. */ +extern const char *cJSON_GetErrorPtr(); + +/* These calls create a cJSON item of the appropriate type. */ +extern cJSON *cJSON_CreateNull(); +extern cJSON *cJSON_CreateTrue(); +extern cJSON *cJSON_CreateFalse(); +extern cJSON *cJSON_CreateBool(int b); +extern cJSON *cJSON_CreateNumber(double num); +extern cJSON *cJSON_CreateString(const char *string); +extern cJSON *cJSON_CreateArray(); +extern cJSON *cJSON_CreateObject(); + +/* These utilities create an Array of count items. */ +extern cJSON *cJSON_CreateIntArray(int *numbers,int count); +extern cJSON *cJSON_CreateFloatArray(float *numbers,int count); +extern cJSON *cJSON_CreateDoubleArray(double *numbers,int count); +extern cJSON *cJSON_CreateStringArray(const char **strings,int count); + +/* Append item to the specified array/object. */ +extern void cJSON_AddItemToArray(cJSON *array, cJSON *item); +extern void cJSON_AddItemToObject(cJSON *object,const char *string,cJSON *item); +/* Append reference to item to the specified array/object. Use this when you want to add an existing cJSON to a new cJSON, but don't want to corrupt your existing cJSON. */ +extern void cJSON_AddItemReferenceToArray(cJSON *array, cJSON *item); +extern void cJSON_AddItemReferenceToObject(cJSON *object,const char *string,cJSON *item); + +/* Remove/Detatch items from Arrays/Objects. */ +extern cJSON *cJSON_DetachItemFromArray(cJSON *array,int which); +extern void cJSON_DeleteItemFromArray(cJSON *array,int which); +extern cJSON *cJSON_DetachItemFromObject(cJSON *object,const char *string); +extern void cJSON_DeleteItemFromObject(cJSON *object,const char *string); + +/* Update array items. */ +extern void cJSON_ReplaceItemInArray(cJSON *array,int which,cJSON *newitem); +extern void cJSON_ReplaceItemInObject(cJSON *object,const char *string,cJSON *newitem); + +/* rger: added helpers */ + +char *cJSON_print_number(cJSON *item); +#define cJSON_AddNullToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateNull()) +#define cJSON_AddTrueToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateTrue()) +#define cJSON_AddFalseToObject(object,name) cJSON_AddItemToObject(object, name, cJSON_CreateFalse()) +#define cJSON_AddNumberToObject(object,name,n) cJSON_AddItemToObject(object, name, cJSON_CreateNumber(n)) +#define cJSON_AddStringToObject(object,name,s) cJSON_AddItemToObject(object, name, cJSON_CreateString(s)) + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/omelasticsearch/cJSON/test.c b/plugins/omelasticsearch/cJSON/test.c new file mode 100644 index 00000000..2cab632a --- /dev/null +++ b/plugins/omelasticsearch/cJSON/test.c @@ -0,0 +1,156 @@ +/* + Copyright (c) 2009 Dave Gamble + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. +*/ + +#include <stdio.h> +#include <stdlib.h> +#include "cJSON.h" + +/* Parse text to JSON, then render back to text, and print! */ +void doit(char *text) +{ + char *out;cJSON *json; + + json=cJSON_Parse(text); + if (!json) {printf("Error before: [%s]\n",cJSON_GetErrorPtr());} + else + { + out=cJSON_Print(json); + cJSON_Delete(json); + printf("%s\n",out); + free(out); + } +} + +/* Read a file, parse, render back, etc. */ +void dofile(char *filename) +{ + FILE *f=fopen(filename,"rb");fseek(f,0,SEEK_END);long len=ftell(f);fseek(f,0,SEEK_SET); + char *data=malloc(len+1);fread(data,1,len,f);fclose(f); + doit(data); + free(data); +} + +/* Used by some code below as an example datatype. */ +struct record {const char *precision;double lat,lon;const char *address,*city,*state,*zip,*country; }; + +/* Create a bunch of objects as demonstration. */ +void create_objects() +{ + cJSON *root,*fmt,*img,*thm,*fld;char *out;int i; /* declare a few. */ + + /* Here we construct some JSON standards, from the JSON site. */ + + /* Our "Video" datatype: */ + root=cJSON_CreateObject(); + cJSON_AddItemToObject(root, "name", cJSON_CreateString("Jack (\"Bee\") Nimble")); + cJSON_AddItemToObject(root, "format", fmt=cJSON_CreateObject()); + cJSON_AddStringToObject(fmt,"type", "rect"); + cJSON_AddNumberToObject(fmt,"width", 1920); + cJSON_AddNumberToObject(fmt,"height", 1080); + cJSON_AddFalseToObject (fmt,"interlace"); + cJSON_AddNumberToObject(fmt,"frame rate", 24); + + out=cJSON_Print(root); cJSON_Delete(root); printf("%s\n",out); free(out); /* Print to text, Delete the cJSON, print it, release the string. + + /* Our "days of the week" array: */ + const char *strings[7]={"Sunday","Monday","Tuesday","Wednesday","Thursday","Friday","Saturday"}; + root=cJSON_CreateStringArray(strings,7); + + out=cJSON_Print(root); cJSON_Delete(root); printf("%s\n",out); free(out); + + /* Our matrix: */ + int numbers[3][3]={{0,-1,0},{1,0,0},{0,0,1}}; + root=cJSON_CreateArray(); + for (i=0;i<3;i++) cJSON_AddItemToArray(root,cJSON_CreateIntArray(numbers[i],3)); + +/* cJSON_ReplaceItemInArray(root,1,cJSON_CreateString("Replacement")); */ + + out=cJSON_Print(root); cJSON_Delete(root); printf("%s\n",out); free(out); + + + /* Our "gallery" item: */ + int ids[4]={116,943,234,38793}; + root=cJSON_CreateObject(); + cJSON_AddItemToObject(root, "Image", img=cJSON_CreateObject()); + cJSON_AddNumberToObject(img,"Width",800); + cJSON_AddNumberToObject(img,"Height",600); + cJSON_AddStringToObject(img,"Title","View from 15th Floor"); + cJSON_AddItemToObject(img, "Thumbnail", thm=cJSON_CreateObject()); + cJSON_AddStringToObject(thm, "Url", "http:/*www.example.com/image/481989943"); + cJSON_AddNumberToObject(thm,"Height",125); + cJSON_AddStringToObject(thm,"Width","100"); + cJSON_AddItemToObject(img,"IDs", cJSON_CreateIntArray(ids,4)); + + out=cJSON_Print(root); cJSON_Delete(root); printf("%s\n",out); free(out); + + /* Our array of "records": */ + struct record fields[2]={ + {"zip",37.7668,-1.223959e+2,"","SAN FRANCISCO","CA","94107","US"}, + {"zip",37.371991,-1.22026e+2,"","SUNNYVALE","CA","94085","US"}}; + + root=cJSON_CreateArray(); + for (i=0;i<2;i++) + { + cJSON_AddItemToArray(root,fld=cJSON_CreateObject()); + cJSON_AddStringToObject(fld, "precision", fields[i].precision); + cJSON_AddNumberToObject(fld, "Latitude", fields[i].lat); + cJSON_AddNumberToObject(fld, "Longitude", fields[i].lon); + cJSON_AddStringToObject(fld, "Address", fields[i].address); + cJSON_AddStringToObject(fld, "City", fields[i].city); + cJSON_AddStringToObject(fld, "State", fields[i].state); + cJSON_AddStringToObject(fld, "Zip", fields[i].zip); + cJSON_AddStringToObject(fld, "Country", fields[i].country); + } + +/* cJSON_ReplaceItemInObject(cJSON_GetArrayItem(root,1),"City",cJSON_CreateIntArray(ids,4)); */ + + out=cJSON_Print(root); cJSON_Delete(root); printf("%s\n",out); free(out); + +} + +int main (int argc, const char * argv[]) { + /* a bunch of json: */ + char text1[]="{\n\"name\": \"Jack (\\\"Bee\\\") Nimble\", \n\"format\": {\"type\": \"rect\", \n\"width\": 1920, \n\"height\": 1080, \n\"interlace\": false,\"frame rate\": 24\n}\n}"; + char text2[]="[\"Sunday\", \"Monday\", \"Tuesday\", \"Wednesday\", \"Thursday\", \"Friday\", \"Saturday\"]"; + char text3[]="[\n [0, -1, 0],\n [1, 0, 0],\n [0, 0, 1]\n ]\n"; + char text4[]="{\n \"Image\": {\n \"Width\": 800,\n \"Height\": 600,\n \"Title\": \"View from 15th Floor\",\n \"Thumbnail\": {\n \"Url\": \"http:/*www.example.com/image/481989943\",\n \"Height\": 125,\n \"Width\": \"100\"\n },\n \"IDs\": [116, 943, 234, 38793]\n }\n }"; + char text5[]="[\n {\n \"precision\": \"zip\",\n \"Latitude\": 37.7668,\n \"Longitude\": -122.3959,\n \"Address\": \"\",\n \"City\": \"SAN FRANCISCO\",\n \"State\": \"CA\",\n \"Zip\": \"94107\",\n \"Country\": \"US\"\n },\n {\n \"precision\": \"zip\",\n \"Latitude\": 37.371991,\n \"Longitude\": -122.026020,\n \"Address\": \"\",\n \"City\": \"SUNNYVALE\",\n \"State\": \"CA\",\n \"Zip\": \"94085\",\n \"Country\": \"US\"\n }\n ]"; + + /* Process each json textblock by parsing, then rebuilding: */ + doit(text1); + doit(text2); + doit(text3); + doit(text4); + doit(text5); + + /* Parse standard testfiles: +/* dofile("../../tests/test1"); */ +/* dofile("../../tests/test2"); */ +/* dofile("../../tests/test3"); */ +/* dofile("../../tests/test4"); */ +/* dofile("../../tests/test5"); */ + + /* Now some samplecode for building objects concisely: */ + create_objects(); + + return 0; +} diff --git a/plugins/omelasticsearch/cJSON/tests/test1 b/plugins/omelasticsearch/cJSON/tests/test1 new file mode 100644 index 00000000..eacfbf5e --- /dev/null +++ b/plugins/omelasticsearch/cJSON/tests/test1 @@ -0,0 +1,22 @@ +{ + "glossary": { + "title": "example glossary", + "GlossDiv": { + "title": "S", + "GlossList": { + "GlossEntry": { + "ID": "SGML", + "SortAs": "SGML", + "GlossTerm": "Standard Generalized Markup Language", + "Acronym": "SGML", + "Abbrev": "ISO 8879:1986", + "GlossDef": { + "para": "A meta-markup language, used to create markup languages such as DocBook.", + "GlossSeeAlso": ["GML", "XML"] + }, + "GlossSee": "markup" + } + } + } + } +} diff --git a/plugins/omelasticsearch/cJSON/tests/test2 b/plugins/omelasticsearch/cJSON/tests/test2 new file mode 100644 index 00000000..5600991a --- /dev/null +++ b/plugins/omelasticsearch/cJSON/tests/test2 @@ -0,0 +1,11 @@ +{"menu": { + "id": "file", + "value": "File", + "popup": { + "menuitem": [ + {"value": "New", "onclick": "CreateNewDoc()"}, + {"value": "Open", "onclick": "OpenDoc()"}, + {"value": "Close", "onclick": "CloseDoc()"} + ] + } +}} diff --git a/plugins/omelasticsearch/cJSON/tests/test3 b/plugins/omelasticsearch/cJSON/tests/test3 new file mode 100644 index 00000000..5662b377 --- /dev/null +++ b/plugins/omelasticsearch/cJSON/tests/test3 @@ -0,0 +1,26 @@ +{"widget": { + "debug": "on", + "window": { + "title": "Sample Konfabulator Widget", + "name": "main_window", + "width": 500, + "height": 500 + }, + "image": { + "src": "Images/Sun.png", + "name": "sun1", + "hOffset": 250, + "vOffset": 250, + "alignment": "center" + }, + "text": { + "data": "Click Here", + "size": 36, + "style": "bold", + "name": "text1", + "hOffset": 250, + "vOffset": 100, + "alignment": "center", + "onMouseUp": "sun1.opacity = (sun1.opacity / 100) * 90;" + } +}}
\ No newline at end of file diff --git a/plugins/omelasticsearch/cJSON/tests/test4 b/plugins/omelasticsearch/cJSON/tests/test4 new file mode 100644 index 00000000..d540b57f --- /dev/null +++ b/plugins/omelasticsearch/cJSON/tests/test4 @@ -0,0 +1,88 @@ +{"web-app": { + "servlet": [ + { + "servlet-name": "cofaxCDS", + "servlet-class": "org.cofax.cds.CDSServlet", + "init-param": { + "configGlossary:installationAt": "Philadelphia, PA", + "configGlossary:adminEmail": "ksm@pobox.com", + "configGlossary:poweredBy": "Cofax", + "configGlossary:poweredByIcon": "/images/cofax.gif", + "configGlossary:staticPath": "/content/static", + "templateProcessorClass": "org.cofax.WysiwygTemplate", + "templateLoaderClass": "org.cofax.FilesTemplateLoader", + "templatePath": "templates", + "templateOverridePath": "", + "defaultListTemplate": "listTemplate.htm", + "defaultFileTemplate": "articleTemplate.htm", + "useJSP": false, + "jspListTemplate": "listTemplate.jsp", + "jspFileTemplate": "articleTemplate.jsp", + "cachePackageTagsTrack": 200, + "cachePackageTagsStore": 200, + "cachePackageTagsRefresh": 60, + "cacheTemplatesTrack": 100, + "cacheTemplatesStore": 50, + "cacheTemplatesRefresh": 15, + "cachePagesTrack": 200, + "cachePagesStore": 100, + "cachePagesRefresh": 10, + "cachePagesDirtyRead": 10, + "searchEngineListTemplate": "forSearchEnginesList.htm", + "searchEngineFileTemplate": "forSearchEngines.htm", + "searchEngineRobotsDb": "WEB-INF/robots.db", + "useDataStore": true, + "dataStoreClass": "org.cofax.SqlDataStore", + "redirectionClass": "org.cofax.SqlRedirection", + "dataStoreName": "cofax", + "dataStoreDriver": "com.microsoft.jdbc.sqlserver.SQLServerDriver", + "dataStoreUrl": "jdbc:microsoft:sqlserver://LOCALHOST:1433;DatabaseName=goon", + "dataStoreUser": "sa", + "dataStorePassword": "dataStoreTestQuery", + "dataStoreTestQuery": "SET NOCOUNT ON;select test='test';", + "dataStoreLogFile": "/usr/local/tomcat/logs/datastore.log", + "dataStoreInitConns": 10, + "dataStoreMaxConns": 100, + "dataStoreConnUsageLimit": 100, + "dataStoreLogLevel": "debug", + "maxUrlLength": 500}}, + { + "servlet-name": "cofaxEmail", + "servlet-class": "org.cofax.cds.EmailServlet", + "init-param": { + "mailHost": "mail1", + "mailHostOverride": "mail2"}}, + { + "servlet-name": "cofaxAdmin", + "servlet-class": "org.cofax.cds.AdminServlet"}, + + { + "servlet-name": "fileServlet", + "servlet-class": "org.cofax.cds.FileServlet"}, + { + "servlet-name": "cofaxTools", + "servlet-class": "org.cofax.cms.CofaxToolsServlet", + "init-param": { + "templatePath": "toolstemplates/", + "log": 1, + "logLocation": "/usr/local/tomcat/logs/CofaxTools.log", + "logMaxSize": "", + "dataLog": 1, + "dataLogLocation": "/usr/local/tomcat/logs/dataLog.log", + "dataLogMaxSize": "", + "removePageCache": "/content/admin/remove?cache=pages&id=", + "removeTemplateCache": "/content/admin/remove?cache=templates&id=", + "fileTransferFolder": "/usr/local/tomcat/webapps/content/fileTransferFolder", + "lookInContext": 1, + "adminGroupID": 4, + "betaServer": true}}], + "servlet-mapping": { + "cofaxCDS": "/", + "cofaxEmail": "/cofaxutil/aemail/*", + "cofaxAdmin": "/admin/*", + "fileServlet": "/static/*", + "cofaxTools": "/tools/*"}, + + "taglib": { + "taglib-uri": "cofax.tld", + "taglib-location": "/WEB-INF/tlds/cofax.tld"}}}
\ No newline at end of file diff --git a/plugins/omelasticsearch/cJSON/tests/test5 b/plugins/omelasticsearch/cJSON/tests/test5 new file mode 100644 index 00000000..49980ca2 --- /dev/null +++ b/plugins/omelasticsearch/cJSON/tests/test5 @@ -0,0 +1,27 @@ +{"menu": { + "header": "SVG Viewer", + "items": [ + {"id": "Open"}, + {"id": "OpenNew", "label": "Open New"}, + null, + {"id": "ZoomIn", "label": "Zoom In"}, + {"id": "ZoomOut", "label": "Zoom Out"}, + {"id": "OriginalView", "label": "Original View"}, + null, + {"id": "Quality"}, + {"id": "Pause"}, + {"id": "Mute"}, + null, + {"id": "Find", "label": "Find..."}, + {"id": "FindAgain", "label": "Find Again"}, + {"id": "Copy"}, + {"id": "CopyAgain", "label": "Copy Again"}, + {"id": "CopySVG", "label": "Copy SVG"}, + {"id": "ViewSVG", "label": "View SVG"}, + {"id": "ViewSource", "label": "View Source"}, + {"id": "SaveAs", "label": "Save As"}, + null, + {"id": "Help"}, + {"id": "About", "label": "About Adobe CVG Viewer..."} + ] +}} diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 00e4dbac..f27fe62b 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -34,6 +34,10 @@ #include <signal.h> #include <errno.h> #include <time.h> +#include <sys/types.h> +#include <sys/stat.h> +#include <fcntl.h> +#include "cJSON/cjson.h" #include "conf.h" #include "syslogd-types.h" #include "srUtils.h" @@ -64,8 +68,10 @@ STATSCOUNTER_DEF(indexSuccess, mutIndexSuccess) */ typedef struct curl_slist HEADER; typedef struct _instanceData { - uchar *server; int port; + int replyLen; + int fdErrFile; /* error file fd or -1 if not open */ + uchar *server; uchar *uid; uchar *pwd; uchar *searchIndex; @@ -73,6 +79,9 @@ typedef struct _instanceData { uchar *parent; uchar *tplName; uchar *timeout; + uchar *restURL; /* last used URL for error reporting */ + uchar *errorFile; + char *reply; sbool dynSrchIdx; sbool dynSrchType; sbool dynParent; @@ -104,6 +113,7 @@ static struct cnfparamdescr actpdescr[] = { { "bulkmode", eCmdHdlrBinary, 0 }, { "asyncrepl", eCmdHdlrBinary, 0 }, { "timeout", eCmdHdlrGetWord, 0 }, + { "errorfile", eCmdHdlrGetWord, 0 }, { "template", eCmdHdlrGetWord, 1 } }; static struct cnfparamblk actpblk = @@ -114,6 +124,8 @@ static struct cnfparamblk actpblk = BEGINcreateInstance CODESTARTcreateInstance + pData->restURL = NULL; + pData->fdErrFile = -1; ENDcreateInstance BEGINisCompatibleWithFeature @@ -132,6 +144,8 @@ CODESTARTfreeInstance curl_easy_cleanup(pData->curlHandle); pData->curlHandle = NULL; } + if(pData->fdErrFile != -1) + close(pData->fdErrFile); free(pData->server); free(pData->uid); free(pData->pwd); @@ -139,6 +153,9 @@ CODESTARTfreeInstance free(pData->searchType); free(pData->parent); free(pData->tplName); + free(pData->timeout); + free(pData->restURL); + free(pData->errorFile); ENDfreeInstance BEGINdbgPrintInstInfo @@ -158,6 +175,8 @@ CODESTARTdbgPrintInstInfo dbgprintf("\tdynamic parent=%d\n", pData->dynParent); dbgprintf("\tasync replication=%d\n", pData->asyncRepl); dbgprintf("\tbulkmode=%d\n", pData->bulkmode); + dbgprintf("\terrorfile='%s'\n", pData->errorFile == NULL ? + (uchar*)"(not configured)" : pData->errorFile); ENDdbgPrintInstInfo @@ -202,12 +221,16 @@ checkConn(instanceData *pData) curl_easy_setopt(curl, CURLOPT_URL, cstr); free(cstr); + pData->reply = NULL; + pData->replyLen = 0; + curl_easy_setopt(curl, CURLOPT_WRITEDATA, pData); res = curl_easy_perform(curl); if(res != CURLE_OK) { DBGPRINTF("omelasticsearch: checkConn() curl_easy_perform() " "failed: %s\n", curl_easy_strerror(res)); ABORT_FINALIZE(RS_RET_SUSPENDED); } + free(pData->reply); DBGPRINTF("omelasticsearch: checkConn() completed with success\n"); finalize_it: @@ -271,7 +294,6 @@ static rsRetVal setCurlURL(instanceData *pData, uchar **tpls) { char authBuf[1024]; - char *restURL; uchar *searchIndex; uchar *searchType; uchar *parent; @@ -305,11 +327,12 @@ setCurlURL(instanceData *pData, uchar **tpls) if(r == 0) r = es_addBuf(&url, "parent=", sizeof("parent=")-1); if(r == 0) r = es_addBuf(&url, (char*)parent, ustrlen(parent)); } - restURL = es_str2cstr(url, NULL); - curl_easy_setopt(pData->curlHandle, CURLOPT_URL, restURL); + + free(pData->restURL); + pData->restURL = (uchar*)es_str2cstr(url, NULL); + curl_easy_setopt(pData->curlHandle, CURLOPT_URL, pData->restURL); es_deleteStr(url); - DBGPRINTF("omelasticsearch: using REST URL: '%s'\n", restURL); - free(restURL); + DBGPRINTF("omelasticsearch: using REST URL: '%s'\n", pData->restURL); if(pData->uid != NULL) { rLocal = snprintf(authBuf, sizeof(authBuf), "%s:%s", pData->uid, @@ -347,9 +370,6 @@ buildBatch(instanceData *pData, uchar *message, uchar **tpls) # define META_END "\"}}\n" getIndexTypeAndParent(pData, tpls, &searchIndex, &searchType, &parent); -dbgprintf("AAA: searchIndex: '%s'\n", searchIndex); -dbgprintf("AAA: searchType: '%s'\n", searchType); -dbgprintf("AAA: parent: '%s'\n", parent); r = es_addBuf(&pData->batch.data, META_STRT, sizeof(META_STRT)-1); if(r == 0) r = es_addBuf(&pData->batch.data, (char*)searchIndex, ustrlen(searchIndex)); @@ -373,22 +393,168 @@ finalize_it: RETiRet; } + +/* write data error request/replies to separate error file + * Note: we open the file but never close it before exit. If it + * needs to be closed, HUP must be sent. + */ +static inline rsRetVal +writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) +{ + char *rendered = NULL; + cJSON *errRoot; + cJSON *req; + cJSON *replyRoot = *pReplyRoot; + size_t toWrite; + ssize_t wrRet; + char errStr[1024]; + DEFiRet; + + if(pData->errorFile == NULL) { + DBGPRINTF("omelasticsearch: no local error logger defined - " + "ignoring ES error information\n"); + FINALIZE; + } + + if(pData->fdErrFile == -1) { + pData->fdErrFile = open((char*)pData->errorFile, + O_WRONLY|O_CREAT|O_APPEND|O_LARGEFILE|O_CLOEXEC, + S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP); + if(pData->fdErrFile == -1) { + rs_strerror_r(errno, errStr, sizeof(errStr)); + DBGPRINTF("omelasticsearch: error opening error file: %s\n", errStr); + ABORT_FINALIZE(RS_RET_ERR); + } + } + if((req=cJSON_CreateObject()) == NULL) ABORT_FINALIZE(RS_RET_ERR); + cJSON_AddItemToObject(req, "url", cJSON_CreateString((char*)pData->restURL)); + cJSON_AddItemToObject(req, "postdata", cJSON_CreateString((char*)reqmsg)); + + if((errRoot=cJSON_CreateObject()) == NULL) ABORT_FINALIZE(RS_RET_ERR); + cJSON_AddItemToObject(errRoot, "request", req); + cJSON_AddItemToObject(errRoot, "reply", replyRoot); + rendered = cJSON_Print(errRoot); + /* we do not do real error-handling on the err file, as this finally complicates + * things way to much. + */ + DBGPRINTF("omelasticsearch: error record: '%s'\n", rendered); + toWrite = strlen(rendered); + wrRet = write(pData->fdErrFile, rendered, toWrite); + if(wrRet != (ssize_t) toWrite) { + DBGPRINTF("omelasticsearch: error %d writing error file, write returns %lld\n", + errno, (long long) wrRet); + } + free(rendered); + cJSON_Delete(errRoot); + *pReplyRoot = NULL; /* tell caller not to delete once again! */ + +finalize_it: + if(rendered != NULL) + free(rendered); + RETiRet; +} + + +static inline rsRetVal +checkResultBulkmode(instanceData *pData, cJSON *root) +{ + int i; + int numitems; + cJSON *items; + cJSON *item; + cJSON *create; + cJSON *ok; + DEFiRet; + + items = cJSON_GetObjectItem(root, "items"); + if(items == NULL || items->type != cJSON_Array) { + DBGPRINTF("omelasticsearch: error in elasticsearch reply: " + "bulkmode insert does not return array, reply is: %s\n", + pData->reply); + ABORT_FINALIZE(RS_RET_DATAFAIL); + } + numitems = cJSON_GetArraySize(items); +DBGPRINTF("omelasticsearch: %d items in reply\n", numitems); + for(i = 0 ; i < numitems ; ++i) { + item = cJSON_GetArrayItem(items, i); + if(item == NULL) { + DBGPRINTF("omelasticsearch: error in elasticsearch reply: " + "cannot obtain reply array item %d\n", i); + ABORT_FINALIZE(RS_RET_DATAFAIL); + } + create = cJSON_GetObjectItem(item, "create"); + if(create == NULL || create->type != cJSON_Object) { + DBGPRINTF("omelasticsearch: error in elasticsearch reply: " + "cannot obtain 'create' item for #%d\n", i); + ABORT_FINALIZE(RS_RET_DATAFAIL); + } + ok = cJSON_GetObjectItem(create, "ok"); + if(ok == NULL || ok->type != cJSON_True) { + DBGPRINTF("omelasticsearch: error in elasticsearch reply: " + "item %d, prop ok (%p) not ok\n", i, ok); + ABORT_FINALIZE(RS_RET_DATAFAIL); + } + } + +finalize_it: + RETiRet; +} + + +static inline rsRetVal +checkResult(instanceData *pData, uchar *reqmsg) +{ + cJSON *root; + cJSON *ok; + DEFiRet; + + root = cJSON_Parse(pData->reply); + if(root == NULL) { + DBGPRINTF("omelasticsearch: could not parse JSON result \n"); + ABORT_FINALIZE(RS_RET_ERR); + } + + if(pData->bulkmode) { + iRet = checkResultBulkmode(pData, root); + } else { + ok = cJSON_GetObjectItem(root, "ok"); + if(ok == NULL || ok->type != cJSON_True) { + iRet = RS_RET_DATAFAIL; + } + } + + /* Note: we ignore errors writing the error file, as we cannot handle + * these in any case. + */ + if(iRet == RS_RET_DATAFAIL) { + writeDataError(pData, &root, reqmsg); + iRet = RS_RET_OK; /* we have handled the problem! */ + } + +finalize_it: + if(root != NULL) + cJSON_Delete(root); + RETiRet; +} + + static rsRetVal -curlPost(instanceData *instance, uchar *message, int msglen, uchar **tpls) +curlPost(instanceData *pData, uchar *message, int msglen, uchar **tpls) { CURLcode code; - CURL *curl = instance->curlHandle; + CURL *curl = pData->curlHandle; DEFiRet; - if(instance->dynSrchIdx || instance->dynSrchType || instance->dynParent) - CHKiRet(setCurlURL(instance, tpls)); + pData->reply = NULL; + pData->replyLen = 0; + + if(pData->dynSrchIdx || pData->dynSrchType || pData->dynParent) + CHKiRet(setCurlURL(pData, tpls)); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, (char *)message); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, pData); curl_easy_setopt(curl, CURLOPT_POSTFIELDS, (char *)message); curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, msglen); -dbgprintf("omelasticsearch: do curl_easy_perform()\n"); code = curl_easy_perform(curl); -DBGPRINTF("omelasticsearch: curl_easy_perform() returned %lld\n", (long long) code); switch (code) { case CURLE_COULDNT_RESOLVE_HOST: case CURLE_COULDNT_RESOLVE_PROXY: @@ -398,12 +564,18 @@ DBGPRINTF("omelasticsearch: curl_easy_perform() returned %lld\n", (long long) co DBGPRINTF("omelasticsearch: we are suspending ourselfs due " "to failure %lld of curl_easy_perform()\n", (long long) code); - return RS_RET_SUSPENDED; + ABORT_FINALIZE(RS_RET_SUSPENDED); default: STATSCOUNTER_INC(indexSubmit, mutIndexSubmit); - return RS_RET_OK; + break; } + + pData->reply[pData->replyLen] = '\0'; /* byte has been reserved in malloc */ + DBGPRINTF("omelasticsearch: es reply: '%s'\n", pData->reply); + + CHKiRet(checkResult(pData, message)); finalize_it: + free(pData->reply); RETiRet; } @@ -424,7 +596,6 @@ CODESTARTdoAction if(pData->bulkmode) { CHKiRet(buildBatch(pData, ppString[0], ppString)); } else { -dbgprintf("omelasticsearch: doAction calling curlPost\n"); CHKiRet(curlPost(pData, ppString[0], strlen((char*)ppString[0]), ppString)); } @@ -449,35 +620,20 @@ ENDendTransaction size_t curlResult(void *ptr, size_t size, size_t nmemb, void *userdata) { - unsigned int i; char *p = (char *)ptr; - char *jsonData = (char *)userdata; - static char ok[] = "{\"ok\":true,"; - - ASSERT(size == 1); -DBGPRINTF("omelasticsearch request: %s\n", jsonData); -DBGPRINTF("omelasticsearch result: "); -for (i = 0; i < nmemb; i++) - DBGPRINTF("%c", p[i]); -DBGPRINTF("\n"); - - if (size == 1 && - nmemb > sizeof(ok)-1 && - strncmp(p, ok, sizeof(ok)-1) == 0) { - STATSCOUNTER_INC(indexSuccess, mutIndexSuccess); -dbgprintf("omelasticsearch ok\n"); - } else { -dbgprintf("omelasticsearch fail\n"); - STATSCOUNTER_INC(indexFailed, mutIndexFailed); - if (Debug) { - DBGPRINTF("omelasticsearch (fail) request: %s\n", jsonData); - DBGPRINTF("omelasticsearch (fail) result: "); - for (i = 0; i < nmemb; i++) - DBGPRINTF("%c", p[i]); - DBGPRINTF("\n"); - } + instanceData *pData = (instanceData*) userdata; + char *buf; + size_t newlen; + + newlen = pData->replyLen + size*nmemb; + if((buf = realloc(pData->reply, newlen + 1)) == NULL) { + DBGPRINTF("omelasticsearch: realloc failed in curlResult\n"); + return 0; /* abort due to failure */ } - return size * nmemb; + memcpy(buf+pData->replyLen, p, size*nmemb); + pData->replyLen = newlen; + pData->reply = buf; + return size*nmemb; } @@ -533,6 +689,7 @@ setInstParamDefaults(instanceData *pData) pData->asyncRepl = 0; pData->bulkmode = 0; pData->tplName = NULL; + pData->errorFile = NULL; } BEGINnewActInst @@ -552,6 +709,8 @@ CODESTARTnewActInst continue; if(!strcmp(actpblk.descr[i].name, "server")) { pData->server = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(actpblk.descr[i].name, "errorfile")) { + pData->errorFile = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "serverport")) { pData->port = (int) pvals[i].val.d.n, NULL; } else if(!strcmp(actpblk.descr[i].name, "uid")) { @@ -693,6 +852,14 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct +BEGINdoHUP +CODESTARTdoHUP + if(pData->fdErrFile != -1) { + close(pData->fdErrFile); + pData->fdErrFile = -1; + } +ENDdoHUP + BEGINmodExit CODESTARTmodExit @@ -707,6 +874,7 @@ CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES CODEqueryEtryPt_IsCompatibleWithFeature_IF_OMOD_QUERIES CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES +CODEqueryEtryPt_doHUP CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */ ENDqueryEtryPt diff --git a/plugins/omlibdbi/omlibdbi.c b/plugins/omlibdbi/omlibdbi.c index 99bcaf9d..390e59d5 100644 --- a/plugins/omlibdbi/omlibdbi.c +++ b/plugins/omlibdbi/omlibdbi.c @@ -81,15 +81,36 @@ typedef struct configSettings_s { uchar *dbName; /* database to use */ } configSettings_t; static configSettings_t cs; +uchar *pszFileDfltTplName; /* name of the default template to use */ + +struct modConfData_s { + rsconf_t *pConf; /* our overall config object */ + uchar *dbiDrvrDir; /* where do the dbi drivers reside? */ + uchar *tplName; /* default template */ +}; + +static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */ +static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */ +static int bLegacyCnfModGlobalsPermitted;/* are legacy module-global config parameters permitted? */ + /* tables for interfacing with the v6 config system */ +/* module-global parameters */ +static struct cnfparamdescr modpdescr[] = { + { "template", eCmdHdlrGetWord, 0 }, + { "driverdirectory", eCmdHdlrGetWord, 0 } +}; +static struct cnfparamblk modpblk = + { CNFPARAMBLK_VERSION, + sizeof(modpdescr)/sizeof(struct cnfparamdescr), + modpdescr + }; /* action (instance) parameters */ static struct cnfparamdescr actpdescr[] = { { "server", eCmdHdlrGetWord, 1 }, { "db", eCmdHdlrGetWord, 1 }, { "uid", eCmdHdlrGetWord, 1 }, { "pwd", eCmdHdlrGetWord, 1 }, - { "driverdirectory", eCmdHdlrGetWord, 0 }, { "driver", eCmdHdlrGetWord, 1 }, { "template", eCmdHdlrGetWord, 0 } }; @@ -99,6 +120,20 @@ static struct cnfparamblk actpblk = actpdescr }; +/* this function gets the default template. It coordinates action between + * old-style and new-style configuration parts. + */ +static inline uchar* +getDfltTpl(void) +{ + if(loadModConf != NULL && loadModConf->tplName != NULL) + return loadModConf->tplName; + else if(pszFileDfltTplName == NULL) + return (uchar*)" StdDBFmt"; + else + return pszFileDfltTplName; +} + BEGINinitConfVars /* (re)set config variables to default values */ CODESTARTinitConfVars @@ -144,7 +179,6 @@ static void closeConn(instanceData *pData) BEGINfreeInstance CODESTARTfreeInstance closeConn(pData); - free(pData->dbiDrvrDir); free(pData->drvrName); free(pData->host); free(pData->usrName); @@ -302,6 +336,79 @@ CODESTARTdoAction ENDdoAction +BEGINbeginCnfLoad +CODESTARTbeginCnfLoad + loadModConf = pModConf; + pModConf->pConf = pConf; + pModConf->tplName = NULL; + bLegacyCnfModGlobalsPermitted = 1; +ENDbeginCnfLoad + +BEGINsetModCnf + struct cnfparamvals *pvals = NULL; + int i; +CODESTARTsetModCnf + pvals = nvlstGetParams(lst, &modpblk, NULL); + if(pvals == NULL) { + errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "omlibdbi: error processing " + "module config parameters [module(...)]"); + ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS); + } + + if(Debug) { + dbgprintf("module (global) param blk for omlibdbi:\n"); + cnfparamsPrint(&modpblk, pvals); + } + + for(i = 0 ; i < modpblk.nParams ; ++i) { + if(!pvals[i].bUsed) + continue; + if(!strcmp(modpblk.descr[i].name, "template")) { + loadModConf->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + if(pszFileDfltTplName != NULL) { + errmsg.LogError(0, RS_RET_DUP_PARAM, "omlibdbi: warning: default template " + "was already set via legacy directive - may lead to inconsistent " + "results."); + } + } else if(!strcmp(modpblk.descr[i].name, "driverdirectory")) { + loadModConf->dbiDrvrDir = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else { + dbgprintf("omlibdbi: program error, non-handled " + "param '%s' in beginCnfLoad\n", modpblk.descr[i].name); + } + } + bLegacyCnfModGlobalsPermitted = 0; +finalize_it: + if(pvals != NULL) + cnfparamvalsDestruct(pvals, &modpblk); +ENDsetModCnf + +BEGINendCnfLoad +CODESTARTendCnfLoad + loadModConf = NULL; /* done loading */ + /* free legacy config vars */ + free(pszFileDfltTplName); + pszFileDfltTplName = NULL; +ENDendCnfLoad + +BEGINcheckCnf +CODESTARTcheckCnf +ENDcheckCnf + +BEGINactivateCnf +CODESTARTactivateCnf + runModConf = pModConf; +ENDactivateCnf + +BEGINfreeCnf +CODESTARTfreeCnf + free(pModConf->tplName); + free(pModConf->dbiDrvrDir); +ENDfreeCnf + + + + static inline void setInstParamDefaults(instanceData *pData) { @@ -311,6 +418,7 @@ setInstParamDefaults(instanceData *pData) BEGINnewActInst struct cnfparamvals *pvals; + uchar *tplToUse; int i; CODESTARTnewActInst if((pvals = nvlstGetParams(lst, &actpblk, NULL)) == NULL) { @@ -332,28 +440,19 @@ CODESTARTnewActInst pData->usrName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "pwd")) { pData->pwd = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); - } else if(!strcmp(actpblk.descr[i].name, "driverdirectory")) { - pData->dbiDrvrDir = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "driver")) { pData->drvrName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(actpblk.descr[i].name, "template")) { pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else { - dbgprintf("ommysql: program error, non-handled " + dbgprintf("omlibdbi: program error, non-handled " "param '%s'\n", actpblk.descr[i].name); } } - if(pData->tplName == NULL) { - CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*) strdup(" StdDBFmt"), - OMSR_RQD_TPL_OPT_SQL)); - } else { - CHKiRet(OMSRsetEntry(*ppOMSR, 0, - (uchar*) strdup((char*) pData->tplName), - OMSR_RQD_TPL_OPT_SQL)); - } + tplToUse = (pData->tplName == NULL) ? (uchar*)strdup((char*)getDfltTpl()) : pData->tplName; + CHKiRet(OMSRsetEntry(*ppOMSR, 0, tplToUse, OMSR_RQD_TPL_OPT_SQL)); CODE_STD_FINALIZERnewActInst -dbgprintf("XXXX: added param, iRet %d\n", iRet); cnfparamvalsDestruct(pvals, &actpblk); ENDnewActInst @@ -380,19 +479,17 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) /* NULL values are supported because drivers have different needs. * They will err out on connect. -- rgerhards, 2008-02-15 */ - if(cs.host != NULL) - if((pData->host = (uchar*) strdup((char*)cs.host)) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + if(cs.host != NULL) + CHKmalloc(pData->host = (uchar*) strdup((char*)cs.host)); if(cs.usrName != NULL) - if((pData->usrName = (uchar*) strdup((char*)cs.usrName)) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - if(cs.dbName != NULL) - if((pData->dbName = (uchar*) strdup((char*)cs.dbName)) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - if(cs.pwd != NULL) - if((pData->pwd = (uchar*) strdup((char*)cs.pwd)) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); + CHKmalloc(pData->usrName = (uchar*) strdup((char*)cs.usrName)); + if(cs.dbName != NULL) + CHKmalloc(pData->dbName = (uchar*) strdup((char*)cs.dbName)); + if(cs.pwd != NULL) + CHKmalloc(pData->pwd = (uchar*) strdup((char*)cs.pwd)); if(cs.dbiDrvrDir != NULL) - if((pData->dbiDrvrDir = (uchar*) strdup((char*)cs.dbiDrvrDir)) == NULL) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - - CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_RQD_TPL_OPT_SQL, (uchar*) " StdDBFmt")); - + CHKmalloc(loadModConf->dbiDrvrDir = (uchar*) strdup((char*)cs.dbiDrvrDir)); + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_RQD_TPL_OPT_SQL, getDfltTpl())); CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct @@ -413,6 +510,8 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES +CODEqueryEtryPt_STD_CONF2_QUERIES +CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES ENDqueryEtryPt @@ -444,7 +543,7 @@ INITLegCnfVars *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"actionlibdbidriverdirectory", 0, eCmdHdlrGetWord, NULL, &cs.dbiDrvrDir, STD_LOADABLE_MODULE_ID)); + CHKiRet(regCfSysLineHdlr2((uchar *)"actionlibdbidriverdirectory", 0, eCmdHdlrGetWord, NULL, &cs.dbiDrvrDir, STD_LOADABLE_MODULE_ID, &bLegacyCnfModGlobalsPermitted)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"actionlibdbidriver", 0, eCmdHdlrGetWord, NULL, &cs.drvrName, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"actionlibdbihost", 0, eCmdHdlrGetWord, NULL, &cs.host, STD_LOADABLE_MODULE_ID)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"actionlibdbiusername", 0, eCmdHdlrGetWord, NULL, &cs.usrName, STD_LOADABLE_MODULE_ID)); diff --git a/plugins/ommongodb/ommongodb.c b/plugins/ommongodb/ommongodb.c index ed77f824..dd997410 100644 --- a/plugins/ommongodb/ommongodb.c +++ b/plugins/ommongodb/ommongodb.c @@ -236,11 +236,11 @@ getDefaultBSON(msg_t *pMsg) gint64 ts_gen, ts_rcv; /* timestamps: generated, received */ int secfrac; - procid = MsgGetProp(pMsg, NULL, PROP_PROGRAMNAME, NULL, &procid_len, &procid_free); - tag = MsgGetProp(pMsg, NULL, PROP_SYSLOGTAG, NULL, &tag_len, &tag_free); - pid = MsgGetProp(pMsg, NULL, PROP_PROCID, NULL, &pid_len, &pid_free); - sys = MsgGetProp(pMsg, NULL, PROP_HOSTNAME, NULL, &sys_len, &sys_free); - msg = MsgGetProp(pMsg, NULL, PROP_MSG, NULL, &msg_len, &msg_free); + procid = MsgGetProp(pMsg, NULL, PROP_PROGRAMNAME, NULL, &procid_len, &procid_free, NULL); + tag = MsgGetProp(pMsg, NULL, PROP_SYSLOGTAG, NULL, &tag_len, &tag_free, NULL); + pid = MsgGetProp(pMsg, NULL, PROP_PROCID, NULL, &pid_len, &pid_free, NULL); + sys = MsgGetProp(pMsg, NULL, PROP_HOSTNAME, NULL, &sys_len, &sys_free, NULL); + msg = MsgGetProp(pMsg, NULL, PROP_MSG, NULL, &msg_len, &msg_free, NULL); // TODO: move to datetime? Refactor in any case! rgerhards, 2012-03-30 ts_gen = (gint64) datetime.syslogTime2time_t(&pMsg->tTIMESTAMP) * 1000; /* ms! */ diff --git a/plugins/ommysql/ommysql.c b/plugins/ommysql/ommysql.c index 69ffb9ac..2dfa29de 100644 --- a/plugins/ommysql/ommysql.c +++ b/plugins/ommysql/ommysql.c @@ -189,7 +189,6 @@ static rsRetVal initMySQL(instanceData *pData, int bSilent) ASSERT(pData != NULL); ASSERT(pData->f_hmysql == NULL); - pData->f_hmysql = mysql_init(NULL); if(pData->f_hmysql == NULL) { errmsg.LogError(0, RS_RET_SUSPENDED, "can not initialize MySQL handle"); @@ -219,10 +218,12 @@ static rsRetVal initMySQL(instanceData *pData, int bSilent) pData->f_dbpwd, pData->f_dbname, pData->f_dbsrvPort, NULL, 0) == NULL) { reportDBError(pData, bSilent); closeMySQL(pData); /* ignore any error we may get */ - iRet = RS_RET_SUSPENDED; + ABORT_FINALIZE(RS_RET_SUSPENDED); } + mysql_autocommit(pData->f_hmysql, 0); } +finalize_it: RETiRet; } @@ -241,6 +242,7 @@ rsRetVal writeMySQL(uchar *psz, instanceData *pData) /* see if we are ready to proceed */ if(pData->f_hmysql == NULL) { CHKiRet(initMySQL(pData, 0)); + } /* try insert */ @@ -272,12 +274,28 @@ CODESTARTtryResume } ENDtryResume +BEGINbeginTransaction +CODESTARTbeginTransaction + CHKiRet(writeMySQL((uchar*)"START TRANSACTION", pData)); +finalize_it: +ENDbeginTransaction + BEGINdoAction CODESTARTdoAction dbgprintf("\n"); - iRet = writeMySQL(ppString[0], pData); + CHKiRet(writeMySQL(ppString[0], pData)); + iRet = RS_RET_DEFER_COMMIT; +finalize_it: ENDdoAction +BEGINendTransaction +CODESTARTendTransaction + if (mysql_commit(pData->f_hmysql) != 0) { + dbgprintf("mysql server error: transaction not committed\n"); + iRet = RS_RET_SUSPENDED; + } +ENDendTransaction + static inline void setInstParamDefaults(instanceData *pData) @@ -305,7 +323,7 @@ CODESTARTnewActInst CHKiRet(createInstance(&pData)); setInstParamDefaults(pData); - CODE_STD_STRING_REQUESTnewActInst(1) + CODE_STD_STRING_REQUESTparseSelectorAct(1) for(i = 0 ; i < actpblk.nParams ; ++i) { if(!pvals[i].bUsed) continue; @@ -437,6 +455,7 @@ BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES +CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */ ENDqueryEtryPt @@ -459,6 +478,11 @@ INITLegCnfVars *ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */ CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); + INITChkCoreFeature(bCoreSupportsBatching, CORE_FEATURE_BATCHING); + if(!bCoreSupportsBatching) { + errmsg.LogError(0, NO_ERRCODE, "ommysql: rsyslog core too old"); + ABORT_FINALIZE(RS_RET_ERR); + } /* we need to init the MySQL library. If that fails, we cannot run */ if( diff --git a/plugins/omruleset/omruleset.c b/plugins/omruleset/omruleset.c index 4c7e25d2..11765507 100644 --- a/plugins/omruleset/omruleset.c +++ b/plugins/omruleset/omruleset.c @@ -120,7 +120,11 @@ CODESTARTdoAction (char*) pData->pszRulesetName, pData->pRuleset); MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY); MsgSetRuleset(pMsg, pData->pRuleset); - submitMsg(pMsg); + /* Note: we intentionally use submitMsg2() here, as we process messages + * that were already run through the rate-limiter. So it is (at least) + * questionable if they were rate-limited again. + */ + submitMsg2(pMsg); finalize_it: ENDdoAction @@ -165,12 +169,8 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1) p += sizeof(":omruleset:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ CHKiRet(createInstance(&pData)); - /* re-enable in v7.3: requires action list to support - * action-like statements, something that is too late to - * do in 7.1. errmsg.LogError(0, RS_RET_DEPRECATED, "warning: omruleset is deprecated, consider " "using the 'call' statement instead"); - */ /* check if a non-standard template is to be applied */ if(*(p-1) == ';') |