Skip to content

Commit a6db44f

Browse files
CaideyipiCaideyipi
authored
[Pipe] Split OPC UA sink requests by advertised server limits (#18486) (#18510)
(cherry picked from commit 91a44fd)
1 parent 0329a16 commit a6db44f

2 files changed

Lines changed: 191 additions & 17 deletions

File tree

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

Lines changed: 99 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363

6464
import java.nio.file.Paths;
6565
import java.util.ArrayList;
66+
import java.util.Arrays;
6667
import java.util.HashSet;
6768
import java.util.List;
6869
import java.util.Map;
@@ -73,10 +74,14 @@
7374
import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
7475
import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
7576
import static org.eclipse.milo.opcua.stack.core.StatusCodes.Bad_Timeout;
77+
import static org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn.Neither;
7678

7779
public class IoTDBOpcUaClient {
7880
private static final Logger LOGGER = LoggerFactory.getLogger(OpcUaNameSpace.class);
7981

82+
private static final int DEFAULT_MAX_NODES_PER_WRITE = 10_000;
83+
private static final int DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT = 250;
84+
8085
// Customized nodes
8186
private static final int NAME_SPACE_INDEX = 2;
8287

@@ -90,6 +95,9 @@ public class IoTDBOpcUaClient {
9095
private final boolean historizing;
9196
private ClientRunner runner;
9297

98+
private int maxNodesPerWrite = DEFAULT_MAX_NODES_PER_WRITE;
99+
private int maxNodesPerNodeManagement = DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT;
100+
93101
public IoTDBOpcUaClient(
94102
final String nodeUrl,
95103
final SecurityPolicy securityPolicy,
@@ -118,6 +126,62 @@ public void run(final OpcUaClient client) throws Exception {
118126
}
119127
break;
120128
}
129+
updateOperationLimits();
130+
}
131+
132+
private void updateOperationLimits() {
133+
try {
134+
final List<DataValue> operationLimits =
135+
client
136+
.readValues(
137+
0.0,
138+
Neither,
139+
Arrays.asList(
140+
Identifiers.Server_ServerCapabilities_OperationLimits_MaxNodesPerWrite,
141+
Identifiers
142+
.Server_ServerCapabilities_OperationLimits_MaxNodesPerNodeManagement))
143+
.get();
144+
maxNodesPerWrite = getOperationLimit(operationLimits, 0, DEFAULT_MAX_NODES_PER_WRITE);
145+
maxNodesPerNodeManagement =
146+
getOperationLimit(operationLimits, 1, DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT);
147+
LOGGER.info(
148+
"OPC UA server operation limits: maxNodesPerWrite={}, maxNodesPerNodeManagement={}",
149+
maxNodesPerWrite,
150+
maxNodesPerNodeManagement);
151+
} catch (final InterruptedException e) {
152+
Thread.currentThread().interrupt();
153+
LOGGER.warn(
154+
"Interrupted while reading OPC UA server operation limits, use defaults: "
155+
+ "maxNodesPerWrite={}, maxNodesPerNodeManagement={}",
156+
DEFAULT_MAX_NODES_PER_WRITE,
157+
DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT);
158+
} catch (final Exception e) {
159+
LOGGER.warn(
160+
"Failed to read OPC UA server operation limits, use defaults: "
161+
+ "maxNodesPerWrite={}, maxNodesPerNodeManagement={}",
162+
DEFAULT_MAX_NODES_PER_WRITE,
163+
DEFAULT_MAX_NODES_PER_NODE_MANAGEMENT,
164+
e);
165+
}
166+
}
167+
168+
private static int getOperationLimit(
169+
final List<DataValue> operationLimits, final int index, final int defaultValue) {
170+
if (Objects.isNull(operationLimits) || operationLimits.size() <= index) {
171+
return defaultValue;
172+
}
173+
174+
final DataValue dataValue = operationLimits.get(index);
175+
if (Objects.isNull(dataValue)
176+
|| Objects.isNull(dataValue.getStatusCode())
177+
|| !dataValue.getStatusCode().isGood()
178+
|| Objects.isNull(dataValue.getValue())
179+
|| !(dataValue.getValue().getValue() instanceof Number)) {
180+
return defaultValue;
181+
}
182+
183+
final long limit = ((Number) dataValue.getValue().getValue()).longValue();
184+
return limit == 0 ? Integer.MAX_VALUE : (int) Math.min(limit, Integer.MAX_VALUE);
121185
}
122186

123187
// Only support tree model & client-server
@@ -185,7 +249,8 @@ private void collectWriteRequests(
185249
if (Objects.nonNull(sink.getValueName()) && !sink.getValueName().equals(name)) {
186250
PipeLogger.log(
187251
LOGGER::warn,
188-
"When the 'with-quality' mode is enabled, the measurement must be either \"value-name\" or \"quality-name\"");
252+
"When the 'with-quality' mode is enabled, the measurement must be either \"value-name\""
253+
+ " or \"quality-name\"");
189254
continue;
190255
}
191256

@@ -263,17 +328,23 @@ private void addMissingNodes(final List<OpcUaWriteRequest> writeRequests) throws
263328
}
264329
}
265330

