File indexing completed on 2026-08-12 09:36:16
0001 """Credential-free orchestration for durable PanDA pause/resume requests."""
0002
0003 import json
0004 import uuid
0005
0006 from django.db import IntegrityError, transaction
0007 from django.utils import timezone
0008
0009 from monitor_app.activemq_connection import ActiveMQConnectionManager
0010 from monitor_app.models import PandaTaskOperation
0011
0012 PAUSE_REJECTED_TASK_STATUSES = frozenset(
0013 ('finished', 'failed', 'done', 'aborted', 'broken', 'paused'))
0014 RESUMABLE_TASK_STATUSES = frozenset(('paused', 'throttled', 'staging'))
0015 BULK_OPERATION_STATUSES = {
0016 'pause': frozenset(('running',)),
0017 'resume': frozenset(('paused',)),
0018 }
0019 MAX_BULK_TASKS = 5000
0020
0021
0022 class PandaTaskOperationError(Exception):
0023 def __init__(self, detail, status=400):
0024 super().__init__(detail)
0025 self.detail = detail
0026 self.status = status
0027
0028
0029 def serialize_operation(operation):
0030 return {
0031 'id': str(operation.id),
0032 'jedi_task_id': operation.jedi_task_id,
0033 'task_name': operation.task_name,
0034 'operation': operation.operation,
0035 'source': operation.source,
0036 'requested_by': operation.requested_by,
0037 'status': operation.status,
0038 'pending': operation.is_pending,
0039 'diagnostic': operation.diagnostic,
0040 'observed_status': operation.observed_status,
0041 'evidence': operation.evidence,
0042 'requested_at': operation.requested_at.isoformat(),
0043 'started_at': (operation.started_at.isoformat()
0044 if operation.started_at else None),
0045 'accepted_at': (operation.accepted_at.isoformat()
0046 if operation.accepted_at else None),
0047 'completed_at': (operation.completed_at.isoformat()
0048 if operation.completed_at else None),
0049 'updated_at': operation.updated_at.isoformat(),
0050 }
0051
0052
0053 def operation_controls(task, *, authenticated, internal_monitor, pending=None):
0054 """Return enabled state and concise disabled reasons for both controls."""
0055 status = str(task.get('status') or '').lower()
0056 pause_state_valid = bool(status) and status not in PAUSE_REJECTED_TASK_STATUSES
0057 resume_state_valid = status in RESUMABLE_TASK_STATUSES
0058 controls = {
0059 'visible': pause_state_valid or resume_state_valid,
0060 'pause': {'enabled': False, 'reason': ''},
0061 'resume': {'enabled': False, 'reason': ''},
0062 }
0063 if not controls['visible']:
0064 return controls
0065 if not internal_monitor:
0066 reason = 'Available on the internal monitor.'
0067 controls['pause']['reason'] = reason
0068 controls['resume']['reason'] = reason
0069 return controls
0070 if not authenticated:
0071 controls['pause']['reason'] = 'Sign in to pause this task.'
0072 controls['resume']['reason'] = 'Sign in to resume this task.'
0073 return controls
0074 if pending:
0075 reason = f'{pending.operation.title()} is already in progress.'
0076 controls['pause']['reason'] = reason
0077 controls['resume']['reason'] = reason
0078 return controls
0079 if resume_state_valid:
0080 controls['resume']['enabled'] = True
0081 controls['resume']['reason'] = f'Resume this {status} task.'
0082 else:
0083 controls['resume']['reason'] = 'Task is not resumable.'
0084 if status == 'paused':
0085 controls['pause']['reason'] = 'Task is already paused.'
0086 return controls
0087 if pause_state_valid:
0088 controls['pause']['enabled'] = True
0089 controls['pause']['reason'] = 'Pause this task.'
0090 return controls
0091
0092
0093 def queue_task_operation(*, task, operation, requested_by, source='manual',
0094 evidence=None):
0095 """Persist and queue one pause/resume request for the prod-ops agent."""
0096 record, created = _persist_task_operation(
0097 task=task,
0098 operation=operation,
0099 requested_by=requested_by,
0100 source=source,
0101 evidence=evidence,
0102 )
0103 if not created:
0104 return record, False
0105 jedi_task_id = record.jedi_task_id
0106 msg = {
0107 'msg_type': 'panda_task_operation',
0108 'namespace': 'prodops',
0109 'operation_id': str(record.id),
0110 'task_name': record.task_name or f'PanDA task {jedi_task_id}',
0111 'jedi_task_id': jedi_task_id,
0112 'operation': operation,
0113 'source': source,
0114 'created_by': requested_by,
0115 }
0116 _send_operation_message(msg, [record])
0117 return record, True
0118
0119
0120 def _persist_task_operation(*, task, operation, requested_by, source,
0121 evidence=None, allowed_statuses=None):
0122 """Validate and create one record without sending an ActiveMQ message."""
0123 if operation not in ('pause', 'resume'):
0124 raise PandaTaskOperationError('operation must be pause or resume')
0125 try:
0126 jedi_task_id = int(task.get('jeditaskid'))
0127 except (TypeError, ValueError):
0128 raise PandaTaskOperationError('Task has no valid JEDI task ID.', 409)
0129
0130 pending = (PandaTaskOperation.objects
0131 .filter(jedi_task_id=jedi_task_id,
0132 status__in=PandaTaskOperation.PENDING_STATUSES)
0133 .first())
0134 if pending:
0135 if pending.operation == operation:
0136 return pending, False
0137 raise PandaTaskOperationError(
0138 f'{pending.operation.title()} is already in progress for this task.',
0139 409,
0140 )
0141
0142 task_status = str(task.get('status') or '').lower()
0143 if allowed_statuses is not None and task_status not in allowed_statuses:
0144 raise PandaTaskOperationError(
0145 f'Task cannot be bulk {operation}d from status {task_status}.', 409)
0146 if operation == 'pause':
0147 if task_status in PAUSE_REJECTED_TASK_STATUSES:
0148 raise PandaTaskOperationError(
0149 f'Task cannot be paused from status {task_status}.', 409)
0150 elif task_status not in RESUMABLE_TASK_STATUSES:
0151 raise PandaTaskOperationError(
0152 f'Task cannot be resumed from status {task_status}.', 409)
0153
0154 try:
0155 with transaction.atomic():
0156 record = PandaTaskOperation.objects.create(
0157 jedi_task_id=jedi_task_id,
0158 task_name=str(task.get('taskname') or ''),
0159 operation=operation,
0160 source=source,
0161 requested_by=requested_by,
0162 evidence=evidence or {'task_status': task_status},
0163 )
0164 except IntegrityError:
0165 pending = (PandaTaskOperation.objects
0166 .filter(jedi_task_id=jedi_task_id,
0167 status__in=PandaTaskOperation.PENDING_STATUSES)
0168 .first())
0169 if pending and pending.operation == operation:
0170 return pending, False
0171 raise PandaTaskOperationError(
0172 'Another operation is already in progress for this task.', 409)
0173 return record, True
0174
0175
0176 def _send_operation_message(message, records):
0177 """Send one operation message and fail every new record if it cannot queue."""
0178 try:
0179 triggered = ActiveMQConnectionManager().send_message(
0180 '/queue/epicprod.ops', json.dumps(message))
0181 except Exception as exc:
0182 triggered = False
0183 failure = str(exc)
0184 else:
0185 failure = 'ops-agent queue unreachable'
0186 if not triggered:
0187 diagnostic = f'Could not queue operation: {failure}'
0188 now = timezone.now()
0189 PandaTaskOperation.objects.filter(
0190 pk__in=[record.pk for record in records]).update(
0191 status='failed', diagnostic=diagnostic, completed_at=now,
0192 updated_at=now)
0193 raise PandaTaskOperationError(diagnostic, 503)
0194
0195
0196 def queue_task_operations(*, tasks, operation, requested_by):
0197 """Persist eligible task records and queue one paced prod-ops batch."""
0198 if operation not in BULK_OPERATION_STATUSES:
0199 raise PandaTaskOperationError('operation must be pause or resume')
0200 if not tasks:
0201 raise PandaTaskOperationError('Select at least one task.')
0202 if len(tasks) > MAX_BULK_TASKS:
0203 raise PandaTaskOperationError(
0204 f'At most {MAX_BULK_TASKS} tasks may be submitted at once.')
0205
0206 batch_id = str(uuid.uuid4())
0207 records = []
0208 new_records = []
0209 rejected = []
0210 allowed_statuses = BULK_OPERATION_STATUSES[operation]
0211 for task in tasks:
0212 jedi_task_id = task.get('jeditaskid')
0213 task_status = str(task.get('status') or '').lower()
0214 try:
0215 record, created = _persist_task_operation(
0216 task=task,
0217 operation=operation,
0218 requested_by=requested_by,
0219 source='manual-bulk',
0220 evidence={'task_status': task_status, 'batch_id': batch_id},
0221 allowed_statuses=allowed_statuses,
0222 )
0223 except PandaTaskOperationError as exc:
0224 rejected.append({
0225 'jedi_task_id': jedi_task_id,
0226 'status': task_status,
0227 'error': exc.detail,
0228 })
0229 continue
0230 records.append(record)
0231 if created:
0232 new_records.append(record)
0233
0234 if not records:
0235 return {'batch_id': batch_id, 'records': [], 'rejected': rejected,
0236 'queued': 0}
0237 if new_records:
0238 items = [
0239 {
0240 'operation_id': str(record.id),
0241 'jedi_task_id': record.jedi_task_id,
0242 'task_name': (record.task_name
0243 or f'PanDA task {record.jedi_task_id}'),
0244 }
0245 for record in new_records
0246 ]
0247 _send_operation_message({
0248 'msg_type': 'panda_task_operations',
0249 'namespace': 'prodops',
0250 'batch_id': batch_id,
0251 'operation': operation,
0252 'items': items,
0253 'source': 'manual-bulk',
0254 'created_by': requested_by,
0255 }, new_records)
0256 return {
0257 'batch_id': batch_id,
0258 'records': [serialize_operation(record) for record in records],
0259 'rejected': rejected,
0260 'queued': len(new_records),
0261 }
0262
0263
0264 def update_task_operation(operation_id, *, status, diagnostic='',
0265 observed_status='', evidence=None):
0266 """Commit an agent-reported lifecycle transition to the durable record."""
0267 valid = {choice[0] for choice in PandaTaskOperation.STATUS_CHOICES}
0268 if status not in valid or status == 'queued':
0269 raise PandaTaskOperationError('Invalid operation status.')
0270 try:
0271 record = PandaTaskOperation.objects.get(pk=operation_id)
0272 except PandaTaskOperation.DoesNotExist:
0273 raise PandaTaskOperationError('Operation not found.', 404)
0274
0275 now = timezone.now()
0276 record.status = status
0277 record.diagnostic = str(diagnostic or '')[:4000]
0278 record.observed_status = str(observed_status or '')[:50]
0279 if evidence:
0280 merged = dict(record.evidence or {})
0281 merged.update(evidence)
0282 record.evidence = merged
0283 if status == 'running' and not record.started_at:
0284 record.started_at = now
0285 if status in ('accepted', 'verified', 'unverified') and not record.accepted_at:
0286 record.accepted_at = now
0287 if status in ('verified', 'failed', 'timeout', 'unverified'):
0288 record.completed_at = now
0289 record.save()
0290 return record
