This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 261db07cae fix(logging): apply plugin body size limits to
request/response logging (#7852)
261db07cae is described below
commit 261db07cae710abc71ad7de8f72280757d54cd4e
Author: Duansg <[email protected]>
AuthorDate: Mon Oct 19 21:11:53 2026 -0700
fix(logging): apply plugin body size limits to request/response logging
(#7852)
---
.../logging/common/AbstractLoggingPlugin.java | 6 +-
.../plugin/logging/common/body/BodyWriter.java | 124 ++++--
.../common/body/LoggingServerHttpRequest.java | 23 +-
.../common/body/LoggingServerHttpResponse.java | 32 +-
.../logging/common/config/GenericGlobalConfig.java | 14 +-
.../common/utils/LogCollectConfigUtils.java | 35 +-
.../common/AbstractLoggingPluginBodyLimitTest.java | 467 +++++++++++++++++++++
.../plugin/logging/common/body/BodyWriterTest.java | 172 ++++++++
.../common/body/LoggingBodySaturatedSizeTest.java | 94 +++++
.../common/utils/LogCollectConfigUtilsTest.java | 46 ++
10 files changed, 943 insertions(+), 70 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPlugin.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPlugin.java
index 3c9ad03be7..79166a9dc6 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPlugin.java
@@ -28,6 +28,7 @@ import org.apache.shenyu.plugin.base.utils.HostAddressUtils;
import org.apache.shenyu.plugin.logging.common.body.LoggingServerHttpRequest;
import org.apache.shenyu.plugin.logging.common.body.LoggingServerHttpResponse;
import org.apache.shenyu.plugin.logging.common.collector.LogCollector;
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
import org.apache.shenyu.plugin.logging.common.constant.GenericLoggingConstant;
import org.apache.shenyu.plugin.logging.common.entity.CommonLoggingRuleHandle;
import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
@@ -115,9 +116,10 @@ public abstract class AbstractLoggingPlugin<L extends
ShenyuRequestLog> extends
requestInfo.setSelectorId(selector.getId());
requestInfo.setRuleId(rule.getId());
requestInfo.setNamespaceId(rule.getNamespaceId());
- LoggingServerHttpRequest<L> loggingServerHttpRequest = new
LoggingServerHttpRequest<>(request, requestInfo);
+ GenericGlobalConfig globalConfig =
LogCollectConfigUtils.getGlobalConfig(selector.getPluginId());
+ LoggingServerHttpRequest<L> loggingServerHttpRequest = new
LoggingServerHttpRequest<>(request, requestInfo,
globalConfig.getMaxRequestBody());
LoggingServerHttpResponse<L> loggingServerHttpResponse = new
LoggingServerHttpResponse<>(exchange.getResponse(),
- requestInfo, this.logCollector(), desensitized, keyWordMatch,
dataDesensitizeAlg);
+ requestInfo, this.logCollector(), desensitized, keyWordMatch,
dataDesensitizeAlg, globalConfig.getMaxResponseBody());
ServerWebExchange webExchange =
exchange.mutate().request(loggingServerHttpRequest)
.response(loggingServerHttpResponse).build();
loggingServerHttpResponse.setExchange(webExchange);
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/BodyWriter.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/BodyWriter.java
index 5a63755138..68832aec5a 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/BodyWriter.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/BodyWriter.java
@@ -26,82 +26,122 @@ import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.nio.charset.StandardCharsets;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.Objects;
/**
* bodyWriter is used to read Body.
+ *
+ * <p>The body is buffered only up to {@code maxBodySize} bytes. Once the
limit is exceeded the
+ * buffer is released (so it can be garbage collected before the request
completes) and only the
+ * total size keeps being tracked. All methods are
+ * synchronized because reactive signals (e.g. cancel-triggered {@code
doFinally}) are not
+ * guaranteed to be serialized with {@code onNext}.</p>
*/
public class BodyWriter {
private static final Logger LOG =
LoggerFactory.getLogger(BodyWriter.class);
- private final ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ private final int maxBodySize;
+
+ /**
+ * Lazily allocated on the first buffered write and released on overflow;
guarded by {@code this}.
+ */
+ private ByteArrayOutputStream stream;
- private final WritableByteChannel channel = Channels.newChannel(stream);
+ private WritableByteChannel channel;
- private final AtomicBoolean isClosed = new AtomicBoolean(false);
+ private long size;
+ private boolean overflow;
+
+ private boolean closed;
+
+ /**
+ * Create a writer without a body limit (legacy behavior).
+ */
+ public BodyWriter() {
+ this(Integer.MAX_VALUE);
+ }
+
+ /**
+ * Create a writer that stops buffering once the body exceeds the limit.
+ *
+ * @param maxBodySize maximum buffered body size in bytes; non-positive
values buffer nothing
+ */
+ public BodyWriter(final int maxBodySize) {
+ this.maxBodySize = Math.max(0, maxBodySize);
+ }
/**
* write ByteBuffer.
*
* @param buffer byte buffer
*/
- public void write(final ByteBuffer buffer) {
- if (!isClosed.get()) {
- try {
- channel.write(buffer);
- } catch (IOException e) {
- isClosed.compareAndSet(false, true);
- LOG.error("write buffer Failed.", e);
- }
+ public synchronized void write(final ByteBuffer buffer) {
+ if (closed || !buffer.hasRemaining()) {
+ return;
+ }
+ size += buffer.remaining();
+ if (overflow) {
+ return;
+ }
+ if (size > maxBodySize) {
+ // release the buffered prefix so its array can be collected
before the request ends;
+ // only the total size is tracked from now on
+ overflow = true;
+ stream = null;
+ channel = null;
+ return;
+ }
+ if (Objects.isNull(stream)) {
+ stream = new ByteArrayOutputStream();
+ channel = Channels.newChannel(stream);
+ }
+ try {
+ channel.write(buffer);
+ } catch (IOException e) {
+ closed = true;
+ LOG.error("write buffer Failed.", e);
}
}
/**
- * judge stream is empty.
+ * Judge whether no body bytes have been observed.
*
- * @return true: stream is empty
+ * @return true when the body is empty
*/
- public boolean isEmpty() {
- return stream.size() == 0;
+ public synchronized boolean isEmpty() {
+ return size == 0;
}
/**
- * get stream size.
+ * Get the observed body size, including bytes discarded after exceeding
the limit.
*
- * @return size of stream
+ * @return observed byte count, capped at {@link Integer#MAX_VALUE}
*/
- public int size() {
- return stream.size();
+ public synchronized int size() {
+ return (int) Math.min(size, Integer.MAX_VALUE);
}
/**
- * output stream value.
+ * Judge whether the body exceeded the limit and was discarded.
*
- * @return string of stream
+ * @return true when the body has been discarded
*/
- public String output() {
- if (isEmpty()) {
+ public synchronized boolean isLimitExceeded() {
+ return overflow;
+ }
+
+ /**
+ * Output the buffered body and stop accepting further writes.
+ *
+ * @return string of stream, or an empty string if the body is empty or
exceeds the limit
+ */
+ public synchronized String output() {
+ closed = true;
+ if (size == 0 || overflow || Objects.isNull(stream)) {
return "";
}
- try {
- isClosed.compareAndSet(false, true);
- return new String(stream.toByteArray(), StandardCharsets.UTF_8);
- } catch (Exception e) {
- LOG.error("Write failed: ", e);
- return "Write failed: " + e.getMessage();
- } finally {
- try {
- stream.close();
- } catch (IOException e) {
- LOG.error("Close stream error: ", e);
- }
- try {
- channel.close();
- } catch (IOException e) {
- LOG.error("Close channel error: ", e);
- }
- }
+ return new String(stream.toByteArray(), StandardCharsets.UTF_8);
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpRequest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpRequest.java
index b0b0fda8c7..58a9d3aa22 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpRequest.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpRequest.java
@@ -17,8 +17,8 @@
package org.apache.shenyu.plugin.logging.common.body;
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
-import org.apache.shenyu.plugin.logging.common.utils.LogCollectConfigUtils;
import org.apache.shenyu.plugin.logging.common.utils.LogCollectUtils;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -33,9 +33,23 @@ public class LoggingServerHttpRequest<L extends
ShenyuRequestLog> extends Server
private final L logInfo;
+ private final int maxRequestBody;
+
public LoggingServerHttpRequest(final ServerHttpRequest delegate, final L
logInfo) {
+ this(delegate, logInfo, GenericGlobalConfig.DEFAULT_MAX_REQUEST_BODY);
+ }
+
+ /**
+ * Create a request decorator using the body limit captured for this
request.
+ *
+ * @param delegate delegate request
+ * @param logInfo access log
+ * @param maxRequestBody maximum logged request body size in bytes
+ */
+ public LoggingServerHttpRequest(final ServerHttpRequest delegate, final L
logInfo, final int maxRequestBody) {
super(delegate);
this.logInfo = logInfo;
+ this.maxRequestBody = maxRequestBody;
}
/**
@@ -46,7 +60,7 @@ public class LoggingServerHttpRequest<L extends
ShenyuRequestLog> extends Server
@Override
@NonNull
public Flux<DataBuffer> getBody() {
- BodyWriter writer = new BodyWriter();
+ BodyWriter writer = new BodyWriter(maxRequestBody);
return super.getBody().doOnNext(dataBuffer -> {
if (LogCollectUtils.isNotBinaryType(getHeaders())) {
try (DataBuffer.ByteBufferIterator bufferIterator =
dataBuffer.readableByteBuffers()) {
@@ -55,9 +69,8 @@ public class LoggingServerHttpRequest<L extends
ShenyuRequestLog> extends Server
}
}).doFinally(signal -> {
- int size = writer.size();
- boolean requestBodyTooLarge =
LogCollectConfigUtils.isRequestBodyTooLarge(size);
- if (size == 0 || requestBodyTooLarge) {
+ // use the writer's overflow state: size() saturates at
Integer.MAX_VALUE
+ if (writer.isEmpty() || writer.isLimitExceeded()) {
return;
}
String body = writer.output();
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpResponse.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpResponse.java
index 506cb6551d..97ce38c87e 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpResponse.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/body/LoggingServerHttpResponse.java
@@ -26,9 +26,9 @@ import org.apache.shenyu.plugin.api.result.ShenyuResult;
import org.apache.shenyu.plugin.api.result.ShenyuResultWrap;
import org.apache.shenyu.plugin.base.utils.MediaTypeUtils;
import org.apache.shenyu.plugin.logging.common.collector.LogCollector;
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
import org.apache.shenyu.plugin.logging.common.constant.GenericLoggingConstant;
import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
-import org.apache.shenyu.plugin.logging.common.utils.LogCollectConfigUtils;
import org.apache.shenyu.plugin.logging.common.utils.LogCollectUtils;
import org.apache.shenyu.plugin.logging.desensitize.api.matcher.KeyWordMatch;
import org.reactivestreams.Publisher;
@@ -76,6 +76,8 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
private final KeyWordMatch keyWordMatch;
+ private final int maxResponseBody;
+
/**
* Constructor LoggingServerHttpResponse.
*
@@ -105,12 +107,30 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
public LoggingServerHttpResponse(final ServerHttpResponse delegate, final
L logInfo,
final LogCollector<L> logCollector, final
boolean desensitized,
final KeyWordMatch keyWordMatch, final
String dataDesensitizeAlg) {
+ this(delegate, logInfo, logCollector, desensitized, keyWordMatch,
dataDesensitizeAlg, GenericGlobalConfig.DEFAULT_MAX_RESPONSE_BODY);
+ }
+
+ /**
+ * Create a response decorator using the body limit captured for this
request.
+ *
+ * @param delegate delegate response
+ * @param logInfo access log
+ * @param logCollector LogCollector instance
+ * @param desensitized desensitize flag
+ * @param keyWordMatch compiled keyword matcher
+ * @param dataDesensitizeAlg desensitize function
+ * @param maxResponseBody maximum logged response body size in bytes
+ */
+ public LoggingServerHttpResponse(final ServerHttpResponse delegate, final
L logInfo,
+ final LogCollector<L> logCollector, final
boolean desensitized,
+ final KeyWordMatch keyWordMatch, final
String dataDesensitizeAlg, final int maxResponseBody) {
super(delegate);
this.logInfo = logInfo;
this.logCollector = logCollector;
this.desensitized = desensitized;
this.dataDesensitizeAlg = dataDesensitizeAlg;
this.keyWordMatch = keyWordMatch;
+ this.maxResponseBody = maxResponseBody;
}
/**
@@ -153,7 +173,7 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
if (MediaTypeUtils.isByteType(mediaType)) {
return Flux.from(body).doFinally(signal ->
logResponse(shenyuContext, null));
}
- BodyWriter writer = new BodyWriter();
+ BodyWriter writer = new BodyWriter(maxResponseBody);
return Flux.from(body).doOnNext(buffer -> {
if (LogCollectUtils.isNotBinaryType(getHeaders())) {
try (DataBuffer.ByteBufferIterator bufferIterator =
buffer.readableByteBuffers()) {
@@ -186,8 +206,8 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
logInfo.setUpstreamIp(getUpstreamIp());
}
if (Objects.nonNull(writer)) {
- int size = writer.size();
- if (size > 0 &&
!LogCollectConfigUtils.isResponseBodyTooLarge(size)) {
+ // use the writer's overflow state: size() saturates at
Integer.MAX_VALUE
+ if (!writer.isEmpty() && !writer.isLimitExceeded()) {
String body = writer.output();
logInfo.setResponseBody(body);
}
@@ -243,7 +263,7 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
/**
* collect access error.
*
- * @param throwable Exception occurred。
+ * @param throwable Exception occurred.
*/
public void logError(final Throwable throwable) {
HttpStatusCode httpStatus = HttpStatus.INTERNAL_SERVER_ERROR;
@@ -282,7 +302,7 @@ public class LoggingServerHttpResponse<L extends
ShenyuRequestLog> extends Serve
}
int size = bytes.length;
- if (size > 0 && !LogCollectConfigUtils.isResponseBodyTooLarge(size)) {
+ if (size > 0 && size <= maxResponseBody) {
String body = new String(bytes, StandardCharsets.UTF_8);
logInfo.setResponseBody(body);
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/config/GenericGlobalConfig.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/config/GenericGlobalConfig.java
index 7bced39ce6..b651d25715 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/config/GenericGlobalConfig.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/config/GenericGlobalConfig.java
@@ -24,6 +24,16 @@ import
org.apache.shenyu.plugin.logging.common.sampler.Sampler;
*/
public class GenericGlobalConfig {
+ /**
+ * Default response body limit in bytes (512 KiB).
+ */
+ public static final int DEFAULT_MAX_RESPONSE_BODY = 524288;
+
+ /**
+ * Default request body limit in bytes (512 KiB).
+ */
+ public static final int DEFAULT_MAX_REQUEST_BODY = 524288;
+
private String sampleRate = "1";
private Sampler sampler;
@@ -31,12 +41,12 @@ public class GenericGlobalConfig {
/**
* default 512KB.
*/
- private int maxResponseBody = 524288;
+ private int maxResponseBody = DEFAULT_MAX_RESPONSE_BODY;
/**
* default 512kb.
*/
- private int maxRequestBody = 524288;
+ private int maxRequestBody = DEFAULT_MAX_REQUEST_BODY;
/**
* default 50000.
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtils.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtils.java
index 43c4fd20e1..6b94ba04ce 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtils.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtils.java
@@ -26,7 +26,6 @@ import
org.apache.shenyu.plugin.logging.common.sampler.CountSampler;
import org.apache.shenyu.plugin.logging.common.sampler.Sampler;
import org.springframework.web.server.ServerWebExchange;
-import java.util.Objects;
import java.util.Optional;
/**
@@ -34,8 +33,6 @@ import java.util.Optional;
*/
public final class LogCollectConfigUtils {
- private static GenericGlobalConfig genericGlobalConfig = new
GenericGlobalConfig();
-
private static Sampler globalSampler = Sampler.ALWAYS_SAMPLE;
private LogCollectConfigUtils() {
@@ -71,29 +68,41 @@ public final class LogCollectConfigUtils {
}
/**
- * judge whether request body too large.
+ * Get the current plugin configuration, falling back to the default
limits when absent.
+ *
+ * @param pluginId plugin id
+ * @return plugin configuration or a new default configuration
+ */
+ public static GenericGlobalConfig getGlobalConfig(final String pluginId) {
+ return Optional.ofNullable(pluginId)
+
.map(AbstractLogPluginDataHandler.getPluginGlobalConfigMap()::get)
+ .orElseGet(GenericGlobalConfig::new);
+ }
+
+ /**
+ * Judge whether the request body exceeds the default limit.
*
* @param bodySize body size
* @return whether request body too large
+ * @deprecated This method only checks the default limit. Use {@link
#getGlobalConfig(String)}
+ * and the plugin's configured request body limit instead.
*/
+ @Deprecated
public static boolean isRequestBodyTooLarge(final int bodySize) {
- if (Objects.isNull(genericGlobalConfig)) {
- return false;
- }
- return bodySize > genericGlobalConfig.getMaxRequestBody();
+ return bodySize > GenericGlobalConfig.DEFAULT_MAX_REQUEST_BODY;
}
/**
- * judge whether response body too large.
+ * Judge whether the response body exceeds the default limit.
*
* @param bodySize body size.
* @return whether response body too large
+ * @deprecated This method only checks the default limit. Use {@link
#getGlobalConfig(String)}
+ * and the plugin's configured response body limit instead.
*/
+ @Deprecated
public static boolean isResponseBodyTooLarge(final int bodySize) {
- if (Objects.isNull(genericGlobalConfig)) {
- return false;
- }
- return bodySize > genericGlobalConfig.getMaxResponseBody();
+ return bodySize > GenericGlobalConfig.DEFAULT_MAX_RESPONSE_BODY;
}
/**
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPluginBodyLimitTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPluginBodyLimitTest.java
new file mode 100644
index 0000000000..188d28f42b
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/AbstractLoggingPluginBodyLimitTest.java
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.logging.common;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.dto.PluginData;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.plugin.api.RemoteAddressResolver;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
+import org.apache.shenyu.plugin.api.result.DefaultShenyuResult;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.apache.shenyu.plugin.logging.common.body.LoggingServerHttpRequest;
+import org.apache.shenyu.plugin.logging.common.body.LoggingServerHttpResponse;
+import org.apache.shenyu.plugin.logging.common.collector.LogCollector;
+import org.apache.shenyu.plugin.logging.common.config.GenericApiConfig;
+import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
+import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
+import
org.apache.shenyu.plugin.logging.common.handler.AbstractLogPluginDataHandler;
+import org.apache.shenyu.plugin.logging.desensitize.api.matcher.KeyWordMatch;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests body limits through the logging plugin and its request and response
decorators.
+ */
+public final class AbstractLoggingPluginBodyLimitTest {
+
+ private static final String PLUGIN_ID = "body-limit-test";
+
+ private static final String OTHER_PLUGIN_ID = "other-body-limit-test";
+
+ private final RecordingLogCollector collector = new
RecordingLogCollector();
+
+ private final TestLoggingPlugin plugin = new TestLoggingPlugin(collector);
+
+ private final TestLogPluginDataHandler dataHandler = new
TestLogPluginDataHandler(collector);
+
+ private ApplicationContext previousContext;
+
+ private GenericApplicationContext applicationContext;
+
+ @BeforeEach
+ public void setUp() {
+ previousContext =
SpringBeanUtils.getInstance().getApplicationContext();
+ applicationContext = new GenericApplicationContext();
+ applicationContext.registerBean(RemoteAddressResolver.class, () -> new
RemoteAddressResolver() {
+ });
+ applicationContext.registerBean(DefaultShenyuResult.class);
+ applicationContext.refresh();
+
SpringBeanUtils.getInstance().setApplicationContext(applicationContext);
+ }
+
+ @AfterEach
+ public void tearDown() {
+
AbstractLogPluginDataHandler.getPluginGlobalConfigMap().remove(PLUGIN_ID);
+
AbstractLogPluginDataHandler.getPluginGlobalConfigMap().remove(OTHER_PLUGIN_ID);
+ SpringBeanUtils.getInstance().setApplicationContext(previousContext);
+ applicationContext.close();
+ }
+
+ @Test
+ public void testConfiguredRequestLimitAndResponseBoundary() {
+ updatePluginConfig(PLUGIN_ID, 4, 8);
+ ShenyuRequestLog log = execute(PLUGIN_ID, "hello", "response");
+ assertNull(log.getRequestBody());
+ assertEquals("response", log.getResponseBody());
+ }
+
+ @Test
+ public void testConfiguredResponseLimitAndRequestBoundary() {
+ updatePluginConfig(PLUGIN_ID, 8, 4);
+ ShenyuRequestLog log = execute(PLUGIN_ID, "request!", "hello");
+ assertEquals("request!", log.getRequestBody());
+ assertNull(log.getResponseBody());
+ }
+
+ @Test
+ public void testLimitsLargerThanDefault() {
+ updatePluginConfig(PLUGIN_ID, 1048576, 1048576);
+ String body = "x".repeat(524289);
+ ShenyuRequestLog log = execute(PLUGIN_ID, body, body);
+ assertTrue(body.equals(log.getRequestBody()), "Request body below the
configured limit should be logged");
+ assertTrue(body.equals(log.getResponseBody()), "Response body below
the configured limit should be logged");
+ }
+
+ @Test
+ public void testDefaultLimitsWhenConfigurationIsMissing() {
+ String body = "x".repeat(524288);
+ ShenyuRequestLog atLimit = execute(PLUGIN_ID, body, body);
+ assertEquals(body, atLimit.getRequestBody());
+ assertEquals(body, atLimit.getResponseBody());
+ ShenyuRequestLog overLimit = execute(PLUGIN_ID, body + "x", body +
"x");
+ assertNull(overLimit.getRequestBody());
+ assertNull(overLimit.getResponseBody());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"{}",
"{\"maxRequestBody\":null,\"maxResponseBody\":null}"})
+ public void testJsonDefaultsReplacePreviousLimits(final String configJson)
{
+ updatePluginConfig(PLUGIN_ID, 4, 4);
+ updatePluginConfig(PLUGIN_ID, configJson);
+ String body = "x".repeat(524288);
+ ShenyuRequestLog atLimit = execute(PLUGIN_ID, body, body);
+ assertEquals(body, atLimit.getRequestBody());
+ assertEquals(body, atLimit.getResponseBody());
+ ShenyuRequestLog overLimit = execute(PLUGIN_ID, body + "x", body +
"x");
+ assertNull(overLimit.getRequestBody());
+ assertNull(overLimit.getResponseBody());
+ }
+
+ @Test
+ public void testMissingJsonRequestLimitUsesDefault() {
+ updatePluginConfig(PLUGIN_ID, 4, 4);
+ updatePluginConfig(PLUGIN_ID, "{\"maxResponseBody\":4}");
+ String body = "x".repeat(524288);
+ ShenyuRequestLog atLimit = execute(PLUGIN_ID, body, "hello");
+ assertEquals(body, atLimit.getRequestBody());
+ assertNull(atLimit.getResponseBody());
+ assertNull(execute(PLUGIN_ID, body + "x", "hello").getRequestBody());
+ }
+
+ @Test
+ public void testMissingJsonResponseLimitUsesDefault() {
+ updatePluginConfig(PLUGIN_ID, 4, 4);
+ updatePluginConfig(PLUGIN_ID, "{\"maxRequestBody\":4}");
+ String body = "x".repeat(524288);
+ ShenyuRequestLog atLimit = execute(PLUGIN_ID, "hello", body);
+ assertNull(atLimit.getRequestBody());
+ assertEquals(body, atLimit.getResponseBody());
+ assertNull(execute(PLUGIN_ID, "hello", body + "x").getResponseBody());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {524288, 524289})
+ public void testLegacyRequestConstructorUsesDefaultLimit(final int
bodySize) {
+ String body = "x".repeat(bodySize);
+ ShenyuRequestLog log = new ShenyuRequestLog();
+ LoggingServerHttpRequest<ShenyuRequestLog> request = new
LoggingServerHttpRequest<>(createExchange(body).getRequest(), log);
+ request.getBody().doOnNext(DataBufferUtils::release).blockLast();
+ if (bodySize > 524288) {
+ assertNull(log.getRequestBody());
+ } else {
+ assertEquals(body, log.getRequestBody());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {524288, 524289})
+ public void testLegacyResponseConstructorUsesDefaultLimit(final int
bodySize) {
+ String body = "x".repeat(bodySize);
+ MockServerWebExchange exchange = createExchange("hello");
+ ShenyuRequestLog log = new ShenyuRequestLog();
+ LoggingServerHttpResponse<ShenyuRequestLog> response = new
LoggingServerHttpResponse<>(exchange.getResponse(),
+ log, collector, false, Collections.emptySet(), "md5");
+ response.setExchange(exchange);
+ response.writeWith(buffers(body)).block();
+ assertEquals(body, exchange.getResponse().getBodyAsString().block());
+ if (bodySize > 524288) {
+ assertNull(log.getResponseBody());
+ } else {
+ assertEquals(body, log.getResponseBody());
+ }
+ }
+
+ @Test
+ public void testPluginLimitsAreIsolated() {
+ updatePluginConfig(PLUGIN_ID, 4, 4);
+ updatePluginConfig(OTHER_PLUGIN_ID, 8, 8);
+ ShenyuRequestLog small = execute(PLUGIN_ID, "hello", "hello");
+ ShenyuRequestLog large = execute(OTHER_PLUGIN_ID, "hello", "hello");
+ final ShenyuRequestLog smallAgain = execute(PLUGIN_ID, "hello",
"hello");
+ assertNull(small.getRequestBody());
+ assertNull(small.getResponseBody());
+ assertEquals("hello", large.getRequestBody());
+ assertEquals("hello", large.getResponseBody());
+ assertNull(smallAgain.getRequestBody());
+ assertNull(smallAgain.getResponseBody());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testNestedPluginsKeepIndependentLimits(final boolean
smallerLimitFirst) {
+ updatePluginConfig(PLUGIN_ID, smallerLimitFirst ? 4 : 8,
smallerLimitFirst ? 4 : 8);
+ updatePluginConfig(OTHER_PLUGIN_ID, smallerLimitFirst ? 8 : 4,
smallerLimitFirst ? 8 : 4);
+ RecordingLogCollector otherCollector = new RecordingLogCollector();
+ TestLoggingPlugin otherPlugin = new TestLoggingPlugin(otherCollector);
+ MockServerWebExchange exchange = createExchange("hello");
+ AtomicReference<String> forwardedRequest = new AtomicReference<>();
+ plugin.doExecute(exchange, outer -> otherPlugin.doExecute(outer,
+ inner -> forward(inner, forwardedRequest, "hello"),
selector(OTHER_PLUGIN_ID), rule(OTHER_PLUGIN_ID)),
+ selector(PLUGIN_ID), rule(PLUGIN_ID)).block();
+ assertEquals("hello", forwardedRequest.get());
+ assertEquals("hello",
exchange.getResponse().getBodyAsString().block());
+ ShenyuRequestLog outerLog = collector.log.get();
+ assertNotNull(outerLog);
+ assertEquals(smallerLimitFirst ? null : "hello",
outerLog.getRequestBody());
+ assertEquals(smallerLimitFirst ? null : "hello",
outerLog.getResponseBody());
+ assertEquals(HttpStatus.CREATED.value(), outerLog.getStatus());
+ assertEquals(5, outerLog.getResponseContentLength());
+ ShenyuRequestLog innerLog = otherCollector.log.get();
+ assertNotNull(innerLog);
+ assertEquals(smallerLimitFirst ? "hello" : null,
innerLog.getRequestBody());
+ assertEquals(smallerLimitFirst ? "hello" : null,
innerLog.getResponseBody());
+ assertEquals(HttpStatus.CREATED.value(), innerLog.getStatus());
+ assertEquals(5, innerLog.getResponseContentLength());
+ }
+
+ @Test
+ public void testConfigUpdateAppliesToNextRequest() {
+ updatePluginConfig(PLUGIN_ID, 8, 8);
+ MockServerWebExchange exchange = createExchange("hello");
+ AtomicReference<String> forwardedRequest = new AtomicReference<>();
+ Mono<Void> inFlight = plugin.doExecute(exchange, decorated ->
forward(decorated, forwardedRequest, "hello"),
+ selector(PLUGIN_ID), rule(PLUGIN_ID));
+ updatePluginConfig(PLUGIN_ID, 4, 4);
+ inFlight.block();
+ assertEquals("hello", collector.log.get().getRequestBody());
+ assertEquals("hello", collector.log.get().getResponseBody());
+ ShenyuRequestLog nextRequest = execute(PLUGIN_ID, "hello", "hello");
+ assertNull(nextRequest.getRequestBody());
+ assertNull(nextRequest.getResponseBody());
+ updatePluginConfig(PLUGIN_ID, 8, 8);
+ ShenyuRequestLog updatedRequest = execute(PLUGIN_ID, "hello", "hello");
+ assertEquals("hello", updatedRequest.getRequestBody());
+ assertEquals("hello", updatedRequest.getResponseBody());
+ }
+
+ @Test
+ public void testUtf8BytesAreAccumulatedAcrossChunks() {
+ // two 3-byte UTF-8 characters, 6 bytes in total
+ final String first = Character.toString(0x20AC);
+ final String second = Character.toString(0x2122);
+ final String body = first + second;
+ updatePluginConfig(PLUGIN_ID, 5, 6);
+ MockServerWebExchange exchange = createExchange(first, second);
+ ShenyuRequestLog log = execute(PLUGIN_ID, exchange, body, first,
second);
+ assertNull(log.getRequestBody());
+ assertEquals(body, log.getResponseBody());
+ updatePluginConfig(PLUGIN_ID, 6, 5);
+ log = execute(PLUGIN_ID, createExchange(first, second), body, first,
second);
+ assertEquals(body, log.getRequestBody());
+ assertNull(log.getResponseBody());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1})
+ public void testNonPositiveLimitsOmitNonEmptyBodies(final int limit) {
+ updatePluginConfig(PLUGIN_ID, limit, limit);
+ ShenyuRequestLog log = execute(PLUGIN_ID, "hello", "hello");
+ assertNull(log.getRequestBody());
+ assertNull(log.getResponseBody());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testErrorResponseUsesConfiguredLimit(final boolean
synchronous) {
+ updatePluginConfig(PLUGIN_ID, 8, 1);
+ ShenyuRequestLog small = executeError(synchronous);
+ assertEquals(500, small.getStatus());
+ assertTrue(small.getResponseContentLength() > 1);
+ assertNull(small.getResponseBody());
+ updatePluginConfig(PLUGIN_ID, 8, 1024);
+ ShenyuRequestLog large = executeError(synchronous);
+ assertEquals(500, large.getStatus());
+ assertNotNull(large.getResponseBody());
+ assertTrue(large.getResponseBody().contains("Internal Server Error"));
+ }
+
+ private ShenyuRequestLog executeError(final boolean synchronous) {
+ IllegalStateException failure = new IllegalStateException("failure");
+ assertSame(failure, assertThrows(IllegalStateException.class, () ->
plugin.doExecute(createExchange("hello"), exchange -> {
+ if (synchronous) {
+ throw failure;
+ }
+ return Mono.error(failure);
+ }, selector(PLUGIN_ID), rule(PLUGIN_ID)).block()));
+ return collector.log.get();
+ }
+
+ private ShenyuRequestLog execute(final String pluginId, final String
requestBody, final String responseBody) {
+ return execute(pluginId, createExchange(requestBody), requestBody,
responseBody);
+ }
+
+ private ShenyuRequestLog execute(final String pluginId, final
MockServerWebExchange exchange,
+ final String requestBody, final String...
responseChunks) {
+ AtomicReference<String> forwardedRequest = new AtomicReference<>();
+ plugin.doExecute(exchange, decorated -> forward(decorated,
forwardedRequest, responseChunks), selector(pluginId), rule(pluginId)).block();
+ String responseBody = String.join("", responseChunks);
+ assertEquals(requestBody, forwardedRequest.get());
+ assertEquals(responseBody,
exchange.getResponse().getBodyAsString().block());
+ ShenyuRequestLog log = collector.log.get();
+ assertNotNull(log);
+ assertEquals(HttpStatus.CREATED.value(), log.getStatus());
+ assertEquals(responseBody.getBytes(StandardCharsets.UTF_8).length,
log.getResponseContentLength());
+ return log;
+ }
+
+ private Mono<Void> forward(final ServerWebExchange exchange, final
AtomicReference<String> forwardedRequest,
+ final String... responseChunks) {
+ return
DataBufferUtils.join(exchange.getRequest().getBody()).doOnNext(buffer -> {
+ forwardedRequest.set(buffer.toString(StandardCharsets.UTF_8));
+ DataBufferUtils.release(buffer);
+ }).then(Mono.defer(() -> {
+ exchange.getResponse().setStatusCode(HttpStatus.CREATED);
+
exchange.getResponse().getHeaders().setContentType(MediaType.TEXT_PLAIN);
+ return exchange.getResponse().writeWith(buffers(responseChunks));
+ }));
+ }
+
+ private MockServerWebExchange createExchange(final String... chunks) {
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.post("http://localhost/body-limit";)
+ .remoteAddress(new InetSocketAddress("127.0.0.1",
8000)).contentType(MediaType.TEXT_PLAIN).body(buffers(chunks)));
+ ShenyuContext context = new ShenyuContext();
+ context.setStartDateTime(LocalDateTime.now());
+ exchange.getAttributes().put(Constants.CONTEXT, context);
+ return exchange;
+ }
+
+ private Flux<DataBuffer> buffers(final String... chunks) {
+ return Flux.fromArray(chunks).map(chunk ->
DefaultDataBufferFactory.sharedInstance.wrap(chunk.getBytes(StandardCharsets.UTF_8)));
+ }
+
+ private void updatePluginConfig(final String pluginId, final int
requestLimit, final int responseLimit) {
+ updatePluginConfig(pluginId, "{\"maxRequestBody\":" + requestLimit +
",\"maxResponseBody\":" + responseLimit + "}");
+ }
+
+ private void updatePluginConfig(final String pluginId, final String
configJson) {
+ PluginData pluginData = new PluginData();
+ pluginData.setId(pluginId);
+ pluginData.setName(PluginEnum.LOGGING_KAFKA.getName());
+ pluginData.setEnabled(true);
+ pluginData.setConfig(configJson);
+ dataHandler.handlerPlugin(pluginData);
+ }
+
+ private SelectorData selector(final String pluginId) {
+ SelectorData selector = new SelectorData();
+ selector.setId(pluginId + "-selector");
+ selector.setPluginId(pluginId);
+ return selector;
+ }
+
+ private RuleData rule(final String pluginId) {
+ RuleData rule = new RuleData();
+ rule.setId(pluginId + "-rule");
+ rule.setSelectorId(pluginId + "-selector");
+ return rule;
+ }
+
+ // Keep the handler's Singleton entry separate from production
configuration classes.
+ private static final class TestGlobalConfig extends GenericGlobalConfig {
+ }
+
+ private static final class TestLogPluginDataHandler extends
AbstractLogPluginDataHandler<TestGlobalConfig, GenericApiConfig> {
+
+ private final LogCollector<ShenyuRequestLog> collector;
+
+ private TestLogPluginDataHandler(final LogCollector<ShenyuRequestLog>
collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ protected LogCollector<ShenyuRequestLog> logCollector() {
+ return collector;
+ }
+
+ @Override
+ protected void doRefreshConfig(final TestGlobalConfig globalLogConfig)
{
+ }
+
+ @Override
+ public String pluginNamed() {
+ return PluginEnum.LOGGING_KAFKA.getName();
+ }
+ }
+
+ private static final class TestLoggingPlugin extends
AbstractLoggingPlugin<ShenyuRequestLog> {
+
+ private final LogCollector<ShenyuRequestLog> collector;
+
+ private TestLoggingPlugin(final LogCollector<ShenyuRequestLog>
collector) {
+ this.collector = collector;
+ }
+
+ @Override
+ protected LogCollector<ShenyuRequestLog> logCollector() {
+ return collector;
+ }
+
+ @Override
+ protected PluginEnum pluginEnum() {
+ return PluginEnum.LOGGING_KAFKA;
+ }
+
+ @Override
+ protected ShenyuRequestLog doLogExecute(final ServerWebExchange
exchange, final SelectorData selector, final RuleData rule) {
+ return new ShenyuRequestLog();
+ }
+ }
+
+ private static final class RecordingLogCollector implements
LogCollector<ShenyuRequestLog> {
+
+ private final AtomicReference<ShenyuRequestLog> log = new
AtomicReference<>();
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public void desensitize(final ShenyuRequestLog requestLog, final
KeyWordMatch keyWordMatch, final String desensitizeAlg) {
+ }
+
+ @Override
+ public void collect(final ShenyuRequestLog requestLog) {
+ log.set(requestLog);
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/BodyWriterTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/BodyWriterTest.java
index e2f1824f97..927d74a2d4 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/BodyWriterTest.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/BodyWriterTest.java
@@ -20,10 +20,19 @@ package org.apache.shenyu.plugin.logging.common.body;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.test.util.ReflectionTestUtils;
+import java.io.ByteArrayOutputStream;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
/**
* The Test Case For BodyWriter.
@@ -69,4 +78,167 @@ public class BodyWriterTest {
String res = writer.output();
Assertions.assertEquals(res, "hello, shenyu");
}
+
+ @Test
+ public void testSplitUtf8BodyAtLimit() {
+ BodyWriter limitedWriter = new BodyWriter(6);
+ // two 3-byte UTF-8 characters, split inside the first character
+ final String body = Character.toString(0x20AC) +
Character.toString(0x2122);
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ limitedWriter.write(ByteBuffer.wrap(bytes, 0, 1).asReadOnlyBuffer());
+ limitedWriter.write(ByteBuffer.wrap(bytes, 1, bytes.length -
1).asReadOnlyBuffer());
+ Assertions.assertFalse(limitedWriter.isLimitExceeded());
+ Assertions.assertEquals(6, limitedWriter.size());
+ Assertions.assertEquals(body, limitedWriter.output());
+ }
+
+ @Test
+ public void testOverflowDropsBufferAndKeepsCounting() {
+ BodyWriter limitedWriter = new BodyWriter(4);
+
limitedWriter.write(ByteBuffer.wrap("test".getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertFalse(limitedWriter.isLimitExceeded());
+ Assertions.assertEquals(4, bufferedSize(limitedWriter));
+ limitedWriter.write(ByteBuffer.wrap(new byte[1]));
+ Assertions.assertTrue(limitedWriter.isLimitExceeded());
+ assertBufferReleased(limitedWriter);
+ for (int i = 0; i < 64; i++) {
+ limitedWriter.write(ByteBuffer.wrap(new byte[1024]));
+ }
+ Assertions.assertEquals(5 + 64 * 1024, limitedWriter.size());
+ assertBufferReleased(limitedWriter);
+ Assertions.assertFalse(limitedWriter.isEmpty());
+ Assertions.assertEquals("", limitedWriter.output());
+ }
+
+ @Test
+ public void testOversizedChunkIsNotBufferedAndSourceUntouched() {
+ BodyWriter limitedWriter = new BodyWriter(4);
+ ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
+ limitedWriter.write(buffer.asReadOnlyBuffer());
+ Assertions.assertEquals(0, buffer.position());
+ Assertions.assertEquals(1024 * 1024, limitedWriter.size());
+ Assertions.assertTrue(limitedWriter.isLimitExceeded());
+ assertBufferReleased(limitedWriter);
+ Assertions.assertEquals("", limitedWriter.output());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, -1, Integer.MIN_VALUE})
+ public void testNonPositiveLimitBuffersNothing(final int limit) {
+ BodyWriter limitedWriter = new BodyWriter(limit);
+ limitedWriter.write(ByteBuffer.allocate(0));
+ Assertions.assertTrue(limitedWriter.isEmpty());
+ Assertions.assertFalse(limitedWriter.isLimitExceeded());
+
limitedWriter.write(ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertEquals(5, limitedWriter.size());
+ Assertions.assertTrue(limitedWriter.isLimitExceeded());
+ assertBufferReleased(limitedWriter);
+ Assertions.assertEquals("", limitedWriter.output());
+ }
+
+ @Test
+ public void testObservedSizeIsCapped() {
+ BodyWriter limitedWriter = new BodyWriter(4);
+ ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
+ for (int i = 0; i < 2049; i++) {
+ limitedWriter.write(buffer.asReadOnlyBuffer());
+ }
+ Assertions.assertEquals(Integer.MAX_VALUE, limitedWriter.size());
+ Assertions.assertTrue(limitedWriter.isLimitExceeded());
+ Assertions.assertEquals("", limitedWriter.output());
+ }
+
+ @Test
+ public void testOverflowDetectedWhenSizeSaturatesAtMaxLimit() {
+ BodyWriter maxLimitWriter = new BodyWriter(Integer.MAX_VALUE);
+ // simulate a body that has already reached the int range without
allocating 2 GiB
+ ReflectionTestUtils.setField(maxLimitWriter, "size", (long)
Integer.MAX_VALUE - 1);
+ maxLimitWriter.write(ByteBuffer.wrap(new byte[2]));
+ Assertions.assertEquals(Integer.MAX_VALUE, maxLimitWriter.size());
+ Assertions.assertTrue(maxLimitWriter.isLimitExceeded());
+ assertBufferReleased(maxLimitWriter);
+ Assertions.assertEquals("", maxLimitWriter.output());
+ }
+
+ @Test
+ public void testOutputStopsFurtherWrites() {
+ BodyWriter limitedWriter = new BodyWriter(4);
+
limitedWriter.write(ByteBuffer.wrap("abc".getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertEquals("abc", limitedWriter.output());
+
limitedWriter.write(ByteBuffer.wrap("hello".getBytes(StandardCharsets.UTF_8)));
+ Assertions.assertEquals(3, limitedWriter.size());
+ Assertions.assertFalse(limitedWriter.isLimitExceeded());
+ Assertions.assertEquals("abc", limitedWriter.output());
+ }
+
+ @Test
+ public void testConcurrentWriteAndOutputDoNotFail() throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ for (int round = 0; round < 200; round++) {
+ BodyWriter concurrentWriter = new BodyWriter(64);
+ CountDownLatch start = new CountDownLatch(1);
+ Future<?> writes = executor.submit(() -> {
+ start.await();
+ for (int i = 0; i < 100; i++) {
+
concurrentWriter.write(ByteBuffer.wrap("abcdefgh".getBytes(StandardCharsets.UTF_8)));
+ }
+ return null;
+ });
+ Future<String> output = executor.submit(() -> {
+ start.await();
+ return concurrentWriter.output();
+ });
+ start.countDown();
+ writes.get(5, TimeUnit.SECONDS);
+ String body = output.get(5, TimeUnit.SECONDS);
+ // output sees either a consistent prefix within the limit, or
nothing after overflow
+ Assertions.assertTrue(body.length() <= 64 && body.length() % 8
== 0, body);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testOverflowReleasesBufferReferences() {
+ BodyWriter limitedWriter = new BodyWriter(512 * 1024);
+ ByteBuffer chunk = ByteBuffer.allocate(8192);
+ for (int i = 0; i < 64; i++) {
+ limitedWriter.write(chunk.asReadOnlyBuffer());
+ }
+ Assertions.assertFalse(limitedWriter.isLimitExceeded());
+ Assertions.assertEquals(512 * 1024, bufferedSize(limitedWriter));
+ limitedWriter.write(ByteBuffer.wrap(new byte[1]));
+ Assertions.assertTrue(limitedWriter.isLimitExceeded());
+ // the writer no longer references the buffered array, so it is
collectable before the request ends
+ assertBufferReleased(limitedWriter);
+ // later writes only count bytes and must not allocate a new buffer
+ for (int i = 0; i < 8; i++) {
+ limitedWriter.write(chunk.asReadOnlyBuffer());
+ }
+ assertBufferReleased(limitedWriter);
+ Assertions.assertEquals(512 * 1024 + 1 + 8 * 8192,
limitedWriter.size());
+ Assertions.assertEquals("", limitedWriter.output());
+ assertBufferReleased(limitedWriter);
+ }
+
+ @Test
+ public void testEmptyBodyDoesNotAllocateBuffer() {
+ BodyWriter limitedWriter = new BodyWriter(4);
+ limitedWriter.write(ByteBuffer.allocate(0));
+ Assertions.assertNull(ReflectionTestUtils.getField(limitedWriter,
"stream"));
+ Assertions.assertEquals("", limitedWriter.output());
+ }
+
+ private static void assertBufferReleased(final BodyWriter bodyWriter) {
+ Assertions.assertNull(ReflectionTestUtils.getField(bodyWriter,
"stream"));
+ Assertions.assertNull(ReflectionTestUtils.getField(bodyWriter,
"channel"));
+ }
+
+ private static int bufferedSize(final BodyWriter bodyWriter) {
+ ByteArrayOutputStream stream = (ByteArrayOutputStream)
ReflectionTestUtils.getField(bodyWriter, "stream");
+ Assertions.assertNotNull(stream);
+ return stream.size();
+ }
}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/LoggingBodySaturatedSizeTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/LoggingBodySaturatedSizeTest.java
new file mode 100644
index 0000000000..0569125f1b
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/body/LoggingBodySaturatedSizeTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shenyu.plugin.logging.common.body;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
+import org.apache.shenyu.plugin.logging.common.collector.LogCollector;
+import org.apache.shenyu.plugin.logging.common.entity.ShenyuRequestLog;
+import org.apache.shenyu.plugin.logging.desensitize.api.matcher.KeyWordMatch;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.http.MediaType;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import reactor.core.publisher.Flux;
+
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Collections;
+
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.when;
+
+/**
+ * Callers must rely on the writer's overflow state, because {@link
BodyWriter#size()} saturates at
+ * {@link Integer#MAX_VALUE} and cannot be compared against a limit of {@link
Integer#MAX_VALUE}.
+ */
+public final class LoggingBodySaturatedSizeTest {
+
+ @Test
+ public void testRequestBodyStaysNullWhenSaturatedWriterOverflows() {
+ ShenyuRequestLog log = new ShenyuRequestLog();
+ MockServerWebExchange exchange = createExchange();
+ try (MockedConstruction<BodyWriter> ignored =
mockConstruction(BodyWriter.class, (writer, context) ->
stubSaturatedOverflow(writer))) {
+ LoggingServerHttpRequest<ShenyuRequestLog> request = new
LoggingServerHttpRequest<>(exchange.getRequest(), log, Integer.MAX_VALUE);
+ request.getBody().blockLast();
+ }
+ Assertions.assertNull(log.getRequestBody());
+ }
+
+ @Test
+ public void testResponseBodyStaysNullWhenSaturatedWriterOverflows() {
+ ShenyuRequestLog log = new ShenyuRequestLog();
+ MockServerWebExchange exchange = createExchange();
+
exchange.getResponse().getHeaders().setContentType(MediaType.TEXT_PLAIN);
+ try (MockedConstruction<BodyWriter> ignored =
mockConstruction(BodyWriter.class, (writer, context) ->
stubSaturatedOverflow(writer))) {
+ final LogCollector<ShenyuRequestLog> noCollector = null;
+ LoggingServerHttpResponse<ShenyuRequestLog> response = new
LoggingServerHttpResponse<>(exchange.getResponse(), log,
+ noCollector, false, new
KeyWordMatch(Collections.emptySet()), "md5", Integer.MAX_VALUE);
+ response.setExchange(exchange);
+ response.writeWith(buffers()).block();
+ }
+ Assertions.assertNull(log.getResponseBody());
+ Assertions.assertEquals(Integer.MAX_VALUE,
log.getResponseContentLength());
+ }
+
+ private static void stubSaturatedOverflow(final BodyWriter writer) {
+ when(writer.size()).thenReturn(Integer.MAX_VALUE);
+ when(writer.isEmpty()).thenReturn(false);
+ when(writer.isLimitExceeded()).thenReturn(true);
+ when(writer.output()).thenReturn("");
+ }
+
+ private static MockServerWebExchange createExchange() {
+ MockServerWebExchange exchange =
MockServerWebExchange.from(MockServerHttpRequest.post("http://localhost/saturated";)
+ .contentType(MediaType.TEXT_PLAIN).body(buffers()));
+ ShenyuContext context = new ShenyuContext();
+ context.setStartDateTime(LocalDateTime.now());
+ exchange.getAttributes().put(Constants.CONTEXT, context);
+ return exchange;
+ }
+
+ private static Flux<DataBuffer> buffers() {
+ return
Flux.just(DefaultDataBufferFactory.sharedInstance.wrap("hello".getBytes(StandardCharsets.UTF_8)));
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtilsTest.java
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtilsTest.java
index a6ab7e99a5..4fa3e728e7 100644
---
a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtilsTest.java
+++
b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/utils/LogCollectConfigUtilsTest.java
@@ -20,6 +20,7 @@ package org.apache.shenyu.plugin.logging.common.utils;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.plugin.api.context.ShenyuContext;
import org.apache.shenyu.plugin.logging.common.config.GenericGlobalConfig;
+import
org.apache.shenyu.plugin.logging.common.handler.AbstractLogPluginDataHandler;
import org.apache.shenyu.plugin.logging.common.sampler.Sampler;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -33,6 +34,9 @@ import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* The Test Case For LogCollectConfigUtils.
@@ -65,4 +69,46 @@ public class LogCollectConfigUtilsTest {
field.setAccessible(true);
assertEquals(field.get("const"), Sampler.ALWAYS_SAMPLE);
}
+
+ @Test
+ public void testGetGlobalConfigUsesPluginId() {
+ String pluginId = "config-utils-body-limit-test";
+ GenericGlobalConfig pluginConfig = new GenericGlobalConfig();
+ pluginConfig.setMaxRequestBody(8);
+ pluginConfig.setMaxResponseBody(16);
+ AbstractLogPluginDataHandler.getPluginGlobalConfigMap().put(pluginId,
pluginConfig);
+ try {
+ assertSame(pluginConfig,
LogCollectConfigUtils.getGlobalConfig(pluginId));
+ } finally {
+
AbstractLogPluginDataHandler.getPluginGlobalConfigMap().remove(pluginId);
+ }
+ }
+
+ @Test
+ public void testMissingConfigDoesNotShareMutableDefaults() {
+ String pluginId = "missing-body-limit-test";
+ GenericGlobalConfig missing =
LogCollectConfigUtils.getGlobalConfig(pluginId);
+ assertEquals(524288, missing.getMaxRequestBody());
+ assertEquals(524288, missing.getMaxResponseBody());
+ missing.setMaxRequestBody(1);
+ missing.setMaxResponseBody(1);
+ GenericGlobalConfig next =
LogCollectConfigUtils.getGlobalConfig(pluginId);
+ assertEquals(524288, next.getMaxRequestBody());
+ assertEquals(524288, next.getMaxResponseBody());
+ }
+
+ @Test
+ public void testNullPluginIdUsesDefaults() {
+ GenericGlobalConfig missing =
LogCollectConfigUtils.getGlobalConfig(null);
+ assertEquals(524288, missing.getMaxRequestBody());
+ assertEquals(524288, missing.getMaxResponseBody());
+ }
+
+ @Test
+ public void testLegacySizeChecksUseDefaultLimits() {
+ assertFalse(LogCollectConfigUtils.isRequestBodyTooLarge(524288));
+ assertTrue(LogCollectConfigUtils.isRequestBodyTooLarge(524289));
+ assertFalse(LogCollectConfigUtils.isResponseBodyTooLarge(524288));
+ assertTrue(LogCollectConfigUtils.isResponseBodyTooLarge(524289));
+ }
}
