ゲートウェイの非ゲット・リクエストの再試行
OpenFeignの場合：
- Get リクエスト：200 以外のレスポンスコード、例外、再試行。
- Get 以外のリクエスト: IOException とリダイレンスブレーカの例外はすべてリトライされ、それ以外は何も行われません。
Spring Cloud Gatewayの場合：
- Get リクエスト：4XX、5XX 応答コード、例外があれば再試行します。
ここで、Spring Cloud Gatewayの非GetリクエストのIOExceptionに対するリトライと、リダイレンスブレーカー例外を実装する必要があります。
既存の設計
org.springframework.cloud.gateway.filter.factory.RetryGatewayFilterFactory
public class RetryGatewayFilterFactory
extends AbstractGatewayFilterFactory<RetryGatewayFilterFactory.RetryConfig> {
/**
* Retry iteration key.ServerWebExchangeのAttributeのキーは
* このアトリビュートは，+1，リトライ回数が超過していないか確認する
*/
public static final String RETRY_ITERATION_KEY = "retry_iteration";
public RetryGatewayFilterFactory() {
super(RetryConfig.class);
}
@Override
public GatewayFilter apply(RetryConfig retryConfig) {
//テスト構成
retryConfig.validate();
Repeat<ServerWebExchange> statusCodeRepeat = null;
//再試行可能な HTTP レスポンス・ステータス・コードが設定されている場合、そのレスポンス・コードが再試行可能かどうかを確認する
if (!retryConfig.getStatuses().isEmpty() || !retryConfig.getSeries().isEmpty()) {
Predicate<RepeatContext<ServerWebExchange>> repeatPredicate = context -> {
ServerWebExchange exchange = context.applicationContext();
//リトライ回数を超えていないか確認する
if (exceedsMaxIterations(exchange, retryConfig)) {
return false;
}
//再試行が可能かどうかを判断する
HttpStatus statusCode = exchange.getResponse().getStatusCode();
boolean retryableStatusCode = retryConfig.getStatuses()
.contains(statusCode);
if (!retryableStatusCode && statusCode != null) {
// try the series
retryableStatusCode = retryConfig.getSeries().stream()
.anyMatch(series -> statusCode.series().equals(series));
}
final boolean finalRetryableStatusCode = retryableStatusCode;
//HTTPメソッドが再試行可能かどうかを判断する
HttpMethod httpMethod = exchange.getRequest().getMethod();
boolean retryableMethod = retryConfig.getMethods().contains(httpMethod);
//メソッドが再試行可能かどうかの HTTP レスポンス・ステータス・コードを返す。
return retryableMethod && finalRetryableStatusCode;
};
//再試行するたびにルートをリセットし、ルートを再解決しなければならない
statusCodeRepeat = Repeat.onlyIf(repeatPredicate)
.doOnRepeat(context -> reset(context.applicationContext()));
//バックオフを設定する
BackoffConfig backoff = retryConfig.getBackoff();
if (backoff != null) {
statusCodeRepeat = statusCodeRepeat.backoff(getBackoff(backoff));
}
}
// TODO: support timeout, backoff, jitter, etc... in Builder
//例外が再試行できるかどうかを判断する
Retry<ServerWebExchange> exceptionRetry = null;
if (!retryConfig.getExceptions().isEmpty()) {
Predicate<RetryContext<ServerWebExchange>> retryContextPredicate = context -> {
ServerWebExchange exchange = context.applicationContext();
if (exceedsMaxIterations(exchange, retryConfig)) {
return false;
}
Throwable exception = context.exception();
for (Class<? extends Throwable> retryableClass : retryConfig
.getExceptions()) {
if (retryableClass.isInstance(exception) || (exception != null
&& retryableClass.isInstance(exception.getCause()))) {
trace("exception or its cause is retryable %s, configured exceptions %s",
() -> getExceptionNameWithCause(exception),
retryConfig::getExceptions);
HttpMethod httpMethod = exchange.getRequest().getMethod();
boolean retryableMethod = retryConfig.getMethods()
.contains(httpMethod);
trace("retryableMethod: %b, httpMethod %s, configured methods %s",
() -> retryableMethod, () -> httpMethod,
retryConfig::getMethods);
return retryableMethod;
}
}
trace("exception or its cause is not retryable %s, configured exceptions %s",
() -> getExceptionNameWithCause(exception),
retryConfig::getExceptions);
return false;
};
exceptionRetry = Retry.onlyIf(retryContextPredicate)
.doOnRetry(context -> reset(context.applicationContext()))
.retryMax(retryConfig.getRetries());
BackoffConfig backoff = retryConfig.getBackoff();
if (backoff != null) {
exceptionRetry = exceptionRetry.backoff(getBackoff(backoff));
}
}
GatewayFilter gatewayFilter = apply(retryConfig.getRouteId(), statusCodeRepeat,
exceptionRetry);
return new GatewayFilter() {
@Override
public Mono<Void> filter(ServerWebExchange exchange,
GatewayFilterChain chain) {
return gatewayFilter.filter(exchange, chain);
}
@Override
public String toString() {
return filterToStringCreator(RetryGatewayFilterFactory.this)
.append("retries", retryConfig.getRetries())
.append("series", retryConfig.getSeries())
.append("statuses", retryConfig.getStatuses())
.append("methods", retryConfig.getMethods())
.append("exceptions", retryConfig.getExceptions()).toString();
}
};
}
private String getExceptionNameWithCause(Throwable exception) {
if (exception != null) {
StringBuilder builder = new StringBuilder(exception.getClass().getName());
Throwable cause = exception.getCause();
if (cause != null) {
builder.append("{cause=").append(cause.getClass().getName()).append("}");
}
return builder.toString();
}
else {
return "null";
}
}
private Backoff getBackoff(BackoffConfig backoff) {
return Backoff.exponential(backoff.firstBackoff, backoff.maxBackoff,
backoff.factor, backoff.basedOnPreviousValue);
}
public boolean exceedsMaxIterations(ServerWebExchange exchange,
RetryConfig retryConfig) {
Integer iteration = exchange.getAttribute(RETRY_ITERATION_KEY);
//再試行可能回数を超えたかどうか
boolean exceeds = iteration != null && iteration >= retryConfig.getRetries();
return exceeds;
}
public void reset(ServerWebExchange exchange) {
//このメソッドは主に
Set<String> addedHeaders = exchange.getAttributeOrDefault(
CLIENT_RESPONSE_HEADER_NAMES, Collections.emptySet());
addedHeaders
.forEach(header -> exchange.getResponse().getHeaders().remove(header));
removeAlreadyRouted(exchange);
}
public GatewayFilter apply(String routeId, Repeat<ServerWebExchange> repeat,
Retry<ServerWebExchange> retry) {
if (routeId != null && getPublisher() != null) {
// send an event to enable caching
getPublisher().publishEvent(new EnableBodyCachingEvent(this, routeId));
}
return (exchange, chain) -> {
trace("Entering retry-filter");
// chain.filter returns a Mono<Void>
Publisher<Void> publisher = chain.filter(exchange)
// .log("retry-filter", Level.INFO)
.doOnSuccessOrError((aVoid, throwable) -> {
int iteration = exchange
.getAttributeOrDefault(RETRY_ITERATION_KEY, -1);
int newIteration = iteration + 1;
trace("setting new iteration in attr %d", () -> newIteration);
exchange.getAttributes().put(RETRY_ITERATION_KEY, newIteration);
});
if (retry != null) {
// retryWhen returns a Mono<Void>
// retry needs to go before repeat
publisher = ((Mono<Void>) publisher)
.retryWhen(retry.withApplicationContext(exchange));
}
if (repeat != null) {
// repeatWhen returns a Flux<Void>
// so this needs to be last and the variable a Publisher<Void>
publisher = ((Mono<Void>) publisher)
.repeatWhen(repeat.withApplicationContext(exchange));
}
return Mono.fromDirect(publisher);
};
}
@SuppressWarnings("unchecked")
public static class RetryConfig implements HasRouteId {
//ルート ID
private String routeId;
//最初の呼び出しを除いた再試行回数はデフォルトで3回、つまり4回呼び出される可能性がある。
private int retries = 3;
//どのHTTPステータスコードを再試行するかについては、SeriesはHttpStatusのセットに対応する
private List<Series> series = toList(Series.SERVER_ERROR);
//HttpStatusはHTTPステータスコードである。
private List<HttpStatus> statuses = new ArrayList<>();
//どのHTTPメソッドに対して再試行するか
private List<HttpMethod> methods = toList(HttpMethod.GET);
//どの例外に対して再試行するか
private List<Class<? extends Throwable>> exceptions = toList(IOException.class,
TimeoutException.class);
//再試行間隔ポリシー
private BackoffConfig backoff;
public void validate() {
//リトライは 10 以上でなければならない
Assert.isTrue(this.retries > 0, "retries must be greater than 0");
//再試行可能なシリーズ、再試行可能なステータスコード、再試行可能な例外は、すべて空にすることはできない。
Assert.isTrue(
!this.series.isEmpty() || !this.statuses.isEmpty()
|| !this.exceptions.isEmpty(),
"series, status and exceptions may not all be empty");
//リトライされた Http メソッドは null にはできない
Assert.notEmpty(this.methods, "methods may not be empty");
if (this.backoff != null) {
this.backoff.validate();
}
}
//コンストラクタ、ゲッター、セッター、およびいくつかのユーティリティメソッドを省略する
}
public static class BackoffConfig {
//最初の再試行間隔
private Duration firstBackoff = Duration.ofMillis(5);
//最大待機間隔
private Duration maxBackoff;
//成長率
private int factor = 2;
//最後のリクエストのリトライ間隔を保持し、次回その間隔からリトライするかどうか
private boolean basedOnPreviousValue = true;
//コンストラクタを省略する，getter，setter
public void validate() {
//最初の再試行間隔は null にはできない
Assert.notNull(this.firstBackoff, "firstBackoff must be present");
}
}
}
要約すると、プロセスは次のように単純化されます：
- このリクエストのHTTPメソッドがRetryConfig.methodsに含まれているかどうか、HTTPレスポンスコードがRetryConfig.seriesの範囲内かステータスの集合に含まれているかどうかを判断し、RetryConfig.seriesの範囲内かステータスの集合に含まれている場合は、このリクエストのretry_iterationを見て、このAttributeのretry_iterationが初回かどうかを確認し、retry_iterationを超える場合はリトライします。retry_iterationを越えていなければ再試行し、越えていれば再試行を停止します。
- このリクエストのHTTPメソッドがRetryConfig.methodsに含まれているかどうか、例外がRetryConfig.exceptionのセットに含まれているかどうかを判断し、含まれている場合は、このリクエストの属性retry_iterationがリトライ回数を超えた回数を確認し、超えていない場合はリトライし、リトライを停止します。もし超えていなければ再試行し、超えていれば再試行を停止します。
設定時に、HTTP メソッドが全てのメソッドを含んでいる場合、 GET リクエストか非 GET リクエストかを区別する方法はありません。 もし二つのフィルタを作って、一つは GET を、もう一つは非 GET をインターセプトすると、 それらが共有する Attribute は毎回 +2 になり、再試行回数は不正確になります。
application.propertiesつまり、GET と非 GET で異なる RetryConfig を使い、GET の方は設定に基づいたまま、非 GET リクエストでは次の例外を強制的に再試行するというものです：
- io.netty.channel.ConnectTimeoutException.class接続タイムアウト
- java.net.ConnectException.classCallNotPermittedException: resilience4j サーキットブレーカー関連の例外: ホストへのルートがない例外
- io.github.resilience4j.circuitbreaker.CallNotPermittedException: resilience4j サーキット・ブレーカ関連の例外
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
ServerHttpRequest request = exchange.getRequest();
//マイクロサービス名を取得する
String serviceName = request.getHeaders().getFirst(CommonConstant.SERVICE_NAME);
HttpMethod method = exchange.getRequest().getMethod();
//GatewayFilterを生成し、gatewayFilterMapに保存する
GatewayFilter gatewayFilter = gatewayFilterMap.computeIfAbsent(serviceName + ":" + method, k -> {
Map<String, RetryConfig> retryConfigMap = apiGatewayRetryConfig.getRetry();
//マイクロサービス名で再試行設定を取得する
RetryConfig retryConfig = retryConfigMap.containsKey(serviceName) ? retryConfigMap.get(serviceName) : apiGatewayRetryConfig.getDefault();
//再試行回数が 0 の場合、再試行しない
if (retryConfig.getRetries() == 0) {
return null;
}
//GET以外のリクエストでは、再試行に制限をかけ、以下の例外bのみを再試行する。
if (!HttpMethod.GET.equals(method)) {
RetryConfig newConfig = new RetryConfig();
BeanUtils.copyProperties(retryConfig, newConfig);
//再試行されるすべてのメソッドを制限することは、ここでは、外側のレイヤーがGETでないことを制限するので、GETでないすべてのメソッドと等価である
newConfig.setMethods(HttpMethod.values());
newConfig.setSeries();
newConfig.setStatuses();
newConfig.setExceptions(//リンクタイムアウト
io.netty.channel.ConnectTimeoutException.class,
//No route to host
java.net.ConnectException.class,
//Resilience4jの例外
CallNotPermittedException.class);
retryConfig = newConfig;
}
return this.apply(retryConfig);
});
return gatewayFilter != null ? gatewayFilter.filter(exchange, chain) : chain.filter(exchange);
}
