Skip to content

Commit 7f9bdf4

Browse files
ArtDuArtDu
committed
Add balancing test with discovery and connections
1 parent 27df177 commit 7f9bdf4

6 files changed

Lines changed: 277 additions & 77 deletions

File tree

‎src/test/java/io/tarantool/driver/integration/ClusterConnectionIT.java‎

Lines changed: 169 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,46 @@
11
package io.tarantool.driver.integration;
22

3-
import io.tarantool.driver.api.TarantoolClientConfig;
4-
import io.tarantool.driver.api.TarantoolClusterAddressProvider;
5-
import io.tarantool.driver.api.TarantoolServerAddress;
6-
import io.tarantool.driver.api.connection.TarantoolConnection;
7-
import io.tarantool.driver.api.retry.TarantoolRequestRetryPolicies;
8-
import io.tarantool.driver.auth.SimpleTarantoolCredentials;
9-
import io.tarantool.driver.core.ClusterTarantoolTupleClient;
10-
import io.tarantool.driver.core.ProxyTarantoolTupleClient;
11-
import io.tarantool.driver.core.RetryingTarantoolTupleClient;
12-
import org.junit.jupiter.api.BeforeAll;
13-
import org.junit.jupiter.api.Test;
14-
import org.rnorth.ducttape.unreliables.Unreliables;
15-
import org.testcontainers.containers.Container;
16-
import org.testcontainers.junit.jupiter.Testcontainers;
17-
183
import java.io.IOException;
194
import java.util.ArrayList;
205
import java.util.Arrays;
216
import java.util.Collections;
227
import java.util.List;
238
import java.util.concurrent.CompletableFuture;
9+
import java.util.concurrent.ExecutionException;
2410
import java.util.concurrent.TimeUnit;
2511
import java.util.concurrent.TimeoutException;
2612
import java.util.concurrent.atomic.AtomicReference;
13+
import java.util.stream.Collectors;
2714

15+
import org.jetbrains.annotations.NotNull;
16+
import org.junit.jupiter.api.BeforeAll;
17+
import org.junit.jupiter.api.Test;
18+
import org.rnorth.ducttape.unreliables.Unreliables;
19+
import org.testcontainers.containers.Container;
20+
import org.testcontainers.junit.jupiter.Testcontainers;
2821
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
2922
import static org.junit.jupiter.api.Assertions.assertEquals;
3023
import static org.junit.jupiter.api.Assertions.assertTrue;
3124

