Skip to content

Commit 6130d3f

Browse files
original-brownbearoriginal-brownbear
committed
Remove ChannelType.Client
There's no usage of a client channel anywhere in prod code. -> remove any notion of a client channel and replace the now irrelevant enum with a boolean to simplify things. Also, cleanup handling of the max header length to make that more obvious and remove the indirection to a `ByteSizeValue` that really just wastes CPU cache and hides the 2G maximum default (however reasonable that number may be).
1 parent daf2c22 commit 6130d3f

File tree

3 files changed

+16
-83
lines changed

3 files changed

+16
-83
lines changed

‎server/src/main/java/org/elasticsearch/transport/InboundDecoder.java‎

Lines changed: 11 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,6 @@
1616
import org.elasticsearch.common.bytes.ReleasableBytesReference;
1717
import org.elasticsearch.common.io.stream.StreamInput;
1818
import org.elasticsearch.common.recycler.Recycler;
19-
import org.elasticsearch.common.unit.ByteSizeUnit;
20-
import org.elasticsearch.common.unit.ByteSizeValue;
2119
import org.elasticsearch.core.CheckedConsumer;
2220
import org.elasticsearch.core.Releasable;
2321
import org.elasticsearch.core.Releasables;
@@ -36,21 +34,17 @@ public class InboundDecoder implements Releasable {
3634
private int bytesConsumed = 0;
3735
private boolean isCompressed = false;
3836
private boolean isClosed = false;
39-
private final ByteSizeValue maxHeaderSize;
40-
private final ChannelType channelType;
37+
private final int maxHeaderSize;
38+
private final boolean isServerChannel;
4139

4240
public InboundDecoder(Recycler<BytesRef> recycler) {
43-
this(recycler, ByteSizeValue.of(2, ByteSizeUnit.GB), ChannelType.MIX);
41+
this(recycler, Integer.MAX_VALUE, false);
4442
}
4543

46-
public InboundDecoder(Recycler<BytesRef> recycler, ChannelType channelType) {
47-
this(recycler, ByteSizeValue.of(2, ByteSizeUnit.GB), channelType);
48-
}
49-
50-
public InboundDecoder(Recycler<BytesRef> recycler, ByteSizeValue maxHeaderSize, ChannelType channelType) {
44+
public InboundDecoder(Recycler<BytesRef> recycler, int maxHeaderSize, boolean isServerChannel) {
5145
this.recycler = recycler;
5246
this.maxHeaderSize = maxHeaderSize;
53-
this.channelType = channelType;
47+
this.isServerChannel = isServerChannel;
5448
}
5549

5650
public int decode(ReleasableBytesReference reference, CheckedConsumer<Object, IOException> fragmentConsumer) throws IOException {
@@ -73,13 +67,13 @@ public int internalDecode(ReleasableBytesReference reference, CheckedConsumer<Ob
7367
fragmentConsumer.accept(PING);
7468
return 6;
7569
} else {
76-
int headerBytesToRead = headerBytesToRead(reference, maxHeaderSize);
70+
int headerBytesToRead = headerBytesToRead(reference);
7771
if (headerBytesToRead == 0) {
7872
return 0;
7973
} else {
8074
totalNetworkSize = messageLength + TcpHeader.BYTES_REQUIRED_FOR_MESSAGE_SIZE;
8175

82-
Header header = readHeader(messageLength, reference, channelType);
76+
Header header = readHeader(messageLength, reference);
8377
bytesConsumed += headerBytesToRead;
8478
if (header.isCompressed()) {
8579
isCompressed = true;
@@ -160,11 +154,7 @@ private boolean isDone() {
160154
return bytesConsumed == totalNetworkSize;
161155
}
162156

163-
private static int headerBytesToRead(BytesReference reference, ByteSizeValue maxHeaderSize) throws StreamCorruptedException {
164-
if (reference.length() < TcpHeader.BYTES_REQUIRED_FOR_VERSION) {
165-
return 0;
166-
}
167-
157+
private int headerBytesToRead(BytesReference reference) throws StreamCorruptedException {
168158
if (reference.length() <= TcpHeader.HEADER_SIZE) {
169159
return 0;
170160
} else {
@@ -173,7 +163,7 @@ private static int headerBytesToRead(BytesReference reference, ByteSizeValue max
173163
throw new StreamCorruptedException("invalid negative variable header size: " + variableHeaderSize);
174164
}
175165
int totalHeaderSize = TcpHeader.HEADER_SIZE + variableHeaderSize;
176-
if (totalHeaderSize > maxHeaderSize.getBytes()) {
166+
if (totalHeaderSize > maxHeaderSize) {
177167
throw new StreamCorruptedException("header size [" + totalHeaderSize + "] exceeds limit of [" + maxHeaderSize + "]");
178168
}
179169
if (totalHeaderSize > reference.length()) {
@@ -184,18 +174,16 @@ private static int headerBytesToRead(BytesReference reference, ByteSizeValue max
184174
}
185175
}
186176

187-
private static Header readHeader(int networkMessageSize, BytesReference bytesReference, ChannelType channelType) throws IOException {
177+
private Header readHeader(int networkMessageSize, BytesReference bytesReference) throws IOException {
188178
try (StreamInput streamInput = bytesReference.streamInput()) {
189179
streamInput.skip(TcpHeader.BYTES_REQUIRED_FOR_MESSAGE_SIZE);
190180
long requestId = streamInput.readLong();
191181
byte status = streamInput.readByte();
192182
int remoteVersion = streamInput.readInt();
193183

194184
Header header = new Header(networkMessageSize, requestId, status, TransportVersion.fromId(remoteVersion));
195-
if (channelType == ChannelType.SERVER && header.isResponse()) {
185+
if (isServerChannel && header.isResponse()) {
196186
throw new IllegalArgumentException("server channels do not accept inbound responses, only requests, closing channel");
197-
} else if (channelType == ChannelType.CLIENT && header.isRequest()) {
198-
throw new IllegalArgumentException("client channels do not accept inbound requests, only responses, closing channel");
199187
}
200188
if (header.isHandshake()) {
201189
checkHandshakeVersionCompatibility(header.getVersion());
@@ -243,9 +231,4 @@ static void checkVersionCompatibility(TransportVersion remoteVersion) {
243231
}
244232
}
245233

246-
public enum ChannelType {
247-
SERVER,
248-
CLIENT,
249-
MIX
250-
}
251234
}

‎server/src/test/java/org/elasticsearch/transport/InboundDecoderTests.java‎

Lines changed: 2 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121
import org.elasticsearch.common.util.concurrent.ThreadContext;
2222
import org.elasticsearch.test.ESTestCase;
2323
import org.elasticsearch.test.TransportVersionUtils;
24-
import org.elasticsearch.transport.InboundDecoder.ChannelType;
2524

2625
import java.io.IOException;
2726
import java.util.ArrayList;
@@ -160,56 +159,6 @@ private void doHandshakeCompatibilityTest(TransportVersion transportVersion, Com
160159
}
161160
}
162161

163-
public void testClientChannelTypeFailsDecodingRequests() throws Exception {
164-
String action = "test-request";
165-
long requestId = randomNonNegativeLong();
166-
if (randomBoolean()) {
167-
final String headerKey = randomAlphaOfLength(10);
168-
final String headerValue = randomAlphaOfLength(20);
169-
if (randomBoolean()) {
170-
threadContext.putHeader(headerKey, headerValue);
171-
} else {
172-
threadContext.addResponseHeader(headerKey, headerValue);
173-
}
174-
}
175-
// a request
176-
final var isHandshake = randomBoolean();
177-
final var version = isHandshake
178-
? randomFrom(TransportHandshaker.ALLOWED_HANDSHAKE_VERSIONS)
179-
: TransportVersionUtils.randomCompatibleVersion(random());
180-
logger.info("--> version = {}", version);
181-
182-
try (RecyclerBytesStreamOutput os = new RecyclerBytesStreamOutput(recycler)) {
183-
final BytesReference bytes = OutboundHandler.serialize(
184-
OutboundHandler.MessageDirection.REQUEST,
185-
action,
186-
requestId,
187-
isHandshake,
188-
version,
189-
randomFrom(Compression.Scheme.DEFLATE, Compression.Scheme.LZ4, null),
190-
new TestRequest(randomAlphaOfLength(100)),
191-
threadContext,
192-
os
193-
);
194-
try (InboundDecoder clientDecoder = new InboundDecoder(recycler, ChannelType.CLIENT)) {
195-
IllegalArgumentException e = expectThrows(
196-
IllegalArgumentException.class,
197-
() -> clientDecoder.decode(wrapAsReleasable(bytes), ignored -> {})
198-
);
199-
assertThat(e.getMessage(), containsString("client channels do not accept inbound requests, only responses"));
200-
}
201-
// the same message will be decoded by a server or mixed decoder
202-
try (InboundDecoder decoder = new InboundDecoder(recycler, randomFrom(ChannelType.SERVER, ChannelType.MIX))) {
203-
final ArrayList<Object> fragments = new ArrayList<>();
204-
int bytesConsumed = decoder.decode(wrapAsReleasable(bytes), fragments::add);
205-
int totalHeaderSize = TcpHeader.HEADER_SIZE + bytes.getInt(TcpHeader.VARIABLE_HEADER_SIZE_POSITION);
206-
assertEquals(totalHeaderSize, bytesConsumed);
207-
final Header header = (Header) fragments.get(0);
208-
assertEquals(requestId, header.getRequestId());
209-
}
210-
}
211-
}
212-
213162
public void testServerChannelTypeFailsDecodingResponses() throws Exception {
214163
long requestId = randomNonNegativeLong();
215164
if (randomBoolean()) {
@@ -239,13 +188,13 @@ public void testServerChannelTypeFailsDecodingResponses() throws Exception {
239188
threadContext,
240189
os
241190
);
242-
try (InboundDecoder decoder = new InboundDecoder(recycler, ChannelType.SERVER)) {
191+
try (InboundDecoder decoder = new InboundDecoder(recycler, Integer.MAX_VALUE, true)) {
243192
final ReleasableBytesReference releasable1 = wrapAsReleasable(bytes);
244193
IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> decoder.decode(releasable1, ignored -> {}));
245194
assertThat(e.getMessage(), containsString("server channels do not accept inbound responses, only requests"));
246195
}
247196
// the same message will be decoded by a client or mixed decoder
248-
try (InboundDecoder decoder = new InboundDecoder(recycler, randomFrom(ChannelType.CLIENT, ChannelType.MIX))) {
197+
try (InboundDecoder decoder = new InboundDecoder(recycler)) {
249198
final ArrayList<Object> fragments = new ArrayList<>();
250199
int bytesConsumed = decoder.decode(wrapAsReleasable(bytes), fragments::add);
251200
int totalHeaderSize = TcpHeader.HEADER_SIZE + bytes.getInt(TcpHeader.VARIABLE_HEADER_SIZE_POSITION);

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/core/security/transport/netty4/SecurityNetty4Transport.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,6 @@
5959
import javax.net.ssl.SSLEngine;
6060
import javax.net.ssl.SSLParameters;
6161

62-
import static org.elasticsearch.transport.InboundDecoder.ChannelType.SERVER;
6362
import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE;
6463
import static org.elasticsearch.xpack.core.XPackSettings.REMOTE_CLUSTER_CLIENT_SSL_ENABLED;
6564
import static org.elasticsearch.xpack.core.XPackSettings.REMOTE_CLUSTER_CLIENT_SSL_PREFIX;
@@ -80,6 +79,7 @@ public class SecurityNetty4Transport extends Netty4Transport {
8079
private final SslConfiguration remoteClusterClientSslConfiguration;
8180
private final RemoteClusterClientBootstrapOptions remoteClusterClientBootstrapOptions;
8281
private final CrossClusterAccessAuthenticationService crossClusterAccessAuthenticationService;
82+
private final int maxHeaderSize;
8383

8484
public SecurityNetty4Transport(
8585
final Settings settings,
@@ -120,6 +120,7 @@ public SecurityNetty4Transport(
120120
this.remoteClusterClientSslConfiguration = null;
121121
}
122122
this.remoteClusterClientBootstrapOptions = RemoteClusterClientBootstrapOptions.fromSettings(settings);
123+
this.maxHeaderSize = Math.toIntExact(RemoteClusterPortSettings.MAX_REQUEST_HEADER_SIZE.get(settings).getBytes());
123124
}
124125

125126
@Override
@@ -167,7 +168,7 @@ protected InboundPipeline getInboundPipeline(Channel channel, boolean isRemoteCl
167168
return new InboundPipeline(
168169
getStatsTracker(),
169170
threadPool.relativeTimeInMillisSupplier(),
170-
new InboundDecoder(recycler, RemoteClusterPortSettings.MAX_REQUEST_HEADER_SIZE.get(settings), SERVER),
171+
new InboundDecoder(recycler, maxHeaderSize, true),
171172
new InboundAggregator(getInflightBreaker(), getRequestHandlers()::getHandler, ignoreDeserializationErrors()),
172173
this::inboundMessage
173174
) {

0 commit comments

Comments
 (0)