Merge pull request #22996 from skizunov/develop

Fix Windows salt-master support for RAET

salt/config.py

@@ -1122,6 +1122,16 @@ def _validate_opts(opts):
                         err.format(key, val, type(val), VALID_OPTS[key])
                     )
 
+    # RAET on Windows uses 'win32file.CreateMailslot()' for IPC. Due to this,
+    # sock_dirs must start with '\\.\mailslot\' and not contain any colons.
+    # We don't expect the user to know this, so we will fix up their path for
+    # them if it isn't compliant.
+    if (salt.utils.is_windows() and opts.get('transport') == 'raet' and
+             'sock_dir' in opts and
+             not opts['sock_dir'].startswith('\\\\.\\mailslot\\')):
+        opts['sock_dir'] = (
+                '\\\\.\\mailslot\\' + opts['sock_dir'].replace(':', ''))
+
     for error in errors:
         log.warning(error)
     if errors:

salt/daemons/flo/core.py

@@ -26,7 +26,7 @@ from raet.lane.stacking import LaneStack
 
 from salt import daemons
 from salt.daemons import salting
-from salt.utils import kinds
+from salt.utils import kinds, is_windows
 from salt.utils.event import tagify
 
 # Import ioflo libs
@@ -76,7 +76,7 @@ class SaltRaetCleanup(ioflo.base.deeding.Deed):
         '''
         Should only run once to cleanup stale lane uxd files.
         '''
-        if self.opts.value.get('sock_dir'):
+        if not is_windows() and self.opts.value.get('sock_dir'):
             sockdirpath = os.path.abspath(self.opts.value['sock_dir'])
             console.concise("Cleaning up uxd files in {0}\n".format(sockdirpath))
             protecteds = self.opts.value.get('raet_cleanup_protecteds', [])

salt/daemons/flo/jobber.py

@@ -24,7 +24,7 @@ from raet import raeting, nacling
 from raet.lane.stacking import LaneStack
 from raet.lane.yarding import RemoteYard
 
-from salt.utils import kinds
+from salt.utils import kinds, is_windows
 from salt.utils.event import tagify
 
 from salt.exceptions import (
@@ -237,12 +237,25 @@ class SaltRaetNixJobber(ioflo.base.deeding.Deed):
                         )
             log.debug('Command details {0}'.format(data))
 
-            process = multiprocessing.Process(
-                    target=self.proc_run,
-                    kwargs={'msg': msg}
-                    )
-            process.start()
-            process.join()
+            if is_windows():
+                # SaltRaetNixJobber is not picklable. Pickling is necessary
+                # when spawning a process in Windows. Since the process will
+                # be spawned and joined on non-Windows platforms, instead of
+                # this, just run the function directly and absorb any thrown
+                # exceptions.
+                try:
+                    self.proc_run(msg)
+                except Exception as exc:
+                    log.error(
+                            'Exception caught by jobber: {0}'.format(exc),
+                            exc_info=True)
+            else:
+                process = multiprocessing.Process(
+                        target=self.proc_run,
+                        kwargs={'msg': msg}
+                        )
+                process.start()
+                process.join()
 
     def proc_run(self, msg):
         '''

salt/daemons/flo/reactor.py

@@ -19,7 +19,7 @@ def reactor_fork(self):
     '''
     Add a reactor object to the process manager
     '''
-    self.proc_mgr.add_process(
+    self.proc_mgr.value.add_process(
             salt.utils.reactor.Reactor,
             args=(self.opts.value,))
 
@@ -33,6 +33,6 @@ def event_return_fork(self):
     '''
     Add a reactor object to the process manager
     '''
-    self.proc_mgr.add_process(
+    self.proc_mgr.value.add_process(
             salt.utils.event.EventReturn,
             args=(self.opts.value,))

salt/utils/raetevent.py

@@ -54,7 +54,7 @@ class RAETEvent(object):
         Prepare the stack objects
         '''
         if not self.stack:
-            if transport.jobber_stack:
+            if hasattr(transport, 'jobber_stack') and transport.jobber_stack:
                 self.stack = transport.jobber_stack
             else:
                 self.stack = transport.jobber_stack = self._setup_stack(ryn=self.ryn)

salt/utils/reactor.py

@@ -138,7 +138,12 @@ class Reactor(multiprocessing.Process, salt.state.Compiler):
         salt.utils.appendproctitle(self.__class__.__name__)
 
         # instantiate some classes inside our new process
-        self.event = salt.utils.event.SaltEvent('master', self.opts['sock_dir'])
+        self.event = salt.utils.event.get_event(
+                'master',
+                self.opts['sock_dir'],
+                self.opts['transport'],
+                opts=self.opts,
+                listen=True)
         self.wrap = ReactWrap(self.opts)
 
         for data in self.event.iter_events(full=True):