File indexing completed on 2026-11-09 10:47:33
0001 """The Event Service reading of one queue: what the node harness reported
0002 across the queue's Event Service jobs over a window, from the job
0003 reports (swf-epicprod docs/NODE_EVENT_DISPATCHER.md, The record), as a
0004 cached product on the uniform mechanism (docs/CACHED_PRODUCTS.md): the
0005 queue detail page serves the stored reading and rebuilds behind the
0006 response.
0007
0008 The reading: jobs with a harness report and their slots, units done and
0009 failed with their events, the unit wall's middle and tail and the
0010 seconds an event costs on this queue, the closes with their events and
0011 wall, the jobs that stopped taking ranges at the deadline, and the
0012 events done per hour of job wall on the queue's Event Service jobs.
0013 """
0014 import json
0015 import logging
0016
0017 from django.db import connections
0018
0019 from .cached_product import get_product
0020 from .panda.constants import PANDA_SCHEMA
0021
0022 logger = logging.getLogger(__name__)
0023
0024 TTL_S = 300
0025 WINDOW_DAYS = 7
0026 JOB_CAP = 3000
0027
0028
0029 def _median(values):
0030 vals = sorted(v for v in values if v is not None)
0031 if not vals:
0032 return None
0033 mid = len(vals) // 2
0034 return round(vals[mid] if len(vals) % 2 else (vals[mid - 1] + vals[mid]) / 2, 1)
0035
0036
0037 def _reading(queue_name, days=WINDOW_DAYS):
0038 """The queue's Event Service reading over ``days``; pure database."""
0039 with connections['panda'].cursor() as cursor:
0040 cursor.execute(
0041 f'SELECT j."pandaid", j."jobstatus", j."jobsubstatus", '
0042 f'EXTRACT(EPOCH FROM (j."endtime" - j."starttime")), m."metadata" '
0043 f'FROM "{PANDA_SCHEMA}"."jobsarchived4" j '
0044 f'LEFT JOIN "{PANDA_SCHEMA}"."metatable" m ON m."pandaid" = j."pandaid" '
0045 f'WHERE j."computingsite" = %s AND j."eventservice" IN (1, 6) '
0046 f'AND j."modificationtime" > NOW() - INTERVAL %s '
0047 f'ORDER BY j."pandaid" DESC LIMIT %s',
0048 [queue_name, f'{int(days)} days', JOB_CAP])
0049 rows = cursor.fetchall()
0050 out = {'queue': queue_name, 'window_days': int(days), 'es_jobs': len(rows),
0051 'capped': len(rows) >= JOB_CAP, 'by_status': {}, 'by_substatus': {}}
0052 reports = 0
0053 slots = []
0054 unit_walls, unit_events, spe = [], [], []
0055 done = failed = events_done = 0
0056 closes = close_failed = close_events = 0
0057 close_walls = []
0058 deadline_marks = 0
0059 job_wall_s = 0.0
0060 for pandaid, status, substatus, wall, raw in rows:
0061 out['by_status'][status] = out['by_status'].get(status, 0) + 1
0062 if substatus:
0063 out['by_substatus'][substatus] = out['by_substatus'].get(substatus, 0) + 1
0064 if wall is not None:
0065 job_wall_s += float(wall)
0066 if not raw:
0067 continue
0068 try:
0069 metadata = json.loads(raw) if isinstance(raw, str) else raw
0070 except (ValueError, TypeError):
0071 continue
0072 es = metadata.get('es') if isinstance(metadata, dict) else None
0073 if not isinstance(es, dict):
0074 continue
0075 reports += 1
0076 if es.get('slots'):
0077 slots.append(int(es['slots']))
0078 if es.get('untaken_at_deadline'):
0079 deadline_marks += 1
0080 for u in es.get('ranges_done') or []:
0081 if not isinstance(u, dict):
0082 continue
0083 done += 1
0084 ev = int(u.get('events_reconstructed') or u.get('events') or 0)
0085 events_done += ev
0086 w = u.get('wall_s')
0087 if w is not None:
0088 unit_walls.append(float(w))
0089 unit_events.append(ev)
0090 if ev:
0091 spe.append(float(w) / ev)
0092 failed += len(es.get('ranges_failed') or [])
0093 for c in es.get('closes') or []:
0094 if not isinstance(c, dict):
0095 continue
0096 closes += 1
0097 if c.get('ok'):
0098 close_events += int(c.get('events') or 0)
0099 else:
0100 close_failed += 1
0101 if c.get('wall_s') is not None:
0102 close_walls.append(float(c['wall_s']))
0103 walls = sorted(unit_walls)
0104 out.update({
0105 'reports': reports,
0106 'slots_min': min(slots) if slots else None, 'slots_max': max(slots) if slots else None,
0107 'units_done': done, 'units_failed': failed, 'events_done': events_done,
0108 'unit_events_median': _median(unit_events),
0109 'unit_wall_median_s': _median(unit_walls),
0110 'unit_wall_p90_s': round(walls[min(len(walls) - 1, int(0.9 * len(walls)))], 1) if walls else None,
0111 's_per_event_median': round(_median(spe), 2) if spe else None,
0112 'closes': closes, 'closes_failed': close_failed, 'close_events': close_events,
0113 'close_wall_median_s': _median(close_walls),
0114 'deadline_marks': deadline_marks,
0115 'job_wall_h': round(job_wall_s / 3600.0, 2),
0116 'events_per_job_hour': round(events_done / (job_wall_s / 3600.0), 1) if job_wall_s else None,
0117 })
0118 return out
0119
0120
0121 def es_reading(queue_name, refresh=False):
0122 """One queue's Event Service reading with its build time; None when
0123 the queue ran no Event Service job in the window; an error record
0124 when the read fails."""
0125 key = 'es_reading:v1:{}'.format(queue_name)
0126 try:
0127 product = get_product(key, lambda: _reading(queue_name),
0128 ttl_seconds=TTL_S, refresh=refresh)
0129 except Exception as e:
0130 logger.error('event service reading failed for %s: %s', queue_name, e)
0131 return {'queue': queue_name, 'error': str(e)}
0132 value = dict((product or {}).get('value') or {})
0133 if not value.get('es_jobs'):
0134 return None
0135 value['built_at'] = (product or {}).get('built_at')
0136 value['age_seconds'] = (product or {}).get('age_seconds')
0137 return value
