feat: Use last valid host name when next event is incomplete

New feature flag. Enables a kv store feature to cache the last host name from a SOURCEIP and use that value if a following event has null, nill or ip rather than host name. Local storage uses sqllite

.github/workflows/ci-main.yaml

@@ -108,6 +108,7 @@ jobs:
         with:
           context: .
           file: package/Dockerfile
+          #platforms: linux/amd64,linux/arm64
           platforms: linux/amd64,linux/arm64
           push: true
           #tags: ${{ needs.meta.outputs.container_tags }}

docs/experiments.md

@@ -0,0 +1,7 @@
+# Current Experimental Features
+
+# > 2.13.0
+
+* In env_file set `SC4S_USE_NAME_CACHE=yes` to enable caching last valid host string and replacing nill, null, or ipv4 with last good value. 
+    - Benefit: More correct host name values in Splunk when source vendor fails to provide valid syslog message
+    - Risk: Potential disk I/O usage (space, iops) Potential reduction in throughput when a high proportion of events are incomplete.

mkdocs.yml

@@ -45,5 +45,6 @@ nav:
   - Troubleshooting:
       - SC4S Startup and Validation: "troubleshooting/troubleshoot_SC4S_server.md"
       - SC4S Logging and Troubleshooting Resources: "troubleshooting/troubleshoot_resources.md"
+  - Experiments: "experiments.md"
   - "Upgrading SC4S": "upgrade.md"
   - "SC4S FAQ": "faq.md"

package/etc/conf.d/conflib/_common/syslog_format.conf

@@ -59,8 +59,12 @@ filter f_msg_is_tcp_json{
     match("tcp_json" value("fields.sc4s_syslog_format"))
 };
 filter f_host_is_nil_or_ip{
-    host('^-') or
-    host('^((((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))|((([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))))$')
+    (
+        host('^-') or
+        host('^((((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))|((([0-9a-fA-F]{1,4}:){7,7}[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,7}:|([0-9a-fA-F]{1,4}:){1,6}:[0-9a-fA-F]{1,4}|([0-9a-fA-F]{1,4}:){1,5}(:[0-9a-fA-F]{1,4}){1,2}|([0-9a-fA-F]{1,4}:){1,4}(:[0-9a-fA-F]{1,4}){1,3}|([0-9a-fA-F]{1,4}:){1,3}(:[0-9a-fA-F]{1,4}){1,4}|([0-9a-fA-F]{1,4}:){1,2}(:[0-9a-fA-F]{1,4}){1,5}|[0-9a-fA-F]{1,4}:((:[0-9a-fA-F]{1,4}){1,6})|:((:[0-9a-fA-F]{1,4}){1,7}|:)|fe80:(:[0-9a-fA-F]{0,4}){0,4}%[0-9a-zA-Z]{1,}|::(ffff(:0{1,4}){0,1}:){0,1}((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])|([0-9a-fA-F]{1,4}:){1,4}:((25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9])\.){3,3}(25[0-5]|(2[0-4]|1{0,1}[0-9]){0,1}[0-9]))))$')
+    )
+    and not '$HOST' eq "127.0.0.1"
+
 };
 
 filter f_host_is_ip{

package/etc/conf.d/conflib/_splunk/fix_dns.conf

@@ -4,6 +4,12 @@ parser p_fix_host_resolver {
     );
 };
 
