Implement ACT_FIELDS_PASSING, test in mongodb

The mongodb test contains only debug printfs.

Example template, inspired by writeMongoDB_msg:

$template MongoTemplate,"%hostname::::sys%%timereported::::time%%timegenerated::::time_rcvd%%msg%%syslogfacility-text::::syslog_fac%%syslogseverity-text::::syslog_server%%syslogtag::::syslog_tag%%programname::::procid%%procid::::pid%%$!foo::::foo%%$!abc::::renamed%"

Note that JSON escaping is actually undesirable in this mode (should it
be silently ignored?), $!all-json doesn't yet work as expected, and all
data is stored as strings.

Signed-off-by: Miloslav Trmač <mitr@redhat.com>

5 files changed, 166 insertions, 7 deletions

diff --git a/action.c b/action.c
index d1dcad89..9c780fb6 100644
--- a/action.c
+++ b/action.c
@@ -823,9 +823,13 @@ static rsRetVal prepareDoActionParams(action_t *pAction, batch_obj_t *pElem)
 			case ACT_MSG_PASSING:
 				pElem->staticActParams[i] = (void*) pMsg;
 				break;
-			case ACT_FIELDS_PASSING:
-				pElem->staticActParams[i] = "";
+			case ACT_FIELDS_PASSING: {
+				struct templateField *fields;
+
+				CHKiRet(tplToFields(pAction->ppTpl[i], pMsg, &fields));
+				pElem->staticActParams[i] = fields;
 				break;
+			}
 			default:dbgprintf("software bug/error: unknown pAction->eParamPassing %d in prepareDoActionParams\n",
 					   (int) pAction->eParamPassing);
 				assert(0); /* software bug if this happens! */
@@ -877,7 +881,6 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch)
 				break;
 			case ACT_STRING_PASSING:
 			case ACT_MSG_PASSING:
-			case ACT_FIELDS_PASSING:
 				/* nothing to do in that case */
 				/* TODO ... and yet we do something ;) This is considered not
 				 * really needed, but I was not bold enough to remove that while
@@ -889,6 +892,21 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch)
 					((uchar**)pElem->staticActParams)[j] = NULL;
 				}
 				break;
+			case ACT_FIELDS_PASSING:
+				for(j = 0 ; j < pAction->iNumTpls ; ++j) {
+					struct templateField *fields;
+
+					fields = pElem->staticActParams[j];
+					if (fields!= NULL) {
+						size_t k;
+
+						for (k = 0; fields[k].fieldName != NULL; k++)
+							free(fields[k].value);
+						free(fields);
+						pElem->staticActParams[j] = NULL;
+					}
+				}
+				break;
 			}
 		}
 	}
diff --git a/plugins/ommongodb/ommongodb.c b/plugins/ommongodb/ommongodb.c
index 00afcf68..8873d5ac 100644
--- a/plugins/ommongodb/ommongodb.c
+++ b/plugins/ommongodb/ommongodb.c
@@ -297,6 +297,57 @@ finalize_it:
 	RETiRet;
 }
 
+/* write to mongodb based on a field template. */
+static rsRetVal
+writeMongoDB_fields(struct templateField *fields, instanceData *pData)
+{
+	bson *doc = NULL;
+	size_t i;
+	DEFiRet;
+
+	/* see if we are ready to proceed */
+	if(pData->conn == NULL) {
+		CHKiRet(initMongoDB(pData, 0));
+	}
+
+	dbgprintf("+++start+++\n");
+	for(i = 0; fields[i].fieldName != NULL; i++) {
+		dbgprintf("%.*s: %s\n", (int)es_strlen(fields[i].fieldName),
+			  es_getBufAddr(fields[i].fieldName), fields[i].value);
+	}
+	dbgprintf("---end---\n");
+#if 0	
+	doc = bson_build(BSON_TYPE_STRING, "sys", sys, sys_len,
+			 BSON_TYPE_UTC_DATETIME, "time", ts_gen,
+			 BSON_TYPE_UTC_DATETIME, "time_rcvd", ts_rcv,
+			 BSON_TYPE_STRING, "msg", msg, msg_len,
+			 BSON_TYPE_INT32, "syslog_fac", facil,
+			 BSON_TYPE_INT32, "syslog_sever", severity,
+			 BSON_TYPE_STRING, "syslog_tag", tag, tag_len,
+			 BSON_TYPE_STRING, "procid", procid, procid_len,
+			 BSON_TYPE_STRING, "pid", pid, pid_len,
+			 BSON_TYPE_STRING, "level", getLumberjackLevel(pMsg->iSeverity), -1,
+			 BSON_TYPE_NONE);
+
+	if(doc == NULL) {
+		reportMongoError(pData);
+		dbgprintf("ommongodb: error creating BSON doc\n");
+		ABORT_FINALIZE(RS_RET_SUSPENDED);
+	}
+	bson_finish(doc);
+	if(!mongo_sync_cmd_insert(pData->conn, (char*)pData->dbNcoll, doc, NULL)) {
+		reportMongoError(pData);
+		dbgprintf("ommongodb: insert error\n");
+		ABORT_FINALIZE(RS_RET_SUSPENDED);
+	}
+#endif
+
+finalize_it:
+	if(doc != NULL)
+		bson_free(doc);
+	RETiRet;
+}
+
 BEGINtryResume
 CODESTARTtryResume
 	if(pData->conn == NULL) {
@@ -308,6 +359,9 @@ BEGINdoAction
 CODESTARTdoAction
 	if(pData->tplName == NULL) {
 		iRet = writeMongoDB_msg((msg_t*)ppString[0], pData);
+	} else {
+		iRet = writeMongoDB_fields((struct templateField *)ppString[0],
+					   pData);
 	}
 ENDdoAction
 
@@ -363,12 +417,9 @@ CODESTARTnewActInst
 	if(pData->tplName == NULL) {
 		CHKiRet(OMSRsetEntry(*ppOMSR, 0, NULL, OMSR_TPL_AS_MSG));
 	} else {
-		errmsg.LogError(0, RS_RET_LEGA_ACT_NOT_SUPPORTED,
-			"ommongodb: templates are not supported in this version");
-		ABORT_FINALIZE(RS_RET_ERR);
 		CHKiRet(OMSRsetEntry(*ppOMSR, 0,
 			(uchar*) strdup((char*) pData->tplName),
-			OMSR_TPL_AS_ARRAY));
+			OMSR_TPL_AS_FIELDS));
 	}
 
 	if(pData->db == NULL)
