[PATCH 0/5] Add Zabbix functionality to suricata-reporter

[PATCH 0/5] Add Zabbix functionality to suricata-reporter

[Robin Roevens]

Hi all,

As discussed here earlier, I've worked on implementing sending
Suricata alerts straight to Zabbix from within suricata-reporter instead
of trying to parse the suricata logging separately using the Zabbix agent.

For this I use the zabbix-utils python library, which I submited here
also as a separate pak (but meanwhile already requires an update, which
I will post soon). This set of patches makes suricata-reporter able to
directly communicate to a Zabbix server without having the zabbix_agentd
pak installed, sending suricata alerts in real-time.

As Zabbix supports sending items in bulk, I have opted to create an
async background task that will send all events from last 1 second in
bulk so that even in the case that there are hundreds of incoming
alerts, Zabbix server is only contacted once per second.

When for some reason sending to Zabbix server fails, it will be retried
3 times and then the background task will be suspended until a new
suricata event comes in. That will wake the task again and retry to send all
pending events. In environments with many events, that may actually not
have that much of an effect. But in the average environment, this will
give the Zabbix Server some breathing space as it failing to receive our
events, may indicate a Zabbix server overload. 

For this I have to keep track which events are sent and which are
pending. So I added a column in the database that keeps track of that.

I have also added an alert_max_age config parameter that allows the user
to set how long suricata-reporter should retry to send events to Zabbix.
Events older than that set age, will no longer be sent to Zabbix.
This also give the user the implicit option to send older events when
only just enabling the zabbix sending functionality, since the DB column
exists and no event was ever sent to Zabbix, all events will be
'pending". At first run with zabbix functionality enabled, all events up
to alert_max_age that are in the database will be sent to zabbix
immediatly.

All events sent to Zabbix contain the timestamp of retrieval by
suricata-reporter, so Zabbix will register and order them as received on that
timestamp independently of the actual time Zabbix itself received the
event.

This is my first adventure in Python async programming, so I hope I did
not make any flagrant mistakes. But the code has been running here for
weeks now without any problem. I have not actually tested large bursts
of events, as I could not simulate that.. But I did make Zabbix server
slow, unavailable and finally replaced it with netcat (to accept the connection, but
not react on it) and I had the connection with the server off for a few
hours to then re-establish the connection to see hundereds of pending events 
being registered in only a few milliseconds.
I did not notice any problems with suricata-reporter in any of these
cases.

Regards

Robin

-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

[PATCH 1/5] Initialize async zabbix sender from zabbix_utils

[Robin Roevens]

When Zabbix is enabled in new config section [zabbix], and the zabbix_utils 
python module is available, a zabbix AsyncSender object will be initialized 
for sending alerts to Zabbix using parameters from the new config section.

Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
---
 src/reporter.conf.in     | 23 +++++++++++++++++++++++
 src/suricata-reporter.in | 37 +++++++++++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/src/reporter.conf.in b/src/reporter.conf.in
index 5943006..bab01b6 100644
--- a/src/reporter.conf.in
+++ b/src/reporter.conf.in
@@ -45,3 +45,26 @@
 ; 3 = Low Severity
 ; 4 = Informational
 ;severity = 3
+
+[zabbix]
+; Enable sending alerts to Zabbix
+;enabled = false
+
+; Path to the Zabbix agent configuration file
+;zabbix_agentd_config = /etc/zabbix_agentd/zabbix_agentd.conf
+
+; Zabbix server ip or hostname (required if zabbix_agentd_config is not set)
+;zabbix_server_host = 127.0.0.1
+
+; Zabbix server port (defaults to 10051 if not set)
+;zabbix_server_port = 10051
+
+; Hostname as defined in Zabbix server to send alerts to (defaults to either the
+; Hostname directive in Zabbix Agent config or system hostname)
+;alert_item_hostname = IPFire
+
+; Zabbix item key to send alerts to
+;alert_item_key = ipfire.suricata.event.get
+
+; Max age (seconds) to retry sending alerts to Zabbix
+;alert_max_age = 3600
\ No newline at end of file
diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index 28b55bc..f9da7b4 100644
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -37,6 +37,13 @@ import socket
 import sqlite3
 import sys
 
+# Load zabbix_utils module if available 
+zabbix_utils_available = True
+try:
+    from zabbix_utils import AsyncSender, ItemValue
+except ImportError:
+    zabbix_utils_available = False
+
 # Fetch the hostname
 HOSTNAME = socket.gethostname()
 
@@ -75,6 +82,10 @@ class Reporter(object):
 		# Remember the last time the database was cleaned
 		self.last_cleanup_at = None
 
+		# Initialize Zabbix sender
+		self.zabbix_sender = None
+		self.init_zabbix_sender()
+
 		# Register any signals
 		for signo in (signal.SIGINT, signal.SIGTERM):
 			self.loop.add_signal_handler(signo, self.terminate)
@@ -97,6 +108,32 @@ class Reporter(object):
 
 		return config
 
+	def init_zabbix_sender(self):
+		"""
+			Initialize the Zabbix async sender if configured
+		"""
+		if not self.config.getboolean('zabbix', 'enabled', fallback=False):
+			return
+
+		if not zabbix_utils_available:
+			log.error("zabbix-utils is not installed. Zabbix alerts will not be sent.")
+			return
+
+		zabbix_config = self.config.get('zabbix', 'zabbix_agentd_config', fallback='')
+		zabbix_server_host = self.config.get('zabbix', 'zabbix_server_host', fallback='')
+		zabbix_server_port = self.config.getint('zabbix', 'zabbix_server_port', fallback=10051)
+
+		if zabbix_config:
+			if not os.path.isfile(zabbix_config):
+				log.error(f"Zabbix agent config file {zabbix_config} does not exist.")
+				return
+			self.zabbix_sender = AsyncSender(use_config=True, config_path=zabbix_config)
+		else:
+			if not zabbix_server_host:
+				log.error("zabbix_server_host must be specified when zabbix_agentd_config is not provided.")
+				return
+			self.zabbix_sender = AsyncSender(server=zabbix_server_host, port=zabbix_server_port)
+
 	def _open_database(self):
 		"""
 			Opens the database
-- 
2.54.0


-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

[PATCH 2/5] Add database column zabbix_pending in alerts table.

[Robin Roevens]

To prevent possible hammering of the Zabbix server in busy environments,
instead of sending alerts immediatly when they arrive, we will send them in bulk
every 1 second. For that we need to know what alerts where not yet sent to 
Zabbix. This will be tracked in this new column zabbix_pending in the alerts
table.

Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
---
 src/suricata-reporter.in | 23 +++++++++++++++++++++++
 1 file changed, 23 insertions(+)

diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index f9da7b4..83e4e97 100644
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -134,6 +134,27 @@ class Reporter(object):
 				return
 			self.zabbix_sender = AsyncSender(server=zabbix_server_host, port=zabbix_server_port)
 
+	def _ensure_zabbix_pending_column(self, db):
+		"""
+			Ensures the database schema always has the zabbix_pending column and index.
+			If they are missing, the database is migrated to add them.
+		"""
+		cursor = db.execute("PRAGMA table_info(alerts)")
+		columns = {row[1] for row in cursor.fetchall()}
+
+		if "zabbix_pending" not in columns:
+			db.execute("ALTER TABLE alerts ADD COLUMN zabbix_pending INTEGER NOT NULL DEFAULT 0")
+			db.commit()
+			log.debug("Database: Added zabbix_pending column to alerts table.")
+
+		cursor = db.execute("PRAGMA index_list(alerts)")
+		indexes = {row[1] for row in cursor.fetchall()}
+
+		if "alerts_zabbix_pending" not in indexes:
+			db.execute("CREATE INDEX IF NOT EXISTS alerts_zabbix_pending ON alerts(zabbix_pending)")
+			db.commit()
+			log.debug("Database: Added alerts_zabbix_pending index on alerts table column zabbix_pending.")
+
 	def _open_database(self):
 		"""
 			Opens the database
