src/monitor_app/worker_disk_space.py

0001 """Worker filesystem headroom from native-pilot heartbeats, with retained history."""
0002 from datetime import datetime, timezone
0003 import re
0004
0005 SITES = ('UM_GREX_PanDA_1',)
0006 FIELDS = ('FreeMB', 'TotalMB', 'IFree', 'ITotal')
0007 # The alarm judges a worker volume by free fraction, free space and free
0008 # inodes; the worst sample by one measure need not be the worst by another.
0009 MEASURES = (lambda row: row['free_fraction'], lambda row: row['free_bytes'],
0010             lambda row: row['free_inodes'] / row['total_inodes'] if row['total_inodes'] else 1)
0011
0012
0013 def summarize(rows, now=None):
0014     now = now or datetime.now(timezone.utc).timestamp()
0015     volumes, worst, running, reporting = {}, {}, 0, 0
0016     for site, pandaid, metrics in rows:
0017         running += 1
0018         values = {key: int(value) for key, value in
0019                   re.findall(r'\b(diskSample|(?:ce|pilot|scratch)(?:FreeMB|TotalMB|IFree|ITotal))=(\d+)', metrics or '')}
0020         stamp = values.get('diskSample')
0021         if not stamp or not -60 <= now - stamp <= 900:
0022             continue
0023         reporting += 1
0024         for label in ('ce', 'pilot', 'scratch'):
0025             if not all(label + key in values for key in FIELDS):
0026                 continue
0027             total, free = values[label + 'TotalMB'], values[label + 'FreeMB']
0028             if total <= 0:
0029                 continue
0030             key = site + ':' + label
0031             row = dict(site=site, volume=label, free_bytes=free * 2**20,
0032                        total_bytes=total * 2**20, free_fraction=free / total,
0033                        free_inodes=values[label + 'IFree'],
0034                        total_inodes=values[label + 'ITotal'],
0035                        observed_at=stamp, pandaid=pandaid)
0036             if label != 'ce':
0037                 # Worker volumes are distinct filesystems: keep the worst
0038                 # fresh sample by each measure the alarm judges.
0039                 for index, measure in enumerate(MEASURES):
0040                     old = worst.get((key, index))
0041                     if old is None or measure(row) < measure(old):
0042                         worst[(key, index)] = row
0043                 continue
0044             old = volumes.get(key)
0045             # The CE spool is one shared filesystem: a newer sample observes
0046             # cleanup that an older low sample cannot.
0047             if old is None or stamp > old['observed_at'] or (
0048                     stamp == old['observed_at'] and row['free_fraction'] < old['free_fraction']):
0049                 volumes[key] = row
0050     rows = list(volumes.values())
0051     for row in worst.values():
0052         if not any(row is kept for kept in rows):
0053             rows.append(row)
0054     return dict(at=int(now), running_jobs=running, reporting_jobs=reporting, volumes=rows)
0055
0056
0057 def build():
0058     from django.db import connections, transaction
0059     from django.utils import timezone as tz
0060     from .models import CachedProduct
0061     with connections['panda'].cursor() as cursor:
0062         cursor.execute('SELECT computingsite,pandaid,jobmetrics FROM doma_panda.jobsactive4 '
0063                        'WHERE computingsite=ANY(%s) AND jobstatus=%s', [list(SITES), 'running'])
0064         result = summarize(cursor.fetchall())
0065     from .cached_product import stores
0066     if not stores():
0067         return result
0068     with transaction.atomic():
0069         history, _ = CachedProduct.objects.select_for_update().get_or_create(
0070             key='worker_disk_space_history:v1', defaults={'value': {'samples': []}, 'built_at': tz.now()})
0071         samples = [s for s in history.value.get('samples', []) if s['at'] > result['at'] - 7 * 86400]
0072         if not samples or result['at'] - samples[-1]['at'] >= 60:
0073             samples.append(result)
0074             history.value = {'samples': samples}
0075             history.built_at = tz.now()
0076             history.save(update_fields=['value', 'built_at', 'updated_at'])
0077     return result
0078
0079
0080 def reading(refresh=False):
0081     from .cached_product import get_product
0082     product = get_product('worker_disk_space:v1', build, 60, refresh=refresh)
0083     value = dict(product.get('value') or {})
0084     value['age_seconds'] = product.get('age_seconds')
0085     return value
0086
0087
0088 from rest_framework.decorators import api_view, permission_classes
0089 from rest_framework.permissions import AllowAny
0090 from rest_framework.response import Response
0091
0092
0093 @api_view(['GET'])
0094 @permission_classes([AllowAny])
0095 def worker_disk_space(request):
0096     return Response(reading(refresh=request.query_params.get('refresh') == '1'))