reporter: Serialize writes to the database

sqlite3 can obviously have only one writer at a time. Some operations
like cleaning up the database can take a couple of seconds during which
other write operations might time out and will be dropped. Therefore we
are introducing a lock so that only one operation at a time is trying to
write to the database. This shouldn't cause any problems during normal
operation, but will block any writers when the database is being cleaned
up or optimised.

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>

diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index e1c25b48b49f7c6818aadfc8ad80439f5c35230b..3afcedf3d96acc4b585f5021e64b5bcbc065080d 100644 (file)
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -261,6 +261,12 @@ class Worker(threading.Thread):
                """
                return self.reporter.config
 
+       # Global write lock for the database
+       # Since only one members can write to the database at the same time, we will
+       # have to serialize those writes in order to avoid any OperationalErrors from
+       # the sqlite3 driver, if an operation is waiting too long for a lock.
+       _write = threading.Lock()
+
        def _open_database(self):
                """
                        Opens the database
@@ -385,9 +391,10 @@ class Worker(threading.Thread):
                                t = (now - retention_days).timestamp()
 
                                # Remove everything
-                               self.db.execute(
-                                       "DELETE FROM alerts WHERE timestamp <= ?", (t,),
-                               )
+                               with self._write:
+                                       self.db.execute(
+                                               "DELETE FROM alerts WHERE timestamp <= ?", (t,),
+                                       )
 
                        # Release the lock
                        finally:
@@ -399,19 +406,9 @@ class Worker(threading.Thread):
                """
                log.debug("Optimizing the database")
 
-               for i in range(5):
+               with self._write:
                        self.db.execute("PRAGMA optimize")
-
-                       try:
-                               self.db.execute("PRAGMA wal_checkpoint = TRUNCATE")
-
-                       # Try again if this has failed (probably because of the database being locked)
-                       except sqlite3.OperationalError:
-                               continue
-
-                       # Terminate the loop if the operation was successful
-                       else:
-                               break
+                       self.db.execute("PRAGMA wal_checkpoint = TRUNCATE")
 
        def process(self, event):
                """
@@ -432,9 +429,10 @@ class Worker(threading.Thread):
                log.debug("Received alert: %s" % event)
 
                # Write the event to the database
-               self.db.execute("INSERT INTO alerts(timestamp, event) VALUES(?, ?)",
-                       (event.timestamp.timestamp(), event.json))
-               self.db.commit()
+               with self._write:
+                       self.db.execute("INSERT INTO alerts(timestamp, event) VALUES(?, ?)",
+                               (event.timestamp.timestamp(), event.json))
+                       self.db.commit()
 
                # Send to syslog
                if self.config.getboolean("syslog", "enabled", fallback=False):