ms
34 lines
2.1 KiB
Python
34 lines
2.1 KiB
Python
"""One-time import of tasks still available in Flower into the persistent outbox."""
|
|
import argparse,os,uuid
|
|
from datetime import datetime,timezone
|
|
from dotenv import load_dotenv
|
|
import requests
|
|
from .outbox import Outbox
|
|
from .main import envelope
|
|
STATES={'SUCCESS':'task-succeeded','FAILURE':'task-failed','REVOKED':'task-revoked','STARTED':'task-started','RECEIVED':'task-received','RETRY':'task-retried','PENDING':'task-sent'}
|
|
def main():
|
|
p=argparse.ArgumentParser();p.add_argument('--env',default='.env');p.add_argument('--limit',type=int,default=100000);args=p.parse_args()
|
|
load_dotenv(args.env);os.umask(0o077)
|
|
if args.limit<1 or args.limit>1000000:raise SystemExit('limit must be between 1 and 1000000')
|
|
out=Outbox(os.getenv('AGENT_SPOOL','var/outbox.sqlite3'))
|
|
base=os.environ['AGENT_FLOWER_URL'].rstrip('/')
|
|
count=0;skipped=0;offset=0
|
|
while offset<args.limit:
|
|
r=requests.get(base+'/api/tasks',params={'limit':min(200,args.limit-offset),'offset':offset},auth=(os.getenv('AGENT_FLOWER_USER',''),os.getenv('AGENT_FLOWER_PASSWORD','')),timeout=30)
|
|
r.raise_for_status();tasks=r.json()
|
|
if not tasks:break
|
|
for task_id,t in tasks.items():
|
|
kind=STATES.get(t.get('state'))
|
|
stamp=t.get('timestamp') or t.get('succeeded') or t.get('failed') or t.get('started') or t.get('received') or t.get('sent')
|
|
if not kind or not isinstance(stamp,(int,float)):
|
|
skipped+=1;continue
|
|
payload={'uuid':str(task_id),'name':str(t.get('name') or ''),'hostname':str(t.get('worker') or t.get('hostname') or '')}
|
|
if isinstance(t.get('runtime'),(int,float)):payload['runtime']=t['runtime']
|
|
e=envelope(kind,payload,datetime.fromtimestamp(stamp,timezone.utc))
|
|
e['id']=str(uuid.uuid5(uuid.NAMESPACE_URL,base+'|'+task_id+'|'+kind+'|'+str(stamp)))
|
|
out.add(e);count+=1
|
|
offset+=len(tasks)
|
|
if len(tasks)<200:break
|
|
out.close();print(f'{count} task snapshots queued; {skipped} skipped because state/timestamp was unavailable.')
|
|
if __name__=='__main__':main()
|