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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Olga Maciaszek-Sharma
2020-12-18 17:58:35 +01:00
parent 8125db910a a40258394a
commit c6bfc2d827
24 changed files with 873 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
docs/src/main/asciidoc/spring-cloud-commons.adoc
View File

@@ -1118,15 +1118,38 @@ public class MyConfiguration {
One type of bean that it may be useful to register using <<custom-loadbalancer-configuration,Custom LoadBalancer configuration>> is `LoadBalancerLifecycle`.
The LoadBalancerLifecycle beans provide callback methods, named `onStart(Request<RC> request)` and `onComplete(CompletionContext<RES, T> completionContext)`, that you should implement to specify what actions should take place before and after load-balancing.
The `LoadBalancerLifecycle` beans provide callback methods, named `onStart(Request<RC> request)`, `onStartRequest(Request<RC> request, Response<T> lbResponse)` and `onComplete(CompletionContext<RES, T, RC> completionContext)`, that you should implement to specify what actions should take place before and after load-balancing.
`onStart(Request<RC> request)` takes a `Request` object as a parameter. It contains data that is used to select an appropriate instance, including the downstream client request and <<spring-cloud-loadbalancer-hints,hint>>. On the other hand, a `CompletionContext` object is provided to the `onComplete(CompletionContext<RES, T> completionContext)` method. It contains the LoadBalancer `Response`, including the selected service instance, the `Status` of the request executed against that service instance and (if available) the response returned to the downstream client, and (if an exception has occurred) the corresponding `Throwable`.
`onStart(Request<RC> request)` takes a `Request` object as a parameter. It contains data that is used to select an appropriate instance, including the downstream client request and <<spring-cloud-loadbalancer-hints,hint>>. `onStartRequest` also takes the `Request` object and, additionally, the `Response<T>` object as parameters. On the other hand, a `CompletionContext` object is provided to the `onComplete(CompletionContext<RES, T, RC> completionContext)` method. It contains the LoadBalancer `Response`, including the selected service instance, the `Status` of the request executed against that service instance and (if available) the response returned to the downstream client, and (if an exception has occurred) the corresponding `Throwable`.
The `supports(Class requestContextClass, Class responseClass,
Class serverTypeClass)` method can be used to determine whether the processor in question handles objects of provided types. If not overridden by the user, it returns `true`.
NOTE: In the preceding method calls, `RC` means `RequestContext` type, `RES` means client response type, and `T` means returned server type.
[[loadbalancer-micrometer-stats-lifecycle]]
=== Spring Cloud LoadBalancer Statistics
We provide a `LoadBalancerLifecycle` bean called `MicrometerStatsLoadBalancerLifecycle`, which uses Micrometer to provide statistics for load-balanced calls.
In order to get this bean added to your application context,
set the value of the `spring.cloud.loadbalancer.stats.micrometer.enabled` to `true` and have a `MeterRegistry` available (for example, by adding https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-features.html[Spring Boot Actuator] to your project).
`MicrometerStatsLoadBalancerLifecycle` registers the following meters in `MeterRegistry`:
* `loadbalancer.requests.active`: A gauge that allows you to monitor the number of currently active requests for any service instance (service instance data available via tags);
* `loadbalancer.requests.success`: A timer that measures the time of execution of any load-balanced requests that have ended in passing a response on to the underlying client;
* `loadbalancer.requests.failed`: A timer that measures the time of execution of any load-balanced requests that have ended with an exception;
* `loadbalancer.requests.discard`: A counter that measures the number of discarded load-balanced requests, i.e. requests where a service instance to run the request on has not been retrieved by the LoadBalancer.
Additional information regarding the service instances, request data, and response data is added to metrics via tags whenever available.
NOTE: For some implementations, such as `BlockingLoadBalancerClient`, request and response data might not be available, as we establish generic types from arguments and might not be able to determine the types and read the data.
NOTE: The meters are registered in the registry when at least one record is added for a given meter.
TIP: You can further configure the behavior of those metrics (for example, add https://micrometer.io/docs/concepts#_histograms_and_percentiles[publishing percentiles and histograms]) by https://docs.spring.io/spring-boot/docs/current/reference/html/production-ready-features.html#production-ready-metrics-per-meter-properties[adding `MeterFilters`].
== Spring Cloud Circuit Breaker
include::spring-cloud-circuitbreaker.adoc[leveloffset=+1]

13
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/AsyncLoadBalancerInterceptor.java
View File

@@ -19,7 +19,6 @@ package org.springframework.cloud.client.loadbalancer;
import java.io.IOException;
import java.net.URI;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.AsyncClientHttpRequestExecution;
import org.springframework.http.client.AsyncClientHttpRequestInterceptor;
@@ -42,14 +41,10 @@ public class AsyncLoadBalancerInterceptor implements AsyncClientHttpRequestInter
final AsyncClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
return this.loadBalancer.execute(serviceName, new LoadBalancerRequest<ListenableFuture<ClientHttpResponse>>() {
@Override
public ListenableFuture<ClientHttpResponse> apply(final ServiceInstance instance) throws Exception {
HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance,
AsyncLoadBalancerInterceptor.this.loadBalancer);
return execution.executeAsync(serviceRequest, body);
}
return this.loadBalancer.execute(serviceName, instance -> {
HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance,
AsyncLoadBalancerInterceptor.this.loadBalancer);
return execution.executeAsync(serviceRequest, body);
});
}

30
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/CompletionContext.java
View File

@@ -25,7 +25,7 @@ import org.springframework.core.style.ToStringCreator;
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
public class CompletionContext<RES, T> {
public class CompletionContext<RES, T, C> {
private final Status status;
@@ -35,25 +35,31 @@ public class CompletionContext<RES, T> {
private final RES clientResponse;
public CompletionContext(Status status) {
this(status, null, null, null);
private final Request<C> loadBalancerRequest;
public CompletionContext(Status status, Request<C> loadBalancerRequest) {
this(status, null, loadBalancerRequest, null, null);
}
public CompletionContext(Status status, Response<T> response) {
this(status, null, response, null);
public CompletionContext(Status status, Request<C> loadBalancerRequest, Response<T> response) {
this(status, null, loadBalancerRequest, response, null);
}
public CompletionContext(Status status, Throwable throwable, Response<T> loadBalancerResponse) {
this(status, throwable, loadBalancerResponse, null);
public CompletionContext(Status status, Throwable throwable, Request<C> loadBalancerRequest,
Response<T> loadBalancerResponse) {
this(status, throwable, loadBalancerRequest, loadBalancerResponse, null);
}
public CompletionContext(Status status, Response<T> loadBalancerResponse, RES clientResponse) {
this(status, null, loadBalancerResponse, clientResponse);
public CompletionContext(Status status, Request<C> loadBalancerRequest, Response<T> loadBalancerResponse,
RES clientResponse) {
this(status, null, loadBalancerRequest, loadBalancerResponse, clientResponse);
}
public CompletionContext(Status status, Throwable throwable, Response<T> loadBalancerResponse, RES clientResponse) {
public CompletionContext(Status status, Throwable throwable, Request<C> loadBalancerRequest,
Response<T> loadBalancerResponse, RES clientResponse) {
this.status = status;
this.throwable = throwable;
this.loadBalancerRequest = loadBalancerRequest;
this.loadBalancerResponse = loadBalancerResponse;
this.clientResponse = clientResponse;
}
@@ -74,6 +80,10 @@ public class CompletionContext<RES, T> {
return clientResponse;
}
public Request<C> getLoadBalancerRequest() {
return loadBalancerRequest;
}
@Override
public String toString() {
ToStringCreator to = new ToStringCreator(this);

14
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/HintRequestContext.java
View File

@@ -25,7 +25,7 @@ import org.springframework.core.style.ToStringCreator;
*
* @author Olga Maciaszek-Sharma
*/
public class HintRequestContext {
public class HintRequestContext implements TimedRequestContext {
/**
* A {@link String} value of hint that can be used to choose the correct service
@@ -33,6 +33,8 @@ public class HintRequestContext {
*/
private String hint = "default";
private long requestStartTime;
public HintRequestContext() {
}
@@ -48,6 +50,16 @@ public class HintRequestContext {
this.hint = hint;
}
@Override
public long getRequestStartTime() {
return requestStartTime;
}
@Override
public void setRequestStartTime(long requestStartTime) {
this.requestStartTime = requestStartTime;
}
@Override
public String toString() {
ToStringCreator to = new ToStringCreator(this);

11
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/LoadBalancerLifecycle.java
View File

@@ -45,11 +45,20 @@ public interface LoadBalancerLifecycle<RC, RES, T> {
*/
void onStart(Request<RC> request);
/**
* A callback method executed after a service instance has been selected, before
* executing the actual load-balanced request.
* @param request the {@link Request} that has been used by the LoadBalancer to select
* a service instance
* @param lbResponse the {@link Response} returned by the LoadBalancer
*/
void onStartRequest(Request<RC> request, Response<T> lbResponse);
/**
* A callback method executed after load-balancing.
* @param completionContext the {@link CompletionContext} containing data relevant to
* the load-balancing and the response returned from the selected service instance
*/
void onComplete(CompletionContext<RES, T> completionContext);
void onComplete(CompletionContext<RES, T, RC> completionContext);
}

46
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/LoadBalancerRequestAdapter.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* Copyright 2012-2020 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
*
* https://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.client.loadbalancer;
import org.springframework.cloud.client.ServiceInstance;
/**
* An adapter class that allows creating {@link Request} objects from previously
* {@link LoadBalancerRequest} objects.
*
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
public class LoadBalancerRequestAdapter<T, RC> extends DefaultRequest<RC> implements LoadBalancerRequest<T> {
private final LoadBalancerRequest<T> delegate;
public LoadBalancerRequestAdapter(LoadBalancerRequest<T> delegate) {
this.delegate = delegate;
}
public LoadBalancerRequestAdapter(LoadBalancerRequest<T> delegate, RC context) {
super(context);
this.delegate = delegate;
}
@Override
public T apply(ServiceInstance instance) throws Exception {
return delegate.apply(instance);
}
}

4
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/RequestDataContext.java
View File

@@ -26,6 +26,10 @@ import org.springframework.http.HttpMethod;
*/
public class RequestDataContext extends DefaultRequestContext {
public RequestDataContext() {
super();
}
public RequestDataContext(RequestData requestData) {
this(requestData, "default");
}

39
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/ResponseData.java
View File

@@ -16,13 +16,18 @@
package org.springframework.cloud.client.loadbalancer;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import org.springframework.core.style.ToStringCreator;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseCookie;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.client.ClientResponse;
@@ -59,6 +64,11 @@ public class ResponseData {
this(response.getStatusCode(), response.getHeaders(), response.getCookies(), requestData);
}
public ResponseData(ClientHttpResponse clientHttpResponse, RequestData requestData) throws IOException {
this(clientHttpResponse.getStatusCode(), clientHttpResponse.getHeaders(),
buildCookiesFromHeaders(clientHttpResponse.getHeaders()), requestData);
}
public HttpStatus getHttpStatus() {
return httpStatus;
}
@@ -82,6 +92,30 @@ public class ResponseData {
return to.toString();
}
static MultiValueMap<String, ResponseCookie> buildCookiesFromHeaders(HttpHeaders headers) {
LinkedMultiValueMap<String, ResponseCookie> newCookies = new LinkedMultiValueMap<>();
if (headers == null) {
return newCookies;
}
List<String> cookiesFromHeaders = headers.get(HttpHeaders.COOKIE);
if (cookiesFromHeaders != null) {
cookiesFromHeaders.forEach(cookie -> {
String[] splitCookie = cookie.split("=");
if (splitCookie.length < 2) {
return;
}
newCookies.put(splitCookie[0],
Collections.singletonList(ResponseCookie.from(splitCookie[0], splitCookie[1]).build()));
});
}
return newCookies;
}
@Override
public int hashCode() {
return Objects.hash(httpStatus, headers, cookies, requestData);
}
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -95,9 +129,4 @@ public class ResponseData {
&& Objects.equals(cookies, that.cookies) && Objects.equals(requestData, that.requestData);
}
@Override
public int hashCode() {
return Objects.hash(httpStatus, headers, cookies, requestData);
}
}

25
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/RetryLoadBalancerInterceptor.java
View File

@@ -89,7 +89,8 @@ public class RetryLoadBalancerInterceptor implements ClientHttpRequestIntercepto
Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(
loadBalancerFactory.getInstances(serviceName, LoadBalancerLifecycle.class),
RequestDataContext.class, ResponseData.class, ServiceInstance.class);
RetryableRequestContext.class, ResponseData.class, ServiceInstance.class);
String hint = getHint(serviceName);
if (serviceInstance == null) {
if (LOG.isDebugEnabled()) {
LOG.debug("Service instance retrieved from LoadBalancedRetryContext: was null. "
@@ -100,7 +101,6 @@ public class RetryLoadBalancerInterceptor implements ClientHttpRequestIntercepto
LoadBalancedRetryContext lbContext = (LoadBalancedRetryContext) context;
previousServiceInstance = lbContext.getPreviousServiceInstance();
}
String hint = getHint(serviceName);
DefaultRequest<RetryableRequestContext> lbRequest = new DefaultRequest<>(
new RetryableRequestContext(previousServiceInstance, new RequestData(request), hint));
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
@@ -112,15 +112,22 @@ public class RetryLoadBalancerInterceptor implements ClientHttpRequestIntercepto
LoadBalancedRetryContext lbContext = (LoadBalancedRetryContext) context;
lbContext.setServiceInstance(serviceInstance);
}
Response<ServiceInstance> lbResponse = new DefaultResponse(serviceInstance);
if (serviceInstance == null) {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<ResponseData, ServiceInstance, RetryableRequestContext>(
CompletionContext.Status.DISCARD,
new DefaultRequest<>(
new RetryableRequestContext(null, new RequestData(request), hint)),
lbResponse)));
}
}
Response<ServiceInstance> lbResponse = new DefaultResponse(serviceInstance);
if (serviceInstance == null) {
supportedLifecycleProcessors
.forEach(lifecycle -> lifecycle.onComplete(new CompletionContext<ResponseData, ServiceInstance>(
CompletionContext.Status.DISCARD, lbResponse)));
}
LoadBalancerRequestAdapter<ClientHttpResponse, RetryableRequestContext> lbRequest = new LoadBalancerRequestAdapter<>(
requestFactory.createRequest(request, body, execution),
new RetryableRequestContext(null, new RequestData(request), hint));
ServiceInstance finalServiceInstance = serviceInstance;
ClientHttpResponse response = RetryLoadBalancerInterceptor.this.loadBalancer.execute(serviceName,
serviceInstance, requestFactory.createRequest(request, body, execution));
finalServiceInstance, lbRequest);
int statusCode = response.getRawStatusCode();
if (retryPolicy != null && retryPolicy.retryableStatusCode(statusCode)) {
if (LOG.isDebugEnabled()) {

17
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/RetryableRequestContext.java
View File

@@ -27,21 +27,22 @@ import org.springframework.core.style.ToStringCreator;
*
* @author Olga Maciaszek-Sharma
*/
public class RetryableRequestContext extends DefaultRequestContext {
public class RetryableRequestContext extends RequestDataContext {
private final ServiceInstance previousServiceInstance;
private ServiceInstance previousServiceInstance;
public RetryableRequestContext(ServiceInstance previousServiceInstance) {
this.previousServiceInstance = previousServiceInstance;
}
public RetryableRequestContext(ServiceInstance previousServiceInstance, Object clientRequest) {
super(clientRequest);
public RetryableRequestContext(ServiceInstance previousServiceInstance, RequestData clientRequestData) {
super(clientRequestData);
this.previousServiceInstance = previousServiceInstance;
}
public RetryableRequestContext(ServiceInstance previousServiceInstance, Object clientRequest, String hint) {
super(clientRequest, hint);
public RetryableRequestContext(ServiceInstance previousServiceInstance, RequestData clientRequestData,
String hint) {
super(clientRequestData, hint);
this.previousServiceInstance = previousServiceInstance;
}
@@ -49,6 +50,10 @@ public class RetryableRequestContext extends DefaultRequestContext {
return previousServiceInstance;
}
public void setPreviousServiceInstance(ServiceInstance previousServiceInstance) {
this.previousServiceInstance = previousServiceInstance;
}
@Override
public String toString() {
ToStringCreator to = new ToStringCreator(this);

31
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/TimedRequestContext.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2012-2020 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
*
* https://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.client.loadbalancer;
/**
* Allows setting and retrieving request start time.
*
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
public interface TimedRequestContext {
long getRequestStartTime();
void setRequestStartTime(long requestStartTime);
}

11
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerExchangeFilterFunction.java
View File

@@ -94,7 +94,7 @@ public class ReactorLoadBalancerExchangeFilterFunction implements LoadBalancedEx
LOG.warn(message);
}
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, lbResponse)));
.onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, lbResponse)));
return Mono.just(ClientResponse.create(HttpStatus.SERVICE_UNAVAILABLE)
.body(serviceInstanceUnavailableMessage(serviceId)).build());
}
@@ -107,13 +107,14 @@ public class ReactorLoadBalancerExchangeFilterFunction implements LoadBalancedEx
ClientRequest newRequest = buildClientRequest(clientRequest, instance,
stickySessionProperties.getInstanceIdCookieName(),
stickySessionProperties.isAddServiceInstanceCookie());
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, lbResponse));
return next.exchange(newRequest)
.doOnError(throwable -> supportedLifecycleProcessors.forEach(
lifecycle -> lifecycle.onComplete(new CompletionContext<ResponseData, ServiceInstance>(
CompletionContext.Status.FAILED, throwable, lbResponse))))
.doOnError(throwable -> supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<ResponseData, ServiceInstance, RequestDataContext>(
CompletionContext.Status.FAILED, throwable, lbRequest, lbResponse))))
.doOnSuccess(clientResponse -> supportedLifecycleProcessors.forEach(
lifecycle -> lifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS,
lbResponse, new ResponseData(clientResponse, requestData)))));
lbRequest, lbResponse, new ResponseData(clientResponse, requestData)))));
});
}

17
spring-cloud-commons/src/main/java/org/springframework/cloud/client/loadbalancer/reactive/RetryableLoadBalancerExchangeFilterFunction.java
View File

@@ -39,7 +39,6 @@ import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycleValida
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.RequestData;
import org.springframework.cloud.client.loadbalancer.RequestDataContext;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.cloud.client.loadbalancer.RetryableRequestContext;
@@ -102,7 +101,7 @@ public class RetryableLoadBalancerExchangeFilterFunction implements LoadBalanced
Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(
loadBalancerFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
RequestDataContext.class, ResponseData.class, ServiceInstance.class);
RetryableRequestContext.class, ResponseData.class, ServiceInstance.class);
String hint = getHint(serviceId, properties.getHint());
RequestData requestData = new RequestData(clientRequest);
DefaultRequest<RetryableRequestContext> lbRequest = new DefaultRequest<>(
@@ -110,14 +109,15 @@ public class RetryableLoadBalancerExchangeFilterFunction implements LoadBalanced
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
return Mono.defer(() -> choose(serviceId, lbRequest).flatMap(lbResponse -> {
ServiceInstance instance = lbResponse.getServer();
lbRequest.setContext(new RetryableRequestContext(instance, clientRequest, hint));
lbRequest.setContext(new RetryableRequestContext(instance, requestData, hint));
if (instance == null) {
String message = serviceInstanceUnavailableMessage(serviceId);
if (LOG.isWarnEnabled()) {
LOG.warn(message);
}
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, lbResponse)));
.onComplete(new CompletionContext<ResponseData, ServiceInstance, RetryableRequestContext>(
CompletionContext.Status.DISCARD, lbRequest, lbResponse)));
return Mono.just(ClientResponse.create(HttpStatus.SERVICE_UNAVAILABLE)
.body(serviceInstanceUnavailableMessage(serviceId)).build());
}
@@ -130,13 +130,14 @@ public class RetryableLoadBalancerExchangeFilterFunction implements LoadBalanced
ClientRequest newRequest = buildClientRequest(clientRequest, instance,
stickySessionProperties.getInstanceIdCookieName(),
stickySessionProperties.isAddServiceInstanceCookie());
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, lbResponse));
return next.exchange(newRequest)
.doOnError(throwable -> supportedLifecycleProcessors.forEach(
lifecycle -> lifecycle.onComplete(new CompletionContext<ResponseData, ServiceInstance>(
CompletionContext.Status.FAILED, throwable, lbResponse))))
.doOnError(throwable -> supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<ResponseData, ServiceInstance, RetryableRequestContext>(
CompletionContext.Status.FAILED, throwable, lbRequest, lbResponse))))
.doOnSuccess(clientResponse -> supportedLifecycleProcessors.forEach(
lifecycle -> lifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS,
lbResponse, new ResponseData(clientResponse, requestData)))))
lbRequest, lbResponse, new ResponseData(clientResponse, requestData)))))
.map(clientResponse -> {
loadBalancerRetryContext.setClientResponse(clientResponse);
if (shouldRetrySameServiceInstance(loadBalancerRetryContext)) {

32
spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/RetryLoadBalancerInterceptorTests.java
View File

@@ -373,16 +373,26 @@ public class RetryLoadBalancerInterceptorTests {
interceptor.intercept(request, new byte[] {}, mock(ClientHttpRequestExecution.class));
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("testLifecycle")).getStartLog()).hasSize(1);
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("testLifecycle")).getStartRequestLog())
.hasSize(0);
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("testLifecycle")).getCompleteLog()).hasSize(0);
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("anotherLifecycle")).getStartLog()).hasSize(1);
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("anotherLifecycle")).getStartRequestLog())
.hasSize(0);
assertThat(((TestLoadBalancerLifecycle) lifecycleProcessors.get("anotherLifecycle")).getCompleteLog())
.hasSize(0);
assertThat(((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("testLifecycle")).getStartLog())
.hasSize(0);
assertThat(
((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("testLifecycle")).getStartRequestLog())
.hasSize(1);
assertThat(((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("testLifecycle")).getCompleteLog())
.hasSize(1);
assertThat(((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("anotherLifecycle")).getStartLog())
.hasSize(0);
assertThat(
((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("testLifecycle")).getStartRequestLog())
.hasSize(1);
assertThat(
((TestLoadBalancerLifecycle) client.getLifecycleProcessors().get("anotherLifecycle")).getCompleteLog())
.hasSize(1);
@@ -499,11 +509,12 @@ public class RetryLoadBalancerInterceptorTests {
Set<LoadBalancerLifecycle> supportedLoadBalancerProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(lifecycleProcessors, DefaultRequestContext.class, Object.class,
ServiceInstance.class);
supportedLoadBalancerProcessors.forEach(lifecycle -> lifecycle.onStartRequest(new DefaultRequest<>(),
new DefaultResponse(serviceInstance)));
T response = (T) new MockClientHttpResponse(new byte[] {}, HttpStatus.OK);
supportedLoadBalancerProcessors
.forEach(lifecycle -> lifecycle.onComplete(new CompletionContext(CompletionContext.Status.SUCCESS,
new DefaultResponse(defaultServiceInstance()))));
new DefaultRequest<>(), new DefaultResponse(defaultServiceInstance()))));
return response;
}
@@ -532,7 +543,9 @@ public class RetryLoadBalancerInterceptorTests {
final ConcurrentHashMap<String, Request<Object>> startLog = new ConcurrentHashMap<>();
final ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> completeLog = new ConcurrentHashMap<>();
final ConcurrentHashMap<String, Request<Object>> startRequestLog = new ConcurrentHashMap<>();
final ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance, Object>> completeLog = new ConcurrentHashMap<>();
@Override
public boolean supports(Class requestContextClass, Class responseClass, Class serverTypeClass) {
@@ -547,7 +560,12 @@ public class RetryLoadBalancerInterceptorTests {
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance> completionContext) {
public void onStartRequest(Request<Object> request, Response<ServiceInstance> lbResponse) {
startRequestLog.put(getName() + UUID.randomUUID(), request);
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance, Object> completionContext) {
completeLog.put(getName() + UUID.randomUUID(), completionContext);
}
@@ -555,10 +573,14 @@ public class RetryLoadBalancerInterceptorTests {
return startLog;
}
ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> getCompleteLog() {
ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance, Object>> getCompleteLog() {
return completeLog;
}
ConcurrentHashMap<String, Request<Object>> getStartRequestLog() {
return startRequestLog;
}
protected String getName() {
return getClass().getSimpleName();
}

31
spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/ReactorLoadBalancerExchangeFilterFunctionTests.java
View File

@@ -43,6 +43,7 @@ import org.springframework.cloud.client.loadbalancer.DefaultRequestContext;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycle;
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpStatus;
@@ -128,19 +129,24 @@ class ReactorLoadBalancerExchangeFilterFunctionTests {
void loadBalancerLifecycleCallbacksExecuted() {
final String callbackTestHint = "callbackTestHint";
loadBalancerProperties.getHint().put("testservice", "callbackTestHint");
final String result = "callbackTestResult";
ClientResponse clientResponse = WebClient.builder().baseUrl("http://testservice").filter(loadBalancerFunction)
.build().get().uri("/callback").exchange().block();
Collection<Request<Object>> lifecycleLogRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle")).getStartLog()
.values();
Collection<CompletionContext<Object, ServiceInstance>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
Collection<Request<Object>> lifecycleStartedLogRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle"))
.getStartRequestLog().values();
Collection<CompletionContext<Object, ServiceInstance, Object>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("anotherLoadBalancerLifecycle"))
.getCompleteLog().values();
then(clientResponse.statusCode()).isEqualTo(HttpStatus.OK);
assertThat(lifecycleLogRequests).extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(lifecycleStartedLogRequests)
.extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(anotherLifecycleLogRequests)
.extracting(completionContext -> ((ResponseData) completionContext.getClientResponse()).getRequestData()
.getUrl().toString())
@@ -204,9 +210,11 @@ class ReactorLoadBalancerExchangeFilterFunctionTests {
protected static class TestLoadBalancerLifecycle implements LoadBalancerLifecycle<Object, Object, ServiceInstance> {
ConcurrentHashMap<String, Request<Object>> startLog = new ConcurrentHashMap<>();
final Map<String, Request<Object>> startLog = new ConcurrentHashMap<>();
ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> completeLog = new ConcurrentHashMap<>();
final Map<String, Request<Object>> startRequestLog = new ConcurrentHashMap<>();
final Map<String, CompletionContext<Object, ServiceInstance, Object>> completeLog = new ConcurrentHashMap<>();
@Override
public void onStart(Request<Object> request) {
@@ -214,18 +222,27 @@ class ReactorLoadBalancerExchangeFilterFunctionTests {
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance> completionContext) {
public void onStartRequest(Request<Object> request, Response<ServiceInstance> lbResponse) {
startRequestLog.put(getName() + UUID.randomUUID(), request);
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance, Object> completionContext) {
completeLog.put(getName() + UUID.randomUUID(), completionContext);
}
ConcurrentHashMap<String, Request<Object>> getStartLog() {
Map<String, Request<Object>> getStartLog() {
return startLog;
}
ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> getCompleteLog() {
Map<String, CompletionContext<Object, ServiceInstance, Object>> getCompleteLog() {
return completeLog;
}
Map<String, Request<Object>> getStartRequestLog() {
return startRequestLog;
}
protected String getName() {
return this.getClass().getSimpleName();
}

26
spring-cloud-commons/src/test/java/org/springframework/cloud/client/loadbalancer/reactive/RetryableLoadBalancerExchangeFilterFunctionIntegrationTests.java
View File

@@ -45,6 +45,7 @@ import org.springframework.cloud.client.loadbalancer.DefaultRequestContext;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycle;
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpMethod;
@@ -108,12 +109,18 @@ class RetryableLoadBalancerExchangeFilterFunctionIntegrationTests {
Collection<Request<Object>> lifecycleLogRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle")).getStartLog()
.values();
Collection<CompletionContext<Object, ServiceInstance>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
Collection<Request<Object>> lifecycleLogStartRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle"))
.getStartRequestLog().values();
Collection<CompletionContext<Object, ServiceInstance, Object>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
.getInstances("testservice", LoadBalancerLifecycle.class).get("anotherLoadBalancerLifecycle"))
.getCompleteLog().values();
then(clientResponse.statusCode()).isEqualTo(HttpStatus.OK);
assertThat(lifecycleLogRequests).extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(lifecycleLogStartRequests)
.extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(anotherLifecycleLogRequests)
.extracting(completionContext -> ((ResponseData) completionContext.getClientResponse()).getRequestData()
.getHttpMethod())
@@ -270,7 +277,9 @@ class RetryableLoadBalancerExchangeFilterFunctionIntegrationTests {
Map<String, Request<Object>> startLog = new ConcurrentHashMap<>();
Map<String, CompletionContext<Object, ServiceInstance>> completeLog = new ConcurrentHashMap<>();
Map<String, Request<Object>> startRequestLog = new ConcurrentHashMap<>();
Map<String, CompletionContext<Object, ServiceInstance, Object>> completeLog = new ConcurrentHashMap<>();
@Override
public void onStart(Request<Object> request) {
@@ -278,7 +287,12 @@ class RetryableLoadBalancerExchangeFilterFunctionIntegrationTests {
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance> completionContext) {
public void onStartRequest(Request<Object> request, Response<ServiceInstance> lbResponse) {
startRequestLog.put(getName() + UUID.randomUUID(), request);
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance, Object> completionContext) {
completeLog.clear();
completeLog.put(getName() + UUID.randomUUID(), completionContext);
}
@@ -287,10 +301,14 @@ class RetryableLoadBalancerExchangeFilterFunctionIntegrationTests {
return startLog;
}
Map<String, CompletionContext<Object, ServiceInstance>> getCompleteLog() {
Map<String, CompletionContext<Object, ServiceInstance, Object>> getCompleteLog() {
return completeLog;
}
Map<String, Request<Object>> getStartRequestLog() {
return startRequestLog;
}
protected String getName() {
return this.getClass().getSimpleName();
}

5
spring-cloud-loadbalancer/pom.xml
View File

@@ -87,6 +87,11 @@
<artifactId>spring-security-oauth2-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

41
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/blocking/client/BlockingLoadBalancerClient.java
View File

@@ -33,11 +33,14 @@ import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycle;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycleValidator;
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequest;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequestAdapter;
import org.springframework.cloud.client.loadbalancer.LoadBalancerUriTools;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer;
import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.util.ReflectionUtils;
import static org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer.REQUEST;
@@ -65,7 +68,7 @@ public class BlockingLoadBalancerClient implements LoadBalancerClient {
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
String hint = getHint(serviceId);
DefaultRequest<DefaultRequestContext> lbRequest = new DefaultRequest<>(
LoadBalancerRequestAdapter<T, DefaultRequestContext> lbRequest = new LoadBalancerRequestAdapter<>(request,
new DefaultRequestContext(request, hint));
Set<LoadBalancerLifecycle> supportedLifecycleProcessors = LoadBalancerLifecycleValidator
.getSupportedLifecycleProcessors(
@@ -74,11 +77,11 @@ public class BlockingLoadBalancerClient implements LoadBalancerClient {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
ServiceInstance serviceInstance = choose(serviceId, lbRequest);
if (serviceInstance == null) {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, new EmptyResponse())));
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, new EmptyResponse())));
throw new IllegalStateException("No instances available for " + serviceId);
}
return execute(serviceId, serviceInstance, request);
return execute(serviceId, serviceInstance, lbRequest);
}
@Override
@@ -89,25 +92,45 @@ public class BlockingLoadBalancerClient implements LoadBalancerClient {
.getSupportedLifecycleProcessors(
loadBalancerClientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
DefaultRequestContext.class, Object.class, ServiceInstance.class);
Request lbRequest = request instanceof Request ? (Request) request : new DefaultRequest<>();
supportedLifecycleProcessors
.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, new DefaultResponse(serviceInstance)));
try {
T response = request.apply(serviceInstance);
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS, defaultResponse, response)));
Object clientResponse = getClientResponse(response);
supportedLifecycleProcessors
.forEach(lifecycle -> lifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS,
lbRequest, defaultResponse, clientResponse)));
return response;
}
catch (IOException iOException) {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.FAILED, iOException, defaultResponse)));
new CompletionContext<>(CompletionContext.Status.FAILED, iOException, lbRequest, defaultResponse)));
throw iOException;
}
catch (Exception exception) {
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.FAILED, exception, defaultResponse)));
supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.FAILED, exception, lbRequest, defaultResponse)));
ReflectionUtils.rethrowRuntimeException(exception);
}
return null;
}
private <T> Object getClientResponse(T response) {
ClientHttpResponse clientHttpResponse = null;
if (response instanceof ClientHttpResponse) {
clientHttpResponse = (ClientHttpResponse) response;
}
if (clientHttpResponse != null) {
try {
return new ResponseData(clientHttpResponse, null);
}
catch (IOException ignored) {
}
}
return response;
}
@Override
public URI reconstructURI(ServiceInstance serviceInstance, URI original) {
return LoadBalancerUriTools.reconstructURI(serviceInstance, original);

45
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/config/LoadBalancerStatsAutoConfiguration.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* Copyright 2012-2020 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
*
* https://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.loadbalancer.config;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.loadbalancer.stats.MicrometerStatsLoadBalancerLifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* Autoconfiguration that provides a {@link MicrometerStatsLoadBalancerLifecycle} bean.
*
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(MeterRegistry.class)
@ConditionalOnProperty(value = "spring.cloud.loadbalancer.stats.micrometer.enabled", havingValue = "true")
public class LoadBalancerStatsAutoConfiguration {
@Bean
@ConditionalOnBean(MeterRegistry.class)
public MicrometerStatsLoadBalancerLifecycle micrometerStatsLifecycle(MeterRegistry meterRegistry) {
return new MicrometerStatsLoadBalancerLifecycle(meterRegistry);
}
}

160
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/stats/LoadBalancerTags.java Normal file
View File

@@ -0,0 +1,160 @@
/*
* Copyright 2012-2020 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
*
* https://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.loadbalancer.stats;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.CompletionContext;
import org.springframework.cloud.client.loadbalancer.RequestData;
import org.springframework.cloud.client.loadbalancer.RequestDataContext;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.util.StringUtils;
/**
* Utility class for building metrics tags for load-balanced calls.
*
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
final class LoadBalancerTags {
static final String UNKNOWN = "UNKNOWN";
private LoadBalancerTags() {
throw new UnsupportedOperationException("Cannot instantiate utility class");
}
static Iterable<Tag> buildSuccessRequestTags(CompletionContext<Object, ServiceInstance, Object> completionContext) {
ServiceInstance serviceInstance = completionContext.getLoadBalancerResponse().getServer();
Tags tags = Tags.of(buildServiceInstanceTags(serviceInstance));
Object clientResponse = completionContext.getClientResponse();
if (clientResponse instanceof ResponseData) {
ResponseData responseData = (ResponseData) clientResponse;
RequestData requestData = responseData.getRequestData();
if (requestData != null) {
tags = tags.and(valueOrUnknown("method", requestData.getHttpMethod()),
valueOrUnknown("uri", getPath(requestData)));
}
else {
tags = tags.and(Tag.of("method", UNKNOWN), Tag.of("uri", UNKNOWN));
}
tags = tags.and(Tag.of("outcome", forStatus(statusValue(responseData))),
valueOrUnknown("status", statusValue(responseData)));
}
else {
tags = tags.and(Tag.of("method", UNKNOWN), Tag.of("uri", UNKNOWN), Tag.of("outcome", UNKNOWN),
Tag.of("status", UNKNOWN));
}
return tags;
}
// In keeping with the way null HttpStatus is handled in Actuator
private static int statusValue(ResponseData responseData) {
return responseData.getHttpStatus() != null ? responseData.getHttpStatus().value() : 200;
}
private static String getPath(RequestData requestData) {
return requestData.getUrl() != null ? requestData.getUrl().getPath() : UNKNOWN;
}
static Iterable<Tag> buildDiscardedRequestTags(
CompletionContext<Object, ServiceInstance, Object> completionContext) {
if (completionContext.getLoadBalancerRequest().getContext() instanceof RequestDataContext) {
RequestData requestData = ((RequestDataContext) completionContext.getLoadBalancerRequest().getContext())
.getClientRequest();
if (requestData != null) {
return Tags.of(valueOrUnknown("method", requestData.getHttpMethod()),
valueOrUnknown("uri", getPath(requestData)), valueOrUnknown("serviceId", getHost(requestData)));
}
}
return Tags.of(valueOrUnknown("method", UNKNOWN), valueOrUnknown("uri", UNKNOWN),
valueOrUnknown("serviceId", UNKNOWN));
}
private static String getHost(RequestData requestData) {
return requestData.getUrl() != null ? requestData.getUrl().getHost() : UNKNOWN;
}
static Iterable<Tag> buildFailedRequestTags(CompletionContext<Object, ServiceInstance, Object> completionContext) {
ServiceInstance serviceInstance = completionContext.getLoadBalancerResponse().getServer();
Tags tags = Tags.of(buildServiceInstanceTags(serviceInstance)).and(exception(completionContext.getThrowable()));
if (completionContext.getLoadBalancerRequest().getContext() instanceof RequestDataContext) {
RequestData requestData = ((RequestDataContext) completionContext.getLoadBalancerRequest().getContext())
.getClientRequest();
if (requestData != null) {
return tags.and(Tags.of(valueOrUnknown("method", requestData.getHttpMethod()),
valueOrUnknown("uri", getPath(requestData))));
}
}
return tags.and(Tags.of(valueOrUnknown("method", UNKNOWN), valueOrUnknown("uri", UNKNOWN)));
}
static Iterable<Tag> buildServiceInstanceTags(ServiceInstance serviceInstance) {
return Tags.of(valueOrUnknown("serviceId", serviceInstance.getServiceId()),
valueOrUnknown("serviceInstance.instanceId", serviceInstance.getInstanceId()),
valueOrUnknown("serviceInstance.host", serviceInstance.getHost()),
valueOrUnknown("serviceInstance.port", String.valueOf(serviceInstance.getPort())));
}
private static Tag valueOrUnknown(String key, String value) {
if (value != null) {
return Tag.of(key, value);
}
return Tag.of(key, UNKNOWN);
}
private static Tag valueOrUnknown(String key, Object value) {
if (value != null) {
return Tag.of(key, String.valueOf(value));
}
return Tag.of(key, UNKNOWN);
}
private static Tag exception(Throwable exception) {
if (exception != null) {
String simpleName = exception.getClass().getSimpleName();
return Tag.of("exception", StringUtils.hasText(simpleName) ? simpleName : exception.getClass().getName());
}
return Tag.of("exception", "None");
}
// Logic from Actuator's `Outcome` class. Copied in here to avoid adding Actuator
// dependency.
public static String forStatus(int status) {
if (status >= 100 && status < 200) {
return "INFORMATIONAL";
}
else if (status >= 200 && status < 300) {
return "SUCCESS";
}
else if (status >= 300 && status < 400) {
return "REDIRECTION";
}
else if (status >= 400 && status < 500) {
return "CLIENT_ERROR";
}
else if (status >= 500 && status < 600) {
return "SERVER_ERROR";
}
return UNKNOWN;
}
}

121
spring-cloud-loadbalancer/src/main/java/org/springframework/cloud/loadbalancer/stats/MicrometerStatsLoadBalancerLifecycle.java Normal file
View File

@@ -0,0 +1,121 @@
/*
* Copyright 2012-2020 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
*
* https://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.loadbalancer.stats;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.CompletionContext;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycle;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.TimedRequestContext;
import static org.springframework.cloud.loadbalancer.stats.LoadBalancerTags.buildDiscardedRequestTags;
import static org.springframework.cloud.loadbalancer.stats.LoadBalancerTags.buildFailedRequestTags;
import static org.springframework.cloud.loadbalancer.stats.LoadBalancerTags.buildServiceInstanceTags;
import static org.springframework.cloud.loadbalancer.stats.LoadBalancerTags.buildSuccessRequestTags;
/**
* An implementation of {@link LoadBalancerLifecycle} that records metrics for
* load-balanced calls.
*
* @author Olga Maciaszek-Sharma
* @since 3.0.0
*/
public class MicrometerStatsLoadBalancerLifecycle implements LoadBalancerLifecycle<Object, Object, ServiceInstance> {
private final MeterRegistry meterRegistry;
private final ConcurrentHashMap<ServiceInstance, AtomicLong> activeRequestsPerInstance = new ConcurrentHashMap<>();
public MicrometerStatsLoadBalancerLifecycle(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
@Override
public boolean supports(Class requestContextClass, Class responseClass, Class serverTypeClass) {
return ServiceInstance.class.isAssignableFrom(serverTypeClass);
}
@Override
public void onStart(Request<Object> request) {
// do nothing
}
@Override
public void onStartRequest(Request<Object> request, Response<ServiceInstance> lbResponse) {
if (request.getContext() instanceof TimedRequestContext) {
((TimedRequestContext) request.getContext()).setRequestStartTime(System.nanoTime());
}
if (!lbResponse.hasServer()) {
return;
}
ServiceInstance serviceInstance = lbResponse.getServer();
AtomicLong activeRequestsCounter = activeRequestsPerInstance.computeIfAbsent(serviceInstance, instance -> {
AtomicLong createdCounter = new AtomicLong();
Gauge.builder("loadbalancer.requests.active", () -> createdCounter)
.tags(buildServiceInstanceTags(serviceInstance)).register(meterRegistry);
return createdCounter;
});
activeRequestsCounter.incrementAndGet();
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance, Object> completionContext) {
long requestFinishedTimestamp = System.nanoTime();
if (CompletionContext.Status.DISCARD.equals(completionContext.status())) {
Counter.builder("loadbalancer.requests.discard").tags(buildDiscardedRequestTags(completionContext))
.register(meterRegistry).increment();
return;
}
ServiceInstance serviceInstance = completionContext.getLoadBalancerResponse().getServer();
AtomicLong activeRequestsCounter = activeRequestsPerInstance.get(serviceInstance);
if (activeRequestsCounter != null) {
activeRequestsCounter.decrementAndGet();
}
Object loadBalancerRequestContext = completionContext.getLoadBalancerRequest().getContext();
if (requestHasBeenTimed(loadBalancerRequestContext)) {
if (CompletionContext.Status.FAILED.equals(completionContext.status())) {
Timer.builder("loadbalancer.requests.failed").tags(buildFailedRequestTags(completionContext))
.register(meterRegistry)
.record(requestFinishedTimestamp
- ((TimedRequestContext) loadBalancerRequestContext).getRequestStartTime(),
TimeUnit.NANOSECONDS);
return;
}
Timer.builder("loadbalancer.requests.success").tags(buildSuccessRequestTags(completionContext))
.register(meterRegistry)
.record(requestFinishedTimestamp
- ((TimedRequestContext) loadBalancerRequestContext).getRequestStartTime(),
TimeUnit.NANOSECONDS);
}
}
private boolean requestHasBeenTimed(Object loadBalancerRequestContext) {
return loadBalancerRequestContext instanceof TimedRequestContext
&& (((TimedRequestContext) loadBalancerRequestContext).getRequestStartTime() != 0L);
}
}

3
spring-cloud-loadbalancer/src/main/resources/META-INF/spring.factories
View File

@@ -3,4 +3,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.loadbalancer.config.LoadBalancerAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.BlockingLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.LoadBalancerCacheAutoConfiguration,\
org.springframework.cloud.loadbalancer.security.OAuth2LoadBalancerClientAutoConfiguration
org.springframework.cloud.loadbalancer.security.OAuth2LoadBalancerClientAutoConfiguration,\
org.springframework.cloud.loadbalancer.config.LoadBalancerStatsAutoConfiguration

30
spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/blocking/client/BlockingLoadBalancerClientTests.java
View File

@@ -20,6 +20,7 @@ import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@@ -173,12 +174,18 @@ class BlockingLoadBalancerClientTests {
Collection<Request<Object>> lifecycleLogRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("myservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle")).getStartLog()
.values();
Collection<CompletionContext<Object, ServiceInstance>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
Collection<Request<Object>> lifecycleLogStartedRequests = ((TestLoadBalancerLifecycle) factory
.getInstances("myservice", LoadBalancerLifecycle.class).get("loadBalancerLifecycle"))
.getStartRequestLog().values();
Collection<CompletionContext<Object, ServiceInstance, Object>> anotherLifecycleLogRequests = ((AnotherLoadBalancerLifecycle) factory
.getInstances("myservice", LoadBalancerLifecycle.class).get("anotherLoadBalancerLifecycle"))
.getCompleteLog().values();
assertThat(actualResult).isEqualTo(result);
assertThat(lifecycleLogRequests).extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(lifecycleLogStartedRequests)
.extracting(request -> ((DefaultRequestContext) request.getContext()).getHint())
.contains(callbackTestHint);
assertThat(anotherLifecycleLogRequests).extracting(CompletionContext::getClientResponse).contains(result);
}
@@ -231,9 +238,11 @@ class BlockingLoadBalancerClientTests {
protected static class TestLoadBalancerLifecycle implements LoadBalancerLifecycle<Object, Object, ServiceInstance> {
final ConcurrentHashMap<String, Request<Object>> startLog = new ConcurrentHashMap<>();
final Map<String, Request<Object>> startLog = new ConcurrentHashMap<>();
final ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> completeLog = new ConcurrentHashMap<>();
final Map<String, Request<Object>> startRequestLog = new ConcurrentHashMap<>();
final Map<String, CompletionContext<Object, ServiceInstance, Object>> completeLog = new ConcurrentHashMap<>();
@Override
public void onStart(Request<Object> request) {
@@ -241,18 +250,27 @@ class BlockingLoadBalancerClientTests {
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance> completionContext) {
public void onStartRequest(Request<Object> request, Response<ServiceInstance> lbResponse) {
startRequestLog.put(getName() + UUID.randomUUID(), request);
}
@Override
public void onComplete(CompletionContext<Object, ServiceInstance, Object> completionContext) {
completeLog.put(getName() + UUID.randomUUID(), completionContext);
}
ConcurrentHashMap<String, Request<Object>> getStartLog() {
Map<String, Request<Object>> getStartLog() {
return startLog;
}
ConcurrentHashMap<String, CompletionContext<Object, ServiceInstance>> getCompleteLog() {
Map<String, CompletionContext<Object, ServiceInstance, Object>> getCompleteLog() {
return completeLog;
}
Map<String, Request<Object>> getStartRequestLog() {
return startRequestLog;
}
protected String getName() {
return this.getClass().getSimpleName();
}

182
spring-cloud-loadbalancer/src/test/java/org/springframework/cloud/loadbalancer/stats/MicrometerStatsLoadBalancerLifecycleTests.java Normal file
View File

@@ -0,0 +1,182 @@
/*
* Copyright 2012-2020 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
*
* https://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.loadbalancer.stats;
import java.net.URI;
import java.util.HashMap;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.junit.jupiter.api.Test;
import org.springframework.cloud.client.DefaultServiceInstance;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.CompletionContext;
import org.springframework.cloud.client.loadbalancer.DefaultRequest;
import org.springframework.cloud.client.loadbalancer.DefaultRequestContext;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.RequestData;
import org.springframework.cloud.client.loadbalancer.RequestDataContext;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.util.MultiValueMapAdapter;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.cloud.loadbalancer.stats.LoadBalancerTags.UNKNOWN;
/**
* Tests for {@link MicrometerStatsLoadBalancerLifecycle}.
*
* @author Olga Maciaszek-Sharma
*/
class MicrometerStatsLoadBalancerLifecycleTests {
MeterRegistry meterRegistry = new SimpleMeterRegistry();
MicrometerStatsLoadBalancerLifecycle statsLifecycle = new MicrometerStatsLoadBalancerLifecycle(meterRegistry);
@Test
void shouldRecordSuccessfulTimedRequest() {
RequestData requestData = new RequestData(HttpMethod.GET, URI.create("http://test.org/test"), new HttpHeaders(),
new HttpHeaders(), new HashMap<>());
Request<Object> lbRequest = new DefaultRequest<>(new RequestDataContext(requestData));
Response<ServiceInstance> lbResponse = new DefaultResponse(
new DefaultServiceInstance("test-1", "test", "test.org", 8080, false, new HashMap<>()));
ResponseData responseData = new ResponseData(HttpStatus.OK, new HttpHeaders(),
new MultiValueMapAdapter<>(new HashMap<>()), requestData);
statsLifecycle.onStartRequest(lbRequest, lbResponse);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(1);
statsLifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.SUCCESS, lbRequest, lbResponse, responseData));
assertThat(meterRegistry.getMeters()).hasSize(2);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(0);
assertThat(meterRegistry.get("loadbalancer.requests.success").timers()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().count()).isEqualTo(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().getId().getTags()).contains(
Tag.of("method", "GET"), Tag.of("outcome", "SUCCESS"), Tag.of("serviceId", "test"),
Tag.of("serviceInstance.host", "test.org"), Tag.of("serviceInstance.instanceId", "test-1"),
Tag.of("serviceInstance.port", "8080"), Tag.of("status", "200"), Tag.of("uri", "/test"));
}
@Test
void shouldRecordFailedTimedRequest() {
RequestData requestData = new RequestData(HttpMethod.GET, URI.create("http://test.org/test"), new HttpHeaders(),
new HttpHeaders(), new HashMap<>());
Request<Object> lbRequest = new DefaultRequest<>(new RequestDataContext(requestData));
Response<ServiceInstance> lbResponse = new DefaultResponse(
new DefaultServiceInstance("test-1", "test", "test.org", 8080, false, new HashMap<>()));
statsLifecycle.onStartRequest(lbRequest, lbResponse);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(1);
statsLifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.FAILED, new IllegalStateException(),
lbRequest, lbResponse));
assertThat(meterRegistry.getMeters()).hasSize(2);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(0);
assertThat(meterRegistry.get("loadbalancer.requests.failed").timers()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.failed").timer().count()).isEqualTo(1);
assertThat(meterRegistry.get("loadbalancer.requests.failed").timer().getId().getTags()).contains(
Tag.of("exception", "IllegalStateException"), Tag.of("method", "GET"), Tag.of("serviceId", "test"),
Tag.of("serviceInstance.host", "test.org"), Tag.of("serviceInstance.instanceId", "test-1"),
Tag.of("serviceInstance.port", "8080"), Tag.of("uri", "/test"));
}
@Test
void shouldNotRecordDiscardedRequest() {
RequestData requestData = new RequestData(HttpMethod.GET, URI.create("http://test.org/test"), new HttpHeaders(),
new HttpHeaders(), new HashMap<>());
Request<Object> lbRequest = new DefaultRequest<>(new RequestDataContext(requestData));
Response<ServiceInstance> lbResponse = new EmptyResponse();
statsLifecycle.onStartRequest(lbRequest, lbResponse);
statsLifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, lbResponse));
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.discard").counter().count()).isEqualTo(1);
}
@Test
void shouldNotRecordUnTimedRequest() {
Request<Object> lbRequest = new DefaultRequest<>(new StatsTestContext());
Response<ServiceInstance> lbResponse = new DefaultResponse(
new DefaultServiceInstance("test-1", "test", "test.org", 8080, false, new HashMap<>()));
ResponseData responseData = new ResponseData(HttpStatus.OK, new HttpHeaders(),
new MultiValueMapAdapter<>(new HashMap<>()), null);
statsLifecycle.onStartRequest(lbRequest, lbResponse);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(1);
statsLifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.SUCCESS, lbRequest, lbResponse, responseData));
assertThat(meterRegistry.getMeters()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(0);
}
@Test
void shouldNotCreateNullTagsWhenNullDataObjects() {
Request<Object> lbRequest = new DefaultRequest<>(new DefaultRequestContext());
Response<ServiceInstance> lbResponse = new DefaultResponse(new DefaultServiceInstance());
statsLifecycle.onStartRequest(lbRequest, lbResponse);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(1);
statsLifecycle
.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS, lbRequest, lbResponse, null));
assertThat(meterRegistry.getMeters()).hasSize(2);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(0);
assertThat(meterRegistry.get("loadbalancer.requests.success").timers()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().count()).isEqualTo(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().getId().getTags()).contains(
Tag.of("method", UNKNOWN), Tag.of("outcome", UNKNOWN), Tag.of("serviceId", UNKNOWN),
Tag.of("serviceInstance.host", UNKNOWN), Tag.of("serviceInstance.instanceId", UNKNOWN),
Tag.of("serviceInstance.port", "0"), Tag.of("status", UNKNOWN), Tag.of("uri", UNKNOWN));
}
@Test
void shouldNotCreateNullTagsWhenEmptyDataObjects() {
RequestData requestData = new RequestData(null, null, null, null, null);
Request<Object> lbRequest = new DefaultRequest<>(new RequestDataContext());
Response<ServiceInstance> lbResponse = new DefaultResponse(new DefaultServiceInstance());
ResponseData responseData = new ResponseData(null, null, null, requestData);
statsLifecycle.onStartRequest(lbRequest, lbResponse);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(1);
statsLifecycle.onComplete(
new CompletionContext<>(CompletionContext.Status.SUCCESS, lbRequest, lbResponse, responseData));
assertThat(meterRegistry.getMeters()).hasSize(2);
assertThat(meterRegistry.get("loadbalancer.requests.active").gauge().value()).isEqualTo(0);
assertThat(meterRegistry.get("loadbalancer.requests.success").timers()).hasSize(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().count()).isEqualTo(1);
assertThat(meterRegistry.get("loadbalancer.requests.success").timer().getId().getTags()).contains(
Tag.of("method", UNKNOWN), Tag.of("outcome", "SUCCESS"), Tag.of("serviceId", UNKNOWN),
Tag.of("serviceInstance.host", UNKNOWN), Tag.of("serviceInstance.instanceId", UNKNOWN),
Tag.of("serviceInstance.port", "0"), Tag.of("status", "200"), Tag.of("uri", UNKNOWN));
}
private static class StatsTestContext {
}
}