@@ -165,6 +186,8 @@ class Reporter(object):
 			CREATE INDEX IF NOT EXISTS alerts_timestamp ON alerts(timestamp);
 		""")
 
+		self._ensure_zabbix_pending_column(db)
+
 		return db
 
 	@property
-- 
2.54.0


-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

[PATCH 3/5] Set zabbix_pending flag when storing new event in DB

[Robin Roevens]

When sending to zabbix is enabled, we need to set the zabbix_pending flag set 
on each new event written to the database.

Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
---
 src/suricata-reporter.in | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index 83e4e97..78bb04d 100644
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -314,9 +314,14 @@ class Reporter(object):
 		"""
 			Writes a single event to the database
 		"""
+		# Determine whether this event should be marked for Zabbix delivery
+		zabbix_pending = 1 if self.config.getboolean('zabbix', 'enabled', fallback=False) else 0
+
 		# Write the event to the database
-		self.db.execute("INSERT INTO alerts(timestamp, event) VALUES(?, ?)",
-			(event.timestamp.timestamp(), event.json))
+		self.db.execute(
+			"INSERT INTO alerts(timestamp, event, zabbix_pending) VALUES(?, ?, ?)",
+			(event.timestamp.timestamp(), event.json, zabbix_pending)
+		)
 
 		# Commit it straight away
 		self.db.commit()
-- 
2.54.0


-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

[PATCH 4/5] Add function to send all pending alerts to Zabbix

[Robin Roevens]

Add abbility to send all alerts from DB marked as pending to Zabbix server in
bulk with errorhandling

Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
---
 src/suricata-reporter.in | 94 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 94 insertions(+)

diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index 78bb04d..47482ab 100644
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -326,6 +326,100 @@ class Reporter(object):
 		# Commit it straight away
 		self.db.commit()
 
+	async def flush_pending_to_zabbix(self):
+		"""
+			Sends all alerts that were marked for Zabbix in a single bulk call.
+		"""
+		if not self.zabbix_sender:
+			return False
+
+		alert_item_hostname = self.config.get('zabbix', 'alert_item_hostname', fallback=(self.zabbix_sender.host or HOSTNAME))
+		alert_item_key = self.config.get('zabbix', 'alert_item_key', fallback='ipfire.suricata.event.get')
+
+		now = datetime.datetime.now()
+		alert_max_age = datetime.timedelta(
+			seconds = self.config.getint('zabbix', 'alert_max_age', fallback=3600)
+		)
+		min_timestamp = (now - alert_max_age).timestamp()
+
+		pending_rows = self.db.execute(
+			"SELECT id, event FROM alerts "
+			"WHERE zabbix_pending = 1 AND timestamp >= ? "
+			"ORDER BY id",
+			(min_timestamp,)
+		).fetchall()
+
+		if not pending_rows:
+			return True
+
+		items = []
+		item_ids = []
+
+		for alert_id, event_json in pending_rows:
+			try:
+				pending_event = Event(event_json)
+			except ValueError as e:
+				log.warning("Skipping malformed pending alert from database: %s" % e)
+				continue
+
+			items.append(ItemValue(
+				alert_item_hostname,
+				alert_item_key,
+				pending_event.json,
+				int(pending_event.timestamp.timestamp())
+			))
+			item_ids.append(alert_id)
+
+		if not items:
+			return True
+
+		try:
+			# Send all pending items in bulk to Zabbix
+			response = await self.zabbix_sender.send(items)
+
+			# Simulate zabbix_utils method of splitting the items into chunks
+			# so if a chunk fails to send, we know what items have failed
+			chunk_size = getattr(self.zabbix_sender, 'chunk_size', 250)
+			chunked_item_ids = [item_ids[i:i + chunk_size] for i in range(0, len(item_ids), chunk_size)]
+			successful_ids = []
+
+			# Check for failures, determine which chunks where sent successfully
+			if response.failed == 0:
+				successful_ids = item_ids
+			elif response.details:
+				for node, chunks in response.details.items():
+					for chunk_index, resp in enumerate(chunks):
+						chunk_ids = chunked_item_ids[chunk_index]
+						if resp.failed == 0:
+							log.debug(f"Zabbix sender: Pending chunk sent successfully to {node} in {resp.time}")
+							successful_ids.extend(chunk_ids)
+						else:
+							log.error(f"Zabbix sender: Failed to send pending chunk to {node} at chunk {resp.chunk}")
+							log.debug(response)
+			else:
+				log.error(f"Zabbix sender: Failed to send {len(item_ids)} pending alerts.")
+				log.debug(response)
+
+			# Mark sent items as no longer pending in the DB
+			if successful_ids:
+				self.db.execute(
+					"UPDATE alerts SET zabbix_pending = 0 WHERE id IN ({})".format(
+						", ".join("?" for _ in successful_ids)
+					),
+					successful_ids
+				)
+				self.db.commit()
+
+			log.debug(f"Zabbix sender: {len(successful_ids)} alerts sent successfully")
+			pending_count = len(item_ids) - len(successful_ids)
+			if pending_count != 0:
+				log.debug(f"Zabbix sender: {pending_count} alerts failed to send and are still pending")
+			return pending_count == 0
+
+		except Exception as e:
+			log.error(f"Zabbix sender: Failed to send {len(item_ids)} pending alerts: {e}")
+			return False
+
 	def optimize(self):
 		"""
 			Called when the process exits to optimize the database
