From: Jiri Vymazal <jvymazal@redhat.com>

Date: Mon, 28 Jun 2018 12:07:55 +0100

Subject: Kubernetes Metadata plugin - mmkubernetes

This plugin is used to annotate records logged by Kubernetes containers.

It will add the namespace uuid, pod uuid, pod and namespace labels and

annotations, and other metadata associated with the pod and namespace.

It will work with either log files in `/var/log/containers/*.log` or

with journald entries with `CONTAINER_NAME` and `CONTAINER_ID_FULL`.

For usage and configuration see syslog-doc

*Credits*

This work is based on https://github.com/fabric8io/fluent-plugin-kubernetes_metadata_filter

and has many of the same features.

(cherry picked from commit a6264bf8f91975c9bc0fc602dcdc6881486f1579)

(cherry picked from commit b8e68366422052dca9e0a9409baa410f20ae88c8)

(cherry picked from commit 77886e21292d8220f93b3404236da0e8f7159255)

(cherry picked from commit e4d1c7b3832eedc8a1545c2ee6bf022f545d0c76)

(cherry picked from commit 3d9f820642b0edc78da0b5bed818590dcd31fa9c)

(cherry picked from commit 1d49aac5cb101704486bfb065fac362ca69f06bc)

(cherry picked from commit fc2ad45f78dd666b8c9e706ad88c17aaff146d2d)

(cherry picked from commit 8cf87f64f6c74a4544112ec7fddc5bf4d43319a7)

---

 Makefile.am                                      |    5 +

 configure.ac                                     |   35 +

 contrib/mmkubernetes/Makefile.am                 |    6 +

 contrib/mmkubernetes/k8s_container_name.rulebase |    3 +

 contrib/mmkubernetes/k8s_filename.rulebase       |    2 +

 contrib/mmkubernetes/mmkubernetes.c              | 1491 +++++++++++++++++++++++

 contrib/mmkubernetes/sample.conf                 |    7 +

 7 files changed, 1549 insertions(+)

 create mode 100644 contrib/mmkubernetes/Makefile.am

 create mode 100644 contrib/mmkubernetes/k8s_container_name.rulebase

 create mode 100644 contrib/mmkubernetes/k8s_filename.rulebase

 create mode 100644 contrib/mmkubernetes/mmkubernetes.c

 create mode 100644 contrib/mmkubernetes/sample.conf

diff --git a/Makefile.am b/Makefile.am

index a276ef9ea..b58ebaf93 100644

--- a/Makefile.am

+++ b/Makefile.am

@@ -275,6 +275,11 @@ if ENABLE_OMTCL

 SUBDIRS += contrib/omtcl

 endif

+# mmkubernetes

+if ENABLE_MMKUBERNETES

+SUBDIRS += contrib/mmkubernetes

+endif

+

 # tests are added as last element, because tests may need different

 # modules that need to be generated first

 SUBDIRS += tests

diff --git a/configure.ac b/configure.ac

index a9411f4be..c664222b9 100644

--- a/configure.ac

+++ b/configure.ac

@@ -1889,6 +1889,39 @@ AM_CONDITIONAL(ENABLE_OMTCL, test x$enable_omtcl = xyes)

 # END TCL SUPPORT

+# mmkubernetes - Kubernetes metadata support

+

+AC_ARG_ENABLE(mmkubernetes,

+        [AS_HELP_STRING([--enable-mmkubernetes],

+            [Enable compilation of the mmkubernetes module @<:@default=no@:>@])],

+        [case "${enableval}" in

+         yes) enable_mmkubernetes="yes" ;;

+          no) enable_mmkubernetes="no" ;;

+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-mmkubernetes) ;;

+         esac],

+        [enable_mmkubernetes=no]

+)

+if test "x$enable_mmkubernetes" = "xyes"; then

+        PKG_CHECK_MODULES([CURL], [libcurl])

+        PKG_CHECK_MODULES(LIBLOGNORM, lognorm >= 2.0.3)

+

+        save_CFLAGS="$CFLAGS"

+        save_LIBS="$LIBS"

+

+        CFLAGS="$CFLAGS $LIBLOGNORM_CFLAGS"

+        LIBS="$LIBS $LIBLOGNORM_LIBS"

+

