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

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(+)
  

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.
> 
>