-- 
2.54.0


-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

[PATCH 5/5] Add background task to send all pending alerts to Zabbix every 1 second

[Robin Roevens]

Background task that will send pending events every 1 second. On failure it will
retry 3 times and then pauses until next alert comes in to prevent hammering a
possibly already overloaded Zabbix server..

Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
---
 src/suricata-reporter.in | 109 +++++++++++++++++++++++++++++++++------
 1 file changed, 94 insertions(+), 15 deletions(-)

diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
index 47482ab..a95f6b7 100644
--- a/src/suricata-reporter.in
+++ b/src/suricata-reporter.in
@@ -82,10 +82,6 @@ class Reporter(object):
 		# Remember the last time the database was cleaned
 		self.last_cleanup_at = None
 
-		# Initialize Zabbix sender
-		self.zabbix_sender = None
-		self.init_zabbix_sender()
-
 		# Register any signals
 		for signo in (signal.SIGINT, signal.SIGTERM):
 			self.loop.add_signal_handler(signo, self.terminate)
@@ -99,6 +95,11 @@ class Reporter(object):
 		# Create the socket
 		self.sock = self._create_socket()
 
+		# Initialize Zabbix sender
+		self.zabbix_sender = None
+		self.init_zabbix_sender()
+		self.zabbix_sender_task = None
+
 	def read_config(self):
 		"""
 			Reads or re-reads the configuration.
@@ -123,6 +124,12 @@ class Reporter(object):
 		zabbix_server_host = self.config.get('zabbix', 'zabbix_server_host', fallback='')
 		zabbix_server_port = self.config.getint('zabbix', 'zabbix_server_port', fallback=10051)
 
+		# Zabbix sender backoff state
+		self.zabbix_consecutive_failures = 0
+		self.zabbix_pause_until_new_event = False
+		self.zabbix_sender_wakeup = asyncio.Event()
+		self.zabbix_pause_after_failures = 3
+
 		if zabbix_config:
 			if not os.path.isfile(zabbix_config):
 				log.error(f"Zabbix agent config file {zabbix_config} does not exist.")
@@ -242,6 +249,67 @@ class Reporter(object):
 		# Return the socket
 		return sock
 
+	def _start_zabbix_sender_task(self):
+		"""
+			Start a background Zabbix sender task
+		"""
+		if not self.config.getboolean("zabbix", "enabled", fallback=False):
+			return
+
+		self.zabbix_sender_task = asyncio.create_task(self._periodic_zabbix_sender_flush())
+
+	async def _stop_zabbix_sender_task(self):
+		"""
+			Stops the background Zabbix sender task
+		"""
+		if not self.zabbix_sender_task:
+			return
+		
+		self.zabbix_sender_task.cancel()
+		try:
+			await self.zabbix_sender_task
+		except asyncio.CancelledError:
+			pass
+			
+		# Final flush of any remaining alerts
+		await self.flush_pending_to_zabbix()
+
+	async def _periodic_zabbix_sender_flush(self):
+		"""
+			Background task that periodically sends all pending Zabbix alerts.
+			Runs every 1 second to batch-send alerts in bulk on heavy load, but 
+			pauses after repeated failures until a new event arrives.
+		"""
+		log.debug("Starting periodic Zabbix sender task")
+		
+		try:
+			while not self.is_terminated.is_set():
+				if self.zabbix_pause_until_new_event:
+					log.warning("Zabbix sending is paused after repeated failures; waiting for the next event")
+					self.zabbix_sender_wakeup.clear()
+					await self.zabbix_sender_wakeup.wait()
+					self.zabbix_sender_wakeup.clear()
+					self.zabbix_pause_until_new_event = False
+					self.zabbix_consecutive_failures = 0
+					continue
+
+				if self.is_terminated.is_set():
+					break
+
+				# Send pending alerts to Zabbix
+				if await self.flush_pending_to_zabbix():
+					self.zabbix_consecutive_failures = 0
+				else:
+					self.zabbix_consecutive_failures += 1
+					if self.zabbix_consecutive_failures >= self.zabbix_pause_after_failures:
+						self.zabbix_pause_until_new_event = True
+
+				await asyncio.sleep(1)
+				
+		except asyncio.CancelledError:
+			log.debug("Periodic Zabbix sender task cancelled")
+			raise
+
 	async def run(self):
 		"""
 			The main loop of the application.
@@ -251,22 +319,29 @@ class Reporter(object):
 		# Cleanup the database at startup
 		self.cleanup()
 
-		# Wait until we have terminated
-		await self.is_terminated.wait()
+		# Start the periodic Zabbix sender task
+		self._start_zabbix_sender_task()
 
-		# Remove the socket so we won't receive any more data
 		try:
-			os.unlink(self.socket_path)
-		except OSError as e:
-			log.error("Failed to remove %s: %s" % (self.socket_path, e))
+			# Wait until we have terminated
+			await self.is_terminated.wait()
+		finally:
+			# Remove the socket so we won't receive any more data
+			try:
+				os.unlink(self.socket_path)
+			except OSError as e:
+				log.error("Failed to remove %s: %s" % (self.socket_path, e))
+
+			# Cancel the periodic Zabbix sender task
+			await self._stop_zabbix_sender_task()
 
-		# We will optimize the database before we exit
-		self.optimize()
+			# We will optimize the database before we exit
+			self.optimize()
 
-		# Close the database
-		self.db.close()
+			# Close the database
+			self.db.close()
 
