Skip to content

Commit e8bdb6c

Browse files
HTHouHTHou
committed
Support IPv6 endpoint URLs
1 parent f3e1612 commit e8bdb6c

35 files changed

Lines changed: 882 additions & 133 deletions

File tree

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ public class ClusterConstant {
4444
public static final String DEFAULT_DATA_NODE_COMMON_PROPERTIES =
4545
"DefaultDataNodeCommonProperties";
4646
public static final String DATA_REGION_PER_DATANODE = "integrationTest.dataRegionPerDataNode";
47+
public static final String TEST_NODE_ADDRESS = "IoTDBTestNodeAddress";
4748

4849
// Cluster Configurations
4950
public static final String CLUSTER_CONFIGURATIONS = "ClusterConfigurations";

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java‎

Lines changed: 12 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import org.apache.iotdb.jdbc.IoTDBConnection;
6565
import org.apache.iotdb.rpc.IoTDBConnectionException;
6666
import org.apache.iotdb.rpc.TSStatusCode;
67+
import org.apache.iotdb.rpc.UrlUtils;
6768
import org.apache.iotdb.session.Session;
6869
import org.apache.iotdb.session.TableSessionBuilder;
6970
import org.apache.iotdb.session.pool.SessionPool;
@@ -150,9 +151,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
150151
final String configNodeMetricContent =
151152
getUrlContent(
152153
Config.IOTDB_HTTP_URL_PREFIX
153-
+ configNode.getIp()
154-
+ ":"
155-
+ configNode.getMetricPort()
154+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
155+
configNode.getIp(), configNode.getMetricPort())
156156
+ "/metrics",
157157
authHeader);
158158
result.add(configNodeMetricContent);
@@ -162,9 +162,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
162162
final String dataNodeMetricContent =
163163
getUrlContent(
164164
Config.IOTDB_HTTP_URL_PREFIX
165-
+ dataNode.getIp()
166-
+ ":"
167-
+ dataNode.getMetricPort()
165+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
166+
dataNode.getIp(), dataNode.getMetricPort())
168167
+ "/metrics",
169168
authHeader);
170169
result.add(dataNodeMetricContent);
@@ -1038,7 +1037,8 @@ protected NodeConnection getWriteConnectionWithSpecifiedDataNode(
10381037
final String password,
10391038
final String sqlDialect)
10401039
throws SQLException {
1041-
final String endpoint = dataNode.getIp() + ":" + dataNode.getPort();
1040+
final String endpoint =
1041+
UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(), dataNode.getPort());
10421042
final Connection writeConnection =
10431043
DriverManager.getConnection(
10441044
Config.IOTDB_URL_PREFIX
@@ -1618,18 +1618,14 @@ public void ensureNodeStatus(
16181618
.forEach(
16191619
node ->
16201620
nodeIds.put(
1621-
node.getInternalEndPoint().getIp()
1622-
+ ":"
1623-
+ node.getInternalEndPoint().getPort(),
1621+
UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getInternalEndPoint()),
16241622
node.getConfigNodeId()));
16251623
showClusterResp
16261624
.getDataNodeList()
16271625
.forEach(
16281626
node ->
16291627
nodeIds.put(
1630-
node.getClientRpcEndPoint().getIp()
1631-
+ ":"
1632-
+ node.getClientRpcEndPoint().getPort(),
1628+
UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getClientRpcEndPoint()),
16331629
node.getDataNodeId()));
16341630
for (int j = 0; j < nodes.size(); j++) {
16351631
BaseNodeWrapper nodeWrapper = nodes.get(j);
@@ -1652,10 +1648,9 @@ public void ensureNodeStatus(
16521648
continue;
16531649
}
16541650
if (nodeWrapper instanceof DataNodeWrapper && targetStatus.equals(NodeStatus.Running)) {
1655-
final String[] ipPort = nodeWrapper.getIpAndPortString().split(":");
1656-
final String ip = ipPort[0];
1657-
final int port = Integer.parseInt(ipPort[1]);
1658-
try (TSocket socket = new TSocket(new TConfiguration(), ip, port, 1000)) {
1651+
try (TSocket socket =
1652+
new TSocket(
1653+
new TConfiguration(), nodeWrapper.getIp(), nodeWrapper.getPort(), 1000)) {
16591654
socket.open();
16601655
} catch (final TTransportException e) {
16611656
errorMessages.add(

‎integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.apache.iotdb.it.env.cluster.config.MppJVMConfig;
2828
import org.apache.iotdb.it.framework.IoTDBTestLogger;
2929
import org.apache.iotdb.itbase.env.BaseNodeWrapper;
30+
import org.apache.iotdb.rpc.UrlUtils;
3031

3132
import org.apache.tsfile.external.commons.io.FileUtils;
3233
import org.apache.tsfile.external.commons.io.file.PathUtils;
@@ -123,6 +124,7 @@
123124
import static org.apache.iotdb.it.env.cluster.ClusterConstant.TARGET;
124125
import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_LIB_PATH;
125126
import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_PATH;
127+
import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEST_NODE_ADDRESS;
126128
import static org.apache.iotdb.it.env.cluster.ClusterConstant.TRIGGER_LIB_DIR;
127129
import static org.apache.iotdb.it.env.cluster.ClusterConstant.UDF_LIB_DIR;
128130
import static org.apache.iotdb.it.env.cluster.ClusterConstant.USER_DIR;
@@ -181,7 +183,7 @@ protected AbstractNodeWrapper(
181183
this.testClassName = testClassName;
182184
this.testMethodName = testMethodName;
183185
this.portList = portList;
184-
this.nodeAddress = "127.0.0.1";
186+
this.nodeAddress = System.getProperty(TEST_NODE_ADDRESS, "127.0.0.1");
185187
this.nodePort = portList[0];
186188
this.metricPort = portList[portList.length - 2];
187189
jmxPort = this.portList[portList.length - 1];
@@ -606,7 +608,7 @@ public void setPort(int port) {
606608

607609
@Override
608610
public final String getIpAndPortString() {
609-
return this.getIp() + ":" + this.getPort();
611+
return UrlUtils.formatTEndPointIpv4AndIpv6Url(this.getIp(), this.getPort());
610612
}
611613

612614
protected String workDirFilePath(String dirName, String fileName) {

‎integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java‎

Lines changed: 13 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.apache.iotdb.jdbc.Config;
4343
import org.apache.iotdb.jdbc.Constant;
4444
import org.apache.iotdb.rpc.IoTDBConnectionException;
45+
import org.apache.iotdb.rpc.UrlUtils;
4546
import org.apache.iotdb.session.Session;
4647
import org.apache.iotdb.session.TableSessionBuilder;
4748
import org.apache.iotdb.session.pool.SessionPool;
@@ -72,6 +73,10 @@ public class RemoteServerEnv implements BaseEnv {
7273
private IClientManager<TEndPoint, SyncConfigNodeIServiceClient> clientManager;
7374
private RemoteClusterConfig clusterConfig = new RemoteClusterConfig();
7475

76+
private String remoteEndpointUrl(String port) {
77+
return UrlUtils.formatTEndPointIpv4AndIpv6Url(ip_addr, Integer.parseInt(port));
78+
}
79+
7580
@Override
7681
public void initClusterEnvironment() {
7782
try (Connection connection = EnvFactory.getEnv().getConnection();
@@ -116,11 +121,11 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
116121
List<String> result = new ArrayList<>();
117122
result.add(
118123
getUrlContent(
119-
Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + configNodeMetricPort + "/metrics",
124+
Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(configNodeMetricPort) + "/metrics",
120125
authHeader));
121126
result.add(
122127
getUrlContent(
123-
Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + dataNodeMetricPort + "/metrics",
128+
Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(dataNodeMetricPort) + "/metrics",
124129
authHeader));
125130
return result;
126131
}
@@ -133,7 +138,7 @@ public Connection getConnection(String username, String password, String sqlDial
133138
Class.forName(Config.JDBC_DRIVER_NAME);
134139
connection =
135140
DriverManager.getConnection(
136-
Config.IOTDB_URL_PREFIX + ip_addr + ":" + port,
141+
Config.IOTDB_URL_PREFIX + remoteEndpointUrl(port),
137142
BaseEnv.constructProperties(username, password, sqlDialect));
138143
} catch (ClassNotFoundException e) {
139144
e.printStackTrace();
@@ -170,9 +175,7 @@ public Connection getConnection(
170175
connection =
171176
DriverManager.getConnection(
172177
Config.IOTDB_URL_PREFIX
173-
+ ip_addr
174-
+ ":"
175-
+ port
178+
+ remoteEndpointUrl(port)
176179
+ "?"
177180
+ VERSION
178181
+ "="
@@ -300,15 +303,15 @@ public ISession getSessionConnection(ZoneId zoneId) throws IoTDBConnectionExcept
300303
@Override
301304
public ITableSession getTableSessionConnection() throws IoTDBConnectionException {
302305
return new TableSessionBuilder()
303-
.nodeUrls(Collections.singletonList(ip_addr + ":" + port))
306+
.nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
304307
.build();
305308
}
306309

307310
@Override
308311
public ITableSession getTableSessionConnectionWithDB(String database)
309312
throws IoTDBConnectionException {
310313
return new TableSessionBuilder()
311-
.nodeUrls(Collections.singletonList(ip_addr + ":" + port))
314+
.nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
312315
.database(database)
313316
.build();
314317
}
@@ -332,7 +335,7 @@ public ITableSession getTableSessionConnection(List<String> nodeUrls)
332335
public ITableSession getTableSessionConnection(String userName, String password)
333336
throws IoTDBConnectionException {
334337
return new TableSessionBuilder()
335-
.nodeUrls(Collections.singletonList(ip_addr + ":" + port))
338+
.nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
336339
.username(userName)
337340
.password(password)
338341
.build();
@@ -356,7 +359,7 @@ public ISession getSessionConnection(String userName, String password)
356359
public ISession getSessionConnection(List<String> nodeUrls) throws IoTDBConnectionException {
357360
Session session =
358361
new Session.Builder()
359-
.nodeUrls(Collections.singletonList(ip_addr + ":" + port))
362+
.nodeUrls(Collections.singletonList(remoteEndpointUrl(port)))
360363
.username(SessionConfig.DEFAULT_USER)
361364
.password(SessionConfig.DEFAULT_PASSWORD)
362365
.fetchSize(SessionConfig.DEFAULT_FETCH_SIZE)

‎integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,124 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.db.it;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
23+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
24+
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
25+
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
26+
import org.apache.iotdb.consensus.ConsensusFactory;
27+
import org.apache.iotdb.it.env.EnvFactory;
28+
import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
29+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
30+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
31+
import org.apache.iotdb.it.utils.IPv6TestUtils;
32+
import org.apache.iotdb.itbase.category.ClusterIT;
33+
import org.apache.iotdb.rpc.TSStatusCode;
34+
35+
import org.junit.AfterClass;
36+
import org.junit.BeforeClass;
37+
import org.junit.Test;
38+
import org.junit.experimental.categories.Category;
39+
import org.junit.runner.RunWith;
40+
41+
import java.sql.Connection;
42+
import java.sql.ResultSet;
43+
import java.sql.Statement;
44+
45+
import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS;
46+
import static org.junit.Assert.assertEquals;
47+
import static org.junit.Assert.assertFalse;
48+
import static org.junit.Assert.assertTrue;
49+
50+
@RunWith(IoTDBTestRunner.class)
51+
@Category({ClusterIT.class})
52+
public class IoTDBIPv6ClusterIT {
53+
54+
private static String previousTestNodeAddress;
55+
56+
@BeforeClass
57+
public static void setUp() {
58+
IPv6TestUtils.assumeIPv6LoopbackAvailable();
59+
previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
60+
EnvFactory.getEnv()
61+
.getConfig()
62+
.getCommonConfig()
63+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
64+
.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
65+
.setSchemaReplicationFactor(3)
66+
.setDataReplicationFactor(3);
67+
EnvFactory.getEnv().initClusterEnvironment(1, 3);
68+
}
69+
70+
@AfterClass
71+
public static void tearDown() {
72+
EnvFactory.getEnv().cleanClusterEnvironment();
73+
IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
74+
}
75+
76+
@Test
77+
public void clusterCanCommunicateThroughIPv6Loopback() throws Exception {
78+
for (ConfigNodeWrapper configNode : EnvFactory.getEnv().getConfigNodeWrapperList()) {
79+
assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getIp());
80+
assertTrue(configNode.getIpAndPortString().startsWith("[::1]:"));
81+
}
82+
for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) {
83+
assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
84+
assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:"));
85+
}
86+
87+
try (SyncConfigNodeIServiceClient client =
88+
(SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
89+
final TShowClusterResp showClusterResp = client.showCluster();
90+
final TSStatus status = showClusterResp.getStatus();
91+
assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
92+
assertEquals(1, showClusterResp.getConfigNodeListSize());
93+
assertEquals(3, showClusterResp.getDataNodeListSize());
94+
showClusterResp
95+
.getConfigNodeList()
96+
.forEach(
97+
configNode ->
98+
assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getInternalEndPoint().getIp()));
99+
showClusterResp.getDataNodeList().forEach(this::assertDataNodeLocationUsesIPv6);
100+
}
101+
102+
try (Connection connection = EnvFactory.getEnv().getConnection();
103+
Statement statement = connection.createStatement()) {
104+
statement.execute("CREATE DATABASE root.ipv6_cluster");
105+
statement.execute("CREATE TIMESERIES root.ipv6_cluster.d1.s1 INT64");
106+
statement.execute("INSERT INTO root.ipv6_cluster.d1(time, s1) VALUES (1, 300)");
107+
try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.ipv6_cluster.d1")) {
108+
assertTrue(resultSet.next());
109+
assertEquals(1L, resultSet.getLong(1));
110+
assertEquals(300L, resultSet.getLong(2));
111+
assertFalse(resultSet.next());
112+
}
113+
}
114+
}
115+
116+
private void assertDataNodeLocationUsesIPv6(final TDataNodeLocation dataNodeLocation) {
117+
assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getClientRpcEndPoint().getIp());
118+
assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getInternalEndPoint().getIp());
119+
assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getMPPDataExchangeEndPoint().getIp());
120+
assertEquals(
121+
IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getSchemaRegionConsensusEndPoint().getIp());
122+
assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getDataRegionConsensusEndPoint().getIp());
123+
}
124+
}

‎integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.it.env.EnvFactory;
2626
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
2727
import org.apache.iotdb.itbase.env.BaseEnv;
28+
import org.apache.iotdb.rpc.UrlUtils;
2829

2930
import org.apache.tsfile.utils.Pair;
3031
import org.awaitility.Awaitility;
@@ -283,7 +284,9 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
283284
LOGGER.info("Step 6: Verifying schema consistency on each DataNode independently...");
284285
List<DataNodeWrapper> dataNodeWrappers = EnvFactory.getEnv().getDataNodeWrapperList();
285286
for (DataNodeWrapper wrapper : dataNodeWrappers) {
286-
String nodeDescription = "DataNode " + wrapper.getIp() + ":" + wrapper.getPort();
287+
String nodeDescription =
288+
"DataNode "
289+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(wrapper.getIp(), wrapper.getPort());
287290
LOGGER.info("Verifying schema on {}", nodeDescription);
288291
Awaitility.await()
289292
.atMost(60, TimeUnit.SECONDS)
@@ -307,7 +310,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
307310
LOGGER.info(
308311
"Step 7: Stopping each DataNode in turn and verifying remaining nodes show consistent schema...");
309312
for (DataNodeWrapper stoppedNode : dataNodeWrappers) {
310-
String stoppedDesc = "DataNode " + stoppedNode.getIp() + ":" + stoppedNode.getPort();
313+
String stoppedDesc =
314+
"DataNode "
315+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
316+
stoppedNode.getIp(), stoppedNode.getPort());
311317
LOGGER.info("Stopping {}", stoppedDesc);
312318
stoppedNode.stopForcibly();
313319
Assert.assertFalse(stoppedDesc + " should be stopped", stoppedNode.isAlive());
@@ -318,7 +324,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception {
318324
if (aliveNode == stoppedNode) {
319325
continue;
320326
}
321-
String aliveDesc = "DataNode " + aliveNode.getIp() + ":" + aliveNode.getPort();
327+
String aliveDesc =
328+
"DataNode "
329+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
330+
aliveNode.getIp(), aliveNode.getPort());
322331
Awaitility.await()
323332
.pollDelay(1, TimeUnit.SECONDS)
324333
.atMost(90, TimeUnit.SECONDS)
@@ -446,7 +455,9 @@ private void verifyTimeSeriesAfterDelete(Statement statement, String context) th
446455
protected void waitForReplicationComplete(DataNodeWrapper leaderNode) {
447456
final long timeoutSeconds = 120;
448457
final String metricsUrl =
449-
"http://" + leaderNode.getIp() + ":" + leaderNode.getMetricPort() + "/metrics";
458+
"http://"
459+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(leaderNode.getIp(), leaderNode.getMetricPort())
460+
+ "/metrics";
450461
LOGGER.info(
451462
"Waiting for consensus pipe syncLag to reach 0 on leader DataNode (url: {}, timeout: {}s)...",
452463
metricsUrl,

0 commit comments

Comments
 (0)