Capture and forward subprocess output

Send all process output to a file

For easier retrieval of output on failure

Author: minrk

ipyparallel/apps/baseapp.py

@@ -202,6 +202,8 @@ def reinit_logging(self):
             # Start logging to the new log file
             log_filename = self.name + u'-' + str(os.getpid()) + u'.log'
             logfile = os.path.join(log_dir, log_filename)
+            if sys.__stderr__:
+                print(f"Sending logs to {logfile}", file=sys.__stderr__)
             open_log_file = open(logfile, 'w')
         else:
             open_log_file = None

ipyparallel/cluster/_win32support.py

@@ -250,15 +250,14 @@ def _classes_default(self):
     daemonize = Bool(
         False,
         config=True,
-        help="""Daemonize the ipcluster program. This implies --log-to-file.
+        help="""Launch the cluster and immediately exit.
+
+        .. versionchanged:: 7.0
+            No longer leaves the ipcluster process itself running.
+            Prior to 7.0, --daemonize did not work on Windows.
         """,
     )
 
-    @observe('daemonize')
-    def _daemonize_changed(self, change):
-        if change['new']:
-            self.log_to_file = True
-
     early_shutdown = Integer(
         30,
         config=True,
@@ -382,6 +381,9 @@ def engines_stopped_early(self, stop_data):
             a public interface.
             """
             )
+            engine_output = self.engine_launcher.get_output(remove=True)
+            if engine_output:
+                self.log.error(f"Engine output:\n{engine_output}")
             self.loop.add_callback(self.stop_cluster)
 
         return self.engines_stopped(stop_data)
@@ -471,7 +473,9 @@ def engines_stopped(self, r):
     async def start_cluster(self):
         await self.cluster.start_cluster()
         if self.daemonize:
-            self.log.info(f"Leaving cluster running: {self.cluster.cluster_file}")
+            print(
+                f"Leaving cluster running: {self.cluster.cluster_file}", file=sys.stderr
+            )
             self.loop.add_callback(self.loop.stop)
         self.cluster._controller.on_stop(self.controller_stopped)
         self.watch_engines()

ipyparallel/cluster/app.py

@@ -357,7 +357,9 @@ def from_dict(cls, d, **kwargs):
             cls = self.engine_launcher_class = import_item(engine_info["class"])
             for engine_set_id, engine_state in engine_info.get("sets", {}).items():
                 self._engine_sets[engine_set_id] = cls.from_dict(
-                    engine_state, parent=self
+                    engine_state,
+                    engine_set_id=engine_set_id,
+                    parent=self,
                 )
 
         return self
@@ -503,6 +505,7 @@ async def start_engines(self, n=None, engine_set_id=None, **kwargs):
             log=self.log,
             profile_dir=self.profile_dir,
             cluster_id=self.cluster_id,
+            engine_set_id=engine_set_id,
             **kwargs,
         )
         if n is None:
@@ -547,6 +550,8 @@ async def stop_engines(self, engine_set_id=None):
         r = engine_set.stop()
         if inspect.isawaitable(r):
             await r
+        # retrieve and cleanup output files
+        engine_set.get_output(remove=True)
         self._engine_sets.pop(engine_set_id)
         self.update_cluster_file()
 
@@ -608,6 +613,9 @@ async def stop_controller(self):
             if inspect.isawaitable(r):
                 await r
 
+        if self._controller:
+            self._controller.get_output(remove=True)
+
         self._controller = None
         self.update_cluster_file()