-		log.debug("Reporter has exited")
+			log.debug("Reporter has exited")
 
 	def terminate(self):
 		"""
@@ -485,6 +560,10 @@ class Reporter(object):
 		# Store the alert
 		self.store(event)
 
+		# Wake the periodic flush task so it can retry immediately on new input
+		if self.config.getboolean("zabbix", "enabled", fallback=False):
+			self.zabbix_sender_wakeup.set()
+
 		# Send to syslog
 		if self.config.getboolean("syslog", "enabled", fallback=False):
 			await self.send_to_syslog(event)
-- 
2.54.0


-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

Re: [PATCH 0/5] Add Zabbix functionality to suricata-reporter

[Robin Roevens]

Small correction to my explanation, the statement I made about sending
older events from the time when zabbix functionality is not yet
enabled, is not true, as when zabbix functionality is disabled, events
recorded in the database are not marked as pending, so they won't be
sent to zabbix when that functionality is enabled on a later time.

It does however give the user the implicit functionality of augmenting
the max_age in case zabbix server was unreachable for longer than
current max_age to recover events it missed due to previous max_age
setting.

Regards
Robin

Robin Roevens schreef op do 30-07-2026 om 21:15 [+0200]:
> Hi all,
> 
> As discussed here earlier, I've worked on implementing sending
> Suricata alerts straight to Zabbix from within suricata-reporter
> instead
> of trying to parse the suricata logging separately using the Zabbix
> agent.
> 
> For this I use the zabbix-utils python library, which I submited here
> also as a separate pak (but meanwhile already requires an update,
> which
> I will post soon). This set of patches makes suricata-reporter able
> to
> directly communicate to a Zabbix server without having the
> zabbix_agentd
> pak installed, sending suricata alerts in real-time.
> 
> As Zabbix supports sending items in bulk, I have opted to create an
> async background task that will send all events from last 1 second in
> bulk so that even in the case that there are hundreds of incoming
> alerts, Zabbix server is only contacted once per second.
> 
> When for some reason sending to Zabbix server fails, it will be
> retried
> 3 times and then the background task will be suspended until a new
> suricata event comes in. That will wake the task again and retry to
> send all
> pending events. In environments with many events, that may actually
> not
> have that much of an effect. But in the average environment, this
> will
> give the Zabbix Server some breathing space as it failing to receive
> our
> events, may indicate a Zabbix server overload. 
> 
> For this I have to keep track which events are sent and which are
> pending. So I added a column in the database that keeps track of
> that.
> 
> I have also added an alert_max_age config parameter that allows the
> user
> to set how long suricata-reporter should retry to send events to
> Zabbix.
> Events older than that set age, will no longer be sent to Zabbix.
> This also give the user the implicit option to send older events when
> only just enabling the zabbix sending functionality, since the DB
> column
> exists and no event was ever sent to Zabbix, all events will be
> 'pending". At first run with zabbix functionality enabled, all events
> up
> to alert_max_age that are in the database will be sent to zabbix
> immediatly.
> 
> All events sent to Zabbix contain the timestamp of retrieval by
> suricata-reporter, so Zabbix will register and order them as received
> on that
> timestamp independently of the actual time Zabbix itself received the
> event.
> 
> This is my first adventure in Python async programming, so I hope I
> did
> not make any flagrant mistakes. But the code has been running here
> for
> weeks now without any problem. I have not actually tested large
> bursts
> of events, as I could not simulate that.. But I did make Zabbix
> server
> slow, unavailable and finally replaced it with netcat (to accept the
> connection, but
> not react on it) and I had the connection with the server off for a
> few
> hours to then re-establish the connection to see hundereds of pending
> events 
> being registered in only a few milliseconds.
> I did not notice any problems with suricata-reporter in any of these
> cases.
> 
> Regards
> 
> Robin

-- 
Dit bericht is gescanned op virussen en andere gevaarlijke
inhoud door MailScanner en lijkt schoon te zijn.

Re: [PATCH 0/5] Add Zabbix functionality to suricata-reporter

[Michael Tremer]

Hello Robin,

Thank you very much for sending these patches.

Before we dig into the code, I have a couple of questions about the design...

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> Hi all,
> 
> As discussed here earlier, I've worked on implementing sending
> Suricata alerts straight to Zabbix from within suricata-reporter instead
> of trying to parse the suricata logging separately using the Zabbix agent.
> 
> For this I use the zabbix-utils python library, which I submited here
> also as a separate pak (but meanwhile already requires an update, which
> I will post soon). This set of patches makes suricata-reporter able to
> directly communicate to a Zabbix server without having the zabbix_agentd
> pak installed, sending suricata alerts in real-time.

Yes, this is a good choice and I like that suricate-reporter will try to load support for Zabbix and if the module is not available, it simply disables support for Zabbix. That allows us to have a smaller configuration file if things like this are auto-detected.

> As Zabbix supports sending items in bulk, I have opted to create an
> async background task that will send all events from last 1 second in
> bulk so that even in the case that there are hundreds of incoming
> alerts, Zabbix server is only contacted once per second.

Okay, this makes sense. But I believe that there is already a small race in the implementation:

If the client side (in this case suricata-reporter) does not finish the call of flush_pending_to_zabbix() within that second, it will be called again which will result in the same rows being selected again, transmitted again, and assuming that there are just thousands of alarms it will take over a second again, the function will be called again, and so on. So the application will stall very quickly.

Although we should not see thousands of alerts per second under normal conditions, there could be other reasons why this is taking some time. For example, the Zabbix host could be in a different location and round-trips around half the planet are taking some time; it could be busy writing other things to its database or the database has just decided to do a little cleanup job. One second isn’t a lot of time then and we will have to make the system a little bit more resilient against this.

> When for some reason sending to Zabbix server fails, it will be retried
> 3 times and then the background task will be suspended until a new
> suricata event comes in. That will wake the task again and retry to send all
> pending events. In environments with many events, that may actually not
> have that much of an effect. But in the average environment, this will
> give the Zabbix Server some breathing space as it failing to receive our
> events, may indicate a Zabbix server overload.

Good thinking here.

> For this I have to keep track which events are sent and which are
> pending. So I added a column in the database that keeps track of that.

So, this is a very crucial thing we probably need to discuss :)

What is the rationale behind this? Obviously there are some easy answers:

1) We don’t want to loose any history if the network or Zabbix is down

2) We can even restart the reporter without losing any alerts

But then I am already running out of ideas why this could be a good idea. The cons that I can see are:

* A lot of additional I/O on the database. Although we would be updating rows very briefly after they have been written to the database, it will create a copy of the row and change the append-only architecture of the database. It will have a lot more cleaning up to do to evict all updated rows.

* You will only ever go back by about 1h by default. Could we just not keep things in RAM for that long?

I am not saying that I hate the idea, but I am not sure whether it is worth paying the price. The good side is that if people are not using Zabbix, there is no overhead except the space for the extra column. But if we would add another monitoring solution, we would potentially have to add another field, and another, and another?

So a possible other solution that I can come up with would be: Creating a separate table with all pending events that have to be transmitted. And every once in a while we truncate it should it become too long. We could even keep a list of IDs in memory only if we want to go down that route.

> I have also added an alert_max_age config parameter that allows the user
> to set how long suricata-reporter should retry to send events to Zabbix.
> Events older than that set age, will no longer be sent to Zabbix.
> This also give the user the implicit option to send older events when
> only just enabling the zabbix sending functionality, since the DB column
> exists and no event was ever sent to Zabbix, all events will be
> 'pending". At first run with zabbix functionality enabled, all events up
> to alert_max_age that are in the database will be sent to zabbix
> immediatly.

I like the mechanism, but whenever I am building something like this, I am never sure what would be a reasonable window.

Locally, suricate-reporter is keeping the events for pretty much forever. So we could even go back three days or something. Or we could give up really quickly. After maybe a minute. I never know what is right, but for the implementation, the length of the window plays a role - see above.

With email and syslog we do more of a “fire and forget” approach. If we send the syslog message and syslog wasn’t ready to receive it, we wouldn’t know and we would not try again...

> All events sent to Zabbix contain the timestamp of retrieval by
> suricata-reporter, so Zabbix will register and order them as received on that
> timestamp independently of the actual time Zabbix itself received the
> event.
> 
> This is my first adventure in Python async programming, so I hope I did
> not make any flagrant mistakes. But the code has been running here for
> weeks now without any problem. I have not actually tested large bursts
> of events, as I could not simulate that.. But I did make Zabbix server
> slow, unavailable and finally replaced it with netcat (to accept the connection, but
> not react on it) and I had the connection with the server off for a few
> hours to then re-establish the connection to see hundereds of pending events 
> being registered in only a few milliseconds.
> I did not notice any problems with suricata-reporter in any of these
> cases.

This is good testing. Usually, if I need to create a lot of events, I enable the “PING” rule in “icmp_info” and just send a lot of ping packets to the firewall. You could try a flood ping with “ping -f”.

I will send some more comments about the code in the other emails.

Best,
-Michael

> 
> Regards
> 
> Robin
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
> 

Re: [PATCH 1/5] Initialize async zabbix sender from zabbix_utils

[Michael Tremer]

Hello,

Just as a warning, I am going to be picky :)

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> When Zabbix is enabled in new config section [zabbix], and the zabbix_utils 
> python module is available, a zabbix AsyncSender object will be initialized 
> for sending alerts to Zabbix using parameters from the new config section.
> 
> Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
> ---
> src/reporter.conf.in     | 23 +++++++++++++++++++++++
> src/suricata-reporter.in | 37 +++++++++++++++++++++++++++++++++++++
> 2 files changed, 60 insertions(+)
> 
> diff --git a/src/reporter.conf.in b/src/reporter.conf.in
> index 5943006..bab01b6 100644
> --- a/src/reporter.conf.in
> +++ b/src/reporter.conf.in
> @@ -45,3 +45,26 @@
> ; 3 = Low Severity
> ; 4 = Informational
> ;severity = 3
> +
> +[zabbix]
> +; Enable sending alerts to Zabbix
> +;enabled = false
> +
> +; Path to the Zabbix agent configuration file
> +;zabbix_agentd_config = /etc/zabbix_agentd/zabbix_agentd.conf
> +
> +; Zabbix server ip or hostname (required if zabbix_agentd_config is not set)
> +;zabbix_server_host = 127.0.0.1
> +
> +; Zabbix server port (defaults to 10051 if not set)
> +;zabbix_server_port = 10051

You seem to like loooong variable names which I would shorten.

You are already in the [zabbix] section of the file, so you could simply call it “host” and “port”. Simple names.

> +
> +; Hostname as defined in Zabbix server to send alerts to (defaults to either the
> +; Hostname directive in Zabbix Agent config or system hostname)
> +;alert_item_hostname = IPFire
> +
> +; Zabbix item key to send alerts to
> +;alert_item_key = ipfire.suricata.event.get
> +
> +; Max age (seconds) to retry sending alerts to Zabbix
> +;alert_max_age = 3600
> \ No newline at end of file
> diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
> index 28b55bc..f9da7b4 100644
> --- a/src/suricata-reporter.in
> +++ b/src/suricata-reporter.in
> @@ -37,6 +37,13 @@ import socket
> import sqlite3
> import sys
> 
> +# Load zabbix_utils module if available 
> +zabbix_utils_available = True
> +try:
> +    from zabbix_utils import AsyncSender, ItemValue
> +except ImportError:
> +    zabbix_utils_available = False
> +

To keep the logic in one block, you could add the “True” statement to the else: clause of the block.

> # Fetch the hostname
> HOSTNAME = socket.gethostname()
> 
> @@ -75,6 +82,10 @@ class Reporter(object):
> # Remember the last time the database was cleaned
> self.last_cleanup_at = None
> 
> + # Initialize Zabbix sender
> + self.zabbix_sender = None
> + self.init_zabbix_sender()

I would have the function just return the sender object. That way, you can make it one line here and you don’t have to remember the name of the class variable in the function below.

> +
> # Register any signals
> for signo in (signal.SIGINT, signal.SIGTERM):
> self.loop.add_signal_handler(signo, self.terminate)
> @@ -97,6 +108,32 @@ class Reporter(object):
> 
> return config
> 
> + def init_zabbix_sender(self):
> + """
> + Initialize the Zabbix async sender if configured
> + """
> + if not self.config.getboolean('zabbix', 'enabled', fallback=False):
> + return
> +
> + if not zabbix_utils_available:
> + log.error("zabbix-utils is not installed. Zabbix alerts will not be sent.")
> + return
> +
> + zabbix_config = self.config.get('zabbix', 'zabbix_agentd_config', fallback='')
> + zabbix_server_host = self.config.get('zabbix', 'zabbix_server_host', fallback='')
> + zabbix_server_port = self.config.getint('zabbix', 'zabbix_server_port', fallback=10051)

Here you are being bitten by the long names again. This would be much shorter:

 host = self.config.get('zabbix', 'host', fallback=‘')

Oh, and you seem to be mixing “ and ‘ a lot in the code :) Both work, but just “ would be fine for me.

> +
> + if zabbix_config:
> + if not os.path.isfile(zabbix_config):
> + log.error(f"Zabbix agent config file {zabbix_config} does not exist.")
> + return
> + self.zabbix_sender = AsyncSender(use_config=True, config_path=zabbix_config)
> + else:
> + if not zabbix_server_host:
> + log.error("zabbix_server_host must be specified when zabbix_agentd_config is not provided.")
> + return
> + self.zabbix_sender = AsyncSender(server=zabbix_server_host, port=zabbix_server_port)
> +
> def _open_database(self):
> """
> Opens the database
> -- 
> 2.54.0
> 
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
> 

