Fix up basic ping between master and minion

salt/transport/base.py

@@ -130,7 +130,7 @@ def publish_client(opts, io_loop, host=None, port=None, path=None, transport=Non
         return salt.transport.tcp.TCPPubClient(
             opts, io_loop, host=host, port=port, path=path
         )
-    elif ttype == "tcp":
+    elif ttype == "ws":
         import salt.transport.ws
 
         return salt.transport.ws.PublishClient(

salt/transport/tcp.py

@@ -25,6 +25,7 @@
 import tornado.netutil
 import tornado.tcpclient
 import tornado.tcpserver
+import tornado.util
 
 import salt.master
 import salt.payload
@@ -33,6 +34,7 @@
 import salt.utils.files
 import salt.utils.msgpack
 import salt.utils.platform
+import salt.utils.process
 import salt.utils.versions
 from salt.exceptions import SaltClientError, SaltReqTimeoutError
 from salt.utils.network import ip_bracket
@@ -42,10 +44,6 @@
 else:
     USE_LOAD_BALANCER = False
 
-if USE_LOAD_BALANCER:
-    import tornado.util
-
-    from salt.utils.process import SignalHandlingProcess
 
 log = logging.getLogger(__name__)
 
@@ -134,7 +132,7 @@ def _set_tcp_keepalive(sock, opts):
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 0)
 
 
-class LoadBalancerServer(SignalHandlingProcess):
+class LoadBalancerServer(salt.utils.process.SignalHandlingProcess):
     """
     Raw TCP server which runs in its own process and will listen
     for incoming connections. Each incoming connection will be

salt/transport/ws.py

@@ -172,7 +172,7 @@ async def recv(self, timeout=None):
         try:
             await self._read_in_progress.acquire(timeout=0.001)
         except tornado.gen.TimeoutError:
-            log.error("Timeout Error")
+            log.error("Unable to acquire read lock")
             return
         try:
             if timeout == 0:
@@ -213,10 +213,8 @@ async def recv(self, timeout=None):
                         if raw_msg.data == "close":
                             await self._ws.close()
                     if raw_msg.type == aiohttp.WSMsgType.BINARY:
-                        log.error("ORIG MSG IS %r", raw_msg.data)
                         self.unpacker.feed(raw_msg.data)
                         for msg in self.unpacker:
-                            log.error("MSG IS %r", msg)
                             framed_msg = salt.transport.frame.decode_embedded_strs(msg)
                             return framed_msg["body"]
                     elif raw_msg.type == aiohttp.WSMsgType.ERROR:
@@ -237,7 +235,6 @@ async def handle_on_recv(self, callback):
                 # XXX
                 log.error("Other exception", exc_info=True)
             else:
-                log.error("on recv got msg %r", msg)
                 callback(msg)
 
     def on_recv(self, callback):
@@ -376,9 +373,7 @@ def pre_fork(self, process_manager):
 
     async def handle_request(self, request):
         ws = aiohttp.web.WebSocketResponse()
-        log.error("perpare request")
         await ws.prepare(request)
-        log.error("request prepared")
         self.clients.add(ws)
         while True:
             await asyncio.sleep(1)
@@ -387,7 +382,11 @@ def connect(self):
         log.debug("Connect pusher %s", self.pull_path)
         self.pub_sock = salt.utils.asynchronous.SyncWrapper(
             _TCPPubServerPublisher,
-            (self.pull_path,),
+            (
+                self.pull_host,
+                self.pull_port,
+                self.pull_path,
+            ),
             loop_kwarg="io_loop",
         )
         self.pub_sock.connect()
@@ -404,7 +403,6 @@ async def publish_payload(self, package, *args):
         payload = salt.transport.frame.frame_msg(package)
         for ws in list(self.clients):
             try:
-                log.error("Publish package %r %r", ws, payload)
                 await ws.send_bytes(payload)
             except ConnectionResetError:
                 self.clients.discard(ws)
@@ -471,15 +469,12 @@ async def handle_message(self, request):
             ws = aiohttp.web.WebSocketResponse()
             await ws.prepare(request)
             async for msg in ws:
-                log.error("got msg %r", msg)
                 if msg.type == aiohttp.WSMsgType.TEXT:
                     if msg.data == "close":
                         await ws.close()
                 if msg.type == aiohttp.WSMsgType.BINARY:
                     payload = salt.payload.loads(msg.data)
-                    log.error("Handle message got %r", payload)
                     reply = await self.message_handler(payload)
-                    log.error("Handle message reply %r", reply)
                     await ws.send_bytes(salt.payload.dumps(reply))
                 elif msg.type == aiohttp.WSMsgType.ERROR:
                     log.error("ws connection closed with exception %s", ws.exception())
@@ -505,21 +500,20 @@ def __init__(self, opts, io_loop):  # pylint: disable=W0231
         self._closed = False
 
     async def connect(self):
-        if self.session is None:
-            self.session = aiohttp.ClientSession()
-            URL = self.get_master_uri(self.opts)
-            self.ws = await self.session.ws_connect(URL)
+        # if self.session is None:
+        self.session = aiohttp.ClientSession()
+        URL = self.get_master_uri(self.opts)
+        self.ws = await self.session.ws_connect(URL)
 
     async def send(self, load, timeout=60):
-        if self.sending:
+        if self.sending or self._closing:
             await asyncio.sleep(0.03)
         self.sending = True
         try:
             await self.connect()
             message = salt.payload.dumps(load)
             await self.ws.send_bytes(message)
             async for msg in self.ws:
-                log.error("Got MSG %r", msg)
                 if msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
                     break
                 data = salt.payload.loads(msg.data)
@@ -531,20 +525,22 @@ async def send(self, load, timeout=60):
     async def _close(self):
         if self.ws is not None:
             await self.ws.close()
+            self.ws = None
         if self.session is not None:
             await self.session.close()
+            self.session = None
         self._closed = True
 
     def close(self):
         if self._closing:
             return
         self._closing = True
-        self.io_loop.spawn_callback(self._close)
+        self.close_task = asyncio.create_task(self._close())
 
     @staticmethod
     def get_master_uri(opts):
         if "master_uri" in opts:
-            return opts["master_uri"]
+            return opts["master_uri"].replace("tcp:", "http:", 1)
         return f"http://{opts['master_ip']}:{opts['master_port']}/ws"
 
     # pylint: disable=W1701