+        AC_CHECK_FUNC([ln_loadSamplesFromString],

+                      [AC_DEFINE([HAVE_LOADSAMPLESFROMSTRING], [1], [Define if ln_loadSamplesFromString exists.])],

+                      [AC_DEFINE([NO_LOADSAMPLESFROMSTRING], [1], [Define if ln_loadSamplesFromString does not exist.])])

+

+        CFLAGS="$save_CFLAGS"

+        LIBS="$save_LIBS"

+fi

+AM_CONDITIONAL(ENABLE_MMKUBERNETES, test x$enable_mmkubernetes = xyes)

+

+# END Kubernetes metadata support

+

 # man pages

 AC_CHECKING([if required man pages already exist])

 have_to_generate_man_pages="no"

@@ -2016,6 +2035,7 @@ AC_CONFIG_FILES([Makefile \

 		contrib/omhttpfs/Makefile \

 		contrib/omamqp1/Makefile \

 		contrib/omtcl/Makefile \

+		contrib/mmkubernetes/Makefile \

 		tests/Makefile])

 AC_OUTPUT

@@ -2090,6 +2110,7 @@ echo "    mmrfc5424addhmac enabled:                 $enable_mmrfc5424addhmac"

 echo "    mmpstrucdata enabled:                     $enable_mmpstrucdata"

 echo "    mmsequence enabled:                       $enable_mmsequence"

 echo "    mmdblookup enabled:                       $enable_mmdblookup"

+echo "    mmkubernetes enabled:                     $enable_mmkubernetes"

 echo

 echo "---{ database support }---"

 echo "    MySql support enabled:                    $enable_mysql"

diff --git a/contrib/mmkubernetes/Makefile.am b/contrib/mmkubernetes/Makefile.am

new file mode 100644

index 000000000..3dcc235a6

--- /dev/null

+++ b/contrib/mmkubernetes/Makefile.am

@@ -0,0 +1,6 @@

+pkglib_LTLIBRARIES = mmkubernetes.la

+

+mmkubernetes_la_SOURCES = mmkubernetes.c

+mmkubernetes_la_CPPFLAGS = $(RSRT_CFLAGS) $(PTHREADS_CFLAGS) $(CURL_CFLAGS) $(LIBLOGNORM_CFLAGS)

+mmkubernetes_la_LDFLAGS = -module -avoid-version

+mmkubernetes_la_LIBADD = $(CURL_LIBS) $(LIBLOGNORM_LIBS)

diff --git a/contrib/mmkubernetes/k8s_container_name.rulebase b/contrib/mmkubernetes/k8s_container_name.rulebase

new file mode 100644

index 000000000..35fbb317c

--- /dev/null

+++ b/contrib/mmkubernetes/k8s_container_name.rulebase

@@ -0,0 +1,3 @@

+version=2

+rule=:%k8s_prefix:char-to:_%_%container_name:char-to:.%.%container_hash:char-to:_%_%pod_name:char-to:_%_%namespace_name:char-to:_%_%not_used_1:char-to:_%_%not_used_2:rest%

+rule=:%k8s_prefix:char-to:_%_%container_name:char-to:_%_%pod_name:char-to:_%_%namespace_name:char-to:_%_%not_used_1:char-to:_%_%not_used_2:rest%

diff --git a/contrib/mmkubernetes/k8s_filename.rulebase b/contrib/mmkubernetes/k8s_filename.rulebase

new file mode 100644

index 000000000..24c0d9138

--- /dev/null

+++ b/contrib/mmkubernetes/k8s_filename.rulebase

@@ -0,0 +1,2 @@

+version=2

+rule=:/var/log/containers/%pod_name:char-to:_%_%namespace_name:char-to:_%_%container_name_and_id:char-to:.%.log

diff --git a/contrib/mmkubernetes/mmkubernetes.c b/contrib/mmkubernetes/mmkubernetes.c

new file mode 100644

index 000000000..5012c54f6

--- /dev/null

+++ b/contrib/mmkubernetes/mmkubernetes.c

@@ -0,0 +1,1491 @@

+/* mmkubernetes.c

+ * This is a message modification module. It uses metadata obtained

+ * from the message to query Kubernetes and obtain additional metadata

+ * relating to the container instance.

+ *

+ * Inspired by:

+ * https://github.com/fabric8io/fluent-plugin-kubernetes_metadata_filter

+ *

+ * NOTE: read comments in module-template.h for details on the calling interface!

+ *

+ * Copyright 2016 Red Hat Inc.

+ *

+ * This file is part of rsyslog.

+ *

+ * 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.

+ * See the License for the specific language governing permissions and

+ * limitations under the License.

+ */

