mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-10-21 21:12:01 +00:00
fix: deduplicate identical digest events
This commit is contained in:
@@ -1602,6 +1602,10 @@ class NotificationManager:
|
|||||||
'vm_fail', 'ct_fail',
|
'vm_fail', 'ct_fail',
|
||||||
'system_shutdown', 'system_reboot',
|
'system_shutdown', 'system_reboot',
|
||||||
})
|
})
|
||||||
|
# A task completion can be observed twice through adjacent collectors.
|
||||||
|
# Keep the daily digest useful by coalescing only byte-for-byte identical
|
||||||
|
# buffered INFO entries that arrive close together.
|
||||||
|
_DIGEST_DUPLICATE_WINDOW = 300 # seconds
|
||||||
|
|
||||||
def _should_buffer_for_digest(self, ch_name: str, severity: str,
|
def _should_buffer_for_digest(self, ch_name: str, severity: str,
|
||||||
event_type: str) -> bool:
|
event_type: str) -> bool:
|
||||||
@@ -1632,12 +1636,29 @@ class NotificationManager:
|
|||||||
conn = sqlite3.connect(str(DB_PATH), timeout=10)
|
conn = sqlite3.connect(str(DB_PATH), timeout=10)
|
||||||
conn.execute('PRAGMA journal_mode=WAL')
|
conn.execute('PRAGMA journal_mode=WAL')
|
||||||
conn.execute('PRAGMA busy_timeout=5000')
|
conn.execute('PRAGMA busy_timeout=5000')
|
||||||
|
# Adjacent collectors can observe the same completed task. Only
|
||||||
|
# coalesce byte-for-byte identical rows, so a different version,
|
||||||
|
# source or result stays visible in the same digest.
|
||||||
|
now = int(time.time())
|
||||||
|
duplicate = conn.execute(
|
||||||
|
'SELECT 1 FROM digest_pending '
|
||||||
|
'WHERE channel = ? AND event_type = ? AND event_group = ? '
|
||||||
|
'AND severity = ? AND title = ? AND body = ? AND ts >= ? '
|
||||||
|
'LIMIT 1',
|
||||||
|
(
|
||||||
|
ch_name, event_type, event_group, severity, title, body,
|
||||||
|
now - self._DIGEST_DUPLICATE_WINDOW,
|
||||||
|
),
|
||||||
|
).fetchone()
|
||||||
|
if duplicate:
|
||||||
|
conn.close()
|
||||||
|
return
|
||||||
conn.execute(
|
conn.execute(
|
||||||
'INSERT INTO digest_pending '
|
'INSERT INTO digest_pending '
|
||||||
'(channel, event_type, event_group, severity, ts, title, body) '
|
'(channel, event_type, event_group, severity, ts, title, body) '
|
||||||
'VALUES (?, ?, ?, ?, ?, ?, ?)',
|
'VALUES (?, ?, ?, ?, ?, ?, ?)',
|
||||||
(ch_name, event_type, event_group, severity,
|
(ch_name, event_type, event_group, severity,
|
||||||
int(time.time()), title, body),
|
now, title, body),
|
||||||
)
|
)
|
||||||
conn.commit()
|
conn.commit()
|
||||||
conn.close()
|
conn.close()
|
||||||
|
|||||||
@@ -707,6 +707,52 @@ class RuntimeCatalogTests(unittest.TestCase):
|
|||||||
self.assertNotIn("System", channel.payload[1])
|
self.assertNotIn("System", channel.payload[1])
|
||||||
self.assertTrue(channel.payload[3]["_quiet_hours_summary"])
|
self.assertTrue(channel.payload[3]["_quiet_hours_summary"])
|
||||||
|
|
||||||
|
def test_digest_buffer_coalesces_only_identical_recent_events(self):
|
||||||
|
manager = notification_manager.NotificationManager()
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
db_path = Path(directory) / "settings.db"
|
||||||
|
conn = sqlite3.connect(db_path)
|
||||||
|
conn.execute(
|
||||||
|
"CREATE TABLE digest_pending ("
|
||||||
|
"id INTEGER PRIMARY KEY, channel TEXT, event_type TEXT, "
|
||||||
|
"event_group TEXT, severity TEXT, ts INTEGER, title TEXT, body TEXT)"
|
||||||
|
)
|
||||||
|
conn.commit()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
with mock.patch.object(notification_manager, "DB_PATH", db_path), \
|
||||||
|
mock.patch.object(notification_manager.time, "time", side_effect=(1000, 1060, 1070, 1301)):
|
||||||
|
manager._buffer_digest_event(
|
||||||
|
"telegram", "lxc_update_applied", "vm_ct", "INFO",
|
||||||
|
"pve01: LXC wireguard (101) update completed", "Source: Manual",
|
||||||
|
)
|
||||||
|
# Same completed task from an adjacent collector: do not list it twice.
|
||||||
|
manager._buffer_digest_event(
|
||||||
|
"telegram", "lxc_update_applied", "vm_ct", "INFO",
|
||||||
|
"pve01: LXC wireguard (101) update completed", "Source: Manual",
|
||||||
|
)
|
||||||
|
# Different result/body for the same LXC remains visible.
|
||||||
|
manager._buffer_digest_event(
|
||||||
|
"telegram", "lxc_update_applied", "vm_ct", "INFO",
|
||||||
|
"pve01: LXC wireguard (101) update completed", "Source: Scheduled",
|
||||||
|
)
|
||||||
|
# The same result is allowed again outside the short window.
|
||||||
|
manager._buffer_digest_event(
|
||||||
|
"telegram", "lxc_update_applied", "vm_ct", "INFO",
|
||||||
|
"pve01: LXC wireguard (101) update completed", "Source: Manual",
|
||||||
|
)
|
||||||
|
|
||||||
|
conn = sqlite3.connect(db_path)
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT ts, body FROM digest_pending ORDER BY id"
|
||||||
|
).fetchall()
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
self.assertEqual(
|
||||||
|
rows,
|
||||||
|
[(1000, "Source: Manual"), (1070, "Source: Scheduled"), (1301, "Source: Manual")],
|
||||||
|
)
|
||||||
|
|
||||||
def test_visible_templates_use_known_backend_and_frontend_groups(self):
|
def test_visible_templates_use_known_backend_and_frontend_groups(self):
|
||||||
visible_groups = {
|
visible_groups = {
|
||||||
template.get("group", "other")
|
template.get("group", "other")
|
||||||
|
|||||||
Reference in New Issue
Block a user