@@ -420,9 +471,31 @@ CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
 ENDqueryEtryPt
 
 BEGINmodInit()
+	rsRetVal localRet;
+	rsRetVal (*pomsrGetSupportedTplOpts)(unsigned long *pOpts);
+	unsigned long opts;
+	int bFieldPassingSupported;
 CODESTARTmodInit
 	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
 CODEmodInit_QueryRegCFSLineHdlr
+	/* check if the rsyslog core supports parameter passing code */
+	bFieldPassingSupported = 0;
+	localRet = pHostQueryEtryPt((uchar*)"OMSRgetSupportedTplOpts",
+			&pomsrGetSupportedTplOpts);
+	if(localRet == RS_RET_OK) {
+		/* found entry point, so let's see if core supports msg passing */
+		CHKiRet((*pomsrGetSupportedTplOpts)(&opts));
+		if(opts & OMSR_TPL_AS_FIELDS)
+			bFieldPassingSupported = 1;
+	} else if(localRet != RS_RET_ENTRY_POINT_NOT_FOUND) {
+		ABORT_FINALIZE(localRet); /* Something else went wrong, not acceptable */
+	}
+	if(!bFieldPassingSupported) {
+		DBGPRINTF("ommongodb: field-passing is not supported by rsyslog core, "
+			  "can not continue.\n");
+		ABORT_FINALIZE(RS_RET_NO_FIELD_PASSING);
+	}
+
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
 	CHKiRet(objUse(datetime, CORE_COMPONENT));
 	INITChkCoreFeature(bCoreSupportsBatching, CORE_FEATURE_BATCHING);
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 57e8a05c..8d91f0e7 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -380,6 +380,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_DUP_PARAM = -2220, /**< config parameter is given more than once */
 	RS_RET_MODULE_ALREADY_IN_CONF = -2221, /**< module already in current configuration */
 	RS_RET_PARAM_NOT_PERMITTED = -2222, /**< legacy parameter no longer permitted (usally already set by v2) */