25+
import io.tarantool.driver.api.TarantoolClient;
26+
import io.tarantool.driver.api.TarantoolClientConfig;
27+
import io.tarantool.driver.api.TarantoolClientFactory;
28+
import io.tarantool.driver.api.TarantoolClusterAddressProvider;
29+
import io.tarantool.driver.api.TarantoolResult;
30+
import io.tarantool.driver.api.TarantoolServerAddress;
31+
import io.tarantool.driver.api.connection.TarantoolConnection;
32+
import io.tarantool.driver.api.retry.TarantoolRequestRetryPolicies;
33+
import io.tarantool.driver.api.tuple.TarantoolTuple;
34+
import io.tarantool.driver.auth.SimpleTarantoolCredentials;
35+
import io.tarantool.driver.auth.TarantoolCredentials;
36+
import io.tarantool.driver.cluster.BinaryClusterDiscoveryEndpoint;
37+
import io.tarantool.driver.cluster.BinaryDiscoveryClusterAddressProvider;
38+
import io.tarantool.driver.cluster.TarantoolClusterDiscoveryConfig;
39+
import io.tarantool.driver.cluster.TestWrappedClusterAddressProvider;
40+
import io.tarantool.driver.core.ClusterTarantoolTupleClient;
41+
import io.tarantool.driver.core.ProxyTarantoolTupleClient;
42+
import io.tarantool.driver.core.RetryingTarantoolTupleClient;
43+
3244
/**
3345
* @author Alexey Kuzin
3446
* @author Artyom Dubinin
@@ -49,17 +61,18 @@ public static void setUp() throws TimeoutException {
4961

5062
private TarantoolClientConfig.Builder prepareConfig() {
5163
return TarantoolClientConfig.builder()
52-
.withCredentials(new SimpleTarantoolCredentials(USER_NAME, PASSWORD))
53-
.withConnectTimeout(1000)
54-
.withReadTimeout(1000);
64+
.withCredentials(new SimpleTarantoolCredentials(USER_NAME, PASSWORD))
65+
.withConnectTimeout(1000)
66+
.withReadTimeout(1000);
5567
}
5668

5769
private RetryingTarantoolTupleClient setupRouterClient(int port, int retries, long delay) {
5870
ClusterTarantoolTupleClient clusterClient = new ClusterTarantoolTupleClient(
5971
prepareConfig().build(), container.getRouterHost(), container.getMappedPort(port));
6072

6173
return new RetryingTarantoolTupleClient(new ProxyTarantoolTupleClient(clusterClient),
62-
TarantoolRequestRetryPolicies.byNumberOfAttempts(retries).withDelay(delay).build());
74+
TarantoolRequestRetryPolicies.byNumberOfAttempts(retries)
75+
.withDelay(delay).build());
6376
}
6477

6578
private RetryingTarantoolTupleClient setupClusterClient(
@@ -70,7 +83,143 @@ private RetryingTarantoolTupleClient setupClusterClient(
7083

7184
ProxyTarantoolTupleClient client = new ProxyTarantoolTupleClient(clusterClient);
7285
return new RetryingTarantoolTupleClient(client,
73-
TarantoolRequestRetryPolicies.byNumberOfAttempts(retries, e -> true).withDelay(delay).build());
86+
TarantoolRequestRetryPolicies.byNumberOfAttempts(retries, e -> true)
87+
.withDelay(delay).build());
88+
}
89+
90+
@Test
91+
void test_roundRobin_shouldWorkCorrectly_withDiscoveryAndConnections()
92+
throws ExecutionException, InterruptedException, IOException {
93+
94+
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> clusterClient =
95+
getTarantoolClusterClientWithDiscovery(2, 5_000);
96+
97+
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> routerClient1 = getSimpleClient(3301);
98+
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> routerClient2 = getSimpleClient(3302);
99+
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> routerClient3 = getSimpleClient(3303);
100+
// 3306 isn't in cluster topology yet
101+
TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> routerClient4 = getSimpleClient(3306);
102+
103+
int callCounter = 15;
104+
for (int i = 0; i < callCounter; i++) {
105+
clusterClient.callForSingleResult(
106+
"simple_long_running_function", Arrays.asList(0, true), Boolean.class).get();
107+
}
108+
109+
String getAllConnectionCalls =
110+
"return box.space.request_counters.index.count:select(0, {iterator = box.index.GT})";
111+
// 15 calls on 3 routers on 2 connection == 15 / 3 == 5 / 2 == 2 or 3 calls per connect
112+
for (TarantoolClient router : Arrays.asList(routerClient1, routerClient2, routerClient3)) {
113+
assertEquals(Arrays.asList(2, 3), getCallCountersPerConnection(getAllConnectionCalls, router));
114+
}
115+
116+
// add new router
117+
// put 3306 in topology as router
118+
routerClient1.eval("cartridge = require('cartridge') " +
119+
"replicasets = { " +
120+
" { " +
121+
" alias = 'app-router-fourth', " +
122+
" roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' }, " +
123+
" join_servers = { { uri = 'localhost:3306' } } " +
124+
" }} " +
125+
"cartridge.admin_edit_topology({ replicasets = replicasets }) ").join();
126+
127+
// wait until discovery get topology
128+
Thread.sleep(5_000);
129+
130+
callCounter = 16;
131+
// 16 / 4 / 2 = 2 requests per connect
132+
for (int i = 0; i < callCounter; i++) {
133+
clusterClient.callForSingleResult(
134+
"simple_long_running_function", Arrays.asList(0, true), Boolean.class).get();
135+
}
136+
137+
for (TarantoolClient router :
138+
Arrays.asList(routerClient1, routerClient2, routerClient3)) {
139+
assertEquals(Arrays.asList(4, 5), getCallCountersPerConnection(getAllConnectionCalls, router));
140+
}
141+
142+
Object routerCallCounterPerConnection = getCallCountersPerConnection(getAllConnectionCalls, routerClient4);
143+
assertEquals(Arrays.asList(2, 2), routerCallCounterPerConnection);
144+
145+
String pid = container.execInContainer("pgrep", "-f", "testapp@second-router")
146+
.getStdout().replace("\n", "");
147+
container.execInContainer("kill", "-9", pid);
148+
// wait until discovery get topology
149+
Thread.sleep(5_000);
150+
151+
callCounter = 12;
152+
// 12 / 3 / 2 = 2 requests per connect
153+
for (int i = 0; i < callCounter; i++) {
154+
clusterClient.callForSingleResult(
155+
"simple_long_running_function", Arrays.asList(0, true), Boolean.class).get();
156+
}
157+
Thread.sleep(5_000);
158+
for (TarantoolClient router :
159+
Arrays.asList(routerClient1, routerClient3)) {
160+
assertEquals(Arrays.asList(6, 7), getCallCountersPerConnection(getAllConnectionCalls, router));
161+
}
162+
routerCallCounterPerConnection = getCallCountersPerConnection(getAllConnectionCalls, routerClient4);
163+
assertEquals(Arrays.asList(4, 4), routerCallCounterPerConnection);
164+
}
165+
166+
private static TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> getSimpleClient(Integer port) {
167+
return TarantoolClientFactory.createClient()
168+
.withAddress(container.getRouterHost(), container.getMappedPort(port))
169+
.withCredentials(USER_NAME, PASSWORD)
170+
.build();
171+
}
172+
173+
private static TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>>
174+
getTarantoolClusterClientWithDiscovery(
175+
int connections, int delay) {
176+
String host = container.getRouterHost();
177+
int port = container.getRouterPort();
178+
179+
TarantoolCredentials credentials = new SimpleTarantoolCredentials(
180+
USER_NAME,
181+
PASSWORD
182+
);
183+
TarantoolClientConfig config = TarantoolClientConfig.builder()
184+
.withCredentials(credentials)
185+
.build();
186+
187+
BinaryClusterDiscoveryEndpoint endpoint = new BinaryClusterDiscoveryEndpoint.Builder()
188+
.withClientConfig(config)
189+
.withEntryFunction("get_routers")
190+
.withEndpointProvider(() -> Collections.singletonList(
191+
new TarantoolServerAddress(
192+
host, port
193+
)))
194+
.build();
195+
196+
TarantoolClusterDiscoveryConfig clusterDiscoveryConfig = new TarantoolClusterDiscoveryConfig.Builder()
197+
.withDelay(delay)
198+
.withEndpoint(endpoint)
199+
.build();
200+
201+
BinaryDiscoveryClusterAddressProvider discoveryProvider = new BinaryDiscoveryClusterAddressProvider(
202+
clusterDiscoveryConfig);
203+
204+
TarantoolClusterAddressProvider wrapperDiscoveryProvider
205+
= new TestWrappedClusterAddressProvider(discoveryProvider, container); // because we use docker ports
206+
207+
return TarantoolClientFactory.createClient()
208+
.withAddressProvider(wrapperDiscoveryProvider)
209+
.withCredentials(USER_NAME, PASSWORD)
210+
.withConnections(connections)
211+
.build();
212+
}
213+
214+
@NotNull
215+
private static Object getCallCountersPerConnection(String getAllConnectionCalls, TarantoolClient router) {
216+
List<?> luaResponse = router.eval(getAllConnectionCalls).join();
217+
ArrayList tuples = (ArrayList) luaResponse.get(0); // because lua has multivalue response
218+
219+
Object routerCallCounterPerConnection = tuples.stream()
220+
.map(item -> ((ArrayList) item).get(1))
221+
.collect(Collectors.toList());
222+
return routerCallCounterPerConnection;
74223
}
75224

76225
@Test

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

Lines changed: 10 additions & 57 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,12 @@
11
local vshard = require('vshard')
22
local cartridge_rpc = require('cartridge.rpc')
3-
local fiber = require('fiber')
43
local crud = require('crud')
54
local uuid = require('uuid')
65
local log = require('log')
76

7+
local metadata_utils = require('utils.metadata')
88
local crud_utils = require('utils.crud')
9+
local counter = require('modules.counter')
910

1011
local function get_schema()
1112
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()
100101
error("Test error: raising_error() called")
101102
end
102103

103-
local function reset_request_counters()
104-
box.space.request_counters:replace({ 1, 0 })
105-
end
106-
107-
local function get_router_name()
108-
return string.sub(box.cfg.custom_proc_title, 9)
109-
end
110-
111-
local function simple_long_running_function(seconds_to_sleep)
112-
fiber.sleep(seconds_to_sleep)
113-
return true
114-
end
115-
116-
local function long_running_function(values)
117-
local seconds_to_sleep = 0
118-
local disabled_router_name = ""
119-
if values ~= nil then
120-
if type(values) == "table" then
121-
values = values or {}
122-
seconds_to_sleep = values[1]
123-
disabled_router_name = values[2]
124-
else
125-
seconds_to_sleep = values
126-
end
127-
end
128-
129-
-- need using number instead field name as string in update function for compatibility with tarantool 1.10
130-
box.space.request_counters:update(1, { { '+', 2, 1 } })
131-
log.info('Executing long-running function ' ..
132-
tostring(box.space.request_counters:get(1)[2]) ..
133-
"(name: " .. disabled_router_name ..
134-
"; sleep: " .. seconds_to_sleep .. ")")
135-
if get_router_name() == disabled_router_name then
136-
return nil, "Disabled by client; router_name = " .. disabled_router_name
137-
end
138-
if seconds_to_sleep then
139-
fiber.sleep(seconds_to_sleep)
140-
end
141-
return true
142-
end
143-
144-
local function get_request_count()
145-
return box.space.request_counters:get(1)[2]
146-
end
147-
148104
-- it's like vshard error throwing
149105
local function box_error_unpack_no_connection()
150106
return nil, box.error.new(box.error.NO_CONNECTION):unpack()
@@ -202,12 +158,6 @@ local function select_router_space()
202158
end
203159

204160
local function init_router_spaces()
205-
local request_counters = box.schema.space.create('request_counters', {
206-
format = { { 'id', 'unsigned' }, { 'count', 'unsigned' } },
207-
if_not_exists = true
208-
})
209-
request_counters:create_index('id', { parts = { 'id' }, if_not_exists = true })
210-
211161
local router_space = box.schema.space.create('router_space', {
212162
format = { { 'id', 'unsigned' } },
213163
if_not_exists = true
@@ -220,6 +170,7 @@ end
220170
local function init(opts)
221171
if opts.is_master then
222172
init_router_spaces()
173+
counter.init_counter_space()
223174
end
224175
patch_crud_methods_for_tests()
225176

@@ -235,11 +186,13 @@ local function init(opts)
235186
rawset(_G, 'retrying_function', retrying_function)
236187
rawset(_G, 'raising_error', raising_error)
237188

238-
rawset(_G, 'reset_request_counters', reset_request_counters)
239-
rawset(_G, 'get_router_name', get_router_name)
240-
rawset(_G, 'long_running_function', long_running_function)
241-
rawset(_G, 'simple_long_running_function', simple_long_running_function)
242-
rawset(_G, 'get_request_count', get_request_count)
189+
rawset(_G, 'get_router_name', metadata_utils.get_router_name)
190+
191+
rawset(_G, 'reset_request_counters', counter.reset_request_counters)
192+
rawset(_G, 'simple_long_running_function', counter.simple_long_running_function)
193+
rawset(_G, 'long_running_function', counter.long_running_function)
194+
rawset(_G, 'get_request_count', counter.get_request_count)
195+
243196
rawset(_G, 'box_error_unpack_no_connection', box_error_unpack_no_connection)
244197
rawset(_G, 'box_error_unpack_timeout', box_error_unpack_timeout)
245198
rawset(_G, 'box_error_timeout', box_error_timeout)

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
local cartridge = require('cartridge')
22

3+
local counter = require('modules.counter')
4+
35
function get_routers()
46
local function table_contains(table, element)
57
for _, value in pairs(table) do
@@ -44,10 +46,16 @@ local function init(opts)
4446
-- luacheck: no unused args
4547
if opts.is_master then
4648
box.schema.user.grant('guest', 'read,write', 'universe', nil, { if_not_exists = true })
49+
counter.init_counter_space()
4750
end
4851

4952
init_httpd()
5053

54+
rawset(_G, 'reset_request_counters', counter.reset_request_counters)
55+
rawset(_G, 'simple_long_running_function', counter.simple_long_running_function)
56+
rawset(_G, 'long_running_function', counter.long_running_function)
57+
rawset(_G, 'get_request_count', counter.get_request_count)
58+
5159
return true
5260
end
5361

‎src/test/resources/cartridge/instances.yml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,11 @@ testapp.third-router:
1313
advertise_uri: localhost:3303
1414
http_port: 8083
1515

16+
testapp.fourth-router:
17+
workdir: ./tmp/db_dev/3306
18+
advertise_uri: localhost:3306
19+
http_port: 8086
20+
1621
testapp.s1-storage:
1722
workdir: ./tmp/db_dev/3304
1823
advertise_uri: localhost:3304

0 commit comments

Comments
 (0)