src/monitor_app/worker_records.py

0001 """Harvester worker records: our copy of PanDA's, kept past its window.
0002
0003 PanDA keeps a harvester worker row three months after its last update
0004 (``harvester_workers``, partitions dropped by date); the job-worker links
0005 and the jobs themselves stay. A batch allocation's start, end and cores are
0006 in the worker row alone, so the allocation page
0007 (``panda.queries.allocation_timeline``) loses the allocation's clock when
0008 the row goes. This module copies the rows of the queues whose allocations
0009 the page draws into ``HarvesterWorkerRecord``, nightly by the prod-ops
0010 agent (``worker_record_capture``), and the timeline reads the copy once
0011 PanDA no longer has the row (swf-epicprod docs/EPICPROD_OPS.md, Harvester
0012 worker records).
0013
0014 Which queues: those whose site name starts with ``SITE_PREFIXES``, the
0015 NERSC Perlmutter queues, whose allocations run many jobs each; about a
0016 hundred rows a day. A row is copied whole but its JDL, and copied again
0017 whenever PanDA's ``lastupdate`` is newer than ours, so a worker captured
0018 while it ran ends with its final state.
0019 """
0020 import logging
0021 from datetime import timedelta
0022 from datetime import timezone as dt_timezone
0023
0024 from django.db import connections
0025 from django.utils import timezone
0026
0027 from .panda.constants import PANDA_SCHEMA
0028
0029 logger = logging.getLogger(__name__)
0030
0031 SITE_PREFIXES = ('NERSC',)
0032 # The first pass reaches the oldest row PanDA still holds.
0033 DEFAULT_DAYS = 100
0034 TYPED = ('computingsite', 'status', 'batchid', 'ncore',
0035          'submittime', 'starttime', 'endtime', 'lastupdate')
0036
0037
0038 def _aware(t):
0039     if t is None:
0040         return None
0041     return t.replace(tzinfo=dt_timezone.utc) if t.tzinfo is None else t
0042
0043
0044 def _jsonable(v):
0045     if hasattr(v, 'isoformat'):
0046         return _aware(v).isoformat()
0047     if isinstance(v, (bytes, memoryview)):
0048         return None
0049     return v
0050
0051
0052 def capture(days=DEFAULT_DAYS, prefixes=SITE_PREFIXES):
0053     """Copy the worker rows updated in the last ``days`` at the sites.
0054     Returns counts: read, created, updated, unchanged."""
0055     from .models import HarvesterWorkerRecord
0056
0057     since = timezone.now() - timedelta(days=days)
0058     like = ' OR '.join('"computingsite" LIKE %s' for _ in prefixes)
0059     with connections['panda'].cursor() as cursor:
0060         cursor.execute(
0061             f'SELECT * FROM "{PANDA_SCHEMA}"."harvester_workers" '
0062             f'WHERE "lastupdate" > %s AND ({like})',
0063             [since.replace(tzinfo=None)] + [p + '%' for p in prefixes])
0064         cols = [c[0].lower() for c in cursor.description]
0065         rows = [dict(zip(cols, r)) for r in cursor.fetchall()]
0066     have = {(h, w): lu for h, w, lu in HarvesterWorkerRecord.objects
0067             .filter(computingsite__regex=r'^(' + '|'.join(prefixes) + ')')
0068             .values_list('harvesterid', 'workerid', 'lastupdate')}
0069     counts = {'read': len(rows), 'created': 0, 'updated': 0, 'unchanged': 0}
0070     for row in rows:
0071         key = (row['harvesterid'], int(row['workerid']))
0072         lastupdate = _aware(row.get('lastupdate'))
0073         if key in have and have[key] is not None and lastupdate is not None \
0074                 and have[key] >= lastupdate:
0075             counts['unchanged'] += 1
0076             continue
0077         fields = {k: (_aware(row.get(k)) if k.endswith('time') or k == 'lastupdate'
0078                       else row.get(k)) for k in TYPED}
0079         fields['status'] = fields['status'] or ''
0080         fields['batchid'] = str(fields['batchid'] or '')
0081         fields['record'] = {k: _jsonable(v) for k, v in row.items() if k != 'jdl'}
0082         _, created = HarvesterWorkerRecord.objects.update_or_create(
0083             harvesterid=key[0], workerid=key[1], defaults=fields)
0084         counts['created' if created else 'updated'] += 1
0085     return counts
0086
0087
0088 def worker_row(harvesterid, workerid):
0089     """Our copy of one worker row in the shape PanDA's has (naive UTC
0090     datetimes), or None."""
0091     from .models import HarvesterWorkerRecord
0092     rec = HarvesterWorkerRecord.objects.filter(
0093         harvesterid=harvesterid, workerid=int(workerid)).first()
0094     if rec is None:
0095         return None
0096     out = dict(rec.record or {})
0097     for k in ('submittime', 'starttime', 'endtime', 'lastupdate'):
0098         t = getattr(rec, k)
0099         out[k] = t.astimezone(dt_timezone.utc).replace(tzinfo=None) if t else None
0100     out['ncore'] = rec.ncore
0101     out['from_copy'] = True
0102     return out