Wrap search calls with a circuit breaker

Make sure we give some time for the service to recover if we enter
a situation where the error rate goes too high.

This patch introduces a circuit breaker for the calls that go
through the parallel scheduler so that we stop accepting queries for
a while (60s by default) if the error rate gets too high (50% by
default).

A health check (under /actuator/health/circuibreaker) has been added
and metrics are exposed via /actuator/prometheus and
/actuator/metrics/resilience4j.*

Id: 789a7c220019c926e8344dcf91efbb5351925dc4
Author: Caio
Commit time: 2018-12-16T14:21:03+01:00

Modified pom.xml

@@ -33,6 +33,7
<logback.version>1.2.3</logback.version>
<lucene.version>7.5.0</lucene.version>
<junit.jupiter.version>5.3.2</junit.jupiter.version>
+ <resilience4j.version>0.13.2</resilience4j.version>
<rocker.version>1.2.0</rocker.version>

<rocker.optimize>true</rocker.optimize>

@@ -111,6 +112,24
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-reactor</artifactId>
+ <version>${resilience4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-micrometer</artifactId>
+ <version>${resilience4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-circuitbreaker</artifactId>
+ <version>${resilience4j.version}</version>
</dependency>

<dependency>

Modified src/main/resources/application.properties

@@ -1,7 +1,11
-#logging.level.web=TRACE
+#logging.level.web=DEBUG
spring.main.banner-mode=off
spring.devtools.restart.enabled=true
+management.server.port=8081
+management.server.address=127.0.0.1
management.endpoint.prometheus.enabled=true
management.endpoints.web.exposure.include=info,health,metrics,prometheus
+management.endpoints.jmx.exposure.exclude=*
+management.endpoint.health.show-details=always
cerberus.search.location=tmp/hue
cerberus.search.timeout=2s

Modified src/main/java/co/caio/cerberus/boot/SearchConfiguration.java

@@ -1,10 +1,15
package co.caio.cerberus.boot;

import co.caio.cerberus.search.Searcher;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import io.github.resilience4j.micrometer.CircuitBreakerMetrics;
import java.nio.file.Path;
import java.time.Duration;
+import java.util.List;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

@@ -34,5 +39,52
@Bean("searchTimeout")
Duration getTimeout() {
return timeout;
+ }
+
+ @Bean("searchCircuitBreaker")
+ CircuitBreaker getSearchCircuitBreaker() {
+ return CircuitBreaker.ofDefaults("searchCircuitBreaker");
+ }
+
+ @Bean
+ CircuitBreakerMetrics registerMetrics(CircuitBreaker breaker) {
+ return CircuitBreakerMetrics.ofIterable(List.of(breaker));
+ }
+
+ @Component("circuitbreaker")
+ static class CircuitBreakerHealthIndicator implements HealthIndicator {
+ private final CircuitBreaker breaker;
+
+ CircuitBreakerHealthIndicator(CircuitBreaker breaker) {
+ this.breaker = breaker;
+ }
+
+ @Override
+ public Health health() {
+ switch (breaker.getState()) {
+ case CLOSED:
+ return withMetadata(Health.up());
+ case OPEN:
+ return withMetadata(Health.down());
+ default:
+ return withMetadata(Health.unknown());
+ }
+ }
+
+ private Health withMetadata(Health.Builder builder) {
+ var config = breaker.getCircuitBreakerConfig();
+ var metrics = breaker.getMetrics();
+
+ builder.withDetail("failureRate", metrics.getFailureRate());
+ builder.withDetail("failureRateThreshold", config.getFailureRateThreshold());
+ builder.withDetail("maxBufferedCalls", metrics.getMaxNumberOfBufferedCalls());
+ builder.withDetail("bufferedCalls", metrics.getNumberOfBufferedCalls());
+ builder.withDetail("failedCalls", metrics.getNumberOfFailedCalls());
+ builder.withDetail("notPermittedCalls", metrics.getNumberOfNotPermittedCalls());
+ builder.withDetail(
+ "openStateDurationSeconds", config.getWaitDurationInOpenState().toSeconds());
+
+ return builder.build();
+ }
}
}

Modified src/main/java/co/caio/cerberus/boot/WebController.java

@@ -1,8 +1,11
package co.caio.cerberus.boot;

import co.caio.cerberus.boot.SearchParameterParser.SearchParameterException;
import co.caio.cerberus.model.SearchQuery;
import co.caio.cerberus.search.Searcher;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import io.github.resilience4j.circuitbreaker.CircuitBreakerOpenException;
+import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator;
import io.micrometer.core.annotation.Timed;
import java.time.Duration;
import java.util.Map;

@@ -27,11 +30,14
private final Searcher searcher;
private final Duration timeout;
private final SearchParameterParser parser;
+ private final CircuitBreaker breaker;

- public WebController(Searcher searcher, @Qualifier("searchTimeout") Duration timeout) {
+ public WebController(
+ Searcher searcher, @Qualifier("searchTimeout") Duration timeout, CircuitBreaker breaker) {
this.searcher = searcher;
this.timeout = timeout;
this.parser = new SearchParameterParser();
+ this.breaker = breaker;
}

@GetMapping("/")

@@ -40,8 +46,8
return views.index.template("Cerberus").render().toString();
}

- @GetMapping("/search")
@Timed
+ @GetMapping("/search")
@ResponseBody
public Mono<String> search(@RequestParam Map<String, String> params) {
SearchQuery query = parser.buildQuery(params);

@@ -51,8 +57,9
// System.out.println(Thread.currentThread().getName());
return views.search.template(searcher.search(query)).render().toString();
})
- .subscribeOn(Schedulers.parallel())
- .timeout(timeout);
+ .publishOn(Schedulers.parallel())
+ .timeout(timeout)
+ .transform(CircuitBreakerOperator.of(breaker));
}

