You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shenyu.apache.org by li...@apache.org on 2026/10/01 04:18:21 UTC
(shenyu) branch master updated: fix(logging): make collector startup idempotent (#7227)
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 891bc5deb9 fix(logging): make collector startup idempotent (#7227)
891bc5deb9 is described below
commit 891bc5deb969591224e2a6416f30ecbc39518116
Author: Liming Deng <li...@gmail.com>
AuthorDate: Thu Oct 1 12:18:14 2026 +0800
fix(logging): make collector startup idempotent (#7227)
---
.../common/collector/AbstractLogCollector.java | 52 ++++++++++---
.../common/collector/AbstractLogCollectorTest.java | 88 +++++++++++++++++++++-
2 files changed, 128 insertions(+), 12 deletions(-)
diff --git a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
index 535022155a..3576996a76 100644
--- a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
+++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/main/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollector.java
@@ -57,9 +57,9 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
private static final Logger LOG = LoggerFactory.getLogger(AbstractLogCollector.class);
- private int bufferSize;
+ private volatile int bufferSize;
- private BlockingQueue<L> bufferQueue;
+ private volatile BlockingQueue<L> bufferQueue;
private final Map<String, BlockingQueue<L>> bufferQueueS = Maps.newConcurrentMap();
@@ -67,12 +67,18 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
private long lastPushTime;
- private final AtomicBoolean started = new AtomicBoolean(true);
+ private final AtomicBoolean started = new AtomicBoolean(false);
+
+ private ShenyuThreadPoolExecutor executor;
@Override
- public void start() {
- bufferSize = getLogCollectConfig().getBufferQueueSize();
- bufferQueue = new LinkedBlockingDeque<>(bufferSize);
+ public synchronized void start() {
+ refreshBufferSize();
+ if (started.get()) {
+ return;
+ }
+ // Admission is bounded in collect(), allowing live capacity changes without dropping queued logs.
+ bufferQueue = new LinkedBlockingDeque<>();
ShenyuConfig config = Optional.ofNullable(Singleton.INST.get(ShenyuConfig.class)).orElse(new ShenyuConfig());
final ShenyuConfig.SharedPool sharedPool = config.getSharedPool();
ShenyuThreadPoolExecutor threadExecutor = new ShenyuThreadPoolExecutor(sharedPool.getCorePoolSize(),
@@ -81,7 +87,22 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
ShenyuThreadFactory.create(config.getSharedPool().getPrefix(), true),
new ThreadPoolExecutor.AbortPolicy());
started.set(true);
- threadExecutor.execute(this::consume);
+ executor = threadExecutor;
+ try {
+ threadExecutor.execute(() -> consume(threadExecutor));
+ } catch (RuntimeException e) {
+ started.set(false);
+ threadExecutor.shutdownNow();
+ throw e;
+ }
+ }
+
+ private void refreshBufferSize() {
+ int configuredSize = getLogCollectConfig().getBufferQueueSize();
+ if (configuredSize <= 0) {
+ throw new IllegalArgumentException("bufferQueueSize must be positive");
+ }
+ bufferSize = configuredSize;
}
@Override
@@ -94,7 +115,13 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
BlockingQueue<L> bufferQueue = bufferQueueS.computeIfAbsent(selectorId, bufferQueueS -> initQueue(selectorId));
bufferQueue.offer(log);
} else {
- bufferQueue.offer(log);
+ BlockingQueue<L> queue = bufferQueue;
+ synchronized (queue) {
+ // On shrink, retain the backlog but reject new logs until it falls below the new limit.
+ if (queue.size() < bufferSize) {
+ queue.offer(log);
+ }
+ }
}
}
@@ -107,8 +134,8 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
/**
* batch and async consume.
*/
- private void consume() {
- while (started.get()) {
+ private void consume(final ThreadPoolExecutor consumerExecutor) {
+ while (!consumerExecutor.isShutdown()) {
int diffTimeMSForPush = 100;
try {
List<L> logs = new ArrayList<>();
@@ -284,8 +311,11 @@ public abstract class AbstractLogCollector<T extends AbstractLogConsumeClient<?,
protected abstract void desensitizeLog(L log, KeyWordMatch keyWordMatch, String desensitizeAlg);
@Override
- public void close() throws Exception {
+ public synchronized void close() throws Exception {
started.set(false);
+ if (Objects.nonNull(executor)) {
+ executor.shutdownNow();
+ }
AbstractLogConsumeClient<?, ?> logCollectClient = getLogConsumeClient();
try {
flushBufferQueues();
diff --git a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollectorTest.java b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollectorTest.java
index 55540e804e..205682125f 100644
--- a/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollectorTest.java
+++ b/shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-common/src/test/java/org/apache/shenyu/plugin/logging/common/collector/AbstractLogCollectorTest.java
@@ -32,18 +32,26 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.stream.IntStream;
+import org.apache.shenyu.common.concurrent.ShenyuThreadPoolExecutor;
+import org.mockito.MockedConstruction;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.inOrder;
@@ -51,6 +59,9 @@ import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.times;
+import static org.mockito.ArgumentMatchers.any;
/**
* The Test Case For AbstractLogCollector.
@@ -59,6 +70,8 @@ public class AbstractLogCollectorTest {
private final AbstractLogConsumeClient<?, ShenyuRequestLog> logConsumeClient = mock(AbstractLogConsumeClient.class);
+ private final GenericGlobalConfig collectorConfig = new GenericGlobalConfig();
+
private final AbstractLogCollector<AbstractLogConsumeClient<?, ShenyuRequestLog>, ShenyuRequestLog, GenericGlobalConfig> collector =
new AbstractLogCollector<>() {
@Override
@@ -68,7 +81,7 @@ public class AbstractLogCollectorTest {
@Override
protected GenericGlobalConfig getLogCollectConfig() {
- return null;
+ return collectorConfig;
}
@Override
@@ -76,6 +89,76 @@ public class AbstractLogCollectorTest {
}
};
+ @Test
+ public void testRepeatedStartAndRestartDoNotLeakConsumers() throws Exception {
+ try {
+ collector.start();
+ ThreadPoolExecutor first = (ThreadPoolExecutor) ReflectionTestUtils.getField(collector, "executor");
+ Object queue = ReflectionTestUtils.getField(collector, "bufferQueue");
+ collector.start();
+ assertSame(first, ReflectionTestUtils.getField(collector, "executor"));
+ assertSame(queue, ReflectionTestUtils.getField(collector, "bufferQueue"));
+ collector.close();
+ collector.start();
+ assertNotSame(first, ReflectionTestUtils.getField(collector, "executor"));
+ assertTrue(first.awaitTermination(2, TimeUnit.SECONDS));
+ } finally {
+ collector.close();
+ }
+ ThreadPoolExecutor last = (ThreadPoolExecutor) ReflectionTestUtils.getField(collector, "executor");
+ assertTrue(last.awaitTermination(2, TimeUnit.SECONDS));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testLiveCapacityChangesPreserveBacklogAndConsumer() throws Exception {
+ try (MockedConstruction<ShenyuThreadPoolExecutor> executors = mockConstruction(ShenyuThreadPoolExecutor.class)) {
+ collectorConfig.setBufferQueueSize(1);
+ collector.start();
+ final BlockingQueue<ShenyuRequestLog> queue = (BlockingQueue<ShenyuRequestLog>) ReflectionTestUtils.getField(collector, "bufferQueue");
+ final ShenyuRequestLog first = new ShenyuRequestLog();
+ final ShenyuRequestLog second = new ShenyuRequestLog();
+ final ShenyuRequestLog third = new ShenyuRequestLog();
+ first.setRequestUri("/first");
+ second.setRequestUri("/second");
+ third.setRequestUri("/third");
+ collector.collect(first);
+ collector.collect(second);
+ assertEquals(List.of(first), new ArrayList<>(queue));
+ collectorConfig.setBufferQueueSize(3);
+ collector.start();
+ collector.collect(second);
+ collector.collect(third);
+ collector.collect(new ShenyuRequestLog());
+ assertEquals(List.of(first, second, third), new ArrayList<>(queue));
+ collectorConfig.setBufferQueueSize(1);
+ collector.start();
+ collector.collect(new ShenyuRequestLog());
+ assertEquals(List.of(first, second, third), new ArrayList<>(queue));
+ queue.drainTo(new ArrayList<>());
+ collector.collect(first);
+ collector.collect(second);
+ assertEquals(List.of(first), new ArrayList<>(queue));
+ assertSame(queue, ReflectionTestUtils.getField(collector, "bufferQueue"));
+ assertEquals(1, executors.constructed().size());
+ verify(executors.constructed().get(0), times(1)).execute(any(Runnable.class));
+ collector.close();
+ }
+ }
+
+ @Test
+ public void testConcurrentProducersRespectLiveCapacity() throws Exception {
+ try (MockedConstruction<ShenyuThreadPoolExecutor> executors = mockConstruction(ShenyuThreadPoolExecutor.class)) {
+ collectorConfig.setBufferQueueSize(10);
+ collector.start();
+ IntStream.range(0, 1000).parallel().forEach(index -> collector.collect(new ShenyuRequestLog()));
+ BlockingQueue<?> queue = (BlockingQueue<?>) ReflectionTestUtils.getField(collector, "bufferQueue");
+ assertEquals(10, queue.size());
+ assertEquals(1, executors.constructed().size());
+ collector.close();
+ }
+ }
+
@Test
public void testSelectorQueueInitializationDoesNotReplaceGlobalQueue() throws Exception {
final GenericGlobalConfig config = new GenericGlobalConfig();
@@ -115,6 +198,7 @@ public class AbstractLogCollectorTest {
@Test
public void testCollectAddsLogWhenBufferQueueHasCapacity() throws Exception {
+ ((AtomicBoolean) ReflectionTestUtils.getField(collector, "started")).set(true);
BlockingQueue<ShenyuRequestLog> bufferQueue = new LinkedBlockingDeque<>(1);
setField(collector, "bufferSize", 1);
setField(collector, "bufferQueue", bufferQueue);
@@ -127,6 +211,7 @@ public class AbstractLogCollectorTest {
@Test
public void testCollectDoesNotThrowWhenBufferQueueIsFull() throws Exception {
+ ((AtomicBoolean) ReflectionTestUtils.getField(collector, "started")).set(true);
ShenyuRequestLog bufferedLog = new ShenyuRequestLog();
BlockingQueue<ShenyuRequestLog> bufferQueue = new StaleSizeLinkedBlockingDeque();
bufferQueue.add(bufferedLog);
@@ -164,6 +249,7 @@ public class AbstractLogCollectorTest {
BlockingQueue<ShenyuRequestLog> bufferQueue = new StaleSizeLinkedBlockingDeque();
bufferQueue.add(bufferedLog);
setField(multiClientCollector, "bufferSize", 1);
+ ((AtomicBoolean) ReflectionTestUtils.getField(multiClientCollector, "started")).set(true);
getBufferQueues(multiClientCollector).put("selector", bufferQueue);
ShenyuRequestLog log = new ShenyuRequestLog();
log.setSelectorId("selector");