management/commands/sweep_panda_associations.py

0001 """Batch PanDA-association sweep — the scheduled counterpart of the lazy path.
0002
0003 Associations between PanDA tasks and PCS campaign tasks are otherwise created
0004 only when a person views a task through the monitor (or at PCS submission).
0005 This command applies the same reconciliation to every recent EIC PanDA task,
0006 so directly submitted production is pulled into the catalog no matter who
0007 ignores the UI. Run nightly by the prod-ops agent's catalog_sync chain; run
0008 once with a wide --days window as the backfill.
0009 """
0010
0011 from django.core.management.base import BaseCommand
0012
0013
0014 class Command(BaseCommand):
0015     help = ('Associate recent PanDA tasks with PCS campaign tasks '
0016             '(batch form of the lazy per-view reconciliation).')
0017
0018     def add_arguments(self, parser):
0019         parser.add_argument('--days', type=int, default=14,
0020                             help='PanDA task modification window (default 14)')
0021         parser.add_argument('--limit', type=int, default=1000,
0022                             help='max PanDA tasks to examine (default 1000)')
0023         parser.add_argument('--no-intake', action='store_true',
0024                             help='associate only; skip auto-intake of '
0025                                  'unmatched group.EIC production tasknames')
0026
0027     def handle(self, *args, **opts):
0028         from datetime import timedelta
0029
0030         from django.db import connections
0031         from django.utils import timezone
0032
0033         from monitor_app.panda.constants import PANDA_SCHEMA
0034         from pcs.services import (intake_direct_panda_task,
0035                                   reconcile_panda_task_association)
0036
0037         # Lean query: only the fields the reconciler and intake read. The
0038         # general list_tasks() augments every task with per-task job counts,
0039         # which this sweep never uses and which blow the time budget at
0040         # backfill windows.
0041         cutoff = timezone.now() - timedelta(days=opts['days'])
0042         fields = ['jeditaskid', 'taskname', 'status', 'site', 'username',
0043                   'workinggroup', 'processingtype']
0044         field_list = ', '.join('"%s"' % f for f in fields)
0045         with connections['panda'].cursor() as cur:
0046             cur.execute(
0047                 f'SELECT {field_list} '
0048                 f'FROM "{PANDA_SCHEMA}"."jedi_tasks" '
0049                 f'WHERE COALESCE("modificationtime", "creationdate") >= %s '
0050                 f'AND "workinggroup" = %s '
0051                 f'ORDER BY "jeditaskid" DESC LIMIT %s',
0052                 [cutoff, 'EIC', opts['limit']])
0053             tasks = [dict(zip(fields, row)) for row in cur.fetchall()]
0054
0055         from pcs.models import PandaTasks
0056
0057         checked = new = existing = unmatched = intaken = skipped = 0
0058         for panda_task in tasks:
0059             # Non-production names (user.*, testbed fastproc) can never match
0060             # a PCS composed name — skip the expensive matching unless an
0061             # association already exists to refresh.
0062             taskname = str(panda_task.get('taskname') or '')
0063             if not taskname.startswith('group.'):
0064                 jedi = panda_task.get('jeditaskid')
0065                 if not PandaTasks.objects.filter(jedi_task_id=jedi).exists():
0066                     skipped += 1
0067                     continue
0068             pcs_task, row, reason = reconcile_panda_task_association(panda_task)
0069             checked += 1
0070             if row is None and not opts['no_intake']:
0071                 # Commissioning/migration policy: directly submitted
0072                 # group.EIC production is auto-intaken into the catalog,
0073                 # then associated by the normal reconciler.
0074                 task, intake_reason = intake_direct_panda_task(panda_task)
0075                 if task is not None:
0076                     intaken += 1
0077                     self.stdout.write(
0078                         f"intaken jediTaskID={panda_task.get('jeditaskid')} "
0079                         f"-> {task.name}")
0080                     pcs_task, row, reason = reconcile_panda_task_association(panda_task)
0081             if row is None:
0082                 unmatched += 1
0083                 self.stdout.write(
0084                     f"unmatched jediTaskID={panda_task.get('jeditaskid')}: {reason}")
0085             elif reason == 'existing jediTaskID association':
0086                 existing += 1
0087             else:
0088                 new += 1
0089                 name = pcs_task.composed_name if pcs_task else '?'
0090                 self.stdout.write(
0091                     f"associated jediTaskID={panda_task.get('jeditaskid')} "
0092                     f"-> {name} ({reason})")
0093
0094         # Summary is the last stdout line; the agent records it in the
0095         # action stream.
0096         self.stdout.write(
0097             f"checked={checked} new={new} existing={existing} "
0098             f"intaken={intaken} unmatched={unmatched} skipped={skipped} "
0099             f"days={opts['days']}")