Skip to content

Commit 396d20b

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

13 files changed

Lines changed: 1123 additions & 97 deletions

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

Lines changed: 55 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -729,7 +729,9 @@ public int getTimeout() {
729729
// build() time. 0 or negative is a documented "disable" value, so
730730
// a Long.MIN_VALUE sentinel keeps it distinguishable from "unset".
731731
private static final long DURABLE_ACK_KEEPALIVE_NOT_SET = Long.MIN_VALUE;
732+
private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS;
732733
private long durableAckKeepaliveIntervalMillis = DURABLE_ACK_KEEPALIVE_NOT_SET;
734+
private boolean gorillaEnabled = true;
733735
// Drives the initial-connect strategy. OFF is fail-fast (default).
734736
// SYNC retries on the user thread up to the reconnect cap. ASYNC
735737
// returns immediately and lets the I/O thread retry in the
@@ -1029,8 +1031,8 @@ public Sender build() {
10291031
}
10301032

10311033
if (protocol == PROTOCOL_WEBSOCKET) {
1032-
if (hosts.size() != 1 || ports.size() != 1) {
1033-
throw new LineSenderException("only a single address (host:port) is supported for WebSocket transport");
1034+
if (hosts.size() < 1 || ports.size() != hosts.size()) {
1035+
throw new LineSenderException("WebSocket transport requires at least one host:port pair");
10341036
}
10351037

10361038
int actualAutoFlushRows = autoFlushRows == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_ROWS : autoFlushRows;
@@ -1134,11 +1136,15 @@ public Sender build() {
11341136
int actualErrorInboxCapacity = errorInboxCapacity != PARAMETER_NOT_SET_EXPLICITLY
11351137
? errorInboxCapacity
11361138
: io.questdb.client.cutlass.qwp.client.sf.cursor.SenderErrorDispatcher.DEFAULT_CAPACITY;
1139+
java.util.List<QwpWebSocketSender.Endpoint> wsEndpoints =
1140+
new java.util.ArrayList<>(hosts.size());
1141+
for (int i = 0, n = hosts.size(); i < n; i++) {
1142+
wsEndpoints.add(new QwpWebSocketSender.Endpoint(hosts.getQuick(i), ports.getQuick(i)));
1143+
}
11371144
QwpWebSocketSender connected;
11381145
try {
11391146
connected = QwpWebSocketSender.connect(
1140-
hosts.getQuick(0),
1141-
ports.getQuick(0),
1147+
wsEndpoints,
11421148
wsTlsConfig,
11431149
actualAutoFlushRows,
11441150
actualAutoFlushBytes,
@@ -1155,7 +1161,8 @@ public Sender build() {
11551161
initialConnectMode,
11561162
errorHandler,
11571163
actualErrorInboxCapacity,
1158-
actualDurableAckKeepaliveIntervalMillis
1164+
actualDurableAckKeepaliveIntervalMillis,
1165+
authTimeoutMillis
11591166
);
11601167
} catch (Throwable t) {
11611168
// connect() failed before ownership of cursorEngine
@@ -1167,6 +1174,7 @@ public Sender build() {
11671174
}
11681175
throw t;
11691176
}
1177+
connected.setGorillaEnabled(gorillaEnabled);
11701178
// connect() succeeded — `connected` now owns cursorEngine
11711179
// via setCursorEngine(engine, true). From here on, ANY
11721180
// failure must close `connected` (which closes the engine
@@ -1971,6 +1979,30 @@ public LineSenderBuilder durableAckKeepaliveIntervalMillis(long millis) {
19711979
return this;
19721980
}
19731981

1982+
/**
1983+
* Per-endpoint timeout on the WebSocket upgrade response read. Default
1984+
* {@value QwpWebSocketSender#DEFAULT_AUTH_TIMEOUT_MS} ms.
1985+
*/
1986+
public LineSenderBuilder authTimeoutMillis(long millis) {
1987+
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
1988+
throw new LineSenderException(
1989+
"auth_timeout is only supported for WebSocket transport");
1990+
}
1991+
if (millis <= 0L) {
1992+
throw new LineSenderException("auth_timeout must be > 0: ").put(millis);
1993+
}
1994+
this.authTimeoutMillis = millis;
1995+
return this;
1996+
}
1997+
1998+
public LineSenderBuilder gorilla(boolean enabled) {
1999+
if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
2000+
throw new LineSenderException("gorilla is only supported for WebSocket transport");
2001+
}
2002+
this.gorillaEnabled = enabled;
2003+
return this;
2004+
}
2005+
19742006
/**
19752007
* Per-outage cap on the cursor I/O loop's reconnect retry budget.
19762008
* Once a wire failure occurs, the loop retries with exponential
@@ -2666,6 +2698,24 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
26662698
}
26672699
pos = getValue(configurationString, pos, sink, "close_flush_timeout_millis");
26682700
closeFlushTimeoutMillis(parseLongValue(sink, "close_flush_timeout_millis"));
2701+
} else if (Chars.equals("auth_timeout", sink)) {
2702+
if (protocol != PROTOCOL_WEBSOCKET) {
2703+
throw new LineSenderException("auth_timeout is only supported for WebSocket transport");
2704+
}
2705+
pos = getValue(configurationString, pos, sink, "auth_timeout");
2706+
authTimeoutMillis(parseLongValue(sink, "auth_timeout"));
2707+
} else if (Chars.equals("gorilla", sink)) {
2708+
if (protocol != PROTOCOL_WEBSOCKET) {
2709+
throw new LineSenderException("gorilla is only supported for WebSocket transport");
2710+
}
2711+
pos = getValue(configurationString, pos, sink, "gorilla");
2712+
if (Chars.equals("on", sink) || Chars.equals("true", sink)) {
2713+
gorilla(true);
2714+
} else if (Chars.equals("off", sink) || Chars.equals("false", sink)) {
2715+
gorilla(false);
2716+
} else {
2717+
throw new LineSenderException("invalid gorilla [value=").put(sink).put(", allowed=[on, off]]");
2718+
}
26692719
} else if (Chars.equals("durable_ack_keepalive_interval_millis", sink)) {
26702720
if (protocol != PROTOCOL_WEBSOCKET) {
26712721
throw new LineSenderException(

‎core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ public abstract class WebSocketClient implements QuietCloseable {
7575
private static final int PARSE_INCOMPLETE = 0;
7676
private static final int PARSE_NEED_MORE = -1;
7777
private static final int PARSE_OK = 1;
78+
private static final String QUESTDB_ROLE_HEADER_NAME = "X-QuestDB-Role:";
7879
private static final String QWP_DURABLE_ACK_ENABLED_VALUE = "enabled";
7980
private static final String QWP_DURABLE_ACK_HEADER_NAME = "X-QWP-Durable-Ack:";
8081
private static final String QWP_VERSION_HEADER_NAME = "X-QWP-Version:";
@@ -133,6 +134,7 @@ public abstract class WebSocketClient implements QuietCloseable {
133134
// setQwpRequestDurableAck) is the early-fail signal.
134135
private boolean serverDurableAckEnabled;
135136
private int serverQwpVersion = 1;
137+
private String upgradeRejectRole;
136138
private boolean upgraded;
137139

138140
public WebSocketClient(HttpClientConfiguration configuration, SocketFactory socketFactory) {
@@ -296,6 +298,16 @@ public int getServerQwpVersion() {
296298
return serverQwpVersion;
297299
}
298300

301+
/**
302+
* If the most recent {@link #upgrade} was rejected with a 503 carrying an
303+
* {@code X-QuestDB-Role} header, returns that role (e.g. {@code REPLICA},
304+
* {@code PRIMARY_CATCHUP}). Returns null otherwise. Read after a failed
305+
* upgrade to classify the rejection by replication role.
306+
*/
307+
public String getUpgradeRejectRole() {
308+
return upgradeRejectRole;
309+
}
310+
299311
/**
300312
* Returns whether the WebSocket is connected and upgraded.
301313
*/
@@ -624,6 +636,23 @@ private static boolean extractDurableAckEnabled(String response) {
624636
return false;
625637
}
626638

639+
private static String extractRoleHeader(String response) {
640+
int headerLen = QUESTDB_ROLE_HEADER_NAME.length();
641+
int responseLen = response.length();
642+
for (int i = 0; i <= responseLen - headerLen; i++) {
643+
if (response.regionMatches(true, i, QUESTDB_ROLE_HEADER_NAME, 0, headerLen)) {
644+
int valueStart = i + headerLen;
645+
int lineEnd = response.indexOf('\r', valueStart);
646+
if (lineEnd < 0) {
647+
lineEnd = responseLen;
648+
}
649+
String value = response.substring(valueStart, lineEnd).trim();
650+
return value.isEmpty() ? null : value;
651+
}
652+
}
653+
return null;
654+
}
655+
627656
private static int extractQwpVersion(String response) {
628657
int headerLen = QWP_VERSION_HEADER_NAME.length();
629658
int responseLen = response.length();
@@ -1031,6 +1060,9 @@ private void validateUpgradeResponse(int headerEnd) {
10311060
// Check status line
10321061
if (!response.startsWith("HTTP/1.1 101")) {
10331062
String statusLine = response.split("\r\n")[0];
1063+
if (statusLine.startsWith("HTTP/1.1 503")) {
1064+
upgradeRejectRole = extractRoleHeader(response);
1065+
}
10341066
throw new HttpClientException("WebSocket upgrade failed: ").put(statusLine);
10351067
}
10361068

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

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
/*******************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
package io.questdb.client.cutlass.qwp.client;
26+
27+
/**
28+
* Per-client bookkeeping that ranks the configured endpoint list when picking
29+
* the next host to try. Mirrors the .NET client's QwpHostHealthTracker.
30+
* <p>
31+
* Within a round, {@link #pickNext()} returns the highest-priority host that
32+
* has not yet been attempted; the caller advances the round via
33+
* {@link #beginRound(boolean)}.
34+
*/
35+
public final class QwpHostHealthTracker {
36+
public enum HostState {
37+
UNKNOWN,
38+
HEALTHY,
39+
TRANSIENT_REJECT,
40+
TRANSPORT_ERROR,
41+
TOPOLOGY_REJECT,
42+
}
43+
44+
private static final HostState[] PRIORITY_ORDER = {
45+
HostState.HEALTHY,
46+
HostState.UNKNOWN,
47+
HostState.TRANSIENT_REJECT,
48+
HostState.TRANSPORT_ERROR,
49+
HostState.TOPOLOGY_REJECT,
50+
};
51+
52+
private final boolean[] attemptedThisRound;
53+
private final int hostCount;
54+
private final Object lock = new Object();
55+
private final HostState[] states;
56+
57+
public QwpHostHealthTracker(int hostCount) {
58+
if (hostCount <= 0) {
59+
throw new IllegalArgumentException("hostCount must be > 0");
60+
}
61+
this.hostCount = hostCount;
62+
this.states = new HostState[hostCount];
63+
this.attemptedThisRound = new boolean[hostCount];
64+
for (int i = 0; i < hostCount; i++) {
65+
states[i] = HostState.UNKNOWN;
66+
}
67+
}
68+
69+
/**
70+
* Resets attempted flags. With {@code forgetClassifications}, every host
71+
* except the last-known {@link HostState#HEALTHY} entry is reset to
72+
* {@link HostState#UNKNOWN}; the sticky-Healthy keeps the last successful
73+
* host first in line on the next round.
74+
*/
75+
public void beginRound(boolean forgetClassifications) {
76+
synchronized (lock) {
77+
int stickyIndex = -1;
78+
if (forgetClassifications) {
79+
for (int i = 0; i < hostCount; i++) {
80+
if (states[i] == HostState.HEALTHY) {
81+
stickyIndex = i;
82+
}
83+
}
84+
}
85+
for (int i = 0; i < hostCount; i++) {
86+
attemptedThisRound[i] = false;
87+
if (forgetClassifications && i != stickyIndex) {
88+
states[i] = HostState.UNKNOWN;
89+
}
90+
}
91+
}
92+
}
93+
94+
public int count() {
95+
return hostCount;
96+
}
97+
98+
public HostState getState(int idx) {
99+
synchronized (lock) {
100+
return states[idx];
101+
}
102+
}
103+
104+
public boolean isRoundExhausted() {
105+
synchronized (lock) {
106+
for (int i = 0; i < hostCount; i++) {
107+
if (!attemptedThisRound[i]) {
108+
return false;
109+
}
110+
}
111+
return true;
112+
}
113+
}
114+
115+
/**
116+
* Returns the highest-priority host not yet attempted this round, or -1
117+
* when the round is exhausted.
118+
*/
119+
public int pickNext() {
120+
synchronized (lock) {
121+
for (HostState p : PRIORITY_ORDER) {
122+
for (int i = 0; i < hostCount; i++) {
123+
if (!attemptedThisRound[i] && states[i] == p) {
124+
return i;
125+
}
126+
}
127+
}
128+
return -1;
129+
}
130+
}
131+
132+
/**
133+
* Demotes a previously-healthy host on send/receive failure so a subsequent
134+
* sticky-Healthy reset doesn't preserve it as the priority entry.
135+
*/
136+
public void recordMidStreamFailure(int idx) {
137+
synchronized (lock) {
138+
if (states[idx] == HostState.HEALTHY) {
139+
states[idx] = HostState.TRANSPORT_ERROR;
140+
}
141+
}
142+
}
143+
144+
public void recordRoleReject(int idx, boolean isTransient) {
145+
synchronized (lock) {
146+
states[idx] = isTransient ? HostState.TRANSIENT_REJECT : HostState.TOPOLOGY_REJECT;
147+
attemptedThisRound[idx] = true;
148+
}
149+
}
150+
151+
public void recordSuccess(int idx) {
152+
synchronized (lock) {
153+
states[idx] = HostState.HEALTHY;
154+
attemptedThisRound[idx] = true;
155+
}
156+
}
157+
158+
public void recordTransportError(int idx) {
159+
synchronized (lock) {
160+
states[idx] = HostState.TRANSPORT_ERROR;
161+
attemptedThisRound[idx] = true;
162+
}
163+
}
164+
}

0 commit comments

Comments
 (0)