Skip to content

Commit 82f7ca6

Browse files
CaideyipiCaideyipi
authored
Pipe: Fixed the OPC UA client connection problem (#17083)
* fix * IT
1 parent d06bced commit 82f7ca6

6 files changed

Lines changed: 188 additions & 36 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeOPCUAIT.java‎

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,11 @@ public void testOPCUAServerSink() throws Exception {
139139
env,
140140
Arrays.asList(
141141
"create aligned timeSeries root.db.opc(value double, quality boolean, other int32)",
142-
"insert into root.db.opc(time, value, quality, other) values (0, 0, true, 1)"),
142+
"create aligned timeSeries root.db.opc1(value double, quality boolean, other int32)",
143+
"create aligned timeSeries root.db.opc2(value double, quality boolean, other int32)",
144+
"insert into root.db.opc(time, value, quality, other) values (0, 0, true, 1)",
145+
"insert into root.db.opc1(time, value, quality, other) values (0, 0, true, 1)",
146+
"insert into root.db.opc2(time, value, quality, other) values (0, 0, true, 1)"),
143147
null);
144148

145149
while (true) {
@@ -175,9 +179,13 @@ public void testOPCUAServerSink() throws Exception {
175179
break;
176180
}
177181

178-
TestUtils.executeNonQuery(
182+
// Test multiple regions
183+
TestUtils.executeNonQueries(
179184
env,
180-
"insert into root.db.opc(time, value, quality, other) values (1, 1, false, 1)",
185+
Arrays.asList(
186+
"insert into root.db.opc(time, value, quality, other) values (1, 1, false, 1)",
187+
"insert into root.db.opc1(time, value, quality, other) values (1, 1, false, 1)",
188+
"insert into root.db.opc2(time, value, quality, other) values (1, 1, false, 1)"),
181189
null);
182190

183191
long startTime = System.currentTimeMillis();
@@ -188,6 +196,22 @@ public void testOPCUAServerSink() throws Exception {
188196
Assert.assertEquals(new Variant(1.0), value.getValue());
189197
Assert.assertEquals(StatusCode.BAD, value.getStatusCode());
190198
Assert.assertEquals(new DateTime(timestampToUtc(1)), value.getSourceTime());
199+
200+
value =
201+
opcUaClient
202+
.readValue(0, TimestampsToReturn.Both, new NodeId(2, "root/db/opc1"))
203+
.get();
204+
Assert.assertEquals(new Variant(1.0), value.getValue());
205+
Assert.assertEquals(StatusCode.BAD, value.getStatusCode());
206+
Assert.assertEquals(new DateTime(timestampToUtc(1)), value.getSourceTime());
207+
208+
value =
209+
opcUaClient
210+
.readValue(0, TimestampsToReturn.Both, new NodeId(2, "root/db/opc2"))
211+
.get();
212+
Assert.assertEquals(new Variant(1.0), value.getValue());
213+
Assert.assertEquals(StatusCode.BAD, value.getStatusCode());
214+
Assert.assertEquals(new DateTime(timestampToUtc(1)), value.getSourceTime());
191215
break;
192216
} catch (final Throwable t) {
193217
if (System.currentTimeMillis() - startTime > 10_000L) {
@@ -345,7 +369,7 @@ private static OpcUaClient getOpcUaClient(
345369
+ UUID.nameUUIDFromBytes(nodeUrl.getBytes(TSFileConfig.STRING_CHARSET));
346370

347371
client = new IoTDBOpcUaClient(nodeUrl, policy, provider, false);
348-
new ClientRunner(client, securityDir, password).run();
372+
new ClientRunner(client, securityDir, password, userName, 10).run();
349373
return client.getClient();
350374
}
351375
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java‎

Lines changed: 62 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,8 @@
100100
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_SECURITY_POLICY_SERVER_DEFAULT_VALUES;
101101
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_DEFAULT_VALUE;
102102
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY;
103+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TIMEOUT_SECONDS_DEFAULT_VALUE;
104+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_TIMEOUT_SECONDS_KEY;
103105
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_DEFAULT_VALUE;
104106
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_VALUE_NAME_KEY;
105107
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_WITH_QUALITY_DEFAULT_VALUE;
@@ -118,6 +120,7 @@
118120
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_SECURITY_DIR_KEY;
119121
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_SECURITY_POLICY_KEY;
120122
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_TCP_BIND_PORT_KEY;
123+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_TIMEOUT_SECONDS_KEY;
121124
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_VALUE_NAME_KEY;
122125
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_WITH_QUALITY_KEY;
123126

@@ -137,8 +140,11 @@ public class OpcUaSink implements PipeConnector {
137140

138141
private static final Map<String, Pair<AtomicInteger, OpcUaNameSpace>>
139142
SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP = new ConcurrentHashMap<>();
143+
private static final Map<String, Pair<AtomicInteger, IoTDBOpcUaClient>>
144+
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP = new ConcurrentHashMap<>();
140145

141146
private String serverKey;
147+
private String nodeUrl;
142148
private boolean isClientServerModel;
143149
private String databaseName;
144150
private String placeHolder4NullTag;
@@ -238,16 +244,15 @@ public void customize(
238244
"When the OPC UA sink sets 'with-quality' to true, the table model data is not supported.");
239245
}
240246

241-
final String nodeUrl =
242-
parameters.getStringByKeys(CONNECTOR_OPC_UA_NODE_URL_KEY, SINK_OPC_UA_NODE_URL_KEY);
247+
nodeUrl = parameters.getStringByKeys(CONNECTOR_OPC_UA_NODE_URL_KEY, SINK_OPC_UA_NODE_URL_KEY);
243248
if (Objects.isNull(nodeUrl)) {
244249
customizeServer(parameters);
245250
} else {
246251
if (PathUtils.isTableModelDatabase(databaseName)) {
247252
throw new PipeException(
248253
"When the OPC UA sink points to an outer server, the table model data is not supported.");
249254
}
250-
customizeClient(nodeUrl, parameters);
255+
customizeClient(parameters);
251256
}
252257
}
253258

@@ -350,7 +355,7 @@ private void customizeServer(final PipeParameters parameters) {
350355
}
351356
}
352357

353-
private void customizeClient(final String nodeUrl, final PipeParameters parameters) {
358+
private void customizeClient(final PipeParameters parameters) {
354359
final SecurityPolicy policy =
355360
getSecurityPolicy(
356361
parameters
@@ -380,15 +385,39 @@ private void customizeClient(final String nodeUrl, final PipeParameters paramete
380385
+ File.separatorChar
381386
+ UUID.nameUUIDFromBytes(nodeUrl.getBytes(TSFileConfig.STRING_CHARSET))));
382387

383-
client =
384-
new IoTDBOpcUaClient(
385-
nodeUrl,
386-
policy,
387-
provider,
388-
parameters.getBooleanOrDefault(
389-
Arrays.asList(CONNECTOR_OPC_UA_HISTORIZING_KEY, SINK_OPC_UA_HISTORIZING_KEY),
390-
CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE));
391-
new ClientRunner(client, securityDir, password).run();
388+
final long timeoutSeconds =
389+
parameters.getLongOrDefault(
390+
Arrays.asList(CONNECTOR_OPC_UA_TIMEOUT_SECONDS_KEY, SINK_OPC_UA_TIMEOUT_SECONDS_KEY),
391+
CONNECTOR_OPC_UA_TIMEOUT_SECONDS_DEFAULT_VALUE);
392+
393+
synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) {
394+
client =
395+
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP
396+
.compute(
397+
nodeUrl,
398+
(key, oldValue) -> {
399+
if (Objects.isNull(oldValue)) {
400+
final IoTDBOpcUaClient result =
401+
new IoTDBOpcUaClient(
402+
nodeUrl,
403+
policy,
404+
provider,
405+
parameters.getBooleanOrDefault(
406+
Arrays.asList(
407+
CONNECTOR_OPC_UA_HISTORIZING_KEY,
408+
SINK_OPC_UA_HISTORIZING_KEY),
409+
CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE));
410+
final ClientRunner runner =
411+
new ClientRunner(result, securityDir, password, userName, timeoutSeconds);
412+
runner.run();
413+
return new Pair<>(new AtomicInteger(0), result);
414+
}
415+
oldValue.getRight().checkEquals(userName, password, securityDir, policy);
416+
return oldValue;
417+
})
418+
.getRight();
419+
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl).getLeft().incrementAndGet();
420+
}
392421
}
393422

394423
private SecurityPolicy getSecurityPolicy(final String securityPolicy) {
@@ -521,10 +550,6 @@ public interface ThrowingBiConsumer<T, U, E extends Exception> {
521550

522551
@Override
523552
public void close() throws Exception {
524-
if (Objects.nonNull(client)) {
525-
client.disconnect();
526-
}
527-
528553
if (serverKey == null) {
529554
return;
530555
}
@@ -544,6 +569,26 @@ public void close() throws Exception {
544569
}
545570
}
546571
}
572+
573+
if (nodeUrl == null) {
574+
return;
575+
}
576+
577+
synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) {
578+
final Pair<AtomicInteger, IoTDBOpcUaClient> pair =
579+
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.get(nodeUrl);
580+
if (pair == null) {
581+
return;
582+
}
583+
584+
if (pair.getLeft().decrementAndGet() <= 0) {
585+
try {
586+
pair.getRight().disconnect();
587+
} finally {
588+
CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP.remove(nodeUrl);
589+
}
590+
}
591+
}
547592
}
548593

