pandaserver/test/killJobLowPrio.py

0001 import optparse
0002 import time
0003
0004 # password
0005 from pandaserver.config import panda_config
0006 from pandaserver.taskbuffer.OraDBProxy import DBProxy
0007 from pandaserver.userinterface import Client
0008
0009 usageStr = """%prog [options] <priority>
0010
0011 Description: kill jobs with low priorities below a given value"""
0012 option_parser = optparse.OptionParser(conflict_handler="resolve", usage=usageStr)
0013 option_parser.add_option(
0014     "-9",
0015     action="store_const",
0016     const=True,
0017     dest="forceKill",
0018     default=False,
0019     help="kill jobs before next heartbeat is coming",
0020 )
0021 option_parser.add_option(
0022     "--running",
0023     action="store_const",
0024     const=True,
0025     dest="killRunning",
0026     default=False,
0027     help="kill running jobs to free up CPU slots. jobs will be killed regardless of job status if omitted",
0028 )
0029 option_parser.add_option("--site", action="store", dest="site", default=None, help="computingSite")
0030 option_parser.add_option("--cloud", action="store", dest="cloud", default=None, help="cloud")
0031 option_parser.add_option(
0032     "--maxJobs",
0033     action="store",
0034     dest="maxJobs",
0035     default=None,
0036     help="max number of jobs to be killed",
0037 )
0038 options, args = option_parser.parse_args()
0039
0040 if options.cloud is None and options.site is None:
0041     option_parser.error("--site=<computingSite> and/or --cloud=<cloud> is required")
0042
0043 proxyS = DBProxy()
0044 proxyS.connect(panda_config.dbhost, panda_config.dbpasswd, panda_config.dbuser, panda_config.dbname)
0045
0046 jobsMap = {}
0047
0048 if len(args) == 0:
0049     option_parser.error("priority is required")
0050
0051 varMap = {}
0052 varMap[":prodSourceLabel"] = "managed"
0053 varMap[":currentPriority"] = args[0]
0054 sql = "SELECT PandaID,currentPriority FROM %s WHERE prodSourceLabel=:prodSourceLabel AND currentPriority<:currentPriority "
0055 if options.killRunning:
0056     sql += "AND jobStatus=:jobStatus "
0057     varMap[":jobStatus"] = "running"
0058 if options.cloud is not None:
0059     sql += "AND cloud=:cloud "
0060     varMap[":cloud"] = options.cloud
0061 if options.site is not None:
0062     sql += "AND computingSite=:site "
0063     varMap[":site"] = options.site
0064 for table in [
0065     "ATLAS_PANDA.jobsActive4",
0066     "ATLAS_PANDA.jobsDefined4",
0067 ]:
0068     status, res = proxyS.querySQLS(sql % table, varMap)
0069     if res is not None:
0070         for id, prio in res:
0071             if prio not in jobsMap:
0072                 jobsMap[prio] = []
0073             if id not in jobsMap[prio]:
0074                 jobsMap[prio].append(id)
0075
0076 # order by PandaID and currentPriority
0077 jobs = []
0078 prioList = sorted(jobsMap)
0079 for prio in prioList:
0080     # reverse order by PandaID to kill newer jobs
0081     ids = sorted(jobsMap[prio])
0082     ids.reverse()
0083     jobs += ids
0084
0085 if options.maxJobs is not None:
0086     jobs = jobs[: int(options.maxJobs)]
0087
0088 print(f"The number of jobs with priorities below {args[0]} : {len(jobs)}")
0089 if len(jobs):
0090     nJob = 100
0091     iJob = 0
0092     while iJob < len(jobs):
0093         print(f"kill {str(jobs[iJob:iJob + nJob])}")
0094         if options.forceKill:
0095             Client.kill_jobs(jobs[iJob : iJob + nJob], 9)
0096         else:
0097             Client.kill_jobs(jobs[iJob : iJob + nJob])
0098         iJob += nJob
0099         time.sleep(1)