mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-02-22 10:26:26 +00:00
Update notification service
This commit is contained in:
@@ -1015,6 +1015,22 @@ class ProxmoxHookWatcher:
|
|||||||
if key not in _reserved and key not in data:
|
if key not in _reserved and key not in data:
|
||||||
data[key] = str(val) if val is not None else ''
|
data[key] = str(val) if val is not None else ''
|
||||||
|
|
||||||
|
# Skip event types that the TaskWatcher already captures with
|
||||||
|
# full details (VMID, VM name, size, duration, etc.).
|
||||||
|
# The PVE webhook sends inferior data for these (no VMID, no name).
|
||||||
|
# Only forward events that tasks/journal do NOT catch reliably.
|
||||||
|
_TASKS_HANDLED = {
|
||||||
|
'backup_complete', 'backup_fail', 'backup_start',
|
||||||
|
'snapshot_complete', 'snapshot_fail',
|
||||||
|
'vm_start', 'vm_stop', 'vm_restart', 'vm_shutdown',
|
||||||
|
'ct_start', 'ct_stop',
|
||||||
|
'migration_start', 'migration_complete', 'migration_fail',
|
||||||
|
'replication_complete', 'replication_fail',
|
||||||
|
}
|
||||||
|
if event_type in _TASKS_HANDLED:
|
||||||
|
return {'accepted': False, 'skipped': True,
|
||||||
|
'reason': f'tasks_watcher_handles_{event_type}'}
|
||||||
|
|
||||||
event = NotificationEvent(
|
event = NotificationEvent(
|
||||||
event_type=event_type,
|
event_type=event_type,
|
||||||
severity=severity,
|
severity=severity,
|
||||||
@@ -1025,31 +1041,6 @@ class ProxmoxHookWatcher:
|
|||||||
raw=payload,
|
raw=payload,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Cross-source dedup: if a tasks/journal watcher already emitted
|
|
||||||
# the same event_type+entity_id within 60s, skip the webhook copy.
|
|
||||||
# The fingerprint is hostname:entity:entity_id:event_type (source-agnostic).
|
|
||||||
import time
|
|
||||||
dedup_key = f"{self._hostname}:{entity}:{entity_id}:{event_type}"
|
|
||||||
now = time.time()
|
|
||||||
if not hasattr(self, '_webhook_dedup'):
|
|
||||||
self._webhook_dedup = {}
|
|
||||||
last_seen = self._webhook_dedup.get(dedup_key, 0)
|
|
||||||
if now - last_seen < 60:
|
|
||||||
return {'accepted': False, 'skipped': True, 'reason': 'duplicate_within_60s'}
|
|
||||||
self._webhook_dedup[dedup_key] = now
|
|
||||||
|
|
||||||
# Also check the pipeline's global dedup (from other sources)
|
|
||||||
if hasattr(self, '_pipeline') and hasattr(self._pipeline, '_recent_fingerprints'):
|
|
||||||
if dedup_key in self._pipeline._recent_fingerprints:
|
|
||||||
fp_time = self._pipeline._recent_fingerprints[dedup_key]
|
|
||||||
if now - fp_time < 60:
|
|
||||||
return {'accepted': False, 'skipped': True, 'reason': 'duplicate_cross_source'}
|
|
||||||
|
|
||||||
# Cleanup old entries periodically
|
|
||||||
if len(self._webhook_dedup) > 200:
|
|
||||||
cutoff = now - 120
|
|
||||||
self._webhook_dedup = {k: v for k, v in self._webhook_dedup.items() if v > cutoff}
|
|
||||||
|
|
||||||
self._queue.put(event)
|
self._queue.put(event)
|
||||||
return {'accepted': True, 'event_type': event_type, 'event_id': event.event_id}
|
return {'accepted': True, 'event_type': event_type, 'event_id': event.event_id}
|
||||||
|
|
||||||
|
|||||||
@@ -491,20 +491,6 @@ class NotificationManager:
|
|||||||
if not self._enabled:
|
if not self._enabled:
|
||||||
return
|
return
|
||||||
|
|
||||||
# Track fingerprint for cross-source dedup (webhook vs tasks/journal).
|
|
||||||
# The webhook handler checks this dict to skip events already
|
|
||||||
# processed from tasks or journal watchers within 60s.
|
|
||||||
import time
|
|
||||||
if not hasattr(self, '_recent_fingerprints'):
|
|
||||||
self._recent_fingerprints = {}
|
|
||||||
self._recent_fingerprints[event.fingerprint] = time.time()
|
|
||||||
# Cleanup old entries
|
|
||||||
if len(self._recent_fingerprints) > 500:
|
|
||||||
cutoff = time.time() - 120
|
|
||||||
self._recent_fingerprints = {
|
|
||||||
k: v for k, v in self._recent_fingerprints.items() if v > cutoff
|
|
||||||
}
|
|
||||||
|
|
||||||
# Check if this event's GROUP is enabled in settings.
|
# Check if this event's GROUP is enabled in settings.
|
||||||
# The UI saves categories by group key: events.vm_ct, events.backup, etc.
|
# The UI saves categories by group key: events.vm_ct, events.backup, etc.
|
||||||
template = TEMPLATES.get(event.event_type, {})
|
template = TEMPLATES.get(event.event_type, {})
|
||||||
@@ -880,7 +866,6 @@ class NotificationManager:
|
|||||||
"""Process incoming Proxmox webhook. Delegates to ProxmoxHookWatcher."""
|
"""Process incoming Proxmox webhook. Delegates to ProxmoxHookWatcher."""
|
||||||
if not self._hook_watcher:
|
if not self._hook_watcher:
|
||||||
self._hook_watcher = ProxmoxHookWatcher(self._event_queue)
|
self._hook_watcher = ProxmoxHookWatcher(self._event_queue)
|
||||||
self._hook_watcher._pipeline = self # For cross-source dedup
|
|
||||||
return self._hook_watcher.process_webhook(payload)
|
return self._hook_watcher.process_webhook(payload)
|
||||||
|
|
||||||
def get_webhook_secret(self) -> str:
|
def get_webhook_secret(self) -> str:
|
||||||
|
|||||||
Reference in New Issue
Block a user