Add balancing test with discovery and connections

src/test/java/io/tarantool/driver/integration/CartridgeMixedInstancesContainer.java

@@ -1,30 +1,30 @@
 package io.tarantool.driver.integration;
 
+import java.time.Duration;
+import java.util.HashMap;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.TarantoolCartridgeContainer;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.containers.wait.strategy.Wait;
 
-import java.time.Duration;
-import java.util.HashMap;
-
 abstract class CartridgeMixedInstancesContainer {
 
     private static final Logger logger = LoggerFactory.getLogger(CartridgeMixedInstancesContainer.class);
 
     protected static final TarantoolCartridgeContainer container;
 
     static {
-        final HashMap<String, String> buildArgs = new HashMap<>();
-        buildArgs.put("TARANTOOL_INSTANCES_FILE", "./instances_mixed.yml");
-        container = new TarantoolCartridgeContainer(
-            "cartridge/instances_mixed.yml",
-            "cartridge/topology_mixed.lua", buildArgs)
-            .withDirectoryBinding("cartridge")
-            .withLogConsumer(new Slf4jLogConsumer(logger))
-            .waitingFor(Wait.forLogMessage(".*Listening HTTP on.*", 3))
-            .withStartupTimeout(Duration.ofMinutes(2));
+        final HashMap<String, String> env = new HashMap<>();
+        env.put("TARANTOOL_INSTANCES_FILE", "./instances_mixed.yml");
+        container = new TarantoolCartridgeContainer("cartridge/instances_mixed.yml",
+                                                    "cartridge/topology_mixed.lua")
+                        .withDirectoryBinding("cartridge")
+                        .withLogConsumer(new Slf4jLogConsumer(logger))
+                        .waitingFor(Wait.forLogMessage(".*Listening HTTP on.*", 3))
+                        .withStartupTimeout(Duration.ofMinutes(2))
+                        .withEnv(env);
     }
 
     protected static void startCluster() {

src/test/java/io/tarantool/driver/integration/ClusterDiscoveryIT.java

@@ -44,7 +44,7 @@ public void httpClusterDiscovererTest() throws TarantoolClientException {
         HTTPDiscoveryClusterAddressProvider addressProvider = getHttpProvider();
         Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();
 
-        assertEquals(nodes.size(), 3);
+        assertEquals(nodes.size(), 4);
         Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));
@@ -70,7 +70,7 @@ public void binaryClusterDiscovererTest() {
         TarantoolClusterAddressProvider addressProvider = getBinaryProvider();
 
         Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();
-        assertEquals(nodes.size(), 3);
+        assertEquals(nodes.size(), 4);
         Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));

src/test/java/io/tarantool/driver/integration/ReconnectIT.java

@@ -25,6 +25,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.shaded.org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -245,6 +246,8 @@ void test_should_closeConnections_ifAddressProviderReturnsNewAddresses() throws
         // restart routers for resetting connections
         stopInstances(Arrays.asList("router", "second-router"));
         startCartridge();
+        String status = container.execInContainer("cartridge", "status", "--run-dir=/tmp/run").getStderr();
+        assertEquals(6, StringUtils.countMatches(status, "RUNNING"));
 
         final TarantoolServerAddress firstAddress =
             new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3301));
@@ -298,7 +301,7 @@ public void setRefreshCallback(Runnable runnable) {
                                 numberOfSwitching.incrementAndGet();
                                 runnable.run();
                             }
-                        }, 500, 100, TimeUnit.MILLISECONDS);
+                        }, 0, 100, TimeUnit.MILLISECONDS);
                     }
                 }).build();
 

src/test/java/io/tarantool/driver/integration/SharedCartridgeContainer.java

@@ -55,6 +55,6 @@ protected static void startInstance(String instanceName) throws IOException, Int
     }
 
     protected static void stopInstance(String instanceName) throws IOException, InterruptedException {
-        container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", "--data-dir=/tmp/data", instanceName);
+        container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", instanceName);
     }
 }

src/test/resources/cartridge/app/roles/api_router.lua

@@ -1,11 +1,12 @@
 local vshard = require('vshard')
 local cartridge_rpc = require('cartridge.rpc')
