swf-monitor/scripts/panda-sandbox-keepalive.py

0001 #!/usr/bin/env python3
0002 """panda-sandbox-keepalive.py — keep retryable tasks' sandbox tarballs alive.
0003
0004 The PanDA server purges its sandbox cache of files whose modification time
0005 is older than seven days (copyArchive.py), and nothing refreshes the files
0006 of tasks that outlive that window, so a retry of an older task fails when
0007 each job's pre-process cannot download the tarball (executor error 5303).
0008 The server API provides ``touch_cache_file`` for exactly this: it resets
0009 the file's modification time so the cleanup passes it by.
0010
0011 This doer touches the sandbox tarball of every task worth keeping
0012 retryable: epic-VO tasks in a non-final state, plus tasks finished, failed,
0013 or exhausted within the retention window (SysConfig
0014 ``panda_sandbox_keepalive_final_days``). Each task's tarball name and
0015 source server come from its stored parameters (``jedi_taskparams``). A
0016 tarball already purged is retained in the run inventory per task — that
0017 task is not natively retryable. Aborted and broken tasks are not kept
0018 alive. Missing tarballs are established state, not a failure of the
0019 current keepalive pass; API and authentication failures remain errors.
0020
0021 The same pass maintains log-dataset lifetimes: each candidate task's
0022 BNL Rucio datasets (located by ``task_id`` metadata) whose expiry falls
0023 inside the retention window are refreshed to the full window, so the
0024 logs of a task that is still active or recently final never expire on
0025 their original registration-time clock. Datasets carrying no expiry are
0026 left untouched.
0027
0028 The prod-ops agent's doer for the nightly ``catalog_sync`` chain step;
0029 Django-bootstrap standalone script — also usable by hand. Auth is the
0030 production x509 proxy (``X509_USER_PROXY``) as TLS client certificate,
0031 the same credential the payload-log fetch uses. See
0032 ``swf-epicprod/docs/JEDI_INTEGRATION.md``.
0033
0034 Usage::
0035
0036     cd /data/wenauseic/github/swf-monitor/src
0037     source ../../swf-testbed/.venv/bin/activate && source ~/.env
0038     python ../scripts/panda-sandbox-keepalive.py [--dry-run]
0039 """
0040 import argparse
0041 import ast
0042 import json
0043 import os
0044 import re
0045 import sys
0046
0047 THIS_DIR = os.path.dirname(os.path.abspath(__file__))
0048 sys.path.insert(0, os.path.join(THIS_DIR, '..', 'src'))
0049 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'swf_monitor_project.settings')
0050
0051 import django  # noqa: E402
0052 django.setup()
0053
0054 import requests  # noqa: E402
0055 from django.db import connections  # noqa: E402
0056 from monitor_app.models import SysConfig  # noqa: E402
0057
0058 X509_PROXY = os.environ.get(
0059     "X509_USER_PROXY", "/data/wenauseic/longproxy-for-rucio")
0060 # The PanDA server's own cache checks run with verify disabled
0061 # (copyArchive.py); honor a configured CA bundle when one is present.
0062 CA_VERIFY = os.environ.get("REQUESTS_CA_BUNDLE") or False
0063
0064 FINAL_STATUSES = ("done", "finished", "failed", "broken", "aborted",
0065                   "exhausted")
0066 RETRYABLE_FINAL_STATUSES = ("finished", "failed", "exhausted")
0067
0068 # Sandbox names come in two prefixes (jobO for API submissions, sources for
0069 # prun) and two UUID spellings (condensed hex and hyphenated).
0070 TARBALL_RE = re.compile(r"(?:jobO|sources)\.[0-9a-f-]+\.tar\.gz")
0071 SOURCE_URL_RE = re.compile(r'"sourceURL":\s*"([^"]+)"')
0072
0073
0074 def _log(msg):
0075     print(msg, file=sys.stderr, flush=True)
0076
0077
0078 def _candidate_tasks(final_days):
0079     cur = connections["panda"].cursor()
0080     cur.execute(
0081         "SELECT jeditaskid, status FROM jedi_tasks"
0082         " WHERE vo = 'epic'"
0083         " AND (NOT (status = ANY(%s))"
0084         "      OR (status = ANY(%s)"
0085         "          AND modificationtime > now() - make_interval(days => %s)))"
0086         " ORDER BY jeditaskid",
0087         [list(FINAL_STATUSES), list(RETRYABLE_FINAL_STATUSES),
0088          int(final_days)],
0089     )
0090     return cur.fetchall()
0091
0092
0093 def _task_sandbox(jedi_task_id):
0094     """Return (source_url, tarball_name) from stored task parameters, or
0095     (None, None) when the task has no cache-resident sandbox."""
0096     cur = connections["panda"].cursor()
0097     cur.execute(
0098         "SELECT taskparams FROM jedi_taskparams WHERE jeditaskid = %s",
0099         [jedi_task_id],
0100     )
0101     row = cur.fetchone()
0102     if not row or not row[0]:
0103         return None, None
0104     text = row[0] if isinstance(row[0], str) else row[0].read()
0105     tarballs = set(TARBALL_RE.findall(text))
0106     sources = set(SOURCE_URL_RE.findall(text))
0107     if not tarballs or not sources:
0108         return None, None
0109     if len(tarballs) > 1 or len(sources) > 1:
0110         _log(f"WARNING: task {jedi_task_id} has ambiguous sandbox refs "
0111              f"tarballs={sorted(tarballs)} sources={sorted(sources)}")
0112     return sorted(sources)[0], sorted(tarballs)[0]
0113
0114
0115 def _refresh_log_lifetimes(jedi_task_ids, final_days):
0116     """Refresh the lifetime of candidate tasks' BNL Rucio datasets.
0117
0118     A dataset whose expiry falls inside the retention window gets a fresh
0119     ``final_days`` lifetime; one with no expiry is left alone. Failures
0120     are reported per dataset, never raised.
0121     """
0122     result = {"checked": 0, "refreshed": 0, "errors": []}
0123     try:
0124         from rucio.client import Client
0125     except ImportError as e:
0126         result["errors"].append(f"rucio client unavailable: {e}")
0127         _log(f"WARNING: log-lifetime refresh skipped: {e}")
0128         return result
0129     rucio_url = os.environ.get("RUCIO_URL", "https://nprucio01.sdcc.bnl.gov:443")
0130     scope = os.environ.get("RUCIO_SCOPE", "group.EIC")
0131     try:
0132         client = Client(
0133             rucio_host=rucio_url, auth_host=rucio_url,
0134             account=os.environ.get("RUCIO_ACCOUNT", "panda"),
0135             auth_type="x509_proxy", creds={"client_proxy": X509_PROXY},
0136             ca_cert=None, vo=os.environ.get("RUCIO_VO", "eic"))
0137         client.whoami()
0138     except Exception as e:
0139         result["errors"].append(f"BNL Rucio auth failed: {e}")
0140         _log(f"WARNING: log-lifetime refresh skipped: auth failed: {e}")
0141         return result
0142     import datetime as dt
0143     horizon = dt.datetime.utcnow() + dt.timedelta(days=final_days)
0144     for jedi_task_id in jedi_task_ids:
0145         try:
0146             names = list(client.list_dids(
0147                 scope=scope, filters={"task_id": int(jedi_task_id)},
0148                 did_type="dataset"))
0149         except Exception as e:
0150             result["errors"].append(f"task {jedi_task_id}: list_dids: {e}")
0151             _log(f"WARNING: lifetime lookup failed for task {jedi_task_id}: {e}")
0152             continue
0153         for name in names:
0154             try:
0155                 result["checked"] += 1
0156                 meta = client.get_metadata(scope=scope, name=name)
0157                 expired_at = meta.get("expired_at")
0158                 if expired_at is None or expired_at >= horizon:
0159                     continue
0160                 client.set_metadata(scope=scope, name=name, key="lifetime",
0161                                     value=final_days * 86400)
0162                 result["refreshed"] += 1
0163                 _log(f"refreshed lifetime for task {jedi_task_id} dataset "
0164                      f"{scope}:{name} (was expiring {expired_at})")
0165             except Exception as e:
0166                 result["errors"].append(f"{name}: {e}")
0167                 _log(f"WARNING: lifetime refresh failed for {scope}:{name}: {e}")
0168     return result
0169
0170
0171 def _touch(source_url, tarball):
0172     """POST touch_cache_file; returns (status, message) where status is
0173     'touched', 'missing', or 'error'."""
0174     url = f"{source_url}/api/v1/file_server/touch_cache_file"
0175     try:
0176         r = requests.post(url, data={"file_name": tarball},
0177                           cert=(X509_PROXY, X509_PROXY),
0178                           verify=CA_VERIFY, timeout=30)
0179     except Exception as e:
0180         return "error", f"{type(e).__name__}: {e}"
0181     if r.status_code != 200:
0182         return "error", f"HTTP {r.status_code}: {r.text[:200]}"
0183     # The endpoint answers with a Python-repr dict, not JSON; accept both.
0184     try:
0185         body = r.json()
0186     except ValueError:
0187         try:
0188             body = ast.literal_eval(r.text.strip())
0189         except (ValueError, SyntaxError):
0190             return "error", f"unparseable response: {r.text[:200]}"
0191     if not isinstance(body, dict):
0192         return "error", f"unexpected response shape: {r.text[:200]}"
0193     if body.get("success"):
0194         return "touched", ""
0195     message = str(body.get("message") or "")
0196     if "FileNotFoundError" in message or "No such file" in message:
0197         return "missing", message
0198     return "error", message
0199
0200
0201 def main():
0202     parser = argparse.ArgumentParser()
0203     parser.add_argument("--dry-run", action="store_true",
0204                         help="enumerate and report; do not touch")
0205     args = parser.parse_args()
0206
0207     final_days = int(SysConfig.get_setting(
0208         "panda_sandbox_keepalive_final_days", 30))
0209
0210     tasks = _candidate_tasks(final_days)
0211     # One sandbox can serve several tasks (sandbox reuse); touch each once.
0212     by_tarball = {}
0213     no_sandbox = []
0214     for jedi_task_id, status in tasks:
0215         source_url, tarball = _task_sandbox(jedi_task_id)
0216         if not tarball:
0217             no_sandbox.append(jedi_task_id)
0218             continue
0219         entry = by_tarball.setdefault(
0220             (source_url, tarball), {"tasks": [], "statuses": []})
0221         entry["tasks"].append(jedi_task_id)
0222         entry["statuses"].append(status)
0223
0224     touched, missing, errors = [], [], []
0225     for (source_url, tarball), entry in sorted(by_tarball.items()):
0226         if args.dry_run:
0227             _log(f"dry-run: would touch {tarball} on {source_url} "
0228                  f"for tasks {entry['tasks']}")
0229             continue
0230         status, message = _touch(source_url, tarball)
0231         record = {"tarball": tarball, "tasks": entry["tasks"],
0232                   "statuses": entry["statuses"]}
0233         if status == "touched":
0234             touched.append(record)
0235         elif status == "missing":
0236             missing.append(record)
0237             _log(f"sandbox already absent for tasks {entry['tasks']} "
0238                  f"({tarball}) — not natively retryable")
0239         else:
0240             record["error"] = message
0241             errors.append(record)
0242             _log(f"ERROR: touch failed for {tarball} on {source_url}: "
0243                  f"{message}")
0244
0245     if args.dry_run:
0246         lifetimes = {"checked": 0, "refreshed": 0, "errors": [],
0247                      "skipped": "dry run"}
0248     else:
0249         lifetimes = _refresh_log_lifetimes(
0250             [jedi_task_id for jedi_task_id, _ in tasks], final_days)
0251
0252     summary = {
0253         "final_days": final_days,
0254         "candidates": len(tasks),
0255         "no_sandbox": no_sandbox,
0256         "tarballs": len(by_tarball),
0257         "touched": len(touched),
0258         "missing": missing,
0259         "errors": errors,
0260         "log_lifetimes": lifetimes,
0261         "dry_run": bool(args.dry_run),
0262         "ok": not errors and not lifetimes["errors"],
0263     }
0264     print(json.dumps(summary, default=str))
0265     return 0 if summary["ok"] else 1
0266
0267
0268 if __name__ == "__main__":
0269     sys.exit(main())