Re: [PATCH 2/5] Add database column zabbix_pending in alerts table.

[Michael Tremer]

Hello,

This works, but it is very over-engineered :)

You can simply let the database do what it is doing best and not worry about it in Python.

You are almost there with statements like this:

  CREATE INDEX IF NOT EXISTS alerts_zabbix_pending ON alerts(zabbix_pending)

Add this to the schema in https://git.ipfire.org/?p=suricata-reporter.git;a=blob;f=src/suricata-reporter.in;h=28b55bc39616f1af2769fb7935a972a1288f69bd;hb=HEAD#l114 and the database will create the index if it isn’t there. If it exists, it will simply do nothing.

You can do the same thing for the ADD COLUMN statement.

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> To prevent possible hammering of the Zabbix server in busy environments,
> instead of sending alerts immediatly when they arrive, we will send them in bulk
> every 1 second. For that we need to know what alerts where not yet sent to 
> Zabbix. This will be tracked in this new column zabbix_pending in the alerts
> table.
> 
> Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
> ---
> src/suricata-reporter.in | 23 +++++++++++++++++++++++
> 1 file changed, 23 insertions(+)
> 
> diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
> index f9da7b4..83e4e97 100644
> --- a/src/suricata-reporter.in
> +++ b/src/suricata-reporter.in
> @@ -134,6 +134,27 @@ class Reporter(object):
> return
> self.zabbix_sender = AsyncSender(server=zabbix_server_host, port=zabbix_server_port)
> 
> + def _ensure_zabbix_pending_column(self, db):
> + """
> + Ensures the database schema always has the zabbix_pending column and index.
> + If they are missing, the database is migrated to add them.
> + """
> + cursor = db.execute("PRAGMA table_info(alerts)")
> + columns = {row[1] for row in cursor.fetchall()}
> +
> + if "zabbix_pending" not in columns:
> + db.execute("ALTER TABLE alerts ADD COLUMN zabbix_pending INTEGER NOT NULL DEFAULT 0")
> + db.commit()
> + log.debug("Database: Added zabbix_pending column to alerts table.")
> +
> + cursor = db.execute("PRAGMA index_list(alerts)")
> + indexes = {row[1] for row in cursor.fetchall()}
> +
> + if "alerts_zabbix_pending" not in indexes:
> + db.execute("CREATE INDEX IF NOT EXISTS alerts_zabbix_pending ON alerts(zabbix_pending)")
> + db.commit()
> + log.debug("Database: Added alerts_zabbix_pending index on alerts table column zabbix_pending.")
> +
> def _open_database(self):
> """
> Opens the database
> @@ -165,6 +186,8 @@ class Reporter(object):
> CREATE INDEX IF NOT EXISTS alerts_timestamp ON alerts(timestamp);
> """)
> 
> + self._ensure_zabbix_pending_column(db)
> +
> return db
> 
> @property
> -- 
> 2.54.0
> 
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
> 