266-
final AddNodesResponse addStatus = client.addNodes(nodesToAdd).get();
267-
for (final AddNodesResult result : addStatus.getResults()) {
268-
if (!result.getStatusCode().equals(StatusCode.GOOD)
269-
&& result.getStatusCode().getValue() != StatusCodes.Bad_NodeIdExists) {
270-
throw new PipeException(
271-
"Failed to create nodes after transfer data value, creation status: "
272-
+ addStatus
273-
+ writeRequests
274-
.get(0)
275-
.getErrorString(new StatusCode(StatusCodes.Bad_NodeIdUnknown)));
331+
for (int startIndex = 0; startIndex < nodesToAdd.size(); ) {
332+
final int endIndex =
333+
getBatchEndIndex(startIndex, nodesToAdd.size(), maxNodesPerNodeManagement);
334+
final AddNodesResponse addStatus =
335+
client.addNodes(nodesToAdd.subList(startIndex, endIndex)).get();
336+
for (final AddNodesResult result : addStatus.getResults()) {
337+
if (!result.getStatusCode().equals(StatusCode.GOOD)
338+
&& result.getStatusCode().getValue() != StatusCodes.Bad_NodeIdExists) {
339+
throw new PipeException(
340+
"Failed to create nodes after transfer data value, creation status: "
341+
+ addStatus
342+
+ writeRequests
343+
.get(0)
344+
.getErrorString(new StatusCode(StatusCodes.Bad_NodeIdUnknown)));
345+
}
276346
}
347+
startIndex = endIndex;
277348
}
278349
}
279350

@@ -290,13 +361,24 @@ private void validateRetriedWrites(
290361

291362
private List<StatusCode> writeValuesOnce(final List<OpcUaWriteRequest> writeRequests)
292363
throws Exception {
293-
final List<NodeId> nodeIds = new ArrayList<>(writeRequests.size());
294-
final List<DataValue> dataValues = new ArrayList<>(writeRequests.size());
295-
for (final OpcUaWriteRequest writeRequest : writeRequests) {
296-
nodeIds.add(writeRequest.nodeId);
297-
dataValues.add(writeRequest.dataValue);
364+
final List<StatusCode> writeStatuses = new ArrayList<>(writeRequests.size());
365+
for (int startIndex = 0; startIndex < writeRequests.size(); ) {
366+
final int endIndex = getBatchEndIndex(startIndex, writeRequests.size(), maxNodesPerWrite);
367+
final List<NodeId> nodeIds = new ArrayList<>(endIndex - startIndex);
368+
final List<DataValue> dataValues = new ArrayList<>(endIndex - startIndex);
369+
for (int i = startIndex; i < endIndex; ++i) {
370+
nodeIds.add(writeRequests.get(i).nodeId);
371+
dataValues.add(writeRequests.get(i).dataValue);
372+
}
373+
writeStatuses.addAll(client.writeValues(nodeIds, dataValues).get());
374+
startIndex = endIndex;
298375
}
299-
return client.writeValues(nodeIds, dataValues).get();
376+
return writeStatuses;
377+
}
378+
379+
private static int getBatchEndIndex(
380+
final int startIndex, final int totalSize, final int batchSize) {
381+
return (int) Math.min((long) totalSize, (long) startIndex + batchSize);
300382
}
301383

302384
private static final class OpcUaWriteRequest {

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

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,12 @@
3535
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
3636
import org.eclipse.milo.opcua.stack.core.StatusCodes;
3737
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
38+
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
39+
import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
3840
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
3941
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
42+
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
43+
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
4044
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesItem;
4145
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResponse;
4246
import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResult;
@@ -53,6 +57,8 @@
5357
import java.util.Map;
5458
import java.util.concurrent.CompletableFuture;
5559

60+
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
61+
5662
public class IoTDBOpcUaClientTest {
5763

5864
@Test
@@ -131,6 +137,78 @@ public void testTransferCreatesAndRetriesOnlyMissingNodes() throws Exception {
131137
.writeValues(Mockito.argThat(nodeIds("root/db/d1/s1")), Mockito.argThat(listWithSize(1)));
132138
}
133139

140+
@Test
141+
public void testTransferSplitsWritesAtServerLimit() throws Exception {
142+
final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
143+
Mockito.when(miloClient.writeValues(Mockito.anyList(), Mockito.anyList()))
144+
.thenAnswer(
145+
invocation -> {
146+
final int size = ((List<?>) invocation.getArguments()[0]).size();
147+
return CompletableFuture.completedFuture(Collections.nCopies(size, StatusCode.GOOD));
148+
});
149+
final IoTDBOpcUaClient client = createClient(miloClient, 1, 250);
150+
151+
client.transfer(createTablet(), createSink());
152+
153+
final InOrder inOrder = Mockito.inOrder(miloClient);
154+
inOrder
155+
.verify(miloClient)
156+
.writeValues(Mockito.argThat(nodeIds("root/db/d1/s1")), Mockito.argThat(listWithSize(1)));
157+
inOrder
158+
.verify(miloClient)
159+
.writeValues(Mockito.argThat(nodeIds("root/db/d1/s2")), Mockito.argThat(listWithSize(1)));
160+
}
161+
162+
@Test
163+
public void testTransferSplitsMissingNodeCreationAtServerLimit() throws Exception {
164+
final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
165+
Mockito.when(miloClient.writeValues(Mockito.anyList(), Mockito.anyList()))
166+
.thenReturn(
167+
CompletableFuture.completedFuture(
168+
Arrays.asList(
169+
new StatusCode(StatusCodes.Bad_NodeIdUnknown),
170+
new StatusCode(StatusCodes.Bad_NodeIdUnknown))))
171+
.thenReturn(
172+
CompletableFuture.completedFuture(Arrays.asList(StatusCode.GOOD, StatusCode.GOOD)));
173+
174+
final AddNodesResponse addNodesResponse = Mockito.mock(AddNodesResponse.class);
175+
final AddNodesResult addNodesResult = Mockito.mock(AddNodesResult.class);
176+
Mockito.when(addNodesResult.getStatusCode()).thenReturn(StatusCode.GOOD);
177+
Mockito.when(addNodesResponse.getResults()).thenReturn(new AddNodesResult[] {addNodesResult});
178+
Mockito.when(miloClient.addNodes(Mockito.anyList()))
179+
.thenReturn(CompletableFuture.completedFuture(addNodesResponse));
180+
181+
final IoTDBOpcUaClient client = Mockito.spy(createClient(miloClient, 10_000, 1));
182+
final AddNodesItem firstNode = Mockito.mock(AddNodesItem.class);
183+
final AddNodesItem secondNode = Mockito.mock(AddNodesItem.class);
184+
final ExpandedNodeId firstNodeId = new NodeId(2, "root/db/d1/s1").expanded();
185+
final ExpandedNodeId secondNodeId = new NodeId(2, "root/db/d1/s2").expanded();
186+
Mockito.when(firstNode.getRequestedNewNodeId()).thenReturn(firstNodeId);
187+
Mockito.when(secondNode.getRequestedNewNodeId()).thenReturn(secondNodeId);
188+
Mockito.doAnswer(
189+
invocation ->
190+
Collections.singletonList(
191+
"s1".equals(invocation.getArguments()[1]) ? firstNode : secondNode))
192+
.when(client)
193+
.getNodesToAdd(
194+
Mockito.any(String[].class),
195+
Mockito.anyString(),
196+
Mockito.any(NodeId.class),
197+
Mockito.any());
198+
199+
client.transfer(createTablet(), createSink());
200+
201+
Mockito.verify(miloClient, Mockito.times(2)).addNodes(Mockito.argThat(listWithSize(1)));
202+
final InOrder inOrder = Mockito.inOrder(miloClient);
203+
inOrder
204+
.verify(miloClient)
205+
.writeValues(Mockito.argThat(listWithSize(2)), Mockito.argThat(listWithSize(2)));
206+
inOrder.verify(miloClient, Mockito.times(2)).addNodes(Mockito.argThat(listWithSize(1)));
207+
inOrder
208+
.verify(miloClient)
209+
.writeValues(Mockito.argThat(listWithSize(2)), Mockito.argThat(listWithSize(2)));
210+
}
211+
134212
@Test
135213
public void testTransferFailsOnNonRecoverableStatus() throws Exception {
136214
final OpcUaClient miloClient = Mockito.mock(OpcUaClient.class);
@@ -152,6 +230,12 @@ public void testTransferFailsOnNonRecoverableStatus() throws Exception {
152230
}
153231

154232
private static IoTDBOpcUaClient createClient(final OpcUaClient miloClient) throws Exception {
233+
return createClient(miloClient, 10_000, 250);
234+
}
235+
236+
private static IoTDBOpcUaClient createClient(
237+
final OpcUaClient miloClient, final int maxNodesPerWrite, final int maxNodesPerNodeManagement)
238+
throws Exception {
155239
final IoTDBOpcUaClient client =
156240
new IoTDBOpcUaClient(
157241
"opc.tcp://127.0.0.1:12686", SecurityPolicy.None, new AnonymousProvider(), false);
@@ -160,6 +244,14 @@ private static IoTDBOpcUaClient createClient(final OpcUaClient miloClient) throw
160244
client.setRunner(runner);
161245
final CompletableFuture<UaClient> connectFuture = CompletableFuture.completedFuture(miloClient);
162246
Mockito.when(miloClient.connect()).thenReturn(connectFuture);
247+
Mockito.when(
248+
miloClient.readValues(
249+
Mockito.anyDouble(), Mockito.eq(TimestampsToReturn.Neither), Mockito.anyList()))
250+
.thenReturn(
251+
CompletableFuture.completedFuture(
252+
Arrays.asList(
253+
new DataValue(new Variant(uint(maxNodesPerWrite))),
254+
new DataValue(new Variant(uint(maxNodesPerNodeManagement))))));
163255
client.run(miloClient);
164256
return client;
165257
}

0 commit comments

Comments
 (0)