swf-monitor/scripts/node-measure-ingest.py

0001 #!/usr/bin/env python3
0002 """node-measure-ingest.py — fold finished jobs' measures into the node
0003 measurement store (site-canary docs/MEASUREMENTS.md).
0004
0005 Every finished ePIC production job leaves a payload report with what
0006 prmon measured per stage and the events it produced; its PanDA record
0007 says where it ran. This reads the jobs finished since its cursor, resolves
0008 each job's workload through PCS, computes per stage the measures the store
0009 keeps, and folds them into the running distributions of that queue,
0010 processor and workload. The cursor is the newest job modification time
0011 folded, kept in a state file beside the storage store, so a job is folded
0012 once.
0013
0014 The ePIC-specific part is here: which report fields mean what, and how
0015 the workload keys are found. The store and its fold know nothing of PCS
0016 (canary.store.measure).
0017
0018 Django-bootstrap standalone script — also usable by hand::
0019
0020     cd /data/wenauseic/github/swf-monitor/src
0021     ../../swf-testbed/.venv/bin/python ../scripts/node-measure-ingest.py \
0022         [--hours 168] [--limit 5000] [--dry-run] [--reset]
0023
0024 The last stdout line is a JSON summary; progress goes to stderr.
0025 Exit codes: 0 ok · 6 the PanDA record could not be read.
0026 """
0027 import argparse
0028 import json
0029 import os
0030 import sys
0031 from datetime import datetime, timedelta, timezone as dt_timezone
0032
0033 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0034 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0035 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0036
0037 import django  # noqa: E402
0038 django.setup()
0039
0040 from django.db import connections  # noqa: E402
0041
0042 from monitor_app.panda.constants import PANDA_SCHEMA  # noqa: E402
0043
0044 STATE_PATH = os.environ.get('NODE_MEASURE_STATE',
0045                             '/data/wenauseic/swf-delivery/node-measure-state.json')
0046 DEFAULT_HOURS = 168
0047 DEFAULT_LIMIT = 5000
0048 # The stages whose event count is the simulated one; every other stage's
0049 # per-event measure divides by the events the job delivered.
0050 SIMULATION_STAGES = ('simulation', 'background', 'evgen', 'input', 'geometry')
0051
0052
0053 def _log(msg):
0054     print(msg, file=sys.stderr, flush=True)
0055
0056
0057 def load_state():
0058     try:
0059         with open(STATE_PATH) as handle:
0060             return json.load(handle) or {}
0061     except FileNotFoundError:
0062         return {}
0063     except (OSError, ValueError) as e:
0064         _log(f'WARNING: state at {STATE_PATH} unreadable: {e}')
0065         return {}
0066
0067
0068 def save_state(state):
0069     try:
0070         tmp = STATE_PATH + '.tmp'
0071         with open(tmp, 'w') as handle:
0072             json.dump(state, handle, indent=1, sort_keys=True, default=str)
0073         os.replace(tmp, STATE_PATH)
0074     except OSError as e:
0075         _log(f'ERROR: state not written to {STATE_PATH}: {e}')
0076
0077
0078 def finished_jobs(since, limit):
0079     """Finished production jobs with a report, oldest first:
0080     (pandaid, jeditaskid, queue, processor, cores, modificationtime,
0081     container, report)."""
0082     sql = f"""
0083         SELECT j."pandaid", j."jeditaskid", j."computingsite",
0084                COALESCE(j."cpuconsumptionunit", 'unknown'),
0085                GREATEST(COALESCE(j."actualcorecount", j."corecount", 1), 1),
0086                j."modificationtime", COALESCE(j."container_name", ''),
0087                m."metadata"
0088         FROM "{PANDA_SCHEMA}"."jobsarchived4" j
0089         JOIN "{PANDA_SCHEMA}"."metatable" m ON m."pandaid" = j."pandaid"
0090         WHERE j."jobstatus" = 'finished' AND j."processingtype" = 'epicproduction'
0091           AND j."modificationtime" > %s
0092         ORDER BY j."modificationtime"
0093         LIMIT %s
0094     """
0095     rows = []
0096     with connections['panda'].cursor() as cursor:
0097         cursor.execute(sql, [since, int(limit)])
0098         for (pandaid, jeditaskid, queue, processor, cores, modtime,
0099              container, raw) in cursor.fetchall():
0100             try:
0101                 metadata = json.loads(raw) if isinstance(raw, str) else raw
0102             except (ValueError, TypeError):
0103                 metadata = None
0104             report = (metadata or {}).get('payload') if isinstance(metadata, dict) else None
0105             # PanDA times are naive UTC; the cursor and the store are aware.
0106             if modtime is not None and modtime.tzinfo is None:
0107                 modtime = modtime.replace(tzinfo=dt_timezone.utc)
0108             rows.append((int(pandaid), jeditaskid, queue, processor, int(cores),
0109                          modtime, container, report))
0110     return rows
0111
0112
0113 def stage_measures(report, cores):
0114     """Per stage, the measures the store keeps, from the report's prmon
0115     summaries and event counts. A stage with no events to divide by keeps
0116     its efficiency and memory and no per-event figure."""
0117     prmon = report.get('prmon') or {}
0118     events = report.get('events') or {}
0119     simulated = events.get('simulated')
0120     delivered = events.get('reconstructed') or simulated
0121     out = {}
0122     for stage, rec in prmon.items():
0123         if not isinstance(rec, dict) or 'error' in rec:
0124             continue
0125         wall = rec.get('wall_s')
0126         cpu = (rec.get('cpu_user_s') or 0) + (rec.get('cpu_sys_s') or 0)
0127         count = simulated if stage in SIMULATION_STAGES else delivered
0128         measures = {}
0129         if count and wall:
0130             measures['cpu_s_per_event'] = cpu / count
0131             measures['wall_s_per_event'] = wall / count
0132         if wall:
0133             measures['cpu_efficiency'] = cpu / (wall * max(int(cores or 1), 1))
0134         if rec.get('rss_max_kb') is not None:
0135             measures['rss_max_kb'] = rec['rss_max_kb']
0136         if measures:
0137             out[stage] = measures
0138     return out
0139
0140
0141 class Workloads:
0142     """The workload keys of a task through PCS, once per task."""
0143
0144     def __init__(self):
0145         self._cache = {}
0146
0147     def for_task(self, jeditaskid):
0148         if jeditaskid in self._cache:
0149             return self._cache[jeditaskid]
0150         from pcs.models import PandaTasks
0151         row = (PandaTasks.objects.filter(jedi_task_id=jeditaskid)
0152                .select_related('prod_task__dataset__physics_config').first())
0153         keys = None
0154         if row is not None:
0155             dataset = row.prod_task.dataset if row.prod_task_id else None
0156             keys = {
0157                 'detector_version': (dataset.detector_version if dataset else '') or '',
0158                 'physics_config': ((dataset.physics_config.label
0159                                     if dataset is not None and dataset.physics_config_id
0160                                     else '') or ''),
0161             }
0162         self._cache[jeditaskid] = keys
0163         return keys
0164
0165
0166 def main():
0167     parser = argparse.ArgumentParser()
0168     parser.add_argument('--hours', type=float, default=DEFAULT_HOURS,
0169                         help='the window when there is no cursor yet')
0170     parser.add_argument('--limit', type=int, default=DEFAULT_LIMIT,
0171                         help='jobs per batch; batches run until one comes back short')
0172     parser.add_argument('--dry-run', action='store_true',
0173                         help='read and compute; fold nothing, move no cursor')
0174     parser.add_argument('--reset', action='store_true',
0175                         help='forget the cursor and start from --hours ago')
0176     args = parser.parse_args()
0177
0178     from canary.store import measure
0179
0180     state = {} if args.reset else load_state()
0181     cursor = state.get('cursor')
0182     if cursor:
0183         since = datetime.fromisoformat(cursor)
0184     else:
0185         since = datetime.now(dt_timezone.utc) - timedelta(hours=args.hours)
0186     summary = {'since': since.isoformat(), 'jobs': 0, 'folded': 0, 'rows': 0,
0187                'no_report': 0, 'no_task': 0, 'no_measures': 0,
0188                'cursor': cursor, 'dry_run': bool(args.dry_run)}
0189     workloads = Workloads()
0190     newest = since
0191
0192     while True:
0193         try:
0194             batch = finished_jobs(since if newest == since else newest, args.limit)
0195         except Exception as e:                                # noqa: BLE001
0196             summary['error'] = f'the PanDA record could not be read: {e}'
0197             print(json.dumps(summary, default=str))
0198             return 6
0199         for pandaid, jeditaskid, queue, processor, cores, modtime, container, report in batch:
0200             summary['jobs'] += 1
0201             if modtime and modtime > newest:
0202                 newest = modtime
0203             if not isinstance(report, dict) or not report.get('prmon'):
0204                 summary['no_report'] += 1
0205                 continue
0206             keys = workloads.for_task(jeditaskid) if jeditaskid else None
0207             if keys is None:
0208                 summary['no_task'] += 1
0209                 continue
0210             stages = stage_measures(report, cores)
0211             if not stages:
0212                 summary['no_measures'] += 1
0213                 continue
0214             workload = dict(keys, container_image=container or '',
0215                             payload_version=str(report.get('payload_version') or ''))
0216             if args.dry_run:
0217                 summary['folded'] += 1
0218                 summary['rows'] += len(stages)
0219                 continue
0220             try:
0221                 summary['rows'] += measure.record_job(
0222                     queue_name=queue, processor=processor, workload=workload,
0223                     stages=stages, job_at=modtime)
0224                 summary['folded'] += 1
0225             except Exception as e:                            # noqa: BLE001
0226                 _log(f'ERROR: job {pandaid} not folded: {e}')
0227         if len(batch) < args.limit:
0228             break
0229         _log(f"{summary['jobs']} jobs read, cursor at {newest.isoformat()}")
0230
0231     if not args.dry_run and newest > since:
0232         state['cursor'] = newest.isoformat()
0233         state['last_run'] = datetime.now(dt_timezone.utc).isoformat()
0234         state['last_summary'] = {k: v for k, v in summary.items() if k != 'cursor'}
0235         save_state(state)
0236     summary['cursor'] = newest.isoformat()
0237     print(json.dumps(summary, default=str))
0238     return 0
0239
0240
0241 if __name__ == '__main__':
0242     sys.exit(main())