+

+/* needed for asprintf */

+#ifndef _GNU_SOURCE

+#  define _GNU_SOURCE

+#endif

+

+#include "config.h"

+#include "rsyslog.h"

+#include <stdio.h>

+#include <stdarg.h>

+#include <stdlib.h>

+#include <string.h>

+#include <assert.h>

+#include <errno.h>

+#include <unistd.h>

+#include <sys/stat.h>

+#include <libestr.h>

+#include <liblognorm.h>

+#include <json.h>

+#include <curl/curl.h>

+#include <curl/easy.h>

+#include <pthread.h>

+#include "conf.h"

+#include "syslogd-types.h"

+#include "module-template.h"

+#include "errmsg.h"

+#include "regexp.h"

+#include "hashtable.h"

+#include "srUtils.h"

+

+/* static data */

+MODULE_TYPE_OUTPUT /* this is technically an output plugin */

+MODULE_TYPE_KEEP /* releasing the module would cause a leak through libcurl */

+MODULE_CNFNAME("mmkubernetes")

+DEF_OMOD_STATIC_DATA

+DEFobjCurrIf(errmsg)

+DEFobjCurrIf(regexp)

+

+#define HAVE_LOADSAMPLESFROMSTRING 1

+#if defined(NO_LOADSAMPLESFROMSTRING)

+#undef HAVE_LOADSAMPLESFROMSTRING

+#endif

+/* original from fluentd plugin:

+ * 'var\.log\.containers\.(?<pod_name>[a-z0-9]([-a-z0-9]*[a-z0-9])?\

+ *   (\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace>[^_]+)_\

+ *   (?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\.log$'

+ * this is for _tag_ match, not actual filename match - in_tail turns filename

+ * into a fluentd tag

+ */

+#define DFLT_FILENAME_LNRULES "rule=:/var/log/containers/%pod_name:char-to:_%_"\

+	"%namespace_name:char-to:_%_%container_name:char-to:-%-%container_id:char-to:.%.log"

+#define DFLT_FILENAME_RULEBASE "/etc/rsyslog.d/k8s_filename.rulebase"

+/* original from fluentd plugin:

+ *   '^(?<name_prefix>[^_]+)_(?<container_name>[^\._]+)\

+ *     (\.(?<container_hash>[^_]+))?_(?<pod_name>[^_]+)_\

+ *     (?<namespace>[^_]+)_[^_]+_[^_]+$'

+ */

+#define DFLT_CONTAINER_LNRULES "rule=:%k8s_prefix:char-to:_%_%container_name:char-to:.%."\

+	"%container_hash:char-to:_%_"\

+	"%pod_name:char-to:_%_%namespace_name:char-to:_%_%not_used_1:char-to:_%_%not_used_2:rest%\n"\

+	"rule=:%k8s_prefix:char-to:_%_%container_name:char-to:_%_"\

+	"%pod_name:char-to:_%_%namespace_name:char-to:_%_%not_used_1:char-to:_%_%not_used_2:rest%"

+#define DFLT_CONTAINER_RULEBASE "/etc/rsyslog.d/k8s_container_name.rulebase"

+#define DFLT_SRCMD_PATH "$!metadata!filename"

+#define DFLT_DSTMD_PATH "$!"

+#define DFLT_DE_DOT 1 /* true */

+#define DFLT_DE_DOT_SEPARATOR "_"

+#define DFLT_CONTAINER_NAME "$!CONTAINER_NAME" /* name of variable holding CONTAINER_NAME value */

+#define DFLT_CONTAINER_ID_FULL "$!CONTAINER_ID_FULL" /* name of variable holding CONTAINER_ID_FULL value */

+#define DFLT_KUBERNETES_URL "https://kubernetes.default.svc.cluster.local:443"

+

+static struct cache_s {

+	const uchar *kbUrl;

+	struct hashtable *mdHt;

+	struct hashtable *nsHt;

+	pthread_mutex_t *cacheMtx;

+} **caches;

+