549594
/////////////////////////////// Getter ///////////////////////////////

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

Lines changed: 49 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,15 +25,18 @@
2525
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
2626
import org.eclipse.milo.opcua.stack.client.security.DefaultClientCertificateValidator;
2727
import org.eclipse.milo.opcua.stack.core.security.DefaultTrustListManager;
28+
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
2829
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
2930
import org.slf4j.Logger;
3031
import org.slf4j.LoggerFactory;
3132

3233
import java.io.File;
34+
import java.nio.file.FileSystems;
3335
import java.nio.file.Files;
3436
import java.nio.file.Path;
3537
import java.nio.file.Paths;
3638
import java.security.Security;
39+
import java.util.Objects;
3740

3841
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
3942

@@ -49,14 +52,23 @@ public class ClientRunner {
4952
private final IoTDBOpcUaClient configurableUaClient;
5053
private final Path securityDir;
5154
private final String password;
55+
private final long timeoutSeconds;
56+
57+
// For conflict checking
58+
private final String user;
5259

5360
public ClientRunner(
5461
final IoTDBOpcUaClient configurableUaClient,
5562
final String securityDir,
56-
final String password) {
63+
final String password,
64+
final String user,
65+
final long timeoutSeconds) {
5766
this.configurableUaClient = configurableUaClient;
5867
this.securityDir = Paths.get(securityDir);
5968
this.password = password;
69+
this.user = user;
70+
this.timeoutSeconds = timeoutSeconds;
71+
configurableUaClient.setRunner(this);
6072
}
6173

6274
private OpcUaClient createClient() throws Exception {
@@ -90,7 +102,9 @@ private OpcUaClient createClient() throws Exception {
90102
.setCertificateChain(loader.getClientCertificateChain())
91103
.setCertificateValidator(certificateValidator)
92104
.setIdentityProvider(configurableUaClient.getIdentityProvider())
93-
.setRequestTimeout(uint(5000))
105+
.setRequestTimeout(uint(timeoutSeconds * 1000L))
106+
.setConnectTimeout(uint(timeoutSeconds * 1000L))
107+
.setMaxResponseMessageSize(uint(0))
94108
.build());
95109
}
96110

@@ -109,4 +123,37 @@ public void run() {
109123
"Error getting opc client: " + e.getClass().getSimpleName() + ": " + e.getMessage(), e);
110124
}
111125
}
126+
127+
long getTimeoutSeconds() {
128+
return timeoutSeconds;
129+
}
130+
131+
/////////////////////////////// Conflict detection ///////////////////////////////
132+
133+
void checkEquals(
134+
final String user,
135+
final String password,
136+
final Path securityDir,
137+
final SecurityPolicy securityPolicy) {
138+
checkEquals("user", this.user, user);
139+
checkEquals("password", this.password, password);
140+
checkEquals(
141+
"security dir",
142+
FileSystems.getDefault().getPath(this.securityDir.toAbsolutePath().toString()),
143+
FileSystems.getDefault().getPath(securityDir.toAbsolutePath().toString()));
144+
checkEquals("securityPolicy", configurableUaClient.getSecurityPolicy(), securityPolicy);
145+
}
146+
147+
private void checkEquals(final String attrName, Object thisAttr, Object thatAttr) {
148+
if (!Objects.equals(thisAttr, thatAttr)) {
149+
if (attrName.equals("password")) {
150+
thisAttr = "****";
151+
thatAttr = "****";
152+
}
153+
throw new PipeException(
154+
String.format(
155+
"The existing server with nodeUrl %s's %s %s conflicts to the new %s %s, reject reusing.",
156+
configurableUaClient.getNodeUrl(), attrName, thisAttr, attrName, thatAttr));
157+
}
158+
}
112159
}

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

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import org.eclipse.milo.opcua.sdk.core.ValueRanks;
3535
import org.eclipse.milo.opcua.stack.core.Identifiers;
3636
import org.eclipse.milo.opcua.stack.core.StatusCodes;
37+
import org.eclipse.milo.opcua.stack.core.UaException;
3738
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
3839
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
3940
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
@@ -55,14 +56,17 @@
5556
import org.slf4j.Logger;
5657
import org.slf4j.LoggerFactory;
5758

