From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from mail02.haj.ipfire.org (localhost [IPv6:::1]) by mail02.haj.ipfire.org (Postfix) with ESMTP id 4hBMdQ6SHWz37DB for ; Fri, 31 Jul 2026 10:24:58 +0000 (UTC) Received: from mail01.ipfire.org (mail01.haj.ipfire.org [172.28.1.202]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange x25519) (Client CN "mail01.haj.ipfire.org", Issuer "YR2" (not verified)) by mail02.haj.ipfire.org (Postfix) with ESMTPS id 4hBMdM1SkGz37B5 for ; Fri, 31 Jul 2026 10:24:55 +0000 (UTC) Received: from [127.0.0.1] (localhost [127.0.0.1]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (No client certificate requested) by mail01.ipfire.org (Postfix) with ESMTPSA id 4hBMdB5YzXz5Ld; Fri, 31 Jul 2026 10:24:46 +0000 (UTC) DKIM-Signature: v=1; a=ed25519-sha256; c=relaxed/relaxed; d=ipfire.org; s=202003ed25519; t=1785493486; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=g3NSi9Y/YkOqm8LOFG8Y7mwzc3+VIU9ToS4B7Iwt/aU=; b=j0TMB962uP8NinVA/wPxkucTXDEXrTOP+hDCQO2KGPerQkvwUquJvszfaiBh3YN3njMTUU nOporTlbxgVBpbDw== DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=ipfire.org; s=202003rsa; t=1785493486; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=g3NSi9Y/YkOqm8LOFG8Y7mwzc3+VIU9ToS4B7Iwt/aU=; b=wefRlbor74j5YkXI6DoEo0ZFXq+qNC4V79WeK+29NlACdtydt8/mEc0hzZBHBMFmAwnZa1 U6UwlNDEOmGdApP9V1EUoimQ+Od+jOjblF9Ya6/q1FEQl9O5Kwgm0RfzWDJQAf01JgWFGf xk7vJKK7DVmPvxMd6UUug03xZphwva3VzbFHPxosn6eCBQTc177PbyfYF62uMF7Sg+1keo Y4A9dnqHY8JB8l74mQ7JFaeF9Pd18sT+T6EgURy8VAgPn/3Dv3337ejol101Wg1yhYUGUr M4N6VciodlrL6JpF2FAmutqJzGgB1VUvHI4tlPwS3/73zfqPL2VMYic8XcY9cg== Content-Type: text/plain; charset=utf-8 Precedence: list List-Id: List-Subscribe: , List-Unsubscribe: , List-Post: List-Help: Sender: Mail-Followup-To: Mime-Version: 1.0 Subject: Re: [PATCH 4/5] Add function to send all pending alerts to Zabbix From: Michael Tremer In-Reply-To: <20260730195148.3278295-5-robin.roevens@disroot.org> Date: Fri, 31 Jul 2026 11:24:30 +0100 Cc: development@lists.ipfire.org Content-Transfer-Encoding: quoted-printable Message-Id: <2EBB7379-DEC8-4DC8-B0CB-851A571AA52F@ipfire.org> References: <20260730195148.3278295-1-robin.roevens@disroot.org> <20260730195148.3278295-5-robin.roevens@disroot.org> To: Robin Roevens 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 = wrote: >=20 > Add abbility to send all alerts from DB marked as pending to Zabbix = server in > bulk with errorhandling >=20 > Signed-off-by: Robin Roevens > --- > src/suricata-reporter.in | 94 ++++++++++++++++++++++++++++++++++++++++ > 1 file changed, 94 insertions(+) >=20 > 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() >=20 > + 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 =3D self.config.get('zabbix', = 'alert_item_hostname', fallback=3D(self.zabbix_sender.host or HOSTNAME)) > + alert_item_key =3D self.config.get('zabbix', 'alert_item_key', = fallback=3D'ipfire.suricata.event.get') > + > + now =3D datetime.datetime.now() > + alert_max_age =3D datetime.timedelta( > + seconds =3D self.config.getint('zabbix', 'alert_max_age', = fallback=3D3600) > + ) > + min_timestamp =3D (now - alert_max_age).timestamp() > + > + pending_rows =3D self.db.execute( > + "SELECT id, event FROM alerts " > + "WHERE zabbix_pending =3D 1 AND timestamp >=3D ? " > + "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 =3D 1 AND timestamp = >=3D CURRENT_TIMESTAMP - ? ORDER BY id ASC; The value would have to be in seconds. I also don=E2=80=99t 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 =3D 1 AND timestamp = >=3D 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 =3D 1; If you would run a query with =E2=80=9CWHERE zabbix_pending =3D 0=E2=80=9D= , the database would have to scan the entire table. Potentially millions = of rows. But we never do that. Asking for =E2=80=9CWHERE zabbix_pending =3D 1=E2=80=9D would simply = open the index. And by using =E2=80=9Ctimestamp ASC, id ASC=E2=80=9D, 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 =3D [] > + item_ids =3D [] > + > + for alert_id, event_json in pending_rows: > + try: > + pending_event =3D 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 =3D 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 =3D getattr(self.zabbix_sender, 'chunk_size', 250) > + chunked_item_ids =3D [item_ids[i:i + chunk_size] for i in range(0, = len(item_ids), chunk_size)] > + successful_ids =3D [] > + > + # Check for failures, determine which chunks where sent successfully > + if response.failed =3D=3D 0: > + successful_ids =3D item_ids > + elif response.details: > + for node, chunks in response.details.items(): > + for chunk_index, resp in enumerate(chunks): > + chunk_ids =3D chunked_item_ids[chunk_index] > + if resp.failed =3D=3D 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 =3D 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=E2=80=99s 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 =3D 1 AND timestamp = >=3D 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 =3D self.db.execute(=E2=80=9CSELECT =E2=80=A6=E2=80=9D).fetchall= () if not rows: break =E2=80=A6 do the rest You won=E2=80=99t 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=E2=80=99s 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=E2=80=99t have to take care of it. We can do that by replacing the SELECT statement with an UPDATE =E2=80=A6 = 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 =3D 1 AND timestamp >=3D CURRENT_TIMESTAMP - ? ORDER BY timestamp ASC, id ASC LIMIT 250 ) UPDATE alerts SET zabbix_pending =3D 0 FROM batch WHERE alerts.id =3D batch.id RETURNING 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 =3D 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=E2=80=99t commit = the transaction, but perform a ROLLBACK which then never manifests the = change to zabbix_pending =3D 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 =3D 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.=20 > + > + log.debug(f"Zabbix sender: {len(successful_ids)} alerts sent = successfully") > + pending_count =3D len(item_ids) - len(successful_ids) > + if pending_count !=3D 0: > + log.debug(f"Zabbix sender: {pending_count} alerts failed to send and = are still pending") > + return pending_count =3D=3D 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 > --=20 > 2.54.0 >=20 >=20 > --=20 > Dit bericht is gescanned op virussen en andere gevaarlijke > inhoud door MailScanner en lijkt schoon te zijn. >=20 >=20