GG-33936 IGNITE-15102 Event handling and monitoring (#54)

(cherry picked from commit 82f29e2)

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

Author: isapego

docs/modules.rst

@@ -30,3 +30,4 @@ of `pygridgain`, intended for end users.
     datatypes/parsers
     datatypes/cache_props
     Exceptions <source/pygridgain.exceptions>
+    Monitoring and handling events <source/pygridgain.monitoring>

docs/source/pygridgain.connection.protocol_context.rst

@@ -0,0 +1,19 @@
+..  Copyright 2021 GridGain Systems, Inc. and Contributors.
+
+..  Licensed under the GridGain Community Edition License (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+..      https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+
+..  Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+pygridgain.connection.protocol_context package
+===========================
+
+.. automodule:: pygridgain.connection.protocol_context
+    :members:

docs/source/pygridgain.connection.rst

@@ -19,3 +19,9 @@ pygridgain.connection package
     :members:
     :undoc-members:
     :show-inheritance:
+
+Submodules
+----------
+
+.. toctree::
+    pygridgain.connection.protocol_context

docs/source/pygridgain.monitoring.rst

@@ -0,0 +1,20 @@
+..  Copyright 2021 GridGain Systems, Inc. and Contributors.
+
+..  Licensed under the GridGain Community Edition License (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+..      https://www.gridgain.com/products/software/community-edition/gridgain-community-edition-license
+
+..  Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+pygridgain.monitoring module
+===========================
+
+.. automodule:: pygridgain.monitoring
+    :members:
+    :member-order: bysource

docs/source/pygridgain.rst

@@ -43,4 +43,5 @@ Submodules
     pygridgain.transaction
     pygridgain.cursors
     pygridgain.exceptions
+    pygridgain.monitoring
 

pygridgain/aio_client.py

@@ -17,7 +17,7 @@
 import random
 import sys
 from itertools import chain
-from typing import Iterable, Type, Union, Any, Dict, Optional
+from typing import Iterable, Type, Union, Any, Dict, Optional, Sequence
 
 from .aio_cluster import AioCluster
 from .api import cache_get_node_partitions_async
@@ -61,7 +61,8 @@ class AioClient(BaseClient):
     Asynchronous Client implementation.
     """
 
-    def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **kwargs):
+    def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
+                 event_listeners: Optional[Sequence] = None, **kwargs):
         """
         Initialize client.
 
@@ -72,9 +73,10 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **
          https://ignite.apache.org/docs/latest/binary-client-protocol/data-format#schema
         :param partition_aware: (optional) try to calculate the exact data
          placement from the key before to issue the key operation to the
-         server node, `True` by default.
+         server node, `True` by default,
+        :param event_listeners: (optional) event listeners.
         """
-        super().__init__(compact_footer, partition_aware, **kwargs)
+        super().__init__(compact_footer, partition_aware, event_listeners, **kwargs)
         self._registry_mux = asyncio.Lock()
         self._affinity_query_mux = asyncio.Lock()
 
@@ -100,9 +102,8 @@ async def _connect(self, nodes):
 
                         # do not try to open more nodes
                         self._current_node = i
-
                 except connection_errors:
-                    conn.failed = True
+                    pass
 
             self._nodes.append(conn)
 
@@ -302,7 +303,7 @@ async def _get_affinity(self, conn: 'AioConnection', caches: Iterable[int]) -> D
         """
         for _ in range(AFFINITY_RETRIES or 1):
             result = await cache_get_node_partitions_async(conn, caches)
-            if result.status == 0 and result.value['partition_mapping']:
+            if result.status == 0:
                 break
             await asyncio.sleep(AFFINITY_DELAY)
 
@@ -342,7 +343,7 @@ async def get_best_node(
 
                             asyncio.ensure_future(
                                 asyncio.gather(
-                                    *[conn.reconnect() for conn in self._nodes if not conn.alive],
+                                    *[node.reconnect() for node in self._nodes if not node.alive],
                                     return_exceptions=True
                                 )
                             )

pygridgain/client.py

@@ -44,7 +44,7 @@
 import random
 import re
 from itertools import chain
-from typing import Iterable, Type, Union, Any, Dict, Optional
+from typing import Iterable, Type, Union, Any, Dict, Optional, Sequence
 
 from .api import cache_get_node_partitions
 from .api.binary import get_binary_type, put_binary_type
@@ -66,6 +66,7 @@
     get_field_by_id, unsigned
 )
 from .binary import GenericObjectMeta
+from .monitoring import _EventListeners
 
 
 __all__ = ['Client']
@@ -76,7 +77,8 @@ class BaseClient:
     _identifier = re.compile(r'[^0-9a-zA-Z_.+$]', re.UNICODE)
     _ident_start = re.compile(r'^[^a-zA-Z_]+', re.UNICODE)
 
-    def __init__(self, compact_footer: bool = None, partition_aware: bool = False, **kwargs):
+    def __init__(self, compact_footer: bool = None, partition_aware: bool = False,
+                 event_listeners: Optional[Sequence] = None, **kwargs):
         self._compact_footer = compact_footer
         self._partition_aware = partition_aware
         self._connection_args = kwargs
@@ -87,6 +89,7 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = False, *
         self.affinity_version = (0, 0)
         self._affinity = {'version': self.affinity_version, 'partition_mapping': defaultdict(dict)}
         self._protocol_context = None
+        self._event_listeners = _EventListeners(event_listeners)
 
     @property
     def protocol_context(self):
@@ -338,7 +341,8 @@ class Client(BaseClient):
     Synchronous Client implementation.
     """
 
-    def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **kwargs):
+    def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
+                 event_listeners: Optional[Sequence] = None, **kwargs):
         """
         Initialize client.
 
@@ -349,9 +353,10 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True, **
          https://ignite.apache.org/docs/latest/binary-client-protocol/data-format#schema
         :param partition_aware: (optional) try to calculate the exact data
          placement from the key before to issue the key operation to the
-         server node, `True` by default.
+         server node, `True` by default,
+        :param event_listeners: (optional) event listeners.
         """
-        super().__init__(compact_footer, partition_aware, **kwargs)
+        super().__init__(compact_footer, partition_aware, event_listeners, **kwargs)
 
     def connect(self, *args):
         """
@@ -382,7 +387,6 @@ def _connect(self, nodes):
                         self._current_node = i
 
             except connection_errors:
-                conn.failed = True
                 if self.partition_aware:
                     # schedule the reconnection
                     conn.reconnect()
@@ -566,7 +570,7 @@ def _get_affinity(self, conn: 'Connection', caches: Iterable[int]) -> Dict:
         """
         for _ in range(AFFINITY_RETRIES or 1):
             result = cache_get_node_partitions(conn, caches)
-            if result.status == 0 and result.value['partition_mapping']:
+            if result.status == 0:
                 break
             time.sleep(AFFINITY_DELAY)
 
@@ -609,9 +613,9 @@ def get_best_node(
 
                 self._update_affinity(full_affinity)
 
-                for conn in self._nodes:
-                    if not conn.alive:
-                        conn.reconnect()
+                for node in self._nodes:
+                    if not node.alive:
+                        node.reconnect()
 
             c_id = cache.cache_id if isinstance(cache, BaseCache) else cache_id(cache)
             parts = self._cache_partition_mapping(c_id).get('number_of_partitions')

pygridgain/connection/aio_connection.py

@@ -178,10 +178,10 @@ async def _connect(self):
             self._on_handshake_fail(e)
             raise e
         except Exception as e:
+            self._on_handshake_fail(e)
             # restore undefined protocol version
             if detecting_protocol:
                 self.client.protocol_context = None
-            self._on_handshake_fail(e)
             raise e
 
         self._on_handshake_success(result)

pygridgain/connection/connection.py

@@ -103,7 +103,9 @@ def _process_handshake_error(self, response):
     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)
+                         self.host, self.port, self.protocol_context)
+        if self._enabled_connection_listener:
+            self._connection_listener.publish_handshake_start(self.host, self.port, self.protocol_context)
 
     def _on_handshake_success(self, result):
         features = BitmaskFeature.from_array(result.get('features', None))
@@ -113,24 +115,45 @@ def _on_handshake_success(self, result):
 
         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)
+                         self.host, self.port, self.uuid, self.protocol_context)
+        if self._enabled_connection_listener:
+            self._connection_listener.publish_handshake_success(self.host, self.port, self.protocol_context, self.uuid)
 
     def _on_handshake_fail(self, err):
+        self.failed = True
+
         if isinstance(err, AuthenticationError):
             logger.error("Authentication failed while connecting to node(address=%s, port=%d): %s",
                          self.host, self.port, err)
+            if self._enabled_connection_listener:
+                self._connection_listener.publish_authentication_fail(self.host, self.port, self.protocol_context, 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)
+                         self.host, self.port, self.protocol_context, err, exc_info=True)
+            if self._enabled_connection_listener:
+                self._connection_listener.publish_handshake_fail(self.host, self.port, self.protocol_context, err)
 
     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)
+        if expected:
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug("Connection closed to node(address=%s, port=%d, node_uuid=%s)",
+                             self.host, self.port, self.uuid)
+            if self._enabled_connection_listener:
+                self._connection_listener.publish_connection_closed(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)
+            if self._enabled_connection_listener:
+                self._connection_listener.publish_connection_lost(self.host, self.port, self.uuid, err)
+
+    @property
+    def _enabled_connection_listener(self):
+        return self.client._event_listeners and self.client._event_listeners.enabled_connection_listener
+
+    @property
+    def _connection_listener(self):
+        return self.client._event_listeners
 
 
 class Connection(BaseConnection):
@@ -221,10 +244,10 @@ def connect(self):
             self._on_handshake_fail(e)
             raise e
         except Exception as e:
+            self._on_handshake_fail(e)
             # restore undefined protocol version
             if detecting_protocol:
                 self.client.protocol_context = None
-            self._on_handshake_fail(e)
             raise e
 
         self._on_handshake_success(result)
@@ -266,7 +289,7 @@ def reconnect(self):
         if self.alive:
             return
 
-        self.close()
+        self.close(on_reconnect=True)
 
         # connect and silence the connection errors
         try:
@@ -358,7 +381,7 @@ def recv(self, flags=None, reconnect=True) -> bytearray:
 
         return data
 
-    def close(self):
+    def close(self, on_reconnect=False):
         """
         Try to mark socket closed, then unlink it. This is recommended but
         not required, since sockets are automatically closed when
@@ -370,5 +393,6 @@ def close(self):
                 self._socket.close()
             except connection_errors:
                 pass
-            self._on_connection_lost(expected=True)
+            if not on_reconnect and not self.failed:
+                self._on_connection_lost(expected=True)
             self._socket = None

pygridgain/connection/protocol_context.py

@@ -45,6 +45,9 @@ def _ensure_consistency(self):
         if not self.is_feature_flags_supported():
             self._features = None
 
+    def copy(self):
+        return ProtocolContext(self.version, self.features)
+
     @property
     def version(self):
         return getattr(self, '_version', None)