Re: [PATCH 3/5] Set zabbix_pending flag when storing new event in DB

[Michael Tremer]

Hello,

The same goes here. Let the database do what it is doing well.

Instead of figuring out whether Zabbix is enabled or not, you could simply change the “DEFAULT” of the field to “true” and leave the INSERT statement unmodified. If someone enables Zabbix afterwards, the downside would be to receive all events from the past. I am not sure if that is a big disadvantage?!

I am thinking towards the future where we could have many more monitoring solutions added here, so for each of them we would have to query the status and adjust the statement. That could become somewhat expensive.

-Michael

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> When sending to zabbix is enabled, we need to set the zabbix_pending flag set 
> on each new event written to the database.
> 
> Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
> ---
> src/suricata-reporter.in | 9 +++++++--
> 1 file changed, 7 insertions(+), 2 deletions(-)
> 
> diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
> index 83e4e97..78bb04d 100644
> --- a/src/suricata-reporter.in
> +++ b/src/suricata-reporter.in
> @@ -314,9 +314,14 @@ class Reporter(object):
> """
> Writes a single event to the database
> """
> + # Determine whether this event should be marked for Zabbix delivery
> + zabbix_pending = 1 if self.config.getboolean('zabbix', 'enabled', fallback=False) else 0
> +
> # Write the event to the database
> - self.db.execute("INSERT INTO alerts(timestamp, event) VALUES(?, ?)",
> - (event.timestamp.timestamp(), event.json))
> + self.db.execute(
> + "INSERT INTO alerts(timestamp, event, zabbix_pending) VALUES(?, ?, ?)",
> + (event.timestamp.timestamp(), event.json, zabbix_pending)
> + )
> 
> # Commit it straight away
> self.db.commit()
> -- 
> 2.54.0
> 
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
> 

Re: [PATCH 4/5] Add function to send all pending alerts to Zabbix

[Michael Tremer]

Hello again,

So this is where the heavy lifting is happening. Before we make any changes to it we should decide if we keep the design as-is or if we want to make some adjustments.

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> Add abbility to send all alerts from DB marked as pending to Zabbix server in
> bulk with errorhandling
> 
> Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
> ---
> src/suricata-reporter.in | 94 ++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 94 insertions(+)
> 
> diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
> index 78bb04d..47482ab 100644
> --- a/src/suricata-reporter.in
> +++ b/src/suricata-reporter.in
> @@ -326,6 +326,100 @@ class Reporter(object):
> # Commit it straight away
> self.db.commit()
> 
> + async def flush_pending_to_zabbix(self):
> + """
> + Sends all alerts that were marked for Zabbix in a single bulk call.
> + """
> + if not self.zabbix_sender:
> + return False
> +
> + alert_item_hostname = self.config.get('zabbix', 'alert_item_hostname', fallback=(self.zabbix_sender.host or HOSTNAME))
> + alert_item_key = self.config.get('zabbix', 'alert_item_key', fallback='ipfire.suricata.event.get')
> +
> + now = datetime.datetime.now()
> + alert_max_age = datetime.timedelta(
> + seconds = self.config.getint('zabbix', 'alert_max_age', fallback=3600)
> + )
> + min_timestamp = (now - alert_max_age).timestamp()
> +
> + pending_rows = self.db.execute(
> + "SELECT id, event FROM alerts "
> + "WHERE zabbix_pending = 1 AND timestamp >= ? "
> + "ORDER BY id",
> + (min_timestamp,)
> + ).fetchall()

Here I would once again push for letting the database do what it is good at. Instead of doing some Python maths, you could simple give the database what you want to know:

  SELECT id, event FROM alerts WHERE zabbix_pending = 1 AND timestamp >= CURRENT_TIMESTAMP - ? ORDER BY id ASC;

The value would have to be in seconds.

I also don’t rely on the ID strictly incrementing only. So I would order by timestamp which is what you want logically and then add the id to make the sorting stable:

  SELECT id, event FROM alerts WHERE zabbix_pending = 1 AND timestamp >= CURRENT_TIMESTAMP - ? ORDER BY timestamp ASC, id ASC;

And to make this all really fast and efficient, I would recommend to create a partial index. Right now you are creating it like this:

  CREATE INDEX IF NOT EXISTS alerts_zabbix_pending ON alerts(zabbix_pending)

