imptcp: added basic stats counters

1 files changed, 32 insertions, 10 deletions

diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c
index 65fe703c..951b24e4 100644
--- a/plugins/imptcp/imptcp.c
+++ b/plugins/imptcp/imptcp.c
@@ -65,6 +65,7 @@
 #include "datetime.h"
 #include "ruleset.h"
 #include "msg.h"
+#include "statsobj.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 ... */
@@ -82,7 +83,7 @@ DEFobjCurrIf(prop)
 DEFobjCurrIf(datetime)
 DEFobjCurrIf(errmsg)
 DEFobjCurrIf(ruleset)
-
+DEFobjCurrIf(statsobj)
 
 
 /* config settings */
@@ -123,7 +124,8 @@ struct ptcpsrv_s {
  * includes support for doubly-linked list.
  */
 struct ptcpsess_s {
-	ptcpsrv_t *pSrv;	/* our server */
+//	ptcpsrv_t *pSrv;	/* our server TODO: check remove! */
+	ptcplstn_t *pLstn;	/* our listener */
 	ptcpsess_t *prev, *next;
 	int sock;
 	epolld_t *epd;
@@ -151,6 +153,8 @@ struct ptcplstn_s {
 	ptcplstn_t *prev, *next;
 	int sock;
 	epolld_t *epd;
+	statsobj_t *stats;	/* listener stats */
+	STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit)
 };
 
 
@@ -492,22 +496,25 @@ static rsRetVal
 doSubmitMsg(ptcpsess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub)
 {
 	msg_t *pMsg;
+	ptcpsrv_t *pSrv;
 	DEFiRet;
 
 	if(pThis->iMsg == 0) {
 		DBGPRINTF("discarding zero-sized message\n");
 		FINALIZE;
 	}
+	pSrv = pThis->pLstn->pSrv;
 
 	/* we now create our own message object and submit it to the queue */
 	CHKiRet(msgConstructWithTime(&pMsg, stTime, ttGenTime));
 	MsgSetRawMsg(pMsg, (char*)pThis->pMsg, pThis->iMsg);
-	MsgSetInputName(pMsg, pThis->pSrv->pInputName);
+	MsgSetInputName(pMsg, pSrv->pInputName);
 	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
 	pMsg->msgFlags  = NEEDS_PARSING | PARSE_HOSTNAME;
 	MsgSetRcvFrom(pMsg, pThis->peerName);
 	CHKiRet(MsgSetRcvFromIP(pMsg, pThis->peerIP));
-	MsgSetRuleset(pMsg, pThis->pSrv->pRuleset);
+	MsgSetRuleset(pMsg, pSrv->pRuleset);
+	STATSCOUNTER_INC(pThis->pLstn->ctrSubmit, pThis->pLstn->mutCtrSubmit);
 
 	if(pMultiSub == NULL) {
 		CHKiRet(submitMsg(pMsg));
@@ -589,7 +596,8 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG
 		}
 
 		if((   (c == '\n')
-		   || ((pThis->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->iAddtlFrameDelim))
+		   || ((pThis->pLstn->pSrv->iAddtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER)
+		       && (c == pThis->pLstn->pSrv->iAddtlFrameDelim))
 		   ) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */
 			doSubmitMsg(pThis, stTime, ttGenTime, pMultiSub);
 			pThis->inputState = eAtStrtFram;
@@ -751,10 +759,19 @@ addLstn(ptcpsrv_t *pSrv, int sock)
 {
 	DEFiRet;
 	ptcplstn_t *pLstn;
+	uchar statname[64];
 
 	CHKmalloc(pLstn = malloc(sizeof(ptcplstn_t)));
 	pLstn->pSrv = pSrv;
 	pLstn->sock = sock;
+	/* support statistics gathering */
+	CHKiRet(statsobj.Construct(&(pLstn->stats)));
+	snprintf((char*)statname, sizeof(statname), "imptcp(%s)", pSrv->port);
+	statname[sizeof(statname)-1] = '\0'; /* just to be on the save side... */
+	CHKiRet(statsobj.SetName(pLstn->stats, statname));
+	CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("submitted"),
+		ctrType_IntCtr, &(pLstn->ctrSubmit)));
+	CHKiRet(statsobj.ConstructFinalize(pLstn->stats));
 
 	/* add to start of server's listener list */
 	pLstn->prev = NULL;
@@ -773,14 +790,15 @@ finalize_it:
 /* add a session to the server 
  */
 static rsRetVal
-addSess(ptcpsrv_t *pSrv, int sock, prop_t *peerName, prop_t *peerIP)
+addSess(ptcplstn_t *pLstn, int sock, prop_t *peerName, prop_t *peerIP)
 {
 	DEFiRet;
 	ptcpsess_t *pSess = NULL;
+	ptcpsrv_t *pSrv = pLstn->pSrv;
 
 	CHKmalloc(pSess = malloc(sizeof(ptcpsess_t)));
 	CHKmalloc(pSess->pMsg = malloc(iMaxLine * sizeof(uchar)));
-	pSess->pSrv = pSrv;
+	pSess->pLstn = pLstn;
 	pSess->sock = sock;
 	pSess->inputState = eAtStrtFram;
 	pSess->iMsg = 0;
@@ -824,7 +842,7 @@ closeSess(ptcpsess_t *pSess)
 		pSess->next->prev = pSess->prev;
 	if(pSess->prev == NULL) {
 		/* need to update root! */
-		pSess->pSrv->pSess = pSess->next;
+		pSess->pLstn->pSrv->pSess = pSess->next;
 	} else {
 		pSess->prev->next = pSess->next;
 	}
@@ -949,7 +967,7 @@ lstnActivity(ptcplstn_t *pLstn)
 		if(localRet == RS_RET_NO_MORE_DATA)
 			break;
 		CHKiRet(localRet);
-		CHKiRet(addSess(pLstn->pSrv, newSock, peerName, peerIP));
+		CHKiRet(addSess(pLstn, newSock, peerName, peerIP));
 	}
 
 finalize_it:
@@ -979,7 +997,7 @@ sessActivity(ptcpsess_t *pSess)
 			CHKiRet(DataRcvd(pSess, rcvBuf, lenRcv));
 		} else if (lenRcv == 0) {
 			/* session was closed, do clean-up */
-			if(pSess->pSrv->bEmitMsgOnClose) {
+			if(pSess->pLstn->pSrv->bEmitMsgOnClose) {
 				uchar *peerName;
 				int lenPeer;
 				prop.GetString(pSess->peerName, &peerName, &lenPeer);
@@ -1085,6 +1103,8 @@ shutdownSrv(ptcpsrv_t *pSrv)
 	pLstn = pSrv->pLstn;
 	while(pLstn != NULL) {
 		close(pLstn->sock);
+		statsobj.Destruct(&(pLstn->stats));
+		/* now unlink listner */
 		lstnDel = pLstn;
 		pLstn = pLstn->next;
 		DBGPRINTF("imptcp shutdown listen socket %d\n", lstnDel->sock);
@@ -1132,6 +1152,7 @@ CODESTARTmodExit
 
 	/* release objects we used */
 	objRelease(glbl, CORE_COMPONENT);
+	objRelease(statsobj, CORE_COMPONENT);
 	objRelease(prop, CORE_COMPONENT);
 	objRelease(net, LM_NET_FILENAME);
 	objRelease(datetime, CORE_COMPONENT);
@@ -1167,6 +1188,7 @@ CODEmodInit_QueryRegCFSLineHdlr
 	initConfigSettings();
 	/* request objects we use */
 	CHKiRet(objUse(glbl, CORE_COMPONENT));
+	CHKiRet(objUse(statsobj, CORE_COMPONENT));
 	CHKiRet(objUse(prop, CORE_COMPONENT));
 	CHKiRet(objUse(net, LM_NET_FILENAME));
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));