Skip to content

Commit 0ef17ba

Browse files
HTHouCaideyipi
HTHou
and
Caideyipi
authored
Support IPv6 endpoint URLs (#18162)
Co-authored-by: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
1 parent dfbe8db commit 0ef17ba

45 files changed

Lines changed: 1532 additions & 145 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎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;
@@ -151,9 +152,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
151152
final String configNodeMetricContent =
152153
getUrlContent(
153154
Config.IOTDB_HTTP_URL_PREFIX
154-
+ configNode.getIp()
155-
+ ":"
156-
+ configNode.getMetricPort()
155+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
156+
configNode.getIp(), configNode.getMetricPort())
157157
+ "/metrics",
158158
authHeader);
159159
result.add(configNodeMetricContent);
@@ -163,9 +163,8 @@ public List<String> getMetricPrometheusReporterContents(String authHeader) {
163163
final String dataNodeMetricContent =
164164
getUrlContent(
165165
Config.IOTDB_HTTP_URL_PREFIX
166-
+ dataNode.getIp()
167-
+ ":"
168-
+ dataNode.getMetricPort()
166+
+ UrlUtils.formatTEndPointIpv4AndIpv6Url(
167+
dataNode.getIp(), dataNode.getMetricPort())
169168
+ "/metrics",
170169
authHeader);
171170
result.add(dataNodeMetricContent);
@@ -1042,7 +1041,8 @@ protected NodeConnection getWriteConnectionWithSpecifiedDataNode(
10421041
final String password,
10431042
final String sqlDialect)
10441043
throws SQLException {
1045-
final String endpoint = dataNode.getIp() + ":" + dataNode.getPort();
1044+
final String endpoint =
1045+
UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(), dataNode.getPort());
10461046
final Connection writeConnection =
10471047
DriverManager.getConnection(
10481048
Config.IOTDB_URL_PREFIX
@@ -1633,18 +1633,14 @@ public void ensureNodeStatus(
16331633
.forEach(
16341634
node ->
16351635
nodeIds.put(
1636-
node.getInternalEndPoint().getIp()
1637-
+ ":"
1638-
+ node.getInternalEndPoint().getPort(),
1636+
UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getInternalEndPoint()),
16391637
node.getConfigNodeId()));
16401638
showClusterResp
16411639
.getDataNodeList()
16421640
.forEach(
16431641
node ->
16441642
nodeIds.put(
1645-
node.getClientRpcEndPoint().getIp()
1646-
+ ":"
1647-
+ node.getClientRpcEndPoint().getPort(),
1643+
UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getClientRpcEndPoint()),
16481644
node.getDataNodeId()));
16491645
for (int j = 0; j < nodes.size(); j++) {
16501646
BaseNodeWrapper nodeWrapper = nodes.get(j);
@@ -1667,10 +1663,9 @@ public void ensureNodeStatus(
16671663
continue;
16681664
}
16691665
if (nodeWrapper instanceof DataNodeWrapper && targetStatus.equals(NodeStatus.Running)) {
1670-
final String[] ipPort = nodeWrapper.getIpAndPortString().split(":");
1671-
final String ip = ipPort[0];
1672-
final int port = Integer.parseInt(ipPort[1]);
1673-
try (TSocket socket = new TSocket(new TConfiguration(), ip, port, 1000)) {
1666+
try (TSocket socket =
1667+
new TSocket(
1668+
new TConfiguration(), nodeWrapper.getIp(), nodeWrapper.getPort(), 1000)) {
16741669
socket.open();
16751670
} catch (final TTransportException e) {
16761671
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+
}

0 commit comments

Comments
 (0)