ms
63 lines
3.9 KiB
Python
63 lines
3.9 KiB
Python
import logging
|
|
from datetime import timedelta
|
|
from celery import shared_task
|
|
from django.conf import settings
|
|
from django.core.mail import send_mail
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from .models import Server,Configuration,Alarm,Notification,Event,Task,Worker,AccessToken,Audit
|
|
log=logging.getLogger(__name__)
|
|
@shared_task
|
|
def evaluate_alarms():
|
|
cfg,_=Configuration.objects.get_or_create(pk=1)
|
|
now=timezone.now(); cutoff=now-timedelta(seconds=cfg.heartbeat_timeout)
|
|
for sid in Server.objects.values_list('pk',flat=True):
|
|
with transaction.atomic():
|
|
s=Server.objects.select_for_update().get(pk=sid)
|
|
desired={}
|
|
if s.enabled and not s.maintenance:
|
|
if not s.latest_sample_at or s.latest_sample_at<cutoff:
|
|
desired['heartbeat']=('critical','Aktuelle Host-Messdaten fehlen')
|
|
else:
|
|
for key,limit in [('cpu',cfg.cpu_warning),('ram',cfg.ram_warning),('disk',cfg.disk_warning)]:
|
|
if s.latest.get(key,0)>=limit: desired[key]=('warning',f'{key.upper()} über Schwellwert: {s.latest[key]:.1f} %')
|
|
for service,online in s.latest.get('services',{}).items():
|
|
if not online: desired['service:'+service]=('critical',f'Dienst nicht erreichbar: {service}')
|
|
for w in Worker.objects.filter(server=s):
|
|
if not w.online or w.last_seen<cutoff: desired['worker:'+str(w.pk)]=('warning',f'Worker-Heartbeat fehlt: {w.hostname}'[:400])
|
|
for kind in ['flower-status','collector-status']:
|
|
e=Event.objects.filter(server=s,kind=kind).order_by('-occurred_at').first()
|
|
if e and (not e.payload.get('online') or e.occurred_at<cutoff): desired[kind]=('warning',f'{kind}: Verbindung unterbrochen oder veraltet')
|
|
Alarm.objects.filter(server=s,resolved_at__isnull=True).exclude(code__in=desired).update(resolved_at=now)
|
|
for code,(severity,message) in desired.items():
|
|
a,created=Alarm.objects.get_or_create(server=s,code=code,resolved_at__isnull=True,defaults={'severity':severity,'message':message})
|
|
if created: Notification.objects.create(alarm=a,next_attempt=now)
|
|
@shared_task
|
|
def deliver_notifications():
|
|
if not settings.ALERT_RECIPIENTS: return
|
|
for nid in Notification.objects.filter(sent_at__isnull=True,next_attempt__lte=timezone.now(),alarm__resolved_at__isnull=True).values_list('pk',flat=True)[:50]:
|
|
with transaction.atomic():
|
|
n=Notification.objects.select_for_update().select_related('alarm__server').get(pk=nid)
|
|
if n.sent_at: continue
|
|
n.attempts+=1
|
|
try:
|
|
send_mail(f'[dilib Operations] {n.alarm.severity}: {n.alarm.server.name}',n.alarm.message,settings.DEFAULT_FROM_EMAIL,settings.ALERT_RECIPIENTS,fail_silently=False)
|
|
n.sent_at=timezone.now(); n.last_error=''
|
|
except Exception as exc:
|
|
# Do not store SMTP credentials or raw exception messages.
|
|
n.last_error=type(exc).__name__; n.next_attempt=timezone.now()+timedelta(seconds=min(3600,30*2**min(n.attempts,7)))
|
|
log.warning('Notification %s failed: %s',n.pk,n.last_error)
|
|
n.save()
|
|
@shared_task
|
|
def cleanup():
|
|
cfg,_=Configuration.objects.get_or_create(pk=1)
|
|
cutoff=timezone.now()-timedelta(days=cfg.retention_days)
|
|
# Bounded batches avoid one enormous transaction; Metric rows cascade.
|
|
for model,field in [(Event,'occurred_at'),(Task,'last_at'),(Worker,'last_seen'),(Audit,'at')]:
|
|
for _ in range(20):
|
|
ids=list(model.objects.filter(**{field+'__lt':cutoff}).values_list('pk',flat=True)[:1000])
|
|
if not ids: break
|
|
model.objects.filter(pk__in=ids).delete()
|
|
Alarm.objects.filter(resolved_at__lt=cutoff).delete()
|
|
AccessToken.objects.filter(expires_at__lt=timezone.now()).delete()
|