+typedef struct {

+	int nmemb;

+	uchar **patterns;

+	regex_t *regexps;

+} annotation_match_t;

+

+/* module configuration data */

+struct modConfData_s {

+	rsconf_t *pConf;	/* our overall config object */

+	uchar *kubernetesUrl;	/* scheme, host, port, and optional path prefix for Kubernetes API lookups */

+	uchar *srcMetadataPath;	/* where to get data for kubernetes queries */

+	uchar *dstMetadataPath;	/* where to put metadata obtained from kubernetes */

+	uchar *caCertFile; /* File holding the CA cert (+optional chain) of CA that issued the Kubernetes server cert */

+	sbool allowUnsignedCerts; /* For testing/debugging - do not check for CA certs (CURLOPT_SSL_VERIFYPEER FALSE) */

+	uchar *token; /* The token value to use to authenticate to Kubernetes - takes precedence over tokenFile */

+	uchar *tokenFile; /* The file whose contents is the token value to use to authenticate to Kubernetes */

+	sbool de_dot; /* If true (default), convert '.' characters in labels & annotations to de_dot_separator */

+	uchar *de_dot_separator; /* separator character (default '_') to use for de_dotting */

+	size_t de_dot_separator_len; /* length of separator character */

+	annotation_match_t annotation_match; /* annotation keys must match these to be included in record */

+	char *fnRules; /* lognorm rules for container log filename match */

+	uchar *fnRulebase; /* lognorm rulebase filename for container log filename match */

+	char *contRules; /* lognorm rules for CONTAINER_NAME value match */

+	uchar *contRulebase; /* lognorm rulebase filename for CONTAINER_NAME value match */

+};

+

+/* action (instance) configuration data */

+typedef struct _instanceData {

+	uchar *kubernetesUrl;	/* scheme, host, port, and optional path prefix for Kubernetes API lookups */

+	msgPropDescr_t *srcMetadataDescr;	/* where to get data for kubernetes queries */

+	uchar *dstMetadataPath;	/* where to put metadata obtained from kubernetes */

+	uchar *caCertFile; /* File holding the CA cert (+optional chain) of CA that issued the Kubernetes server cert */

+	sbool allowUnsignedCerts; /* For testing/debugging - do not check for CA certs (CURLOPT_SSL_VERIFYPEER FALSE) */

+	uchar *token; /* The token value to use to authenticate to Kubernetes - takes precedence over tokenFile */

+	uchar *tokenFile; /* The file whose contents is the token value to use to authenticate to Kubernetes */

+	sbool de_dot; /* If true (default), convert '.' characters in labels & annotations to de_dot_separator */

+	uchar *de_dot_separator; /* separator character (default '_') to use for de_dotting */

+	size_t de_dot_separator_len; /* length of separator character */

+	annotation_match_t annotation_match; /* annotation keys must match these to be included in record */

+	char *fnRules; /* lognorm rules for container log filename match */

+	uchar *fnRulebase; /* lognorm rulebase filename for container log filename match */

+	ln_ctx fnCtxln;	/**< context to be used for liblognorm */

+	char *contRules; /* lognorm rules for CONTAINER_NAME value match */

+	uchar *contRulebase; /* lognorm rulebase filename for CONTAINER_NAME value match */

+	ln_ctx contCtxln;	/**< context to be used for liblognorm */

+	msgPropDescr_t *contNameDescr; /* CONTAINER_NAME field */

+	msgPropDescr_t *contIdFullDescr; /* CONTAINER_ID_FULL field */

+	struct cache_s *cache;

+} instanceData;

+

+typedef struct wrkrInstanceData {

+	instanceData *pData;

+	CURL *curlCtx;

+	struct curl_slist *curlHdr;

+	char *curlRply;

+	size_t curlRplyLen;

+} wrkrInstanceData_t;

+

+/* module parameters (v6 config format) */

