From 71121003e273dc9fb2d6dd2aac427fb2c6329d23 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 25 Jul 2012 18:30:18 +0200 Subject: omelasticsearch: parse JSON response (in regard to data errors) note: bulkmode response processing is still mostly missing --- plugins/omelasticsearch/omelasticsearch.c | 102 +++++++++++++++++++----------- 1 file changed, 65 insertions(+), 37 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index f77caeca..b49caf38 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -34,6 +34,7 @@ #include #include #include +#include "cjson.h" #include "conf.h" #include "syslogd-types.h" #include "srUtils.h" @@ -64,8 +65,9 @@ STATSCOUNTER_DEF(indexSuccess, mutIndexSuccess) */ typedef struct curl_slist HEADER; typedef struct _instanceData { - uchar *server; int port; + int replyLen; + uchar *server; uchar *uid; uchar *pwd; uchar *searchIndex; @@ -73,6 +75,7 @@ typedef struct _instanceData { uchar *parent; uchar *tplName; uchar *timeout; + char *reply; sbool dynSrchIdx; sbool dynSrchType; sbool dynParent; @@ -139,6 +142,7 @@ CODESTARTfreeInstance free(pData->searchType); free(pData->parent); free(pData->tplName); + free(pData->timeout); ENDfreeInstance BEGINdbgPrintInstInfo @@ -202,12 +206,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: @@ -373,22 +381,51 @@ finalize_it: RETiRet; } +static inline rsRetVal +checkResult(instanceData *pData) +{ + 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) { + //TODO: implement + //iRet = checkResultBulkmode(pData, root); + } else { + ok = cJSON_GetObjectItem(root, "ok"); + if(ok == NULL || ok->type != cJSON_True) { + ABORT_FINALIZE(RS_RET_DATAFAIL); + } + } +finalize_it: + 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; - curl_easy_setopt(curl, CURLOPT_WRITEDATA, (char *)message); + if(pData->dynSrchIdx || pData->dynSrchType || pData->dynParent) + CHKiRet(setCurlURL(pData, tpls)); + + 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 +435,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)); finalize_it: + free(pData->reply); RETiRet; } @@ -449,35 +492,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; } -- cgit v1.2.3 From 575686bd68a47fc5d9f59c1ed610f6680e1bb6fa Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 30 Jul 2012 11:01:11 +0200 Subject: omelasticsearch: mileston: bulk reply is parsed --- plugins/omelasticsearch/omelasticsearch.c | 50 +++++++++++++++++++++++++++++-- 1 file changed, 48 insertions(+), 2 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index b49caf38..04113170 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -381,6 +381,53 @@ finalize_it: 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 ok (%p) not ok\n", ok); + ABORT_FINALIZE(RS_RET_DATAFAIL); + } +dbgprintf("omelasticsearch: item %d is OK\n", i); + } + +finalize_it: + RETiRet; +} + + static inline rsRetVal checkResult(instanceData *pData) { @@ -395,8 +442,7 @@ checkResult(instanceData *pData) } if(pData->bulkmode) { - //TODO: implement - //iRet = checkResultBulkmode(pData, root); + iRet = checkResultBulkmode(pData, root); } else { ok = cJSON_GetObjectItem(root, "ok"); if(ok == NULL || ok->type != cJSON_True) { -- cgit v1.2.3 From 36f7fbd5f458c3e8b2924a72f74621e055f319b2 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 30 Jul 2012 12:40:49 +0200 Subject: omelasticsearch: support for writing data errors to local file added --- plugins/omelasticsearch/omelasticsearch.c | 109 ++++++++++++++++++++++++++---- 1 file changed, 95 insertions(+), 14 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 04113170..fea85f22 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -34,6 +34,9 @@ #include #include #include +#include +#include +#include #include "cjson.h" #include "conf.h" #include "syslogd-types.h" @@ -67,6 +70,7 @@ typedef struct curl_slist HEADER; typedef struct _instanceData { int port; int replyLen; + int fdErrFile; /* error file fd or -1 if not open */ uchar *server; uchar *uid; uchar *pwd; @@ -75,6 +79,8 @@ 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; @@ -107,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 = @@ -117,6 +124,8 @@ static struct cnfparamblk actpblk = BEGINcreateInstance CODESTARTcreateInstance + pData->restURL = NULL; + pData->fdErrFile = -1; ENDcreateInstance BEGINisCompatibleWithFeature @@ -135,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); @@ -143,6 +154,8 @@ CODESTARTfreeInstance free(pData->parent); free(pData->tplName); free(pData->timeout); + free(pData->restURL); + free(pData->errorFile); ENDfreeInstance BEGINdbgPrintInstInfo @@ -162,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 @@ -279,7 +294,6 @@ static rsRetVal setCurlURL(instanceData *pData, uchar **tpls) { char authBuf[1024]; - char *restURL; uchar *searchIndex; uchar *searchType; uchar *parent; @@ -313,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, @@ -355,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)); @@ -381,6 +393,55 @@ 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; + char errStr[1024]; + DEFiRet; + + if(pData->errorFile == NULL) + 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); +DBGPRINTF("omelasticsearch: error record: '%s'\n", rendered); + write(pData->fdErrFile, rendered, strlen(rendered)); + 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) { @@ -420,7 +481,6 @@ DBGPRINTF("omelasticsearch: %d items in reply\n", numitems); "item ok (%p) not ok\n", ok); ABORT_FINALIZE(RS_RET_DATAFAIL); } -dbgprintf("omelasticsearch: item %d is OK\n", i); } finalize_it: @@ -429,7 +489,7 @@ finalize_it: static inline rsRetVal -checkResult(instanceData *pData) +checkResult(instanceData *pData, uchar *reqmsg) { cJSON *root; cJSON *ok; @@ -446,11 +506,21 @@ checkResult(instanceData *pData) } else { ok = cJSON_GetObjectItem(root, "ok"); if(ok == NULL || ok->type != cJSON_True) { - ABORT_FINALIZE(RS_RET_DATAFAIL); + 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: - cJSON_Delete(root); + if(root != NULL) + cJSON_Delete(root); RETiRet; } @@ -490,7 +560,7 @@ curlPost(instanceData *pData, uchar *message, int msglen, uchar **tpls) pData->reply[pData->replyLen] = '\0'; /* byte has been reserved in malloc */ DBGPRINTF("omelasticsearch: es reply: '%s'\n", pData->reply); - CHKiRet(checkResult(pData)); + CHKiRet(checkResult(pData, message)); finalize_it: free(pData->reply); RETiRet; @@ -513,7 +583,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)); } @@ -607,6 +676,7 @@ setInstParamDefaults(instanceData *pData) pData->asyncRepl = 0; pData->bulkmode = 0; pData->tplName = NULL; + pData->errorFile = NULL; } BEGINnewActInst @@ -626,6 +696,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")) { @@ -767,6 +839,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 @@ -781,6 +861,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 -- cgit v1.2.3 From cb12ed5dd211f9910860e86fd937180895c459de Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 30 Jul 2012 13:06:44 +0200 Subject: omelasticsearch: improved debug logging --- plugins/omelasticsearch/omelasticsearch.c | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index fea85f22..48cfef7d 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -408,8 +408,11 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) char errStr[1024]; DEFiRet; - if(pData->errorFile == NULL) + 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, @@ -478,7 +481,7 @@ DBGPRINTF("omelasticsearch: %d items in reply\n", numitems); ok = cJSON_GetObjectItem(create, "ok"); if(ok == NULL || ok->type != cJSON_True) { DBGPRINTF("omelasticsearch: error in elasticsearch reply: " - "item ok (%p) not ok\n", ok); + "item %d, prop ok (%p) not ok\n", i, ok); ABORT_FINALIZE(RS_RET_DATAFAIL); } } -- cgit v1.2.3 From b11c85aac8b41eb74fb14c486f88940df8819bbc Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 29 Oct 2012 11:49:07 +0100 Subject: move cJSON to omelasticsearch the rsyslog runtime itself now uses json-c, but omelasticsearch currently depends on cJSON. We will change this, but not yet. Let's merge as is and see that it works well ;) --- plugins/omelasticsearch/omelasticsearch.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 48cfef7d..a5833b15 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -37,7 +37,7 @@ #include #include #include -#include "cjson.h" +#include "cJSON/cjson.h" #include "conf.h" #include "syslogd-types.h" #include "srUtils.h" -- cgit v1.2.3 From e5ef73eb25c8dda2e19c593ad2fc0a960aa8873b Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 1 Nov 2012 17:05:07 +0100 Subject: bugfix: invalid rsyslog-internal macro API use This had no bad effect, because the macro did the same as the one that should have been used. --- plugins/omelasticsearch/omelasticsearch.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index a5833b15..982f4318 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -773,7 +773,7 @@ CODESTARTnewActInst if(pData->dynSrchType) ++iNumTpls; if(pData->dynParent) ++iNumTpls; DBGPRINTF("omelasticsearch: requesting %d templates\n", iNumTpls); - CODE_STD_STRING_REQUESTparseSelectorAct(iNumTpls) + CODE_STD_STRING_REQUESTnewActInst(iNumTpls) CHKiRet(OMSRsetEntry(*ppOMSR, 0, (uchar*)strdup((pData->tplName == NULL) ? " StdJSONFmt" : (char*)pData->tplName), -- cgit v1.2.3 From 3352d2c605567c29840cc93a358d60881c865cb7 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 14 Dec 2012 09:20:51 +0100 Subject: minor cleanup --- plugins/omelasticsearch/omelasticsearch.c | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 982f4318..499a2bb2 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -405,6 +405,8 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) cJSON *errRoot; cJSON *req; cJSON *replyRoot = *pReplyRoot; + size_t toWrite; + ssize_t wrRet; char errStr[1024]; DEFiRet; @@ -432,8 +434,16 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) cJSON_AddItemToObject(errRoot, "request", req); cJSON_AddItemToObject(errRoot, "reply", replyRoot); rendered = cJSON_Print(errRoot); -DBGPRINTF("omelasticsearch: error record: '%s'\n", rendered); - write(pData->fdErrFile, rendered, strlen(rendered)); + /* 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! */ -- cgit v1.2.3 From 2f68d5f0d8a5c3ffef6bf52f27abc726ad27d764 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=B4me=20Renard?= Date: Wed, 24 Apr 2013 10:56:52 +0200 Subject: omelasticsearch: _id field support for bulk operations also max number of templates for plugin use has been increased to five closes: http://bugzilla.adiscon.com/show_bug.cgi?id=392 --- plugins/omelasticsearch/omelasticsearch.c | 122 +++++++++++++++++++++++++----- 1 file changed, 105 insertions(+), 17 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index f27fe62b..33e58c1a 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -11,11 +11,11 @@ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 * -or- * see COPYING.ASL20 in the source distribution - * + * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. @@ -79,12 +79,14 @@ typedef struct _instanceData { uchar *parent; uchar *tplName; uchar *timeout; + uchar *bulkId; uchar *restURL; /* last used URL for error reporting */ uchar *errorFile; char *reply; sbool dynSrchIdx; sbool dynSrchType; sbool dynParent; + sbool dynBulkId; sbool bulkmode; sbool asyncRepl; struct { @@ -114,7 +116,9 @@ static struct cnfparamdescr actpdescr[] = { { "asyncrepl", eCmdHdlrBinary, 0 }, { "timeout", eCmdHdlrGetWord, 0 }, { "errorfile", eCmdHdlrGetWord, 0 }, - { "template", eCmdHdlrGetWord, 1 } + { "template", eCmdHdlrGetWord, 1 }, + { "dynbulkid", eCmdHdlrBinary, 0 }, + { "bulkid", eCmdHdlrGetWord, 0 }, }; static struct cnfparamblk actpblk = { CNFPARAMBLK_VERSION, @@ -156,6 +160,7 @@ CODESTARTfreeInstance free(pData->timeout); free(pData->restURL); free(pData->errorFile); + free(pData->bulkId); ENDfreeInstance BEGINdbgPrintInstInfo @@ -177,6 +182,8 @@ CODESTARTdbgPrintInstInfo dbgprintf("\tbulkmode=%d\n", pData->bulkmode); dbgprintf("\terrorfile='%s'\n", pData->errorFile == NULL ? (uchar*)"(not configured)" : pData->errorFile); + dbgprintf("\tdynbulkid=%d\n", pData->dynBulkId); + dbgprintf("\tbulkid='%s'\n", pData->bulkId); ENDdbgPrintInstInfo @@ -220,7 +227,7 @@ checkConn(instanceData *pData) cstr = es_str2cstr(url, NULL); curl_easy_setopt(curl, CURLOPT_URL, cstr); free(cstr); - + pData->reply = NULL; pData->replyLen = 0; curl_easy_setopt(curl, CURLOPT_WRITEDATA, pData); @@ -250,7 +257,8 @@ ENDtryResume /* get the current index and type for this message */ static inline void getIndexTypeAndParent(instanceData *pData, uchar **tpls, - uchar **srchIndex, uchar **srchType, uchar **parent) + uchar **srchIndex, uchar **srchType, uchar **parent, + uchar **bulkId) { if(pData->dynSrchIdx) { *srchIndex = tpls[1]; @@ -258,15 +266,27 @@ getIndexTypeAndParent(instanceData *pData, uchar **tpls, *srchType = tpls[2]; if(pData->dynParent) { *parent = tpls[3]; + if(pData->dynBulkId) { + *bulkId = tpls[4]; + } } else { *parent = pData->parent; + if(pData->dynBulkId) { + *bulkId = tpls[3]; + } } } else { *srchType = pData->searchType; if(pData->dynParent) { *parent = tpls[2]; + if(pData->dynBulkId) { + *bulkId = tpls[3]; + } } else { *parent = pData->parent; + if(pData->dynBulkId) { + *bulkId = tpls[2]; + } } } } else { @@ -275,15 +295,27 @@ getIndexTypeAndParent(instanceData *pData, uchar **tpls, *srchType = tpls[1]; if(pData->dynParent) { *parent = tpls[2]; + if(pData->dynBulkId) { + *bulkId = tpls[3]; + } } else { *parent = pData->parent; + if(pData->dynBulkId) { + *bulkId = tpls[2]; + } } } else { *srchType = pData->searchType; if(pData->dynParent) { *parent = tpls[1]; + if(pData->dynBulkId) { + *bulkId = tpls[2]; + } } else { *parent = pData->parent; + if(pData->dynBulkId) { + *bulkId = tpls[1]; + } } } } @@ -297,6 +329,7 @@ setCurlURL(instanceData *pData, uchar **tpls) uchar *searchIndex; uchar *searchType; uchar *parent; + uchar *bulkId; es_str_t *url; int rLocal; int r; @@ -308,7 +341,7 @@ setCurlURL(instanceData *pData, uchar **tpls) r = es_addBuf(&url, "_bulk", sizeof("_bulk")-1); parent = NULL; } else { - getIndexTypeAndParent(pData, tpls, &searchIndex, &searchType, &parent); + getIndexTypeAndParent(pData, tpls, &searchIndex, &searchType, &parent, &bulkId); r = es_addBuf(&url, (char*)searchIndex, ustrlen(searchIndex)); if(r == 0) r = es_addChar(&url, '/'); if(r == 0) r = es_addBuf(&url, (char*)searchType, ustrlen(searchType)); @@ -330,7 +363,7 @@ setCurlURL(instanceData *pData, uchar **tpls) free(pData->restURL); pData->restURL = (uchar*)es_str2cstr(url, NULL); - curl_easy_setopt(pData->curlHandle, CURLOPT_URL, pData->restURL); + curl_easy_setopt(pData->curlHandle, CURLOPT_URL, pData->restURL); es_deleteStr(url); DBGPRINTF("omelasticsearch: using REST URL: '%s'\n", pData->restURL); @@ -343,7 +376,7 @@ setCurlURL(instanceData *pData, uchar **tpls) rLocal); ABORT_FINALIZE(RS_RET_ERR); } - curl_easy_setopt(pData->curlHandle, CURLOPT_USERPWD, authBuf); + curl_easy_setopt(pData->curlHandle, CURLOPT_USERPWD, authBuf); curl_easy_setopt(pData->curlHandle, CURLOPT_PROXYAUTH, CURLAUTH_ANY); } finalize_it: @@ -363,13 +396,15 @@ buildBatch(instanceData *pData, uchar *message, uchar **tpls) uchar *searchIndex; uchar *searchType; uchar *parent; + uchar *bulkId = NULL; DEFiRet; # define META_STRT "{\"index\":{\"_index\": \"" # define META_TYPE "\",\"_type\":\"" # define META_PARENT "\",\"_parent\":\"" +# define META_ID "\", \"_id\":\"" # define META_END "\"}}\n" - getIndexTypeAndParent(pData, tpls, &searchIndex, &searchType, &parent); + getIndexTypeAndParent(pData, tpls, &searchIndex, &searchType, &parent, &bulkId); 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)); @@ -380,6 +415,10 @@ buildBatch(instanceData *pData, uchar *message, uchar **tpls) if(r == 0) r = es_addBuf(&pData->batch.data, META_PARENT, sizeof(META_PARENT)-1); if(r == 0) r = es_addBuf(&pData->batch.data, (char*)parent, ustrlen(parent)); } + if(bulkId != NULL) { + if(r == 0) r = es_addBuf(&pData->batch.data, META_ID, sizeof(META_ID)-1); + if(r == 0) r = es_addBuf(&pData->batch.data, (char*)bulkId, ustrlen(bulkId)); + } if(r == 0) r = es_addBuf(&pData->batch.data, META_END, sizeof(META_END)-1); if(r == 0) r = es_addBuf(&pData->batch.data, (char*)message, length); if(r == 0) r = es_addBuf(&pData->batch.data, "\n", sizeof("\n")-1); @@ -409,7 +448,7 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) ssize_t wrRet; char errStr[1024]; DEFiRet; - + if(pData->errorFile == NULL) { DBGPRINTF("omelasticsearch: no local error logger defined - " "ignoring ES error information\n"); @@ -524,7 +563,7 @@ checkResult(instanceData *pData, uchar *reqmsg) } /* Note: we ignore errors writing the error file, as we cannot handle - * these in any case. + * these in any case. */ if(iRet == RS_RET_DATAFAIL) { writeDataError(pData, &root, reqmsg); @@ -552,8 +591,8 @@ curlPost(instanceData *pData, uchar *message, int msglen, uchar **tpls) CHKiRet(setCurlURL(pData, tpls)); curl_easy_setopt(curl, CURLOPT_WRITEDATA, pData); - curl_easy_setopt(curl, CURLOPT_POSTFIELDS, (char *)message); - curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, msglen); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, (char *)message); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, msglen); code = curl_easy_perform(curl); switch (code) { case CURLE_COULDNT_RESOLVE_HOST: @@ -649,10 +688,10 @@ curlSetup(instanceData *pData) } header = curl_slist_append(NULL, "Content-Type: text/json; charset=utf-8"); - curl_easy_setopt(handle, CURLOPT_HTTPHEADER, header); + curl_easy_setopt(handle, CURLOPT_HTTPHEADER, header); curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curlResult); - curl_easy_setopt(handle, CURLOPT_POST, 1); + curl_easy_setopt(handle, CURLOPT_POST, 1); pData->curlHandle = handle; pData->postHeader = header; @@ -690,6 +729,8 @@ setInstParamDefaults(instanceData *pData) pData->bulkmode = 0; pData->tplName = NULL; pData->errorFile = NULL; + pData->dynBulkId= 0; + pData->bulkId = NULL; } BEGINnewActInst @@ -737,12 +778,16 @@ CODESTARTnewActInst pData->asyncRepl = pvals[i].val.d.n; } else if(!strcmp(actpblk.descr[i].name, "template")) { pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); + } else if(!strcmp(actpblk.descr[i].name, "dynbulkid")) { + pData->dynBulkId = pvals[i].val.d.n; + } else if(!strcmp(actpblk.descr[i].name, "bulkid")) { + pData->bulkId = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else { dbgprintf("omelasticsearch: program error, non-handled " "param '%s'\n", actpblk.descr[i].name); } } - + if(pData->pwd != NULL && pData->uid == NULL) { errmsg.LogError(0, RS_RET_UID_MISSING, "omelasticsearch: password is provided, but no uid " @@ -767,6 +812,12 @@ CODESTARTnewActInst "name for parent template given - action definition invalid"); ABORT_FINALIZE(RS_RET_CONFIG_ERROR); } + if(pData->dynBulkId && pData->bulkId == NULL) { + errmsg.LogError(0, RS_RET_CONFIG_ERROR, + "omelasticsearch: requested dynamic bulkid, but no " + "name for bulkid template given - action definition invalid"); + ABORT_FINALIZE(RS_RET_CONFIG_ERROR); + } if(pData->bulkmode) { pData->batch.currTpl1 = NULL; @@ -782,6 +833,7 @@ CODESTARTnewActInst if(pData->dynSrchIdx) ++iNumTpls; if(pData->dynSrchType) ++iNumTpls; if(pData->dynParent) ++iNumTpls; + if(pData->dynBulkId) ++iNumTpls; DBGPRINTF("omelasticsearch: requesting %d templates\n", iNumTpls); CODE_STD_STRING_REQUESTnewActInst(iNumTpls) @@ -803,11 +855,29 @@ CODESTARTnewActInst if(pData->dynParent) { CHKiRet(OMSRsetEntry(*ppOMSR, 3, ustrdup(pData->parent), OMSR_NO_RQD_TPL_OPTS)); + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 4, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } + } else { + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 3, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } } } else { if(pData->dynParent) { CHKiRet(OMSRsetEntry(*ppOMSR, 2, ustrdup(pData->parent), OMSR_NO_RQD_TPL_OPTS)); + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 3, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } + } else { + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 2, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } } } } else { @@ -817,12 +887,30 @@ CODESTARTnewActInst if(pData->dynParent) { CHKiRet(OMSRsetEntry(*ppOMSR, 2, ustrdup(pData->parent), OMSR_NO_RQD_TPL_OPTS)); + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 3, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } + } else { + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 2, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } } } else { if(pData->dynParent) { CHKiRet(OMSRsetEntry(*ppOMSR, 1, ustrdup(pData->parent), OMSR_NO_RQD_TPL_OPTS)); - } + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 2, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } + } else { + if(pData->dynBulkId) { + CHKiRet(OMSRsetEntry(*ppOMSR, 1, ustrdup(pData->bulkId), + OMSR_NO_RQD_TPL_OPTS)); + } + } } } -- cgit v1.2.3 From 80f88242982c9c6ad6ce8628fc5b94ea74051cf4 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Tue, 2 Jul 2013 11:48:52 +0200 Subject: bugfix: double-free in omelasticsearch closes: http://bugzilla.adiscon.com/show_bug.cgi?id=461 Thanks to Marius Ionescu for providing a detailled bug report --- plugins/omelasticsearch/omelasticsearch.c | 1 - 1 file changed, 1 deletion(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 33e58c1a..57abefdd 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -483,7 +483,6 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) 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! */ -- cgit v1.2.3 From ebdcba5a6a5c9a20a529b62273a7ad852a793f09 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 4 Jul 2013 18:54:55 +0200 Subject: minor code reduction --- plugins/omelasticsearch/omelasticsearch.c | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 57abefdd..cb96fb6b 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -487,8 +487,7 @@ writeDataError(instanceData *pData, cJSON **pReplyRoot, uchar *reqmsg) *pReplyRoot = NULL; /* tell caller not to delete once again! */ finalize_it: - if(rendered != NULL) - free(rendered); + free(rendered); RETiRet; } -- cgit v1.2.3 From c1fbb1801229f9afd13abfa96ebf5f20acb3a033 Mon Sep 17 00:00:00 2001 From: Andre Lorbach Date: Tue, 23 Jul 2013 12:33:33 +0200 Subject: Fixed bugid #464 in elasticsearch output module Added some libcurl options in checkConn to disable the bodyport of the request within this functions. Otherwise libcurl could have created a sigsegv when curl_easy_perform() was called. --- plugins/omelasticsearch/omelasticsearch.c | 6 ++++++ 1 file changed, 6 insertions(+) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index cb96fb6b..5c53ba19 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -224,6 +224,12 @@ checkConn(instanceData *pData) DBGPRINTF("omelasticsearch: checkConn() curl_easy_init() failed\n"); ABORT_FINALIZE(RS_RET_SUSPENDED); } + /* Bodypart of request not needed, so set curl opt to nobody and httpget, otherwise lib-curl could sigsegv */ + curl_easy_setopt(curl, CURLOPT_HTTPGET, TRUE); + curl_easy_setopt(curl, CURLOPT_NOBODY, TRUE); + /* Only enable for debugging + curl_easy_setopt(curl, CURLOPT_VERBOSE, TRUE); */ + cstr = es_str2cstr(url, NULL); curl_easy_setopt(curl, CURLOPT_URL, cstr); free(cstr); -- cgit v1.2.3 From 030e8360702c2c5e227fa70b257f8b3fd048333b Mon Sep 17 00:00:00 2001 From: Andre Lorbach Date: Wed, 24 Jul 2013 09:50:14 +0200 Subject: Fixed a bug in curlPost related to http://bugzilla.adiscon.com/show_bug.cgi?id=464 A Check was added for replylen before writting the 0 Byte to the end of reply data. --- plugins/omelasticsearch/omelasticsearch.c | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index 5c53ba19..b98a7828 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -613,8 +613,11 @@ curlPost(instanceData *pData, uchar *message, int msglen, uchar **tpls) break; } - pData->reply[pData->replyLen] = '\0'; /* byte has been reserved in malloc */ - DBGPRINTF("omelasticsearch: es reply: '%s'\n", pData->reply); + DBGPRINTF("omelasticsearch: pData replyLen = '%d'\n", pData->replyLen); + if (pData->replyLen > 0) { + pData->reply[pData->replyLen] = '\0'; /* Append 0 Byte if replyLen is above 0 - byte has been reserved in malloc */ + } + DBGPRINTF("omelasticsearch: pData reply: '%s'\n", pData->reply); CHKiRet(checkResult(pData, message)); finalize_it: -- cgit v1.2.3 From 372c7264a598e72f6b9e8f734c9ea6a4a9427c8b Mon Sep 17 00:00:00 2001 From: Andre Lorbach Date: Wed, 24 Jul 2013 11:10:01 +0200 Subject: Fixed another issue related to http://bugzilla.adiscon.com/show_bug.cgi?id=464 Added check for empty batch data in endTransaction function. Avoids sending an unneeded curlPost, can happen when elasticsearch is shutdown while sending data to. --- plugins/omelasticsearch/omelasticsearch.c | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index b98a7828..aea8e321 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -651,12 +651,17 @@ ENDdoAction BEGINendTransaction - char *cstr; + char *cstr = NULL; CODESTARTendTransaction dbgprintf("omelasticsearch: endTransaction init\n"); - cstr = es_str2cstr(pData->batch.data, NULL); - dbgprintf("omelasticsearch: endTransaction, batch: '%s'\n", cstr); - CHKiRet(curlPost(pData, (uchar*) cstr, strlen(cstr), NULL)); + /* End Transaction only if batch data is not empty */ + if (pData->batch.data != NULL ) { + cstr = es_str2cstr(pData->batch.data, NULL); + dbgprintf("omelasticsearch: endTransaction, batch: '%s'\n", cstr); + CHKiRet(curlPost(pData, (uchar*) cstr, strlen(cstr), NULL)); + } + else + dbgprintf("omelasticsearch: endTransaction, pData->batch.data is NULL, nothing to send. \n"); finalize_it: free(cstr); dbgprintf("omelasticsearch: endTransaction done with %d\n", iRet); -- cgit v1.2.3 From c2dd607f9bac232f862158080741f449f834eeef Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Wed, 11 Sep 2013 16:26:09 +0200 Subject: omelasticsearch: straighten out statistics counters Conflicts: plugins/omelasticsearch/omelasticsearch.c --- plugins/omelasticsearch/omelasticsearch.c | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) (limited to 'plugins/omelasticsearch/omelasticsearch.c') diff --git a/plugins/omelasticsearch/omelasticsearch.c b/plugins/omelasticsearch/omelasticsearch.c index aea8e321..d49667f9 100644 --- a/plugins/omelasticsearch/omelasticsearch.c +++ b/plugins/omelasticsearch/omelasticsearch.c @@ -58,10 +58,9 @@ DEFobjCurrIf(errmsg) DEFobjCurrIf(statsobj) statsobj_t *indexStats; -STATSCOUNTER_DEF(indexConFail, mutIndexConFail) STATSCOUNTER_DEF(indexSubmit, mutIndexSubmit) -STATSCOUNTER_DEF(indexFailed, mutIndexFailed) -STATSCOUNTER_DEF(indexSuccess, mutIndexSuccess) +STATSCOUNTER_DEF(indexHTTPFail, mutIndexHTTPFail) +STATSCOUNTER_DEF(indexESFail, mutIndexESFail) /* REST API for elasticsearch hits this URL: * http://:// @@ -540,6 +539,9 @@ DBGPRINTF("omelasticsearch: %d items in reply\n", numitems); } finalize_it: + if(iRet != RS_RET_OK) { + STATSCOUNTER_INC(indexESFail, mutIndexESFail); + } RETiRet; } @@ -603,13 +605,12 @@ curlPost(instanceData *pData, uchar *message, int msglen, uchar **tpls) case CURLE_COULDNT_RESOLVE_PROXY: case CURLE_COULDNT_CONNECT: case CURLE_WRITE_ERROR: - STATSCOUNTER_INC(indexConFail, mutIndexConFail); + STATSCOUNTER_INC(indexHTTPFail, mutIndexHTTPFail); DBGPRINTF("omelasticsearch: we are suspending ourselfs due " "to failure %lld of curl_easy_perform()\n", (long long) code); ABORT_FINALIZE(RS_RET_SUSPENDED); default: - STATSCOUNTER_INC(indexSubmit, mutIndexSubmit); break; } @@ -639,6 +640,7 @@ ENDbeginTransaction BEGINdoAction CODESTARTdoAction + STATSCOUNTER_INC(indexSubmit, mutIndexSubmit); if(pData->bulkmode) { CHKiRet(buildBatch(pData, ppString[0], ppString)); } else { @@ -993,15 +995,13 @@ CODEmodInit_QueryRegCFSLineHdlr /* support statistics gathering */ CHKiRet(statsobj.Construct(&indexStats)); - CHKiRet(statsobj.SetName(indexStats, (uchar *)"elasticsearch")); - CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"connfail", - ctrType_IntCtr, &indexConFail)); - CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"submits", + CHKiRet(statsobj.SetName(indexStats, (uchar *)"omelasticsearch")); + CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"submitted", ctrType_IntCtr, &indexSubmit)); - CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"failed", - ctrType_IntCtr, &indexFailed)); - CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"success", - ctrType_IntCtr, &indexSuccess)); + CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"failed.http", + ctrType_IntCtr, &indexHTTPFail)); + CHKiRet(statsobj.AddCounter(indexStats, (uchar *)"failed.es", + ctrType_IntCtr, &indexESFail)); CHKiRet(statsobj.ConstructFinalize(indexStats)); ENDmodInit -- cgit v1.2.3