Add cluster discovery function to mesh connection

Make Mesh class compatible with original Connection
Make mesh addrs validation parts
Add warning on mesh config validation.
Reworked tests on mesh class
Reworked strategy class
Autodocs on mesh class

Author: denis-ignatenko

doc/api/class-mesh-connection.rst

@@ -0,0 +1,8 @@
+
+.. currentmodule:: tarantool.mesh_connection
+
+class :class:`MeshConnection`
+-----------------------------
+
+.. autoclass:: MeshConnection
+

doc/index.rst

@@ -40,6 +40,7 @@ API Reference
 
    api/module-tarantool.rst
    api/class-connection.rst
+   api/class-mesh-connection.rst
    api/class-space.rst
    api/class-response.rst
 

doc/index.ru.rst

@@ -40,6 +40,7 @@
 
    api/module-tarantool.rst
    api/class-connection.rst
+   api/class-mesh-connection.rst
    api/class-space.rst
    api/class-response.rst
 

tarantool/const.py

@@ -86,3 +86,5 @@
 RECONNECT_MAX_ATTEMPTS = 10
 # Default delay between attempts to reconnect (seconds)
 RECONNECT_DELAY = 0.1
+# Default cluster nodes list refresh interval (seconds)
+CLUSTER_DISCOVERY_DELAY = 60

tarantool/error.py

@@ -152,6 +152,10 @@ class NetworkWarning(UserWarning):
     pass
 
 
+class ConfigurationWarning(UserWarning):
+    '''Warning in configuration validation'''
+    pass
+
 # always print this warnings
 warnings.filterwarnings("always", category=NetworkWarning)
 

tarantool/mesh_connection.py

