File indexing completed on 2026-11-09 10:47:38
0001 """Worker-start cohorts and native pilot admission evidence."""
0002 from collections import Counter
0003 from datetime import timedelta
0004 import time
0005
0006 from django.db import connections
0007 from django.utils import timezone
0008
0009 from .cached_product import get_product
0010 from .host_reports import HARVESTER_HOSTS, latest
0011
0012
0013 def osg_image_demand(memory_mb, cores):
0014 """One indexed active-job aggregate; no history scans or per-job reads."""
0015 queue = 'BNL_OSG_PanDA_1'
0016 began = time.monotonic()
0017 with connections['panda'].cursor() as cursor:
0018 cursor.execute('''SELECT container_name,
0019 COUNT(*) FILTER (WHERE jobstatus = 'activated'
0020 AND COALESCE(minramcount, 0) <= %s
0021 AND COALESCE(corecount, 1) <= %s) AS runnable,
0022 COUNT(*) FILTER (WHERE jobstatus IN ('running','starting','sent')),
0023 COUNT(*) FILTER (WHERE jobstatus = 'activated') AS activated
0024 FROM doma_panda.jobsactive4
0025 WHERE computingsite = %s
0026 AND jobstatus IN ('activated','running','starting','sent')
0027 AND prodsourcelabel IN ('managed','test','prod_test','install','user','panda')
0028 GROUP BY container_name''', [memory_mb, cores, queue])
0029 rows = [{'image': image or '', 'activated': runnable, 'running_jobs': running,
0030 'resource_ineligible': activated - runnable}
0031 for image, runnable, running, activated in cursor.fetchall()]
0032 return {'queue': queue, 'observed_at': timezone.now().strftime('%Y-%m-%dT%H:%M:%SZ'),
0033 'images': rows, 'query_ms': round(1000 * (time.monotonic() - began), 1),
0034 'disk_check': 'pilot dispatcher uses actual free space; Condor reservation is not that limit'}
0035
0036
0037 def outcome(queue, status, exit_code):
0038 """Keep deliberate native pre-job refusals apart from failed payloads."""
0039 try:
0040 exit_code = int(exit_code) if exit_code is not None else None
0041 except (TypeError, ValueError):
0042 exit_code = None
0043 if queue == 'BNL_OSG_PanDA_1' and exit_code == 73:
0044 return 'declined'
0045 if status == 'cancelled':
0046 return 'cancelled'
0047 if status == 'finished' and exit_code not in (None, 0):
0048 return 'failed'
0049 return status or 'unknown'
0050
0051
0052 def build(queues, hours):
0053 end = timezone.now().replace(tzinfo=None)
0054 start = end - timedelta(hours=hours)
0055 cohorts = Counter()
0056 declines_seconds = 0
0057 reasons = []
0058 recent = []
0059 with connections['panda'].cursor() as cursor:
0060 cursor.execute('''SELECT computingsite, status, nativeexitcode, COUNT(*),
0061 SUM(GREATEST(0, EXTRACT(EPOCH FROM (endtime-starttime))))
0062 FROM doma_panda.harvester_workers
0063 WHERE computingsite = ANY(%s) AND starttime > %s AND starttime <= %s
0064 GROUP BY 1, 2, 3''', [queues, start, end])
0065 for queue, status, code, count, seconds in cursor.fetchall():
0066 label = outcome(queue, status, code)
0067 cohorts[label] += count
0068 if label == 'declined':
0069 declines_seconds += float(seconds or 0)
0070 reasons.append({'outcome': label, 'raw_status': status,
0071 'exit_code': code, 'count': count})
0072 cursor.execute('''SELECT status, COUNT(*) FROM doma_panda.harvester_workers
0073 WHERE computingsite = ANY(%s) AND status IN ('running','submitted','ready','idle')
0074 GROUP BY 1''', [queues])
0075 workers_now = dict(cursor.fetchall())
0076 cursor.execute('''SELECT COUNT(*) FROM doma_panda.jobsactive4
0077 WHERE computingsite = ANY(%s) AND jobstatus='running' ''', [queues])
0078 jobs_now = cursor.fetchone()[0]
0079 cursor.execute('''SELECT workerid, harvesterid, batchid, computingsite,
0080 status, nativeexitcode, starttime, endtime, batchlog, diagmessage
0081 FROM doma_panda.harvester_workers
0082 WHERE computingsite = ANY(%s) AND starttime > %s AND starttime <= %s
0083 ORDER BY starttime DESC LIMIT 30''', [queues, start, end])
0084 for worker, harvester, batch, queue, status, code, began, ended, log, diag in cursor.fetchall():
0085 label = outcome(queue, status, code)
0086 recent.append({'workerid': worker, 'harvesterid': harvester,
0087 'batchid': batch, 'queue': queue, 'outcome': label,
0088 'exit_code': code, 'started': began.isoformat() if began else None,
0089 'ended': ended.isoformat() if ended else None, 'batch_log': log,
0090 'reason': ('Admission declined before taking a job: insufficient remaining lifetime'
0091 if label == 'declined' else str(diag or ''))})
0092 starts = sum(cohorts.values())
0093 return {'since': start.isoformat() + 'Z', 'until': end.isoformat() + 'Z',
0094 'hours': hours, 'starts': starts, 'declined': cohorts['declined'],
0095 'declined_pct': round(100 * cohorts['declined'] / starts, 1) if starts else 0,
0096 'still_running': cohorts['running'], 'finished': cohorts['finished'],
0097 'failed': cohorts['failed'], 'cancelled': cohorts['cancelled'],
0098 'other': sum(value for key, value in cohorts.items()
0099 if key not in ('declined','running','finished','failed','cancelled')),
0100 'declined_worker_hours': round(declines_seconds / 3600, 1),
0101 'workers_now': workers_now, 'jobs_now': jobs_now,
0102 'outcomes': sorted(reasons, key=lambda row: -row['count']), 'recent': recent}
0103
0104
0105 def product(queues, hours=1, refresh=False):
0106 key = 'worker_admission:v1:' + ','.join(sorted(queues)) + ':' + str(hours)
0107 return get_product(key, lambda: build(queues, hours), ttl_seconds=60,
0108 refresh=refresh)
0109
0110
0111 def startup_evidence(queues):
0112 for host in HARVESTER_HOSTS:
0113 report = latest(host) or {}
0114 data = ((report.get('record') or {}).get('harvester_db') or {}).get('admission_declines')
0115 if data and 'BNL_OSG_PanDA_1' in queues:
0116 return dict(data, reported_at=report.get('reported_at'),
0117 age_seconds=report.get('age_seconds'), host=host)
0118 return None
