From 185aa3968bd38cfe021f7bcec9bae02c12ae4c0b Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 29 Mar 2017 22:17:41 -0400 Subject: [PATCH 1/2] Log when workflow scheduling is delayed about why. Currently the logs contain fairly uninformative step delayed messages that don't allow admins to understand why parts of the workflow are being delayed during scheduling. Here are some examples of before and after. When running the test: ``` ./run_tests.sh -api test/api/test_workflows.py:WorkflowsApiTestCase.test_workflow_pause ``` Before these lines would show up: ``` galaxy.workflow.run DEBUG 2017-03-30 09:23:47,327 Workflow step 2 of invocation 1 invoked (174.149 ms) galaxy.workflow.run DEBUG 2017-03-30 09:23:47,347 Workflow step 3 of invocation 1 invoked (19.355 ms) galaxy.workflow.run DEBUG 2017-03-30 09:23:47,364 Workflow step 4 of invocation 1 delayed (16.262 ms) ``` Now these same lines are as follows: ``` galaxy.workflow.run DEBUG 2017-03-30 09:19:28,601 Workflow step 2 of invocation 1 invoked (172.849 ms) galaxy.workflow.run DEBUG 2017-03-30 09:19:28,619 Marking step 3 outputs delayed (executing pause step) galaxy.workflow.run DEBUG 2017-03-30 09:19:28,620 Workflow step 3 of invocation 1 invoked (17.999 ms) galaxy.workflow.run DEBUG 2017-03-30 09:19:28,633 Workflow step 4 of invocation 1 delayed (dependent step [3] delayed, so this step must be delayed) (13.398 ms) ``` Also, when running the test: ``` ./run_tests.sh -api test/api/test_workflows.py:WorkflowsApiTestCase.test_workflow_run_dynamic_output_collections_3 ``` Before these lines would be printed: ``` galaxy.workflow.run DEBUG 2017-03-30 09:25:35,910 Workflow step 4 of invocation 1 invoked (281.479 ms) galaxy.workflow.run DEBUG 2017-03-30 09:25:35,937 Workflow step 5 of invocation 1 delayed (25.904 ms) galaxy.workflow.run DEBUG 2017-03-30 09:25:35,970 Workflow step 6 of invocation 1 delayed (32.597 ms) ``` Now these same lines are as follows: ``` galaxy.workflow.run DEBUG 2017-03-30 09:27:54,270 Workflow step 4 of invocation 1 invoked (295.826 ms) galaxy.workflow.run DEBUG 2017-03-30 09:27:54,315 Workflow step 5 of invocation 1 delayed (dependent collection [1] not yet populated with datasets) (44.581 ms) galaxy.workflow.run DEBUG 2017-03-30 09:27:54,326 Workflow step 6 of invocation 1 delayed (dependent step [5] delayed, so this step must be delayed) (10.804 ms) ``` When running workflows with steps that depend explicitly on other steps (instead of implicitly between dataset connections), lines such as: ``` galaxy.workflow.run DEBUG 2017-03-30 09:33:52,966 Marking step 3 outputs delayed (workflow paused at this step waiting for review) galaxy.workflow.run DEBUG 2017-03-30 09:33:53,001 Workflow step 4 of invocation 1 delayed (dependent step [3] delayed, so this step must be delayed) (0.157 ms) galaxy.workflow.run DEBUG 2017-03-30 09:33:53,001 Workflow step 5 of invocation 1 delayed (depends on step [4] but that step has not been invoked yet) (0.084 ms) ``` and ``` galaxy.workflow.run DEBUG 2017-03-30 09:33:57,090 Workflow step 5 of invocation 1 delayed (depends on step [4] but one or more jobs created from that step have not finished yet) (0.144 ms) ``` now may appear. --- lib/galaxy/workflow/modules.py | 12 ++++++++---- lib/galaxy/workflow/run.py | 28 +++++++++++++++++++--------- test/api/test_workflows.py | 2 -- 3 files changed, 27 insertions(+), 15 deletions(-) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 87fc5894eb5..fd6571660f7 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -508,7 +508,7 @@ class PauseModule( WorkflowModule ): return state def execute( self, trans, progress, invocation, step ): - progress.mark_step_outputs_delayed( step ) + progress.mark_step_outputs_delayed( step, why="executing pause step" ) return None def recover_mapping( self, step, step_invocations, progress ): @@ -522,7 +522,8 @@ class PauseModule( WorkflowModule ): return elif action is False: raise CancelWorkflowEvaluation() - raise DelayedWorkflowEvaluation() + delayed_why = "workflow paused at this step waiting for review" + raise DelayedWorkflowEvaluation(why=delayed_why) def do_invocation_step_action( self, step, action ): """ Update or set the workflow invocation state action - generic @@ -834,7 +835,8 @@ class ToolModule( WorkflowModule ): workflow_invocation_uuid=invocation.uuid.hex ) except ToolInputsNotReadyException: - raise DelayedWorkflowEvaluation() + delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id + raise DelayedWorkflowEvaluation(why=delayed_why) if collection_info: step_outputs = dict( execution_tracker.implicit_collections ) @@ -1020,7 +1022,9 @@ def load_module_sections( trans ): class DelayedWorkflowEvaluation(Exception): - pass + + def __init__(self, why=None): + self.why = why class CancelWorkflowEvaluation(Exception): diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 190c2d9d67e..0619f19d6df 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -147,6 +147,7 @@ class WorkflowInvoker( object ): workflow_invocation = self.workflow_invocation remaining_steps = self.progress.remaining_steps() delayed_steps = False + more_info = "" for step in remaining_steps: step_delayed = False step_timer = ExecutionTimer() @@ -165,8 +166,9 @@ class WorkflowInvoker( object ): # https://github.com/galaxyproject/galaxy/issues/2259 if job: workflow_invocation_step.job_id = job.id - except modules.DelayedWorkflowEvaluation: + except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True + more_info = "(%s)" % de.why self.progress.mark_step_outputs_delayed( step ) except Exception: log.exception( @@ -177,7 +179,8 @@ class WorkflowInvoker( object ): raise step_verb = "invoked" if not step_delayed else "delayed" - log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, step_verb, step_timer)) + status = step_verb + (" %s" % more_info if more_info else "") + log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, status, step_timer)) if delayed_steps: state = model.WorkflowInvocation.states.READY @@ -207,14 +210,16 @@ class WorkflowInvoker( object ): # No steps created yet - have to delay evaluation. if not step_invocations: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but that step has not been invoked yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) for step_invocation in step_invocations: job = step_invocation.job if job: # At least one job in incomplete. if not job.finished: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but one or more jobs created from that step have not finished yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) if job.state != job.states.OK: raise modules.CancelWorkflowEvaluation() @@ -292,7 +297,8 @@ class WorkflowProgress( object ): raise Exception(message) step_outputs = self.outputs[ output_step_id ] if step_outputs is STEP_OUTPUT_DELAYED: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent step [%s] delayed, so this step must be delayed" % output_step_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) output_name = connection.output_name try: replacement = step_outputs[ output_name ] @@ -313,7 +319,8 @@ class WorkflowProgress( object ): # TODO: consider distinguish between cancelled and failed? raise modules.CancelWorkflowEvaluation() - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent collection [%s] not yet populated with datasets" % replacement.id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) return replacement def get_replacement_workflow_output( self, workflow_output ): @@ -338,7 +345,10 @@ class WorkflowProgress( object ): def set_step_outputs(self, step, outputs): self.outputs[ step.id ] = outputs - def mark_step_outputs_delayed(self, step): + def mark_step_outputs_delayed(self, step, why=None): + if why: + message = "Marking step %s outputs delayed (%s)" % (step.id, why) + log.debug(message) self.outputs[ step.id ] = STEP_OUTPUT_DELAYED def _subworkflow_invocation(self, step): @@ -395,8 +405,8 @@ class WorkflowProgress( object ): def _recover_mapping( self, step, step_invocations ): try: step.module.recover_mapping( step, step_invocations, self ) - except modules.DelayedWorkflowEvaluation: - self.mark_step_outputs_delayed( step ) + except modules.DelayedWorkflowEvaluation as de: + self.mark_step_outputs_delayed( step, de.why ) __all__ = ( 'invoke', 'WorkflowRunConfig' ) diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 38261d06812..2c42a528596 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -869,8 +869,6 @@ test_data: @skip_without_tool( "cat1" ) @skip_without_tool( "collection_paired_test" ) def test_workflow_run_zip_collections( self ): - # A more advanced output collection workflow, testing regression of - # https://github.com/galaxyproject/galaxy/issues/776 history_id = self.dataset_populator.new_history() workflow_id = self._upload_yaml_workflow(""" class: GalaxyWorkflow From b8692309484c3a671145684277201ed7aaf444dd Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 30 Mar 2017 16:32:33 -0400 Subject: [PATCH 2/2] Re-arrange workflow scheduling logic according to @nsoranzo's suggestions. --- lib/galaxy/workflow/run.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 0619f19d6df..2e8848f380b 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -147,7 +147,6 @@ class WorkflowInvoker( object ): workflow_invocation = self.workflow_invocation remaining_steps = self.progress.remaining_steps() delayed_steps = False - more_info = "" for step in remaining_steps: step_delayed = False step_timer = ExecutionTimer() @@ -168,8 +167,7 @@ class WorkflowInvoker( object ): workflow_invocation_step.job_id = job.id except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True - more_info = "(%s)" % de.why - self.progress.mark_step_outputs_delayed( step ) + self.progress.mark_step_outputs_delayed( step, why=de.why ) except Exception: log.exception( "Failed to schedule %s, problem occurred on %s.", @@ -178,9 +176,8 @@ class WorkflowInvoker( object ): ) raise - step_verb = "invoked" if not step_delayed else "delayed" - status = step_verb + (" %s" % more_info if more_info else "") - log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, status, step_timer)) + if not step_delayed: + log.debug("Workflow step %s of invocation %s invoked %s" % (step.id, workflow_invocation.id, step_timer)) if delayed_steps: state = model.WorkflowInvocation.states.READY @@ -347,7 +344,7 @@ class WorkflowProgress( object ): def mark_step_outputs_delayed(self, step, why=None): if why: - message = "Marking step %s outputs delayed (%s)" % (step.id, why) + message = "Marking step %s outputs of invocation %s delayed (%s)" % (step.id, self.workflow_invocation.id, why) log.debug(message) self.outputs[ step.id ] = STEP_OUTPUT_DELAYED