Manages salt service during watchmaker install

src/watchmaker/managers/base.py

@@ -14,7 +14,7 @@
 import zipfile
 
 from six import add_metaclass
-from six.moves import urllib
+from six.moves import queue, urllib
 
 from watchmaker.exceptions import WatchmakerException
 
@@ -188,21 +188,33 @@ def create_working_dir(self, basedir, prefix):
         return working_dir
 
     @staticmethod
-    def _pipe_logger(pipe, logger, prefix_msg=''):
+    def _pipe_logger(pipe, logger, prefix_msg='', pipe_queue=None):
         try:
             for line in iter(pipe.readline, b''):
                 logger('%s%s', prefix_msg, line.rstrip())
+                if pipe_queue:
+                    pipe_queue.put(line)
         finally:
             pipe.close()
 
-    def call_process(self, cmd):
+    def call_process(self, cmd, stdout=False):
         """
         Execute a shell command.
 
         Args:
             cmd (:obj:`list`):
                 Command to execute.
+            stdout (:obj:`bool`):
+                (Defaults to ``False``) Switch to control whether to return
+                stdout.
+
+        Returns:
+            :obj:`None` unless ``stdout`` is ``True``. In that case, the stdout
+            is returned.
         """
+        ret = None
+        stdout_queue = queue.Queue() if stdout else None
+
         if not isinstance(cmd, list):
             msg = 'Command is not a list: {0}'.format(cmd)
             self.log.critical(msg)
@@ -217,7 +229,11 @@ def call_process(self, cmd):
 
         stdout_reader = threading.Thread(
             target=self._pipe_logger,
-            args=(process.stdout, self.log.debug, 'Command stdout: '))
+            args=(
+                process.stdout,
+                self.log.debug,
+                'Command stdout: ',
+                stdout_queue))
         stdout_reader.daemon = True
         stdout_reader.start()
 
@@ -238,6 +254,19 @@ def call_process(self, cmd):
             self.log.critical(msg)
             raise WatchmakerException(msg)
 
+        if stdout_queue:
+            # Return stdout
+            ret = b''
+            while not stdout_queue.empty():
+                try:
+                    ret = ret + stdout_queue.get(False)
+                except queue.Empty:
+                    continue
+                stdout_queue.task_done()
+            stdout_queue.join()
+
+        return ret
+
     def cleanup(self):
         """Delete working directory."""
         self.log.info('Cleanup Time...')

src/watchmaker/workers/salt.py

@@ -228,7 +228,7 @@ def _set_grain(self, grain, value):
         ]
         self.run_salt(cmd)
 
-    def run_salt(self, command):
+    def run_salt(self, command, stdout=False):
         """
         Execute salt command.
 
@@ -238,6 +238,7 @@ def run_salt(self, command):
                 Watchmaker will always begin the command with the options
                 ``--local``, ``--retcode-passthrough``, and ``--no-color``, so
                 do not specify those options in the command.
+            stdout (obj:`bool`) Switch to control whether to return stdout.
         """
         cmd = [
             self.salt_call,
@@ -249,7 +250,63 @@ def run_salt(self, command):
             cmd.extend(command)
         else:
             cmd.append(command)
-        self.call_process(cmd)
+        ret = self.call_process(cmd, stdout=stdout)
+        if stdout:
+            return ret
+
+    def get_service_status(self, service):
+        """
+        Get the service status using salt.
+
+        Args:
+            service (obj:`str`): Name of the service to query.
+
+        Returns:
+            obj:`bool`. ``True`` if the service is running. ``False`` if the
+            service is not running or not present.
+        """
+        cmd = [
+            'service.status', service,
+            '--out', 'newline_values_only'
+        ]
+        ret = self.run_salt(cmd, stdout=True)
+        return ret.strip().lower() == b'true'
+
+    def stop_service(self, service):
+        """
+        Stop a service status using salt.
+
+        Args:
+            service (obj:`str`): Name of the service to stop.
+
+        Returns:
+            obj:`bool`. ``True`` if the service was stopped. ``False`` if the
+            service could not be stopped.
+        """
+        cmd = [
+            'service.stop', service,
+            '--out', 'newline_values_only'
+        ]
+        ret = self.run_salt(cmd, stdout=True)
+        return ret.strip().lower() == b'true'
+
+    def start_service(self, service):
+        """
+        Start a service status using salt.
+
+        Args:
+            service (obj:`str`): Name of the service to start.
+
+        Returns:
+            obj:`bool`. ``True`` if the service was started. ``False`` if the
+            service could not be started.
+        """
+        cmd = [
+            'service.start', service,
+            '--out', 'newline_values_only'
+        ]
+        ret = self.run_salt(cmd, stdout=True)
+        return ret.strip().lower() == b'true'
 
     def process_grains(self):
         """Set salt grains."""
@@ -427,8 +484,17 @@ def install(self):
         """Install salt and execute salt states."""
         self._configuration_validation()
         self._prepare_for_install()
+
+        status_salt = False
+        if os.path.exists(self.salt_call):
+            status_salt = self.get_service_status('salt-minion')
         self._install_package()
+        stopped_salt = self.stop_service('salt-minion')
         self._build_salt_formula(self.salt_srv)
+        if status_salt and stopped_salt:
+            started_salt = self.start_service('salt-minion')
+            if not started_salt:
+                self.log.error('Failed to restart salt-minion service')
 
         self.process_grains()
         self.process_states(self.salt_states)
@@ -524,8 +590,17 @@ def _set_grain(self, grain, value):
     def install(self):
         """Install salt and execute salt states."""
         self._prepare_for_install()
+
+        status_salt = False
+        if os.path.exists(self.salt_call):
+            status_salt = self.get_service_status('salt-minion')
         self._install_package()
+        stopped_salt = self.stop_service('salt-minion')
         self._build_salt_formula(self.salt_srv)
+        if status_salt and stopped_salt:
+            started_salt = self.start_service('salt-minion')
+            if not started_salt:
+                self.log.error('Failed to restart salt-minion service')
 
         if self.ash_role and self.ash_role != 'None':
             role = {'role': str(self.ash_role)}