Even more logging.

 - Log every upload not just dynamically collected files (more cohesive with download logging).
 - Log all staging actions (in and out) with a complete description of FileAction.
 - Log full description of FileAction object on each staging up/staging down failure.

Requested by @natefoo.

Author: jmchilton

pulsar/client/action_mapper.py

@@ -282,6 +282,16 @@ def staging_needed(self):
     def staging_action_local(self):
         return self.staging == STAGING_ACTION_LOCAL
 
+    def to_dict(self):
+        return dict(action_type=self.action_type)
+
+    def __str__(self):
+        as_dict = self.to_dict()
+        attribute_str = ""
+        for key, value in as_dict.items():
+            attribute_str += "%s=%s" % (key, value)
+        return "FileAction[%s]" % attribute_str
+
 
 class NoneAction(BaseAction):
     """ This action indicates the corresponding path does not require any

pulsar/client/staging/down.py

@@ -111,7 +111,7 @@ def __collect_other_working_directory_files(self):
             if name in self.downloaded_working_directory_files:
                 continue
             if self.client_outputs.dynamic_match(name):
-                log.info("collecting %s" % name)
+                log.debug("collecting dynamic output %s" % name)
                 output_file = join(working_directory, self.pulsar_outputs.path_helper.local_name(name))
                 if self._attempt_collect_output(output_type='output_workdir', path=output_file, name=name):
                     self.downloaded_working_directory_files.append(name)
@@ -129,6 +129,7 @@ def _attempt_collect_output(self, output_type, path, name=None):
         return collected
 
     def _collect_output(self, output_type, action, name):
+        log.info("collecting output %s with action %s" % (name, action))
         return self.output_collector.collect_output(self, output_type, action, name)
 
 

pulsar/managers/staging/post.py

@@ -55,7 +55,8 @@ def collect_output(self, results_collector, output_type, action, name):
             name = os.path.basename(action.path)
 
         pulsar_path = self.job_directory.calculate_path(name, output_type)
-        self.action_executor.execute(lambda: action.write_from_path(pulsar_path))
+        description = "staging out file %s via %s" % (pulsar_path, action)
+        self.action_executor.execute(lambda: action.write_from_path(pulsar_path), description)
 
 
 def __pulsar_outputs(job_directory):

pulsar/managers/staging/pre.py

@@ -12,7 +12,8 @@ def preprocess(job_directory, setup_actions, action_executor):
         input_type = setup_action["type"]
         action = from_dict(setup_action["action"])
         path = job_directory.calculate_path(name, input_type)
-        log.debug("Staging %s '%s' via %s to %s", input_type, name, action.__class__.__name__, path)
-        action_executor.execute(lambda: action.write_to_path(path))
+        description = "Staging %s '%s' via %s to %s" % (input_type, name, action, path)
+        log.debug(description)
+        action_executor.execute(lambda: action.write_to_path(path), "action[%s]" % description)
 
 __all__ = ['preprocess']

pulsar/managers/util/retry.py

@@ -27,13 +27,16 @@ def __init__(self, **kwds):
         self.errback = kwds.get("errback", self.__default_errback)
         self.catch = kwds.get("catch", DEFAULT_CATCH)
 
-        self.description = kwds.get("description", DEFAULT_DESCRIPTION)
+        self.default_description = kwds.get("description", DEFAULT_DESCRIPTION)
 
-    def execute(self, action):
+    def execute(self, action, description=None):
         def on_error(exc, intervals, retries, interval=0):
             interval = next(intervals)
             if self.errback:
-                self.errback(exc, interval)
+                errback_args = [exc, interval]
+                if description is not None:
+                    errback_args.append(description)
+                self.errback(exc, interval, description)
             return interval
 
         return _retry_over_time(
@@ -46,10 +49,11 @@ def on_error(exc, intervals, retries, interval=0):
             errback=on_error,
         )
 
-    def __default_errback(self, exc, interval):
+    def __default_errback(self, exc, interval, description=None):
+        description = description or self.default_description
         log.info(
             "Failed to execute %s, retrying in %s seconds.",
-            self.description,
+            description,
             interval,
             exc_info=True
         )

test/retry_action_test.py

@@ -25,7 +25,7 @@ def test_third_execution_fine():
         RetryActionExecutor(max_retries=2, interval_start=.01, interval_step=.01).execute(action_tracker.execute)
     except Exception:
         exception_raised = True
-    assert action_tracker.count == 3
+    assert action_tracker.count == 3, action_tracker.count
     assert not exception_raised