File indexing completed on 2026-07-21 08:46:24
0001
0002 """
0003 Run one credentialed PanDA task operation for an existing JEDI task.
0004
0005 The web tier only queues these requests. This doer sources the PanDA client
0006 environment, uses the cached production token, and calls the PanDA client API.
0007 """
0008 import argparse
0009 import json
0010 import os
0011 import subprocess
0012 import sys
0013 import tempfile
0014
0015
0016 DEFAULT_PCLIENT_SETUP = os.path.expanduser("~/pclient/run/setup.sh")
0017 DEFAULT_AUTH_VO = "EIC.production"
0018
0019
0020 def _log(msg):
0021 print(msg, file=sys.stderr, flush=True)
0022
0023
0024 def _run_inside_pclient(args):
0025 payload = {
0026 "operation": args.operation,
0027 "jedi_task_id": args.jedi_task_id,
0028 "increase": args.increase,
0029 "new_parameters": args.new_parameters,
0030 }
0031 with tempfile.TemporaryDirectory(prefix="panda-task-operation.") as tmpdir:
0032 payload_path = os.path.join(tmpdir, "payload.json")
0033 runner = os.path.join(tmpdir, "run.sh")
0034 with open(payload_path, "w") as f:
0035 json.dump(payload, f)
0036 with open(runner, "w") as f:
0037 f.write("#!/bin/bash\nset -e\n")
0038 f.write(f"source {args.pclient_setup}\n")
0039 f.write(f"export PANDA_AUTH_VO={args.auth_vo}\n")
0040 f.write(f"python3 {os.path.abspath(__file__)} --inside-pclient --payload {payload_path}\n")
0041 try:
0042 p = subprocess.run(["bash", runner], capture_output=True, text=True,
0043 timeout=args.timeout)
0044 except subprocess.TimeoutExpired:
0045 _log(f"ERROR: PanDA operation timed out after {args.timeout}s")
0046 return 4
0047 if p.stdout:
0048 print(p.stdout, end="")
0049 if p.stderr:
0050 print(p.stderr, end="", file=sys.stderr)
0051 return p.returncode
0052
0053
0054 def _inside_pclient(payload_path):
0055 from pandaclient import panda_api
0056
0057 with open(payload_path) as f:
0058 payload = json.load(f)
0059
0060 client = panda_api.get_api()
0061 operation = payload["operation"]
0062 jedi_task_id = int(payload["jedi_task_id"])
0063
0064 if operation == "increase_attempts":
0065 result = client.increase_attempt_nr(jedi_task_id, int(payload.get("increase") or 1))
0066 elif operation == "retry_failures":
0067 new_parameters = payload.get("new_parameters") or None
0068 result = client.retry_task(jedi_task_id, new_parameters=new_parameters)
0069 else:
0070 raise ValueError(f"unknown operation {operation!r}")
0071
0072 ok, diagnostic = _panda_result_ok(result)
0073 print(json.dumps({
0074 "operation": operation,
0075 "jedi_task_id": jedi_task_id,
0076 "ok": ok,
0077 "diagnostic": diagnostic,
0078 "result": result,
0079 }, default=str))
0080 if not ok:
0081 _log(f"ERROR: PanDA returned failure for {operation} on {jedi_task_id}: {diagnostic}")
0082 return 1
0083 return 0
0084
0085
0086 def _panda_result_ok(result):
0087 """Interpret PanDA client (transport_status, operation_result) returns."""
0088 if isinstance(result, (list, tuple)) and result:
0089 status = result[0]
0090 if status != 0:
0091 return False, f"transport status {status}"
0092 if len(result) == 1:
0093 return True, "transport succeeded"
0094 payload = result[1]
0095 if isinstance(payload, (list, tuple)) and payload:
0096 code = payload[0]
0097 message = payload[1] if len(payload) > 1 else ""
0098 return code == 0, f"return code {code}: {message}"
0099 if isinstance(payload, dict):
0100 if "success" in payload:
0101 return bool(payload["success"]), payload.get("message") or str(payload)
0102 if "code" in payload:
0103 return payload.get("code") == 0, payload.get("message") or str(payload)
0104 if payload is None:
0105 return False, "operation returned None"
0106 return True, str(payload)
0107 return result is not None, str(result)
0108
0109
0110 def main():
0111 ap = argparse.ArgumentParser(description="Run an existing PanDA task operation.")
0112 ap.add_argument("--operation", choices=["increase_attempts", "retry_failures"])
0113 ap.add_argument("--jedi-task-id", type=int)
0114 ap.add_argument("--increase", type=int, default=1)
0115 ap.add_argument("--new-parameters", default="", help="JSON object for retry_task new_parameters")
0116 ap.add_argument("--auth-vo", default=DEFAULT_AUTH_VO)
0117 ap.add_argument("--pclient-setup", default=DEFAULT_PCLIENT_SETUP)
0118 ap.add_argument("--timeout", type=int, default=120)
0119 ap.add_argument("--inside-pclient", action="store_true")
0120 ap.add_argument("--payload")
0121 args = ap.parse_args()
0122
0123 if args.inside_pclient:
0124 if not args.payload:
0125 _log("ERROR: --payload required with --inside-pclient")
0126 return 2
0127 return _inside_pclient(args.payload)
0128
0129 if not args.operation or not args.jedi_task_id:
0130 _log("ERROR: --operation and --jedi-task-id are required")
0131 return 2
0132 if args.increase < 1:
0133 _log("ERROR: --increase must be >= 1")
0134 return 2
0135 if args.new_parameters:
0136 try:
0137 new_parameters = json.loads(args.new_parameters)
0138 except ValueError as e:
0139 _log(f"ERROR: --new-parameters is not valid JSON: {e}")
0140 return 2
0141 if not isinstance(new_parameters, dict):
0142 _log("ERROR: --new-parameters must be a JSON object")
0143 return 2
0144 args.new_parameters = new_parameters
0145 else:
0146 args.new_parameters = None
0147
0148 return _run_inside_pclient(args)
0149
0150
0151 if __name__ == "__main__":
0152 sys.exit(main())