+static struct cnfparamdescr modpdescr[] = {

+	{ "kubernetesurl", eCmdHdlrString, 0 },

+	{ "srcmetadatapath", eCmdHdlrString, 0 },

+	{ "dstmetadatapath", eCmdHdlrString, 0 },

+	{ "tls.cacert", eCmdHdlrString, 0 },

+	{ "allowunsignedcerts", eCmdHdlrBinary, 0 },

+	{ "token", eCmdHdlrString, 0 },

+	{ "tokenfile", eCmdHdlrString, 0 },

+	{ "annotation_match", eCmdHdlrArray, 0 },

+	{ "de_dot", eCmdHdlrBinary, 0 },

+	{ "de_dot_separator", eCmdHdlrString, 0 },

+	{ "filenamerulebase", eCmdHdlrString, 0 },

+	{ "containerrulebase", eCmdHdlrString, 0 }

+#if HAVE_LOADSAMPLESFROMSTRING == 1

+	,

+	{ "filenamerules", eCmdHdlrArray, 0 },

+	{ "containerrules", eCmdHdlrArray, 0 }

+#endif

+};

+static struct cnfparamblk modpblk = {

+	CNFPARAMBLK_VERSION,

+	sizeof(modpdescr)/sizeof(struct cnfparamdescr),

+	modpdescr

+};

+

+/* action (instance) parameters (v6 config format) */

+static struct cnfparamdescr actpdescr[] = {

+	{ "kubernetesurl", eCmdHdlrString, 0 },

+	{ "srcmetadatapath", eCmdHdlrString, 0 },

+	{ "dstmetadatapath", eCmdHdlrString, 0 },

+	{ "tls.cacert", eCmdHdlrString, 0 },

+	{ "allowunsignedcerts", eCmdHdlrBinary, 0 },

+	{ "token", eCmdHdlrString, 0 },

+	{ "tokenfile", eCmdHdlrString, 0 },

+	{ "annotation_match", eCmdHdlrArray, 0 },

+	{ "de_dot", eCmdHdlrBinary, 0 },

+	{ "de_dot_separator", eCmdHdlrString, 0 },

+	{ "filenamerulebase", eCmdHdlrString, 0 },

+	{ "containerrulebase", eCmdHdlrString, 0 }

+#if HAVE_LOADSAMPLESFROMSTRING == 1

+	,

+	{ "filenamerules", eCmdHdlrArray, 0 },

+	{ "containerrules", eCmdHdlrArray, 0 }

+#endif

+};

+static struct cnfparamblk actpblk =

+	{ CNFPARAMBLK_VERSION,

+	  sizeof(actpdescr)/sizeof(struct cnfparamdescr),

+	  actpdescr

+	};

+

+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 void free_annotationmatch(annotation_match_t *match) {

+	if (match) {

+		for(int ii = 0 ; ii < match->nmemb; ++ii) {

+			if (match->patterns)

+				free(match->patterns[ii]);

+			if (match->regexps)

+				regexp.regfree(&match->regexps[ii]);

+		}

+		free(match->patterns);

+		match->patterns = NULL;

+		free(match->regexps);

+		match->regexps = NULL;

+		match->nmemb = 0;

+	}

+}

+

+static int init_annotationmatch(annotation_match_t *match, struct cnfarray *ar) {

+	DEFiRet;

+

+	match->nmemb = ar->nmemb;

+	CHKmalloc(match->patterns = calloc(sizeof(uchar*), match->nmemb));

+	CHKmalloc(match->regexps = calloc(sizeof(regex_t), match->nmemb));

+	for(int jj = 0; jj < ar->nmemb; ++jj) {

+		int rexret = 0;

+		match->patterns[jj] = (uchar*)es_str2cstr(ar->arr[jj], NULL);

+		rexret = regexp.regcomp(&match->regexps[jj],

+				(char *)match->patterns[jj], REG_EXTENDED|REG_NOSUB);

+		if (0 != rexret) {

+			char errMsg[512];

+			regexp.regerror(rexret, &match->regexps[jj], errMsg, sizeof(errMsg));

+			iRet = RS_RET_CONFIG_ERROR;

+			errmsg.LogError(0, iRet,

+					"error: could not compile annotation_match string [%s]"

+					" into an extended regexp - %d: %s\n",

+					match->patterns[jj], rexret, errMsg);

+			break;

+		}

+	}

+finalize_it:

+	if (iRet)

+		free_annotationmatch(match);

+	RETiRet;

+}

+

