Merge pull request #25 from Mirantis/stop
Stop execution by marking all tasks SKIPPED
This commit is contained in:
commit
69927ee69a
@ -39,10 +39,11 @@ def update(uid, plan):
|
||||
@click.argument('uid')
|
||||
def report(uid):
|
||||
colors = {
|
||||
'PENDING': 'blue',
|
||||
'PENDING': 'cyan',
|
||||
'ERROR': 'red',
|
||||
'SUCCESS': 'green',
|
||||
'INPROGRESS': 'yellow'}
|
||||
'INPROGRESS': 'yellow',
|
||||
'SKIPPED': 'blue'}
|
||||
|
||||
report = graph.report_topo(uid)
|
||||
for item in report:
|
||||
@ -81,7 +82,14 @@ def stop(uid):
|
||||
# using revoke(terminate=True) will lead to inability to restart execution
|
||||
# research possibility of customizations of
|
||||
# app.control and Panel.register in celery
|
||||
graph.soft_stop(uid)
|
||||
tasks.soft_stop.apply_async(args=[uid], queue='scheduler')
|
||||
|
||||
|
||||
@orchestration.command()
|
||||
@click.argument('uid')
|
||||
def resume(uid):
|
||||
graph.reset(uid, ['SKIPPED'])
|
||||
tasks.schedule_start.apply_async(args=[uid], queue='scheduler')
|
||||
|
||||
|
||||
@orchestration.command()
|
||||
@ -97,10 +105,11 @@ def dg(uid):
|
||||
plan = graph.get_graph(uid)
|
||||
|
||||
colors = {
|
||||
'PENDING': 'blue',
|
||||
'PENDING': 'cyan',
|
||||
'ERROR': 'red',
|
||||
'SUCCESS': 'green',
|
||||
'INPROGRESS': 'yellow'}
|
||||
'INPROGRESS': 'yellow',
|
||||
'SKIPPED': 'blue'}
|
||||
|
||||
for n in plan:
|
||||
color = colors[plan.node[n]['status']]
|
||||
|
@ -102,14 +102,6 @@ def reset(uid, states=None):
|
||||
save_graph(uid, dg)
|
||||
|
||||
|
||||
def soft_stop(uid):
|
||||
"""Graph will stop when all currently inprogress tasks will be finished
|
||||
"""
|
||||
dg = get_graph(uid)
|
||||
dg.graph['stop'] = True
|
||||
save_graph(uid, dg)
|
||||
|
||||
|
||||
def report_topo(uid):
|
||||
|
||||
dg = get_graph(uid)
|
||||
|
@ -116,10 +116,18 @@ def schedule_start(plan_uid, start=None, end=None):
|
||||
- apply different policies to tasks
|
||||
"""
|
||||
dg = graph.get_graph(plan_uid)
|
||||
dg.graph['stop'] = False
|
||||
schedule(plan_uid, dg)
|
||||
|
||||
|
||||
@app.task
|
||||
def soft_stop(plan_uid):
|
||||
dg = graph.get_graph(plan_uid)
|
||||
for n in dg:
|
||||
if dg.node[n]['status'] == 'PENDING':
|
||||
dg.node[n]['status'] = 'SKIPPED'
|
||||
graph.save_graph(plan_uid, dg)
|
||||
|
||||
|
||||
@app.task
|
||||
def schedule_next(task_id, status, errmsg=None):
|
||||
plan_uid, task_name = task_id.rsplit(':', 1)
|
||||
@ -147,7 +155,7 @@ def traverse(dg):
|
||||
visited = set()
|
||||
for node in dg:
|
||||
data = dg.node[node]
|
||||
if data['status'] not in ('PENDING', 'INPROGRESS'):
|
||||
if data['status'] not in ('PENDING', 'INPROGRESS', 'SKIPPED'):
|
||||
visited.add(node)
|
||||
|
||||
for node in dg:
|
||||
@ -155,7 +163,7 @@ def traverse(dg):
|
||||
|
||||
if node in visited:
|
||||
continue
|
||||
elif data['status'] == 'INPROGRESS':
|
||||
elif data['status'] in ('INPROGRESS', 'SKIPPED'):
|
||||
continue
|
||||
|
||||
predecessors = set(dg.predecessors(node))
|
||||
|
Loading…
Reference in New Issue
Block a user