[2617] separated real recv error and close-on-recv; treat the latter as debug.

src/bin/msgq/msgq.py.in

@@ -70,6 +70,13 @@ SPECFILE_LOCATION = SPECFILE_PATH + "/msgq.spec"
 
 class MsgQReceiveError(Exception): pass
 
+class MsgQCloseOnReceive(Exception):
+    def __init__(self, reason, partial_read):
+        self.partial_read = partial_read
+        self.__reason = reason
+    def __str__(self):
+        return self.__reason
+
 class SubscriptionManager:
     def __init__(self, cfgmgr_ready):
         """
@@ -332,7 +339,7 @@ class MsgQ:
             del self.sendbuffs[fd]
         logger.debug(TRACE_BASIC, MSGQ_SOCK_CLOSE, fd)
 
-    def getbytes(self, fd, sock, length):
+    def __getbytes(self, fd, sock, length, continued):
         """Get exactly the requested bytes, or raise an exception if
            EOF."""
         received = b''
@@ -340,16 +347,19 @@ class MsgQ:
             try:
                 data = sock.recv(length - len(received))
             except socket.error as err:
+                if err.errno == errno.ECONNRESET:
+                    raise MsgQCloseOnReceive(str(err), continued)
                 raise MsgQReceiveError(str(err))
             if len(data) == 0:
-                raise MsgQReceiveError("EOF")
+                raise MsgQCloseOnReceive("EOF", continued)
             received += data
+            continued = True
         return received
 
     def read_packet(self, fd, sock):
         """Read a correctly formatted packet.  Will raise exceptions if
            something fails."""
-        lengths = self.getbytes(fd, sock, 6)
+        lengths = self.__getbytes(fd, sock, 6, False)
         overall_length, routing_length = struct.unpack(">IH", lengths)
         if overall_length < 2:
             raise MsgQReceiveError("overall_length < 2")
@@ -360,9 +370,9 @@ class MsgQ:
             raise MsgQReceiveError("routing_length == 0")
         data_length = overall_length - routing_length
         # probably need to sanity check lengths here...
-        routing = self.getbytes(fd, sock, routing_length)
+        routing = self.__getbytes(fd, sock, routing_length, True)
         if data_length > 0:
-            data = self.getbytes(fd, sock, data_length)
+            data = self.__getbytes(fd, sock, data_length, True)
         else:
             data = None
         return (routing, data)
@@ -371,8 +381,11 @@ class MsgQ:
         """Process one packet."""
         try:
             routing, data = self.read_packet(fd, sock)
-        except MsgQReceiveError as err:
-            logger.error(MSGQ_RECV_ERR, fd, err)
+        except (MsgQReceiveError, MsgQCloseOnReceive) as err:
+            if isinstance(err, MsgQCloseOnReceive) and not err.partial_read:
+                logger.debug(TRACE_BASIC, MSGQ_CLOSE_ON_RECV, fd)
+            else:
+                logger.error(MSGQ_RECV_ERR, fd, err)
             self.kill_socket(fd, sock)
             return
 

src/bin/msgq/msgq_messages.mes

@@ -70,7 +70,11 @@ the authors when figuring the problem out.
 
 % MSGQ_RECV_ERR Error reading from socket %1: %2
 There was a low-level error when reading from a socket. The error is logged and
-the corresponding socket is dropped.
+the corresponding socket is dropped.  The errors include receiving
+broken or (non empty but) incomplete data.  In either case it suggests
+something unexpected happens within the BIND 10 system; it's probably
+better to restart the system, and if it continues it should be
+reported as a bug.
 
 % MSGQ_RECV_HDR Received header: %1
 Debug message. This message includes the whole routing header of a packet.
@@ -123,3 +127,9 @@ data structure.
 % MSGQ_SUBS_NEW_TARGET Creating new target for subscription to group '%1' for instance '%2'
 Debug message. Creating a new subscription. Also creating a new data structure
 to hold it.
+
+% MSGQ_CLOSE_ON_RECV Reading from socket canceled as it's closed: FD=%1
+A debug message.  The msgq daemon was notified of a read event on a
+socket, but its initial read operation failed because the remote
+client has closed its socket.  This is possible in a normal operation
+when a module shuts down.

src/bin/msgq/tests/msgq_test.py

@@ -572,6 +572,7 @@ class SocketTests(unittest.TestCase):
         '''A mock socket used instead of standard socket objects.'''
         def __init__(self):
             self.ex_on_send = None # raised from send() if not None
+            self.recv_result = b'test' # dummy data or exception
             self.blockings = [] # history of setblocking() params
         def setblocking(self, on):
             self.blockings.append(on)
@@ -579,6 +580,12 @@ class SocketTests(unittest.TestCase):
             if self.ex_on_send is not None:
                 raise self.ex_on_send
             return 10           # arbitrary choice
+        def recv(self, len):
+            if isinstance(self.recv_result, Exception):
+                raise self.recv_result
+            ret = self.recv_result
+            self.recv_result = b'' # if called again, return empty data
+            return ret
         def fileno(self):
             return 42           # arbitrary choice
 
@@ -587,6 +594,7 @@ class SocketTests(unittest.TestCase):
         def __init__(self, logger):
             self.error_called = 0
             self.warn_called = 0
+            self.debug_called = 0
             self.orig_logger = logger
         def error(self, *args):
             self.error_called += 1
@@ -594,6 +602,9 @@ class SocketTests(unittest.TestCase):
         def warn(self, *args):
             self.warn_called += 1
             self.orig_logger.warn(*args)
+        def debug(self, *args):
+            self.debug_called += 1
+            self.orig_logger.debug(*args)
 
     def mock_kill_socket(self, fileno, sock):
         '''A replacement of MsgQ.kill_socket method for inspection.'''
@@ -687,6 +698,34 @@ class SocketTests(unittest.TestCase):
         # This just shouldn't crash
         self.__msgq._process_fd(4200, True, False, False)
 
+    def test_process_packet(self):
+        '''Check some failure cases in handling an incoming message.'''
+        expected_errors = 0
+        expected_debugs = 0
+
+        for eno in [errno.ENOBUFS, errno.ECONNRESET]:
+            self.__sock_error.errno = eno
+            self.__sock.recv_result = self.__sock_error
+            self.__msgq.process_packet(42, self.__sock)
+            self.assertEqual((42, self.__sock), self.__killed_socket)
+            if eno == errno.ENOBUFS:
+                expected_errors += 1
+            else:
+                expected_debugs += 1
+            self.assertEqual(expected_errors, self.__logger.error_called)
+            self.assertEqual(expected_debugs, self.__logger.debug_called)
+
+        for recv_data in [b'', b'short']:
+            self.__sock.recv_result = recv_data
+            self.__msgq.process_packet(42, self.__sock)
+            self.assertEqual((42, self.__sock), self.__killed_socket)
+            if len(recv_data) == 0:
+                expected_debugs += 1
+            else:
+                expected_errors += 1
+            self.assertEqual(expected_errors, self.__logger.error_called)
+            self.assertEqual(expected_debugs, self.__logger.debug_called)
+
 if __name__ == '__main__':
     isc.log.resetUnitTestRootLogger()
     unittest.main()