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

Merge remote-tracking branch 'origin/2.2.x'

Browse Source 
# Conflicts:
#	docs/src/main/asciidoc/_configprops.adoc
#	spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/LoadBalancerProperties.java
#	spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java
#	spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java
This commit is contained in:
Olga Maciaszek-Sharma
2020-12-01 11:31:55 +01:00
parent 48d315fa5b cbedae2e82
commit c65dfbca6b
8 changed files with 192 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplier.java
View File

@@ -25,6 +25,7 @@ import org.apache.commons.logging.LogFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.retry.Repeat;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
@@ -61,13 +62,18 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns
public HealthCheckServiceInstanceListSupplier(ServiceInstanceListSupplier delegate,
LoadBalancerProperties.HealthCheck healthCheck, WebClient webClient) {
super(delegate);
this.healthCheck = healthCheck;
defaultHealthCheckPath = healthCheck.getPath().getOrDefault("default", "/actuator/health");
this.webClient = webClient;
aliveInstancesReplay = Flux.defer(delegate).delaySubscription(healthCheck.getInitialDelay())
this.healthCheck = healthCheck;
Repeat<Object> aliveInstancesReplayRepeat = Repeat
.onlyIf(repeatContext -> this.healthCheck.getRefetchInstances())
.fixedBackoff(healthCheck.getRefetchInstancesInterval());
Flux<List<ServiceInstance>> aliveInstancesFlux = Flux.defer(delegate)
.switchMap(serviceInstances -> healthCheckFlux(serviceInstances)
.map(alive -> Collections.unmodifiableList(new ArrayList<>(alive))))
.replay(1).refCount(1);
.repeatWhen(aliveInstancesReplayRepeat);
aliveInstancesReplay = aliveInstancesFlux.delaySubscription(healthCheck.getInitialDelay()).replay(1)
.refCount(1);
}
@Override
@@ -80,6 +86,8 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns
}
protected Flux<List<ServiceInstance>> healthCheckFlux(List<ServiceInstance> instances) {
Repeat<Object> healthCheckFluxRepeat = Repeat.onlyIf(repeatContext -> healthCheck.getRepeatHealthCheck())
.fixedBackoff(healthCheck.getInterval());
return Flux.defer(() -> {
List<Mono<ServiceInstance>> checks = new ArrayList<>(instances.size());
for (ServiceInstance instance : instances) {
@@ -110,7 +118,7 @@ public class HealthCheckServiceInstanceListSupplier extends DelegatingServiceIns
result.add(alive);
return result;
}).defaultIfEmpty(result);
}).repeatWhen(restart -> restart.delayElements(healthCheck.getInterval()));
}).repeatWhen(healthCheckFluxRepeat);
}
@Override

60
spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/core/HealthCheckServiceInstanceListSupplierTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.loadbalancer.core;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -47,6 +48,8 @@ import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* Tests for {@link HealthCheckServiceInstanceListSupplier}.
@@ -136,11 +139,11 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class);
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1);
Mockito.doReturn(Mono.just(false)).when(mock).isAlive(serviceInstance2);
@@ -165,11 +168,11 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class);
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance2);
@@ -195,11 +198,11 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class);
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.doReturn(Mono.just(true)).when(mock).isAlive(serviceInstance1);
Mockito.doReturn(Mono.error(new RuntimeException("boom"))).when(mock).isAlive(serviceInstance2);
@@ -224,7 +227,7 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) {
@@ -251,7 +254,7 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1)));
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) {
@@ -277,11 +280,11 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1, serviceInstance2)));
HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class);
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.doReturn(Mono.just(false), Mono.just(true)).when(mock).isAlive(serviceInstance1);
Mockito.doReturn(Mono.error(new RuntimeException("boom"))).when(mock).isAlive(serviceInstance2);
@@ -306,11 +309,11 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1)));
HealthCheckServiceInstanceListSupplier mock = Mockito.mock(HealthCheckServiceInstanceListSupplier.class);
HealthCheckServiceInstanceListSupplier mock = mock(HealthCheckServiceInstanceListSupplier.class);
Mockito.when(mock.isAlive(serviceInstance1)).thenReturn(Mono.never(), Mono.just(true));
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) {
@@ -336,7 +339,7 @@ class HealthCheckServiceInstanceListSupplierTests {
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Flux<List<ServiceInstance>> instances = Flux.just(Lists.list(serviceInstance1))
.concatWith(Flux.just(Lists.list(serviceInstance1, serviceInstance2))
@@ -358,6 +361,33 @@ class HealthCheckServiceInstanceListSupplierTests {
.thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldRefetchInstances() {
healthCheck.setInitialDelay(Duration.ofSeconds(1));
healthCheck.setRepeatHealthCheck(false);
healthCheck.setRefetchInstancesInterval(Duration.ofSeconds(1));
healthCheck.setRefetchInstances(true);
ServiceInstance serviceInstance1 = new DefaultServiceInstance("ignored-service-1", SERVICE_ID, "127.0.0.1",
port, false);
ServiceInstance serviceInstance2 = new DefaultServiceInstance("ignored-service-2", SERVICE_ID, "127.0.0.2",
port, false);
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
when(delegate.get()).thenReturn(Flux.just(Collections.singletonList(serviceInstance1)))
.thenReturn(Flux.just(Collections.singletonList(serviceInstance2)));
listSupplier = new HealthCheckServiceInstanceListSupplier(delegate, healthCheck, webClient) {
@Override
protected Mono<Boolean> isAlive(ServiceInstance serviceInstance) {
return Mono.just(true);
}
};
return listSupplier.get();
}).expectSubscription().expectNoEvent(healthCheck.getInitialDelay()).expectNext(Lists.list(serviceInstance1))
.thenAwait(healthCheck.getRefetchInstancesInterval()).expectNext(Lists.list(serviceInstance2))
.thenCancel().verify(VERIFY_TIMEOUT);
}
@Test
void shouldCacheResultIfAfterPropertiesSetInvoked() {
healthCheck.setInitialDelay(Duration.ofSeconds(1));
@@ -367,7 +397,7 @@ class HealthCheckServiceInstanceListSupplierTests {
AtomicInteger emitCounter = new AtomicInteger();
StepVerifier.withVirtualTime(() -> {
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.getServiceId()).thenReturn(SERVICE_ID);
Mockito.when(delegate.get()).thenReturn(Flux.just(Lists.list(serviceInstance1)));
@@ -397,7 +427,7 @@ class HealthCheckServiceInstanceListSupplierTests {
final AtomicInteger instancesCanceled = new AtomicInteger();
final AtomicBoolean subscribed = new AtomicBoolean();
ServiceInstanceListSupplier delegate = Mockito.mock(ServiceInstanceListSupplier.class);
ServiceInstanceListSupplier delegate = mock(ServiceInstanceListSupplier.class);
Mockito.when(delegate.get()).thenReturn(Flux.<List<ServiceInstance>>never()
.doOnSubscribe(subscription -> subscribed.set(true)).doOnCancel(instancesCanceled::incrementAndGet));