diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java index 691a13509d9ee..9118f51f436dc 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java @@ -44,6 +44,7 @@ public class ClusterConstant { public static final String DEFAULT_DATA_NODE_COMMON_PROPERTIES = "DefaultDataNodeCommonProperties"; public static final String DATA_REGION_PER_DATANODE = "integrationTest.dataRegionPerDataNode"; + public static final String TEST_NODE_ADDRESS = "IoTDBTestNodeAddress"; // Cluster Configurations public static final String CLUSTER_CONFIGURATIONS = "ClusterConfigurations"; diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java index ae724c7081dbb..3819bd32a8d62 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java @@ -64,6 +64,7 @@ import org.apache.iotdb.jdbc.IoTDBConnection; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.session.Session; import org.apache.iotdb.session.TableSessionBuilder; import org.apache.iotdb.session.pool.SessionPool; @@ -151,9 +152,8 @@ public List getMetricPrometheusReporterContents(String authHeader) { final String configNodeMetricContent = getUrlContent( Config.IOTDB_HTTP_URL_PREFIX - + configNode.getIp() - + ":" - + configNode.getMetricPort() + + UrlUtils.formatTEndPointIpv4AndIpv6Url( + configNode.getIp(), configNode.getMetricPort()) + "/metrics", authHeader); result.add(configNodeMetricContent); @@ -163,9 +163,8 @@ public List getMetricPrometheusReporterContents(String authHeader) { final String dataNodeMetricContent = getUrlContent( Config.IOTDB_HTTP_URL_PREFIX - + dataNode.getIp() - + ":" - + dataNode.getMetricPort() + + UrlUtils.formatTEndPointIpv4AndIpv6Url( + dataNode.getIp(), dataNode.getMetricPort()) + "/metrics", authHeader); result.add(dataNodeMetricContent); @@ -1042,7 +1041,8 @@ protected NodeConnection getWriteConnectionWithSpecifiedDataNode( final String password, final String sqlDialect) throws SQLException { - final String endpoint = dataNode.getIp() + ":" + dataNode.getPort(); + final String endpoint = + UrlUtils.formatTEndPointIpv4AndIpv6Url(dataNode.getIp(), dataNode.getPort()); final Connection writeConnection = DriverManager.getConnection( Config.IOTDB_URL_PREFIX @@ -1633,18 +1633,14 @@ public void ensureNodeStatus( .forEach( node -> nodeIds.put( - node.getInternalEndPoint().getIp() - + ":" - + node.getInternalEndPoint().getPort(), + UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getInternalEndPoint()), node.getConfigNodeId())); showClusterResp .getDataNodeList() .forEach( node -> nodeIds.put( - node.getClientRpcEndPoint().getIp() - + ":" - + node.getClientRpcEndPoint().getPort(), + UrlUtils.convertTEndPointIpv4AndIpv6Url(node.getClientRpcEndPoint()), node.getDataNodeId())); for (int j = 0; j < nodes.size(); j++) { BaseNodeWrapper nodeWrapper = nodes.get(j); @@ -1667,10 +1663,9 @@ public void ensureNodeStatus( continue; } if (nodeWrapper instanceof DataNodeWrapper && targetStatus.equals(NodeStatus.Running)) { - final String[] ipPort = nodeWrapper.getIpAndPortString().split(":"); - final String ip = ipPort[0]; - final int port = Integer.parseInt(ipPort[1]); - try (TSocket socket = new TSocket(new TConfiguration(), ip, port, 1000)) { + try (TSocket socket = + new TSocket( + new TConfiguration(), nodeWrapper.getIp(), nodeWrapper.getPort(), 1000)) { socket.open(); } catch (final TTransportException e) { errorMessages.add( diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java index 5a271c1ead26f..8d3f5f2cc9d9e 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java @@ -27,6 +27,7 @@ import org.apache.iotdb.it.env.cluster.config.MppJVMConfig; import org.apache.iotdb.it.framework.IoTDBTestLogger; import org.apache.iotdb.itbase.env.BaseNodeWrapper; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.external.commons.io.file.PathUtils; @@ -123,6 +124,7 @@ import static org.apache.iotdb.it.env.cluster.ClusterConstant.TARGET; import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_LIB_PATH; import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEMPLATE_NODE_PATH; +import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEST_NODE_ADDRESS; import static org.apache.iotdb.it.env.cluster.ClusterConstant.TRIGGER_LIB_DIR; import static org.apache.iotdb.it.env.cluster.ClusterConstant.UDF_LIB_DIR; import static org.apache.iotdb.it.env.cluster.ClusterConstant.USER_DIR; @@ -181,7 +183,7 @@ protected AbstractNodeWrapper( this.testClassName = testClassName; this.testMethodName = testMethodName; this.portList = portList; - this.nodeAddress = "127.0.0.1"; + this.nodeAddress = System.getProperty(TEST_NODE_ADDRESS, "127.0.0.1"); this.nodePort = portList[0]; this.metricPort = portList[portList.length - 2]; jmxPort = this.portList[portList.length - 1]; @@ -606,7 +608,7 @@ public void setPort(int port) { @Override public final String getIpAndPortString() { - return this.getIp() + ":" + this.getPort(); + return UrlUtils.formatTEndPointIpv4AndIpv6Url(this.getIp(), this.getPort()); } protected String workDirFilePath(String dirName, String fileName) { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java index 586eff60494dc..3790c37dcea1c 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java @@ -42,6 +42,7 @@ import org.apache.iotdb.jdbc.Config; import org.apache.iotdb.jdbc.Constant; import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.session.Session; import org.apache.iotdb.session.TableSessionBuilder; import org.apache.iotdb.session.pool.SessionPool; @@ -72,6 +73,10 @@ public class RemoteServerEnv implements BaseEnv { private IClientManager clientManager; private RemoteClusterConfig clusterConfig = new RemoteClusterConfig(); + private String remoteEndpointUrl(String port) { + return UrlUtils.formatTEndPointIpv4AndIpv6Url(ip_addr, Integer.parseInt(port)); + } + @Override public void initClusterEnvironment() { try (Connection connection = EnvFactory.getEnv().getConnection(); @@ -116,11 +121,11 @@ public List getMetricPrometheusReporterContents(String authHeader) { List result = new ArrayList<>(); result.add( getUrlContent( - Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + configNodeMetricPort + "/metrics", + Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(configNodeMetricPort) + "/metrics", authHeader)); result.add( getUrlContent( - Config.IOTDB_HTTP_URL_PREFIX + ip_addr + ":" + dataNodeMetricPort + "/metrics", + Config.IOTDB_HTTP_URL_PREFIX + remoteEndpointUrl(dataNodeMetricPort) + "/metrics", authHeader)); return result; } @@ -133,7 +138,7 @@ public Connection getConnection(String username, String password, String sqlDial Class.forName(Config.JDBC_DRIVER_NAME); connection = DriverManager.getConnection( - Config.IOTDB_URL_PREFIX + ip_addr + ":" + port, + Config.IOTDB_URL_PREFIX + remoteEndpointUrl(port), BaseEnv.constructProperties(username, password, sqlDialect)); } catch (ClassNotFoundException e) { e.printStackTrace(); @@ -170,9 +175,7 @@ public Connection getConnection( connection = DriverManager.getConnection( Config.IOTDB_URL_PREFIX - + ip_addr - + ":" - + port + + remoteEndpointUrl(port) + "?" + VERSION + "=" @@ -300,7 +303,7 @@ public ISession getSessionConnection(ZoneId zoneId) throws IoTDBConnectionExcept @Override public ITableSession getTableSessionConnection() throws IoTDBConnectionException { return new TableSessionBuilder() - .nodeUrls(Collections.singletonList(ip_addr + ":" + port)) + .nodeUrls(Collections.singletonList(remoteEndpointUrl(port))) .build(); } @@ -308,7 +311,7 @@ public ITableSession getTableSessionConnection() throws IoTDBConnectionException public ITableSession getTableSessionConnectionWithDB(String database) throws IoTDBConnectionException { return new TableSessionBuilder() - .nodeUrls(Collections.singletonList(ip_addr + ":" + port)) + .nodeUrls(Collections.singletonList(remoteEndpointUrl(port))) .database(database) .build(); } @@ -332,7 +335,7 @@ public ITableSession getTableSessionConnection(List nodeUrls) public ITableSession getTableSessionConnection(String userName, String password) throws IoTDBConnectionException { return new TableSessionBuilder() - .nodeUrls(Collections.singletonList(ip_addr + ":" + port)) + .nodeUrls(Collections.singletonList(remoteEndpointUrl(port))) .username(userName) .password(password) .build(); @@ -356,7 +359,7 @@ public ISession getSessionConnection(String userName, String password) public ISession getSessionConnection(List nodeUrls) throws IoTDBConnectionException { Session session = new Session.Builder() - .nodeUrls(Collections.singletonList(ip_addr + ":" + port)) + .nodeUrls(Collections.singletonList(remoteEndpointUrl(port))) .username(SessionConfig.DEFAULT_USER) .password(SessionConfig.DEFAULT_PASSWORD) .fetchSize(SessionConfig.DEFAULT_FETCH_SIZE) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java new file mode 100644 index 0000000000000..c01e89c3b98e9 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBIPv6ClusterIT.java @@ -0,0 +1,124 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.it; + +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.it.utils.IPv6TestUtils; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; + +import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category({ClusterIT.class}) +public class IoTDBIPv6ClusterIT { + + private static String previousTestNodeAddress; + + @BeforeClass + public static void setUp() { + IPv6TestUtils.assumeIPv6LoopbackAvailable(); + previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback(); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaReplicationFactor(3) + .setDataReplicationFactor(3); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress); + } + + @Test + public void clusterCanCommunicateThroughIPv6Loopback() throws Exception { + for (ConfigNodeWrapper configNode : EnvFactory.getEnv().getConfigNodeWrapperList()) { + assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getIp()); + assertTrue(configNode.getIpAndPortString().startsWith("[::1]:")); + } + for (DataNodeWrapper dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) { + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp()); + assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:")); + } + + try (SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + final TShowClusterResp showClusterResp = client.showCluster(); + final TSStatus status = showClusterResp.getStatus(); + assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + assertEquals(1, showClusterResp.getConfigNodeListSize()); + assertEquals(3, showClusterResp.getDataNodeListSize()); + showClusterResp + .getConfigNodeList() + .forEach( + configNode -> + assertEquals(IPV6_LOOPBACK_ADDRESS, configNode.getInternalEndPoint().getIp())); + showClusterResp.getDataNodeList().forEach(this::assertDataNodeLocationUsesIPv6); + } + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.ipv6_cluster"); + statement.execute("CREATE TIMESERIES root.ipv6_cluster.d1.s1 INT64"); + statement.execute("INSERT INTO root.ipv6_cluster.d1(time, s1) VALUES (1, 300)"); + try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.ipv6_cluster.d1")) { + assertTrue(resultSet.next()); + assertEquals(1L, resultSet.getLong(1)); + assertEquals(300L, resultSet.getLong(2)); + assertFalse(resultSet.next()); + } + } + } + + private void assertDataNodeLocationUsesIPv6(final TDataNodeLocation dataNodeLocation) { + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getClientRpcEndPoint().getIp()); + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getInternalEndPoint().getIp()); + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getMPPDataExchangeEndPoint().getIp()); + assertEquals( + IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getSchemaRegionConsensusEndPoint().getIp()); + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNodeLocation.getDataRegionConsensusEndPoint().getIp()); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java b/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java index e07e6e4ebc138..094b5aebb94e7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/iotconsensusv2/IoTDBIoTConsensusV23C3DBasicITBase.java @@ -25,6 +25,7 @@ import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.itbase.env.BaseEnv; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.tsfile.utils.Pair; import org.awaitility.Awaitility; @@ -283,7 +284,9 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception { LOGGER.info("Step 6: Verifying schema consistency on each DataNode independently..."); List dataNodeWrappers = EnvFactory.getEnv().getDataNodeWrapperList(); for (DataNodeWrapper wrapper : dataNodeWrappers) { - String nodeDescription = "DataNode " + wrapper.getIp() + ":" + wrapper.getPort(); + String nodeDescription = + "DataNode " + + UrlUtils.formatTEndPointIpv4AndIpv6Url(wrapper.getIp(), wrapper.getPort()); LOGGER.info("Verifying schema on {}", nodeDescription); Awaitility.await() .atMost(60, TimeUnit.SECONDS) @@ -307,7 +310,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception { LOGGER.info( "Step 7: Stopping each DataNode in turn and verifying remaining nodes show consistent schema..."); for (DataNodeWrapper stoppedNode : dataNodeWrappers) { - String stoppedDesc = "DataNode " + stoppedNode.getIp() + ":" + stoppedNode.getPort(); + String stoppedDesc = + "DataNode " + + UrlUtils.formatTEndPointIpv4AndIpv6Url( + stoppedNode.getIp(), stoppedNode.getPort()); LOGGER.info("Stopping {}", stoppedDesc); stoppedNode.stopForcibly(); Assert.assertFalse(stoppedDesc + " should be stopped", stoppedNode.isAlive()); @@ -318,7 +324,10 @@ public void testDeleteTimeSeriesReplicaConsistency() throws Exception { if (aliveNode == stoppedNode) { continue; } - String aliveDesc = "DataNode " + aliveNode.getIp() + ":" + aliveNode.getPort(); + String aliveDesc = + "DataNode " + + UrlUtils.formatTEndPointIpv4AndIpv6Url( + aliveNode.getIp(), aliveNode.getPort()); Awaitility.await() .pollDelay(1, TimeUnit.SECONDS) .atMost(90, TimeUnit.SECONDS) @@ -446,7 +455,9 @@ private void verifyTimeSeriesAfterDelete(Statement statement, String context) th protected void waitForReplicationComplete(DataNodeWrapper leaderNode) { final long timeoutSeconds = 120; final String metricsUrl = - "http://" + leaderNode.getIp() + ":" + leaderNode.getMetricPort() + "/metrics"; + "http://" + + UrlUtils.formatTEndPointIpv4AndIpv6Url(leaderNode.getIp(), leaderNode.getMetricPort()) + + "/metrics"; LOGGER.info( "Waiting for consensus pipe syncLag to reach 0 on leader DataNode (url: {}, timeout: {}s)...", metricsUrl, diff --git a/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java new file mode 100644 index 0000000000000..aed309c1ccee3 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/it/utils/IPv6TestUtils.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.it.utils; + +import org.junit.Assume; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; + +import static org.apache.iotdb.it.env.cluster.ClusterConstant.TEST_NODE_ADDRESS; + +public final class IPv6TestUtils { + + public static final String IPV6_LOOPBACK_ADDRESS = "::1"; + + private IPv6TestUtils() { + // Utility class + } + + public static void assumeIPv6LoopbackAvailable() { + try (ServerSocket socket = new ServerSocket()) { + socket.bind(new InetSocketAddress(InetAddress.getByName(IPV6_LOOPBACK_ADDRESS), 0)); + } catch (IOException e) { + Assume.assumeNoException("IPv6 loopback is not available", e); + } + } + + public static String setTestNodeAddressToIPv6Loopback() { + final String previousAddress = System.getProperty(TEST_NODE_ADDRESS); + System.setProperty(TEST_NODE_ADDRESS, IPV6_LOOPBACK_ADDRESS); + return previousAddress; + } + + public static void restoreTestNodeAddress(final String previousAddress) { + if (previousAddress == null) { + System.clearProperty(TEST_NODE_ADDRESS); + } else { + System.setProperty(TEST_NODE_ADDRESS, previousAddress); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java index c99eb5d941a16..a5e505f4fd75e 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java @@ -130,7 +130,7 @@ public void testMachineDowntimeSync() { private void testMachineDowntime(String sink) { StringBuilder a = new StringBuilder(); for (DataNodeWrapper nodeWrapper : receiverEnv.getDataNodeWrapperList()) { - a.append(nodeWrapper.getIp()).append(":").append(nodeWrapper.getPort()); + a.append(nodeWrapper.getIpAndPortString()); a.append(","); } a.deleteCharAt(a.length() - 1); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java new file mode 100644 index 0000000000000..716d188c2b137 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.pipe.it.dual.treemodel.auto.basic; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.it.env.MultiEnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.it.utils.IPv6TestUtils; +import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic; +import org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS; + +@RunWith(IoTDBTestRunner.class) +@Category({MultiClusterIT2DualTreeAutoBasic.class}) +public class IoTDBPipeIPv6IT extends AbstractPipeDualTreeModelAutoIT { + + private static String previousTestNodeAddress; + + @BeforeClass + public static void setUpIPv6() { + IPv6TestUtils.assumeIPv6LoopbackAvailable(); + previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback(); + } + + @AfterClass + public static void tearDownIPv6() { + IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress); + } + + @Override + @Before + public void setUp() { + MultiEnvFactory.createEnv(2); + senderEnv = MultiEnvFactory.getEnv(0); + receiverEnv = MultiEnvFactory.getEnv(1); + setupConfig(); + senderEnv.initClusterEnvironment(1, 1); + receiverEnv.initClusterEnvironment(1, 1); + } + + @Override + protected void setupConfig() { + super.setupConfig(); + senderEnv + .getConfig() + .getCommonConfig() + .setDataReplicationFactor(1) + .setSchemaReplicationFactor(1); + receiverEnv + .getConfig() + .getCommonConfig() + .setDataReplicationFactor(1) + .setSchemaReplicationFactor(1); + } + + @Test + public void testSyncAndAsyncSinksThroughIPv6Loopback() throws Exception { + assertDataNodesUseIPv6(senderEnv.getDataNodeWrapperList()); + assertDataNodesUseIPv6(receiverEnv.getDataNodeWrapperList()); + + final String receiverNodeUrl = receiverEnv.getDataNodeWrapper(0).getIpAndPortString(); + Assert.assertTrue(receiverNodeUrl.startsWith("[::1]:")); + + try (final SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + createAndStartPipe( + client, + "ipv6_async_pipe", + "root.ipv6_async.**", + BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(), + receiverNodeUrl); + createAndStartPipe( + client, + "ipv6_sync_pipe", + "root.ipv6_sync.**", + BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(), + receiverNodeUrl); + } + + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList( + "insert into root.ipv6_async.d1(time, s1) values (1, 1)", + "insert into root.ipv6_sync.d1(time, s1) values (1, 1)", + "flush"), + null); + + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select count(s1) from root.ipv6_async.d1", + "count(root.ipv6_async.d1.s1),", + Collections.singleton("1,")); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, + "select count(s1) from root.ipv6_sync.d1", + "count(root.ipv6_sync.d1.s1),", + Collections.singleton("1,")); + } + + private static void createAndStartPipe( + final SyncConfigNodeIServiceClient client, + final String pipeName, + final String sourcePath, + final String sinkName, + final String receiverNodeUrl) + throws Exception { + final Map sourceAttributes = new HashMap<>(); + sourceAttributes.put("source.path", sourcePath); + + final Map sinkAttributes = new HashMap<>(); + sinkAttributes.put("sink", sinkName); + sinkAttributes.put("sink.batch.enable", "false"); + sinkAttributes.put("sink.node-urls", receiverNodeUrl); + + final TSStatus status = + client.createPipe( + new TCreatePipeReq(pipeName, sinkAttributes).setExtractorAttributes(sourceAttributes)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe(pipeName).getCode()); + } + + private static void assertDataNodesUseIPv6(final Iterable dataNodes) { + for (final DataNodeWrapper dataNode : dataNodes) { + Assert.assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp()); + Assert.assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:")); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java index 15ae9fca4c539..529147a77b084 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java @@ -134,7 +134,7 @@ public void testMachineDowntimeSync() { private void testMachineDowntime(String sink) { StringBuilder a = new StringBuilder(); for (DataNodeWrapper nodeWrapper : receiverEnv.getDataNodeWrapperList()) { - a.append(nodeWrapper.getIp()).append(":").append(nodeWrapper.getPort()); + a.append(nodeWrapper.getIpAndPortString()); a.append(","); } a.deleteCharAt(a.length() - 1); diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java new file mode 100644 index 0000000000000..32d0d059ac782 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBIPv6ClientIT.java @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.session.it; + +import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.it.utils.IPv6TestUtils; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.jdbc.Config; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; + +import org.apache.tsfile.read.common.RowRecord; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Collections; + +import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBIPv6ClientIT { + + private static String previousTestNodeAddress; + + @BeforeClass + public static void setUp() { + IPv6TestUtils.assumeIPv6LoopbackAvailable(); + previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback(); + EnvFactory.getEnv().initClusterEnvironment(1, 1); + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress); + } + + @Test + public void clientsCanConnectThroughIPv6Loopback() throws Exception { + final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0); + assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp()); + assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:")); + + try (Connection connection = + DriverManager.getConnection( + Config.IOTDB_URL_PREFIX + dataNode.getIpAndPortString(), + System.getProperty("User", "root"), + System.getProperty("Password", "root")); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.ipv6_client"); + statement.execute("CREATE TIMESERIES root.ipv6_client.d1.s1 INT64"); + statement.execute("INSERT INTO root.ipv6_client.d1(time, s1) VALUES (1, 100)"); + try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.ipv6_client.d1")) { + assertTrue(resultSet.next()); + assertEquals(1L, resultSet.getLong(1)); + assertEquals(100L, resultSet.getLong(2)); + assertFalse(resultSet.next()); + } + } + + try (ISession session = + EnvFactory.getEnv() + .getSessionConnection(Collections.singletonList(dataNode.getIpAndPortString()))) { + try (SessionDataSet dataSet = + session.executeQueryStatement("SELECT s1 FROM root.ipv6_client.d1")) { + Assert.assertTrue(dataSet.hasNext()); + final RowRecord record = dataSet.next(); + assertEquals(1L, record.getTimestamp()); + assertEquals(100L, record.getFields().get(0).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + try (SessionDataSet dataSet = + session.executeQueryStatement("SELECT s1 FROM root.ipv6_client.d1")) { + Assert.assertTrue(dataSet.hasNext()); + final RowRecord record = dataSet.next(); + assertEquals(1L, record.getTimestamp()); + assertEquals(100L, record.getFields().get(0).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + + verifyTableSession(dataNode); + } + + private static void verifyTableSession(final DataNodeWrapper dataNode) + throws IoTDBConnectionException, StatementExecutionException { + try (ITableSession session = + EnvFactory.getEnv() + .getTableSessionConnection(Collections.singletonList(dataNode.getIpAndPortString()))) { + session.executeNonQueryStatement("CREATE DATABASE IF NOT EXISTS ipv6_table"); + session.executeNonQueryStatement("USE ipv6_table"); + session.executeNonQueryStatement("CREATE TABLE table1(tag1 STRING TAG, s1 INT64 FIELD)"); + session.executeNonQueryStatement("INSERT INTO table1(time, tag1, s1) VALUES (1, 'd1', 200)"); + try (SessionDataSet dataSet = + session.executeQueryStatement("SELECT time, s1 FROM table1 WHERE tag1 = 'd1'")) { + Assert.assertTrue(dataSet.hasNext()); + final RowRecord record = dataSet.next(); + assertEquals(1L, record.getFields().get(0).getLongV()); + assertEquals(200L, record.getFields().get(1).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + } +} diff --git a/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp b/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp index 604099a82d1bf..55fd13a08f46a 100644 --- a/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp +++ b/iotdb-client/client-cpp/src/rpc/NodesSupplier.cpp @@ -17,6 +17,7 @@ * under the License. */ #include "NodesSupplier.h" +#include "RpcCommon.h" #include "Session.h" #include "SessionDataSet.h" #include @@ -188,7 +189,7 @@ std::vector NodesSupplier::fetchLatestEndpoints() { port = record->fields.at(columnPortIdx).intV.value(); } - if (ip == "0.0.0.0") { + if (UrlUtils::isWildcardAddress(ip)) { log_warn("Skipping invalid node: " + ip + ":" + std::to_string(port)); continue; } @@ -230,4 +231,4 @@ void NodesSupplier::stopBackgroundRefresh() noexcept { refreshThread_.join(); } } -} \ No newline at end of file +} diff --git a/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp b/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp index c83b39a5fb58d..3011a81de724e 100644 --- a/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp +++ b/iotdb-client/client-cpp/src/rpc/RpcCommon.cpp @@ -212,3 +212,27 @@ TEndPoint UrlUtils::parseTEndPointIpv4AndIpv6Url(const string& endPointUrl) { return endPoint; } + +bool UrlUtils::isWildcardAddress(const string& host) { + if (host.empty()) { + return false; + } + + string normalizedHost = host; + if (normalizedHost.size() >= 2 && normalizedHost.front() == '[' && normalizedHost.back() == ']') { + normalizedHost = normalizedHost.substr(1, normalizedHost.size() - 2); + } + + if (normalizedHost == "0.0.0.0") { + return true; + } + if (normalizedHost.find(PORT_SEPARATOR) == string::npos) { + return false; + } + for (char ch : normalizedHost) { + if (ch != ':' && ch != '0') { + return false; + } + } + return true; +} diff --git a/iotdb-client/client-cpp/src/rpc/RpcCommon.h b/iotdb-client/client-cpp/src/rpc/RpcCommon.h index 721150bd6cadc..48bff46ffe6e6 100644 --- a/iotdb-client/client-cpp/src/rpc/RpcCommon.h +++ b/iotdb-client/client-cpp/src/rpc/RpcCommon.h @@ -76,6 +76,8 @@ class UrlUtils { public: static TEndPoint parseTEndPointIpv4AndIpv6Url(const std::string& endPointUrl); + + static bool isWildcardAddress(const std::string& host); }; #endif diff --git a/iotdb-client/client-cpp/src/session/Session.cpp b/iotdb-client/client-cpp/src/session/Session.cpp index 7dd0a7813ed50..cf429af0c4d4e 100644 --- a/iotdb-client/client-cpp/src/session/Session.cpp +++ b/iotdb-client/client-cpp/src/session/Session.cpp @@ -2096,7 +2096,7 @@ void Session::Impl::handleQueryRedirection(TEndPoint endPoint) { void Session::Impl::handleRedirection(const std::string& deviceId, TEndPoint endPoint) { if (!enableRedirection_) return; - if (endPoint.ip == "0.0.0.0") + if (UrlUtils::isWildcardAddress(endPoint.ip)) return; getDefaultSessionConnection(); deviceIdToEndpoint[deviceId] = endPoint; @@ -2122,7 +2122,7 @@ void Session::Impl::handleRedirection(const std::shared_ptr& TEndPoint endPoint) { if (!enableRedirection_) return; - if (endPoint.ip == "0.0.0.0") + if (UrlUtils::isWildcardAddress(endPoint.ip)) return; getDefaultSessionConnection(); tableModelDeviceIdToEndpoint[deviceId] = endPoint; diff --git a/iotdb-client/client-py/iotdb/Session.py b/iotdb-client/client-py/iotdb/Session.py index 0a5cb43fd5440..ae4a9c2cc983b 100644 --- a/iotdb-client/client-py/iotdb/Session.py +++ b/iotdb-client/client-py/iotdb/Session.py @@ -16,48 +16,50 @@ # under the License. # +import ipaddress import logging import random -import sys import struct +import sys import warnings + +from iotdb.utils.SessionDataSet import SessionDataSet from thrift.protocol import TBinaryProtocol, TCompactProtocol from thrift.transport import TSocket, TTransport from tzlocal import get_localzone_name -from iotdb.utils.SessionDataSet import SessionDataSet from .template.Template import Template from .template.TemplateQueryType import TemplateQueryType from .thrift.common.ttypes import TEndPoint from .thrift.rpc.IClientRPCService import ( Client, - TSCreateTimeseriesReq, + TSAppendSchemaTemplateReq, + TSCloseSessionReq, TSCreateAlignedTimeseriesReq, + TSCreateMultiTimeseriesReq, + TSCreateSchemaTemplateReq, + TSCreateTimeseriesReq, + TSDropSchemaTemplateReq, + TSExecuteStatementReq, TSInsertRecordReq, + TSInsertRecordsOfOneDeviceReq, + TSInsertRecordsReq, TSInsertStringRecordReq, TSInsertTabletReq, - TSExecuteStatementReq, - TSOpenSessionReq, - TSCreateMultiTimeseriesReq, - TSCloseSessionReq, TSInsertTabletsReq, - TSInsertRecordsReq, - TSInsertRecordsOfOneDeviceReq, - TSCreateSchemaTemplateReq, - TSDropSchemaTemplateReq, - TSAppendSchemaTemplateReq, + TSOpenSessionReq, TSPruneSchemaTemplateReq, + TSQueryTemplateReq, TSSetSchemaTemplateReq, TSUnsetSchemaTemplateReq, - TSQueryTemplateReq, ) from .thrift.rpc.ttypes import ( TSDeleteDataReq, + TSInsertStringRecordsOfOneDeviceReq, + TSLastDataQueryReq, TSProtocolVersion, - TSSetTimeZoneReq, TSRawDataQueryReq, - TSLastDataQueryReq, - TSInsertStringRecordsOfOneDeviceReq, + TSSetTimeZoneReq, ) from .tsfile.utils.date_utils import parse_date_to_int from .utils import rpc_utils @@ -67,6 +69,44 @@ warnings.simplefilter("always", DeprecationWarning) +def _parse_endpoint_url(endpoint_url): + if endpoint_url.startswith("["): + end_index = endpoint_url.find("]") + if end_index <= 1 or end_index + 1 >= len(endpoint_url): + raise RuntimeError("invalid node url: {}".format(endpoint_url)) + if endpoint_url[end_index + 1] != ":": + raise RuntimeError("invalid node url: {}".format(endpoint_url)) + return endpoint_url[1:end_index], _parse_endpoint_port( + endpoint_url, endpoint_url[end_index + 2 :] + ) + + if "[" in endpoint_url or "]" in endpoint_url: + raise RuntimeError("invalid node url: {}".format(endpoint_url)) + split = endpoint_url.rsplit(":", 1) + if len(split) != 2: + raise RuntimeError("invalid node url: {}".format(endpoint_url)) + return split[0], _parse_endpoint_port(endpoint_url, split[1]) + + +def _parse_endpoint_port(endpoint_url, port): + try: + return int(port) + except ValueError as e: + raise RuntimeError("invalid node url: {}".format(endpoint_url)) from e + + +def _is_wildcard_address(host): + if host is None: + return False + normalized_host = host + if normalized_host.startswith("[") and normalized_host.endswith("]"): + normalized_host = normalized_host[1:-1] + try: + return ipaddress.ip_address(normalized_host).is_unspecified + except ValueError: + return False + + class Session(object): DEFAULT_FETCH_SIZE = 5000 DEFAULT_USER = "root" @@ -158,9 +198,9 @@ def init_from_node_urls( session.__hosts = [] session.__ports = [] for node_url in node_urls: - split = node_url.split(":") - session.__hosts.append(split[0]) - session.__ports.append(int(split[1])) + host, port = _parse_endpoint_url(node_url) + session.__hosts.append(host) + session.__ports.append(port) session.__host = session.__hosts[0] session.__port = session.__ports[0] session.__default_endpoint = TEndPoint(session.__host, session.__port) @@ -259,6 +299,7 @@ def init_connection(self, endpoint): def __get_transport(self, endpoint): if self.__use_ssl: import ssl + from thrift.transport import TSSLSocket if sys.version_info >= (3, 10): @@ -1986,7 +2027,7 @@ def get_connection(self, device_id): def handle_redirection(self, device_id, endpoint: TEndPoint): if self.__enable_redirection: - if endpoint.ip == "0.0.0.0": + if _is_wildcard_address(endpoint.ip): return 0 if ( device_id not in self.__device_id_to_endpoint diff --git a/iotdb-client/client-py/tests/unit/test_session_node_url.py b/iotdb-client/client-py/tests/unit/test_session_node_url.py new file mode 100644 index 0000000000000..3967906198be1 --- /dev/null +++ b/iotdb-client/client-py/tests/unit/test_session_node_url.py @@ -0,0 +1,82 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +import pytest +from iotdb.Session import Session, _is_wildcard_address + + +def test_init_from_node_urls_with_bracketed_ipv6(): + session = Session.init_from_node_urls(["[::1]:6667"]) + + assert session._Session__hosts == ["::1"] + assert session._Session__ports == [6667] + + +def test_init_from_node_urls_with_legacy_ipv6(): + session = Session.init_from_node_urls(["::1:6667"]) + + assert session._Session__hosts == ["::1"] + assert session._Session__ports == [6667] + + +def test_init_from_node_urls_with_hostname(): + session = Session.init_from_node_urls(["localhost:6667"]) + + assert session._Session__hosts == ["localhost"] + assert session._Session__ports == [6667] + + +@pytest.mark.parametrize( + "node_url", + [ + "[::1]6667", + "[::1]:", + "[::1:6667", + ], +) +def test_init_from_node_urls_rejects_malformed_bracketed_ipv6(node_url): + with pytest.raises(RuntimeError): + Session.init_from_node_urls([node_url]) + + +@pytest.mark.parametrize( + "host", + [ + "0.0.0.0", + "::", + "[::]", + "0:0:0:0:0:0:0:0", + "0::0", + ], +) +def test_wildcard_address(host): + assert _is_wildcard_address(host) + + +@pytest.mark.parametrize( + "host", + [ + None, + "127.0.0.1", + "localhost", + "::1", + "[::1]", + ], +) +def test_not_wildcard_address(host): + assert not _is_wildcard_address(host) diff --git a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java index 412093f7386e8..49aa0860a39e7 100644 --- a/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java +++ b/iotdb-client/jdbc/src/main/java/org/apache/iotdb/jdbc/Utils.java @@ -60,41 +60,58 @@ static IoTDBConnectionParams parseUrl(String url, Properties info) throws IoTDBU String suffixURL = null; if (url.startsWith(Config.IOTDB_URL_PREFIX)) { String subURL = url.substring(Config.IOTDB_URL_PREFIX.length()); - int i = subURL.lastIndexOf(COLON); - host = subURL.substring(0, i); - params.setHost(host); - i++; - // parse port - int port = 0; - for (; i < subURL.length() && Character.isDigit(subURL.charAt(i)); i++) { - port = port * 10 + (subURL.charAt(i) - '0'); + int i = -1; + if (subURL.startsWith("[")) { + int endIndex = subURL.indexOf("]"); + if (endIndex > 1 + && endIndex + 1 < subURL.length() + && COLON.equals(subURL.substring(endIndex + 1, endIndex + 2))) { + host = subURL.substring(1, endIndex); + i = endIndex + 2; + } + } else if (!subURL.contains("[") && !subURL.contains("]")) { + i = subURL.lastIndexOf(COLON); + if (i > 0) { + host = subURL.substring(0, i); + i++; + } } - suffixURL = i < subURL.length() ? subURL.substring(i) : ""; - // legal port - if (port >= 1 && port <= 65535) { - params.setPort(port); - - // parse database - if (i < subURL.length() && subURL.charAt(i) == SLASH) { - int endIndex = subURL.indexOf(PARAMETER_SEPARATOR, i + 1); - String database; - if (endIndex <= i + 1) { - if (i + 1 == subURL.length()) { - database = null; + + if (i > 0) { + params.setHost(host); + // parse port + int port = 0; + int portStartIndex = i; + for (; i < subURL.length() && Character.isDigit(subURL.charAt(i)); i++) { + port = port * 10 + (subURL.charAt(i) - '0'); + } + suffixURL = i < subURL.length() ? subURL.substring(i) : ""; + // legal port + if (i > portStartIndex && port >= 1 && port <= 65535) { + params.setPort(port); + + // parse database + if (i < subURL.length() && subURL.charAt(i) == SLASH) { + int endIndex = subURL.indexOf(PARAMETER_SEPARATOR, i + 1); + String database; + if (endIndex <= i + 1) { + if (i + 1 == subURL.length()) { + database = null; + } else { + database = subURL.substring(i + 1); + } + suffixURL = ""; } else { - database = subURL.substring(i + 1); + database = subURL.substring(i + 1, endIndex); + suffixURL = subURL.substring(endIndex); } - suffixURL = ""; - } else { - database = subURL.substring(i + 1, endIndex); - suffixURL = subURL.substring(endIndex); + params.setDb(database); } - params.setDb(database); - } - matcher = SUFFIX_URL_PATTERN.matcher(suffixURL); - if (matcher.matches() && parseUrlParam(subURL, info)) { - isUrlLegal = true; + matcher = SUFFIX_URL_PATTERN.matcher(suffixURL); + if (matcher.matches() && parseUrlParam(subURL, info)) { + isUrlLegal = true; + } } } } diff --git a/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java b/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java index a84010ad4ecaa..cd350a539cebb 100644 --- a/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java +++ b/iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/UtilsTest.java @@ -27,6 +27,7 @@ import org.junit.Before; import org.junit.Test; +import java.util.Optional; import java.util.Properties; import static org.junit.Assert.assertEquals; @@ -90,6 +91,38 @@ public void testParseIPV6URL() throws IoTDBURLException { assertEquals(params.getPassword(), userPwd); } + @Test + public void testParseBracketedIPV6URL() throws IoTDBURLException { + String host = "AD80:E32B:CA25:B3AE:DA4A:DAAF:EEAE:BBBE"; + int port = 6667; + Properties properties = new Properties(); + + IoTDBConnectionParams params = + Utils.parseUrl(String.format(Config.IOTDB_URL_PREFIX + "[%s]:%s/", host, port), properties); + assertEquals(host, params.getHost()); + assertEquals(port, params.getPort()); + + params = + Utils.parseUrl(String.format(Config.IOTDB_URL_PREFIX + "[%s]:%s", host, port), properties); + assertEquals(host, params.getHost()); + assertEquals(port, params.getPort()); + } + + @Test(expected = IoTDBURLException.class) + public void testParseBracketedIPV6URLWithoutPortSeparator() throws IoTDBURLException { + Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1]6667", new Properties()); + } + + @Test(expected = IoTDBURLException.class) + public void testParseBracketedIPV6URLWithoutPort() throws IoTDBURLException { + Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1]:", new Properties()); + } + + @Test(expected = IoTDBURLException.class) + public void testParseBracketedIPV6URLWithoutEndMark() throws IoTDBURLException { + Utils.parseUrl(Config.IOTDB_URL_PREFIX + "[::1:6667", new Properties()); + } + @Test(expected = IoTDBURLException.class) public void testParseWrongUrl1() throws IoTDBURLException { Properties properties = new Properties(); @@ -198,4 +231,17 @@ public void testParseSslConfigFromUrl() throws IoTDBURLException { assertEquals("/tmp/url-keystore.p12", params.getKeyStore()); assertEquals("url_key_pass", params.getKeyStorePwd()); } + + @Test + public void testParseSslConfigFromBracketedIpv6Url() throws IoTDBURLException { + IoTDBConnectionParams params = + Utils.parseUrl( + "jdbc:iotdb://[::1]:6667/ipv6_db?use_ssl=true&sql_dialect=table", new Properties()); + + assertEquals("::1", params.getHost()); + assertEquals(6667, params.getPort()); + assertEquals(Optional.of("ipv6_db"), params.getDb()); + assertEquals(Constant.TABLE_DIALECT, params.getSqlDialect()); + assertTrue(params.isUseSSL()); + } } diff --git a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java index 39b2390a7368c..621ef21c81cf8 100644 --- a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java +++ b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/UrlUtils.java @@ -21,10 +21,14 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import java.net.InetAddress; +import java.net.UnknownHostException; + /** The UrlUtils */ public class UrlUtils { private static final String PORT_SEPARATOR = ":"; - private static final String ABB_COLON = "["; + private static final String IPV6_BEGIN_MARK = "["; + private static final String IPV6_END_MARK = "]"; private UrlUtils() {} @@ -32,22 +36,80 @@ private UrlUtils() {} * Parse TEndPoint from a given TEndPointUrl * example:[D80:0000:0000:0000:ABAA:0000:00C2:0002]:22227 * - * @param endPointUrl ip:port - * @return TEndPoint null if parse error + * @param endPointUrl host:port or [ipv6]:port + * @return parsed TEndPoint + * @throws NumberFormatException if the bracketed endpoint format is invalid or the port is not a + * number */ public static TEndPoint parseTEndPointIpv4AndIpv6Url(String endPointUrl) { TEndPoint endPoint = new TEndPoint(); + if (endPointUrl.startsWith(IPV6_BEGIN_MARK)) { + int endIndex = endPointUrl.indexOf(IPV6_END_MARK); + if (endIndex <= 1 + || endIndex + 1 >= endPointUrl.length() + || !PORT_SEPARATOR.equals(endPointUrl.substring(endIndex + 1, endIndex + 2))) { + throw new NumberFormatException(); + } + endPoint.setIp(endPointUrl.substring(1, endIndex)); + endPoint.setPort(Integer.parseInt(endPointUrl.substring(endIndex + 2))); + return endPoint; + } + if (endPointUrl.contains(IPV6_BEGIN_MARK) || endPointUrl.contains(IPV6_END_MARK)) { + throw new NumberFormatException(); + } if (endPointUrl.contains(PORT_SEPARATOR)) { int point_position = endPointUrl.lastIndexOf(PORT_SEPARATOR); String port = endPointUrl.substring(endPointUrl.lastIndexOf(PORT_SEPARATOR) + 1); String ip = endPointUrl.substring(0, point_position); - // If the ip/host part is provided as IPv6 address, cut off the surrounding square brackets. - if (ip.contains(ABB_COLON)) { - ip = ip.substring(1, ip.length() - 1); - } endPoint.setIp(ip); endPoint.setPort(Integer.parseInt(port)); } return endPoint; } + + /** + * Convert TEndPoint to a URL string. IPv6 literals are surrounded by square brackets so the last + * colon remains an unambiguous port separator. + */ + public static String convertTEndPointIpv4AndIpv6Url(TEndPoint endPoint) { + return formatTEndPointIpv4AndIpv6Url(endPoint.getIp(), endPoint.getPort()); + } + + /** Format host and port as host:port or [ipv6]:port. This method expects host without a port. */ + public static String formatTEndPointIpv4AndIpv6Url(String host, int port) { + String formattedHost = host; + if (isIpv6Literal(host) && !isBracketedIpv6Literal(host)) { + formattedHost = IPV6_BEGIN_MARK + host + IPV6_END_MARK; + } + return formattedHost + PORT_SEPARATOR + port; + } + + public static boolean isWildcardAddress(String host) { + if (host == null) { + return false; + } + String normalizedHost = host; + if (isBracketedIpv6Literal(host)) { + normalizedHost = host.substring(1, host.length() - 1); + } + if ("0.0.0.0".equals(normalizedHost)) { + return true; + } + if (!isIpv6Literal(normalizedHost)) { + return false; + } + try { + return InetAddress.getByName(normalizedHost).isAnyLocalAddress(); + } catch (UnknownHostException e) { + return false; + } + } + + private static boolean isIpv6Literal(String host) { + return host.contains(PORT_SEPARATOR); + } + + private static boolean isBracketedIpv6Literal(String host) { + return host.startsWith(IPV6_BEGIN_MARK) && host.endsWith(IPV6_END_MARK); + } } diff --git a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java index 1d206973d1333..e754df4286e6b 100644 --- a/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java +++ b/iotdb-client/service-rpc/src/test/java/org/apache/iotdb/rpc/UrlUtilsTest.java @@ -24,6 +24,8 @@ import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; public class UrlUtilsTest { @@ -51,6 +53,21 @@ public void testParseIPV6AbbURL() { assertEquals(endPoint.getPort(), 22227); } + @Test(expected = NumberFormatException.class) + public void testParseBracketedIPV6URLWithoutPortSeparator() { + UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0]22227"); + } + + @Test(expected = NumberFormatException.class) + public void testParseBracketedIPV6URLWithoutPort() { + UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0]:"); + } + + @Test(expected = NumberFormatException.class) + public void testParseBracketedIPV6URLWithoutEndMark() { + UrlUtils.parseTEndPointIpv4AndIpv6Url("[D80::ABAA:0:22227"); + } + @Test public void testParseIPV4URL() { String hostAndPoint = "192.0.0.1:22227"; @@ -58,4 +75,52 @@ public void testParseIPV4URL() { assertEquals(endPoint.getIp(), "192.0.0.1"); assertEquals(endPoint.getPort(), 22227); } + + @Test + public void testConvertIPV4URL() { + assertEquals( + "192.0.0.1:22227", + UrlUtils.convertTEndPointIpv4AndIpv6Url(new TEndPoint("192.0.0.1", 22227))); + } + + @Test + public void testConvertHostNameURL() { + assertEquals("localhost:22227", UrlUtils.formatTEndPointIpv4AndIpv6Url("localhost", 22227)); + } + + @Test + public void testConvertIPV6URL() { + assertEquals( + "[D80:0000:0000:0000:ABAA:00BB:EEAA:BBDD]:22227", + UrlUtils.convertTEndPointIpv4AndIpv6Url( + new TEndPoint("D80:0000:0000:0000:ABAA:00BB:EEAA:BBDD", 22227))); + } + + @Test + public void testConvertIPV6AbbURL() { + assertEquals( + "[D80::ABAA:0]:22227", + UrlUtils.convertTEndPointIpv4AndIpv6Url(new TEndPoint("D80::ABAA:0", 22227))); + } + + @Test + public void testConvertBracketedIPV6URL() { + assertEquals( + "[D80::ABAA:0]:22227", UrlUtils.formatTEndPointIpv4AndIpv6Url("[D80::ABAA:0]", 22227)); + } + + @Test + public void testWildcardAddress() { + assertTrue(UrlUtils.isWildcardAddress("0.0.0.0")); + assertTrue(UrlUtils.isWildcardAddress("::")); + assertTrue(UrlUtils.isWildcardAddress("[::]")); + assertTrue(UrlUtils.isWildcardAddress("0:0:0:0:0:0:0:0")); + assertTrue(UrlUtils.isWildcardAddress("0::0")); + + assertFalse(UrlUtils.isWildcardAddress(null)); + assertFalse(UrlUtils.isWildcardAddress("127.0.0.1")); + assertFalse(UrlUtils.isWildcardAddress("localhost")); + assertFalse(UrlUtils.isWildcardAddress("::1")); + assertFalse(UrlUtils.isWildcardAddress("[::1]")); + } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java index c1f43fa30d433..81d83450f72e6 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/NodesSupplier.java @@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.session.i18n.SessionMessages; import org.slf4j.Logger; @@ -257,8 +258,8 @@ private void updateAvailableNodes(SessionDataSet sessionDataSet) throws Exceptio List res = new ArrayList<>(); while (iterator.next()) { String ip = iterator.getString(IP_COLUMN_NAME); - // ignore 0.0.0.0 - if (!"0.0.0.0".equals(ip)) { + // ignore wildcard addresses + if (!UrlUtils.isWildcardAddress(ip)) { String port = iterator.getString(PORT_COLUMN_NAME); if (ip != null && port != null) { res.add(new TEndPoint(ip, Integer.parseInt(port))); diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index 6819325a9b27b..36581b9690cd8 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -32,6 +32,7 @@ import org.apache.iotdb.rpc.NoValidValueException; import org.apache.iotdb.rpc.RedirectException; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.service.rpc.thrift.TCreateTimeseriesUsingSchemaTemplateReq; import org.apache.iotdb.service.rpc.thrift.TSAppendSchemaTemplateReq; import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp; @@ -1414,7 +1415,7 @@ public void removeBrokenSessionConnection(SessionConnection sessionConnection) { private void handleRedirection(String deviceId, TEndPoint endpoint) { if (enableRedirection) { // no need to redirection - if (endpoint.ip.equals("0.0.0.0")) { + if (UrlUtils.isWildcardAddress(endpoint.ip)) { return; } if (!deviceIdToEndpoint.containsKey(deviceId) @@ -1440,7 +1441,7 @@ private void handleRedirection(String deviceId, TEndPoint endpoint) { private void handleRedirection(IDeviceID deviceId, TEndPoint endpoint) { if (enableRedirection) { // no need to redirection - if (endpoint.ip.equals("0.0.0.0")) { + if (UrlUtils.isWildcardAddress(endpoint.ip)) { return; } if (!tableModelDeviceIdToEndpoint.containsKey(deviceId) diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java index ad6699aaf87f3..b498784ad2c79 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java @@ -30,6 +30,7 @@ import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.service.rpc.thrift.IClientRPCService; import org.apache.iotdb.service.rpc.thrift.TCreateTimeseriesUsingSchemaTemplateReq; import org.apache.iotdb.service.rpc.thrift.TSAggregationQueryReq; @@ -83,7 +84,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.StringJoiner; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -1314,14 +1314,11 @@ private String logForReconnectionFailure() { if (endPointList == null) { return MSG_RECONNECTION_FAIL; } - StringJoiner urls = new StringJoiner(","); + List urls = new ArrayList<>(); for (TEndPoint end : endPointList) { - StringJoiner url = new StringJoiner(":"); - url.add(end.getIp()); - url.add(String.valueOf(end.getPort())); - urls.add(url.toString()); + urls.add(UrlUtils.convertTEndPointIpv4AndIpv6Url(end)); } - return MSG_RECONNECTION_FAIL.concat(urls.toString()); + return MSG_RECONNECTION_FAIL.concat(String.join(",", urls)); } @Override diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java index 07c0bc4d7e679..b3de497df3980 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java @@ -32,6 +32,7 @@ import org.apache.iotdb.isession.util.Version; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.service.rpc.thrift.TSBackupConfigurationResp; import org.apache.iotdb.service.rpc.thrift.TSConnectionInfoResp; import org.apache.iotdb.session.DummyNodesSupplier; @@ -426,7 +427,7 @@ public SessionPool( this.version = version; this.thriftDefaultBufferSize = thriftDefaultBufferSize; this.thriftMaxFrameSize = thriftMaxFrameSize; - this.formattedNodeUrls = String.format("%s:%s", host, port); + this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host, port); initThreadPool(); initAvailableNodes(Collections.singletonList(new TEndPoint(host, port))); } @@ -469,7 +470,7 @@ public SessionPool( this.version = version; this.thriftDefaultBufferSize = thriftDefaultBufferSize; this.thriftMaxFrameSize = thriftMaxFrameSize; - this.formattedNodeUrls = String.format("%s:%s", host, port); + this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host, port); this.useSSL = useSSL; this.trustStore = trustStore; this.trustStorePwd = trustStorePwd; @@ -574,7 +575,7 @@ public SessionPool(AbstractSessionPoolBuilder builder) { this.host = builder.host; this.port = builder.rpcPort; this.nodeUrls = null; - this.formattedNodeUrls = String.format("%s:%s", host, port); + this.formattedNodeUrls = UrlUtils.formatTEndPointIpv4AndIpv6Url(host, port); if (enableAutoFetch) { initAvailableNodes(Collections.singletonList(new TEndPoint(host, port))); } else { diff --git a/iotdb-core/ainode/iotdb/ainode/core/config.py b/iotdb-core/ainode/iotdb/ainode/core/config.py index 4995dda7bf337..53217f6c27b5b 100644 --- a/iotdb-core/ainode/iotdb/ainode/core/config.py +++ b/iotdb-core/ainode/iotdb/ainode/core/config.py @@ -466,13 +466,24 @@ def load_properties(filepath, sep="=", comment_char="#"): def parse_endpoint_url(endpoint_url: str) -> TEndPoint: """Parse TEndPoint from a given endpoint url. Args: - endpoint_url: an endpoint url, format: ip:port + endpoint_url: an endpoint url, format: host:port or [ipv6]:port Returns: TEndPoint Raises: - BadNodeUrlError + BadNodeUrlException """ - split = endpoint_url.split(":") + if endpoint_url.startswith("["): + end_index = endpoint_url.find("]") + if end_index <= 1 or end_index + 1 >= len(endpoint_url): + raise BadNodeUrlException(endpoint_url) + if endpoint_url[end_index + 1] != ":": + raise BadNodeUrlException(endpoint_url) + split = [endpoint_url[1:end_index], endpoint_url[end_index + 2 :]] + else: + if "[" in endpoint_url or "]" in endpoint_url: + raise BadNodeUrlException(endpoint_url) + split = endpoint_url.rsplit(":", 1) + if len(split) != 2: raise BadNodeUrlException(endpoint_url) diff --git a/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py b/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py index be1dd9bf1c2b3..bfe4f56d5185e 100644 --- a/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py +++ b/iotdb-core/ainode/iotdb/ainode/core/ingress/iotdb.py @@ -50,6 +50,13 @@ def _cache_enable() -> bool: return AINodeDescriptor().get_config().get_ain_data_storage_cache_size() > 0 +def _format_endpoint_url(ip: str, port: int) -> str: + formatted_ip = ip + if ":" in ip and not (ip.startswith("[") and ip.endswith("]")): + formatted_ip = f"[{ip}]" + return f"{formatted_ip}:{port}" + + class IoTDBTreeModelDataset(BasicDatabaseForecastDataset): cache = MemoryLRUCache() @@ -84,7 +91,7 @@ def __init__( self.TIME_CONDITION = " where time>=%s and time<%s" self.session = Session.init_from_node_urls( - node_urls=[f"{ip}:{port}"], + node_urls=[_format_endpoint_url(ip, port)], user=username, password=password, use_ssl=AINodeDescriptor() @@ -262,7 +269,7 @@ def __init__( ) table_session_config = TableSessionConfig( - node_urls=[f"{ip}:{port}"], + node_urls=[_format_endpoint_url(ip, port)], username=username, password=password, use_ssl=AINodeDescriptor() diff --git a/iotdb-core/ainode/resources/conf/iotdb-ainode.properties b/iotdb-core/ainode/resources/conf/iotdb-ainode.properties index 5b653d678a684..6a8f02a62f74c 100644 --- a/iotdb-core/ainode/resources/conf/iotdb-ainode.properties +++ b/iotdb-core/ainode/resources/conf/iotdb-ainode.properties @@ -23,11 +23,12 @@ cluster_name=defaultCluster # ConfigNode address registered at AINode startup. # Allow modifications only before starting the service for the first time. +# Format: address:port or [ipv6]:port, e.g. 127.0.0.1:10710 or [::1]:10710 # Datatype: String ain_seed_config_node=127.0.0.1:10710 # Used for connection of DataNode/ConfigNode clients -# Could set 127.0.0.1(for local test) or ipv4 address +# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname. # Datatype: String ain_rpc_address=127.0.0.1 diff --git a/iotdb-core/ainode/tests/test_config.py b/iotdb-core/ainode/tests/test_config.py new file mode 100644 index 0000000000000..52955c511378f --- /dev/null +++ b/iotdb-core/ainode/tests/test_config.py @@ -0,0 +1,71 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +import sys +import types +import unittest +from importlib import import_module + + +class TEndPoint: + def __init__(self, ip=None, port=None): + self.ip = ip + self.port = port + + +thrift_module = types.ModuleType("iotdb.thrift") +common_module = types.ModuleType("iotdb.thrift.common") +ttypes_module = types.ModuleType("iotdb.thrift.common.ttypes") +ttypes_module.TEndPoint = TEndPoint +sys.modules.setdefault("iotdb.thrift", thrift_module) +sys.modules.setdefault("iotdb.thrift.common", common_module) +sys.modules.setdefault("iotdb.thrift.common.ttypes", ttypes_module) + +config_module = import_module("iotdb.ainode.core.config") +exception_module = import_module("iotdb.ainode.core.exception") +parse_endpoint_url = config_module.parse_endpoint_url +BadNodeUrlException = exception_module.BadNodeUrlException + + +class ParseEndpointUrlTest(unittest.TestCase): + def test_parse_bracketed_ipv6_endpoint(self): + endpoint = parse_endpoint_url("[::1]:6667") + + self.assertEqual("::1", endpoint.ip) + self.assertEqual(6667, endpoint.port) + + def test_parse_legacy_ipv6_endpoint(self): + endpoint = parse_endpoint_url("::1:6667") + + self.assertEqual("::1", endpoint.ip) + self.assertEqual(6667, endpoint.port) + + def test_reject_malformed_endpoint(self): + malformed_urls = [ + "[]:123", + "foo[::1]:6667", + "foo]bar:6667", + ] + + for endpoint_url in malformed_urls: + with self.subTest(endpoint_url=endpoint_url): + with self.assertRaises(BadNodeUrlException): + parse_endpoint_url(endpoint_url) + + +if __name__ == "__main__": + unittest.main() diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java index cb71ed3fa4d80..7b052cebf9793 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java @@ -30,6 +30,7 @@ import org.apache.iotdb.consensus.i18n.ConsensusMessages; import org.apache.iotdb.rpc.AutoScalingBufferWriteTransport; import org.apache.iotdb.rpc.RpcSslUtils; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.ratis.client.RaftClientConfigKeys; import org.apache.ratis.conf.Parameters; @@ -82,7 +83,7 @@ public class Utils { private Utils() {} public static String hostAddress(TEndPoint endpoint) { - return String.format("%s:%d", endpoint.getIp(), endpoint.getPort()); + return UrlUtils.convertTEndPointIpv4AndIpv6Url(endpoint); } public static String fromTEndPointToString(TEndPoint endpoint) { @@ -103,8 +104,7 @@ public static RaftPeerId fromNodeIdToRaftPeerId(int nodeId) { } public static TEndPoint fromRaftPeerAddressToTEndPoint(String address) { - String[] items = address.split(":"); - return new TEndPoint(items[0], Integer.parseInt(items[1])); + return UrlUtils.parseTEndPointIpv4AndIpv6Url(address); } public static int fromRaftPeerIdToNodeId(RaftPeerId id) { @@ -112,8 +112,7 @@ public static int fromRaftPeerIdToNodeId(RaftPeerId id) { } public static TEndPoint fromRaftPeerProtoToTEndPoint(RaftPeerProto proto) { - String[] items = proto.getAddress().split(":"); - return new TEndPoint(items[0], Integer.parseInt(items[1])); + return fromRaftPeerAddressToTEndPoint(proto.getAddress()); } // priority is used as ordinal of leader election diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java index b1ba21950063d..5633459d38735 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/utils/UtilsTest.java @@ -19,11 +19,13 @@ package org.apache.iotdb.consensus.ratis.utils; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.consensus.ConfigRegionId; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.consensus.config.RatisConfig; import org.apache.ratis.protocol.RaftGroupId; +import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.util.TimeDuration; import org.junit.Assert; import org.junit.Test; @@ -64,4 +66,56 @@ public void testMaxRetryCalculation() { TimeDuration.valueOf(52700, TimeUnit.MILLISECONDS), Utils.getMaxRetrySleepTime(clientConfig)); } + + @Test + public void testRaftPeerAddressRoundTripWithIpv4() { + TEndPoint endPoint = new TEndPoint("192.0.0.1", 10720); + + Assert.assertEquals("192.0.0.1:10720", Utils.hostAddress(endPoint)); + Assert.assertEquals( + endPoint, Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint))); + } + + @Test + public void testRaftPeerAddressRoundTripWithHostName() { + TEndPoint endPoint = new TEndPoint("localhost", 10720); + + Assert.assertEquals("localhost:10720", Utils.hostAddress(endPoint)); + Assert.assertEquals( + endPoint, Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint))); + } + + @Test + public void testRaftPeerAddressRoundTripWithIpv6() { + TEndPoint endPoint = new TEndPoint("::1", 10720); + + Assert.assertEquals("[::1]:10720", Utils.hostAddress(endPoint)); + Assert.assertEquals( + endPoint, Utils.fromRaftPeerAddressToTEndPoint(Utils.hostAddress(endPoint))); + Assert.assertEquals(endPoint, Utils.fromRaftPeerAddressToTEndPoint("::1:10720")); + } + + @Test(expected = NumberFormatException.class) + public void testRaftPeerAddressRejectsBracketedIpv6WithoutPortSeparator() { + Utils.fromRaftPeerAddressToTEndPoint("[::1]10720"); + } + + @Test(expected = NumberFormatException.class) + public void testRaftPeerAddressRejectsBracketedIpv6WithoutPort() { + Utils.fromRaftPeerAddressToTEndPoint("[::1]:"); + } + + @Test(expected = NumberFormatException.class) + public void testRaftPeerAddressRejectsBracketedIpv6WithoutEndMark() { + Utils.fromRaftPeerAddressToTEndPoint("[::1:10720"); + } + + @Test + public void testRaftPeerProtoRoundTripWithIpv6() { + TEndPoint endPoint = new TEndPoint("0:0:0:0:0:0:0:1", 10720); + RaftPeer raftPeer = Utils.fromNodeInfoAndPriorityToRaftPeer(1, endPoint, 0); + + Assert.assertEquals("[0:0:0:0:0:0:0:1]:10720", raftPeer.getAddress()); + Assert.assertEquals(endPoint, Utils.fromRaftPeerProtoToTEndPoint(raftPeer.getRaftPeerProto())); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java index 460c3c5f8461b..230bc80c98f57 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java @@ -39,6 +39,7 @@ import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp; import org.apache.thrift.TException; @@ -404,7 +405,10 @@ private void waitHandshakeFinished(final AtomicBoolean isHandshakeFinished) { } public void updateLeaderCache(final String deviceId, final TEndPoint endPoint) { - if (!useLeaderCache || deviceId == null || endPoint == null) { + if (!useLeaderCache + || deviceId == null + || endPoint == null + || UrlUtils.isWildcardAddress(endPoint.getIp())) { return; } @@ -561,7 +565,7 @@ private static String generateClientResourceKey( "%s-%s", receiverAttributes, endPoints.stream() - .map(endPoint -> String.format("%s:%s", endPoint.getIp(), endPoint.getPort())) + .map(UrlUtils::convertTEndPointIpv4AndIpv6Url) .distinct() .sorted() .collect(Collectors.joining(",", "[", "]"))); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java index 841982accd3e6..a97d003c68807 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; @@ -115,7 +116,10 @@ public Pair getClient(final TEndPoint endPoint) { } public void updateLeaderCache(final String deviceId, final TEndPoint endPoint) { - if (!useLeaderCache || deviceId == null || endPoint == null) { + if (!useLeaderCache + || deviceId == null + || endPoint == null + || UrlUtils.isWildcardAddress(endPoint.getIp())) { return; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index 213e43ab0b844..3be7336badbdf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -66,6 +66,7 @@ import org.apache.iotdb.pipe.api.exception.PipeConnectionException; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq; import com.google.common.collect.ImmutableSet; @@ -876,7 +877,7 @@ private static boolean isSuccess(final TSStatus status) { } private static String format(final TEndPoint endPoint) { - return Objects.isNull(endPoint) ? null : endPoint.getIp() + ":" + endPoint.getPort(); + return Objects.isNull(endPoint) ? null : UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint); } //////////////////////////// Operations for close //////////////////////////// diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java index 483bb3ae1ff9f..e71af67cafd10 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/TestConnectionTask.java @@ -33,6 +33,7 @@ import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask; import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.UrlUtils; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; @@ -173,7 +174,7 @@ private static String connectionResultToString(TTestConnectionResult result) { } private static String endPointToString(TEndPoint endPoint) { - return endPoint.getIp() + ":" + endPoint.getPort(); + return UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint); } private static void sortTestConnectionResp(TTestConnectionResp origin) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java index f564dbbd70d03..fc39fb4a0751d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java @@ -22,16 +22,24 @@ import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager; +import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeCacheLeaderClientManager; +import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager; import org.junit.Assert; import org.junit.Test; import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.concurrent.ExecutorService; public class IoTDBDataNodeAsyncClientManagerTest { + private static final List WILDCARD_ADDRESSES = + Arrays.asList("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0", "0::0"); + @Test public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges() throws Exception { final IoTDBDataNodeAsyncClientManager managerWithSkipIf = @@ -114,6 +122,85 @@ public void testClientResourcesShouldDifferentiateEndPoints() throws Exception { } } + @Test + public void testAsyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() { + final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667); + final List endPoints = new ArrayList<>(Collections.singletonList(originalEndPoint)); + final IoTDBDataNodeAsyncClientManager manager = createAsyncManager(endPoints); + + try { + for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) { + final String deviceId = "async-wildcard-device-" + i; + manager.updateLeaderCache(deviceId, new TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i)); + + Assert.assertEquals(Collections.singletonList(originalEndPoint), endPoints); + Assert.assertNull( + IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId)); + } + + final String ipv6DeviceId = "async-ipv6-device"; + final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677); + manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint); + + Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint), endPoints); + Assert.assertEquals( + ipv6EndPoint, + IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint( + ipv6DeviceId)); + } finally { + manager.close(); + } + } + + @Test + public void testSyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() { + final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667); + final List endPoints = new ArrayList<>(Collections.singletonList(originalEndPoint)); + final TestIoTDBDataNodeSyncClientManager manager = + new TestIoTDBDataNodeSyncClientManager(endPoints); + + try { + for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) { + final String deviceId = "sync-wildcard-device-" + i; + manager.updateLeaderCache(deviceId, new TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i)); + + Assert.assertEquals(Collections.singletonList(originalEndPoint), endPoints); + Assert.assertEquals(0, manager.getReconstructionCount()); + Assert.assertNull( + IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId)); + } + + final String ipv6DeviceId = "sync-ipv6-device"; + final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677); + manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint); + + Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint), endPoints); + Assert.assertEquals(1, manager.getReconstructionCount()); + Assert.assertEquals( + ipv6EndPoint, + IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint( + ipv6DeviceId)); + } finally { + manager.close(); + } + } + + private static IoTDBDataNodeAsyncClientManager createAsyncManager( + final List endPoints) { + return new IoTDBDataNodeAsyncClientManager( + endPoints, + true, + "round-robin", + new UserEntity(1L, "user", "cli-host"), + "password", + true, + "sync", + true, + true, + false, + true); + } + private static String getReceiverAttributes(final IoTDBDataNodeAsyncClientManager manager) throws Exception { final Field field = @@ -142,4 +229,35 @@ private static ExecutorService getExecutor(final IoTDBDataNodeAsyncClientManager field.setAccessible(true); return (ExecutorService) field.get(manager); } + + private static class TestIoTDBDataNodeSyncClientManager extends IoTDBDataNodeSyncClientManager { + + private int reconstructionCount; + + private TestIoTDBDataNodeSyncClientManager(final List endPoints) { + super( + endPoints, + false, + null, + null, + true, + "round-robin", + new UserEntity(1L, "user", "cli-host"), + "password", + true, + "sync", + true, + true, + true); + } + + @Override + protected void reconstructClient(final TEndPoint endPoint) { + reconstructionCount++; + } + + private int getReconstructionCount() { + return reconstructionCount; + } + } } diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 4c06d0c429887..9bb7e154542ac 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -36,7 +36,7 @@ cluster_name=defaultCluster # For other ConfigNodes that to join the cluster, cn_seed_config_node points to any running ConfigNode's cn_internal_address:cn_internal_port. # Note: After this ConfigNode successfully joins the cluster for the first time, this parameter is no longer used. # Each node automatically maintains the list of ConfigNodes and traverses connections when restarting. -# Format: address:port e.g. 127.0.0.1:10710 +# Format: address:port or [ipv6]:port e.g. 127.0.0.1:10710 or [::1]:10710 # effectiveMode: first_start # Datatype: String cn_seed_config_node=127.0.0.1:10710 @@ -44,7 +44,7 @@ cn_seed_config_node=127.0.0.1:10710 # dn_seed_config_node points to any running ConfigNode's cn_internal_address:cn_internal_port. # Note: After this DataNode successfully joins the cluster for the first time, this parameter is no longer used. # Each node automatically maintains the list of ConfigNodes and traverses connections when restarting. -# Format: address:port e.g. 127.0.0.1:10710 +# Format: address:port or [ipv6]:port e.g. 127.0.0.1:10710 or [::1]:10710 # effectiveMode: first_start # Datatype: String dn_seed_config_node=127.0.0.1:10710 @@ -54,7 +54,7 @@ dn_seed_config_node=127.0.0.1:10710 #################### # Used for RPC communication inside cluster. -# Could set 127.0.0.1(for local test) or ipv4 address. +# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname. # effectiveMode: first_start # Datatype: String cn_internal_address=127.0.0.1 @@ -70,7 +70,7 @@ cn_internal_port=10710 cn_consensus_port=10720 # Used for connection of IoTDB native clients(Session) -# Could set 127.0.0.1(for local test) or ipv4 address +# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname. # effectiveMode: restart # Datatype: String dn_rpc_address=127.0.0.1 @@ -82,7 +82,7 @@ dn_rpc_address=127.0.0.1 dn_rpc_port=6667 # Used for communication inside cluster. -# could set 127.0.0.1(for local test) or ipv4 address. +# Could set 127.0.0.1(for local test), ipv4/ipv6 address, or hostname. # effectiveMode: first_start # Datatype: String dn_internal_address=127.0.0.1 diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java index 0c6aab025661f..87d50fc75c79d 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/NodeUrlUtils.java @@ -49,13 +49,10 @@ public class NodeUrlUtils { * Convert TEndPoint to TEndPointUrl * * @param endPoint TEndPoint - * @return TEndPointUrl with format ip:port + * @return TEndPointUrl with format host:port or [ipv6]:port */ public static String convertTEndPointUrl(TEndPoint endPoint) { - StringJoiner url = new StringJoiner(":"); - url.add(endPoint.getIp()); - url.add(String.valueOf(endPoint.getPort())); - return url.toString(); + return UrlUtils.convertTEndPointIpv4AndIpv6Url(endPoint); } /** @@ -75,7 +72,7 @@ public static String convertTEndPointUrls(List endPoints) { /** * Parse TEndPoint from a given TEndPointUrl * - * @param endPointUrl ip:port + * @param endPointUrl host:port or [ipv6]:port * @return TEndPoint * @throws BadNodeUrlException Throw when unable to parse */ diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java index 6f6e86ac8ded5..df1c6a216d4f4 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/NodeUrlUtilsTest.java @@ -69,6 +69,9 @@ public void parseAndConvertTEndPointUrlsIPV4AndIPV6Test() throws BadNodeUrlExcep new TEndPoint("::13.1.68.3", 6669)); final String endPointUrls = "AD80:E32B:CA25:B3AE:DC4C:DAAF:CCDE:2345:6667,[0:0:0:0:0:FFFF:129.144.52.38]:6668,[::13.1.68.3]:6669"; + final String convertedEndPointUrls = + "[AD80:E32B:CA25:B3AE:DC4C:DAAF:CCDE:2345]:6667,[0:0:0:0:0:FFFF:129.144.52.38]:6668,[::13.1.68.3]:6669"; + Assert.assertEquals(convertedEndPointUrls, NodeUrlUtils.convertTEndPointUrls(endPoints)); Assert.assertEquals(endPoints, NodeUrlUtils.parseTEndPointUrls(endPointUrls)); } @@ -90,6 +93,10 @@ public void parseAndConvertTConfigNodeUrlsIPV4AndIPV6Test() throws BadNodeUrlExc new TEndPoint("AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD", 22282))); final String configNodeUrls = "0,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22277,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22278;1,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22279,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22280;2,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22281,AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD:22282"; + final String convertedConfigNodeUrls = + "0,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22277,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22278;1,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22279,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22280;2,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22281,[AD80:E32B:CA25:B3AE:DC4C:DAAF:CDDE:ABFD]:22282"; + Assert.assertEquals( + convertedConfigNodeUrls, NodeUrlUtils.convertTConfigNodeUrls(configNodeLocations)); Assert.assertEquals(configNodeLocations, NodeUrlUtils.parseTConfigNodeUrls(configNodeUrls)); } }