Merge pull request #25149 from jacksontj/saltnado

Saltnado multiprocess support

salt/netapi/rest_tornado/__init__.py

@@ -46,6 +46,12 @@ def start():
     if 'num_processes' not in mod_opts:
         mod_opts['num_processes'] = 1
 
+    if mod_opts['num_processes'] > 1 and mod_opts.get('debug', False) is True:
+        raise Exception((
+            'Tornado\'s debug implementation is not compatible with multiprocess. '
+            'Either disable debug, or set num_processes to 1.'
+        ))
+
     paths = [
         (r"/", saltnado.SaltAPIHandler),
         (r"/login", saltnado.SaltAuthHandler),
@@ -81,7 +87,6 @@ def start():
     application.opts = __opts__
     application.mod_opts = mod_opts
     application.auth = salt.auth.LoadAuth(__opts__)
-    application.event_listener = saltnado.EventListener(mod_opts, __opts__)
 
     # the kwargs for the HTTPServer
     kwargs = {}
@@ -107,7 +112,7 @@ def start():
                          )
         http_server.start(mod_opts['num_processes'])
     except:
-        print('Rest_tornado unable to bind to port {0}'.format(mod_opts['port']))
+        logger.error('Rest_tornado unable to bind to port {0}'.format(mod_opts['port']), exc_info=True)
         raise SystemExit(1)
 
     try:

salt/netapi/rest_tornado/saltnado.py

@@ -257,7 +257,8 @@ def __init__(self, mod_opts, opts):
             'master',
             opts['sock_dir'],
             opts['transport'],
-            opts=opts)
+            opts=opts,
+        )
 
         self.event.subscribe()  # start listening for events immediately
 
@@ -270,8 +271,10 @@ def __init__(self, mod_opts, opts):
         # map of future -> timeout_callback
         self.timeout_map = {}
 
-        self.stream = zmqstream.ZMQStream(self.event.sub,
-                                          io_loop=tornado.ioloop.IOLoop.current())
+        self.stream = zmqstream.ZMQStream(
+            self.event.sub,
+            io_loop=tornado.ioloop.IOLoop.current(),
+        )
         self.stream.on_recv(self._handle_event_socket_recv)
 
     def clean_timeout_futures(self, request):
@@ -390,6 +393,17 @@ def _verify_client(self, client):
             self.write("We don't serve your kind here")
             self.finish()
 
+    def initialize(self):
+        '''
+        Initialize the handler before requests are called
+        '''
+        if not hasattr(self.application, 'event_listener'):
+            logger.critical('init a listener')
+            self.application.event_listener =  EventListener(
+                self.application.mod_opts,
+                self.application.opts,
+            )
+
     @property
     def token(self):
         '''