Skip to content
This repository was archived by the owner on Sep 12, 2024. It is now read-only.

Rejector update #576

Merged
sharad1126 merged 3 commits into from
Jun 11, 2020
Merged

Rejector update #576

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b5015a3
added a test mode
drkovalskyi Jun 4, 2020
5c4c05b
enforce single-core in steps where coresponding tasks were single-core
drkovalskyi Jun 5, 2020
01d6810
bug fix
drkovalskyi Jun 9, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
96 changes: 63 additions & 33 deletions Unified/rejector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,9 @@
username = getpass.getuser()

def rejector(url, specific, options=None):

if options.test:
print "Test mode - no changes propagate to the production system"

if not componentInfo(soft=['wtc','jira']).check(): return

Expand Down Expand Up @@ -62,19 +65,30 @@ def rejector(url, specific, options=None):

comment=""
if options.comments: comment = ", reason: "+options.comments
if options.keep:
wfi.sendLog('rejector','invalidating the workflow by unified operator {}{}'.format(username, comment))
if options.test:
if options.keep:
print 'invalidating the workflow by unified operator {}{}'.format(username, comment)
else:
print 'invalidating the workflow and outputs by unified operator {}{}'.format(username, comment)
results = [True]
else:
wfi.sendLog('rejector','invalidating the workflow and outputs by unified operator {}{}'.format(username, comment))
if options.keep:
wfi.sendLog('rejector','invalidating the workflow by unified operator {}{}'.format(username, comment))
else:
wfi.sendLog('rejector','invalidating the workflow and outputs by unified operator {}{}'.format(username, comment))

results = invalidate(url, wfi, only_resub=True, with_output= (not options.keep))
results = invalidate(url, wfi, only_resub=True, with_output= (not options.keep))

if all(results):
print wfo.name,"rejected"
if options and options.clone:
wfo.status = 'trouble'
session.commit()
if not options.test:
wfo.status = 'trouble'
session.commit()
schema = wfi.getSchema()
if options.test:
print "Original schema"
print json.dumps( schema, indent=2 )
schema['Requestor'] = os.getenv('USER')
schema['Group'] = 'DATAOPS'
schema['OriginalRequestName'] = wfo.name
Expand Down Expand Up @@ -206,11 +220,20 @@ def rejector(url, specific, options=None):
if 'Task%d'%step in schema:
sname = 'Step%d'%step
schema[sname] = schema.pop('Task%d'%step)
tmcore = schema[sname].pop('Multicore')
if 'Multicore' in schema[sname] and schema[sname]['Multicore']==1:
# enforce single-core mode assuming that all Tasks with
# Multicore=1 are not thread-safe
tmcore = schema[sname]['Multicore']
else:
# remove explicit assignment of the number of cores
tmcore = schema[sname].pop('Multicore')
tmem = schema[sname].pop('Memory')
if mcore and tmcore != mcore:
wfi.sendLog('rejector','the conversion to stepchain encoutered different value of Multicore %d != %d'%( tmcore, mcore))
sendLog('rejector','the conversion of %s to stepchain encoutered different value of Multicore %d != %d'%( wfo.name, tmcore, mcore), level='critical')
if options.test:
print 'the conversion of %s to stepchain encoutered different value of Multicore %d != %d' % (wfo.name, tmcore, mcore)
else:
wfi.sendLog('rejector','the conversion to stepchain encoutered different value of Multicore %d != %d'%( tmcore, mcore))
sendLog('rejector','the conversion of %s to stepchain encoutered different value of Multicore %d != %d'%( wfo.name, tmcore, mcore), level='critical')
mcore = max(mcore, tmcore)
mem = max(mem, tmem)
schema[sname]['StepName'] = schema[sname].pop('TaskName')
Expand Down Expand Up @@ -239,40 +262,47 @@ def rejector(url, specific, options=None):
break
schema['Multicore'] = mcore
schema['Memory'] = mem
print "New request schema"
print json.dumps( schema, indent=2 )
newWorkflow = reqMgrClient.submitWorkflow(url, schema)
if not newWorkflow:
msg = "Error in cloning {}".format(wfo.name)
print(msg)
wfi.sendLog('rejector',msg)

# Get the error message
time.sleep(5)
data = reqMgrClient.requestManagerPost(url, "/reqmgr2/data/request", schema)
wfi.sendLog('rejector',data)

print json.dumps( schema, indent=2 )
return
print newWorkflow
if not options.test:
newWorkflow = reqMgrClient.submitWorkflow(url, schema)
if not newWorkflow:
msg = "Error in cloning {}".format(wfo.name)
print(msg)
wfi.sendLog('rejector',msg)

# Get the error message
time.sleep(5)
data = reqMgrClient.requestManagerPost(url, "/reqmgr2/data/request", schema)
wfi.sendLog('rejector',data)

data = reqMgrClient.setWorkflowApproved(url, newWorkflow)
print data
wfi.sendLog('rejector','Cloned into %s by unified operator %s'%( newWorkflow, comment ))
#wfi.notifyRequestor('Cloned into %s by unified operator %s'%( newWorkflow, comment ),do_batch=False)
print json.dumps( schema, indent=2 )
return
print newWorkflow

data = reqMgrClient.setWorkflowApproved(url, newWorkflow)
print data
wfi.sendLog('rejector','Cloned into %s by unified operator %s'%( newWorkflow, comment ))
#wfi.notifyRequestor('Cloned into %s by unified operator %s'%( newWorkflow, comment ),do_batch=False)
else:
Comment thread
sharad1126 marked this conversation as resolved.
wfo.status = 'trouble' if options.set_trouble else 'forget'
wfi.notifyRequestor('Rejected by unified operator %s'%( comment ),do_batch=False)
session.commit()
if options.test:
print 'Rejected by unified operator %s'%( comment )
else:
wfo.status = 'trouble' if options.set_trouble else 'forget'
wfi.notifyRequestor('Rejected by unified operator %s'%( comment ),do_batch=False)
session.commit()

else:
Comment thread
sharad1126 marked this conversation as resolved.
msg = "Error in rejecting {}: {}".format(wfo.name,results)
print(msg)
wfi.sendLog('rejector',msg)
msg = "Error in rejecting {}: {}".format(wfo.name, results)
print msg
if not options.test:
wfi.sendLog('rejector', msg)

if __name__ == "__main__":
url = reqmgr_url

parser = optparse.OptionParser()
parser.add_option('-t', '--test', help="test mode - no changes are made", default=False, action="store_true")
parser.add_option('-c','--clone',help="clone the workflow",default=False,action="store_true")
parser.add_option('--comments', help="Give a comment to the clone",default="")
parser.add_option('-k','--keep',help="keep the outpuy in current status", default=False,action="store_true")
Expand Down