source/sphenixprod/dispatch_productions.py

0001 #!/usr/bin/env python
0002
0003 import argparse
0004 import subprocess
0005 from pathlib import Path
0006 import sys
0007 import time
0008
0009 from simpleLogger import slogger, CustomFormatter, CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL
0010 from logging.handlers import RotatingFileHandler
0011 from datetime import datetime
0012
0013 def setup_dispatcher_logging(args):
0014     """Set up a rotating log handler for the dispatcher."""
0015     logdir = '/tmp/sphenixprod/sphenixprod/dispatcher'
0016     Path(logdir).mkdir(parents=True, exist_ok=True)
0017
0018     handler = RotatingFileHandler(
0019         filename=f"{logdir}/{str(datetime.today().date())}.log",
0020         mode='a',
0021         maxBytes=5*1024*1024, # 5 MB
0022         backupCount=5,
0023         encoding=None,
0024         delay=0
0025     )
0026     handler.setFormatter(CustomFormatter())
0027     slogger.addHandler(handler)
0028     slogger.setLevel(args.loglevel)
0029     INFO("Dispatcher logging setup complete.")
0030
0031 def main():
0032     parser = argparse.ArgumentParser(description="Dispatch multiple production_control jobs.")
0033     parser.add_argument('--steer-list', required=True, help="A text file listing the steer files to process, one per line.")
0034     parser.add_argument('--stagger', type=int, default=60, help="Seconds to wait between dispatching each job.")
0035
0036     vgroup = parser.add_argument_group('Logging level')
0037     exclusive_vgroup = vgroup.add_mutually_exclusive_group()
0038     exclusive_vgroup.add_argument('-v', '--verbose', help="Prints more information per repetition", action='count', default=0)
0039     exclusive_vgroup.add_argument('-d', '--debug', help="Prints even more information", action="store_true")
0040     exclusive_vgroup.add_argument('-c', '--chatty', help="Prints the most information", action="store_true")
0041     exclusive_vgroup.add_argument('--loglevel', dest='loglevel', default='INFO',
0042                                   help="Specific logging level (CHATTY, DEBUG, INFO, WARN, ERROR, CRITICAL)")
0043
0044     args = parser.parse_args()
0045
0046     if args.verbose == 1:
0047         args.loglevel = 'INFO'
0048     if args.debug or args.verbose == 2:
0049         args.loglevel = 'DEBUG'
0050     if args.chatty or args.verbose == 3:
0051         args.loglevel = 'CHATTY'
0052
0053     setup_dispatcher_logging(args)
0054
0055     steer_list_file = Path(args.steer_list)
0056     if not steer_list_file.is_file():
0057         ERROR(f"Steer list file '{steer_list_file}' not found.")
0058         sys.exit(1)
0059
0060     with open(steer_list_file, 'r') as f:
0061         steer_files = [line.strip() for line in f if line.strip() and not line.strip().startswith('#')]
0062
0063     if not steer_files:
0064         INFO(f"No active steer files found in '{steer_list_file}'. Exiting.")
0065         return
0066
0067     INFO(f"Found {len(steer_files)} steer files to process from {steer_list_file}.")
0068
0069     for i, steer_file_path in enumerate(steer_files):
0070         steer_file = Path(steer_file_path)
0071         if not steer_file.is_file():
0072             WARN(f"Steer file '{steer_file}' not found, skipping.")
0073             continue
0074
0075         command = [ "production_control.py", "--steerfile", str(steer_file), f"--loglevel={args.loglevel}" ]
0076
0077         INFO(f"Dispatching: {' '.join(command)}")
0078         result = subprocess.run(command)
0079
0080         if result.returncode == 2:
0081             DEBUG(f"production_control exited with 2 (host not in steer file), skipping stagger.")
0082         elif args.stagger > 0 and i < len(steer_files) - 1:
0083             INFO(f"Waiting for {args.stagger} seconds before next dispatch.")
0084             time.sleep(args.stagger)
0085
0086     INFO("All production control jobs dispatched.")
0087
0088 if __name__ == "__main__":
0089     main()