monitor_app/panda/queue_inventory.py

0001 """The monitor's PanDA queue inventory (``PandaQueue``), kept current from
0002 PanDA's own queue table (``schedconfig_json``, the configuration PanDA
0003 itself reads) rather than from a static file.
0004
0005 Every EIC queue PanDA knows is created or updated with its configuration
0006 and status; the monitor's own additions (``metadata``: descriptions,
0007 classes) are left alone. A queue PanDA no longer lists keeps its row,
0008 marked ``absent``, so its history and descriptions stay reachable.
0009 Run with every pressure-front cycle (``swf_epicprod.front.run_cycle``)
0010 and behind the queue list's update button.
0011 """
0012 import logging
0013
0014 from django.db import connections, transaction
0015
0016 from ..models import PandaQueue
0017 from .constants import PANDA_SCHEMA
0018
0019 logger = logging.getLogger(__name__)
0020
0021 ABSENT = 'absent'
0022
0023
0024 def sync_queue_inventory(vo='eic'):
0025     """Bring ``PandaQueue`` in line with PanDA's queue table for ``vo``.
0026     Returns {created, updated, absent, queues}."""
0027     with connections['panda'].cursor() as cursor:
0028         cursor.execute(
0029             f'SELECT "panda_queue", "data" FROM "{PANDA_SCHEMA}"."schedconfig_json" '
0030             "WHERE \"data\"->>'vo_name' = %s", [vo])
0031         rows = cursor.fetchall()
0032     live = {}
0033     for name, data in rows:
0034         if isinstance(data, str):
0035             import json
0036             data = json.loads(data)
0037         live[str(name)] = data or {}
0038     created = updated = 0
0039     with transaction.atomic():
0040         existing = {q.queue_name: q for q in PandaQueue.objects.all()}
0041         for name, data in live.items():
0042             fields = {'site': str(data.get('site') or data.get('panda_site') or ''),
0043                       'queue_type': str(data.get('type') or ''),
0044                       'status': str(data.get('status') or ''),
0045                       'config_data': data}
0046             row = existing.get(name)
0047             if row is None:
0048                 PandaQueue.objects.create(queue_name=name, **fields)
0049                 created += 1
0050                 continue
0051             changed = [k for k, v in fields.items() if getattr(row, k) != v]
0052             if changed:
0053                 for k in changed:
0054                     setattr(row, k, fields[k])
0055                 row.save(update_fields=changed + ['updated_at'])
0056                 updated += 1
0057         gone = [name for name, row in existing.items()
0058                 if name not in live and row.status != ABSENT]
0059         if gone:
0060             PandaQueue.objects.filter(queue_name__in=gone).update(status=ABSENT)
0061     if created or gone:
0062         logger.info('queue inventory: %d created, %d updated, %d absent (%s)',
0063                     created, updated, len(gone), ', '.join(gone))
0064     return {'created': created, 'updated': updated, 'absent': len(gone),
0065             'queues': len(live)}