Kubernetes metadata enrichment

dev-requirements.txt

@@ -8,4 +8,5 @@ astroid==1.6.1
 logilab-common<=0.63.0
 psutil
 pytest
+pytest-mock
 pre-commit

modules/python-modules/Makefile.am

@@ -47,6 +47,8 @@ EXTRA_DIST += \
 	\
 	modules/python-modules/syslogng/modules/example/scl/example.conf	\
 	modules/python-modules/syslogng/modules/example/__init__.py		\
+	modules/python-modules/syslogng/modules/kubernetes/scl/kubernetes.conf	\
+	modules/python-modules/syslogng/modules/kubernetes/__init__.py		\
 	modules/python-modules/setup.py						\
 	modules/python-modules/README.md
 

modules/python-modules/setup.py

@@ -35,4 +35,8 @@
         "syslogng",
         "syslogng.debuggercli",
         "syslogng.modules.example",
+        "syslogng.modules.kubernetes",
+      ],
+      install_requires=[
+          "kubernetes"
       ])

modules/python-modules/syslogng/message.py

@@ -28,6 +28,26 @@
     warnings.warn("You have imported the syslogng package outside of syslog-ng, thus some of the functionality is not available. Defining fake classes for those exported by the underlying syslog-ng code")
 
     class LogMessage(dict):
+
+        # syslog-ng internally stores everything as a sequence of bytes that
+        # is strongly "preferred" to be an utf8 string.  It takes a bytes
+        # and stores it verbatim.  Strings are converted to utf8 (not the
+        # local character set!)
+
         def __init__(self, msg):
-            super().__init__()
-            self['MESSAGE'] = msg
+            super().__init__([('MESSAGE', msg)])
+
+        def __getitem__(self, key):
+            if isinstance(key, str):
+                key = key.encode('utf8')
+            return super().__getitem__(key)
+
+        def __setitem__(self, key, value):
+            if isinstance(key, str):
+                key = key.encode('utf8')
+            if isinstance(value, str):
+                value = value.encode('utf8')
+            super().__setitem__(key, value)
+
+        def __eq__(self):
+            return False

modules/python-modules/syslogng/modules/kubernetes/__init__.py

@@ -0,0 +1,93 @@
+#############################################################################
+# Copyright (c) 2022 Attila Szakacs <szakacs.attila96@gmail.com>
+#
+# This program is free software; you can redistribute it and/or modify it
+# under the terms of the GNU General Public License version 2 as published
+# by the Free Software Foundation, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+#
+# As an additional exemption you are allowed to compile & link against the
+# OpenSSL libraries as published by the OpenSSL project. See the file
+# COPYING for details.
+#
+#############################################################################
+
+from syslogng import Logger, LogParser
+import kubernetes
+import logging
+
+
+class KubernetesAPIEnrichment(LogParser):
+    logger = logging.getLogger(__name__)
+
+    def init(self, options):
+        self.__metadata = {}
+        self.__prefix = options["prefix"]
+
+        kubernetes.config.load_config()
+        self.__client_api = kubernetes.client.CoreV1Api()
+
+        return True
+
+    def add_prefix(self, string):
+        return self.__prefix + string
+
+    def __get_cached_pod_metadata(self, namespace_name, pod_name):
+        self.logger.debug("Trying to find cached metadata for pod {}".format(pod_name))
+        return self.__metadata[namespace_name][pod_name]
+
+    def __query_pod_metadata_from_kubernetes_api_server(self, namespace_name, pod_name):
+        self.logger.debug("Trying to query metadata for pod {}/{} from the Kubernetes API".format(namespace_name, pod_name))
+
+        cached_namespace = self.__metadata.setdefault(namespace_name, {})
+        cached_pod_metadata = cached_namespace.setdefault(pod_name, {})
+
+        pod = self.__client_api.list_namespaced_pod(namespace_name, field_selector="metadata.name=={}".format(pod_name)).items[0]
+        try:
+            cached_pod_metadata[self.add_prefix("pod_uuid")] = pod.metadata.uid
+        except (AttributeError, IndexError, TypeError):
+            self.logger.warning("Error querying pod.metadata.uid for pod {}/{} from the Kubernetes API".format(namespace_name, pod_name))
+
+        for name, value in (pod.metadata.labels or {}).items():
+            cached_pod_metadata[self.add_prefix("labels.{}".format(name))] = value
+
+        for name, value in (pod.metadata.annotations or {}).items():
+            cached_pod_metadata[self.add_prefix("annotations.{}".format(name))] = value
+
+        try:
+            container_status = pod.status.container_statuses[0]
+            cached_pod_metadata[self.add_prefix("container_name")] = container_status.name
+            cached_pod_metadata[self.add_prefix("container_image")] = container_status.image
+            cached_pod_metadata[self.add_prefix("container_hash")] = container_status.image_id.replace("docker-pullable://", "", 1)
+            cached_pod_metadata[self.add_prefix("docker_id")] = container_status.container_id.replace("docker://", "", 1)
+        except (AttributeError, IndexError, TypeError):
+            self.logger.warning("Error querying container_status for pod {}/{} from the Kubernetes API".format(namespace_name, pod_name))
+        return cached_pod_metadata
+
+    def get_pod_metadata(self, namespace_name, pod_name):
+        try:
+            return self.__get_cached_pod_metadata(namespace_name, pod_name)
+        except KeyError:
+            return self.__query_pod_metadata_from_kubernetes_api_server(namespace_name, pod_name)
+
+    def parse(self, msg):
+        namespace_name = msg[self.add_prefix("namespace_name")].decode()
+        pod_name = msg[self.add_prefix("pod_name")].decode()
+
+        pod_metadata = self.get_pod_metadata(namespace_name, pod_name)
+        for name, value in pod_metadata.items():
+            try:
+                if not msg[name]:
+                    msg[name] = value
+            except KeyError:
+                msg[name] = value
+
+        return True

scl/kubernetes/kubernetes.conf → modules/python-modules/syslogng/modules/kubernetes/scl/kubernetes.conf

@@ -78,6 +78,12 @@ block parser kubernetes-metadata-parser (prefix() template("$FILE_NAME")) {
                 template(`template`)
                 prefix(`prefix`)
             );
+            python(
+                class("syslogng.modules.kubernetes.KubernetesAPIEnrichment")
+                options(
+                    "prefix" => "`prefix`"
+                )
+            );
         };
 
         # make the container-id of 12 characters long as usual in cli