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 1797bd3729 fix: add lightweight circuit breaker to divide retry path
(#7507)
1797bd3729 is described below
commit 1797bd3729b90156664d45bb68f4700e49f7a078
Author: wy471x <[email protected]>
AuthorDate: Sat Oct 10 13:51:30 2026 +0800
fix: add lightweight circuit breaker to divide retry path (#7507)
Co-authored-by: wy471x <[email protected]>
---
.../circuitbreaker/UpstreamCircuitBreaker.java | 274 +++++++++++++++++++++
.../circuitbreaker/UpstreamCircuitBreakerTest.java | 209 ++++++++++++++++
.../plugin/httpclient/DefaultRetryStrategy.java | 13 +-
.../plugin/httpclient/RetryStrategyTest.java | 83 +++++++
.../apache/shenyu/plugin/divide/DividePlugin.java | 78 +++++-
.../divide/DividePluginCircuitBreakerTest.java | 181 ++++++++++++++
6 files changed, 824 insertions(+), 14 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
new file mode 100644
index 0000000000..18c3c40918
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/main/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreaker.java
@@ -0,0 +1,274 @@
+/*
+ * 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.base.circuitbreaker;
+
+import org.apache.shenyu.loadbalancer.entity.Upstream;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * A lightweight built-in circuit breaker for divide/http proxy upstreams.
+ *
+ * <p>State is tracked per selector-upstream pair. When an upstream accumulates
+ * {@code failureThreshold} consecutive failures, its breaker opens and further
+ * requests fail fast without sending any traffic to the upstream. After
+ * {@code halfOpenWaitMillis} the breaker allows a single half-open probe
+ * request; if the probe succeeds the breaker closes again, otherwise it
+ * opens for another wait window.
+ *
+ * <p>This breaker is enabled by default and is fully independent from the
+ * optional resilience4j plugin: when no failure occurs it only performs a few
+ * map lookups, so the overhead on the happy path is negligible.
+ */
+public final class UpstreamCircuitBreaker {
+
+ /**
+ * The default number of consecutive failures before the breaker opens.
+ */
+ public static final int DEFAULT_FAILURE_THRESHOLD = 3;
+
+ /**
+ * The default wait time in milliseconds before a half-open probe is
allowed.
+ */
+ public static final long DEFAULT_HALF_OPEN_WAIT_MILLIS = 10000L;
+
+ private static final UpstreamCircuitBreaker INSTANCE =
+ new UpstreamCircuitBreaker(DEFAULT_FAILURE_THRESHOLD,
DEFAULT_HALF_OPEN_WAIT_MILLIS);
+
+ private final int failureThreshold;
+
+ private final long halfOpenWaitMillis;
+
+ private final ConcurrentMap<String, CircuitBreakerState> breakers;
+
+ private UpstreamCircuitBreaker(final int failureThreshold, final long
halfOpenWaitMillis) {
+ this.failureThreshold = failureThreshold;
+ this.halfOpenWaitMillis = halfOpenWaitMillis;
+ this.breakers = new ConcurrentHashMap<>(16);
+ }
+
+ /**
+ * Get the singleton instance used by the gateway plugins.
+ *
+ * @return the shared {@link UpstreamCircuitBreaker} instance
+ */
+ public static UpstreamCircuitBreaker getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Create a standalone breaker with custom thresholds, mainly for testing.
+ *
+ * @param failureThreshold consecutive failures required to open the
breaker
+ * @param halfOpenWaitMillis wait time in milliseconds before a half-open
probe is allowed
+ * @return a new {@link UpstreamCircuitBreaker} instance
+ */
+ public static UpstreamCircuitBreaker create(final int failureThreshold,
final long halfOpenWaitMillis) {
+ return new UpstreamCircuitBreaker(failureThreshold,
halfOpenWaitMillis);
+ }
+
+ /**
+ * Build the breaker key for the given selector and upstream.
+ *
+ * @param selectorId the selector id the upstream belongs to
+ * @param upstream the upstream
+ * @return the breaker key
+ */
+ public static String buildKey(final String selectorId, final Upstream
upstream) {
+ return selectorId + ":" + upstream.buildDomain();
+ }
+
+ /**
+ * Whether a request to the given upstream is currently allowed.
+ *
+ * <p>Requests are allowed when the breaker is closed, half-open, or open
+ * but already past its half-open wait window (i.e. eligible to be probed).
+ *
+ * @param key the breaker key
+ * @return true if a request may be sent to the upstream
+ */
+ public boolean isRequestAllowed(final String key) {
+ CircuitBreakerState state = this.breakers.get(key);
+ if (Objects.isNull(state)) {
+ return true;
+ }
+ return state.isRequestAllowed();
+ }
+
+ /**
+ * Atomically move an open breaker past its wait window into the half-open
+ * state and grant a single probe request.
+ *
+ * @param key the breaker key
+ * @return true if the caller is granted the probe request
+ */
+ public boolean tryAcquireHalfOpenProbe(final String key) {
+ CircuitBreakerState state = this.breakers.get(key);
+ if (Objects.isNull(state)) {
+ return false;
+ }
+ return state.tryAcquireHalfOpen();
+ }
+
+ /**
+ * Record a successful request, closing the breaker and clearing failures.
+ *
+ * @param key the breaker key
+ */
+ public void recordSuccess(final String key) {
+ CircuitBreakerState state = this.breakers.get(key);
+ if (Objects.nonNull(state)) {
+ state.recordSuccess();
+ }
+ }
+
+ /**
+ * Record a failed request; the breaker opens once the consecutive failure
+ * threshold is reached or a half-open probe fails.
+ *
+ * @param key the breaker key
+ */
+ public void recordFailure(final String key) {
+ this.breakers.computeIfAbsent(key, k -> new
CircuitBreakerState(this.failureThreshold, this.halfOpenWaitMillis))
+ .recordFailure();
+ }
+
+ /**
+ * Whether the breaker currently blocks requests for the given key.
+ *
+ * @param key the breaker key
+ * @return true if the breaker is open and still within its wait window
+ */
+ public boolean isBlocking(final String key) {
+ CircuitBreakerState state = this.breakers.get(key);
+ if (Objects.isNull(state)) {
+ return false;
+ }
+ return state.isBlocking();
+ }
+
+ /**
+ * Remove the state for the given key, mainly for testing.
+ *
+ * @param key the breaker key
+ */
+ public void reset(final String key) {
+ this.breakers.remove(key);
+ }
+
+ /**
+ * Wrap a forwarded {@link Mono} so that its terminal signal is recorded as
+ * a success or a failure of the upstream identified by the given key.
+ *
+ * @param <T> the element type of the source mono
+ * @param source the mono executing the upstream call
+ * @param key the breaker key
+ * @return the wrapped mono with outcome recording attached
+ */
+ public static <T> Mono<T> recordOutcome(final Mono<T> source, final String
key) {
+ return source.doFinally(signal -> {
+ if (signal == SignalType.ON_COMPLETE) {
+ INSTANCE.recordSuccess(key);
+ } else if (signal == SignalType.ON_ERROR) {
+ INSTANCE.recordFailure(key);
+ }
+ });
+ }
+
+ /**
+ * Mutable state of a single breaker.
+ */
+ private static final class CircuitBreakerState {
+
+ private enum State {
+ /**
+ * Requests flow normally.
+ */
+ CLOSED,
+ /**
+ * Requests fail fast until the wait window elapses.
+ */
+ OPEN,
+ /**
+ * A single probe request is in flight.
+ */
+ HALF_OPEN
+ }
+
+ private final int failureThreshold;
+
+ private final long halfOpenWaitMillis;
+
+ private State state;
+
+ private int consecutiveFailures;
+
+ private long openedAtMillis;
+
+ CircuitBreakerState(final int failureThreshold, final long
halfOpenWaitMillis) {
+ this.failureThreshold = failureThreshold;
+ this.halfOpenWaitMillis = halfOpenWaitMillis;
+ this.state = State.CLOSED;
+ }
+
+ synchronized boolean isRequestAllowed() {
+ if (this.state == State.OPEN) {
+ return System.currentTimeMillis() - this.openedAtMillis >=
this.halfOpenWaitMillis;
+ }
+ return true;
+ }
+
+ synchronized boolean isBlocking() {
+ return !isRequestAllowed();
+ }
+
+ synchronized boolean tryAcquireHalfOpen() {
+ if (this.state == State.OPEN
+ && System.currentTimeMillis() - this.openedAtMillis >=
this.halfOpenWaitMillis) {
+ this.state = State.HALF_OPEN;
+ return true;
+ }
+ return false;
+ }
+
+ synchronized void recordSuccess() {
+ this.state = State.CLOSED;
+ this.consecutiveFailures = 0;
+ }
+
+ synchronized void recordFailure() {
+ if (this.state == State.HALF_OPEN) {
+ open();
+ return;
+ }
+ this.consecutiveFailures++;
+ if (this.consecutiveFailures >= this.failureThreshold) {
+ open();
+ }
+ }
+
+ private void open() {
+ this.state = State.OPEN;
+ this.openedAtMillis = System.currentTimeMillis();
+ }
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
new file mode 100644
index 0000000000..35b3c718c3
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-base/src/test/java/org/apache/shenyu/plugin/base/circuitbreaker/UpstreamCircuitBreakerTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.base.circuitbreaker;
+
+import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Test for {@link UpstreamCircuitBreaker}.
+ */
+public final class UpstreamCircuitBreakerTest {
+
+ private static final String KEY = "selector:http://127.0.0.1:9090";;
+
+ @Test
+ public void testRequestsAllowedByDefault() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(3,
100L);
+ assertTrue(breaker.isRequestAllowed(KEY));
+ assertFalse(breaker.isBlocking(KEY));
+ assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
+ }
+
+ @Test
+ public void testOpensAfterConsecutiveFailures() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(3,
100L);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ assertFalse(breaker.isRequestAllowed(KEY));
+ }
+
+ @Test
+ public void testSuccessResetsConsecutiveFailures() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(3,
100L);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ breaker.recordSuccess(KEY);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testSuccessClosesOpenBreaker() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(1,
100000L);
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ breaker.recordSuccess(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ assertTrue(breaker.isRequestAllowed(KEY));
+ }
+
+ @Test
+ public void testHalfOpenProbeAllowedAfterWaitWindow() throws
InterruptedException {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(1,
100L);
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ // within the wait window no probe is granted
+ assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
+ TimeUnit.MILLISECONDS.sleep(150L);
+ // the first caller is granted the single probe
+ assertTrue(breaker.tryAcquireHalfOpenProbe(KEY));
+ // no second probe while the first one is in flight
+ assertFalse(breaker.tryAcquireHalfOpenProbe(KEY));
+ // a failed probe re-opens the breaker
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testSuccessfulProbeClosesBreaker() throws InterruptedException
{
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(2,
100L);
+ breaker.recordFailure(KEY);
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ TimeUnit.MILLISECONDS.sleep(150L);
+ assertTrue(breaker.tryAcquireHalfOpenProbe(KEY));
+ breaker.recordSuccess(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ // breaker fully recovered, a single new failure stays below the
threshold
+ breaker.recordFailure(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testConcurrentFailuresOpenBreaker() throws
InterruptedException {
+ final int threads = 32;
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(3,
100L);
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ for (int i = 0; i < threads; i++) {
+ executor.execute(() -> {
+ try {
+ start.await();
+ breaker.recordFailure(KEY);
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ assertTrue(done.await(10, TimeUnit.SECONDS));
+ executor.shutdownNow();
+ assertTrue(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testOnlyOneConcurrentHalfOpenProbeGranted() throws
InterruptedException {
+ final int threads = 16;
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(1, 50L);
+ breaker.recordFailure(KEY);
+ TimeUnit.MILLISECONDS.sleep(100L);
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ AtomicInteger granted = new AtomicInteger();
+ for (int i = 0; i < threads; i++) {
+ executor.execute(() -> {
+ try {
+ start.await();
+ if (breaker.tryAcquireHalfOpenProbe(KEY)) {
+ granted.incrementAndGet();
+ }
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ assertTrue(done.await(10, TimeUnit.SECONDS));
+ executor.shutdownNow();
+ assertEquals(1, granted.get());
+ }
+
+ @Test
+ public void testRecordOutcomeRecordsTerminalSignals() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(2,
100L);
+
StepVerifier.create(UpstreamCircuitBreaker.recordOutcome(Mono.error(new
IllegalStateException("upstream down")), KEY))
+ .expectError(IllegalStateException.class)
+ .verify();
+
StepVerifier.create(UpstreamCircuitBreaker.recordOutcome(Mono.just("ok"), KEY))
+ .expectNext("ok")
+ .verifyComplete();
+ // one failure recorded and then reset by the success, so one more
failure keeps the breaker closed
+ breaker.recordFailure(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testBuildKeyUsesSelectorIdAndDomain() {
+ Upstream upstream =
Upstream.builder().protocol("http://";).url("127.0.0.1:9090").build();
+ assertEquals("selector:http://127.0.0.1:9090";,
UpstreamCircuitBreaker.buildKey("selector", upstream));
+ }
+
+ @Test
+ public void testResetClearsState() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(1,
100000L);
+ breaker.recordFailure(KEY);
+ assertTrue(breaker.isBlocking(KEY));
+ breaker.reset(KEY);
+ assertFalse(breaker.isBlocking(KEY));
+ }
+
+ @Test
+ public void testMultipleKeysAreIndependent() {
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.create(1,
100000L);
+ List<String> keys = List.of("selector-a:http://a:8080";,
"selector-b:http://b:8080";);
+ breaker.recordFailure(keys.get(0));
+ assertTrue(breaker.isBlocking(keys.get(0)));
+ assertFalse(breaker.isBlocking(keys.get(1)));
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
index 79b6727dfd..5ab621c1ad 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/main/java/org/apache/shenyu/plugin/httpclient/DefaultRetryStrategy.java
@@ -24,6 +24,7 @@ import org.apache.shenyu.common.utils.UriUtils;
import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
import org.apache.shenyu.loadbalancer.entity.Upstream;
import org.apache.shenyu.plugin.api.utils.RequestUrlUtils;
+import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
import org.apache.shenyu.plugin.base.utils.LoadbalancerUtils;
import org.apache.shenyu.plugin.httpclient.exception.ShenyuTimeoutException;
import org.slf4j.Logger;
@@ -120,7 +121,11 @@ public class DefaultRetryStrategy<R> implements
RetryStrategy<R> {
}
}
return true;
- }).collect(Collectors.toList());
+ })
+ //skip upstreams that are still blocked by the built-in
circuit breaker
+ .filter(data -> UpstreamCircuitBreaker.getInstance()
+
.isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId, data)))
+ .collect(Collectors.toList());
if (upstreamList.isEmpty()) {
// no need to retry anymore
return Mono.error(new
ShenyuException("CANNOT_FIND_HEALTHY_UPSTREAM_URL_AFTER_FAILOVER"));
@@ -133,9 +138,11 @@ public class DefaultRetryStrategy<R> implements
RetryStrategy<R> {
final URI newUri = RequestUrlUtils.buildRequestUri(exchange,
upstream.buildDomain());
// in order not to affect the next retry call, newUri needs to be
excluded
exclude.add(newUri);
- return httpClientPlugin.doRequest(exchange,
exchange.getRequest().getMethod().name(), newUri,
httpClientPlugin.getCachedRequestBody(exchange))
+ final String breakerKey =
UpstreamCircuitBreaker.buildKey(selectorId, upstream);
+ return UpstreamCircuitBreaker.recordOutcome(httpClientPlugin
+ .doRequest(exchange,
exchange.getRequest().getMethod().name(), newUri,
httpClientPlugin.getCachedRequestBody(exchange))
.timeout(duration, Mono.error(() -> new
TimeoutException("Response took longer than timeout: " + duration)))
- .doOnError(e -> LOG.error(e.getMessage(), e));
+ .doOnError(e -> LOG.error(e.getMessage(), e)), breakerKey);
});
}
diff --git
a/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
index 7b8e7e71ba..efea42791f 100644
---
a/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
+++
b/shenyu-plugin/shenyu-plugin-httpclient/src/test/java/org/apache/shenyu/plugin/httpclient/RetryStrategyTest.java
@@ -22,6 +22,7 @@ import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.enums.RetryEnum;
import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
@@ -44,7 +45,9 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -158,6 +161,86 @@ public class RetryStrategyTest {
assertEquals(URI.create(standby.buildDomain() + "/test?mode=retry"),
uriCaptor.getValue());
}
+ @Test
+ void testFailoverSkipsBlockedUpstream() {
+ AbstractHttpClientPlugin<String> plugin =
mock(AbstractHttpClientPlugin.class);
+ ServerWebExchange exchange =
createFailoverExchange("http://localhost:8080/test?mode=retry";);
+ Upstream failed = Upstream.builder().url("localhost:8080").build();
+ Upstream standby =
Upstream.builder().protocol("http://";).url("localhost:8081").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId("selector-7475")).thenReturn(Arrays.asList(failed,
standby));
+ final String standbyKey =
UpstreamCircuitBreaker.buildKey("selector-7475", standby);
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.getInstance();
+ for (int i = 0; i < 3; i++) {
+ breaker.recordFailure(standbyKey);
+ }
+ try (MockedStatic<UpstreamCacheManager> cacheMock =
mockStatic(UpstreamCacheManager.class)) {
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ StepVerifier.create(new
DefaultRetryStrategy<>(plugin).execute(Mono.error(new
TimeoutException("upstream failed")),
+ exchange, Duration.ofSeconds(5), 2))
+ .expectErrorSatisfies(this::assertFailoverExhausted)
+ .verify();
+ } finally {
+ breaker.reset(standbyKey);
+ }
+ verifyNoInteractions(plugin);
+ }
+
+ @Test
+ void testFailoverResendRecordsFailureOutcome() {
+ AbstractHttpClientPlugin<String> plugin =
mock(AbstractHttpClientPlugin.class);
+ ServerWebExchange exchange =
createFailoverExchange("http://localhost:8080/test?mode=retry";);
+ Upstream failed = Upstream.builder().url("localhost:8080").build();
+ Upstream standby =
Upstream.builder().protocol("http://";).url("localhost:8081").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId("selector-7475")).thenReturn(Arrays.asList(failed,
standby));
+ when(plugin.getCachedRequestBody(exchange)).thenReturn(Flux.empty());
+ when(plugin.doRequest(eq(exchange), eq("GET"), any(URI.class),
any())).thenReturn(Mono.error(new TimeoutException("resend failed")));
+ final String standbyKey =
UpstreamCircuitBreaker.buildKey("selector-7475", standby);
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.getInstance();
+ try (MockedStatic<UpstreamCacheManager> cacheMock =
mockStatic(UpstreamCacheManager.class)) {
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ StepVerifier.create(new
DefaultRetryStrategy<>(plugin).execute(Mono.error(new
TimeoutException("upstream failed")),
+ exchange, Duration.ofSeconds(5), 2))
+ .expectErrorSatisfies(this::assertFailoverExhausted)
+ .verify();
+ }
+ // the failed resend recorded one failure, one more below the
threshold keeps the breaker closed
+ breaker.recordFailure(standbyKey);
+ assertFalse(breaker.isBlocking(standbyKey));
+ breaker.recordFailure(standbyKey);
+ assertTrue(breaker.isBlocking(standbyKey));
+ breaker.reset(standbyKey);
+ }
+
+ @Test
+ void testFailoverResendRecordsSuccessOutcome() {
+ AbstractHttpClientPlugin<String> plugin =
mock(AbstractHttpClientPlugin.class);
+ ServerWebExchange exchange =
createFailoverExchange("http://localhost:8080/test?mode=retry";);
+ Upstream failed = Upstream.builder().url("localhost:8080").build();
+ Upstream standby =
Upstream.builder().protocol("http://";).url("localhost:8081").build();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId("selector-7475")).thenReturn(Arrays.asList(failed,
standby));
+ when(plugin.getCachedRequestBody(exchange)).thenReturn(Flux.empty());
+ when(plugin.doRequest(eq(exchange), eq("GET"), any(URI.class),
any())).thenReturn(Mono.just("success"));
+ final String standbyKey =
UpstreamCircuitBreaker.buildKey("selector-7475", standby);
+ UpstreamCircuitBreaker breaker = UpstreamCircuitBreaker.getInstance();
+ breaker.recordFailure(standbyKey);
+ breaker.recordFailure(standbyKey);
+ try (MockedStatic<UpstreamCacheManager> cacheMock =
mockStatic(UpstreamCacheManager.class)) {
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ StepVerifier.create(new
DefaultRetryStrategy<>(plugin).execute(Mono.error(new
TimeoutException("upstream failed")),
+ exchange, Duration.ofSeconds(5), 2))
+ .expectNext("success")
+ .verifyComplete();
+ }
+ // the successful resend reset the failure counter, so two more
failures stay below the threshold
+ breaker.recordFailure(standbyKey);
+ breaker.recordFailure(standbyKey);
+ assertFalse(breaker.isBlocking(standbyKey));
+ breaker.reset(standbyKey);
+ }
+
@Test
void testFixedRetryStrategyRetriesTransientGetFailures() {
AbstractHttpClientPlugin<String> httpClientPlugin =
mock(AbstractHttpClientPlugin.class);
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
index 0c55ed7250..ef8deefe99 100644
---
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/main/java/org/apache/shenyu/plugin/divide/DividePlugin.java
@@ -37,6 +37,7 @@ import org.apache.shenyu.plugin.api.result.ShenyuResultWrap;
import org.apache.shenyu.plugin.api.utils.RequestUrlUtils;
import org.apache.shenyu.plugin.api.utils.WebFluxResultUtils;
import org.apache.shenyu.plugin.base.AbstractShenyuPlugin;
+import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
import org.apache.shenyu.plugin.base.utils.CacheKeyUtils;
import org.apache.shenyu.plugin.base.utils.LoadbalancerUtils;
import org.apache.shenyu.plugin.divide.handler.DividePluginDataHandler;
@@ -46,7 +47,9 @@ import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -96,11 +99,17 @@ public class DividePlugin extends AbstractShenyuPlugin {
Object error = ShenyuResultWrap.error(exchange,
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL);
return WebFluxResultUtils.result(exchange, error);
}
+ final List<Upstream> breakeredUpstreamList =
filterByCircuitBreaker(selector.getId(), upstreamList);
+ if (CollectionUtils.isEmpty(breakeredUpstreamList)) {
+ LOG.warn("all upstreams of selector {} are blocked by the circuit
breaker, fail fast", selector.getId());
+ Object error = ShenyuResultWrap.error(exchange,
ShenyuResultEnum.CANNOT_FIND_HEALTHY_UPSTREAM_URL);
+ return WebFluxResultUtils.result(exchange, error);
+ }
List<String> specifyDomains =
exchange.getRequest().getHeaders().get(Constants.SPECIFY_DOMAIN);
Upstream upstream;
if (CollectionUtils.isNotEmpty(specifyDomains)) {
String requested = specifyDomains.get(0);
- upstream = upstreamList.stream()
+ upstream = breakeredUpstreamList.stream()
.filter(u -> u.getUrl().equals(requested))
.findFirst()
.map(u -> Upstream.builder()
@@ -110,9 +119,9 @@ public class DividePlugin extends AbstractShenyuPlugin {
.warmup(u.getWarmup())
.status(u.isStatus())
.build())
- .orElseGet(() ->
LoadbalancerUtils.getForExchange(upstreamList, ruleHandle.getLoadBalance(),
exchange));
+ .orElseGet(() ->
LoadbalancerUtils.getForExchange(breakeredUpstreamList,
ruleHandle.getLoadBalance(), exchange));
} else {
- upstream = LoadbalancerUtils.getForExchange(upstreamList,
ruleHandle.getLoadBalance(), exchange);
+ upstream = LoadbalancerUtils.getForExchange(breakeredUpstreamList,
ruleHandle.getLoadBalance(), exchange);
}
if (Objects.isNull(upstream)) {
LOG.error("divide has no upstream");
@@ -130,14 +139,7 @@ public class DividePlugin extends AbstractShenyuPlugin {
exchange.getAttributes().put(Constants.RETRY_STRATEGY,
StringUtils.defaultIfEmpty(ruleHandle.getRetryStrategy(),
RetryEnum.CURRENT.getName()));
exchange.getAttributes().put(Constants.LOAD_BALANCE,
StringUtils.defaultIfEmpty(ruleHandle.getLoadBalance(),
LoadBalanceEnum.RANDOM.getName()));
exchange.getAttributes().put(Constants.DIVIDE_SELECTOR_ID,
selector.getId());
- if (ruleHandle.getLoadBalance().equals(P2C)) {
- return chain.execute(exchange).doFinally(signalType ->
responseTrigger(upstream));
- } else if (ruleHandle.getLoadBalance().equals(SHORTEST_RESPONSE)) {
- long beginTime = System.currentTimeMillis();
- return chain.execute(exchange).doOnSuccess(e ->
successResponseTrigger(upstream, beginTime
- ));
- }
- return chain.execute(exchange);
+ return forwardWithCircuitBreaker(exchange, chain, ruleHandle,
selector.getId(), upstream);
}
@Override
@@ -165,10 +167,64 @@ public class DividePlugin extends AbstractShenyuPlugin {
return WebFluxResultUtils.noRuleResult(pluginName, exchange);
}
+ /**
+ * Forward the request through the plugin chain and record its outcome into
+ * the built-in circuit breaker, keeping the load-balance specific
triggers.
+ *
+ * @param exchange the current server exchange
+ * @param chain the plugin chain
+ * @param ruleHandle the divide rule handle
+ * @param selectorId the selector id
+ * @param upstream the chosen upstream
+ * @return {@code Mono<Void>} to indicate when request processing is
complete
+ */
+ private Mono<Void> forwardWithCircuitBreaker(final ServerWebExchange
exchange, final ShenyuPluginChain chain,
+ final DivideRuleHandle
ruleHandle, final String selectorId, final Upstream upstream) {
+ String breakerKey = UpstreamCircuitBreaker.buildKey(selectorId,
upstream);
+ if (ruleHandle.getLoadBalance().equals(P2C)) {
+ return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange)
+ .doFinally(signalType -> responseTrigger(upstream)),
breakerKey);
+ } else if (ruleHandle.getLoadBalance().equals(SHORTEST_RESPONSE)) {
+ long beginTime = System.currentTimeMillis();
+ return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange)
+ .doOnSuccess(e -> successResponseTrigger(upstream,
beginTime)), breakerKey);
+ }
+ return UpstreamCircuitBreaker.recordOutcome(chain.execute(exchange),
breakerKey);
+ }
+
private DivideRuleHandle buildRuleHandle(final RuleData rule) {
return
DividePluginDataHandler.CACHED_HANDLE.get().obtainHandle(CacheKeyUtils.INST.getKey(rule));
}
+ /**
+ * Filter out upstreams blocked by the built-in circuit breaker.
+ *
+ * <p>When every upstream of the selector is blocked, a single half-open
+ * probe request is granted to one of them so the breaker can recover;
+ * otherwise the caller should fail fast.
+ *
+ * @param selectorId the selector id the upstreams belong to
+ * @param upstreamList the healthy upstream list from the cache
+ * @return the upstreams allowed to receive requests, possibly empty
+ */
+ private List<Upstream> filterByCircuitBreaker(final String selectorId,
final List<Upstream> upstreamList) {
+ List<Upstream> allowed = new ArrayList<>(upstreamList.size());
+ for (Upstream upstream : upstreamList) {
+ if
(UpstreamCircuitBreaker.getInstance().isRequestAllowed(UpstreamCircuitBreaker.buildKey(selectorId,
upstream))) {
+ allowed.add(upstream);
+ }
+ }
+ if (CollectionUtils.isNotEmpty(allowed)) {
+ return allowed;
+ }
+ for (Upstream upstream : upstreamList) {
+ if
(UpstreamCircuitBreaker.getInstance().tryAcquireHalfOpenProbe(UpstreamCircuitBreaker.buildKey(selectorId,
upstream))) {
+ return Collections.singletonList(upstream);
+ }
+ }
+ return Collections.emptyList();
+ }
+
private void responseTrigger(final Upstream upstream) {
long now = System.currentTimeMillis();
upstream.getInflight().decrementAndGet();
diff --git
a/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
new file mode 100644
index 0000000000..69e67575e8
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-proxy/shenyu-plugin-divide/src/test/java/org/apache/shenyu/plugin/divide/DividePluginCircuitBreakerTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.divide;
+
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.dto.RuleData;
+import org.apache.shenyu.common.dto.SelectorData;
+import org.apache.shenyu.common.dto.convert.rule.impl.DivideRuleHandle;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.loadbalancer.cache.UpstreamCacheManager;
+import org.apache.shenyu.loadbalancer.entity.Upstream;
+import org.apache.shenyu.plugin.api.ShenyuPluginChain;
+import org.apache.shenyu.plugin.api.context.ShenyuContext;
+import org.apache.shenyu.plugin.api.result.DefaultShenyuResult;
+import org.apache.shenyu.plugin.api.result.ShenyuResult;
+import org.apache.shenyu.plugin.api.utils.SpringBeanUtils;
+import org.apache.shenyu.plugin.base.circuitbreaker.UpstreamCircuitBreaker;
+import org.apache.shenyu.plugin.divide.handler.DividePluginDataHandler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.springframework.context.ConfigurableApplicationContext;
+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.Mono;
+import reactor.test.StepVerifier;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test for the built-in circuit breaker behaviour of {@link DividePlugin}.
+ */
+public final class DividePluginCircuitBreakerTest {
+
+ private static final String SELECTOR_ID = "breaker-selector";
+
+ private static final String UPSTREAM_URL = "127.0.0.1:9090";
+
+ private DividePlugin dividePlugin;
+
+ private ShenyuPluginChain chain;
+
+ private ServerWebExchange exchange;
+
+ private SelectorData selectorData;
+
+ private RuleData ruleData;
+
+ private Upstream upstream;
+
+ private String breakerKey;
+
+ private MockedStatic<UpstreamCacheManager> cacheMock;
+
+ @BeforeEach
+ public void setup() {
+ this.dividePlugin = new DividePlugin();
+ this.chain = mock(ShenyuPluginChain.class);
+ this.selectorData = mock(SelectorData.class);
+ this.ruleData = mock(RuleData.class);
+ this.exchange =
MockServerWebExchange.from(MockServerHttpRequest.get("http://localhost/test";).build());
+ ShenyuContext context = mock(ShenyuContext.class);
+ exchange.getAttributes().put(Constants.CONTEXT, context);
+ this.upstream =
Upstream.builder().protocol("http://";).url(UPSTREAM_URL).build();
+ this.breakerKey = UpstreamCircuitBreaker.buildKey(SELECTOR_ID,
upstream);
+ when(selectorData.getId()).thenReturn(SELECTOR_ID);
+ when(ruleData.getHandle()).thenReturn(GsonUtils.getGson().toJson(new
DivideRuleHandle()));
+ DividePluginDataHandler dividePluginDataHandler = new
DividePluginDataHandler();
+ dividePluginDataHandler.handlerRule(ruleData);
+ dividePluginDataHandler.handlerSelector(selectorData);
+ ConfigurableApplicationContext applicationContext =
mock(ConfigurableApplicationContext.class);
+
SpringBeanUtils.getInstance().setApplicationContext(applicationContext);
+ when(applicationContext.getBean(ShenyuResult.class)).thenReturn(new
DefaultShenyuResult());
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+
when(cacheManager.findUpstreamListBySelectorId(SELECTOR_ID)).thenReturn(Collections.singletonList(upstream));
+ this.cacheMock = mockStatic(UpstreamCacheManager.class);
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ cacheMock.close();
+ UpstreamCircuitBreaker.getInstance().reset(breakerKey);
+ }
+
+ @Test
+ public void testFailFastWhenAllUpstreamsBlocked() {
+ openBreaker();
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+ verify(chain, times(0)).execute(any());
+ }
+
+ @Test
+ public void testConsecutiveFailuresOpenBreakerThenFailFast() {
+ when(chain.execute(exchange)).thenReturn(Mono.error(new
IllegalStateException("upstream down")));
+ for (int i = 0; i < 3; i++) {
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .expectError(IllegalStateException.class)
+ .verify();
+ }
+
assertTrue(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
+ // the next request fails fast without reaching the chain any more
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+ verify(chain, times(3)).execute(any());
+ }
+
+ @Test
+ public void testHalfOpenProbeRecoversBreaker() throws Exception {
+ openBreaker();
+ // rewind the open timestamp so the half-open wait window has elapsed
+ Field breakersField =
UpstreamCircuitBreaker.class.getDeclaredField("breakers");
+ breakersField.setAccessible(true);
+ ConcurrentMap<String, ?> breakers = (ConcurrentMap<String, ?>)
breakersField.get(UpstreamCircuitBreaker.getInstance());
+ Object state = breakers.get(breakerKey);
+ Field openedAtField =
state.getClass().getDeclaredField("openedAtMillis");
+ openedAtField.setAccessible(true);
+ openedAtField.setLong(state, System.currentTimeMillis() - 60000L);
+ // the probe request is forwarded and its success closes the breaker
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+ verify(chain, times(1)).execute(any());
+
assertFalse(UpstreamCircuitBreaker.getInstance().isBlocking(breakerKey));
+ }
+
+ @Test
+ public void testHealthyUpstreamsBypassBreaker() {
+ Upstream healthy =
Upstream.builder().protocol("http://";).url("127.0.0.1:9091").build();
+
UpstreamCircuitBreaker.getInstance().reset(UpstreamCircuitBreaker.buildKey(SELECTOR_ID,
healthy));
+ openBreaker();
+ UpstreamCacheManager cacheManager = mock(UpstreamCacheManager.class);
+ List<Upstream> upstreams = List.of(upstream, healthy);
+
when(cacheManager.findUpstreamListBySelectorId(SELECTOR_ID)).thenReturn(upstreams);
+
cacheMock.when(UpstreamCacheManager::getInstance).thenReturn(cacheManager);
+ // the blocked upstream is filtered out, the request still goes
through the healthy one
+ when(chain.execute(exchange)).thenReturn(Mono.empty());
+ StepVerifier.create(dividePlugin.doExecute(exchange, chain,
selectorData, ruleData))
+ .verifyComplete();
+ verify(chain, times(1)).execute(any());
+
UpstreamCircuitBreaker.getInstance().reset(UpstreamCircuitBreaker.buildKey(SELECTOR_ID,
healthy));
+ }
+
+ private void openBreaker() {
+ for (int i = 0; i < 3; i++) {
+ UpstreamCircuitBreaker.getInstance().recordFailure(breakerKey);
+ }
+ }
+}