@@ -4,42 +4,173 @@
 between tarantool instances and basic Round-Robin strategy.
 '''
 
+import time
+from itertools import cycle
+
+
 from tarantool.connection import Connection
-from tarantool.error import NetworkError
+from tarantool.error import warn, NetworkError, ConfigurationWarning
 from tarantool.utils import ENCODING_DEFAULT
 from tarantool.const import (
+    CONNECTION_TIMEOUT,
     SOCKET_TIMEOUT,
     RECONNECT_MAX_ATTEMPTS,
-    RECONNECT_DELAY
+    RECONNECT_DELAY,
+    CLUSTER_DISCOVERY_DELAY,
+)
+
+from tarantool.request import (
+    RequestCall
 )
 
+try:
+    string_types = basestring
+except NameError:
+    string_types = str
+
 
 class RoundRobinStrategy(object):
+    """
+    Simple round-robin address rotation
+    """
     def __init__(self, addrs):
-        self.addrs = addrs
-        self.pos = 0
+        self.reload(addrs)
+    
+    def __validate(self, address):
+        err = []
+
+        if "host" not in address:
+            err.append("host key must be set")
+        elif not isinstance(address["host"], string_types):
+            err.append("host value must be string type")
+
+        if "port" not in address:
+            err.append("port is not set")
+        elif not isinstance(address["port"], int):
+            err.append("port value must be int type")
+        elif address["port"] == 0:
+            err.append("port value must not be zero")
+
+        if err:
+            wrn = 'Address validation :  %s ' % ", ".join(err)
+            warn(wrn,  ConfigurationWarning)
+            return False
+        return True
+
+    def reload(self, addrs):
+        self.addrs = []
+        for i in addrs:
+            if self.__validate(i) and i not in self.addrs:
+                self.addrs.append(i)
+        
+        self.__gen = cycle(self.addrs)
 
     def getnext(self):
-        tmp = self.pos
-        self.pos = (self.pos + 1) % len(self.addrs)
-        return self.addrs[tmp]
+        return next(self.__gen)
+
+
+def parse_uri(uri_str):
+    if not uri_str or uri_str.count(":") != 1:
+        return
+    uri = uri_str.split(':', 1)
+    host = uri[0]
+    try:
+        port = int(uri[1])
+    except ValueError:
+        return
+
+    if host and port:
+        return {'host': host, 'port': port}
 
 
 class MeshConnection(Connection):
-    def __init__(self, addrs,
+    '''
+   Represents a connection to a cluster of Tarantool servers.
+
+   This class uses Connection to connect to one of the nodes of the cluster.
+   The initial list of nodes is passed to the constructor in 'addrs' parameter.
+   The class set in 'strategy_class' parameter is used to select a node from
+   the list and switch nodes in case of unavailability of the current node.
+
+   'cluster_discovery_function' param of the constructor sets the name of a stored
+   Lua function used to refresh the list of available nodes. The function takes
+   no parameters and returns a list of strings in format 'host:port'. A generic
+   function for getting the list of nodes looks like this:
+
+    .. code-block:: lua
+
+        function get_cluster_nodes()
+            return {
+                '192.168.0.1:3301',
+                '192.168.0.2:3302',
+                -- ...
+            }
+        end
+
+    You may put in this list whatever you need depending on your cluster
+    topology. Chances are you'll want to make the list of nodes from nodes'
+    replication config. Here is an example for it:
+
+    .. code-block:: lua
+
+        local uri_lib = require('uri')
+
+        function get_cluster_nodes()
+            local nodes = {}
+
+            local replicas = box.cfg.replication
+
+            for i = 1, #replicas do
+                local uri = uri_lib.parse(replicas[i])
+
+                if uri.host and uri.service then
+                    table.insert(nodes, uri.host .. ':' .. uri.service)
+                end
+            end
+
+            -- if your replication config doesn't contain the current node
+            -- you have to add it manually like this:
+            table.insert(nodes, '192.168.0.1:3301')
+
+            return nodes
+        end
+    '''
+
+    def __init__(self, host=None, port=None,
                  user=None,
                  password=None,
                  socket_timeout=SOCKET_TIMEOUT,
                  reconnect_max_attempts=RECONNECT_MAX_ATTEMPTS,
                  reconnect_delay=RECONNECT_DELAY,
                  connect_now=True,
                  encoding=ENCODING_DEFAULT,
-                 strategy_class=RoundRobinStrategy):
-        self.nattempts = 2 * len(addrs) + 1
+                 call_16=False,
+                 connection_timeout=CONNECTION_TIMEOUT,
+                 addrs=None,
+                 strategy_class=RoundRobinStrategy,
+                 cluster_discovery_function=None,
+                 cluster_discovery_delay=CLUSTER_DISCOVERY_DELAY):
+
+        if addrs is None:
+            addrs = []
+
+        if host and port:
+            addrs.insert(0, {'host': host, 'port': port})
+
+        self.strategy_class = strategy_class
         self.strategy = strategy_class(addrs)
+
+        if not self.strategy.addrs and connect_now:
+            raise ValueError("Host/port or addrs list must be set")
+
         addr = self.strategy.getnext()
         host = addr['host']
         port = addr['port']
+
+        self.cluster_discovery_function = cluster_discovery_function
+        self.cluster_discovery_delay = cluster_discovery_delay
+        self.last_nodes_refresh = 0
+
         super(MeshConnection, self).__init__(host=host,
                                              port=port,
                                              user=user,
@@ -48,18 +179,82 @@ def __init__(self, addrs,
                                              reconnect_max_attempts=reconnect_max_attempts,
                                              reconnect_delay=reconnect_delay,
                                              connect_now=connect_now,
-                                             encoding=encoding)
+                                             encoding=encoding,
+                                             call_16=call_16,
+                                             connection_timeout=connection_timeout)
+        if connect_now and self.cluster_discovery_function:
+            self._opt_refresh_instances()
 
     def _opt_reconnect(self):
-        nattempts = self.nattempts
-        while nattempts > 0:
+        '''
+        Use original opt_reconnect with address rotation
+        '''
+
+        for i in range(len(self.strategy.addrs)+1):
             try:
                 super(MeshConnection, self)._opt_reconnect()
-                break
             except NetworkError:
-                nattempts -= 1
                 addr = self.strategy.getnext()
-                self.host = addr['host']
-                self.port = addr['port']
-        else:
+                self.host = addr["host"]
+                self.port = addr["port"]
+                continue
+
+        if not self._socket:
             raise NetworkError
+
+    def _opt_refresh_instances(self):
+        '''
+        Refresh list of cluster instances.
+        If current connection not in server list will change connection.
+        '''
+        now = time.time()
+
+        if not self.connected or not self.cluster_discovery_function:
+            return
+
+        if now - self.last_nodes_refresh > self.cluster_discovery_delay:
+            request = RequestCall(self,
+                                  self.cluster_discovery_function,
+                                  (),
+                                  self.call_16)
+            
+            self._opt_reconnect()
+            resp = self._send_request_wo_reconnect(request)
+            
+            # got data to refresh
+            if resp.data and resp.data[0]:
+                addrs = []
+                for i in resp.data[0]:
+                    addr = parse_uri(i)
+                    if addr:
+                        addrs.append(addr)
+                
+                self.strategy.reload(addrs)
+                self.last_nodes_refresh = now
+                
+                if not self.strategy.addrs:
+                    raise NetworkError('No addrs to connect')
+
+                # Check current addr in list
+                if {'host': self.host, 'port': self.port} not in self.strategy.addrs:
+                    self.close()
+                    addr = self.strategy.getnext()
+                    self.host = addr['host']
+                    self.port = addr['port']
+                    self._opt_reconnect()
+
+    def _send_request(self, request):
+        '''
+        Send the request to the server through the socket.
+        Return an instance of `Response` class.
+
+        Update instances list from server `cluster_discovery_function` function.
+
+        :param request: object representing a request
+        :type request: `Request` instance
+
+        :rtype: `Response` instance
+        '''
+        self._opt_refresh_instances()
+
+        return super(MeshConnection, self)._send_request(request)

unit/suites/__init__.py

@@ -8,9 +8,10 @@
 from .test_dml import TestSuite_Request
 from .test_protocol import TestSuite_Protocol
 from .test_reconnect import TestSuite_Reconnect
+from .test_mesh import TestSuite_Mesh
 
 test_cases = (TestSuite_Schema, TestSuite_Request, TestSuite_Protocol,
-              TestSuite_Reconnect)
+              TestSuite_Reconnect, TestSuite_Mesh)
 
 def load_tests(loader, tests, pattern):
     suite = unittest.TestSuite()