Skip to content

Commit f32b8b9

Browse files
liubao68liubao68
authored
[#1005]improve instance isolation policies when instance isolated (#1006)
1 parent 7a30a74 commit f32b8b9

File tree

6 files changed

+222
-63
lines changed

6 files changed

+222
-63
lines changed

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/feign/GovernanceFeignBlockingLoadBalancerClient.java‎

Lines changed: 43 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import java.io.IOException;
2020
import java.net.URI;
2121
import java.nio.charset.StandardCharsets;
22+
import java.time.Duration;
2223
import java.util.ArrayList;
2324
import java.util.Arrays;
2425
import java.util.Collection;
@@ -37,7 +38,6 @@
3738
import org.apache.servicecomb.governance.processor.injection.FaultInjectionDecorators;
3839
import org.apache.servicecomb.governance.processor.injection.FaultInjectionDecorators.FaultInjectionDecorateCheckedSupplier;
3940
import org.apache.servicecomb.http.client.common.HttpUtils;
40-
import org.apache.servicecomb.service.center.client.DiscoveryEvents.PullInstanceEvent;
4141
import org.slf4j.Logger;
4242
import org.slf4j.LoggerFactory;
4343
import org.springframework.cloud.client.ServiceInstance;
@@ -60,9 +60,10 @@
6060
import com.huaweicloud.common.context.InvocationContext;
6161
import com.huaweicloud.common.context.InvocationContextHolder;
6262
import com.huaweicloud.common.context.InvocationStage;
63-
import com.huaweicloud.common.event.EventManager;
6463
import com.huaweicloud.common.disovery.InstanceIDAdapter;
64+
import com.huaweicloud.common.event.EventManager;
6565
import com.huaweicloud.governance.adapters.loadbalancer.RetryContext;
66+
import com.huaweicloud.governance.event.InstanceIsolatedEvent;
6667

6768
import io.github.resilience4j.bulkhead.Bulkhead;
6869
import io.github.resilience4j.bulkhead.BulkheadFullException;
@@ -161,15 +162,15 @@ private Response decorateWithFault(Request request, Options options, URI origina
161162
headers.put("Content-Type", Arrays.asList("application/json"));
162163
if (result == null) {
163164
return Response.builder().status(200)
164-
.request(request)
165-
.headers(headers)
166-
.build();
165+
.request(request)
166+
.headers(headers)
167+
.build();
167168
}
168169
return Response.builder().status(200)
169-
.request(request)
170-
.headers(headers)
171-
.body(HttpUtils.serialize(result).getBytes(
172-
StandardCharsets.UTF_8)).build();
170+
.request(request)
171+
.headers(headers)
172+
.body(HttpUtils.serialize(result).getBytes(
173+
StandardCharsets.UTF_8)).build();
173174
}
174175
} catch (Throwable e) {
175176
throw new RuntimeException(e);
@@ -336,42 +337,47 @@ private Response executeWithInstanceIsolation(GovernanceRequestExtractor governa
336337
org.springframework.cloud.client.loadbalancer.Response<ServiceInstance> lbResponse,
337338
Set<LoadBalancerLifecycle> supportedLifecycleProcessors, boolean useRawStatusCodes) {
338339

339-
try {
340-
CircuitBreakerPolicy circuitBreakerPolicy = instanceIsolationHandler.matchPolicy(governanceRequest);
341-
if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
342-
return Response.builder().status(503)
343-
.reason("Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests").request(feignRequest)
344-
.build();
340+
CircuitBreakerPolicy circuitBreakerPolicy = instanceIsolationHandler.matchPolicy(governanceRequest);
341+
if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
342+
return Response.builder().status(503)
343+
.reason("Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests").request(feignRequest)
344+
.build();
345+
}
346+
347+
if (circuitBreakerPolicy != null && !circuitBreakerPolicy.isForceClosed()) {
348+
CircuitBreaker circuitBreaker = instanceIsolationHandler.getActuator(governanceRequest);
349+
if (circuitBreaker == null) {
350+
return executeWithInstanceBulkhead(governanceRequest, feignClient, options, feignRequest, lbRequest,
351+
lbResponse, supportedLifecycleProcessors, useRawStatusCodes);
345352
}
346353

347-
if (circuitBreakerPolicy != null && !circuitBreakerPolicy.isForceClosed()) {
348-
CircuitBreaker circuitBreaker = instanceIsolationHandler.getActuator(governanceRequest);
349-
if (circuitBreaker == null) {
350-
return executeWithInstanceBulkhead(governanceRequest, feignClient, options, feignRequest, lbRequest,
351-
lbResponse, supportedLifecycleProcessors, useRawStatusCodes);
352-
}
354+
CheckedFunction0<Response> next = () -> executeWithInstanceBulkhead(governanceRequest, feignClient, options,
355+
feignRequest, lbRequest, lbResponse,
356+
supportedLifecycleProcessors, useRawStatusCodes);
353357

354-
CheckedFunction0<Response> next = () -> executeWithInstanceBulkhead(governanceRequest, feignClient, options,
355-
feignRequest, lbRequest, lbResponse,
356-
supportedLifecycleProcessors, useRawStatusCodes);
358+
DecorateCheckedSupplier<Response> dcs = Decorators.ofCheckedSupplier(next);
359+
dcs.withCircuitBreaker(circuitBreaker);
357360

358-
DecorateCheckedSupplier<Response> dcs = Decorators.ofCheckedSupplier(next);
359-
dcs.withCircuitBreaker(circuitBreaker);
361+
try {
360362
return dcs.get();
361-
}
363+
} catch (Throwable e) {
364+
if (e instanceof CallNotPermittedException) {
365+
// when instance isolated, request to pull instances.
366+
LOG.error("instance isolated [{}], [{}]", governanceRequest.instanceId(), e.getMessage());
367+
EventManager.post(new InstanceIsolatedEvent(governanceRequest.instanceId(),
368+
Duration.parse(circuitBreakerPolicy.getWaitDurationInOpenState())));
369+
return Response.builder().status(503).reason("instance isolated.").request(feignRequest).build();
370+
}
362371

363-
return executeWithInstanceBulkhead(governanceRequest, feignClient, options, feignRequest, lbRequest,
364-
lbResponse, supportedLifecycleProcessors, useRawStatusCodes);
365-
} catch (Throwable e) {
366-
if (e instanceof CallNotPermittedException) {
367-
// when instance isolated, request to pull instances.
368-
LOG.error("instance isolated [{}]", governanceRequest.instanceId());
369-
EventManager.post(new PullInstanceEvent());
370-
return Response.builder().status(503).reason("instance isolated.").request(feignRequest).build();
372+
if (e instanceof RuntimeException) {
373+
throw (RuntimeException) e;
374+
}
375+
throw new RuntimeException(e);
371376
}
372-
373-
throw new RuntimeException(e);
374377
}
378+
379+
return executeWithInstanceBulkhead(governanceRequest, feignClient, options, feignRequest, lbRequest,
380+
lbResponse, supportedLifecycleProcessors, useRawStatusCodes);
375381
}
376382

377383
private Response executeWithInstanceBulkhead(GovernanceRequestExtractor governanceRequest, Client feignClient,

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/gateway/InstanceIsolationGlobalFilter.java‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,11 @@
1717

1818
package com.huaweicloud.governance.adapters.gateway;
1919

20+
import java.time.Duration;
21+
2022
import org.apache.servicecomb.governance.handler.InstanceIsolationHandler;
2123
import org.apache.servicecomb.governance.marker.GovernanceRequestExtractor;
2224
import org.apache.servicecomb.governance.policy.CircuitBreakerPolicy;
23-
import org.apache.servicecomb.service.center.client.DiscoveryEvents.PullInstanceEvent;
2425
import org.slf4j.Logger;
2526
import org.slf4j.LoggerFactory;
2627
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
@@ -32,6 +33,7 @@
3233
import org.springframework.web.server.ServerWebExchange;
3334

3435
import com.huaweicloud.common.event.EventManager;
36+
import com.huaweicloud.governance.event.InstanceIsolatedEvent;
3537

3638
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
3739
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
@@ -66,6 +68,12 @@ public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
6668

6769
private Mono<Void> addInstanceIsolation(ServerWebExchange exchange, GovernanceRequestExtractor governanceRequest,
6870
Mono<Void> toRun) {
71+
CircuitBreakerPolicy circuitBreakerPolicy = isolationHandler.matchPolicy(governanceRequest);
72+
if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
73+
return Mono.error(new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
74+
"Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests"));
75+
}
76+
6977
CircuitBreaker circuitBreaker = isolationHandler.getActuator(governanceRequest);
7078
Mono<Void> mono = toRun;
7179
if (circuitBreaker != null) {
@@ -74,8 +82,9 @@ private Mono<Void> addInstanceIsolation(ServerWebExchange exchange, GovernanceRe
7482
.transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
7583
.then()
7684
.onErrorResume(CallNotPermittedException.class, (t) -> {
77-
LOGGER.error("instance isolated [{}]", t.getMessage());
78-
EventManager.post(new PullInstanceEvent());
85+
LOGGER.error("instance isolated [{}], [{}]", governanceRequest.instanceId(), t.getMessage());
86+
EventManager.post(new InstanceIsolatedEvent(governanceRequest.instanceId(),
87+
Duration.parse(circuitBreakerPolicy.getWaitDurationInOpenState())));
7988
return Mono.error(new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
8089
"instance isolated.", t));
8190
});

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/loadbalancer/InstanceIsolationServiceInstanceFilter.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
/*
2+
3+
* Copyright (C) 2020-2022 Huawei Technologies Co., Ltd. All rights reserved.
4+
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package com.huaweicloud.governance.adapters.loadbalancer;
18+
19+
import java.util.ArrayList;
20+
import java.util.Iterator;
21+
import java.util.List;
22+
import java.util.Map;
23+
import java.util.concurrent.ConcurrentHashMap;
24+
25+
import org.springframework.cloud.client.ServiceInstance;
26+
import org.springframework.cloud.client.loadbalancer.Request;
27+
import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier;
28+
import org.springframework.core.Ordered;
29+
30+
import com.google.common.eventbus.Subscribe;
31+
import com.huaweicloud.common.disovery.InstanceIDAdapter;
32+
import com.huaweicloud.common.event.EventManager;
33+
import com.huaweicloud.governance.event.InstanceIsolatedEvent;
34+
35+
public class InstanceIsolationServiceInstanceFilter implements ServiceInstanceFilter {
36+
private final Object lock = new Object();
37+
38+
private final Map<String, Long> isolatedInstances = new ConcurrentHashMap<>();
39+
40+
public InstanceIsolationServiceInstanceFilter() {
41+
EventManager.register(this);
42+
}
43+
44+
@Subscribe
45+
public void onInstanceIsolatedEvent(InstanceIsolatedEvent event) {
46+
synchronized (lock) {
47+
for (Iterator<String> iterator = isolatedInstances.keySet().iterator(); iterator.hasNext(); ) {
48+
Long duration = isolatedInstances.get(iterator.next());
49+
if (System.currentTimeMillis() - duration > 0) {
50+
iterator.remove();
51+
}
52+
}
53+
54+
isolatedInstances.put(event.getInstanceId(),
55+
System.currentTimeMillis() + event.getWaitDurationInHalfOpenState().toMillis());
56+
}
57+
}
58+
59+
@Override
60+
public List<ServiceInstance> filter(ServiceInstanceListSupplier supplier, List<ServiceInstance> instances,
61+
Request<?> request) {
62+
if (isolatedInstances.isEmpty() || instances.isEmpty()) {
63+
return instances;
64+
}
65+
List<ServiceInstance> result = new ArrayList<>(instances.size());
66+
for (ServiceInstance serviceInstance : instances) {
67+
Long duration = isolatedInstances.get(InstanceIDAdapter.instanceId(serviceInstance));
68+
if (duration == null) {
69+
result.add(serviceInstance);
70+
continue;
71+
}
72+
73+
if (System.currentTimeMillis() - duration < 0) {
74+
continue;
75+
}
76+
77+
synchronized (lock) {
78+
isolatedInstances.remove(InstanceIDAdapter.instanceId(serviceInstance));
79+
}
80+
result.add(serviceInstance);
81+
}
82+
83+
if (result.isEmpty()) {
84+
return instances;
85+
}
86+
return result;
87+
}
88+
89+
@Override
90+
public int getOrder() {
91+
return Ordered.LOWEST_PRECEDENCE;
92+
}
93+
}

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/loadbalancer/LoadbalancerConfiguration.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,4 +47,9 @@ public void setTransformers(List<LoadBalancerRequestTransformer> transformers) {
4747
public LoadBalancerRequestFactory loadBalancerRequestFactory(LoadBalancerClient loadBalancerClient) {
4848
return new DecorateLoadBalancerRequestFactory(loadBalancerClient, this.transformers);
4949
}
50+
51+
@Bean
52+
public InstanceIsolationServiceInstanceFilter instanceIsolationServiceInstanceFilter() {
53+
return new InstanceIsolationServiceInstanceFilter();
54+
}
5055
}

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/adapters/web/IsolationClientHttpRequestInterceptor.java‎

Lines changed: 31 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,12 @@
1717

1818
package com.huaweicloud.governance.adapters.web;
1919

20+
import java.io.IOException;
21+
import java.time.Duration;
22+
2023
import org.apache.servicecomb.governance.handler.InstanceIsolationHandler;
2124
import org.apache.servicecomb.governance.marker.GovernanceRequestExtractor;
2225
import org.apache.servicecomb.governance.policy.CircuitBreakerPolicy;
23-
import org.apache.servicecomb.service.center.client.DiscoveryEvents.PullInstanceEvent;
2426
import org.slf4j.Logger;
2527
import org.slf4j.LoggerFactory;
2628
import org.springframework.core.Ordered;
@@ -31,6 +33,7 @@
3133

3234
import com.huaweicloud.common.adapters.web.FallbackClientHttpResponse;
3335
import com.huaweicloud.common.event.EventManager;
36+
import com.huaweicloud.governance.event.InstanceIsolatedEvent;
3437

3538
import io.github.resilience4j.circuitbreaker.CallNotPermittedException;
3639
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
@@ -50,34 +53,39 @@ public IsolationClientHttpRequestInterceptor(InstanceIsolationHandler instanceIs
5053
}
5154

5255
@Override
53-
public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution) {
56+
public ClientHttpResponse intercept(HttpRequest request, byte[] body, ClientHttpRequestExecution execution)
57+
throws IOException {
5458
GovernanceRequestExtractor governanceRequest = RestTemplateUtils.createGovernanceRequest(request);
55-
try {
56-
CircuitBreakerPolicy circuitBreakerPolicy = instanceIsolationHandler.matchPolicy(governanceRequest);
57-
if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
58-
return new FallbackClientHttpResponse(503,
59-
"Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests");
60-
}
6159

62-
if (circuitBreakerPolicy != null && !circuitBreakerPolicy.isForceClosed()) {
63-
CircuitBreaker circuitBreaker = instanceIsolationHandler.getActuator(governanceRequest);
64-
if (circuitBreaker != null) {
65-
CheckedFunction0<ClientHttpResponse> next = () -> execution.execute(request, body);
66-
DecorateCheckedSupplier<ClientHttpResponse> dcs = Decorators.ofCheckedSupplier(next);
67-
dcs.withCircuitBreaker(circuitBreaker);
60+
CircuitBreakerPolicy circuitBreakerPolicy = instanceIsolationHandler.matchPolicy(governanceRequest);
61+
if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
62+
return new FallbackClientHttpResponse(503,
63+
"Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests");
64+
}
65+
66+
if (circuitBreakerPolicy != null && !circuitBreakerPolicy.isForceClosed()) {
67+
CircuitBreaker circuitBreaker = instanceIsolationHandler.getActuator(governanceRequest);
68+
if (circuitBreaker != null) {
69+
CheckedFunction0<ClientHttpResponse> next = () -> execution.execute(request, body);
70+
DecorateCheckedSupplier<ClientHttpResponse> dcs = Decorators.ofCheckedSupplier(next);
71+
dcs.withCircuitBreaker(circuitBreaker);
72+
try {
6873
return dcs.get();
74+
} catch (Throwable e) {
75+
if (e instanceof CallNotPermittedException) {
76+
LOG.error("instance isolated [{}], [{}]", governanceRequest.instanceId(), e.getMessage());
77+
EventManager.post(new InstanceIsolatedEvent(governanceRequest.instanceId(),
78+
Duration.parse(circuitBreakerPolicy.getWaitDurationInOpenState())));
79+
return new FallbackClientHttpResponse(503, "instance isolated");
80+
}
81+
if (e instanceof RuntimeException) {
82+
throw (RuntimeException) e;
83+
}
84+
throw new RuntimeException(e);
6985
}
7086
}
71-
return execution.execute(request, body);
72-
} catch (Throwable e) {
73-
if (e instanceof CallNotPermittedException) {
74-
// when instance isolated, request to pull instances.
75-
LOG.warn("instance isolated [{}]", governanceRequest.instanceId());
76-
EventManager.post(new PullInstanceEvent());
77-
return new FallbackClientHttpResponse(503, "instance isolated");
78-
}
79-
throw new RuntimeException(e);
8087
}
88+
return execution.execute(request, body);
8189
}
8290

8391
@Override

‎spring-cloud-huawei-governance/src/main/java/com/huaweicloud/governance/event/InstanceIsolatedEvent.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
/*
2+
3+
* Copyright (C) 2020-2022 Huawei Technologies Co., Ltd. All rights reserved.
4+
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package com.huaweicloud.governance.event;
18+
19+
import java.time.Duration;
20+
21+
public class InstanceIsolatedEvent {
22+
private final String instanceId;
23+
24+
private final Duration waitDurationInHalfOpenState;
25+
26+
public InstanceIsolatedEvent(String instanceId, Duration waitDurationInHalfOpenState) {
27+
this.instanceId = instanceId;
28+
this.waitDurationInHalfOpenState = waitDurationInHalfOpenState;
29+
}
30+
31+
public String getInstanceId() {
32+
return instanceId;
33+
}
34+
35+
public Duration getWaitDurationInHalfOpenState() {
36+
return waitDurationInHalfOpenState;
37+
}
38+
}

0 commit comments

Comments
 (0)