src/monitor_app/task_resources.py

0001 """Twelve-hour observed job resource peaks, grouped by task and quarter hour."""
0002 from datetime import datetime, timedelta, timezone
0003
0004 from django.db import connections, transaction
0005
0006 from .cached_product import get_product
0007 from .panda.constants import PANDA_SCHEMA
0008
0009 PRODUCT_KEY = 'task_resource_peaks:v2'
0010 BIN_SECONDS = 900
0011
0012
0013 def assemble(rows, now):
0014     """Preserve missing telemetry as gaps; never substitute task resource requests."""
0015     first = int((now - timedelta(hours=12)).timestamp()) // BIN_SECONDS * BIN_SECONDS
0016     last = int(now.timestamp()) // BIN_SECONDS * BIN_SECONDS
0017     times = list(range(first, last + 1, BIN_SECONDS))
0018     tasks = {}
0019     scale_peaks = {'memory': 0, 'disk': 0, 'files': 0}
0020     for task_id, name, bucket, jobs, memory, disk, files, mem_n, disk_n, files_n, mem_scale, disk_scale, files_scale in rows:
0021         task = tasks.setdefault(int(task_id), {'id': int(task_id), 'name': name or str(task_id),
0022                     'memory': [None] * len(times), 'disk': [None] * len(times),
0023                     'files': [None] * len(times), 'jobs': [0] * len(times),
0024                     'samples': {key: [0] * len(times) for key in ('memory', 'disk', 'files')},
0025                     'coverage': {'memory': 0, 'disk': 0, 'files': 0}})
0026         i = (int(bucket) - first) // BIN_SECONDS
0027         if not 0 <= i < len(times):
0028             continue
0029         task['jobs'][i] = int(jobs)
0030         for metric, value, count in [('memory', memory, mem_n), ('disk', disk, disk_n), ('files', files, files_n)]:
0031             task[metric][i] = float(value) if value is not None else None
0032             task['samples'][metric][i] = int(count)
0033             task['coverage'][metric] += int(count)
0034         for metric, value in [('memory', mem_scale), ('disk', disk_scale), ('files', files_scale)]:
0035             if value is not None:
0036                 scale_peaks[metric] = max(scale_peaks[metric], float(value))
0037     return {'from': (now - timedelta(hours=12)).isoformat(),
0038             'through': now.isoformat(), 'times': [datetime.fromtimestamp(t, timezone.utc).isoformat() for t in times],
0039             'bin_minutes': 15, 'tasks': sorted(tasks.values(), key=lambda t: t['id']),
0040             'units': {'memory': 'GiB', 'disk': 'GiB', 'files': 'files'},
0041             'scale_peaks': scale_peaks,
0042             'basis': 'Production jobs only. Maximum reported job peak in each 15-minute observation/completion bin; running jobs use their latest update.'}
0043
0044
0045 def build():
0046     now = datetime.now(timezone.utc)
0047     start = now - timedelta(hours=12)
0048     bounds = [start.replace(tzinfo=None), now.replace(tzinfo=None)]
0049     fields = '''pandaid,jeditaskid,jobstatus,COALESCE(endtime,modificationtime) observed,
0050                 maxrss,jobmetrics'''
0051     metrics = ('memory', 'disk', 'files')
0052     quantile_fields = ','.join(
0053         f"percentile_cont(ARRAY[0.25,0.75]) WITHIN GROUP (ORDER BY {metric}) AS {metric}, COUNT({metric}) AS {metric}_n"
0054         for metric in metrics)
0055     # Bound scales using individual jobs, so a sparsely populated task has
0056     # the same per-job weight as every other production task. Keep full
0057     # bin maxima separately for curves, hover details and overflow markers.
0058     scale_fields = ','.join(
0059         f'''MAX(m.{metric}) FILTER (WHERE limits.{metric}_n < 4 OR m.{metric} <=
0060             CASE WHEN limits.{metric}[2] > limits.{metric}[1]
0061               THEN limits.{metric}[2] + 3*(limits.{metric}[2]-limits.{metric}[1])
0062               ELSE 3*limits.{metric}[2] END)'''
0063         for metric in metrics)
0064     sql = f'''WITH raw AS (
0065         SELECT {fields} FROM "{PANDA_SCHEMA}".jobsarchived4
0066           WHERE endtime >= %s AND endtime <= %s AND workinggroup='EIC'
0067             AND modificationtime >= %s
0068             AND prodsourcelabel='managed'
0069         UNION ALL
0070         SELECT {fields} FROM "{PANDA_SCHEMA}".jobsactive4
0071           WHERE modificationtime >= %s AND modificationtime <= %s AND workinggroup='EIC'
0072             AND prodsourcelabel='managed'
0073             AND jobstatus IN ('running','starting','holding','finished','failed','closed','cancelled')
0074       ), jobs AS (
0075         SELECT DISTINCT ON (pandaid) * FROM raw ORDER BY pandaid, observed DESC
0076       ), measured AS (
0077         SELECT jeditaskid, FLOOR(EXTRACT(EPOCH FROM observed)/900)::bigint*900 bucket,
0078           CASE WHEN maxrss>0 THEN maxrss/1048576.0 END memory,
0079           substring(jobmetrics from '(?:^|[[:space:]])scratchPeakMB=([0-9]+)')::double precision/1024 disk,
0080           substring(jobmetrics from '(?:^|[[:space:]])scratchPeakFiles=([0-9]+)')::bigint files
0081         FROM jobs WHERE observed >= %s AND observed <= %s AND jeditaskid IS NOT NULL
0082       ), limits AS MATERIALIZED (SELECT {quantile_fields} FROM measured)
0083       SELECT m.jeditaskid,t.taskname,m.bucket,COUNT(*),MAX(m.memory),MAX(m.disk),MAX(m.files),
0084           COUNT(m.memory),COUNT(m.disk),COUNT(m.files),{scale_fields}
0085         FROM measured m CROSS JOIN limits
0086         LEFT JOIN "{PANDA_SCHEMA}".jedi_tasks t ON t.jeditaskid=m.jeditaskid
0087         GROUP BY m.jeditaskid,t.taskname,m.bucket ORDER BY m.jeditaskid,m.bucket'''
0088     with transaction.atomic(using='panda'):
0089         with connections['panda'].cursor() as cursor:
0090             cursor.execute("SET LOCAL statement_timeout='20s'")
0091             cursor.execute(sql, bounds + [bounds[0]] + bounds * 2)
0092             rows = cursor.fetchall()
0093     return assemble(rows, now)
0094
0095
0096 def product(refresh=False):
0097     return get_product(PRODUCT_KEY, build, 60, refresh=refresh, async_first_fill=True)