Skip to content

Commit 9463d6a

Browse files
kafka1991kafka1991
committed
1. ingress support multi hosts 2. egress failover optimise
1 parent 396d20b commit 9463d6a

4 files changed

Lines changed: 257 additions & 29 deletions

File tree

‎core/src/main/java/io/questdb/client/Sender.java‎

Lines changed: 117 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -835,6 +835,93 @@ public LineSenderBuilder address(CharSequence address) {
835835
return this;
836836
}
837837

838+
private void addAddressEntry(CharSequence src, int start, int end) {
839+
int hostStart;
840+
int hostEnd;
841+
int portStart;
842+
if (src.charAt(start) == '[') {
843+
int closeBracket = Chars.indexOf(src, start + 1, end, ']');
844+
if (closeBracket < 0) {
845+
throw new LineSenderException("missing closing ']' in IPv6 addr entry [address=")
846+
.put(src.subSequence(start, end)).put("]");
847+
}
848+
hostStart = start + 1;
849+
hostEnd = closeBracket;
850+
if (closeBracket == end - 1) {
851+
portStart = -1;
852+
} else if (src.charAt(closeBracket + 1) != ':') {
853+
throw new LineSenderException("expected ':' after ']' in IPv6 addr entry [address=")
854+
.put(src.subSequence(start, end)).put("]");
855+
} else {
856+
portStart = closeBracket + 2;
857+
}
858+
} else {
859+
int firstColon = Chars.indexOf(src, start, end, ':');
860+
int lastColon = Chars.indexOf(src, start, end, ':', -1);
861+
if (firstColon != lastColon) {
862+
hostStart = start;
863+
hostEnd = end;
864+
portStart = -1;
865+
} else if (firstColon < 0) {
866+
hostStart = start;
867+
hostEnd = end;
868+
portStart = -1;
869+
} else {
870+
hostStart = start;
871+
hostEnd = firstColon;
872+
portStart = firstColon + 1;
873+
}
874+
}
875+
if (hostStart == hostEnd) {
876+
throw new LineSenderException("empty host in addr entry [address=")
877+
.put(src.subSequence(start, end)).put("]");
878+
}
879+
int parsedPort = -1;
880+
if (portStart >= 0) {
881+
if (portStart >= end) {
882+
throw new LineSenderException("invalid address, use IPv4 address or a domain name [address=")
883+
.put(src.subSequence(start, end)).put("]");
884+
}
885+
try {
886+
parsedPort = Numbers.parseInt(src, portStart, end);
887+
if (parsedPort < 1 || parsedPort > 65535) {
888+
throw new LineSenderException("invalid port [port=").put(parsedPort).put("]");
889+
}
890+
} catch (NumericException e) {
891+
throw new LineSenderException("cannot parse a port from the address, use IPv4 address or a domain name")
892+
.put(" [address=").put(src.subSequence(start, end)).put("]");
893+
}
894+
}
895+
if (parsedPort != -1) {
896+
for (int i = 0, n = hosts.size(); i < n; i++) {
897+
String storedHost = hosts.get(i);
898+
if (charsEqualsRange(storedHost, src, hostStart, hostEnd)) {
899+
if (ports.size() > i && ports.getQuick(i) == parsedPort) {
900+
throw new LineSenderException("duplicated addresses are not allowed [address=")
901+
.put(src.subSequence(start, end)).put("]");
902+
}
903+
}
904+
}
905+
}
906+
hosts.add(src.subSequence(hostStart, hostEnd).toString());
907+
if (parsedPort != -1) {
908+
ports.add(parsedPort);
909+
}
910+
}
911+
912+
private static boolean charsEqualsRange(CharSequence a, CharSequence b, int bStart, int bEnd) {
913+
int len = bEnd - bStart;
914+
if (a.length() != len) {
915+
return false;
916+
}
917+
for (int i = 0; i < len; i++) {
918+
if (a.charAt(i) != b.charAt(bStart + i)) {
919+
return false;
920+
}
921+
}
922+
return true;
923+
}
924+
838925
/**
839926
* Advanced TLS configuration. Most users should not need to use this.
840927
*
@@ -1162,7 +1249,8 @@ public Sender build() {
11621249
errorHandler,
11631250
actualErrorInboxCapacity,
11641251
actualDurableAckKeepaliveIntervalMillis,
1165-
authTimeoutMillis
1252+
authTimeoutMillis,
1253+
gorillaEnabled
11661254
);
11671255
} catch (Throwable t) {
11681256
// connect() failed before ownership of cursorEngine
@@ -1174,7 +1262,6 @@ public Sender build() {
11741262
}
11751263
throw t;
11761264
}
1177-
connected.setGorillaEnabled(gorillaEnabled);
11781265
// connect() succeeded — `connected` now owns cursorEngine
11791266
// via setCursorEngine(engine, true). From here on, ANY
11801267
// failure must close `connected` (which closes the engine
@@ -1986,10 +2073,10 @@ public LineSenderBuilder durableAckKeepaliveIntervalMillis(long millis) {
19862073
public LineSenderBuilder authTimeoutMillis(long millis) {
19872074
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
19882075
throw new LineSenderException(
1989-
"auth_timeout is only supported for WebSocket transport");
2076+
"auth_timeout_ms is only supported for WebSocket transport");
19902077
}
19912078
if (millis <= 0L) {
1992-
throw new LineSenderException("auth_timeout must be > 0: ").put(millis);
2079+
throw new LineSenderException("auth_timeout_ms must be > 0: ").put(millis);
19932080
}
19942081
this.authTimeoutMillis = millis;
19952082
return this;
@@ -2466,13 +2553,28 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
24662553
}
24672554
if (Chars.equals("addr", sink)) {
24682555
pos = getValue(configurationString, pos, sink, "address");
2469-
address(sink);
2470-
if (ports.size() == hosts.size() - 1) {
2471-
// not set
2472-
port(protocol == PROTOCOL_HTTP ? DEFAULT_HTTP_PORT
2473-
: protocol == PROTOCOL_UDP ? DEFAULT_UDP_PORT
2474-
: protocol == PROTOCOL_WEBSOCKET ? DEFAULT_WEBSOCKET_PORT
2475-
: DEFAULT_TCP_PORT);
2556+
int defaultPort = protocol == PROTOCOL_HTTP ? DEFAULT_HTTP_PORT
2557+
: protocol == PROTOCOL_UDP ? DEFAULT_UDP_PORT
2558+
: protocol == PROTOCOL_WEBSOCKET ? DEFAULT_WEBSOCKET_PORT
2559+
: DEFAULT_TCP_PORT;
2560+
int valLen = sink.length();
2561+
int entryStart = 0;
2562+
for (int i = 0; i <= valLen; i++) {
2563+
if (i == valLen || sink.charAt(i) == ',') {
2564+
int s = entryStart;
2565+
int e = i;
2566+
while (s < e && sink.charAt(s) == ' ') s++;
2567+
while (e > s && sink.charAt(e - 1) == ' ') e--;
2568+
if (s == e) {
2569+
throw new LineSenderException("empty addr entry");
2570+
}
2571+
int portsBefore = ports.size();
2572+
addAddressEntry(sink, s, e);
2573+
if (ports.size() == portsBefore) {
2574+
port(defaultPort);
2575+
}
2576+
entryStart = i + 1;
2577+
}
24762578
}
24772579
} else if (Chars.equals("user", sink)) {
24782580
// deprecated key: user, new key: username
@@ -2698,12 +2800,12 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
26982800
}
26992801
pos = getValue(configurationString, pos, sink, "close_flush_timeout_millis");
27002802
closeFlushTimeoutMillis(parseLongValue(sink, "close_flush_timeout_millis"));
2701-
} else if (Chars.equals("auth_timeout", sink)) {
2803+
} else if (Chars.equals("auth_timeout_ms", sink)) {
27022804
if (protocol != PROTOCOL_WEBSOCKET) {
2703-
throw new LineSenderException("auth_timeout is only supported for WebSocket transport");
2805+
throw new LineSenderException("auth_timeout_ms is only supported for WebSocket transport");
27042806
}
2705-
pos = getValue(configurationString, pos, sink, "auth_timeout");
2706-
authTimeoutMillis(parseLongValue(sink, "auth_timeout"));
2807+
pos = getValue(configurationString, pos, sink, "auth_timeout_ms");
2808+
authTimeoutMillis(parseLongValue(sink, "auth_timeout_ms"));
27072809
} else if (Chars.equals("gorilla", sink)) {
27082810
if (protocol != PROTOCOL_WEBSOCKET) {
27092811
throw new LineSenderException("gorilla is only supported for WebSocket transport");

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpQueryClient.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import java.util.ArrayList;
4343
import java.util.Base64;
4444
import java.util.List;
45+
import java.util.concurrent.ThreadLocalRandom;
4546
import java.util.concurrent.atomic.AtomicBoolean;
4647
import java.util.concurrent.atomic.AtomicReference;
4748

@@ -891,10 +892,9 @@ public void execute(String sql, QwpBindSetter binds, QwpColumnBatchHandler handl
891892
if (failoverInitialBackoffMs > 0L) {
892893
long base = failoverInitialBackoffMs << Math.min(attempt - 1, 30);
893894
long capped = Math.min(base, failoverMaxBackoffMs);
894-
long jitter = capped > 0L
895-
? java.util.concurrent.ThreadLocalRandom.current().nextLong(capped)
895+
long delay = capped > 0L
896+
? ThreadLocalRandom.current().nextLong(capped + 1L)
896897
: 0L;
897-
long delay = Math.min(capped + jitter, failoverMaxBackoffMs);
898898
if (delay > 0L) {
899899
try {
900900
Thread.sleep(delay);

‎core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -137,10 +137,8 @@ public class QwpWebSocketSender implements Sender {
137137
private final GlobalSymbolDictionary globalSymbolDictionary;
138138
private final List<Endpoint> endpoints;
139139
private final QwpHostHealthTracker hostTracker;
140-
private final String host;
141140
private final int inFlightWindowSize;
142141
private final int maxSchemasPerConnection;
143-
private final int port;
144142
private volatile int currentEndpointIdx = -1;
145143
private final CharSequenceObjHashMap<QwpTableBuffer> tableBuffers;
146144
// null means plain text (no TLS)
@@ -236,8 +234,6 @@ private QwpWebSocketSender(
236234
this.endpoints = Collections.unmodifiableList(new ArrayList<>(endpoints));
237235
this.hostTracker = new QwpHostHealthTracker(this.endpoints.size());
238236
this.authorizationHeader = authorizationHeader;
239-
this.host = this.endpoints.get(0).host;
240-
this.port = this.endpoints.get(0).port;
241237
this.tlsConfig = tlsConfig;
242238
this.encoder = new QwpWebSocketEncoder(DEFAULT_BUFFER_SIZE);
243239
this.tableBuffers = new CharSequenceObjHashMap<>();
@@ -498,7 +494,7 @@ public static QwpWebSocketSender connect(
498494
closeFlushTimeoutMillis, reconnectMaxDurationMillis,
499495
reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
500496
initialConnectMode, errorHandler, errorInboxCapacity,
501-
durableAckKeepaliveIntervalMillis, DEFAULT_AUTH_TIMEOUT_MS);
497+
durableAckKeepaliveIntervalMillis, DEFAULT_AUTH_TIMEOUT_MS, true);
502498
}
503499

504500
/**
@@ -524,7 +520,8 @@ public static QwpWebSocketSender connect(
524520
SenderErrorHandler errorHandler,
525521
int errorInboxCapacity,
526522
long durableAckKeepaliveIntervalMillis,
527-
long authTimeoutMs
523+
long authTimeoutMs,
524+
boolean gorillaEnabled
528525
) {
529526
QwpWebSocketSender sender = new QwpWebSocketSender(
530527
endpoints, tlsConfig,
@@ -534,6 +531,8 @@ public static QwpWebSocketSender connect(
534531
try {
535532
sender.requestDurableAck = requestDurableAck;
536533
sender.authTimeoutMs = authTimeoutMs;
534+
sender.gorillaEnabled = gorillaEnabled;
535+
sender.encoder.setGorillaEnabled(gorillaEnabled);
537536
sender.closeFlushTimeoutMillis = closeFlushTimeoutMillis;
538537
sender.reconnectMaxDurationMillis = reconnectMaxDurationMillis;
539538
sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
@@ -1894,7 +1893,7 @@ private void atNanos(long timestampNanos) {
18941893
* attempt (and, in the follow-up commit, schedules a backoff retry
18951894
* within the per-outage time cap).
18961895
*/
1897-
private WebSocketClient buildAndConnect() {
1896+
private synchronized WebSocketClient buildAndConnect() {
18981897
int previousIdx = currentEndpointIdx;
18991898
if (previousIdx >= 0) {
19001899
hostTracker.recordMidStreamFailure(previousIdx);
@@ -2146,22 +2145,25 @@ private void ensureConnected() {
21462145
client.close();
21472146
client = null;
21482147
}
2148+
Endpoint ep = currentEndpoint();
21492149
throw new LineSenderException(
2150-
"Failed to start cursor I/O thread for " + host + ":" + port, t);
2150+
"Failed to start cursor I/O thread for " + ep.host + ":" + ep.port, t);
21512151
}
21522152

21532153
if (client != null) {
2154+
Endpoint ep = currentEndpoint();
21542155
encoder.setVersion((byte) client.getServerQwpVersion());
21552156
LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}, qwpVersion={}]",
2156-
host, port, inFlightWindowSize, client.getServerQwpVersion());
2157+
ep.host, ep.port, inFlightWindowSize, client.getServerQwpVersion());
21572158
} else {
21582159
// Async mode: I/O thread will drive the connect. Encoder uses
21592160
// its default version (V1). Schema state still gets reset for
21602161
// consistency with the sync path; the post-connect replay path
21612162
// does not need a producer-side reset signal because every
21622163
// cursor frame is self-sufficient.
2163-
LOG.info("Async initial connect deferred to I/O thread [host={}, port={}, windowSize={}]",
2164-
host, port, inFlightWindowSize);
2164+
Endpoint ep = endpoints.get(0);
2165+
LOG.info("Async initial connect deferred to I/O thread [host={}, port={}, endpointCount={}, windowSize={}]",
2166+
ep.host, ep.port, endpoints.size(), inFlightWindowSize);
21652167
}
21662168
// Server starts fresh on each connection — discard any schema IDs
21672169
// retained from prior state. Cursor frames are self-sufficient (every
@@ -2459,6 +2461,11 @@ private static List<Endpoint> singleEndpoint(String host, int port) {
24592461
return Collections.singletonList(new Endpoint(host, port));
24602462
}
24612463

2464+
private Endpoint currentEndpoint() {
2465+
int idx = currentEndpointIdx;
2466+
return endpoints.get(Math.max(idx, 0));
2467+
}
2468+
24622469
public static final class Endpoint {
24632470
public final String host;
24642471
public final int port;

0 commit comments

Comments
 (0)