src/monitor_app/site_core_metrics.py

0001 """Core completions and queued core capacity for the production dashboard."""
0002 from collections import defaultdict
0003 from datetime import datetime, timedelta, timezone
0004
0005 from django.db import connections, transaction
0006
0007 from .cached_product import get_product
0008 from .models import SysConfig
0009 from .panda.census import NOT_STARTED
0010 from .panda.queries import PANDA_SCHEMA
0011
0012 KEY = 'site_core_metrics:v1'
0013 TTL = 60
0014 PRODUCTION = """workinggroup='EIC'
0015     AND (prodsourcelabel='managed' OR processingtype='epicproduction')
0016     AND COALESCE(processingtype,'') <> 'canary'"""
0017 CORES = """GREATEST(1, COALESCE(
0018     NULLIF(SUBSTRING(COALESCE(jobmetrics,'') FROM 'payloadThreads=([0-9]+)')::integer,0),
0019     NULLIF(actualcorecount,0), NULLIF(corecount,0), 1))"""
0020
0021
0022 def ready_capacity(jobs, nodes, cores_per_node):
0023     return min(max(0, jobs), max(0, nodes)) * cores_per_node
0024
0025
0026 def build():
0027     now = datetime.now(timezone.utc)
0028     since = (now - timedelta(hours=13)).replace(tzinfo=None)
0029     hour = (now - timedelta(hours=1)).replace(tzinfo=None)
0030     queues = defaultdict(lambda: {'finished_cores_1h': 0, 'failed_cores_1h': 0,
0031                                   'finished_jobs_1h': 0, 'failed_jobs_1h': 0,
0032                                   'queued_cores': 0, 'queued_jobs': 0})
0033     history = defaultdict(dict)
0034     fields = 'pandaid,computingsite,jobstatus,endtime,actualcorecount,corecount,jobmetrics,modificationtime'
0035     sql = f'''WITH jobs AS (
0036         SELECT {fields} FROM "{PANDA_SCHEMA}".jobsarchived4
0037         WHERE modificationtime >= %s AND endtime >= %s
0038           AND jobstatus IN ('finished','failed') AND {PRODUCTION}
0039         UNION ALL
0040         SELECT {fields} FROM "{PANDA_SCHEMA}".jobsactive4
0041         WHERE endtime >= %s AND jobstatus IN ('finished','failed') AND {PRODUCTION}
0042       ), unique_jobs AS (
0043         SELECT DISTINCT ON (pandaid) * FROM jobs ORDER BY pandaid,modificationtime DESC
0044       )
0045       SELECT computingsite,jobstatus,
0046         to_timestamp(CEIL(EXTRACT(EPOCH FROM endtime)/900)*900) AT TIME ZONE 'UTC',
0047         SUM({CORES}), COALESCE(SUM({CORES}) FILTER(WHERE endtime >= %s),0),
0048         COUNT(*) FILTER(WHERE endtime >= %s)
0049       FROM unique_jobs GROUP BY 1,2,3'''
0050     with transaction.atomic(using='panda'):
0051         with connections['panda'].cursor() as cursor:
0052             cursor.execute("SET LOCAL statement_timeout = '25s'")
0053             cursor.execute(sql, [since, since, since, hour, hour])
0054             ended = cursor.fetchall()
0055             cursor.execute(f'''SELECT computingsite,COUNT(*),SUM({CORES})
0056                 FROM "{PANDA_SCHEMA}".jobsactive4
0057                 WHERE jobstatus=ANY(%s) AND {PRODUCTION} GROUP BY 1''', [list(NOT_STARTED)])
0058             pending = cursor.fetchall()
0059     for queue, state, edge, cores, recent, jobs in ended:
0060         queues[queue][f'{state}_cores_1h'] += int(recent)
0061         queues[queue][f'{state}_jobs_1h'] += int(jobs)
0062         stamp = edge.replace(tzinfo=timezone.utc).isoformat()
0063         bucket = history[stamp].setdefault(queue, {'finished': 0, 'failed': 0})
0064         bucket[state] += int(cores)
0065     for queue, jobs, cores in pending:
0066         queues[queue].update(queued_jobs=int(jobs), queued_cores=int(cores))
0067     cfg = SysConfig.get_config()
0068     google = {'queue': cfg.get('gke_pilots.queue') or 'BNL_ePIC_GOOGLE_es',
0069               'nodes': int(cfg.get('gke_pilots.max_pods') or 0),
0070               'cores_per_node': int(float(cfg.get('gke_pilots.cpu') or 14))}
0071     queued = queues[google['queue']]
0072     queued['queued_cores'] = ready_capacity(queued['queued_jobs'], google['nodes'], google['cores_per_node'])
0073     queued['queued_core_demand'] = queued['queued_jobs'] * google['cores_per_node']
0074     # Perlmutter's whole-node jobs declare 256 cores (two threads a core);
0075     # each runs 128 work lanes, the unit its running figure counts, so a
0076     # waiting job counts 128 (Torre Wenaus, 2026-10-29).
0077     from swf_epicprod import perlmutter_production
0078     perl = queues[perlmutter_production.QUEUE]
0079     perl['queued_cores'] = perl['queued_jobs'] * int(perlmutter_production.SHAPE['slots'])
0080     return {'queues': dict(queues), 'ended': dict(history), 'google': google,
0081             'since': since.replace(tzinfo=timezone.utc).isoformat(), 'observed_at': now.isoformat()}
0082
0083
0084 def product(refresh=False):
0085     return get_product(KEY, build, TTL, refresh=refresh, async_first_fill=True)