@ExceptionHandler({

@@ -74,6 +81,14
return ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
.contentType(MediaType.TEXT_HTML)
.body(views.index.template("Timeout: " + ex.getMessage()).render().toString());
+ }
+
+ @ExceptionHandler
+ @ResponseBody
+ ResponseEntity<String> handleCircuitBreaker(CircuitBreakerOpenException ex) {
+ return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
+ .contentType(MediaType.TEXT_HTML)
+ .body(views.index.template("Circuit Breaker: " + ex.getMessage()).render().toString());
}

@ExceptionHandler

Modified src/test/java/co/caio/cerberus/boot/WebControllerTest.java

@@ -5,8 +5,10

import co.caio.cerberus.model.SearchResult;
import co.caio.cerberus.search.Searcher;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import java.time.Duration;
import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;

@@ -22,14 +24,26
class WebControllerTest {

@Autowired WebTestClient testClient;
+ @Autowired CircuitBreaker breaker;
+
@MockBean Searcher searcher;

@TestConfiguration
static class TestConfig {
@Bean("searchTimeout")
- static Duration getTimeout() {
+ Duration getTimeout() {
return Duration.ofMillis(100);
}
+
+ @Bean
+ CircuitBreaker getCircuitBreaker() {
+ return CircuitBreaker.ofDefaults("test");
+ }
+ }
+
+ @BeforeEach
+ void resetCircuitBreakerState() {
+ breaker.reset();
}

@Test

@@ -50,6 +64,17
for (String badQuery : badQueries) {
assertGet("/search?" + badQuery, HttpStatus.BAD_REQUEST);
}
+ }
+
+ @Test
+ void circuitOpensAfterManyErrors() {
+ given(searcher.search(any())).willThrow(RuntimeException.class);
+ // error rate of 100%, but the default ring buffer is of 100 so
+ // the circuit should only open after the 100th request
+ for (int i = 0; i < 100; i++) {
+ assertGet("/search?q=bacon", HttpStatus.INTERNAL_SERVER_ERROR);
+ }
+ assertGet("/search?q=bacon", HttpStatus.SERVICE_UNAVAILABLE);
}

@Test

Created src/test/java/co/caio/cerberus/boot/SearchConfigurationTest.java

@@ -1,0 +1,33
+package co.caio.cerberus.boot;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import co.caio.cerberus.boot.SearchConfiguration.CircuitBreakerHealthIndicator;
+import io.github.resilience4j.circuitbreaker.CircuitBreaker;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.actuate.health.Status;
+
+class SearchConfigurationTest {
+
+ @Test
+ void circuitBreakerHealthEndpoint() {
+ var breaker = CircuitBreaker.ofDefaults("name");
+ var indicator = new CircuitBreakerHealthIndicator(breaker);
+
+ // Default status is closed -> healthy
+ assertEquals(Status.UP, indicator.health().getStatus());
+
+ breaker.transitionToOpenState();
+ assertEquals(Status.DOWN, indicator.health().getStatus());
+
+ breaker.transitionToDisabledState();
+ assertEquals(Status.UNKNOWN, indicator.health().getStatus());
+ breaker.transitionToForcedOpenState();
+ assertEquals(Status.UNKNOWN, indicator.health().getStatus());
+ breaker.transitionToHalfOpenState();
+ assertEquals(Status.UNKNOWN, indicator.health().getStatus());
+
+ breaker.transitionToClosedState();
+ assertEquals(Status.UP, indicator.health().getStatus());
+ }
+}