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

[Robin Roevens <robin.roevens@disroot.org>]

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.