"Generic worker script that runs actual worker script" import os import sys import shlex import argparse import traceback import subprocess from time import sleep from datetime import timedelta from tempfile import TemporaryDirectory from redis import Redis from qc_app import jobs from qc_app.check_connections import check_redis def parse_args(): "Parse the command-line arguments" parser = argparse.ArgumentParser( prog="worker", description = ( "Generic worker to launch and manage specific worker scripts")) parser.add_argument( "redisurl", default="redis:///", help="URL to the redis server") parser.add_argument( "redisprefix", type=str, help="The prefix before the job ID.") parser.add_argument("job_id", help="The id of the job being processed") args = parser.parse_args() check_redis(args.redisurl) return args def run_job(rconn: Redis, job: dict, redisprefix: str): "Run the actual job." try: jobid = job["jobid"] with TemporaryDirectory() as tmpdir: stderrpath = f"{tmpdir}/{jobid}.stderr" with open(stderrpath, "w+b") as tmpfl: with subprocess.Popen( shlex.split(job["command"]), stdout=subprocess.PIPE, stderr=tmpfl) as process: while process.poll() is None: jobs.update_status(rconn, redisprefix, jobid, "running") jobs.update_stdout_stderr(rconn, redisprefix, jobid, process.stdout.read1(),#type: ignore[union-attr] "stdout") sleep(1) jobs.update_status( rconn, redisprefix, jobid, ("error" if process.returncode != 0 else "success")) with open(stderrpath, "rb") as stderr: stderr_content = stderr.read() jobs.update_stdout_stderr(rconn, redisprefix, jobid, stderr_content, "stderr") os.remove(stderrpath) return process.poll() except Exception as exc:# pylint: disable=[broad-except,unused-variable] jobs.update_status(rconn, redisprefix, jobid, "error") jobs.update_stdout_stderr( rconn, redisprefix, jobid, traceback.format_exc().encode("utf-8"), "stderr") print(traceback.format_exc(), file=sys.stderr) return 4 def main(): "Entry point function" args = parse_args() with Redis.from_url(args.redisurl, decode_responses=True) as rconn: job = jobs.job(rconn, args.redisprefix, args.job_id) if job: return run_job(rconn, job, args.redisprefix) jobs.update_status(rconn, args.redisprefix, args.job_id, "error") fqjobid = jobs.job_key(args.redisprefix, args.jobid) rconn.hset(fqjobid, "stderr", f"No such job. '{args.job_id}'.") rconn.expire(name=jobs.job_key(args.redisprefix, args.job_id), time=timedelta(seconds=(2 * 60 * 60))) print(f"No such job. '{args.job_id}'.", file=sys.stderr) return 2 return 3 if __name__ == "__main__": sys.exit(main())