+	RS_RET_NO_FIELD_PASSING = -2223, /**< output module interface parameter passing mode "FIELDS" is not available but required */
 
 	/* RainerScript error messages (range 1000.. 1999) */
 	RS_RET_SYSVAR_NOT_FOUND = 1001, /**< system variable could not be found (maybe misspelled) */
diff --git a/template.c b/template.c
index 768608f1..d422f8ee 100644
--- a/template.c
+++ b/template.c
@@ -269,6 +269,65 @@ finalize_it:
 	RETiRet;
 }
 
+/* This functions converts a template into an array of templateField structures.
+ * For further general details, see the very similar funtion
+ * tpltoString().
+ * The caller is repsonsible for destroying that array as well as all of its
+ * elements (but not the fieldName strings).
+ */
+rsRetVal tplToFields(struct template *pTpl, msg_t *pMsg,
+		     struct templateField **pFields)
+{
+	DEFiRet;
+	struct templateEntry *pTpe;
+	struct templateField *fields;
+	size_t i, propLen;
+	unsigned short bMustBeFreed;
+	uchar *pVal;
+
+	assert(pTpl != NULL);
+	assert(pMsg != NULL);
+	assert(pFields != NULL);
+
+	/* loop through the template. We obtain one value, create a
+	 * private copy (if necessary), add it to the string array
+	 * and then on to the next until we have processed everything.
+	 */
+
+	/* The zeroization implicitly terminates the output array. */
+	CHKmalloc(fields = calloc(pTpl->tpenElements + 1, sizeof(*fields)));
+
+	i = 0;
+	for(pTpe = pTpl->pEntryRoot; pTpe != NULL; pTpe = pTpe->pNext) {
+		if(pTpe->eEntryType == CONSTANT) {
+			/* Completely ingore this - we don't know a field name
+			   to use anyway. */
+		} else if(pTpe->eEntryType == FIELD) {
+			fields[i].fieldName = pTpe->data.field.fieldName;
+			pVal = MsgGetProp(pMsg, pTpe, pTpe->data.field.propid,
+					  pTpe->data.field.propName, &propLen,
+					  &bMustBeFreed);
+			if(bMustBeFreed) { /* if it must be freed, it is our own private copy... */
+				fields[i].value = pVal; /* ... so we can use it! */
+			} else {
+				CHKmalloc(fields[i].value = (uchar*)strdup((char*) pVal));
+			}
+			i++;
+		}
+	}
+
+finalize_it:
+	if(iRet != RS_RET_OK && fields != NULL) {
+		for(i = 0; fields[i].fieldName != NULL; i++)
+			free(fields[i].value);
+		free(fields);
+		fields = NULL;
+	}
+	*pFields = fields;
+
+	RETiRet;
+}
+
 
 /* Helper to doEscape. This is called if doEscape
  * runs out of memory allocating the escaped string.
diff --git a/template.h b/template.h
index 9f6a4c33..218f8b27 100644
--- a/template.h
+++ b/template.h
@@ -119,6 +119,13 @@ struct templateEntry {
 	} data;
 };
 
+/* A field in a list returned by OMSR_TPL_AS_FIELDS / ACT_FIELDS_PASSING.
+   Terminated by an entry with NULL fieldName. */
+struct templateField {
+	es_str_t *fieldName;	/* Borrowed reference, do not free! */
+	uchar *value;		/* For free() */
+};
+
 
 /* interfaces */
 BEGINinterface(tpl) /* name must also be changed in ENDinterface macro! */
@@ -144,6 +151,7 @@ rsRetVal ExtendBuf(uchar **pBuf, size_t *pLenBuf, size_t iMinSize);
  * rgerhards, 2007-08-06
  */
 rsRetVal tplToArray(struct template *pTpl, msg_t *pMsg, uchar*** ppArr);
+rsRetVal tplToFields(struct template *pTpl, msg_t *pMsg, struct templateField **pFields);
 rsRetVal tplToString(struct template *pTpl, msg_t *pMsg, uchar** ppSz, size_t *);
 rsRetVal doEscape(uchar **pp, size_t *pLen, unsigned short *pbMustBeFreed, int escapeMode);