diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java index 358c3158..90c18cc8 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java @@ -17,57 +17,114 @@ package org.springframework.cloud.gateway.filter.factory; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.function.Predicate; +import java.util.logging.Level; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Mono; import reactor.retry.Repeat; import reactor.retry.RepeatContext; +import reactor.retry.Retry; +import reactor.retry.RetryContext; import org.springframework.cloud.gateway.filter.GatewayFilter; +import org.springframework.cloud.gateway.support.ServerWebExchangeUtils; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; import org.springframework.http.HttpStatus.Series; import org.springframework.util.Assert; import org.springframework.web.server.ServerWebExchange; -public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory { +public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory { + private static final Log log = LogFactory.getLog(RetryGatewayFilterFactory.class); public RetryGatewayFilterFactory() { - super(Retry.class); + super(RetryConfig.class); } @Override - public GatewayFilter apply(Retry retry) { - retry.validate(); + public GatewayFilter apply(RetryConfig retryConfig) { + retryConfig.validate(); Predicate> predicate = context -> { ServerWebExchange exchange = context.applicationContext(); + if (exceedsMaxIterations(exchange, retryConfig)) { + return false; + } + HttpStatus statusCode = exchange.getResponse().getStatusCode(); HttpMethod httpMethod = exchange.getRequest().getMethod(); - boolean retryableStatusCode = retry.getStatuses().contains(statusCode); + boolean retryableStatusCode = retryConfig.getStatuses().contains(statusCode); - if (!retryableStatusCode) { + if (!retryableStatusCode && statusCode != null) { // null status code might mean a network exception? // try the series - retryableStatusCode = retry.getSeries().stream() + retryableStatusCode = retryConfig.getSeries().stream() .anyMatch(series -> statusCode.series().equals(series)); } - boolean retryableMethod = retry.getMethods().contains(httpMethod); + boolean retryableMethod = retryConfig.getMethods().contains(httpMethod); return retryableMethod && retryableStatusCode; }; - Repeat repeat = Repeat.create(predicate, retry.getRetries()); + Repeat repeat = Repeat.onlyIf(predicate) + .doOnRepeat(context -> reset(context.applicationContext())); //TODO: support timeout, backoff, jitter, etc... in Builder - return apply(repeat); + + Predicate> retryContextPredicate = context -> { + if (exceedsMaxIterations(context.applicationContext(), retryConfig)) { + return false; + } + + for (Class clazz : retryConfig.getExceptions()) { + if (clazz.isInstance(context.exception())) { + return true; + } + } + return false; + }; + + Retry reactorRetry = Retry.onlyIf(retryContextPredicate) + .doOnRetry(context -> reset(context.applicationContext())) + .retryMax(retryConfig.getRetries()); + return apply(repeat, reactorRetry); } + public boolean exceedsMaxIterations(ServerWebExchange exchange, RetryConfig retryConfig) { + Integer iteration = exchange.getAttribute("retry_iteration"); + + //TODO: deal with null iteration + return iteration != null && iteration >= retryConfig.getRetries(); + } + + public void reset(ServerWebExchange exchange) { + //TODO: what else to do to reset SWE? + exchange.getAttributes().remove(ServerWebExchangeUtils.GATEWAY_ALREADY_ROUTED_ATTR); + } + + @Deprecated public GatewayFilter apply(Repeat repeat) { - return (exchange, chain) -> chain.filter(exchange).repeatWhen( - repeat.withApplicationContext(exchange)).next(); + return apply(repeat, Retry.onlyIf(ctxt -> false)); + } + + public GatewayFilter apply(Repeat repeat, Retry retry) { + return (exchange, chain) -> { + log.trace("Entering retry-filter"); + + int iteration = exchange.getAttributeOrDefault("retry_iteration", -1); + exchange.getAttributes().put("retry_iteration", iteration + 1); + + return Mono.fromDirect(chain.filter(exchange) + .log("retry-filter", Level.INFO) + .retryWhen(retry.withApplicationContext(exchange)) + .repeatWhen(repeat.withApplicationContext(exchange))); + }; } private static List toList(T item) { @@ -76,7 +133,7 @@ public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory series = toList(Series.SERVER_ERROR); @@ -84,31 +141,38 @@ public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory statuses = new ArrayList<>(); private List methods = toList(HttpMethod.GET); - - public Retry setRetries(int retries) { + + private List> exceptions = toList(IOException.class); + + public RetryConfig setRetries(int retries) { this.retries = retries; return this; } - public Retry setSeries(Series... series) { + public RetryConfig setSeries(Series... series) { this.series = Arrays.asList(series); return this; } - public Retry setStatuses(HttpStatus... statuses) { + public RetryConfig setStatuses(HttpStatus... statuses) { this.statuses = Arrays.asList(statuses); return this; } - public Retry setMethods(HttpMethod... methods) { + public RetryConfig setMethods(HttpMethod... methods) { this.methods = Arrays.asList(methods); return this; } - public Retry allMethods() { + public RetryConfig allMethods() { return setMethods(HttpMethod.values()); } + public RetryConfig setExceptions(Class... exceptions) { + this.exceptions = Arrays.asList(exceptions); + return this; + } + public void validate() { Assert.isTrue(this.retries > 0, "retries must be greater than 0"); Assert.isTrue(!this.series.isEmpty() || !this.statuses.isEmpty(), @@ -131,5 +195,10 @@ public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory getMethods() { return methods; } + + public List> getExceptions() { + return exceptions; + } + } } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/FilteringWebHandler.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/FilteringWebHandler.java index 161e1796..53ea2623 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/FilteringWebHandler.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/FilteringWebHandler.java @@ -89,22 +89,34 @@ public class FilteringWebHandler implements WebHandler { private static class DefaultGatewayFilterChain implements GatewayFilterChain { - private int index; + private final int index; private final List filters; public DefaultGatewayFilterChain(List filters) { this.filters = filters; + this.index = 0; + } + + private DefaultGatewayFilterChain(DefaultGatewayFilterChain parent, int index) { + this.filters = parent.getFilters(); + this.index = index; + } + + public List getFilters() { + return filters; } @Override public Mono filter(ServerWebExchange exchange) { - if (this.index < filters.size()) { - GatewayFilter filter = filters.get(this.index++); - return filter.filter(exchange, this); - } - else { - return Mono.empty(); // complete - } + return Mono.defer(() -> { + if (this.index < filters.size()) { + GatewayFilter filter = filters.get(this.index); + DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this, this.index + 1); + return filter.filter(exchange, chain); + } else { + return Mono.empty(); // complete + } + }); } } diff --git a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java index bff2fef6..38340257 100644 --- a/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java +++ b/spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java @@ -59,6 +59,7 @@ import org.springframework.http.HttpStatus; import org.springframework.web.server.ServerWebExchange; import reactor.retry.Repeat; +import reactor.retry.Retry; public class GatewayFilterSpec extends UriSpec { @@ -219,17 +220,23 @@ public class GatewayFilterSpec extends UriSpec { */ public GatewayFilterSpec retry(int retries) { return filter(getBean(RetryGatewayFilterFactory.class) - .apply(retry -> retry.setRetries(retries))); + .apply(retryConfig -> retryConfig.setRetries(retries))); } - public GatewayFilterSpec retry(Consumer retryConsumer) { + public GatewayFilterSpec retry(Consumer retryConsumer) { return filter(getBean(RetryGatewayFilterFactory.class).apply(retryConsumer)); } + public GatewayFilterSpec retry(Repeat repeat, Retry retry) { + return filter(getBean(RetryGatewayFilterFactory.class).apply(repeat, retry)); + } + + @Deprecated public GatewayFilterSpec retry(Repeat repeat) { return filter(getBean(RetryGatewayFilterFactory.class).apply(repeat)); } + @SuppressWarnings("unchecked") public GatewayFilterSpec secureHeaders() { return filter(getBean(SecureHeadersGatewayFilterFactory.class).apply(c -> {})); } @@ -262,6 +269,7 @@ public class GatewayFilterSpec extends UriSpec { .apply(c -> c.setStatus(status))); } + @SuppressWarnings("unchecked") public GatewayFilterSpec saveSession() { return filter(getBean(SaveSessionGatewayFilterFactory.class).apply(c -> {})); } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryConfigGatewayFilterFactoryIntegrationTests.java similarity index 70% rename from spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java rename to spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryConfigGatewayFilterFactoryIntegrationTests.java index 7bda5d92..bb2d2a7f 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryConfigGatewayFilterFactoryIntegrationTests.java @@ -17,9 +17,12 @@ package org.springframework.cloud.gateway.filter.factory; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; +import com.netflix.loadbalancer.Server; +import com.netflix.loadbalancer.ServerList; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.Test; @@ -29,9 +32,12 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.SpringBootConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.web.server.LocalServerPort; import org.springframework.cloud.gateway.route.RouteLocator; import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; import org.springframework.cloud.gateway.test.BaseWebClientTests; +import org.springframework.cloud.netflix.ribbon.RibbonClient; +import org.springframework.cloud.netflix.ribbon.StaticServerList; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.http.HttpHeaders; @@ -41,12 +47,13 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; +import static org.assertj.core.api.Assertions.assertThat; import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT; @RunWith(SpringRunner.class) @SpringBootTest(webEnvironment = RANDOM_PORT) @DirtiesContext -public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTests { +public class RetryConfigGatewayFilterFactoryIntegrationTests extends BaseWebClientTests { @Test public void retryFilterGet() { @@ -77,10 +84,29 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest // .expectBody(String.class).isEqualTo("3"); } + @Test + @SuppressWarnings("unchecked") + public void retryFilterLoadBalancedWithMultipleServers() { + String host = "www.retrywithloadbalancer.org"; + testClient.get() + .uri("/get") + .header(HttpHeaders.HOST, host) + .exchange() + .expectStatus().isOk() + .expectBody(Map.class) + .consumeWith(res -> { + Map body = res.getResponseBody(); + assertThat(body).isNotNull(); + Map headers = (Map) body.get("headers"); + assertThat(headers).containsEntry("X-Forwarded-Host", host); + }); + } + @RestController @EnableAutoConfiguration @SpringBootConfiguration @Import(DefaultTestConfig.class) + @RibbonClient(name = "badservice", configuration = TestBadRibbonConfig.class) public static class TestConfig { Log log = LogFactory.getLog(getClass()); @@ -108,8 +134,23 @@ public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTest .filters(f -> f.prefixPath("/httpbin") .retry(config -> config.setRetries(2))) .uri(uri)) + .route("retry_with_loadbalancer", r -> r.host("**.retrywithloadbalancer.org") + .filters(f -> f.prefixPath("/httpbin") + .retry(config -> config.setRetries(2))) + .uri("lb://badservice")) .build(); } } + protected static class TestBadRibbonConfig { + + @LocalServerPort + protected int port = 0; + + @Bean + public ServerList ribbonServerList() { + return new StaticServerList<>(new Server("https", "localhost.domain.doesnot.exist", this.port), new Server("localhost", this.port)); + } + } + } diff --git a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/HttpBinCompatibleController.java b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/HttpBinCompatibleController.java index 38a961aa..f2e9782f 100644 --- a/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/HttpBinCompatibleController.java +++ b/spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/HttpBinCompatibleController.java @@ -17,8 +17,18 @@ package org.springframework.cloud.gateway.test; +import java.io.IOException; +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.http.codec.multipart.FilePart; @@ -30,15 +40,6 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.server.ServerWebExchange; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.io.IOException; -import java.time.Duration; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; @RestController @RequestMapping("/httpbin") @@ -52,39 +53,40 @@ public class HttpBinCompatibleController { @RequestMapping(path = "/headers", method = { RequestMethod.GET, RequestMethod.POST}, produces = MediaType.APPLICATION_JSON_VALUE) - public Mono> headers(ServerWebExchange exchange) { - return getHeaders(exchange); + public Map headers(ServerWebExchange exchange) { + Map result = new HashMap<>(); + result.put("headers", getHeaders(exchange)); + return result; } @RequestMapping(path = "/delay/{sec}", produces = MediaType.APPLICATION_JSON_VALUE) public Mono> get(ServerWebExchange exchange, @PathVariable int sec) throws InterruptedException { int delay = Math.min(sec, 10); - return get(exchange).delayElement(Duration.ofSeconds(delay)); + return Mono.just(get(exchange)).delayElement(Duration.ofSeconds(delay)); } @RequestMapping(path = "/anything/{anything}", produces = MediaType.APPLICATION_JSON_VALUE) - public Mono> anything(ServerWebExchange exchange, @PathVariable(required = false) String anything) { + public Map anything(ServerWebExchange exchange, @PathVariable(required = false) String anything) { return get(exchange); } @RequestMapping(path = "/get", produces = MediaType.APPLICATION_JSON_VALUE) - public Mono> get(ServerWebExchange exchange) { - return getHeaders(exchange).map(map -> { - HashMap result = new HashMap<>(map); - HashMap params = new HashMap<>(); - exchange.getRequest().getQueryParams().forEach((name, values) -> { - params.put(name, values.get(0)); - }); - result.put("args", params); - return result; - }); + public Map get(ServerWebExchange exchange) { + HashMap result = new HashMap<>(); + HashMap params = new HashMap<>(); + exchange.getRequest().getQueryParams().forEach((name, values) -> { + params.put(name, values.get(0)); + }); + result.put("args", params); + result.put("headers", getHeaders(exchange)); + return result; } @RequestMapping(value = "/post", consumes = MediaType.MULTIPART_FORM_DATA_VALUE, produces = MediaType.APPLICATION_JSON_VALUE) public Mono> postFormData(@RequestBody Mono> parts) { // StringDecoder decoder = StringDecoder.allMimeTypes(true); return parts.flux().flatMap(map -> Flux.fromIterable(map.values())) - .flatMap(map -> Flux.fromIterable(map)) + .flatMap(Flux::fromIterable) .filter(part -> part instanceof FilePart) .reduce(new HashMap(), (files, part) -> { MediaType contentType = part.headers().getContentType(); @@ -104,6 +106,7 @@ public class HttpBinCompatibleController { public Mono> post(ServerWebExchange exchange, @RequestBody(required = false) String body) throws IOException { HashMap ret = new HashMap<>(); + ret.put("headers", getHeaders(exchange)); ret.put("data", body); HashMap form = new HashMap<>(); ret.put("form", form); @@ -123,13 +126,7 @@ public class HttpBinCompatibleController { return ResponseEntity.status(status).body("Failed with "+status); } - private Mono> getHeaders(ServerWebExchange exchange) { - return Flux.fromIterable(exchange.getRequest().getHeaders().entrySet()) - .collectMap(entry -> entry.getKey(), entry -> entry.getValue().get(0)) - .map(map -> { - Map result = new HashMap<>(); - result.put("headers", map); - return result; - }); + public Map getHeaders(ServerWebExchange exchange) { + return exchange.getRequest().getHeaders().toSingleValueMap(); } }