+static int copy_annotationmatch(annotation_match_t *src, annotation_match_t *dest) {

+	DEFiRet;

+

+	dest->nmemb = src->nmemb;

+	CHKmalloc(dest->patterns = malloc(sizeof(uchar*) * dest->nmemb));

+	CHKmalloc(dest->regexps = calloc(sizeof(regex_t), dest->nmemb));

+	for(int jj = 0 ; jj < src->nmemb ; ++jj) {

+		CHKmalloc(dest->patterns[jj] = (uchar*)strdup((char *)src->patterns[jj]));

+		/* assumes was already successfully compiled */

+		regexp.regcomp(&dest->regexps[jj], (char *)dest->patterns[jj], REG_EXTENDED|REG_NOSUB);

+	}

+finalize_it:

+    if (iRet)

+    	free_annotationmatch(dest);

+	RETiRet;

+}

+

+/* takes a hash of annotations and returns another json object hash containing only the

+ * keys that match - this logic is taken directly from fluent-plugin-kubernetes_metadata_filter

+ * except that we do not add the key multiple times to the object to be returned

+ */

+static struct json_object *match_annotations(annotation_match_t *match,

+		struct json_object *annotations) {

+	struct json_object *ret = NULL;

+

+	for (int jj = 0; jj < match->nmemb; ++jj) {

+		struct json_object_iterator it = json_object_iter_begin(annotations);

+		struct json_object_iterator itEnd = json_object_iter_end(annotations);

+		for (;!json_object_iter_equal(&it, &itEnd); json_object_iter_next(&it)) {

+			const char *const key = json_object_iter_peek_name(&it);

+			if (!ret || !fjson_object_object_get_ex(ret, key, NULL)) {

+				if (!regexp.regexec(&match->regexps[jj], key, 0, NULL, 0)) {

+					if (!ret) {

+						ret = json_object_new_object();

+					}

+					json_object_object_add(ret, key,

+						json_object_get(json_object_iter_peek_value(&it)));

+				}

+			}

+		}

+	}

+	return ret;

+}

+

+/* This will take a hash of labels or annotations and will de_dot the keys.

+ * It will return a brand new hash.  AFAICT, there is no safe way to

+ * iterate over the hash while modifying it in place.

+ */

+static struct json_object *de_dot_json_object(struct json_object *jobj,

+		const char *delim, size_t delim_len) {

+	struct json_object *ret = NULL;

+	struct json_object_iterator it = json_object_iter_begin(jobj);

+	struct json_object_iterator itEnd = json_object_iter_end(jobj);

+	es_str_t *new_es_key = NULL;

+	DEFiRet;

+

+	ret = json_object_new_object();

+	while (!json_object_iter_equal(&it, &itEnd)) {

+		const char *const key = json_object_iter_peek_name(&it);

+		const char *cc = strstr(key, ".");

+		if (NULL == cc) {

+			json_object_object_add(ret, key,

+					json_object_get(json_object_iter_peek_value(&it)));

+		} else {

+			char *new_key = NULL;

+			const char *prevcc = key;

+			new_es_key = es_newStrFromCStr(key, (es_size_t)(cc-prevcc));

+			while (cc) {

+				if (es_addBuf(&new_es_key, (char *)delim, (es_size_t)delim_len))

+					ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);

+				cc += 1; /* one past . */

+				prevcc = cc; /* beginning of next substring */

+				if ((cc = strstr(prevcc, ".")) || (cc = strchr(prevcc, '\0'))) {

+					if (es_addBuf(&new_es_key, (char *)prevcc, (es_size_t)(cc-prevcc)))

+						ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);

+					if (!*cc)

+						cc = NULL; /* EOS - done */

+				}

+			}

+			new_key = es_str2cstr(new_es_key, NULL);

+			es_deleteStr(new_es_key);

+			new_es_key = NULL;

+			json_object_object_add(ret, new_key,

+					json_object_get(json_object_iter_peek_value(&it)));

+			free(new_key);

+		}

+		json_object_iter_next(&it);

+	}

+finalize_it:

+	if (iRet != RS_RET_OK) {

+		json_object_put(ret);

+		ret = NULL;

+	}

+	if (new_es_key)

+		es_deleteStr(new_es_key);

+	return ret;

+}

+

+/* given a "metadata" object field, do

+ * - make sure "annotations" field has only the matching keys

+ * - de_dot the "labels" and "annotations" fields keys

+ * This modifies the jMetadata object in place

+ */

