GG-33939 IGNITE-15118 Implement handshake timeout (#55)

(cherry picked from commit de07126)

Co-authored-by: Ivan Daschinsky <[email protected]>

Author: isapego

.travis.yml

@@ -52,4 +52,4 @@ jobs:
       env: TOXENV=py39
 
 install: pip install tox
-script: tox
+script: tox

examples/transactions.py

@@ -130,6 +130,12 @@ def sync_example():
 
 
 if __name__ == '__main__':
+    client = Client()
+    with client.connect('127.0.0.1', 10800):
+        if not client.protocol_context.is_transactions_supported():
+            print("'Transactions' API is not supported by cluster. Finishing...")
+            exit(0)
+
     print("Starting sync example")
     sync_example()
 

pygridgain/aio_client.py

@@ -66,6 +66,9 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
         """
         Initialize client.
 
+        For the use of the SSL-related parameters see
+        https://docs.python.org/3/library/ssl.html#ssl-certificates.
+
         :param compact_footer: (optional) use compact (True, recommended) or
          full (False) schema approach when serializing Complex objects.
          Default is to use the same approach the server is using (None).
@@ -74,7 +77,37 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
         :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,
-        :param event_listeners: (optional) event listeners.
+        :param event_listeners: (optional) event listeners,
+        :param handshake_timeout: (optional) sets timeout (in seconds) for performing handshake (connection)
+         with node. Default is 10.0 seconds,
+        :param use_ssl: (optional) set to True if Ignite server uses SSL
+         on its binary connector. Defaults to use SSL when username
+         and password has been supplied, not to use SSL otherwise,
+        :param ssl_version: (optional) SSL version constant from standard
+         `ssl` module. Defaults to TLS v1.2,
+        :param ssl_ciphers: (optional) ciphers to use. If not provided,
+         `ssl` default ciphers are used,
+        :param ssl_cert_reqs: (optional) determines how the remote side
+         certificate is treated:
+
+         * `ssl.CERT_NONE` − remote certificate is ignored (default),
+         * `ssl.CERT_OPTIONAL` − remote certificate will be validated,
+           if provided,
+         * `ssl.CERT_REQUIRED` − valid remote certificate is required,
+
+        :param ssl_keyfile: (optional) a path to SSL key file to identify
+         local (client) party,
+        :param ssl_keyfile_password: (optional) password for SSL key file,
+         can be provided when key file is encrypted to prevent OpenSSL
+         password prompt,
+        :param ssl_certfile: (optional) a path to ssl certificate file
+         to identify local (client) party,
+        :param ssl_ca_certfile: (optional) a path to a trusted certificate
+         or a certificate chain. Required to check the validity of the remote
+         (server-side) certificate,
+        :param username: (optional) user name to authenticate to Ignite
+         cluster,
+        :param password: (optional) password to authenticate to Ignite cluster.
         """
         super().__init__(compact_footer, partition_aware, event_listeners, **kwargs)
         self._registry_mux = asyncio.Lock()

pygridgain/client.py

@@ -346,6 +346,9 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
         """
         Initialize client.
 
+        For the use of the SSL-related parameters see
+        https://docs.python.org/3/library/ssl.html#ssl-certificates.
+
         :param compact_footer: (optional) use compact (True, recommended) or
          full (False) schema approach when serializing Complex objects.
          Default is to use the same approach the server is using (None).
@@ -354,7 +357,41 @@ def __init__(self, compact_footer: bool = None, partition_aware: bool = True,
         :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,
-        :param event_listeners: (optional) event listeners.
+        :param event_listeners: (optional) event listeners,
+        :param timeout: (optional) sets timeout (in seconds) for each socket
+         operation including `connect`. 0 means non-blocking mode, which is
+         virtually guaranteed to fail. Can accept integer or float value.
+         Default is None (blocking mode),
+        :param handshake_timeout: (optional) sets timeout (in seconds) for performing handshake (connection)
+         with node. Default is 10.0 seconds,
+        :param use_ssl: (optional) set to True if Ignite server uses SSL
+         on its binary connector. Defaults to use SSL when username
+         and password has been supplied, not to use SSL otherwise,
+        :param ssl_version: (optional) SSL version constant from standard
+         `ssl` module. Defaults to TLS v1.2,
+        :param ssl_ciphers: (optional) ciphers to use. If not provided,
+         `ssl` default ciphers are used,
+        :param ssl_cert_reqs: (optional) determines how the remote side
+         certificate is treated:
+
+         * `ssl.CERT_NONE` − remote certificate is ignored (default),
+         * `ssl.CERT_OPTIONAL` − remote certificate will be validated,
+           if provided,
+         * `ssl.CERT_REQUIRED` − valid remote certificate is required,
+
+        :param ssl_keyfile: (optional) a path to SSL key file to identify
+         local (client) party,
+        :param ssl_keyfile_password: (optional) password for SSL key file,
+         can be provided when key file is encrypted to prevent OpenSSL
+         password prompt,
+        :param ssl_certfile: (optional) a path to ssl certificate file
+         to identify local (client) party,
+        :param ssl_ca_certfile: (optional) a path to a trusted certificate
+         or a certificate chain. Required to check the validity of the remote
+         (server-side) certificate,
+        :param username: (optional) user name to authenticate to Ignite
+         cluster,
+        :param password: (optional) password to authenticate to Ignite cluster.
         """
         super().__init__(compact_footer, partition_aware, event_listeners, **kwargs)
 

pygridgain/connection/aio_connection.py

@@ -18,8 +18,6 @@
 from collections import OrderedDict
 from typing import Union
 
-from tzlocal import get_localzone
-
 from pygridgain.constants import PROTOCOLS, PROTOCOL_BYTE_ORDER
 from pygridgain.exceptions import HandshakeError, SocketError, connection_errors, AuthenticationError
 from .bitmask_feature import BitmaskFeature
@@ -106,11 +104,13 @@ def __init__(self, client: 'AioClient', host: str, port: int, username: str = No
         :param client: Ignite client object,
         :param host: Ignite server node's host name or IP,
         :param port: Ignite server node's port number,
+        :param handshake_timeout: (optional) sets timeout (in seconds) for performing handshake (connection)
+         with node. Default is 10.0 seconds,
         :param use_ssl: (optional) set to True if Ignite server uses SSL
          on its binary connector. Defaults to use SSL when username
          and password has been supplied, not to use SSL otherwise,
         :param ssl_version: (optional) SSL version constant from standard
-         `ssl` module. Defaults to TLS v1.1, as in Ignite 2.5,
+         `ssl` module. Defaults to TLS v1.2,
         :param ssl_ciphers: (optional) ciphers to use. If not provided,
          `ssl` default ciphers are used,
         :param ssl_cert_reqs: (optional) determines how the remote side
@@ -153,7 +153,6 @@ async def connect(self):
         """
         if self.alive:
             return
-        self._closed = False
         await self._connect()
 
     async def _connect(self):
@@ -164,27 +163,28 @@ async def _connect(self):
             detecting_protocol = True
             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
+        while True:
+            try:
+                self._on_handshake_start()
                 result = await self._connect_version()
-            else:
+                self._on_handshake_success(result)
+                return
+            except HandshakeError as e:
+                if e.expected_version in PROTOCOLS:
+                    self.client.protocol_context.version = e.expected_version
+                    continue
+                else:
+                    self._on_handshake_fail(e)
+                    raise e
+            except AuthenticationError as e:
                 self._on_handshake_fail(e)
                 raise e
-        except AuthenticationError as e:
-            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
-            raise e
-
-        self._on_handshake_success(result)
+            except Exception as e:
+                self._on_handshake_fail(e)
+                # restore undefined protocol version
+                if detecting_protocol:
+                    self.client.protocol_context = None
+                raise e
 
     def process_connection_lost(self, err, reconnect=False):
         self.failed = True
@@ -213,11 +213,13 @@ async def _connect_version(self) -> Union[dict, OrderedDict]:
 
         ssl_context = create_ssl_context(self.ssl_params)
         handshake_fut = self._loop.create_future()
-        self._transport, _ = await self._loop.create_connection(
-            lambda: BaseProtocol(self, handshake_fut),
-            host=self.host, port=self.port, ssl=ssl_context)
-
-        hs_response = await handshake_fut
+        self._closed = False
+        self._transport, _ = await self._loop.create_connection(lambda: BaseProtocol(self, handshake_fut),
+                                                                host=self.host, port=self.port, ssl=ssl_context)
+        try:
+            hs_response = await asyncio.wait_for(handshake_fut, self.handshake_timeout)
+        except asyncio.TimeoutError:
+            raise ConnectionError('timed out')
 
         if hs_response.op_code == 0:
             await self.close()

pygridgain/connection/connection.py

@@ -21,7 +21,7 @@
 from tzlocal import get_localzone
 
 from pygridgain.constants import PROTOCOLS, DEFAULT_HOST, DEFAULT_PORT, PROTOCOL_BYTE_ORDER
-from pygridgain.exceptions import HandshakeError, SocketError, connection_errors, AuthenticationError
+from pygridgain.exceptions import HandshakeError, SocketError, connection_errors, AuthenticationError, ParameterError
 from .bitmask_feature import BitmaskFeature
 
 from .handshake import HandshakeRequest, HandshakeResponse
@@ -36,8 +36,9 @@
 
 class BaseConnection:
     def __init__(self, client, host: str = None, port: int = None, username: str = None, password: str = None,
-                 **ssl_params):
+                 handshake_timeout: float = 10.0, **ssl_params):
         self.client = client
+        self.handshake_timeout = handshake_timeout
         self.host = host if host else DEFAULT_HOST
         self.port = port if port else DEFAULT_PORT
         self.username = username
@@ -46,6 +47,9 @@ def __init__(self, client, host: str = None, port: int = None, username: str = N
 
         self.timezone = get_localzone().tzname(None)
 
+        if handshake_timeout <= 0.0:
+            raise ParameterError("handshake_timeout should be positive")
+
         check_ssl_params(ssl_params)
 
         if self.username and self.password and 'use_ssl' not in ssl_params:
@@ -166,8 +170,9 @@ class Connection(BaseConnection):
      * binary protocol connector. Encapsulates handshake and failover reconnection.
     """
 
-    def __init__(self, client: 'Client', host: str, port: int, timeout: float = None,
-                 username: str = None, password: str = None, **ssl_params):
+    def __init__(self, client: 'Client', host: str, port: int, username: str = None, password: str = None,
+                 timeout: float = None, handshake_timeout: float = 10.0,
+                 **ssl_params):
         """
         Initialize connection.
 
@@ -181,11 +186,13 @@ def __init__(self, client: 'Client', host: str, port: int, timeout: float = None
          operation including `connect`. 0 means non-blocking mode, which is
          virtually guaranteed to fail. Can accept integer or float value.
          Default is None (blocking mode),
+        :param handshake_timeout: (optional) sets timeout (in seconds) for performing handshake (connection)
+         with node. Default is 10.0.
         :param use_ssl: (optional) set to True if GridGain server uses SSL
          on its binary connector. Defaults to use SSL when username
          and password has been supplied, not to use SSL otherwise,
         :param ssl_version: (optional) SSL version constant from standard
-         `ssl` module. Defaults to TLS v1.1, as in GridGain 8.5,
+         `ssl` module. Defaults to TLS v1.2,
         :param ssl_ciphers: (optional) ciphers to use. If not provided,
          `ssl` default ciphers are used,
         :param ssl_cert_reqs: (optional) determines how the remote side
@@ -211,7 +218,7 @@ def __init__(self, client: 'Client', host: str, port: int, timeout: float = None
         :param password: (optional) password to authenticate to GridGain
          cluster.
         """
-        super().__init__(client, host, port, username, password, **ssl_params)
+        super().__init__(client, host, port, username, password, handshake_timeout, **ssl_params)
         self.timeout = timeout
         self._socket = None
 
@@ -230,27 +237,29 @@ def connect(self):
             detecting_protocol = True
             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
+        while True:
+            try:
+                self._on_handshake_start()
                 result = self._connect_version()
-            else:
+                self._socket.settimeout(self.timeout)
+                self._on_handshake_success(result)
+                return
+            except HandshakeError as e:
+                if e.expected_version in PROTOCOLS:
+                    self.client.protocol_context.version = e.expected_version
+                    continue
+                else:
+                    self._on_handshake_fail(e)
+                    raise e
+            except AuthenticationError as e:
                 self._on_handshake_fail(e)
                 raise e
-        except AuthenticationError as e:
-            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
-            raise e
-
-        self._on_handshake_success(result)
+            except Exception as e:
+                self._on_handshake_fail(e)
+                # restore undefined protocol version
+                if detecting_protocol:
+                    self.client.protocol_context = None
+                raise e
 
     def _connect_version(self) -> Union[dict, OrderedDict]:
         """
@@ -259,7 +268,7 @@ def _connect_version(self) -> Union[dict, OrderedDict]:
         """
 
         self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self._socket.settimeout(self.timeout)
+        self._socket.settimeout(self.handshake_timeout)
         self._socket = wrap(self._socket, self.ssl_params)
         self._socket.connect((self.host, self.port))
 

pygridgain/connection/protocol_context.py

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

pygridgain/transaction.py

@@ -24,7 +24,7 @@
 
 
 def _validate_int_enum_param(value: Union[int, IntEnum], cls: Type[IntEnum]):
-    if value not in cls:
+    if value not in set(v.value for v in cls):  # Use this trick to disable warning on python 3.7
         raise ValueError(f'{value} not in {cls}')
     return value