Skip to content

Commit f57955c

Browse files
CaideyipiCaideyipi
committed
Fix IPv6 handling in Pipe redirection
1 parent 34d64dc commit f57955c

4 files changed

Lines changed: 293 additions & 2 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeIPv6IT.java‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
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.pipe.it.dual.treemodel.auto.basic;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
24+
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
25+
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
26+
import org.apache.iotdb.db.it.utils.TestUtils;
27+
import org.apache.iotdb.it.env.MultiEnvFactory;
28+
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
29+
import org.apache.iotdb.it.framework.IoTDBTestRunner;
30+
import org.apache.iotdb.it.utils.IPv6TestUtils;
31+
import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
32+
import org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
33+
import org.apache.iotdb.rpc.TSStatusCode;
34+
35+
import org.junit.AfterClass;
36+
import org.junit.Assert;
37+
import org.junit.Before;
38+
import org.junit.BeforeClass;
39+
import org.junit.Test;
40+
import org.junit.experimental.categories.Category;
41+
import org.junit.runner.RunWith;
42+
43+
import java.util.Arrays;
44+
import java.util.Collections;
45+
import java.util.HashMap;
46+
import java.util.Map;
47+
48+
import static org.apache.iotdb.it.utils.IPv6TestUtils.IPV6_LOOPBACK_ADDRESS;
49+
50+
@RunWith(IoTDBTestRunner.class)
51+
@Category({MultiClusterIT2DualTreeAutoBasic.class})
52+
public class IoTDBPipeIPv6IT extends AbstractPipeDualTreeModelAutoIT {
53+
54+
private static String previousTestNodeAddress;
55+
56+
@BeforeClass
57+
public static void setUpIPv6() {
58+
IPv6TestUtils.assumeIPv6LoopbackAvailable();
59+
previousTestNodeAddress = IPv6TestUtils.setTestNodeAddressToIPv6Loopback();
60+
}
61+
62+
@AfterClass
63+
public static void tearDownIPv6() {
64+
IPv6TestUtils.restoreTestNodeAddress(previousTestNodeAddress);
65+
}
66+
67+
@Override
68+
@Before
69+
public void setUp() {
70+
MultiEnvFactory.createEnv(2);
71+
senderEnv = MultiEnvFactory.getEnv(0);
72+
receiverEnv = MultiEnvFactory.getEnv(1);
73+
setupConfig();
74+
senderEnv.initClusterEnvironment(1, 1);
75+
receiverEnv.initClusterEnvironment(1, 1);
76+
}
77+
78+
@Override
79+
protected void setupConfig() {
80+
super.setupConfig();
81+
senderEnv
82+
.getConfig()
83+
.getCommonConfig()
84+
.setDataReplicationFactor(1)
85+
.setSchemaReplicationFactor(1);
86+
receiverEnv
87+
.getConfig()
88+
.getCommonConfig()
89+
.setDataReplicationFactor(1)
90+
.setSchemaReplicationFactor(1);
91+
}
92+
93+
@Test
94+
public void testSyncAndAsyncSinksThroughIPv6Loopback() throws Exception {
95+
assertDataNodesUseIPv6(senderEnv.getDataNodeWrapperList());
96+
assertDataNodesUseIPv6(receiverEnv.getDataNodeWrapperList());
97+
98+
final String receiverNodeUrl = receiverEnv.getDataNodeWrapper(0).getIpAndPortString();
99+
Assert.assertTrue(receiverNodeUrl.startsWith("[::1]:"));
100+
101+
try (final SyncConfigNodeIServiceClient client =
102+
(SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
103+
createAndStartPipe(
104+
client,
105+
"ipv6_async_pipe",
106+
"root.ipv6_async.**",
107+
BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
108+
receiverNodeUrl);
109+
createAndStartPipe(
110+
client,
111+
"ipv6_sync_pipe",
112+
"root.ipv6_sync.**",
113+
BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
114+
receiverNodeUrl);
115+
}
116+
117+
TestUtils.executeNonQueries(
118+
senderEnv,
119+
Arrays.asList(
120+
"insert into root.ipv6_async.d1(time, s1) values (1, 1)",
121+
"insert into root.ipv6_sync.d1(time, s1) values (1, 1)",
122+
"flush"),
123+
null);
124+
125+
TestUtils.assertDataEventuallyOnEnv(
126+
receiverEnv,
127+
"select count(s1) from root.ipv6_async.d1",
128+
"count(root.ipv6_async.d1.s1),",
129+
Collections.singleton("1,"));
130+
TestUtils.assertDataEventuallyOnEnv(
131+
receiverEnv,
132+
"select count(s1) from root.ipv6_sync.d1",
133+
"count(root.ipv6_sync.d1.s1),",
134+
Collections.singleton("1,"));
135+
}
136+
137+
private static void createAndStartPipe(
138+
final SyncConfigNodeIServiceClient client,
139+
final String pipeName,
140+
final String sourcePath,
141+
final String sinkName,
142+
final String receiverNodeUrl)
143+
throws Exception {
144+
final Map<String, String> sourceAttributes = new HashMap<>();
145+
sourceAttributes.put("source.path", sourcePath);
146+
147+
final Map<String, String> sinkAttributes = new HashMap<>();
148+
sinkAttributes.put("sink", sinkName);
149+
sinkAttributes.put("sink.batch.enable", "false");
150+
sinkAttributes.put("sink.node-urls", receiverNodeUrl);
151+
152+
final TSStatus status =
153+
client.createPipe(
154+
new TCreatePipeReq(pipeName, sinkAttributes).setExtractorAttributes(sourceAttributes));
155+
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
156+
Assert.assertEquals(
157+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.startPipe(pipeName).getCode());
158+
}
159+
160+
private static void assertDataNodesUseIPv6(final Iterable<DataNodeWrapper> dataNodes) {
161+
for (final DataNodeWrapper dataNode : dataNodes) {
162+
Assert.assertEquals(IPV6_LOOPBACK_ADDRESS, dataNode.getIp());
163+
Assert.assertTrue(dataNode.getIpAndPortString().startsWith("[::1]:"));
164+
}
165+
}
166+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -405,7 +405,10 @@ private void waitHandshakeFinished(final AtomicBoolean isHandshakeFinished) {
405405
}
406406

407407
public void updateLeaderCache(final String deviceId, final TEndPoint endPoint) {
408-
if (!useLeaderCache || deviceId == null || endPoint == null) {
408+
if (!useLeaderCache
409+
|| deviceId == null
410+
|| endPoint == null
411+
|| UrlUtils.isWildcardAddress(endPoint.getIp())) {
409412
return;
410413
}
411414

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeSyncClientManager.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
3030
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
3131
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
32+
import org.apache.iotdb.rpc.UrlUtils;
3233

3334
import org.apache.tsfile.utils.Pair;
3435
import org.slf4j.Logger;
@@ -115,7 +116,10 @@ public Pair<IoTDBSyncClient, Boolean> getClient(final TEndPoint endPoint) {
115116
}
116117

117118
public void updateLeaderCache(final String deviceId, final TEndPoint endPoint) {
118-
if (!useLeaderCache || deviceId == null || endPoint == null) {
119+
if (!useLeaderCache
120+
|| deviceId == null
121+
|| endPoint == null
122+
|| UrlUtils.isWildcardAddress(endPoint.getIp())) {
119123
return;
120124
}
121125

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/IoTDBDataNodeAsyncClientManagerTest.java‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,24 @@
2222
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
2323
import org.apache.iotdb.commons.audit.UserEntity;
2424
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeAsyncClientManager;
25+
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeCacheLeaderClientManager;
26+
import org.apache.iotdb.db.pipe.sink.client.IoTDBDataNodeSyncClientManager;
2527

2628
import org.junit.Assert;
2729
import org.junit.Test;
2830

2931
import java.lang.reflect.Field;
32+
import java.util.ArrayList;
33+
import java.util.Arrays;
3034
import java.util.Collections;
35+
import java.util.List;
3136
import java.util.concurrent.ExecutorService;
3237

3338
public class IoTDBDataNodeAsyncClientManagerTest {
3439

40+
private static final List<String> WILDCARD_ADDRESSES =
41+
Arrays.asList("0.0.0.0", "::", "[::]", "0:0:0:0:0:0:0:0", "0::0");
42+
3543
@Test
3644
public void testReceiverAttributesShouldDifferentiateSkipIfNoPrivileges() throws Exception {
3745
final IoTDBDataNodeAsyncClientManager managerWithSkipIf =
@@ -114,6 +122,85 @@ public void testClientResourcesShouldDifferentiateEndPoints() throws Exception {
114122
}
115123
}
116124

125+
@Test
126+
public void testAsyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() {
127+
final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667);
128+
final List<TEndPoint> endPoints = new ArrayList<>(Collections.singletonList(originalEndPoint));
129+
final IoTDBDataNodeAsyncClientManager manager = createAsyncManager(endPoints);
130+
131+
try {
132+
for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) {
133+
final String deviceId = "async-wildcard-device-" + i;
134+
manager.updateLeaderCache(deviceId, new TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i));
135+
136+
Assert.assertEquals(Collections.singletonList(originalEndPoint), endPoints);
137+
Assert.assertNull(
138+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
139+
}
140+
141+
final String ipv6DeviceId = "async-ipv6-device";
142+
final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677);
143+
manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint);
144+
145+
Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint), endPoints);
146+
Assert.assertEquals(
147+
ipv6EndPoint,
148+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
149+
ipv6DeviceId));
150+
} finally {
151+
manager.close();
152+
}
153+
}
154+
155+
@Test
156+
public void testSyncManagerShouldIgnoreWildcardAndAcceptIPv6LeaderEndPoints() {
157+
final TEndPoint originalEndPoint = new TEndPoint("127.0.0.1", 6667);
158+
final List<TEndPoint> endPoints = new ArrayList<>(Collections.singletonList(originalEndPoint));
159+
final TestIoTDBDataNodeSyncClientManager manager =
160+
new TestIoTDBDataNodeSyncClientManager(endPoints);
161+
162+
try {
163+
for (int i = 0; i < WILDCARD_ADDRESSES.size(); i++) {
164+
final String deviceId = "sync-wildcard-device-" + i;
165+
manager.updateLeaderCache(deviceId, new TEndPoint(WILDCARD_ADDRESSES.get(i), 6667 + i));
166+
167+
Assert.assertEquals(Collections.singletonList(originalEndPoint), endPoints);
168+
Assert.assertEquals(0, manager.getReconstructionCount());
169+
Assert.assertNull(
170+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(deviceId));
171+
}
172+
173+
final String ipv6DeviceId = "sync-ipv6-device";
174+
final TEndPoint ipv6EndPoint = new TEndPoint("::1", 6677);
175+
manager.updateLeaderCache(ipv6DeviceId, ipv6EndPoint);
176+
177+
Assert.assertEquals(Arrays.asList(originalEndPoint, ipv6EndPoint), endPoints);
178+
Assert.assertEquals(1, manager.getReconstructionCount());
179+
Assert.assertEquals(
180+
ipv6EndPoint,
181+
IoTDBDataNodeCacheLeaderClientManager.LEADER_CACHE_MANAGER.getLeaderEndPoint(
182+
ipv6DeviceId));
183+
} finally {
184+
manager.close();
185+
}
186+
}
187+
188+
private static IoTDBDataNodeAsyncClientManager createAsyncManager(
189+
final List<TEndPoint> endPoints) {
190+
return new IoTDBDataNodeAsyncClientManager(
191+
endPoints,
192+
true,
193+
"round-robin",
194+
new UserEntity(1L, "user", "cli-host"),
195+
"password",
196+
true,
197+
"sync",
198+
true,
199+
true,
200+
false,
201+
true);
202+
}
203+
117204
private static String getReceiverAttributes(final IoTDBDataNodeAsyncClientManager manager)
118205
throws Exception {
119206
final Field field =
@@ -142,4 +229,35 @@ private static ExecutorService getExecutor(final IoTDBDataNodeAsyncClientManager
142229
field.setAccessible(true);
143230
return (ExecutorService) field.get(manager);
144231
}
232+
233+
private static class TestIoTDBDataNodeSyncClientManager extends IoTDBDataNodeSyncClientManager {
234+
235+
private int reconstructionCount;
236+
237+
private TestIoTDBDataNodeSyncClientManager(final List<TEndPoint> endPoints) {
238+
super(
239+
endPoints,
240+
false,
241+
null,
242+
null,
243+
true,
244+
"round-robin",
245+
new UserEntity(1L, "user", "cli-host"),
246+
"password",
247+
true,
248+
"sync",
249+
true,
250+
true,
251+
true);
252+
}
253+
254+
@Override
255+
protected void reconstructClient(final TEndPoint endPoint) {
256+
reconstructionCount++;
257+
}
258+
259+
private int getReconstructionCount() {
260+
return reconstructionCount;
261+
}
262+
}
145263
}

0 commit comments

Comments
 (0)