It will index the entire database and store all rows by zabbix_pending being either 1 or 0. But you never really care about the zero case, so you could make the index MUCH smaller by never including them:

  CREATE INDEX alerts_zabbix_pending ON alerts(timestamp ASC, id ASC) WHERE zabbix_pending = 1;

If you would run a query with “WHERE zabbix_pending = 0”, the database would have to scan the entire table. Potentially millions of rows. But we never do that.

Asking for “WHERE zabbix_pending = 1” would simply open the index. And by using “timestamp ASC, id ASC”, we already have the sorting done. So the SELECT query above is basically becoming VERY cheap where it only has to open the index. Most of the time, the index is actually empty because everything has been submitted already and so the database will never ever have to read a single block from disk.

Without the WHERE clause in the CREATE INDEX statement, you would have an index that has an inventory of millions of rows that you would never ever read from.

But we are not done, yet...

> +
> + if not pending_rows:
> + return True
> +
> + items = []
> + item_ids = []
> +
> + for alert_id, event_json in pending_rows:
> + try:
> + pending_event = Event(event_json)
> + except ValueError as e:
> + log.warning("Skipping malformed pending alert from database: %s" % e)
> + continue
> +
> + items.append(ItemValue(
> + alert_item_hostname,
> + alert_item_key,
> + pending_event.json,
> + int(pending_event.timestamp.timestamp())
> + ))
> + item_ids.append(alert_id)
> +
> + if not items:
> + return True
> +
> + try:
> + # Send all pending items in bulk to Zabbix
> + response = await self.zabbix_sender.send(items)
> +
> + # Simulate zabbix_utils method of splitting the items into chunks
> + # so if a chunk fails to send, we know what items have failed
> + chunk_size = getattr(self.zabbix_sender, 'chunk_size', 250)
> + chunked_item_ids = [item_ids[i:i + chunk_size] for i in range(0, len(item_ids), chunk_size)]
> + successful_ids = []
> +
> + # Check for failures, determine which chunks where sent successfully
> + if response.failed == 0:
> + successful_ids = item_ids
> + elif response.details:
> + for node, chunks in response.details.items():
> + for chunk_index, resp in enumerate(chunks):
> + chunk_ids = chunked_item_ids[chunk_index]
> + if resp.failed == 0:
> + log.debug(f"Zabbix sender: Pending chunk sent successfully to {node} in {resp.time}")
> + successful_ids.extend(chunk_ids)
> + else:
> + log.error(f"Zabbix sender: Failed to send pending chunk to {node} at chunk {resp.chunk}")
> + log.debug(response)
> + else:
> + log.error(f"Zabbix sender: Failed to send {len(item_ids)} pending alerts.")
> + log.debug(response)
> +
> + # Mark sent items as no longer pending in the DB
> + if successful_ids:
> + self.db.execute(
> + "UPDATE alerts SET zabbix_pending = 0 WHERE id IN ({})".format(
> + ", ".join("?" for _ in successful_ids)
> + ),
> + successful_ids
> + )
> + self.db.commit()

In this loop, you have a couple of problems that we could simply eliminate to make the whole thing perform better and become more resilient:

First of all, you are sending a query to the database and you are fetching ALL of the rows. That could be hundreds of thousands (or whoever many alerts you might have received in a window of an hour). So you are loading them all into memory which will easily blow through your one second window. You will need hundreds of megabytes of RAM, and in a second, the function will fire again doing the same and so on. So let’s not do that.

Let the database create the chunks and only fetch what you actually need. So move the query into a loop and change it as follows:

  SELECT id, event FROM alerts WHERE zabbix_pending = 1 AND timestamp >= CURRENT_TIMESTAMP - ? ORDER BY timestamp ASC, id ASC LIMIT 250;

So this will give you at most 250 rows. A chunk that is possible to deal with at a time and that is safe enough to read into memory. If the database does not return any rows, you simply abort like so:

  while True:
     rows = self.db.execute(“SELECT …”).fetchall()
     if not rows:
       break

     … do the rest

You won’t need any extra code to create the chunks and so on, you already get a perfectly sliced piece of data.

And then you have the UPDATE statement which could also go very wrong in case the function is running more than once. You are updating very late and second run could already be reading the same rows and sending them to Zabbix. That would require Zabbix to know this and de-duplicate, but I am not sure we can rely on that. Even if so, we would be wasting bandwidth.

So let’s be lazy again and let the database do its job. We can let the database handle the entire update in one single step. It would then lock the rows making sure that no other process can read them at the same time. It would exactly remember which rows it has given to you, so you don’t have to take care of it.

We can do that by replacing the SELECT statement with an UPDATE … RETURNING statement: Unfortunately UPDATE does not support ORDER and LIMIT, but that can be fixed:

  WITH batch AS (
      SELECT id FROM alerts
       WHERE zabbix_pending = 1
         AND timestamp >= CURRENT_TIMESTAMP - ?
       ORDER BY timestamp ASC, id ASC
       LIMIT 250
  )
  UPDATE alerts
     SET zabbix_pending = 0
    FROM batch
    WHERE alerts.id = batch.id
    RETURNING alerts.id <http://alerts.id/>, alerts.event;

Although it looks complicated, the statement is doing everything in one go: It selects the data that has to be submitted to Zabbix, and then immediately sets zabbix_pending = 0. But that is not yet written to the database because we will have to wrap everything in a transaction. After we have received the selected rows, we will send them to Zabbix and then call COMMIT on the database. Only then, the change will be written. Before that, nobody else can touch the rows because they are locked (with SQLite, the whole database is in fact) and if the submit function would be called again, it will not be able to update anything, yet. Instead it will wait until the previous transaction has been commmitted and then continue, ensuring that each batch of events is only ever read once. If writing to Zabbix has failed, we won’t commit the transaction, but perform a ROLLBACK which then never manifests the change to zabbix_pending = 0 and the same rows will be returned next time.

What do we do in case of a partial success from the Zabbix callback? We could simply set zabbix_pending = 1 for the failed IDs. This sounds complicated, but given how small the chance is that only a few of the messages fail, I would say this is worth doing. The locking would still stay in place.

That way, the code is becoming much shorter, will never have any races on the data inside the database. We would however lock the database for as long as it takes to submit the data to Zabbix. 

