SpringCloud/spring-cloud-gateway
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adds support for retrying customizable list of exceptions.

Browse Source 
fixes gh-223
This commit is contained in:
Spencer Gibb
2018-04-25 16:22:33 -04:00
parent b07d805b78
commit 9d8c3d98dd
5 changed files with 189 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactory.java
View File

@@ -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<RetryGatewayFilterFactory.Retry> {
public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory<RetryGatewayFilterFactory.RetryConfig> {
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<? super RepeatContext<ServerWebExchange>> 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<ServerWebExchange> repeat = Repeat.create(predicate, retry.getRetries());
Repeat<ServerWebExchange> repeat = Repeat.onlyIf(predicate)
.doOnRepeat(context -> reset(context.applicationContext()));
//TODO: support timeout, backoff, jitter, etc... in Builder
return apply(repeat);
Predicate<RetryContext<ServerWebExchange>> retryContextPredicate = context -> {
if (exceedsMaxIterations(context.applicationContext(), retryConfig)) {
return false;
}
for (Class<? extends Throwable> clazz : retryConfig.getExceptions()) {
if (clazz.isInstance(context.exception())) {
return true;
}
}
return false;
};
Retry<ServerWebExchange> 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<ServerWebExchange> repeat) {
return (exchange, chain) -> chain.filter(exchange).repeatWhen(
repeat.withApplicationContext(exchange)).next();
return apply(repeat, Retry.onlyIf(ctxt -> false));
}
public GatewayFilter apply(Repeat<ServerWebExchange> repeat, Retry<ServerWebExchange> 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 <T> List<T> toList(T item) {
@@ -76,7 +133,7 @@ public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory<Retr
return list;
}
public static class Retry {
public static class RetryConfig {
private int retries = 3;
private List<Series> series = toList(Series.SERVER_ERROR);
@@ -84,31 +141,38 @@ public class RetryGatewayFilterFactory extends AbstractGatewayFilterFactory<Retr
private List<HttpStatus> statuses = new ArrayList<>();
private List<HttpMethod> methods = toList(HttpMethod.GET);
public Retry setRetries(int retries) {
private List<Class<? extends Throwable>> 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<? extends Throwable>... 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<Retr
public List<HttpMethod> getMethods() {
return methods;
}
public List<Class<? extends Throwable>> getExceptions() {
return exceptions;
}
}
}

28
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/handler/FilteringWebHandler.java
View File

@@ -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<GatewayFilter> filters;
public DefaultGatewayFilterChain(List<GatewayFilter> filters) {
this.filters = filters;
this.index = 0;
}
private DefaultGatewayFilterChain(DefaultGatewayFilterChain parent, int index) {
this.filters = parent.getFilters();
this.index = index;
}
public List<GatewayFilter> getFilters() {
return filters;
}
@Override
public Mono<Void> 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
}
});
}
}

12
spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/route/builder/GatewayFilterSpec.java
View File

@@ -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<RetryGatewayFilterFactory.Retry> retryConsumer) {
public GatewayFilterSpec retry(Consumer<RetryGatewayFilterFactory.RetryConfig> retryConsumer) {
return filter(getBean(RetryGatewayFilterFactory.class).apply(retryConsumer));
}
public GatewayFilterSpec retry(Repeat<ServerWebExchange> repeat, Retry<ServerWebExchange> retry) {
return filter(getBean(RetryGatewayFilterFactory.class).apply(repeat, retry));
}
@Deprecated
public GatewayFilterSpec retry(Repeat<ServerWebExchange> 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 -> {}));
}

43
spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java → spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryConfigGatewayFilterFactoryIntegrationTests.java
View File

@@ -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<String, Object> headers = (Map<String, Object>) 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<Server> ribbonServerList() {
return new StaticServerList<>(new Server("https", "localhost.domain.doesnot.exist", this.port), new Server("localhost", this.port));
}
}
}

61
spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/test/HttpBinCompatibleController.java
View File

@@ -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<Map<String, Object>> headers(ServerWebExchange exchange) {
return getHeaders(exchange);
public Map<String, Object> headers(ServerWebExchange exchange) {
Map<String, Object> result = new HashMap<>();
result.put("headers", getHeaders(exchange));
return result;
}
@RequestMapping(path = "/delay/{sec}", produces = MediaType.APPLICATION_JSON_VALUE)
public Mono<Map<String, Object>> 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<Map<String, Object>> anything(ServerWebExchange exchange, @PathVariable(required = false) String anything) {
public Map<String, Object> anything(ServerWebExchange exchange, @PathVariable(required = false) String anything) {
return get(exchange);
}
@RequestMapping(path = "/get", produces = MediaType.APPLICATION_JSON_VALUE)
public Mono<Map<String, Object>> get(ServerWebExchange exchange) {
return getHeaders(exchange).map(map -> {
HashMap<String, Object> result = new HashMap<>(map);
HashMap<String, String> params = new HashMap<>();
exchange.getRequest().getQueryParams().forEach((name, values) -> {
params.put(name, values.get(0));
});
result.put("args", params);
return result;
});
public Map<String, Object> get(ServerWebExchange exchange) {
HashMap<String, Object> result = new HashMap<>();
HashMap<String, String> 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<Map<String, Object>> postFormData(@RequestBody Mono<MultiValueMap<String, Part>> 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<String, Object>(), (files, part) -> {
MediaType contentType = part.headers().getContentType();
@@ -104,6 +106,7 @@ public class HttpBinCompatibleController {
public Mono<Map<String, Object>> post(ServerWebExchange exchange,
@RequestBody(required = false) String body) throws IOException {
HashMap<String, Object> ret = new HashMap<>();
ret.put("headers", getHeaders(exchange));
ret.put("data", body);
HashMap<String, Object> form = new HashMap<>();
ret.put("form", form);
@@ -123,13 +126,7 @@ public class HttpBinCompatibleController {
return ResponseEntity.status(status).body("Failed with "+status);
}
private Mono<Map<String, Object>> getHeaders(ServerWebExchange exchange) {
return Flux.fromIterable(exchange.getRequest().getHeaders().entrySet())
.collectMap(entry -> entry.getKey(), entry -> entry.getValue().get(0))
.map(map -> {
Map<String, Object> result = new HashMap<>();
result.put("headers", map);
return result;
});
public Map<String, String> getHeaders(ServerWebExchange exchange) {
return exchange.getRequest().getHeaders().toSingleValueMap();
}
}