-
Notifications
You must be signed in to change notification settings - Fork 967
/
resubmit.py
66 lines (59 loc) · 2.91 KB
/
resubmit.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import logging
from galaxy import model
__all__ = ['failure']
log = logging.getLogger(__name__)
MESSAGES = dict(
walltime_reached = 'it reached the walltime',
memory_limit_reached = 'it exceeded the amount of allocated memory'
)
def failure(app, job_runner, job_state):
if (getattr(job_state, 'runner_state', None)
and job_state.runner_state in
(job_state.runner_states.WALLTIME_REACHED,
job_state.runner_states.MEMORY_LIMIT_REACHED)):
# Intercept jobs that hit the walltime and have a walltime or
# nonspecific resubmit destination configured
for resubmit in job_state.job_destination.get('resubmit'):
if (resubmit.get('condition', None) and resubmit['condition'] !=
job_state.runner_state):
# There is a resubmit defined for the destination but
# its condition is not for walltime_reached
continue
log.info("(%s/%s) Job will be resubmitted to '%s' because %s at "
"the '%s' destination",
job_state.job_wrapper.job_id,
job_state.job_id,
resubmit['destination'],
MESSAGES[job_state.runner_state],
job_state.job_wrapper.job_destination.id )
# fetch JobDestination for the id or tag
new_destination = app.job_config.get_destination(
resubmit['destination'])
# Resolve dynamic if necessary
new_destination = (job_state.job_wrapper.job_runner_mapper
.cache_job_destination(new_destination))
# Reset job state
job = job_state.job_wrapper.get_job()
if resubmit.get('handler', None):
log.debug('(%s/%s) Job reassigned to handler %s',
job_state.job_wrapper.job_id, job_state.job_id,
resubmit['handler'])
job.set_handler(resubmit['handler'])
job_runner.sa_session.add( job )
# Is this safe to do here?
job_runner.sa_session.flush()
# Cache the destination to prevent rerunning dynamic after
# resubmit
job_state.job_wrapper.job_runner_mapper \
.cached_job_destination = new_destination
job_state.job_wrapper.set_job_destination(new_destination)
# Clear external ID (state change below flushes the change)
job.job_runner_external_id = None
# Allow the UI to query for resubmitted state
if job.params is None:
job.params = {}
job_state.runner_state_handled = True
info = "This job was resubmitted to the queue because %s on its " \
"compute resource." % MESSAGES[job_state.runner_state])
job_runner.mark_as_resubmitted(job_state, info=info)
return