> +
> + log.debug(f"Zabbix sender: {len(successful_ids)} alerts sent successfully")
> + pending_count = len(item_ids) - len(successful_ids)
> + if pending_count != 0:
> + log.debug(f"Zabbix sender: {pending_count} alerts failed to send and are still pending")
> + return pending_count == 0
> +
> + except Exception as e:
> + log.error(f"Zabbix sender: Failed to send {len(item_ids)} pending alerts: {e}")
> + return False
> +
> def optimize(self):
> """
> Called when the process exits to optimize the database
> -- 
> 2.54.0
> 
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
> 

Re: [PATCH 5/5] Add background task to send all pending alerts to Zabbix every 1 second

[Michael Tremer]

Hello,

I like that Zabbix gets paused when we know for sure that there has not been another event.

But why the consecutive failure thing?

-Michael

> On 30 Jul 2026, at 20:15, Robin Roevens <robin.roevens@disroot.org> wrote:
> 
> Background task that will send pending events every 1 second. On failure it will
> retry 3 times and then pauses until next alert comes in to prevent hammering a
> possibly already overloaded Zabbix server..
> 
> Signed-off-by: Robin Roevens <robin.roevens@disroot.org>
> ---
> src/suricata-reporter.in | 109 +++++++++++++++++++++++++++++++++------
> 1 file changed, 94 insertions(+), 15 deletions(-)
> 
> diff --git a/src/suricata-reporter.in b/src/suricata-reporter.in
> index 47482ab..a95f6b7 100644
> --- a/src/suricata-reporter.in
> +++ b/src/suricata-reporter.in
> @@ -82,10 +82,6 @@ class Reporter(object):
> # Remember the last time the database was cleaned
> self.last_cleanup_at = None
> 
> - # Initialize Zabbix sender
> - self.zabbix_sender = None
> - self.init_zabbix_sender()
> -
> # Register any signals
> for signo in (signal.SIGINT, signal.SIGTERM):
> self.loop.add_signal_handler(signo, self.terminate)
> @@ -99,6 +95,11 @@ class Reporter(object):
> # Create the socket
> self.sock = self._create_socket()
> 
> + # Initialize Zabbix sender
> + self.zabbix_sender = None
> + self.init_zabbix_sender()
> + self.zabbix_sender_task = None
> +
> def read_config(self):
> """
> Reads or re-reads the configuration.
> @@ -123,6 +124,12 @@ class Reporter(object):
> zabbix_server_host = self.config.get('zabbix', 'zabbix_server_host', fallback='')
> zabbix_server_port = self.config.getint('zabbix', 'zabbix_server_port', fallback=10051)
> 
> + # Zabbix sender backoff state
> + self.zabbix_consecutive_failures = 0
> + self.zabbix_pause_until_new_event = False
> + self.zabbix_sender_wakeup = asyncio.Event()
> + self.zabbix_pause_after_failures = 3
> +
> if zabbix_config:
> if not os.path.isfile(zabbix_config):
> log.error(f"Zabbix agent config file {zabbix_config} does not exist.")
> @@ -242,6 +249,67 @@ class Reporter(object):
> # Return the socket
> return sock
> 
> + def _start_zabbix_sender_task(self):
> + """
> + Start a background Zabbix sender task
> + """
> + if not self.config.getboolean("zabbix", "enabled", fallback=False):
> + return
> +
> + self.zabbix_sender_task = asyncio.create_task(self._periodic_zabbix_sender_flush())
> +
> + async def _stop_zabbix_sender_task(self):
> + """
> + Stops the background Zabbix sender task
> + """
> + if not self.zabbix_sender_task:
> + return
> + 
> + self.zabbix_sender_task.cancel()
> + try:
> + await self.zabbix_sender_task
> + except asyncio.CancelledError:
> + pass
> + 
> + # Final flush of any remaining alerts
> + await self.flush_pending_to_zabbix()
> +
> + async def _periodic_zabbix_sender_flush(self):
> + """
> + Background task that periodically sends all pending Zabbix alerts.
> + Runs every 1 second to batch-send alerts in bulk on heavy load, but 
> + pauses after repeated failures until a new event arrives.
> + """
> + log.debug("Starting periodic Zabbix sender task")
> + 
> + try:
> + while not self.is_terminated.is_set():
> + if self.zabbix_pause_until_new_event:
> + log.warning("Zabbix sending is paused after repeated failures; waiting for the next event")
> + self.zabbix_sender_wakeup.clear()
> + await self.zabbix_sender_wakeup.wait()
> + self.zabbix_sender_wakeup.clear()
> + self.zabbix_pause_until_new_event = False
> + self.zabbix_consecutive_failures = 0
> + continue
> +
> + if self.is_terminated.is_set():
> + break
> +
> + # Send pending alerts to Zabbix
> + if await self.flush_pending_to_zabbix():
> + self.zabbix_consecutive_failures = 0
> + else:
> + self.zabbix_consecutive_failures += 1
> + if self.zabbix_consecutive_failures >= self.zabbix_pause_after_failures:
> + self.zabbix_pause_until_new_event = True
> +
> + await asyncio.sleep(1)
> + 
> + except asyncio.CancelledError:
> + log.debug("Periodic Zabbix sender task cancelled")
> + raise
> +
> async def run(self):
> """
> The main loop of the application.
> @@ -251,22 +319,29 @@ class Reporter(object):
> # Cleanup the database at startup
> self.cleanup()
> 
> - # Wait until we have terminated
> - await self.is_terminated.wait()
> + # Start the periodic Zabbix sender task
> + self._start_zabbix_sender_task()
> 
> - # Remove the socket so we won't receive any more data
> try:
> - os.unlink(self.socket_path)
> - except OSError as e:
> - log.error("Failed to remove %s: %s" % (self.socket_path, e))
> + # Wait until we have terminated
> + await self.is_terminated.wait()
> + finally:
> + # Remove the socket so we won't receive any more data
> + try:
> + os.unlink(self.socket_path)
> + except OSError as e:
> + log.error("Failed to remove %s: %s" % (self.socket_path, e))
> +
> + # Cancel the periodic Zabbix sender task
> + await self._stop_zabbix_sender_task()
> 
> - # We will optimize the database before we exit
> - self.optimize()
> + # We will optimize the database before we exit
> + self.optimize()
> 
> - # Close the database
> - self.db.close()
> + # Close the database
> + self.db.close()
> 
> - log.debug("Reporter has exited")
> + log.debug("Reporter has exited")
> 
> def terminate(self):
> """
> @@ -485,6 +560,10 @@ class Reporter(object):
> # Store the alert
> self.store(event)
> 
> + # Wake the periodic flush task so it can retry immediately on new input
> + if self.config.getboolean("zabbix", "enabled", fallback=False):
> + self.zabbix_sender_wakeup.set()
> +
> # Send to syslog
> if self.config.getboolean("syslog", "enabled", fallback=False):
> await self.send_to_syslog(event)
> -- 
> 2.54.0
> 
> 
> -- 
> Dit bericht is gescanned op virussen en andere gevaarlijke
> inhoud door MailScanner en lijkt schoon te zijn.
> 
>