monitor_app/viewdir/analysis.py

0001 """The Analysis page: user analysis on the production platform.
0002
0003 In-development landing page for the analysis capability
0004 (swf-epicprod/docs/PANDA_USER_JOBS.md): the analysis-capable queue set,
0005 analysis task activity, and analysis weather (queue wait times).
0006 Weather is computed from PanDA accounting data and cached; the page
0007 never runs the percentile scan on a warm cache.
0008 """
0009 from django.core.cache import cache
0010 from django.db import connections
0011 from django.shortcuts import render
0012
0013 from ..panda import list_queues
0014
0015 WEATHER_CACHE_KEY = 'analysis_weather_v1'
0016 WEATHER_CACHE_TTL = 3600  # seconds; hourly refresh is ample for a 14-day window
0017 WEATHER_WINDOW_DAYS = 14
0018
0019
0020 def _analysis_queues():
0021     """Unified queues from live PanDA schedconfig (the source the EIC
0022     queues page uses) — the queues serving both job classes."""
0023     result = list_queues(vo='eic')
0024     return [q for q in result.get('queues', [])
0025             if q.get('type') == 'unified']
0026
0027
0028 def _analysis_activity(limit=100):
0029     """Recent user-label tasks across the instance (PanDA DB)."""
0030     cur = connections['panda'].cursor()
0031     cur.execute(
0032         "SELECT jeditaskid, taskname, username, status, creationdate, "
0033         "modificationtime FROM doma_panda.jedi_tasks "
0034         "WHERE prodsourcelabel = 'user' "
0035         "AND modificationtime > now() - interval '30 days' "
0036         "ORDER BY jeditaskid DESC LIMIT %s", [limit])
0037     cols = [c[0] for c in cur.description]
0038     return [dict(zip(cols, row)) for row in cur.fetchall()]
0039
0040
0041 def _analysis_weather():
0042     """Per-queue job wait profile (creation to start), cached hourly."""
0043     rows = cache.get(WEATHER_CACHE_KEY)
0044     if rows is not None:
0045         return rows
0046     cur = connections['panda'].cursor()
0047     cur.execute(
0048         "SELECT computingsite, count(*) AS jobs, "
0049         "round((percentile_cont(0.5) WITHIN GROUP (ORDER BY "
0050         "EXTRACT(EPOCH FROM (starttime - creationtime))/60.0))::numeric, 1) "
0051         "AS median_wait_min, "
0052         "round((percentile_cont(0.9) WITHIN GROUP (ORDER BY "
0053         "EXTRACT(EPOCH FROM (starttime - creationtime))/60.0))::numeric, 1) "
0054         "AS p90_wait_min "
0055         "FROM doma_panda.jobsarchived4 "
0056         "WHERE creationtime > now() - interval '%s days' "
0057         "AND starttime IS NOT NULL AND starttime > creationtime "
0058         "AND jobstatus IN ('finished','failed') "
0059         "GROUP BY computingsite HAVING count(*) > 50 "
0060         "ORDER BY median_wait_min" % WEATHER_WINDOW_DAYS)
0061     cols = [c[0] for c in cur.description]
0062     rows = [dict(zip(cols, row)) for row in cur.fetchall()]
0063     cache.set(WEATHER_CACHE_KEY, rows, WEATHER_CACHE_TTL)
0064     return rows
0065
0066
0067 def _analysis_shares():
0068     """The global-share tree and the latest usage snapshot by share stamp.
0069
0070     Usage comes from the newest jobs_share_stats aggregation; jobs
0071     stamped before the tree existed appear under their old stamp until
0072     they drain."""
0073     cur = connections['panda'].cursor()
0074     cur.execute(
0075         "SELECT name, value, prodsourcelabel FROM doma_panda.global_shares "
0076         "ORDER BY value DESC")
0077     shares = [dict(zip(['name', 'value', 'labels'], r))
0078               for r in cur.fetchall()]
0079     cur.execute(
0080         "SELECT gshare, "
0081         "sum(hs) FILTER (WHERE jobstatus IN ('sent','running','starting')) "
0082         "AS executing_hs, "
0083         "sum(hs) FILTER (WHERE jobstatus = 'activated') AS queued_hs "
0084         "FROM doma_panda.jobs_share_stats "
0085         "WHERE ts = (SELECT max(ts) FROM doma_panda.jobs_share_stats) "
0086         "GROUP BY gshare ORDER BY gshare")
0087     usage = [dict(zip(['gshare', 'executing_hs', 'queued_hs'], r))
0088              for r in cur.fetchall()]
0089     total = sum(float(u['executing_hs'] or 0) for u in usage)
0090     for u in usage:
0091         u['executing_pct'] = (
0092             round(100 * float(u['executing_hs'] or 0) / total, 1)
0093             if total else None)
0094     return shares, usage
0095
0096
0097 def analysis_view(request):
0098     queues, activity, weather, shares, share_usage = [], [], [], [], []
0099     error = ''
0100     try:
0101         queues = _analysis_queues()
0102         activity = _analysis_activity()
0103         weather = _analysis_weather()
0104         shares, share_usage = _analysis_shares()
0105     except Exception as exc:
0106         error = str(exc)
0107     return render(request, 'monitor_app/analysis.html', {
0108         'queues': queues,
0109         'activity': activity,
0110         'weather': weather,
0111         'weather_window_days': WEATHER_WINDOW_DAYS,
0112         'shares': shares,
0113         'share_usage': share_usage,
0114         'error': error,
0115     })