Skip to content

Commit b86a099

Browse files
vietjvietj
committed
Improve service resolver testing.
- use an HTTP proxy between Vert.x HTTP client and the Kubernetes API server so we can operate on the traffic - improve the checking of available pods in tests
1 parent e22a16b commit b86a099

File tree

9 files changed

+319
-265
lines changed

9 files changed

+319
-265
lines changed

‎src/main/generated/io/vertx/serviceresolver/kube/KubeResolverOptionsConverter.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22

33
import io.vertx.core.json.JsonObject;
44
import io.vertx.core.json.JsonArray;
5-
import java.time.Instant;
6-
import java.time.format.DateTimeFormatter;
75

86
/**
97
* Converter and mapper for {@link io.vertx.serviceresolver.kube.KubeResolverOptions}.

‎src/main/generated/io/vertx/serviceresolver/srv/SrvResolverOptionsConverter.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22

33
import io.vertx.core.json.JsonObject;
44
import io.vertx.core.json.JsonArray;
5-
import java.time.Instant;
6-
import java.time.format.DateTimeFormatter;
75

86
/**
97
* Converter and mapper for {@link io.vertx.serviceresolver.srv.SrvResolverOptions}.

‎src/test/java/io/vertx/tests/HttpProxy.java‎

Lines changed: 69 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -2,32 +2,42 @@
22

33
import io.vertx.core.Vertx;
44
import io.vertx.core.http.*;
5-
import io.vertx.core.net.JksOptions;
5+
import io.vertx.core.net.KeyCertOptions;
66
import io.vertx.core.net.SocketAddress;
7+
import io.vertx.core.net.TrustOptions;
78

89
import java.util.Set;
910
import java.util.concurrent.ConcurrentHashMap;
1011
import java.util.concurrent.TimeUnit;
1112
import java.util.concurrent.atomic.AtomicBoolean;
13+
import java.util.function.Predicate;
1214

1315
public class HttpProxy {
1416

1517
private final Vertx vertx;
1618
private HttpServer server;
1719
private HttpClient httpClient;
18-
private WebSocketClient wsClient;
19-
private SocketAddress origin;
20-
private int port;
20+
private WebSocketClient webSocketClient;
2121
private final Set<WebSocketBase> webSockets = ConcurrentHashMap.newKeySet();
22+
private HttpClientOptions httpConfig = new HttpClientOptions().setTrustAll(true);
23+
private WebSocketClientOptions webSocketConfig = new WebSocketClientOptions().setTrustAll(true);
24+
private int port;
25+
private Predicate<HttpServerRequest> requestHandler = req -> true;
2226

2327
public HttpProxy(Vertx vertx) {
2428
this.vertx = vertx;
25-
this.httpClient = vertx.createHttpClient();
26-
this.wsClient = vertx.createWebSocketClient();
29+
}
30+
31+
public HttpProxy requestHandler(Predicate<HttpServerRequest> requestHandler) {
32+
this.requestHandler = requestHandler == null ? req -> true : requestHandler;
33+
return this;
2734
}
2835

2936
public HttpProxy origin(SocketAddress origin) {
30-
this.origin = origin;
37+
httpConfig.setDefaultPort(origin.port());
38+
httpConfig.setDefaultHost(origin.host());
39+
webSocketConfig.setDefaultPort(origin.port());
40+
webSocketConfig.setDefaultHost(origin.host());
3141
return this;
3242
}
3343

@@ -36,13 +46,33 @@ public HttpProxy port(int port) {
3646
return this;
3747
}
3848

49+
public HttpProxy keyCertOptions(KeyCertOptions keyCertOptions) {
50+
httpConfig.setKeyCertOptions(keyCertOptions.copy());
51+
webSocketConfig.setKeyCertOptions(keyCertOptions.copy());
52+
return this;
53+
}
54+
55+
public HttpProxy trustOptions(TrustOptions trustOptions) {
56+
httpConfig.setTrustOptions(trustOptions.copy());
57+
webSocketConfig.setTrustOptions(trustOptions.copy());
58+
return this;
59+
}
60+
61+
public HttpProxy protocol(HttpVersion version) {
62+
httpConfig.setProtocolVersion(version);
63+
return this;
64+
}
65+
66+
public HttpProxy ssl(boolean ssl) {
67+
httpConfig.setSsl(ssl);
68+
webSocketConfig.setSsl(ssl);
69+
return this;
70+
}
71+
3972
public void start() throws Exception {
40-
HttpServer server = vertx.createHttpServer(new HttpServerOptions()
41-
.setSsl(true)
42-
.setKeyCertOptions(new JksOptions()
43-
.setPath("server-keystore.jks")
44-
.setPassword("wibble"))
45-
);
73+
this.httpClient = vertx.createHttpClient(httpConfig);
74+
this.webSocketClient = vertx.createWebSocketClient(webSocketConfig);
75+
server = vertx.createHttpServer();
4676
server.requestHandler(this::handleRequest);
4777
server
4878
.listen(port)
@@ -52,18 +82,22 @@ public void start() throws Exception {
5282
}
5383

5484
private void handleRequest(HttpServerRequest serverRequest) {
55-
if (serverRequest.getHeader("upgrade") != null) {
85+
Predicate<HttpServerRequest> handler = requestHandler;
86+
if (!handler.test(serverRequest)) {
87+
return;
88+
}
89+
if (serverRequest.canUpgradeToWebSocket()) {
5690
WebSocketConnectOptions options = new WebSocketConnectOptions();
57-
options.setServer(origin);
5891
options.setURI(serverRequest.uri());
5992
serverRequest.pause();
60-
wsClient.connect(options).onComplete(ar -> {
93+
webSocketClient.connect(options).onComplete(ar -> {
6194
if (ar.succeeded()) {
6295
WebSocket wsc = ar.result();
6396
AtomicBoolean closed = new AtomicBoolean();
6497
wsc.closeHandler(v -> {
6598
closed.set(true);
6699
});
100+
wsc.pause();
67101
serverRequest.toWebSocket().onComplete(ar2 -> {
68102
if (!closed.get()) {
69103
if (ar2.succeeded()) {
@@ -83,6 +117,7 @@ private void handleRequest(HttpServerRequest serverRequest) {
83117
wss.close();
84118
});
85119
webSockets.add(wss);
120+
wsc.resume();
86121
} else {
87122
wsc.close();
88123
}
@@ -98,37 +133,35 @@ private void handleRequest(HttpServerRequest serverRequest) {
98133
});
99134
} else {
100135
RequestOptions options = new RequestOptions()
101-
.setServer(origin)
102136
.setMethod(serverRequest.method())
103137
.setURI(serverRequest.uri());
104-
serverRequest.body().onComplete(ar_ -> {
105-
if (ar_.succeeded()) {
106-
httpClient.request(options).onComplete(ar -> {
138+
serverRequest.body().onSuccess(requestBody -> {
139+
httpClient
140+
.request(options)
141+
.compose(clientRequest -> clientRequest
142+
.send(requestBody)
143+
.compose(clientResponse -> {
144+
serverRequest.response().setStatusCode(clientResponse.statusCode());
145+
return clientResponse.body();
146+
}))
147+
.onComplete(ar -> {
107148
if (ar.succeeded()) {
108-
HttpClientRequest clientRequest = ar.result();
109-
clientRequest.send(ar_.result()).onComplete(ar2 -> {
110-
if (ar2.succeeded()) {
111-
HttpClientResponse clientResponse = ar2.result();
112-
HttpServerResponse serverResponse = serverRequest.response();
113-
serverResponse.putHeader(HttpHeaders.CONTENT_LENGTH, clientResponse.getHeader(HttpHeaders.CONTENT_LENGTH));
114-
clientResponse.pipeTo(serverResponse);
115-
} else {
116-
serverRequest.response().setStatusCode(500).end();
117-
}
118-
});
149+
serverRequest.response().end(ar.result());
119150
} else {
120-
ar.cause().printStackTrace();
121-
serverRequest.response().reset();
151+
serverRequest.response().setStatusCode(500).end();
122152
}
123153
});
124-
} else {
125-
// Nothing to do ? (compose?)
126-
}
127154
});
128155
}
129156
}
130157

131158
public Set<WebSocketBase> webSockets() {
132159
return webSockets;
133160
}
161+
162+
public void close() {
163+
webSocketClient.close().await();
164+
httpClient.close().await();
165+
server.close().await();
166+
}
134167
}

‎src/test/java/io/vertx/tests/ServiceResolverTestBase.java‎

Lines changed: 73 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import io.vertx.core.Vertx;
66
import io.vertx.core.buffer.Buffer;
77
import io.vertx.core.http.*;
8+
import io.vertx.core.net.AddressResolver;
89
import io.vertx.core.net.SocketAddress;
910
import io.vertx.ext.unit.junit.VertxUnitRunner;
1011
import io.vertx.serviceresolver.ServiceAddress;
@@ -13,24 +14,20 @@
1314
import org.junit.Before;
1415
import org.junit.runner.RunWith;
1516

16-
import java.lang.reflect.UndeclaredThrowableException;
17-
import java.util.ArrayList;
18-
import java.util.HashSet;
19-
import java.util.List;
20-
import java.util.Set;
17+
import java.util.*;
2118
import java.util.concurrent.Callable;
22-
import java.util.concurrent.ExecutionException;
2319
import java.util.concurrent.TimeUnit;
24-
import java.util.concurrent.TimeoutException;
2520
import java.util.function.Predicate;
2621
import java.util.stream.Collectors;
2722

23+
import static org.junit.Assert.fail;
24+
2825
@RunWith(VertxUnitRunner.class)
2926
public abstract class ServiceResolverTestBase {
3027

3128
protected Vertx vertx;
32-
protected HttpClient client;
3329
protected List<HttpServer> pods;
30+
protected Client client;
3431

3532
@Before
3633
public void setUp() throws Exception {
@@ -40,6 +37,9 @@ public void setUp() throws Exception {
4037

4138
@After
4239
public void tearDown() throws Exception {
40+
if (client != null) {
41+
client.close();
42+
}
4343
vertx.close()
4444
.toCompletionStage()
4545
.toCompletableFuture()
@@ -91,12 +91,71 @@ protected void assertWaitUntil(Callable<Boolean> cond) throws Exception {
9191
}
9292

9393
protected Buffer get(ServiceAddress addr) throws Exception {
94+
Client c;
95+
synchronized (this) {
96+
c = client;
97+
if (c == null) {
98+
c = client();
99+
this.client = c;
100+
}
101+
}
102+
return c.get(addr);
103+
}
104+
105+
protected Client client() {
106+
return client(resolver());
107+
}
108+
109+
protected Client client(AddressResolver<?> resolver) {
110+
return new Client(resolver);
111+
}
112+
113+
protected abstract AddressResolver<?> resolver();
114+
115+
protected class Client {
116+
117+
protected HttpClient client;
94118

95-
Future<Buffer> fut = client
96-
.request(new RequestOptions()
97-
.setServer(addr))
98-
.compose(req -> req.send()
99-
.compose(HttpClientResponse::body));
100-
return fut.await(20, TimeUnit.SECONDS);
119+
public Client(AddressResolver<?> resolver) {
120+
this.client = vertx.httpClientBuilder()
121+
.withAddressResolver(resolver)
122+
.build();
123+
}
124+
125+
public Buffer get(ServiceAddress addr) throws Exception {
126+
Future<Buffer> fut = client
127+
.request(new RequestOptions()
128+
.setServer(addr))
129+
.compose(req -> req.send()
130+
.compose(HttpClientResponse::body));
131+
return fut.await(20, TimeUnit.SECONDS);
132+
}
133+
134+
public void close() {
135+
client.close().await();
136+
}
137+
}
138+
139+
protected void checkEndpoints(ServiceAddress svc, String... values) {
140+
Set<String> expected = new HashSet<>(Arrays.asList(values));
141+
int retries = 5;
142+
for (int r = 0;r < retries;r++) {
143+
Set<String> found = new HashSet<>();
144+
for (int i = 0;i < values.length * 10;i++) {
145+
try {
146+
found.add(get(svc).toString());
147+
} catch (Exception ignore) {
148+
}
149+
}
150+
if (found.equals(expected)) {
151+
return;
152+
}
153+
try {
154+
Thread.sleep(10);
155+
} catch (InterruptedException e) {
156+
throw new RuntimeException(e);
157+
}
158+
}
159+
fail();
101160
}
102161
}

‎src/test/java/io/vertx/tests/kube/KubeServiceResolverKindTest.java‎

Lines changed: 1 addition & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,17 @@
11
package io.vertx.tests.kube;
22

33
import com.dajudge.kindcontainer.ApiServerContainer;
4-
import io.fabric8.kubernetes.client.Config;
5-
import io.fabric8.kubernetes.client.http.TlsVersion;
64
import io.netty.util.NetUtil;
75
import io.vertx.core.Handler;
8-
import io.vertx.core.buffer.Buffer;
9-
import io.vertx.core.http.HttpClientOptions;
106
import io.vertx.core.http.HttpServerRequest;
11-
import io.vertx.core.http.HttpVersion;
12-
import io.vertx.core.http.WebSocketClientOptions;
137
import io.vertx.core.net.*;
14-
import io.vertx.serviceresolver.kube.KubeResolver;
15-
import io.vertx.serviceresolver.kube.KubeResolverOptions;
168
import org.junit.Rule;
179

1810
import java.net.Inet4Address;
1911
import java.net.InetAddress;
2012
import java.net.NetworkInterface;
21-
import java.net.URL;
2213
import java.util.*;
2314
import java.util.stream.Collectors;
24-
import java.util.stream.Stream;
25-
26-
import static io.fabric8.kubernetes.client.Config.fromKubeconfig;
2715

2816
public class KubeServiceResolverKindTest extends KubeServiceResolverTestBase {
2917

@@ -32,40 +20,8 @@ public class KubeServiceResolverKindTest extends KubeServiceResolverTestBase {
3220
public ApiServerContainer<?> K8S = new ApiServerContainer<>();
3321

3422
public void setUp() throws Exception {
35-
super.setUp();
36-
3723
kubernetesMocking = new KubernetesMocking(K8S);
38-
39-
Config cfg = fromKubeconfig(K8S.getKubeconfig());
40-
URL url = new URL(cfg.getMasterUrl());
41-
HttpClientOptions httpClientOptions = new HttpClientOptions();
42-
WebSocketClientOptions wsClientOptions = new WebSocketClientOptions();
43-
if (cfg.getTlsVersions() != null && cfg.getTlsVersions().length > 0) {
44-
Stream.of(cfg.getTlsVersions()).map(TlsVersion::javaName).forEach(httpClientOptions::addEnabledSecureTransportProtocol);
45-
Stream.of(cfg.getTlsVersions()).map(TlsVersion::javaName).forEach(wsClientOptions::addEnabledSecureTransportProtocol);
46-
}
47-
if (cfg.isHttp2Disable()) {
48-
httpClientOptions.setProtocolVersion(HttpVersion.HTTP_1_1);
49-
}
50-
Buffer caCert = Buffer.buffer(Base64.getDecoder().decode(cfg.getCaCertData()));
51-
Buffer clientKey = Buffer.buffer(Base64.getDecoder().decode(cfg.getClientKeyData()));
52-
Buffer clientCert = Buffer.buffer(Base64.getDecoder().decode(cfg.getClientCertData()));
53-
KeyCertOptions keyCerts = new PemKeyCertOptions().addKeyValue(clientKey).addCertValue(clientCert);
54-
TrustOptions trust = new PemTrustOptions().addCertValue(caCert);
55-
httpClientOptions
56-
.setSsl(true)
57-
.setKeyCertOptions(keyCerts)
58-
.setTrustOptions(trust);
59-
wsClientOptions
60-
.setSsl(true)
61-
.setKeyCertOptions(keyCerts)
62-
.setTrustOptions(trust);
63-
KubeResolverOptions options = new KubeResolverOptions()
64-
.setNamespace(kubernetesMocking.defaultNamespace())
65-
.setServer(SocketAddress.inetSocketAddress(url.getPort(), url.getHost()))
66-
.setHttpClientOptions(httpClientOptions)
67-
.setWebSocketClientOptions(wsClientOptions);
68-
client = vertx.httpClientBuilder().withAddressResolver(KubeResolver.create(options)).build();
24+
super.setUp();
6925
}
7026

7127
private String determineHostAddress() {

0 commit comments

Comments
 (0)