+parser p_host_cache {
+    python(
+        class("parser_source_cache.psc_parse")
+    );
+};
+
 parser p_add_context_host {
     add-contextual-data(
         selector("${SOURCEIP}"),

package/etc/conf.d/destinations/dest_psc/plugin.conf

@@ -0,0 +1,7 @@
+destination d_psc {
+    python(
+        class("parser_source_cache.psc_dest")
+        batch-timeout(3000)
+        batch-lines(100)
+    );
+};

package/etc/conf.d/log_paths/2/lp-dest-psc.conf

@@ -0,0 +1,8 @@
+log {
+    if {
+        filter(f_host_is_nil_or_ip);
+    } else {
+        destination(d_psc);
+    };
+    flags(catchall);
+};

package/etc/conf.d/sources/source_syslog/plugin.jinja

@@ -166,12 +166,19 @@ source s_{{ port_id }} {
                 filter(f_host_is_nil_or_ip);
                 if {
                     parser(p_add_context_host);
+                };
+                {%- if use_namecache == True %}
+                if {
+                    filter(f_host_is_nil_or_ip);
+                    parser(p_host_cache);
+                };
+                {%- endif %}
                 {%- if use_reverse_dns == True %}
-                elif {
+                if {
                     filter(f_host_is_nil_or_ip);
                     parser(p_fix_host_resolver);
-                {%- endif %}
                 };
+                {%- endif %}
             };              
             rewrite {
                 set('$(lowercase "$HOST")' value(HOST));
@@ -327,12 +334,19 @@ source s_{{ port_id }} {
             filter(f_host_is_nil_or_ip);
             if {
                 parser(p_add_context_host);
+            };
+            {%- if use_namecache == True %}
+            if {
+                filter(f_host_is_nil_or_ip);
+                parser(p_host_cache);
+            };
+            {%- endif %}
             {%- if use_reverse_dns == True %}
-            elif {
+            if {
                 filter(f_host_is_nil_or_ip);
                 parser(p_fix_host_resolver);
-            {%- endif %}
             };
+            {%- endif %}
         };
 
         rewrite {

package/etc/conf.d/sources/source_syslog/plugin.py

@@ -43,6 +43,17 @@
 else:
     use_reverse_dns = False
 
+if os.getenv(f"SC4S_USE_NAME_CACHE", "no").lower() in [
+    "true",
+    "1",
+    "t",
+    "y",
+    "yes",
+]:
+    use_namecache = True
+else:
+    use_namecache = False
+
 if os.getenv(f"SC4S_SOURCE_TLS_ENABLE", "no").lower() in [
     "true",
     "1",
@@ -68,6 +79,7 @@
         store_raw_message=store_raw_message,
         port_id=port_id,
         use_reverse_dns=use_reverse_dns,
+        use_namecache=use_namecache,
         use_tls=use_tls,
         tls_dir=os.getenv(f"SC4S_TLS", "/etc/syslog-ng/tls"),
         cert_file=cert_file,

package/etc/pylib/parser_fix_dns.py

@@ -26,7 +26,7 @@ def parse(self, log_message):
             if len(parts) > 1:
                 log_message["HOST"] = name
         except:
-            pass
+            return False
 
         # return True, other way message is dropped
         return True

package/etc/pylib/parser_source_cache.py

@@ -0,0 +1,93 @@
+
+import sys
+import traceback
+import socket
+import struct
+from sqlitedict import SqliteDict
+
+import time
+try:
+    import syslogng
+except:
+    pass
+
+
+def ip2int(addr):
+    return struct.unpack("!I", socket.inet_aton(addr))[0]
+
+def int2ip(addr):
+    return socket.inet_ntoa(struct.pack("!I", addr))
+
+hostdict = str("/var/lib/syslog-ng/hostip")
+
+class psc_parse(object):
+    def init(self, options):
+        self.logger = syslogng.Logger()
+        self.db = SqliteDict(f"{hostdict}.sqlite")            
+        return True
+
+    def deinit(self):
+        self.db.close()
+
+    def parse(self, log_message):
+        try:
+            ipaddr = log_message["SOURCEIP"].decode("utf-8")
+            ip_int = ip2int(ipaddr)
+            self.logger.debug(f'psc.parse sourceip={ipaddr} int={ip_int}')
+            name = self.db[ip_int]
+            self.logger.debug(f'psc.parse host={name}')
+            log_message["HOST"]=name
+
+        except:
+            exc_type, exc_value, exc_traceback = sys.exc_info()
+            lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
+            self.logger.debug(''.join('!! ' + line for line in lines))
+            return False
+        self.logger.debug(f'psc.parse complete')
+        return True
+
+class psc_dest(object):
+    def init(self, options):
+        self.logger = syslogng.Logger()
+        try:
+            self.db = SqliteDict(f"{hostdict}.sqlite",autocommit=True)            
+        except:
+            exc_type, exc_value, exc_traceback = sys.exc_info()
+            lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
+            self.logger.debug(''.join('!! ' + line for line in lines))
+            return False
+        return True
+
+    def deinit(self):
+        """Close the connection to the target service"""
+        self.db.commit()
+        self.db.close()
+
+    def send(self, log_message):
+        try:
+            ipaddr = log_message["SOURCEIP"].decode("utf-8")
+            ip_int = ip2int(ipaddr)
+            self.logger.debug(f'psc.send sourceip={ipaddr} int={ip_int} host={log_message["HOST"]}')
+            if ip_int in self.db:
+                current = self.db[ip_int]
+                if current != log_message["HOST"]:
+                    self.db[ip_int] =log_message["HOST"]    
+            else:
+                self.db[ip_int] =log_message["HOST"]
+
+        except:
+            exc_type, exc_value, exc_traceback = sys.exc_info()
+            lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
+            self.logger.debug(''.join('!! ' + line for line in lines))
+            return False
+        self.logger.debug('psc.send complete')
+        return True
+
+    def flush(self):
+        self.db.commit()
+
+if __name__ == "__main__":
+    db = SqliteDict(f"{hostdict}.sqlite",autocommit=True)
+    db[0]="seed"
+    db.commit()
+    db.close()

package/etc/pylib/psc_dump.py

@@ -0,0 +1,13 @@
+
+import sys
+import traceback
+import socket
+import struct
+from sqlitedict import SqliteDict
+
+
+hostdict = str("/var/lib/syslog-ng/cache/hostip")
+db = SqliteDict(f"{hostdict}.sqlite")            
+
+for k,v in db.items():
+    print(f"key={k}={v}")

package/sbin/entrypoint.sh

@@ -3,6 +3,8 @@ function join_by { local d=$1; shift; local f=$1; shift; printf %s "$f" "${@/#/$
 
 export PYTHONPATH=/etc/syslog-ng/pylib
 
+python3 /etc/syslog-ng/pylib/parser_source_cache.py
+
 export SC4S_LISTEN_STATUS_PORT=${SC4S_LISTEN_STATUS_PORT:=8080}
 
 # These path variables allow for a single entrypoint script to be utilized for both Container and BYOE runtimes

pyproject.toml

@@ -8,6 +8,8 @@ license = "Apache-2.0"
 [tool.poetry.dependencies]
 python = "^3.9"
 Jinja2 = "^3.0.3"
+sqlitedict = "^2.0.0"
+
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.0.1"

tests/docker-compose.yml

@@ -72,6 +72,7 @@ services:
       - SC4S_LISTEN_PROOFPOINT_PPS_TLS_PORT=7000
       #- SC4S_DEST_SPLUNK_HEC_GLOBAL=no
       #- SC4S_DEST_CEF_SPLUNK_HEC=yes
+      - SC4S_USE_NAME_CACHE=yes
 
   splunk:
     image: docker.io/splunk/splunk:latest