Merge branch 'retry'

spring-cloud-gateway-core/pom.xml

@@ -76,6 +76,10 @@
 			<artifactId>kotlin-reflect</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.addons</groupId>
+			<artifactId>reactor-extra</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>

spring-cloud-gateway-core/src/main/java/org/springframework/cloud/gateway/config/GatewayAutoConfiguration.java

@@ -53,6 +53,7 @@ import org.springframework.cloud.gateway.filter.factory.RemoveNonProxyHeadersGat
 import org.springframework.cloud.gateway.filter.factory.RemoveRequestHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RemoveResponseHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RequestRateLimiterGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.RetryGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RewritePathGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SaveSessionGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SecureHeadersGatewayFilterFactory;
@@ -380,6 +381,11 @@ public class GatewayAutoConfiguration {
 		return new RewritePathGatewayFilterFactory();
 	}
 
+	@Bean
+	public RetryGatewayFilterFactory retryGatewayFilterFactory() {
+		return new RetryGatewayFilterFactory();
+	}
+
 	@Bean
 	public SetPathGatewayFilterFactory setPathGatewayFilterFactory() {
 		return new SetPathGatewayFilterFactory();

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

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed 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.springframework.cloud.gateway.filter.factory;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Predicate;
+
+import reactor.retry.Repeat;
+import reactor.retry.RepeatContext;
+
+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.tuple.Tuple;
+import org.springframework.util.Assert;
+import org.springframework.web.server.ServerWebExchange;
+
+public class RetryGatewayFilterFactory implements GatewayFilterFactory {
+	@Override
+	public GatewayFilter apply(Tuple args) {
+		Retry retry = new Retry();
+
+		if (args.hasFieldName("retries")) {
+			retry.retries(args.getInt("retries"));
+		}
+
+		// TODO: list of statusSeries
+		if (args.hasFieldName("statusSeries")) {
+			int statusSeries = args.getInt("statusSeries");
+			retry.series(Series.valueOf(statusSeries));
+		}
+
+		// TODO: list of status
+		if (args.hasFieldName("status")) {
+			retry.statuses(ServerWebExchangeUtils.parse(args.getRawString("status")));
+		}
+
+		// TODO: list of methods
+		if (args.hasFieldName("method")) {
+			retry.methods(HttpMethod.resolve(args.getString("method").toUpperCase()));
+		}
+
+		return apply(retry);
+	}
+
+    public GatewayFilter apply(Retry retry) {
+		retry.validate();
+
+		Predicate<? super RepeatContext<ServerWebExchange>> predicate = context -> {
+			boolean retryableAttempt = context.iteration() < retry.getRetries();
+
+			if (!retryableAttempt) {
+				return false;
+			}
+
+			ServerWebExchange exchange = context.applicationContext();
+			HttpStatus statusCode = exchange.getResponse().getStatusCode();
+			HttpMethod httpMethod = exchange.getRequest().getMethod();
+
+			boolean retryableStatusCode = retry.getStatuses().contains(statusCode);
+
+			if (!retryableStatusCode) {
+				// try the series
+				retryableStatusCode = retry.getSeries().stream()
+						.anyMatch(series -> statusCode.series().equals(series));
+			}
+
+			boolean retryableMethod = retry.getMethods().contains(httpMethod);
+			return retryableMethod && retryableStatusCode;
+		};
+
+		//TODO: use Repeat statics once updated with a create() like method
+		Repeat<ServerWebExchange> repeat = Repeat.onlyIf(predicate);
+
+		//TODO: support timeout, backoff, jitter, etc... in Builder
+		return apply(repeat);
+	}
+
+	public GatewayFilter apply(Repeat<ServerWebExchange> repeat) {
+		return (exchange, chain) -> chain.filter(exchange).repeatWhen(
+                repeat.withApplicationContext(exchange)).next();
+	}
+
+	public static class Retry {
+		private int retries = 3;
+		
+		private List<Series> series = Collections.singletonList(Series.SERVER_ERROR);
+		
+		private List<HttpStatus> statuses = Collections.emptyList();
+		
+		private List<HttpMethod> methods = Collections.singletonList(HttpMethod.GET);
+		
+		public Retry retries(int retries) {
+			this.retries = retries;
+			return this;
+		}
+		
+		public Retry series(Series... series) {
+			this.series = Arrays.asList(series);
+			return this;
+		}
+		
+		public Retry statuses(HttpStatus... statuses) {
+			this.statuses = Arrays.asList(statuses);
+			return this;
+		}
+		
+		public Retry methods(HttpMethod... methods) {
+			this.methods = Arrays.asList(methods);
+			return this;
+		}
+
+		public Retry allMethods() {
+			return methods(HttpMethod.values());
+		}
+
+		public void validate() {
+			Assert.isTrue(this.retries > 0, "retries must be greater than 0");
+			Assert.isTrue(!this.series.isEmpty() || !this.statuses.isEmpty(),
+					"series and status may not both be empty");
+			Assert.notEmpty(this.methods, "methods may not be empty");
+		}
+
+		public int getRetries() {
+			return retries;
+		}
+
+		public List<Series> getSeries() {
+			return series;
+		}
+
+		public List<HttpStatus> getStatuses() {
+			return statuses;
+		}
+
+		public List<HttpMethod> getMethods() {
+			return methods;
+		}
+	}
+}

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

@@ -22,8 +22,11 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
+import com.netflix.hystrix.HystrixObservableCommand;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import reactor.retry.Repeat;
+
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.cloud.gateway.filter.GatewayFilter;
 import org.springframework.cloud.gateway.filter.OrderedGatewayFilter;
@@ -38,6 +41,7 @@ import org.springframework.cloud.gateway.filter.factory.RemoveNonProxyHeadersGat
 import org.springframework.cloud.gateway.filter.factory.RemoveRequestHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RemoveResponseHeaderGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RequestRateLimiterGatewayFilterFactory;
+import org.springframework.cloud.gateway.filter.factory.RetryGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.RewritePathGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SaveSessionGatewayFilterFactory;
 import org.springframework.cloud.gateway.filter.factory.SecureHeadersGatewayFilterFactory;
@@ -49,10 +53,10 @@ import org.springframework.cloud.gateway.filter.factory.StripPrefixGatewayFilter
 import org.springframework.cloud.gateway.filter.ratelimit.KeyResolver;
 import org.springframework.cloud.gateway.route.Route;
 import org.springframework.core.Ordered;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.tuple.Tuple;
-
-import com.netflix.hystrix.HystrixObservableCommand;
+import org.springframework.web.server.ServerWebExchange;
 
 import static org.springframework.tuple.TupleBuilder.tuple;
 
@@ -202,6 +206,36 @@ public class GatewayFilterSpec extends UriSpec {
 		return filter(getBean(RewritePathGatewayFilterFactory.class).apply(regex, replacement));
 	}
 
+	/**
+	 * 5xx errors and GET are retryable
+	 * @param retries max number of retries
+	 */
+	public GatewayFilterSpec retry(int retries) {
+		return filter(getBean(RetryGatewayFilterFactory.class)
+				.apply(new RetryGatewayFilterFactory.Retry()
+						.retries(retries)));
+	}
+
+	/**
+	 * @param retries max number of retries
+	 * @param httpStatusSeries the http status series that is retryable
+	 * @param httpMethod the http method that is retryable
+	 */
+	public GatewayFilterSpec retry(int retries, HttpStatus.Series httpStatusSeries, HttpMethod httpMethod) {
+		return retry(new RetryGatewayFilterFactory.Retry()
+						.retries(retries)
+						.series(httpStatusSeries)
+						.methods(httpMethod));
+	}
+
+	public GatewayFilterSpec retry(RetryGatewayFilterFactory.Retry retry) {
+		return filter(getBean(RetryGatewayFilterFactory.class).apply(retry));
+	}
+
+	public GatewayFilterSpec retry(Repeat<ServerWebExchange> repeat) {
+		return filter(getBean(RetryGatewayFilterFactory.class).apply(repeat));
+	}
+
 	public GatewayFilterSpec secureHeaders() {
 		return filter(getBean(SecureHeadersGatewayFilterFactory.class).apply(EMPTY_TUPLE));
 	}

spring-cloud-gateway-core/src/test/java/org/springframework/cloud/gateway/filter/factory/RetryGatewayFilterFactoryIntegrationTests.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2013-2018 the original author or authors.
+ *
+ * Licensed 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.springframework.cloud.gateway.filter.factory;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.gateway.test.BaseWebClientTests;
+import org.springframework.context.annotation.Import;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = RANDOM_PORT)
+@DirtiesContext
+public class RetryGatewayFilterFactoryIntegrationTests extends BaseWebClientTests {
+
+	@Test
+	public void retryFilterGet() {
+		testClient.get()
+				.uri("/retry?key=get")
+				.exchange()
+				.expectStatus().isOk()
+				.expectBody(String.class).isEqualTo("3");
+	}
+
+	@Test
+	//TODO: support post
+	public void retryFilterPost() {
+		testClient.post()
+				.uri("/retry?key=post")
+				.exchange()
+				.expectStatus().is5xxServerError();
+				// .expectBody(String.class).isEqualTo("3");
+	}
+
+	@RestController
+	@EnableAutoConfiguration
+	@SpringBootConfiguration
+	@Import(DefaultTestConfig.class)
+	public static class TestConfig {
+		Log log = LogFactory.getLog(getClass());
+
+		ConcurrentHashMap<String, AtomicInteger> map = new ConcurrentHashMap<>();
+
+		@RequestMapping("/httpbin/retry")
+		public String retry(@RequestParam("key") String key) {
+			AtomicInteger count = map.computeIfAbsent(key, s -> new AtomicInteger());
+			int i = count.incrementAndGet();
+			log.warn("Retry count: "+i);
+			if (i < 3) {
+				throw new RuntimeException("temporarily broken");
+			}
+			return String.valueOf(i);
+		}
+	}
+
+}

spring-cloud-gateway-core/src/test/resources/application.yml

@@ -131,6 +131,14 @@ spring:
         - AddResponseHeader=X-Request-Foo, Bar
         - RemoveResponseHeader=X-Request-Foo
 
+      # =====================================
+      - id: retry_test
+        uri: ${test.uri}
+        predicates:
+        - Path=/retry
+        filters:
+        - Retry
+
       # =====================================
       - id: secure_headers_test
         uri: ${test.uri}