+static void parse_labels_annotations(struct json_object *jMetadata,

+		annotation_match_t *match, sbool de_dot,

+		const char *delim, size_t delim_len) {

+	struct json_object *jo = NULL;

+

+	if (fjson_object_object_get_ex(jMetadata, "annotations", &jo)) {

+		if ((jo = match_annotations(match, jo)))

+			json_object_object_add(jMetadata, "annotations", jo);

+		else

+			json_object_object_del(jMetadata, "annotations");

+	}

+	/* dedot labels and annotations */

+	if (de_dot) {

+		struct json_object *jo2 = NULL;

+		if (fjson_object_object_get_ex(jMetadata, "annotations", &jo)) {

+			if ((jo2 = de_dot_json_object(jo, delim, delim_len))) {

+				json_object_object_add(jMetadata, "annotations", jo2);

+			}

+		}

+		if (fjson_object_object_get_ex(jMetadata, "labels", &jo)) {

+			if ((jo2 = de_dot_json_object(jo, delim, delim_len))) {

+				json_object_object_add(jMetadata, "labels", jo2);

+			}

+		}

+	}

+}

+

+#if HAVE_LOADSAMPLESFROMSTRING == 1

+static int array_to_rules(struct cnfarray *ar, char **rules) {

+	DEFiRet;

+	es_str_t *tmpstr = NULL;

+	es_size_t size = 0;

+

+	if (rules == NULL)

+		FINALIZE;

+	*rules = NULL;

+	if (!ar->nmemb)

+		FINALIZE;

+	for (int jj = 0; jj < ar->nmemb; jj++)

+		size += es_strlen(ar->arr[jj]);

+	if (!size)

+		FINALIZE;

+	CHKmalloc(tmpstr = es_newStr(size));

+	CHKiRet((es_addStr(&tmpstr, ar->arr[0])));

+	CHKiRet((es_addBufConstcstr(&tmpstr, "\n")));

+	for(int jj=1; jj < ar->nmemb; ++jj) {

+		CHKiRet((es_addStr(&tmpstr, ar->arr[jj])));

+		CHKiRet((es_addBufConstcstr(&tmpstr, "\n")));

+	}

+	CHKiRet((es_addBufConstcstr(&tmpstr, "\0")));

+	CHKmalloc(*rules = es_str2cstr(tmpstr, NULL));

+finalize_it:

+	if (tmpstr) {

+		es_deleteStr(tmpstr);

+	}

+    if (iRet != RS_RET_OK) {

+    	free(*rules);

+    	*rules = NULL;

+    }

+	RETiRet;

+}

+#endif

+

+/* callback for liblognorm error messages */

+static void

+errCallBack(void __attribute__((unused)) *cookie, const char *msg,

+	    size_t __attribute__((unused)) lenMsg)

+{

+	errmsg.LogError(0, RS_RET_ERR_LIBLOGNORM, "liblognorm error: %s", msg);

+}

+

+static rsRetVal

+set_lnctx(ln_ctx *ctxln, char *instRules, uchar *instRulebase, char *modRules, uchar *modRulebase)

+{

+	DEFiRet;

+	if (ctxln == NULL)

+		FINALIZE;

+	CHKmalloc(*ctxln = ln_initCtx());

+	ln_setErrMsgCB(*ctxln, errCallBack, NULL);

+	if(instRules) {

+#if HAVE_LOADSAMPLESFROMSTRING == 1

+		if(ln_loadSamplesFromString(*ctxln, instRules) !=0) {

+			errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: normalization rules '%s' "

+					"could not be loaded", instRules);

+			ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD);

+		}

+#else

+		(void)instRules;

+#endif

+	} else if(instRulebase) {

+		if(ln_loadSamples(*ctxln, (char*) instRulebase) != 0) {

+			errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: normalization rulebase '%s' "

+					"could not be loaded", instRulebase);

+			ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD);

+		}

+	} else if(modRules) {

+#if HAVE_LOADSAMPLESFROMSTRING == 1

+		if(ln_loadSamplesFromString(*ctxln, modRules) !=0) {

+			errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: normalization rules '%s' "

+					"could not be loaded", modRules);

+			ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD);

+		}

+#else

+		(void)modRules;

+#endif

+	} else if(modRulebase) {

+		if(ln_loadSamples(*ctxln, (char*) modRulebase) != 0) {

+			errmsg.LogError(0, RS_RET_NO_RULEBASE, "error: normalization rulebase '%s' "

+					"could not be loaded", modRulebase);

+			ABORT_FINALIZE(RS_RET_ERR_LIBLOGNORM_SAMPDB_LOAD);

+		}

+	}

