feat: update sqlite caching

package/etc/pylib/parser_source_cache.py

@@ -2,9 +2,7 @@
 import traceback
 import socket
 import struct
-from sqlitedict import SqliteDict
-
-import time
+import sqlite3
 
 try:
     import syslogng
@@ -48,13 +46,14 @@ def int_to_ip6(num):
         return int_to_ip6(addr)
 
 
-hostdict = str("/var/lib/syslog-ng/hostip")
+hostdict = "/var/lib/syslog-ng/hostip.sqlite"
 
 
 class psc_parse(LogParser):
     def init(self, options):
         self.logger = syslogng.Logger()
-        self.db = SqliteDict(f"{hostdict}.sqlite")
+        self.db = sqlite3.connect(hostdict)
+        self.cursor = self.db.cursor()
         return True
 
     def deinit(self):
@@ -65,9 +64,14 @@ def parse(self, log_message):
             ipaddr = log_message.get_as_str("SOURCEIP", "", repr="internal")
             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
+            self.cursor.execute("SELECT host FROM hosts WHERE ip_int=?", (ip_int,))
+            result = self.cursor.fetchone()
+            if result:
+                name = result[0]
+                self.logger.debug(f"psc.parse host={name}")
+                log_message["HOST"] = name
+            else:
+                self.logger.debug(f"No entry found for sourceip={ipaddr}")
 
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
@@ -82,7 +86,15 @@ class psc_dest(LogDestination):
     def init(self, options):
         self.logger = syslogng.Logger()
         try:
-            self.db = SqliteDict(f"{hostdict}.sqlite", autocommit=True)
+            self.db = sqlite3.connect(hostdict)
+            self.cursor = self.db.cursor()
+            # Create table if not exists
+            self.cursor.execute("""
+                CREATE TABLE IF NOT EXISTS hosts (
+                    ip_int INTEGER PRIMARY KEY,
+                    host TEXT
+                )
+            """)
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
             lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
@@ -99,15 +111,16 @@ def send(self, log_message):
         try:
             ipaddr = log_message.get_as_str("SOURCEIP", "", repr="internal")
             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"]
+            host = log_message["HOST"]
+            self.logger.debug(f'psc.send sourceip={ipaddr} int={ip_int} host={host}')
+            self.cursor.execute("SELECT host FROM hosts WHERE ip_int=?", (ip_int,))
+            result = self.cursor.fetchone()
+            if result:
+                current = result[0]
+                if current != host:
+                    self.cursor.execute("UPDATE hosts SET host=? WHERE ip_int=?", (host, ip_int))
             else:
-                self.db[ip_int] = log_message["HOST"]
+                self.cursor.execute("INSERT INTO hosts (ip_int, host) VALUES (?, ?)", (ip_int, host))
 
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
@@ -123,7 +136,14 @@ def flush(self):
 
 
 if __name__ == "__main__":
-    db = SqliteDict(f"{hostdict}.sqlite", autocommit=True)
-    db[0] = "seed"
+    db = sqlite3.connect(hostdict)
+    cursor = db.cursor()
+    cursor.execute("""
+        CREATE TABLE IF NOT EXISTS hosts (
+            ip_int INTEGER PRIMARY KEY,
+            host TEXT
+        )
+    """)
+    cursor.execute("INSERT OR REPLACE INTO hosts (ip_int, host) VALUES (?, ?)", (0, "seed"))
     db.commit()
-    db.close()
+    db.close()

package/etc/pylib/parser_vps_cache.py

@@ -1,10 +1,6 @@
 import sys
 import traceback
-import socket
-import struct
-from sqlitedict import SqliteDict
-
-import time
+import sqlite3
 
 try:
     import syslogng
@@ -18,13 +14,14 @@ class LogDestination:
         pass
 
 
-hostdict = str("/var/lib/syslog-ng/vps")
+hostdict = "/var/lib/syslog-ng/vps.sqlite"
 
 
 class vpsc_parse(LogParser):
     def init(self, options):
         self.logger = syslogng.Logger()
-        self.db = SqliteDict(f"{hostdict}.sqlite")
+        self.db = sqlite3.connect(hostdict)
+        self.cursor = self.db.cursor()
         return True
 
     def deinit(self):
@@ -34,10 +31,15 @@ def parse(self, log_message):
         try:
             host = log_message.get_as_str("HOST", "")
             self.logger.debug(f"vpsc.parse host={host}")
