Skip to content

Use smaller default datagrams for strictly unidirectional air-gap UDP #19083

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 1 commit into
base: master
Choose a base branch
from
+84 −4
Open

Use smaller default datagrams for strictly unidirectional air-gap UDP #19083

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,10 @@
package org.apache.iotdb.db.i18n;

public final class DataNodePipeMessages {
public static final String
LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC =
"Air gap UDP receive buffer is {} bytes, below the requested {} bytes. Increase the operating system receive-buffer limit and restart the DataNode to reduce packet loss.";

public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 =
"Subscription replay stalled for {} ms: {}; diagnostics={}. Inspect prefetch worker stacks and WAL I/O.";

Expand Down
4 changes: 4 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,10 @@
package org.apache.iotdb.db.i18n;

public final class DataNodePipeMessages {
public static final String
LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC =
"Air gap UDP 接收缓冲区实际为 {} 字节，小于请求的 {} 字节。请提高操作系统接收缓冲区上限并重启 DataNode，以减少丢包。";

public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 =
"订阅回放已停滞 {} ms：{}；diagnostics={}。请检查预取工作线程栈和 WAL I/O。";

Expand Down
16 changes: 14 additions & 2 deletions ...main/java/org/apache/iotdb/db/pipe/receiver/protocol/airgap/IoTDBAirGapReceiverAgent.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,10 +262,22 @@ public void start() throws StartupException {
try {
serverSocket = new ServerSocket(PipeConfig.getInstance().getPipeAirGapReceiverPort());
datagramSocket = new DatagramSocket(PipeConfig.getInstance().getPipeAirGapReceiverPort());
datagramSocket.setReceiveBufferSize(
PipeConfig.getInstance().getPipeAirGapReceiverUdpSocketBufferSizeInBytes());
final int requestedBufferSize =
PipeConfig.getInstance().getPipeAirGapReceiverUdpSocketBufferSizeInBytes();
datagramSocket.setReceiveBufferSize(requestedBufferSize);
final int actualBufferSize = datagramSocket.getReceiveBufferSize();
if (actualBufferSize < requestedBufferSize) {
LOGGER.warn(
DataNodePipeMessages
.LOG_AIR_GAP_UDP_RECEIVE_BUFFER_IS_ARG_BYTES_BELOW_THE_REQUESTED_ARG_BYTES_INCREASE_THE_OPERATING_SYSTEM_RECEIVE_BUFFER_LIMIT_AND_RESTART_THE_DATANODE_TO_REDUCE_PACKET_LOSS_89B20AFC,
actualBufferSize,
requestedBufferSize);
}
datagramSocket.setSoTimeout(UDP_SESSION_CLEANUP_INTERVAL_MS);
} catch (final IOException e) {
if (Objects.nonNull(datagramSocket)) {
datagramSocket.close();
}
if (Objects.nonNull(serverSocket)) {
try {
serverSocket.close();
Expand Down
57 changes: 56 additions & 1 deletion ...est/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,50 @@
import java.util.zip.CRC32;

public class IoTDBDataRegionAirGapSinkTest {
@Test
public void testStrictUdpDefaultPacketSizeAndExplicitOverrides() throws Exception {
for (final String packetSizeKey :
Arrays.asList(
null,
PipeSinkConstant.CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY,
PipeSinkConstant.SINK_AIR_GAP_UDP_PACKET_SIZE_KEY)) {
try (final DatagramSocket receiverSocket = new DatagramSocket(0);
final UdpTestingIoTDBDataRegionAirGapSink sink =
new UdpTestingIoTDBDataRegionAirGapSink()) {
final PipeParameters parameters = buildParameters(false);
parameters.getAttribute().put(PipeSinkConstant.SINK_AIR_GAP_TRANSPORT_KEY, "udp");
parameters
.getAttribute()
.put(PipeSinkConstant.SINK_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY, "true");
if (packetSizeKey != null) {
parameters.getAttribute().put(packetSizeKey, "4096");
}
sink.validate(new PipeParameterValidator(parameters));
sink.customize(
parameters,
new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1)));

final byte[] request = new byte[2500];
request[0] = 1;
request[2] = (byte) PipeRequestType.TRANSFER_TABLET_RAW.getType();
Assert.assertTrue(sink.sendUdp(receiverSocket.getLocalPort(), request));
receiverSocket.setSoTimeout(1000);
final int packetCount = packetSizeKey == null ? 2 : 1;
for (int i = 0; i < packetCount; i++) {
final byte[] buffer = new byte[4097];
final DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
receiverSocket.receive(packet);
Assert.assertTrue(packet.getLength() <= (packetSizeKey == null ? 1400 : 4096));
Assert.assertEquals(
packetSizeKey == null
? PipeRequestType.TRANSFER_SLICE.getType()
: PipeRequestType.TRANSFER_TABLET_RAW.getType(),
decodeAirGapDatagram(buffer, packet.getLength()).type);
}
}
}
}

@Test
public void testStrictUdpSendIntervalIsValidatedAndAppliedToEverySlice() throws Exception {
for (final String key :
Expand Down Expand Up @@ -266,9 +310,20 @@ public void testUdpDefaultAndExplicitFalseRequireAcknowledgement() throws Except
sink.customize(
parameters,
new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1)));
final byte[] request = new byte[2500];
request[0] = 1;
request[2] = (byte) PipeRequestType.TRANSFER_TABLET_RAW.getType();
Assert.assertThrows(
SocketTimeoutException.class,
() -> sink.sendUdp(receiverSocket.getLocalPort(), new byte[] {1, 0, 11}));
() -> sink.sendUdp(receiverSocket.getLocalPort(), request));
receiverSocket.setSoTimeout(1000);
final byte[] buffer = new byte[3000];
final DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
receiverSocket.receive(packet);
Assert.assertEquals(request.length + 3 * Long.BYTES, packet.getLength());
Assert.assertEquals(
PipeRequestType.TRANSFER_TABLET_RAW.getType(),
decodeAirGapDatagram(buffer, packet.getLength()).type);
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions ...commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,6 +209,8 @@ private static String getDefaultConnectorOrSinkName(final PipeParameters paramet
public static final String SINK_AIR_GAP_UDP_PACKET_SIZE_KEY =
"sink.air-gap.udp.packet-size-bytes";
public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE = 60 * 1024;
// Leave room for IP/UDP headers and common encapsulation on a 1500-byte MTU link.
public static final int CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE = 1400;
public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_MIN_VALUE = 1024;
public static final int CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_MAX_VALUE = 65_507;
public static final String CONNECTOR_AIR_GAP_UDP_SEND_INTERVAL_MS_KEY =
Expand Down
5 changes: 4 additions & 1 deletion ...de-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_E_LANGUAGE_ENABLE_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_HANDSHAKE_TIMEOUT_MS_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_HANDSHAKE_TIMEOUT_MS_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_DEFAULT_VALUE;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_STRICT_UNIDIRECTIONAL_KEY;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_AIR_GAP_TRANSPORT_DEFAULT_VALUE;
Expand Down Expand Up @@ -322,12 +323,14 @@
if (strictUnidirectional) {
LOGGER.warn(
PipeMessages
.LOG_STRICTLY_UNIDIRECTIONAL_UDP_TRANSPORT_IS_ENABLED_REQUESTS_ARE_SENT_WITHOUT_ACKNOWLEDGEMENTS_DELIVERY_AND_RECEIVER_ERRORS_CANNOT_BE_CONFIRMED_0A2F4382);

Check warning on line 326 in iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/sink/protocol/IoTDBAirGapSink.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Line is longer than 100 characters (found 170). 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AaGbt9N79XzzSUl3DsRz&open=AaGbt9N79XzzSUl3DsRz&pullRequest=19083
}
udpPacketSizeInBytes =
parameters.getIntOrDefault(
Arrays.asList(CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_KEY, SINK_AIR_GAP_UDP_PACKET_SIZE_KEY),
CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE);
strictUnidirectional
? CONNECTOR_AIR_GAP_STRICT_UDP_PACKET_SIZE_DEFAULT_VALUE
: CONNECTOR_AIR_GAP_UDP_PACKET_SIZE_DEFAULT_VALUE);
strictUdpSendIntervalNanos =
TimeUnit.MILLISECONDS.toNanos(
parameters.getIntOrDefault(
Expand Down
Loading