59+
import java.nio.file.Paths;
5860
import java.util.ArrayList;
5961
import java.util.Arrays;
6062
import java.util.List;
6163
import java.util.Objects;
64+
import java.util.concurrent.ExecutionException;
6265
import java.util.function.Predicate;
6366

6467
import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
6568
import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
69+
import static org.eclipse.milo.opcua.stack.core.StatusCodes.Bad_Timeout;
6670

6771
public class IoTDBOpcUaClient {
6872
private static final Logger LOGGER = LoggerFactory.getLogger(OpcUaNameSpace.class);
@@ -78,6 +82,7 @@ public class IoTDBOpcUaClient {
7882
private final IdentityProvider identityProvider;
7983
private OpcUaClient client;
8084
private final boolean historizing;
85+
private ClientRunner runner;
8186

8287
public IoTDBOpcUaClient(
8388
final String nodeUrl,
@@ -93,7 +98,20 @@ public IoTDBOpcUaClient(
9398
public void run(final OpcUaClient client) throws Exception {
9499
// synchronous connect
95100
this.client = client;
96-
client.connect().get();
101+
long startTime = System.currentTimeMillis();
102+
while (System.currentTimeMillis() - startTime < runner.getTimeoutSeconds() * 1000L) {
103+
try {
104+
client.connect().get();
105+
} catch (final ExecutionException e) {
106+
if (e.getCause() instanceof UaException
107+
&& ((UaException) e.getCause()).getStatusCode().getValue() == Bad_Timeout) {
108+
Thread.sleep(1000L);
109+
continue;
110+
}
111+
throw e;
112+
}
113+
break;
114+
}
97115
}
98116

99117
// Only support tree model & client-server
@@ -300,4 +318,18 @@ private static ObjectAttributes createFolderAttributes(final String name) {
300318
null // notifier
301319
);
302320
}
321+
322+
/////////////////////////////// Conflict detection ///////////////////////////////
323+
324+
public void setRunner(ClientRunner runner) {
325+
this.runner = runner;
326+
}
327+
328+
public void checkEquals(
329+
final String user,
330+
final String password,
331+
final String securityDir,
332+
final SecurityPolicy securityPolicy) {
333+
runner.checkEquals(user, password, Paths.get(securityDir), securityPolicy);
334+
}
303335
}

0 commit comments

Comments
 (0)