-local fiber = require('fiber')
 local crud = require('crud')
 local uuid = require('uuid')
 local log = require('log')
 
+local metadata_utils = require('utils.metadata')
 local crud_utils = require('utils.crud')
+local counter = require('modules.counter')
 
 local function get_schema()
     for _, instance_uri in pairs(cartridge_rpc.get_candidates('app.roles.api_storage', { leader_only = true })) do
@@ -100,51 +101,6 @@ local function raising_error()
     error("Test error: raising_error() called")
 end
 
-local function reset_request_counters()
-    box.space.request_counters:replace({ 1, 0 })
-end
-
-local function get_router_name()
-    return string.sub(box.cfg.custom_proc_title, 9)
-end
-
-local function simple_long_running_function(seconds_to_sleep)
-    fiber.sleep(seconds_to_sleep)
-    return true
-end
-
-local function long_running_function(values)
-    local seconds_to_sleep = 0
-    local disabled_router_name = ""
-    if values ~= nil then
-        if type(values) == "table" then
-            values = values or {}
-            seconds_to_sleep = values[1]
-            disabled_router_name = values[2]
-        else
-            seconds_to_sleep = values
-        end
-    end
-
-    -- need using number instead field name as string in update function for compatibility with tarantool 1.10
-    box.space.request_counters:update(1, { { '+', 2, 1 } })
-    log.info('Executing long-running function ' ..
-        tostring(box.space.request_counters:get(1)[2]) ..
-        "(name: " .. disabled_router_name ..
-        "; sleep: " .. seconds_to_sleep .. ")")
-    if get_router_name() == disabled_router_name then
-        return nil, "Disabled by client; router_name = " .. disabled_router_name
-    end
-    if seconds_to_sleep then
-        fiber.sleep(seconds_to_sleep)
-    end
-    return true
-end
-
-local function get_request_count()
-    return box.space.request_counters:get(1)[2]
-end
-
 -- it's like vshard error throwing
 local function box_error_unpack_no_connection()
     return nil, box.error.new(box.error.NO_CONNECTION):unpack()
@@ -202,12 +158,6 @@ local function select_router_space()
 end
 
 local function init_router_spaces()