+finalize_it:

+	if (iRet != RS_RET_OK){

+		ln_exitCtx(*ctxln);

+		*ctxln = NULL;

+	}

+	RETiRet;

+}

+

+BEGINbeginCnfLoad

+CODESTARTbeginCnfLoad

+	loadModConf = pModConf;

+	pModConf->pConf = pConf;

+ENDbeginCnfLoad

+

+

+BEGINsetModCnf

+	struct cnfparamvals *pvals = NULL;

+	int i;

+	FILE *fp;

+	int ret;

+CODESTARTsetModCnf

+	pvals = nvlstGetParams(lst, &modpblk, NULL);

+	if(pvals == NULL) {

+		errmsg.LogError(0, RS_RET_MISSING_CNFPARAMS, "mmkubernetes: "

+			"error processing module config parameters [module(...)]");

+		ABORT_FINALIZE(RS_RET_MISSING_CNFPARAMS);

+	}

+

+	if(Debug) {

+		dbgprintf("module (global) param blk for mmkubernetes:\n");

+		cnfparamsPrint(&modpblk, pvals);

+	}

+

+	loadModConf->de_dot = DFLT_DE_DOT;

+	for(i = 0 ; i < modpblk.nParams ; ++i) {

+		if(!pvals[i].bUsed) {

+			continue;

+		} else if(!strcmp(modpblk.descr[i].name, "kubernetesurl")) {

+			free(loadModConf->kubernetesUrl);

+			loadModConf->kubernetesUrl = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+		} else if(!strcmp(modpblk.descr[i].name, "srcmetadatapath")) {

+			free(loadModConf->srcMetadataPath);

+			loadModConf->srcMetadataPath = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+			/* todo: sanitize the path */

+		} else if(!strcmp(modpblk.descr[i].name, "dstmetadatapath")) {

+			free(loadModConf->dstMetadataPath);

+			loadModConf->dstMetadataPath = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+			/* todo: sanitize the path */

+		} else if(!strcmp(modpblk.descr[i].name, "tls.cacert")) {

+			free(loadModConf->caCertFile);

+			loadModConf->caCertFile = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+			fp = fopen((const char*)loadModConf->caCertFile, "r");

+			if(fp == NULL) {

+				char errStr[1024];

+				rs_strerror_r(errno, errStr, sizeof(errStr));

+				iRet = RS_RET_NO_FILE_ACCESS;

+				errmsg.LogError(0, iRet,

+						"error: certificate file %s couldn't be accessed: %s\n",

+						loadModConf->caCertFile, errStr);

+				ABORT_FINALIZE(iRet);

+			} else {

+				fclose(fp);

+			}

+		} else if(!strcmp(modpblk.descr[i].name, "allowunsignedcerts")) {

+			loadModConf->allowUnsignedCerts = pvals[i].val.d.n;

+		} else if(!strcmp(modpblk.descr[i].name, "token")) {

+			free(loadModConf->token);

+			loadModConf->token = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+		} else if(!strcmp(modpblk.descr[i].name, "tokenfile")) {

+			free(loadModConf->tokenFile);

+			loadModConf->tokenFile = (uchar *) es_str2cstr(pvals[i].val.d.estr, NULL);

+			fp = fopen((const char*)loadModConf->tokenFile, "r");

+			if(fp == NULL) {

+				char errStr[1024];

+				rs_strerror_r(errno, errStr, sizeof(errStr));

+				iRet = RS_RET_NO_FILE_ACCESS;

+				errmsg.LogError(0, iRet,

+						"error: token file %s couldn't be accessed: %s\n",

+						loadModConf->tokenFile, errStr);

+				ABORT_FINALIZE(iRet);

+			} else {

+				fclose(fp);

+			}

+		} else if(!strcmp(modpblk.descr[i].name, "annotation_match")) {

+			free_annotationmatch(&loadModConf->annotation_match);

+			if ((ret = init_annotationmatch(&loadModConf->annotation_match, pvals[i].val.d.ar)))

+				ABORT_FINALIZE(ret);

+		} else if(!strcmp(modpblk.descr[i].name, "de_dot")) {

+			loadModConf->de_dot = pvals[i].val.d.n;