File indexing completed on 2026-10-20 10:15:25
0001 """The declared record: downtime and status declared for the ePIC queues,
0002 storage endpoints and sites, as read from CRIC (swf-epicprod
0003 docs/CONTINUOUS_PRODUCTION.md, Declared downtime).
0004
0005 Three CRIC records carry declared state: PanDA queue status rules
0006 (``pandaqueuestatus``: OFFLINE, BROKEROFF or TEST per queue and activity,
0007 with an expiration, a reason and who set it), DDM endpoint status rules
0008 (``ddmendpointstatus``: the same per endpoint and activity) and downtime
0009 objects (``downtime``: a window with a start, an end, a severity and the
0010 services it takes down). The collector reads the three with the
0011 production proxy and keeps one entry per rule or window in the entry
0012 store (kind ``declared``, context ``cric``), named by its source
0013 identity, with a standing: active, future, expired or cleared. Every
0014 change is a version of the entry; the sync's action names what appeared,
0015 expired and cleared.
0016
0017 The readers here are pure: ``rules_from_pandaqueuestatus``,
0018 ``rules_from_ddmendpointstatus`` and ``windows_from_downtime`` turn the
0019 CRIC documents into records, ``standing_of`` places a record in time,
0020 ``sync`` writes the store, ``declared_for`` and ``history_for`` serve the
0021 pages and the readers (attribution, node guard, canary, the front).
0022 """
0023 import logging
0024 from datetime import datetime, timedelta, timezone as dt_timezone
0025
0026 logger = logging.getLogger(__name__)
0027
0028 DECLARED_KIND = 'declared'
0029 DECLARED_CONTEXT = 'cric'
0030 STANDINGS = ('active', 'future', 'expired', 'cleared')
0031
0032
0033
0034 CACHE_LAG = timedelta(minutes=10)
0035
0036 CRIC_BASE = 'https://datalake-cric.cern.ch/api'
0037 CRIC_QUEUE_STATUS = f'{CRIC_BASE}/atlas/pandaqueuestatus/query/?json&showall=1'
0038 CRIC_ENDPOINT_STATUS = f'{CRIC_BASE}/atlas/ddmendpointstatus/query/?json&showall=1'
0039 CRIC_DOWNTIME = f'{CRIC_BASE}/core/downtime/query/?json'
0040 CRIC_QUEUES = f'{CRIC_BASE}/atlas/pandaqueue/query/?json&vo_name=eic'
0041
0042
0043 def _dt(text):
0044 """A CRIC timestamp (naive UTC, '2026-09-14T23:21:00' or with
0045 microseconds) as an aware datetime; None for nothing."""
0046 if not text:
0047 return None
0048 try:
0049 dt = datetime.fromisoformat(str(text).replace('Z', '+00:00'))
0050 except ValueError:
0051 return None
0052 if dt.tzinfo is None:
0053 dt = dt.replace(tzinfo=dt_timezone.utc)
0054 return dt
0055
0056
0057 def _iso(dt):
0058 return dt.isoformat(timespec='seconds') if dt else None
0059
0060
0061
0062
0063 def rules_from_pandaqueuestatus(doc, queues):
0064 """The status rules of ``queues`` in a ``pandaqueuestatus`` document:
0065 ``{queue: {activity: {'status': {...}, 'mode': {VALUE: {probe: {...}}}}}}``.
0066 One record per queue, activity, mode and probe; the source identity
0067 is that tuple, so a rule re-set with a new expiration is the same
0068 entry with a new version."""
0069 out = []
0070 for queue, activities in (doc or {}).items():
0071 if queue not in queues:
0072 continue
0073 for activity, rec in (activities or {}).items():
0074 modes = (rec or {}).get('mode') or {}
0075 for value, probes in modes.items():
0076 for probe, r in (probes or {}).items():
0077 out.append({
0078 'name': f'queue:{queue}:{activity}:{value}:{probe}',
0079 'kind': 'queue', 'target': queue,
0080 'source': 'cric:pandaqueuestatus',
0081 'value': value, 'activity': activity, 'probe': probe,
0082 'reason': (r or {}).get('reason') or '',
0083 'declared_by': (r or {}).get('operationdn') or '',
0084 'declared_at': _iso(_dt((r or {}).get('updated'))),
0085 'start': None,
0086 'end': _iso(_dt((r or {}).get('expiration'))),
0087 'info_url': '',
0088 'severity': '', 'classification': '', 'description': '',
0089 })
0090 return out
0091
0092
0093 def rules_from_ddmendpointstatus(doc, endpoints):
0094 """The status rules of ``endpoints`` in a ``ddmendpointstatus``
0095 document, the same shape as the queue rules."""
0096 out = []
0097 for endpoint, activities in (doc or {}).items():
0098 if endpoint not in endpoints:
0099 continue
0100 for activity, rec in (activities or {}).items():
0101 modes = (rec or {}).get('mode') or {}
0102 for value, probes in modes.items():
0103 for probe, r in (probes or {}).items():
0104 out.append({
0105 'name': f'endpoint:{endpoint}:{activity}:{value}:{probe}',
0106 'kind': 'endpoint', 'target': endpoint,
0107 'source': 'cric:ddmendpointstatus',
0108 'value': value, 'activity': activity, 'probe': probe,
0109 'reason': (r or {}).get('reason') or '',
0110 'declared_by': (r or {}).get('operationdn') or '',
0111 'declared_at': _iso(_dt((r or {}).get('updated'))),
0112 'start': None,
0113 'end': _iso(_dt((r or {}).get('expiration'))),
0114 'info_url': '',
0115 'severity': '', 'classification': '', 'description': '',
0116 })
0117 return out
0118
0119
0120 def windows_from_downtime(doc, rcsites):
0121 """The downtime windows of ``rcsites`` in a ``downtime`` document (a
0122 dict or list of downtime objects). One record per downtime, on the
0123 resource centre; the affected services ride along. ``rcsites`` is a
0124 set of names, or a mapping of name to the queues behind it, which
0125 the record then carries as ``queues`` so a queue's readers find the
0126 site's windows."""
0127 items = list(doc.values()) if isinstance(doc, dict) else list(doc or [])
0128 queues_of = rcsites if isinstance(rcsites, dict) else {}
0129 out = []
0130 for d in items:
0131 if not isinstance(d, dict):
0132 continue
0133 rc = d.get('rc_site') or ''
0134 if rc not in rcsites:
0135 continue
0136 ident = d.get('id') or d.get('pid') or f"{rc}:{d.get('start_time')}"
0137 out.append({
0138 'name': f'site:{rc}:{ident}',
0139 'kind': 'site', 'target': rc,
0140 'queues': sorted(queues_of.get(rc) or []),
0141 'source': 'cric:downtime',
0142 'value': str(d.get('severity') or 'OUTAGE').upper(),
0143 'activity': '', 'probe': d.get('provider') or '',
0144 'reason': d.get('description') or '',
0145 'declared_by': d.get('provider') or '',
0146 'declared_at': _iso(_dt(d.get('create_time'))),
0147 'start': _iso(_dt(d.get('start_time'))),
0148 'end': _iso(_dt(d.get('end_time'))),
0149 'info_url': d.get('info_url') or '',
0150 'severity': str(d.get('severity') or ''),
0151 'classification': str(d.get('classification') or ''),
0152 'description': d.get('description') or '',
0153 'services': [s if isinstance(s, str) else (s or {}).get('name', '')
0154 for s in (d.get('affected_services') or d.get('services') or [])],
0155 })
0156 return out
0157
0158
0159 def standing_of(record, now):
0160 """Where a declared record stands at ``now``: future (a start not yet
0161 reached), active (in force), expired (its end passed). A rule with no
0162 start acts from the moment it is set; a record with no end holds
0163 until CRIC drops it (cleared, which only the sync can say)."""
0164 start = _dt(record.get('start'))
0165 end = _dt(record.get('end'))
0166 if start and now < start:
0167 return 'future'
0168 if end and now >= end:
0169 return 'expired'
0170 return 'active'
0171
0172
0173 def _concerns(record, kind, target):
0174 """Whether a record is about ``target`` of ``kind``: its own target,
0175 or, for a queue, a site window whose ``queues`` name it."""
0176 if kind and record.get('kind') != kind:
0177 if not (kind == 'queue' and record.get('kind') == 'site'
0178 and target and target in (record.get('queues') or [])):
0179 return False
0180 return True
0181 if target and record.get('target') != target:
0182 return False
0183 return True
0184
0185
0186 def declared_at(records, when, kind=None, target=None, lag=CACHE_LAG):
0187 """The records that covered ``when`` (an aware datetime): in force
0188 then, or ended within ``lag`` before it. Pure; for the attribution
0189 of a failure to a declaration."""
0190 out = []
0191 for r in records:
0192 if not _concerns(r, kind, target):
0193 continue
0194 start = _dt(r.get('start')) or _dt(r.get('declared_at'))
0195 end = _dt(r.get('end'))
0196 cleared = _dt(r.get('cleared_at'))
0197 stop = min(t for t in (end, cleared) if t) if (end or cleared) else None
0198 if start and when < start:
0199 continue
0200 if stop and when >= stop + lag:
0201 continue
0202 out.append(r)
0203 return out
0204
0205
0206
0207
0208 def _context():
0209 from .models import EntryContext
0210 ctx, _ = EntryContext.objects.get_or_create(
0211 name=DECLARED_CONTEXT,
0212 defaults={'title': 'Declared state from CRIC',
0213 'description': 'Downtime and status declared for the ePIC queues, '
0214 'endpoints and sites, one entry per CRIC rule or '
0215 'window (CONTINUOUS_PRODUCTION.md, Declared downtime).'})
0216 return ctx
0217
0218
0219 def entries():
0220 """Every declared entry as stored, not archived."""
0221 from .models import Entry
0222 return list(Entry.objects.filter(kind=DECLARED_KIND, context_id=DECLARED_CONTEXT,
0223 archived=False, deleted_at__isnull=True))
0224
0225
0226 def records():
0227 """The declared records: each entry's data with its standing and name."""
0228 out = []
0229 for e in entries():
0230 d = dict(e.data or {})
0231 d['name'] = e.name
0232 d['standing'] = e.status or ''
0233 out.append(d)
0234 return out
0235
0236
0237 _RECORDS_TTL_S = 60.0
0238 _cache = {'records': None, 'at': 0.0}
0239
0240
0241 def cached_records():
0242 """The declared records, read at most once a minute: the job error
0243 root and the job lists read them per job. A failed read logs and
0244 serves the last copy (or nothing), never raises into a page."""
0245 import time as time_mod
0246 now = time_mod.monotonic()
0247 if _cache['records'] is None or now - _cache['at'] > _RECORDS_TTL_S:
0248 try:
0249 _cache['records'] = records()
0250 _cache['at'] = now
0251 except Exception as e:
0252 logger.error('declared records read failed: %s', e)
0253 return _cache['records'] or []
0254 return _cache['records']
0255
0256
0257
0258
0259 def job_declared(computingsite, endtime):
0260 """The declaration a failed job fell under, for the per-job error
0261 root: the queue's rule or its site's window in force at the job's
0262 end, or ended within CACHE_LAG before it (the pilot's cached
0263 queuedata lags a rule's expiration). Returns ``{'label', 'line',
0264 'queue', 'record', 'grade'}`` or None."""
0265 when = _dt(endtime)
0266 if not computingsite or when is None:
0267 return None
0268 hits = declared_at(cached_records(), when, kind='queue', target=str(computingsite))
0269 if not hits:
0270 return None
0271 r = hits[0]
0272 return {'label': 'declared downtime', 'line': summary_line(r),
0273 'queue': str(computingsite), 'record': r.get('name', ''),
0274 'grade': 'declared record (CRIC)'}
0275
0276
0277 def windows_between(after, before, kind='queue', lag=CACHE_LAG):
0278 """The (target, start, stop) spans of ``kind`` that overlap
0279 [after, before): a rule from its declared instant to its end or
0280 clearing plus the lag, a window from its start; for the summary's
0281 count of failures under declaration. Site windows are given per
0282 queue."""
0283 out = []
0284 for r in cached_records():
0285 targets = []
0286 if r.get('kind') == kind:
0287 targets = [r['target']]
0288 elif kind == 'queue' and r.get('kind') == 'site':
0289 targets = list(r.get('queues') or [])
0290 if not targets:
0291 continue
0292 start = _dt(r.get('start')) or _dt(r.get('declared_at'))
0293 end, cleared = _dt(r.get('end')), _dt(r.get('cleared_at'))
0294 stop = min(t for t in (end, cleared) if t) if (end or cleared) else None
0295 stop = stop + lag if stop else None
0296 if start is None:
0297 continue
0298 if before is not None and start >= before:
0299 continue
0300 if stop is not None and after is not None and stop <= after:
0301 continue
0302 for target in targets:
0303 out.append((target, start, stop, summary_line(r)))
0304 return out
0305
0306
0307 def set_aside_declared(rows, records=None, lag=CACHE_LAG):
0308 """The node guard's rows with those that ended under a declared
0309 downtime of their queue (or the cache lag after it) set aside: a
0310 node is not judged on a downtime's deaths. Rows carry ``queue`` and
0311 ``endtime`` (ISO). Returns (kept, set_aside_by_queue); pure over
0312 ``records`` (the cached records when None)."""
0313 records = cached_records() if records is None else records
0314 by_queue = {}
0315 if not records:
0316 return list(rows), by_queue
0317 kept = []
0318 for row in rows:
0319 when = _dt(row.get('endtime'))
0320 queue = row.get('queue')
0321 if when is not None and queue and declared_at(records, when, kind='queue',
0322 target=str(queue), lag=lag):
0323 by_queue[str(queue)] = by_queue.get(str(queue), 0) + 1
0324 continue
0325 kept.append(row)
0326 return kept, by_queue
0327
0328
0329 def canary_declared(queue_names, now, lag=CACHE_LAG):
0330 """What the site canary reads per queue (site-canary
0331 canary/declared.py, over GET /api/declared/): the rule in force now
0332 with its line, the spans it was declared over (start to end or
0333 clearing plus the cache lag) and the latest span end. ``queue_names``
0334 None means every queue on the record."""
0335 out = {}
0336 names = set(queue_names) if queue_names is not None else None
0337 for r in cached_records():
0338 if r.get('kind') == 'queue':
0339 targets = [r['target']]
0340 elif r.get('kind') == 'site':
0341 targets = list(r.get('queues') or [])
0342 else:
0343 continue
0344 if names is not None:
0345 targets = [t for t in targets if t in names]
0346 if not targets:
0347 continue
0348 start = _dt(r.get('start')) or _dt(r.get('declared_at'))
0349 end, cleared = _dt(r.get('end')), _dt(r.get('cleared_at'))
0350 stop = min(t for t in (end, cleared) if t) if (end or cleared) else None
0351 stop_lagged = stop + lag if stop else None
0352 in_force = (r.get('standing') != 'cleared' and standing_of(r, now) == 'active')
0353 for target in targets:
0354 slot = out.setdefault(target, {'in_force': '', 'spans': [], 'last_end': None})
0355 if in_force and not slot['in_force']:
0356 slot['in_force'] = summary_line(r)
0357 if start is not None:
0358 slot['spans'].append((_iso(start), _iso(stop_lagged)))
0359 if stop_lagged is not None and (slot['last_end'] is None
0360 or _iso(stop_lagged) > slot['last_end']):
0361 slot['last_end'] = _iso(stop_lagged)
0362 return out
0363
0364
0365 def gate_for_queue(queue, horizon_h, now=None):
0366 """The front's declared gate for one queue: red with the rule in
0367 force, or a window starting within ``horizon_h`` hours; the reason is
0368 the record's line. Pure over the cached records."""
0369 from django.utils import timezone
0370 now = now or timezone.now()
0371 horizon = now + timedelta(hours=float(horizon_h or 0))
0372 in_force, coming = [], []
0373 for r in cached_records():
0374 if r.get('standing') == 'cleared' or not _concerns(r, 'queue', queue):
0375 continue
0376 standing = standing_of(r, now)
0377 if standing == 'active':
0378 in_force.append(r)
0379 elif standing == 'future':
0380 start = _dt(r.get('start'))
0381 if start and start <= horizon:
0382 coming.append(r)
0383 in_force.sort(key=lambda r: r.get('end') or '')
0384 coming.sort(key=lambda r: r.get('start') or '')
0385 if in_force:
0386 return {'red': True, 'state': 'in_force', 'reason': summary_line(in_force[0])}
0387 if coming:
0388 return {'red': True, 'state': 'coming', 'reason': summary_line(coming[0])}
0389 return {'red': False, 'state': '', 'reason': ''}
0390
0391
0392 def sync(found, now, changed_by='cric_declared_state'):
0393 """Write what CRIC holds now onto the store: a record seen for the
0394 first time is created with its standing; a known one takes its new
0395 fields and standing; a record CRIC no longer returns is ``cleared``
0396 (with the instant) if it was in force or future, ``expired`` if its
0397 end had passed. Returns the changes: appeared, changed, expired,
0398 cleared, unchanged (counts), and the names in each."""
0399 import time as time_mod
0400 from django.db import transaction
0401 from .models import Entry
0402 from .signals import set_changed_by
0403 set_changed_by(changed_by)
0404 changes = {'appeared': [], 'changed': [], 'expired': [], 'cleared': [], 'unchanged': 0}
0405 with transaction.atomic():
0406 ctx = _context()
0407 known = {e.name: e for e in Entry.objects.select_for_update().filter(
0408 kind=DECLARED_KIND, context=ctx, archived=False, deleted_at__isnull=True)}
0409 seen = set()
0410 for r in found:
0411 name = r['name']
0412 seen.add(name)
0413 standing = standing_of(r, now)
0414 e = known.get(name)
0415 data = {k: v for k, v in r.items() if k != 'name'}
0416 data['observed_at'] = _iso(now)
0417 if e is None:
0418 data['first_seen'] = _iso(now)
0419 e = Entry(kind=DECLARED_KIND, context=ctx, name=name, status=standing,
0420 title=summary_line(r), data=data)
0421 e.save()
0422 changes['appeared'].append(name)
0423 continue
0424 before = dict(e.data or {})
0425 data['first_seen'] = before.get('first_seen') or _iso(now)
0426 if before.get('cleared_at'):
0427
0428 data.pop('cleared_at', None)
0429 comparable = {k: v for k, v in data.items() if k not in ('observed_at',)}
0430 before_comparable = {k: v for k, v in before.items() if k not in ('observed_at',)}
0431 if comparable == before_comparable and (e.status or '') == standing:
0432 changes['unchanged'] += 1
0433 continue
0434 if standing == 'expired' and (e.status or '') != 'expired':
0435 changes['expired'].append(name)
0436 else:
0437 changes['changed'].append(name)
0438 e.data = data
0439 e.status = standing
0440 e.title = summary_line(r)
0441 e.timestamp_modified = time_mod.time()
0442 e.save()
0443 for name, e in known.items():
0444 if name in seen or (e.status or '') in ('expired', 'cleared'):
0445 continue
0446 data = dict(e.data or {})
0447 end = _dt(data.get('end'))
0448 if end and now >= end:
0449 e.status = 'expired'
0450 changes['expired'].append(name)
0451 else:
0452 e.status = 'cleared'
0453 data['cleared_at'] = _iso(now)
0454 changes['cleared'].append(name)
0455 e.data = data
0456 e.timestamp_modified = time_mod.time()
0457 e.save()
0458 return changes
0459
0460
0461
0462
0463 def _fmt_when(text):
0464 dt = _dt(text)
0465 return dt.strftime('%m/%d %H:%M UTC') if dt else ''
0466
0467
0468 def summary_line(r):
0469 """One line for a queue line or an endpoint line: the rule or window
0470 as an operator reads it. 'offline until 09/15 00:21 UTC: scheduled
0471 downtime (xzhao@bnl.gov)'; 'downtime 09/20 08:00 to 16:00 UTC:
0472 <description>'."""
0473 who = r.get('declared_by') or ''
0474 reason = (r.get('reason') or r.get('description') or '').strip()
0475 tail = (f': {reason}' if reason else '') + (f' ({who})' if who else '')
0476 if r.get('kind') == 'site':
0477 span = _fmt_when(r.get('start'))
0478 if r.get('end'):
0479 span += f" to {_fmt_when(r.get('end'))}"
0480 return f"{(r.get('value') or 'downtime').lower()} {span}{tail}".strip()
0481 value = (r.get('value') or '').lower()
0482 activity = r.get('activity') or ''
0483 act = f" ({activity})" if activity and activity != 'a' else ''
0484 until = f" until {_fmt_when(r.get('end'))}" if r.get('end') else ''
0485 return f"{value}{act}{until}{tail}".strip()
0486
0487
0488 def declared_for(kind, now=None):
0489 """What stands declared per target of ``kind`` (queue, endpoint,
0490 site) at ``now``: ``{target: {'active': [records], 'future':
0491 [records], 'line': text}}`` with the line the list pages show: the
0492 rule in force, else the next future window, else ''. The standing is
0493 recomputed at read time so a rule that expired between two syncs
0494 reads expired."""
0495 from django.utils import timezone
0496 now = now or timezone.now()
0497 out = {}
0498 for r in cached_records():
0499 if r.get('standing') == 'cleared':
0500 continue
0501 if r.get('kind') == kind:
0502 targets = [r['target']]
0503 elif kind == 'queue' and r.get('kind') == 'site':
0504
0505 targets = list(r.get('queues') or [])
0506 else:
0507 continue
0508 standing = standing_of(r, now)
0509 if standing == 'expired':
0510 continue
0511 for target in targets:
0512 slot = out.setdefault(target, {'active': [], 'future': [], 'line': ''})
0513 slot[standing].append(r)
0514 for target, slot in out.items():
0515 slot['active'].sort(key=lambda r: r.get('end') or '')
0516 slot['future'].sort(key=lambda r: r.get('start') or '')
0517 first = (slot['active'] or slot['future'] or [None])[0]
0518 slot['line'] = summary_line(first) if first else ''
0519 return out
0520
0521
0522 def history_for(kind, target, limit=20):
0523 """Every declared record of one target, newest first by its declared
0524 instant, with standing; for the detail pages."""
0525 rows = [r for r in records() if r.get('kind') == kind and r.get('target') == target]
0526 rows.sort(key=lambda r: r.get('declared_at') or r.get('start') or '', reverse=True)
0527 return rows[:limit]
0528
0529
0530 STANDING_CLASS = {'active': 'offline_fill', 'future': 'pending_fill'}
0531
0532
0533 def for_detail_page(kind, target, now=None):
0534 """What a detail page shows for one target: ``now`` (the active and
0535 future records with their lines) and ``history`` (every record, with
0536 its line, its declared instant and its standing as the page words
0537 them). One read of the store."""
0538 from django.utils import timezone
0539 now = now or timezone.now()
0540 current = {'active': [], 'future': [], 'line': ''}
0541 history = []
0542 for r in history_for(kind, target, limit=50):
0543 r = dict(r)
0544 r['line'] = summary_line(r)
0545 r['when'] = _fmt_when(r.get('declared_at') or r.get('start'))
0546 r['cleared_when'] = _fmt_when(r.get('cleared_at')) if r.get('cleared_at') else ''
0547 standing = r.get('standing') if r.get('standing') == 'cleared' else standing_of(r, now)
0548 r['standing'] = standing
0549 r['standing_class'] = STANDING_CLASS.get(standing, '')
0550 history.append(r)
0551 if standing in ('active', 'future'):
0552 current[standing].append(r)
0553 current['active'].sort(key=lambda r: r.get('end') or '')
0554 current['future'].sort(key=lambda r: r.get('start') or '')
0555 first = (current['active'] or current['future'] or [None])[0]
0556 current['line'] = first['line'] if first else ''
0557 return current, history