-            fields = self.db[host]
-            self.logger.debug(f"vpsc.parse host={host} fields={fields}")
-            for k, v in fields.items():
-                log_message[k] = v
+            self.cursor.execute("SELECT fields FROM hosts WHERE host=?", (host,))
+            result = self.cursor.fetchone()
+            if result:
+                fields = eval(result[0])
+                self.logger.debug(f"vpsc.parse host={host} fields={fields}")
+                for k, v in fields.items():
+                    log_message[k] = v
+            else:
+                self.logger.debug(f"No fields found for host={host}")
 
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
@@ -52,7 +54,15 @@ class vpsc_dest(LogDestination):
     def init(self, options):
         self.logger = syslogng.Logger()
         try:
-            self.db = SqliteDict(f"{hostdict}.sqlite", autocommit=True)
+            self.db = sqlite3.connect(hostdict)
+            self.cursor = self.db.cursor()
+            # Create table if not exists
+            self.cursor.execute("""
+                CREATE TABLE IF NOT EXISTS hosts (
+                    host TEXT PRIMARY KEY,
+                    fields TEXT
+                )
+            """)
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
             lines = traceback.format_exception(exc_type, exc_value, exc_traceback)
@@ -69,20 +79,19 @@ def send(self, log_message):
         try:
             host = log_message.get_as_str("HOST", "")
             fields = {}
-            fields[".netsource.sc4s_vendor"] = log_message.get_as_str(
-                "fields.sc4s_vendor"
-            )
-            fields[".netsource.sc4s_product"] = log_message.get_as_str(
-                "fields.sc4s_product"
-            )
+            fields[".netsource.sc4s_vendor"] = log_message.get_as_str("fields.sc4s_vendor")
+            fields[".netsource.sc4s_product"] = log_message.get_as_str("fields.sc4s_product")
 
             self.logger.debug(f"vpsc.send host={host} fields={fields}")
-            if host in self.db:
-                current = self.db[host]
+            self.cursor.execute("SELECT fields FROM hosts WHERE host=?", (host,))
+            result = self.cursor.fetchone()
+            if result:
+                current = eval(result[0])
                 if current != fields:
-                    self.db[host] = fields
+                    self.cursor.execute("UPDATE hosts SET fields=? WHERE host=?", (str(fields), host))
             else:
-                self.db[host] = fields
+                self.cursor.execute("INSERT INTO hosts (host, fields) VALUES (?, ?)", (host, str(fields)))
+            self.db.commit()
 
         except Exception:
             exc_type, exc_value, exc_traceback = sys.exc_info()
@@ -98,4 +107,4 @@ def flush(self):
 
 
 if __name__ == "__main__":
-    pass
+    pass

package/etc/pylib/psc_dump.py

@@ -1,13 +1,29 @@
+import sqlite3
+import os
 
-import sys
-import traceback
-import socket
-import struct
-from sqlitedict import SqliteDict
+db_path = "/var/lib/syslog-ng/hostip.sqlite"
 
+if not os.path.exists(db_path):
+    raise FileNotFoundError(f"Database file not found at {db_path}")
 
-hostdict = str("/var/lib/syslog-ng/cache/hostip")
-db = SqliteDict(f"{hostdict}.sqlite")
+try:
+    with sqlite3.connect(db_path) as db:
+        cursor = db.cursor()
 
-for k,v in db.items():
-    print(f"key={k}={v}")
+        # Ensure the table exists
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS hosts (
+                ip_int INTEGER PRIMARY KEY,
+                host TEXT
+            )
+        """)
+
+        cursor.execute("SELECT ip_int, host FROM hosts")
+        rows = cursor.fetchall()
+        for k, v in rows:
+            print(f"key={k}={v}")
+
+except sqlite3.Error as e:
+    print(f"Database error: {e}")
+except Exception as e:
+    print(f"Unexpected error: {e}")

package/etc/pylib/vps_dump.py

@@ -0,0 +1,29 @@
+import sqlite3
+import os
+
+db_path = "/var/lib/syslog-ng/vps.sqlite"
+
+if not os.path.exists(db_path):
+    raise FileNotFoundError(f"Database file not found at {db_path}")
+
+try:
+    with sqlite3.connect(db_path) as db:
+        cursor = db.cursor()
+
+        # Ensure the table exists
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS hosts (
+                host TEXT PRIMARY KEY,
+                fields TEXT
+            )
+        """)
+
+        cursor.execute("SELECT host, fields FROM hosts")
+        rows = cursor.fetchall()
+        for k, v in rows:
+            print(f"key={k}={v}")
+
+except sqlite3.Error as e:
+    print(f"Database error: {e}")
+except Exception as e:
+    print(f"Unexpected error: {e}")

pyproject.toml

@@ -8,7 +8,6 @@ license = "Apache-2.0"
 [tool.poetry.dependencies]
 python = "^3.9"
 Jinja2 = "^3.1.3"
-sqlitedict = "^2.0.0"
 requests = "^2.28.1"
 shortuuid = "^1.0.11"
 pyyaml = "6.0.2"