Merge pull request #3849 from jmchilton/improved_scheduling_delay_logging

Log reason when workflow scheduling of a step is delayed.

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):

lib/galaxy/workflow/run.py

@@ -176,9 +176,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
-                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.",
@@ -187,8 +187,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))
+            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
@@ -218,14 +218,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()
@@ -303,7 +305,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 ]
@@ -324,7 +327,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 ):
@@ -349,7 +353,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 of invocation %s delayed (%s)" % (step.id, self.workflow_invocation.id, why)
+            log.debug(message)
         self.outputs[ step.id ] = STEP_OUTPUT_DELAYED
 
     def _subworkflow_invocation(self, step):
@@ -406,8 +413,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' )

test/api/test_workflows.py

@@ -857,8 +857,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