-    local request_counters = box.schema.space.create('request_counters', {
-        format = { { 'id', 'unsigned' }, { 'count', 'unsigned' } },
-        if_not_exists = true
-    })
-    request_counters:create_index('id', { parts = { 'id' }, if_not_exists = true })
-
     local router_space = box.schema.space.create('router_space', {
         format = { { 'id', 'unsigned' } },
         if_not_exists = true
@@ -220,6 +170,7 @@ end
 local function init(opts)
     if opts.is_master then
         init_router_spaces()
+        counter.init_counter_space()
     end
     patch_crud_methods_for_tests()
 
@@ -235,11 +186,13 @@ local function init(opts)
     rawset(_G, 'retrying_function', retrying_function)
     rawset(_G, 'raising_error', raising_error)
 
-    rawset(_G, 'reset_request_counters', reset_request_counters)
-    rawset(_G, 'get_router_name', get_router_name)
-    rawset(_G, 'long_running_function', long_running_function)
-    rawset(_G, 'simple_long_running_function', simple_long_running_function)
-    rawset(_G, 'get_request_count', get_request_count)
+    rawset(_G, 'get_router_name', metadata_utils.get_router_name)
+
+    rawset(_G, 'reset_request_counters', counter.reset_request_counters)
+    rawset(_G, 'simple_long_running_function', counter.simple_long_running_function)
+    rawset(_G, 'long_running_function', counter.long_running_function)
+    rawset(_G, 'get_request_count', counter.get_request_count)
+
     rawset(_G, 'box_error_unpack_no_connection', box_error_unpack_no_connection)
     rawset(_G, 'box_error_unpack_timeout', box_error_unpack_timeout)
     rawset(_G, 'box_error_timeout', box_error_timeout)

src/test/resources/cartridge/app/roles/custom.lua

@@ -1,5 +1,7 @@
 local cartridge = require('cartridge')
 
+local counter = require('modules.counter')
+
 function get_routers()
     local function table_contains(table, element)
         for _, value in pairs(table) do
@@ -44,10 +46,16 @@ local function init(opts)
     -- luacheck: no unused args
     if opts.is_master then
         box.schema.user.grant('guest', 'read,write', 'universe', nil, { if_not_exists = true })
+        counter.init_counter_space()
     end
 
     init_httpd()
 
+    rawset(_G, 'reset_request_counters', counter.reset_request_counters)
+    rawset(_G, 'simple_long_running_function', counter.simple_long_running_function)
+    rawset(_G, 'long_running_function', counter.long_running_function)
+    rawset(_G, 'get_request_count', counter.get_request_count)
+
     return true
 end
 

src/test/resources/cartridge/instances.yml

@@ -13,6 +13,11 @@ testapp.third-router:
   advertise_uri: localhost:3303
   http_port: 8083
 
+testapp.fourth-router:
+  workdir: ./tmp/db_dev/3306
+  advertise_uri: localhost:3306
+  http_port: 8086
+
 testapp.s1-storage:
   workdir: ./tmp/db_dev/3304
   advertise_uri: localhost:3304

src/test/resources/cartridge/modules/counter.lua

@@ -0,0 +1,78 @@
+local fiber = require('fiber')
+local log = require('log')
+local metadata_utils = require('utils.metadata')
+
+local function reset_request_counters()
+    box.space.request_counters:replace({ 1, 0 })
+end
+
+local function update_request_counters(with_session_id)
+    with_session_id = with_session_id or false
+    -- need using number instead field name as string in update function for compatibility with tarantool 1.10
+    if with_session_id then
+        box.space.request_counters:update(box.session.id(), { { '+', 2, 1 } })
+    else
+        box.space.request_counters:update(1, { { '+', 2, 1 } })
+    end
+end
+
+local function get_request_count()
+    return box.space.request_counters:get(1)[2]
+end
+
+local function simple_long_running_function(seconds_to_sleep, with_session_id)
+    update_request_counters(with_session_id)
+    fiber.sleep(seconds_to_sleep)
+    return true
+end
+
+local function long_running_function(values, with_session_id)
+    local seconds_to_sleep = 0
+    local disabled_router_name = ""
+    if values ~= nil then
+        if type(values) == "table" then
+            values = values or {}
+            seconds_to_sleep = values[1]
+            disabled_router_name = values[2]
+        else
+            seconds_to_sleep = values
+        end
+    end
+
+    update_request_counters(with_session_id)
+    log.info('Executing long-running function ' ..
+        tostring(box.space.request_counters:get(1)[2]) ..
+        "(name: " .. disabled_router_name ..
+        "; sleep: " .. seconds_to_sleep .. ")")
+    if metadata_utils.get_router_name() == disabled_router_name then
+        return nil, "Disabled by client; router_name = " .. disabled_router_name
+    end
+    if seconds_to_sleep then
+        fiber.sleep(seconds_to_sleep)
+    end
+    return true
+end
+
+local function reset_request_counters_on_connect()
+    box.space.request_counters:replace({ box.session.id(), 0 })
+end
+
+local function init_counter_space()
+    local request_counters = box.schema.space.create('request_counters', {
+        format = { { 'id', 'unsigned' }, { 'count', 'unsigned' } },
+        if_not_exists = true
+    })
+    request_counters:create_index('id', { parts = { 'id' }, if_not_exists = true })
+    request_counters:create_index('count', { parts = { 'count' }, if_not_exists = true, unique = false })
+
+
+    box.session.on_connect(reset_request_counters_on_connect)
+end
+
+return {
+    reset_request_counters = reset_request_counters,
+    simple_long_running_function = simple_long_running_function,
+    long_running_function = long_running_function,
+    get_request_count = get_request_count,
+    init_counter_space = init_counter_space,
+}

src/test/resources/cartridge/utils/metadata.lua

@@ -0,0 +1,7 @@
+local function get_router_name()
+    return string.sub(box.cfg.custom_proc_title, 9)
+end
+
+return {
+    get_router_name = get_router_name
+}