GG-33923 IGNITE-15103 Implement logging of connections and queries (#53)

(cherry picked from commit 8fc14f8)

Co-authored-by: Ivan Daschinsky <ivandasch@apache.org>

Author: isapego

pygridgain/connection/aio_connection.py

@@ -21,7 +21,7 @@
 from tzlocal import get_localzone
 
 from pygridgain.constants import PROTOCOLS, PROTOCOL_BYTE_ORDER
-from pygridgain.exceptions import HandshakeError, SocketError, connection_errors
+from pygridgain.exceptions import HandshakeError, SocketError, connection_errors, AuthenticationError
 from .bitmask_feature import BitmaskFeature
 from .connection import BaseConnection
 
@@ -56,7 +56,7 @@ def data_received(self, data: bytes) -> None:
                 hs_response = self.__parse_handshake(packet, self._conn.client)
                 self._handshake_fut.set_result(hs_response)
             else:
-                self._conn.on_message(packet)
+                self._conn.process_message(packet)
             self._buffer = self._buffer[packet_sz:len(self._buffer)]
 
     def __has_full_response(self):
@@ -72,7 +72,7 @@ def __process_connection_error(self, exc):
         connected = self._handshake_fut.done()
         if not connected:
             self._handshake_fut.set_exception(exc)
-        self._conn.on_connection_lost(exc, connected)
+        self._conn.process_connection_lost(exc, connected)
 
     @staticmethod
     def __send_handshake(transport, conn):
@@ -165,38 +165,41 @@ async def _connect(self):
             self.client.protocol_context = ProtocolContext(max(PROTOCOLS), BitmaskFeature.all_supported())
 
         try:
+            self._on_handshake_start()
             result = await self._connect_version()
         except HandshakeError as e:
             if e.expected_version in PROTOCOLS:
                 self.client.protocol_context.version = e.expected_version
                 result = await self._connect_version()
             else:
+                self._on_handshake_fail(e)
                 raise e
-        except connection_errors:
+        except AuthenticationError as e:
+            self._on_handshake_fail(e)
+            raise e
+        except Exception as e:
             # restore undefined protocol version
             if detecting_protocol:
                 self.client.protocol_context = None
-            raise
+            self._on_handshake_fail(e)
+            raise e
 
-        # connection is ready for end user
-        features = BitmaskFeature.from_array(result.get('features', None))
-        self.client.protocol_context.features = features
-        self.uuid = result.get('node_uuid', None)  # version-specific (1.4+)
-        self.failed = False
+        self._on_handshake_success(result)
 
-    def on_connection_lost(self, error, reconnect=False):
+    def process_connection_lost(self, err, reconnect=False):
         self.failed = True
         for _, fut in self._pending_reqs.items():
-            fut.set_exception(error)
+            fut.set_exception(err)
         self._pending_reqs.clear()
 
         if self._transport_closed_fut and not self._transport_closed_fut.done():
             self._transport_closed_fut.set_result(None)
 
         if reconnect and not self._closed:
+            self._on_connection_lost(err)
             self._loop.create_task(self._reconnect())
 
-    def on_message(self, data):
+    def process_message(self, data):
         req_id = int.from_bytes(data[4:12], byteorder=PROTOCOL_BYTE_ORDER, signed=True)
         if req_id in self._pending_reqs:
             self._pending_reqs[req_id].set_result(data)
@@ -217,7 +220,7 @@ async def _connect_version(self) -> Union[dict, OrderedDict]:
         hs_response = await handshake_fut
 
         if hs_response.op_code == 0:
-            await self._close_transport()
+            await self.close()
             self._process_handshake_error(hs_response)
 
         return hs_response
@@ -271,4 +274,5 @@ async def _close_transport(self):
             except asyncio.TimeoutError:
                 pass
             finally:
+                self._on_connection_lost(expected=True)
                 self._transport_closed_fut = None

pygridgain/connection/connection.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 
+import logging
 from collections import OrderedDict
 import socket
 from typing import Union
@@ -30,6 +31,8 @@
 
 CLIENT_STATUS_AUTH_FAILURE = 2000
 
+logger = logging.getLogger('.'.join(__name__.split('.')[:-1]))
+
 
 class BaseConnection:
     def __init__(self, client, host: str = None, port: int = None, username: str = None, password: str = None,
@@ -82,21 +85,53 @@ def protocol_context(self):
         return self.client.protocol_context
 
     def _process_handshake_error(self, response):
-        error_text = f'Handshake error: {response.message}'
         # if handshake fails for any reason other than protocol mismatch
         # (i.e. authentication error), server version is 0.0.0
+        if response.client_status == CLIENT_STATUS_AUTH_FAILURE:
+            raise AuthenticationError(response.message)
+
         protocol_version = self.client.protocol_context.version
         server_version = (response.version_major, response.version_minor, response.version_patch)
-
+        error_text = f'Handshake error: {response.message}'
         if any(server_version):
             error_text += f' Server expects binary protocol version ' \
                           f'{server_version[0]}.{server_version[1]}.{server_version[2]}. ' \
                           f'Client provides ' \
                           f'{protocol_version[0]}.{protocol_version[1]}.{protocol_version[2]}.'
-        elif response.client_status == CLIENT_STATUS_AUTH_FAILURE:
-            raise AuthenticationError(error_text)
         raise HandshakeError(server_version, error_text)
 
+    def _on_handshake_start(self):
+        if logger.isEnabledFor(logging.DEBUG):
+            logger.debug("Connecting to node(address=%s, port=%d) with protocol context %s",
+                         self.host, self.port, self.client.protocol_context)
+
+    def _on_handshake_success(self, result):
+        features = BitmaskFeature.from_array(result.get('features', None))
+        self.client.protocol_context.features = features
+        self.uuid = result.get('node_uuid', None)  # version-specific (1.4+)
+        self.failed = False
+
+        if logger.isEnabledFor(logging.DEBUG):
+            logger.debug("Connected to node(address=%s, port=%d, node_uuid=%s) with protocol context %s",
+                         self.host, self.port, self.uuid, self.client.protocol_context)
+
+    def _on_handshake_fail(self, err):
+        if isinstance(err, AuthenticationError):
+            logger.error("Authentication failed while connecting to node(address=%s, port=%d): %s",
+                         self.host, self.port, err)
+        else:
+            logger.error("Failed to perform handshake, connection to node(address=%s, port=%d) "
+                         "with protocol context %s failed: %s",
+                         self.host, self.port, self.client.protocol_context, err, exc_info=True)
+
+    def _on_connection_lost(self, err=None, expected=False):
+        if expected and logger.isEnabledFor(logging.DEBUG):
+            logger.debug("Connection closed to node(address=%s, port=%d, node_uuid=%s)",
+                         self.host, self.port, self.uuid)
+        else:
+            logger.info("Connection lost to node(address=%s, port=%d, node_uuid=%s): %s",
+                        self.host, self.port, self.uuid, err)
+
 
 class Connection(BaseConnection):
     """
@@ -173,24 +208,26 @@ def connect(self):
             self.client.protocol_context = ProtocolContext(max(PROTOCOLS), BitmaskFeature.all_supported())
 
         try:
+            self._on_handshake_start()
             result = self._connect_version()
         except HandshakeError as e:
             if e.expected_version in PROTOCOLS:
                 self.client.protocol_context.version = e.expected_version
                 result = self._connect_version()
             else:
+                self._on_handshake_fail(e)
                 raise e
-        except connection_errors:
+        except AuthenticationError as e:
+            self._on_handshake_fail(e)
+            raise e
+        except Exception as e:
             # restore undefined protocol version
             if detecting_protocol:
                 self.client.protocol_context = None
-            raise
+            self._on_handshake_fail(e)
+            raise e
 
-        # connection is ready for end user
-        features = BitmaskFeature.from_array(result.get('features', None))
-        self.client.protocol_context.features = features
-        self.uuid = result.get('node_uuid', None)  # version-specific (1.4+)
-        self.failed = False
+        self._on_handshake_success(result)
 
     def _connect_version(self) -> Union[dict, OrderedDict]:
         """
@@ -264,11 +301,12 @@ def send(self, data: Union[bytes, bytearray], flags=None, reconnect=True):
 
         try:
             self._socket.sendall(data, **kwargs)
-        except connection_errors:
+        except connection_errors as e:
             self.failed = True
             if reconnect:
+                self._on_connection_lost(e)
                 self.reconnect()
-            raise
+            raise e
 
     def recv(self, flags=None, reconnect=True) -> bytearray:
         """
@@ -293,11 +331,12 @@ def recv(self, flags=None, reconnect=True) -> bytearray:
                 if bytes_received == 0:
                     raise SocketError('Connection broken.')
                 bytes_total_received += bytes_received
-            except connection_errors:
+            except connection_errors as e:
                 self.failed = True
                 if reconnect:
+                    self._on_connection_lost(e)
                     self.reconnect()
-                raise
+                raise e
 
             if bytes_total_received < 4:
                 continue
@@ -331,5 +370,5 @@ def close(self):
                 self._socket.close()
             except connection_errors:
                 pass
-
+            self._on_connection_lost(expected=True)
             self._socket = None

pygridgain/connection/protocol_context.py

@@ -38,6 +38,9 @@ def __eq__(self, other):
             self.version == other.version and \
             self.features == other.features
 
+    def __str__(self):
+        return f'ProtocolContext(version={self._version}, features={self._features})'
+
     def _ensure_consistency(self):
         if not self.is_feature_flags_supported():
             self._features = None