ms
40 lines
2.4 KiB
Python
40 lines
2.4 KiB
Python
from django.db import transaction
|
|
from django.utils import timezone
|
|
from .models import Server,Event,Metric,Worker,Task
|
|
STATES={'task-sent':'SENT','task-received':'RECEIVED','task-started':'STARTED','task-succeeded':'SUCCESS','task-failed':'FAILURE','task-retried':'RETRY','task-revoked':'REVOKED','task-rejected':'REJECTED'}
|
|
TERMINAL={'SUCCESS','FAILURE','REVOKED','REJECTED'}
|
|
@transaction.atomic
|
|
def ingest(server_id,events):
|
|
server=Server.objects.select_for_update().get(pk=server_id)
|
|
count=0
|
|
for item in events:
|
|
event,created=Event.objects.get_or_create(id=item['id'],defaults={'server':server,'kind':item['kind'],'occurred_at':item['occurred_at'],'payload':item['payload']})
|
|
if not created:
|
|
if event.server_id!=server.id: raise ValueError('Event-ID gehört zu einem anderen Agenten')
|
|
continue
|
|
count+=1
|
|
p=event.payload; at=event.occurred_at
|
|
if event.kind=='metric':
|
|
Metric.objects.create(event=event,server=server,at=at,cpu=p['cpu'],ram=p['ram'],disk=p['disk'])
|
|
if not server.latest_sample_at or at>=server.latest_sample_at:
|
|
server.latest_sample_at=at; server.latest=p
|
|
elif event.kind.startswith('worker-'):
|
|
w,new=Worker.objects.get_or_create(server=server,hostname=p['hostname'],defaults={'last_seen':at})
|
|
if at>=w.last_seen:
|
|
w.last_seen=at; w.online=event.kind!='worker-offline'; w.active=p.get('active',0); w.processed=p.get('processed',0); w.save()
|
|
elif event.kind in STATES:
|
|
state=STATES[event.kind]
|
|
t,new=Task.objects.get_or_create(server=server,task_id=p['uuid'],defaults={'first_at':at,'last_at':at,'state':state})
|
|
t.first_at=min(t.first_at,at)
|
|
# A late RECEIVED must never overwrite a terminal result. Retry is nonterminal.
|
|
if (state in TERMINAL and t.state not in TERMINAL) or (at>=t.last_at and (t.state not in TERMINAL or state in TERMINAL)):
|
|
t.state=state; t.last_at=at
|
|
if p.get('runtime') is not None: t.runtime=p['runtime']
|
|
t.retries=max(t.retries,p.get('retries',0))
|
|
if p.get('name'): t.name=p['name']
|
|
if p.get('hostname'): t.worker=p['hostname']
|
|
t.save()
|
|
server.last_seen=timezone.now()
|
|
server.save(update_fields=['last_seen','latest_sample_at','latest'])
|
|
return count
|