This is an automated email from the ASF dual-hosted git repository. liubao pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/servicecomb-java-chassis.git
commit 4167f3b03b3b47cb0e415717ee5c2a47c8090271 Author: liubao <[email protected]> AuthorDate: Tue Jan 5 10:59:04 2021 +0800 [SCB-2116]add consumer implementation and add test cases --- .../pom.xml | 4 + .../demo/zeroconfig/client/GovernanceEndpoint.java | 61 ++++++++ .../src/main/resources/application.yml | 65 ++++++++- .../pom.xml | 4 + .../demo/zeroconfig/server/GovernanceEndpoint.java | 67 +++++++++ .../src/main/resources/application.yml | 8 +- .../demo/zeroconfig/tests/GovernanceTest.java | 157 +++++++++++++++++++++ .../src/main/resources/registry.yaml | 1 + dependencies/bom/pom.xml | 5 + distribution/pom.xml | 4 + .../servicecomb/governance/policy/RetryPolicy.java | 4 +- .../governance/ConsumerGovernanceHandler.java | 157 +++++++++++++++++++++ .../governance/ProviderGovernanceHandler.java | 46 ++++-- .../governance/ServiceCombInvocationContext.java | 60 ++++++++ .../governance/ServiceCombRetryExtension.java | 69 +++++++++ .../src/main/resources/config/cse.handler.xml | 2 + .../loadbalance/LoadbalanceHandler.java | 14 +- solutions/solution-basic/pom.xml | 4 + 18 files changed, 711 insertions(+), 21 deletions(-) diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/pom.xml b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/pom.xml index f087a38..0f154b1 100644 --- a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/pom.xml +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/pom.xml @@ -42,6 +42,10 @@ </dependency> <dependency> <groupId>org.apache.servicecomb</groupId> + <artifactId>handler-governance</artifactId> + </dependency> + <dependency> + <groupId>org.apache.servicecomb</groupId> <artifactId>registry-schema-discovery</artifactId> </dependency> <dependency> diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/java/org/apache/servicecomb/demo/zeroconfig/client/GovernanceEndpoint.java b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/java/org/apache/servicecomb/demo/zeroconfig/client/GovernanceEndpoint.java new file mode 100644 index 0000000..0d99ffe --- /dev/null +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/java/org/apache/servicecomb/demo/zeroconfig/client/GovernanceEndpoint.java @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.demo.zeroconfig.client; + +import org.apache.servicecomb.provider.rest.common.RestSchema; +import org.apache.servicecomb.provider.springmvc.reference.RestTemplateBuilder; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.client.RestTemplate; + +@RestSchema(schemaId = "GovernanceEndpoint") +@RequestMapping("/governance") +public class GovernanceEndpoint { + private static final String SERVER = "servicecomb://demo-zeroconfig-schemadiscovery-registry-server"; + + private RestTemplate restTemplate = RestTemplateBuilder.create(); + + private int count = 0; + + @GetMapping("/hello") + public String hello() { + return restTemplate.getForObject(SERVER + "/governance/hello", String.class); + } + + @GetMapping("/retry") + public String retry(@RequestParam(name = "invocationID") String invocationID) { + return restTemplate + .getForObject(SERVER + "/governance/retry?invocationID={1}", String.class, + invocationID); + } + + @GetMapping("/circuitBreaker") + public String circuitBreaker() throws Exception { + count++; + if (count % 3 == 0) { + return "ok"; + } + throw new RuntimeException("test error"); + } + + @GetMapping("/bulkhead") + public String bulkhead() { + return restTemplate.getForObject(SERVER + "/governance/hello", String.class); + } +} diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/resources/application.yml b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/resources/application.yml index 6bbfed6..3d8453f 100644 --- a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/resources/application.yml +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-client/src/main/resources/application.yml @@ -25,9 +25,68 @@ service_description: name: demo-zeroconfig-schemadiscovery-registry-client version: 0.0.2 servicecomb: + rest: + address: 0.0.0.0:8082 + handler: chain: Consumer: - default: loadbalance - rest: - address: 0.0.0.0:8082 \ No newline at end of file + default: governance-consumer,loadbalance + Provider: + default: governance-provider + + matchGroup: + demo-rateLimiting: | + matches: + - apiPath: + exact: "GovernanceEndpoint.hello" + name: xx + ## services is optional in configuration file + services: demo-zeroconfig-schemadiscovery-registry-client + demo-retry: | + matches: + - apiPath: + exact: "GovernanceEndpoint.retry" + name: xx + ## services is optional in configuration file + services: demo-zeroconfig-schemadiscovery-registry-client + demo-circuitBreaker: | + matches: + - apiPath: + exact: "GovernanceEndpoint.circuitBreaker" + name: xx + ## services is optional in configuration file + services: demo-zeroconfig-schemadiscovery-registry-client + demo-bulkhead: | + matches: + - apiPath: + exact: "GovernanceEndpoint.bulkhead" + name: xx + ## services is optional in configuration file + services: demo-zeroconfig-schemadiscovery-registry-client + rateLimiting: + demo-rateLimiting: | + rules: + match: demo-rateLimiting.xx + rate: 10 + name: xx + retry: + demo-retry: | + rules: + match: demo-retry.xx + maxAttempts: 3 + name: xx + circuitBreaker: + demo-circuitBreaker: | + rules: + match: demo-circuitBreaker.xx + minimumNumberOfCalls: 10 + slidingWindowSize: 10 + failureRateThreshold: 20 + name: xx + bulkhead: + demo-bulkhead: | + rules: + match: demo-bulkhead.xx + maxConcurrentCalls: 5 + name: xx \ No newline at end of file diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/pom.xml b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/pom.xml index 77d7783..4ad699f 100644 --- a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/pom.xml +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/pom.xml @@ -42,6 +42,10 @@ </dependency> <dependency> <groupId>org.apache.servicecomb</groupId> + <artifactId>handler-governance</artifactId> + </dependency> + <dependency> + <groupId>org.apache.servicecomb</groupId> <artifactId>registry-schema-discovery</artifactId> </dependency> <dependency> diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/java/org/apache/servicecomb/demo/zeroconfig/server/GovernanceEndpoint.java b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/java/org/apache/servicecomb/demo/zeroconfig/server/GovernanceEndpoint.java new file mode 100644 index 0000000..718a242 --- /dev/null +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/java/org/apache/servicecomb/demo/zeroconfig/server/GovernanceEndpoint.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.demo.zeroconfig.server; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.servicecomb.provider.rest.common.RestSchema; +import org.apache.servicecomb.swagger.invocation.exception.InvocationException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; + +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; + +@RestSchema(schemaId = "GovernanceEndpoint") +@RequestMapping("/governance") +public class GovernanceEndpoint { + private static final Logger LOGGER = LoggerFactory.getLogger(GovernanceEndpoint.class); + + private Map<String, Integer> retryTimes = new HashMap<>(); + + @GetMapping("/hello") + public String sayHello() { + return "Hello world!"; + } + + @GetMapping("/retry") + @ApiResponses({ + @ApiResponse(code = 200, response = String.class, message = ""), + @ApiResponse(code = 502, response = String.class, message = "")}) + public String retry(@RequestParam(name = "invocationID") String invocationID) { + LOGGER.info("invoke service: {}", invocationID); + retryTimes.putIfAbsent(invocationID, 0); + retryTimes.put(invocationID, retryTimes.get(invocationID) + 1); + + int retry = retryTimes.get(invocationID); + + if (retry == 3) { + return "try times: " + retry; + } + throw new InvocationException(502, "retry", "retry"); + } + + @GetMapping("/circuitBreaker") + public String circuitBreaker() { + throw new RuntimeException("circuitBreaker by provider."); + } +} diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/resources/application.yml b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/resources/application.yml index 99e7de1..9f6d98b 100644 --- a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/resources/application.yml +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-server/src/main/resources/application.yml @@ -26,4 +26,10 @@ service_description: version: 0.0.2 servicecomb: rest: - address: 0.0.0.0:8080 \ No newline at end of file + address: 0.0.0.0:8080 + handler: + chain: + Consumer: + default: governance-consumer,loadbalance + Provider: + default: governance-provider \ No newline at end of file diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/java/org/apache/servicecomb/demo/zeroconfig/tests/GovernanceTest.java b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/java/org/apache/servicecomb/demo/zeroconfig/tests/GovernanceTest.java new file mode 100644 index 0000000..40161bf --- /dev/null +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/java/org/apache/servicecomb/demo/zeroconfig/tests/GovernanceTest.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.demo.zeroconfig.tests; + +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.servicecomb.demo.CategorizedTestCase; +import org.apache.servicecomb.demo.TestMgr; +import org.apache.servicecomb.provider.springmvc.reference.RestTemplateBuilder; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestTemplate; + +@Component +public class GovernanceTest implements CategorizedTestCase { + String url = "servicecomb://demo-zeroconfig-schemadiscovery-registry-client/governance"; + + RestTemplate template = RestTemplateBuilder.create(); + + @Override + public void testRestTransport() throws Exception { + testCircuitBreaker(); + testBulkhead(); + testRateLimiting(); + testRetry(); + } + + private void testRetry() { + String invocationID = UUID.randomUUID().toString(); + String result = template.getForObject(url + "/retry?invocationID={1}", String.class, invocationID); + TestMgr.check(result, "try times: 3"); + } + + private void testCircuitBreaker() throws Exception { + CountDownLatch latch = new CountDownLatch(100); + AtomicBoolean expectedFailed = new AtomicBoolean(false); + AtomicBoolean notExpectedFailed = new AtomicBoolean(false); + + for (int i = 0; i < 10; i++) { + for (int j = 0; j < 10; j++) { + String name = "t-" + i + "-" + j; + new Thread(name) { + public void run() { + try { + String result = template.getForObject(url + "/circuitBreaker", String.class); + if (!"ok".equals(result)) { + notExpectedFailed.set(true); + } + } catch (Exception e) { + if (!"InvocationException: code=429;msg=CommonExceptionData [message=circuitBreaker is open.]" + .equals(e.getMessage()) + && !e.getMessage().equals( + "InvocationException: code=500;msg={message=Unexpected exception when processing the request.}")) { + notExpectedFailed.set(true); + } + if ("InvocationException: code=429;msg=CommonExceptionData [message=circuitBreaker is open.]" + .equals(e.getMessage())) { + expectedFailed.set(true); + } + } + latch.countDown(); + } + }.start(); + } + Thread.sleep(100); + } + + latch.await(20, TimeUnit.SECONDS); + TestMgr.check(true, expectedFailed.get()); + TestMgr.check(false, notExpectedFailed.get()); + } + + private void testBulkhead() throws Exception { + CountDownLatch latch = new CountDownLatch(100); + AtomicBoolean expectedFailed = new AtomicBoolean(false); + AtomicBoolean notExpectedFailed = new AtomicBoolean(false); + + for (int i = 0; i < 10; i++) { + for (int j = 0; j < 10; j++) { + String name = "t-" + i + "-" + j; + new Thread(name) { + public void run() { + try { + String result = template.getForObject(url + "/bulkhead", String.class); + if (!"Hello world!".equals(result)) { + notExpectedFailed.set(true); + } + } catch (Exception e) { + if (!"InvocationException: code=429;msg=CommonExceptionData [message=bulkhead is full and does not permit further calls.]" + .equals(e.getMessage())) { + notExpectedFailed.set(true); + } + expectedFailed.set(true); + } + latch.countDown(); + } + }.start(); + } + Thread.sleep(100); + } + + latch.await(20, TimeUnit.SECONDS); + TestMgr.check(true, expectedFailed.get()); + TestMgr.check(false, notExpectedFailed.get()); + } + + private void testRateLimiting() throws Exception { + CountDownLatch latch = new CountDownLatch(100); + AtomicBoolean expectedFailed = new AtomicBoolean(false); + AtomicBoolean notExpectedFailed = new AtomicBoolean(false); + + for (int i = 0; i < 10; i++) { + for (int j = 0; j < 10; j++) { + String name = "t-" + i + "-" + j; + new Thread(name) { + public void run() { + try { + String result = template.getForObject(url + "/hello", String.class); + if (!"Hello world!".equals(result)) { + notExpectedFailed.set(true); + } + } catch (Exception e) { + if (!"InvocationException: code=429;msg=CommonExceptionData [message=rate limited.]" + .equals(e.getMessage())) { + notExpectedFailed.set(true); + } + expectedFailed.set(true); + } + latch.countDown(); + } + }.start(); + } + Thread.sleep(100); + } + + latch.await(20, TimeUnit.SECONDS); + TestMgr.check(true, expectedFailed.get()); + TestMgr.check(false, notExpectedFailed.get()); + } +} diff --git a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/resources/registry.yaml b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/resources/registry.yaml index df728fc..cdd8fc7 100644 --- a/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/resources/registry.yaml +++ b/demo/demo-zeroconfig-schemadiscovery-registry/demo-zeroconfig-schemadiscovery-registry-tests/src/main/resources/registry.yaml @@ -22,6 +22,7 @@ demo-zeroconfig-schemadiscovery-registry-client: schemaIds: - ClientServerEndpoint - SchemaDiscoveryEndpoint + - GovernanceEndpoint instances: - endpoints: - rest://localhost:8082 diff --git a/dependencies/bom/pom.xml b/dependencies/bom/pom.xml index 56c7626..f1a877e 100644 --- a/dependencies/bom/pom.xml +++ b/dependencies/bom/pom.xml @@ -205,6 +205,11 @@ </dependency> <dependency> <groupId>org.apache.servicecomb</groupId> + <artifactId>handler-governance</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.servicecomb</groupId> <artifactId>handler-loadbalance</artifactId> <version>${project.version}</version> </dependency> diff --git a/distribution/pom.xml b/distribution/pom.xml index 9b070d2..7a5793b 100644 --- a/distribution/pom.xml +++ b/distribution/pom.xml @@ -170,6 +170,10 @@ </dependency> <dependency> <groupId>org.apache.servicecomb</groupId> + <artifactId>handler-governance</artifactId> + </dependency> + <dependency> + <groupId>org.apache.servicecomb</groupId> <artifactId>handler-loadbalance</artifactId> </dependency> <dependency> diff --git a/governance/src/main/java/org/apache/servicecomb/governance/policy/RetryPolicy.java b/governance/src/main/java/org/apache/servicecomb/governance/policy/RetryPolicy.java index 4efd065..b6ec4b9 100644 --- a/governance/src/main/java/org/apache/servicecomb/governance/policy/RetryPolicy.java +++ b/governance/src/main/java/org/apache/servicecomb/governance/policy/RetryPolicy.java @@ -30,14 +30,14 @@ public class RetryPolicy extends AbstractPolicy { public static final int DEFAULT_MAX_ATTEMPTS = 3; - public static final int DEFAULT_WAIT_DURATION = 0; + public static final int DEFAULT_WAIT_DURATION = 1; public static final String DEFAULT_RETRY_ON_RESPONSE_STATUS = "502"; //最多尝试次数 private int maxAttempts = DEFAULT_MAX_ATTEMPTS; - //每次重试尝试等待的时间，默认给0 + //每次重试尝试等待的时间，默认给1。 在异步场景下，这个值必须大于0，否则不会重试。 private int waitDuration = DEFAULT_WAIT_DURATION; //需要重试的http status, 逗号分隔 diff --git a/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ConsumerGovernanceHandler.java b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ConsumerGovernanceHandler.java new file mode 100644 index 0000000..8443e73 --- /dev/null +++ b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ConsumerGovernanceHandler.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.handler.governance; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import org.apache.servicecomb.core.Handler; +import org.apache.servicecomb.core.Invocation; +import org.apache.servicecomb.core.provider.consumer.SyncResponseExecutor; +import org.apache.servicecomb.foundation.common.utils.BeanUtils; +import org.apache.servicecomb.governance.MatchersManager; +import org.apache.servicecomb.governance.handler.RetryHandler; +import org.apache.servicecomb.governance.marker.GovHttpRequest; +import org.apache.servicecomb.governance.policy.RetryPolicy; +import org.apache.servicecomb.governance.properties.RetryProperties; +import org.apache.servicecomb.registry.RegistrationManager; +import org.apache.servicecomb.swagger.invocation.AsyncResponse; +import org.apache.servicecomb.swagger.invocation.Response; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.github.resilience4j.decorators.Decorators; +import io.github.resilience4j.decorators.Decorators.DecorateCompletionStage; + +public class ConsumerGovernanceHandler implements Handler { + private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerGovernanceHandler.class); + + private MatchersManager matchersManager = BeanUtils.getBean(MatchersManager.class); + + private RetryHandler retryHandler = BeanUtils.getBean(RetryHandler.class); + + private RetryProperties retryProperties = BeanUtils.getBean(RetryProperties.class); + + private static final ScheduledExecutorService RETRY_POOL = Executors.newScheduledThreadPool(2, new ThreadFactory() { + private AtomicInteger count = new AtomicInteger(0); + + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, "governance-retry-pool-thread-" + count.getAndIncrement()); + // avoid block shutdown + thread.setDaemon(true); + return thread; + } + }); + + @Override + public void handle(Invocation invocation, AsyncResponse asyncResp) throws Exception { + Supplier<CompletionStage<Response>> next = createBusinessCompletionStageSupplier(invocation); + DecorateCompletionStage<Response> dcs = Decorators.ofCompletionStage(next); + GovHttpRequest request = createGovHttpRequest(invocation); + + try { + ServiceCombInvocationContext.setInvocationContext(invocation); + addRetry(dcs, request); + } finally { + ServiceCombInvocationContext.removeInvocationContext(); + } + + final SyncResponseExecutor originalExecutor; + final Executor newExecutor; + if (invocation.getResponseExecutor() instanceof SyncResponseExecutor) { + originalExecutor = (SyncResponseExecutor) invocation.getResponseExecutor(); + newExecutor = command -> { + // retry的场景，对于同步调用, 同步调用的主线程已经被挂起，无法再主线程中进行重试; + // 重试的场景，主线程等待响应线程唤醒。因此需要转换主线程，响应唤醒新的主线程，在重试逻辑成功后，再唤醒原来的主线程。 + // 重试也不能在网络线程（event-loop）中进行，未被保护的阻塞操作会导致网络线程挂起 + RETRY_POOL.submit(command); + }; + invocation.setResponseExecutor(newExecutor); + } else { + originalExecutor = null; + newExecutor = null; + } + + dcs.get().whenComplete((r, e) -> { + if (e == null) { + if (originalExecutor != null) { + originalExecutor.execute(() -> { + asyncResp.complete(r); + }); + } else { + asyncResp.complete(r); + } + return; + } + + if (originalExecutor != null) { + originalExecutor.execute(() -> { + asyncResp.consumerFail(e); + }); + } else { + asyncResp.consumerFail(e); + } + }); + } + + private void addRetry(DecorateCompletionStage<Response> dcs, GovHttpRequest request) { + RetryPolicy retryPolicy = matchersManager.match(request, retryProperties.getParsedEntity()); + if (retryPolicy != null) { + dcs.withRetry(retryHandler.getActuator(retryPolicy), RETRY_POOL); + } + } + + private Supplier<CompletionStage<Response>> createBusinessCompletionStageSupplier(Invocation invocation) { + final int currentHandler = invocation.getHandlerIndex(); + final AtomicBoolean isRetryHolder = new AtomicBoolean(false); + + return () -> { + CompletableFuture<Response> result = new CompletableFuture<>(); + if (isRetryHolder.getAndSet(true)) { + invocation.setHandlerIndex(currentHandler); + LOGGER.info("retry operation {}, trace id {}", + invocation.getOperationMeta().getMicroserviceQualifiedName(), invocation.getTraceId()); + } + try { + invocation.next(response -> { + result.complete(response); + }); + } catch (Exception e) { + result.completeExceptionally(e); + } + return result; + }; + } + + private GovHttpRequest createGovHttpRequest(Invocation invocation) { + GovHttpRequest request = new GovHttpRequest(RegistrationManager.INSTANCE.getMicroservice().getServiceName(), + RegistrationManager.INSTANCE.getMicroservice().getVersion()); + request.setUri(invocation.getSchemaId() + "." + invocation.getOperationName()); + request.setMethod(invocation.getOperationMeta().getHttpMethod()); + request.setHeaders(invocation.getContext()); + return request; + } +} diff --git a/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ProviderGovernanceHandler.java b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ProviderGovernanceHandler.java index 06d0c50..b3623c3 100644 --- a/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ProviderGovernanceHandler.java +++ b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ProviderGovernanceHandler.java @@ -73,13 +73,14 @@ public class ProviderGovernanceHandler implements Handler { DecorateCompletionStage<Response> dcs = Decorators.ofCompletionStage(next); GovHttpRequest request = createGovHttpRequest(invocation); - RateLimitingPolicy rateLimitingPolicy = matchersManager.match(request, rateLimitProperties.getParsedEntity()); - dcs.withRateLimiter(rateLimitingHandler.getActuator(rateLimitingPolicy)); - CircuitBreakerPolicy circuitBreakerPolicy = matchersManager - .match(request, circuitBreakerProperties.getParsedEntity()); - dcs.withCircuitBreaker(circuitBreakerHandler.getActuator(circuitBreakerPolicy)); - BulkheadPolicy bulkheadPolicy = matchersManager.match(request, bulkheadProperties.getParsedEntity()); - dcs.withBulkhead(bulkheadHandler.getActuator(bulkheadPolicy)); + try { + ServiceCombInvocationContext.setInvocationContext(invocation); + addRateLimiting(dcs, request); + addCircuitBreaker(dcs, request); + addBulkhead(dcs, request); + } finally { + ServiceCombInvocationContext.removeInvocationContext(); + } dcs.get().whenComplete((r, e) -> { if (e == null) { @@ -107,12 +108,41 @@ public class ProviderGovernanceHandler implements Handler { }); } + private void addBulkhead(DecorateCompletionStage<Response> dcs, GovHttpRequest request) { + BulkheadPolicy bulkheadPolicy = matchersManager.match(request, bulkheadProperties.getParsedEntity()); + if (bulkheadPolicy != null) { + dcs.withBulkhead(bulkheadHandler.getActuator(bulkheadPolicy)); + } + } + + private void addCircuitBreaker(DecorateCompletionStage<Response> dcs, GovHttpRequest request) { + CircuitBreakerPolicy circuitBreakerPolicy = matchersManager + .match(request, circuitBreakerProperties.getParsedEntity()); + if (circuitBreakerPolicy != null) { + dcs.withCircuitBreaker(circuitBreakerHandler.getActuator(circuitBreakerPolicy)); + } + } + + private void addRateLimiting(DecorateCompletionStage<Response> dcs, GovHttpRequest request) { + RateLimitingPolicy rateLimitingPolicy = matchersManager.match(request, rateLimitProperties.getParsedEntity()); + if (rateLimitingPolicy != null) { + dcs.withRateLimiter(rateLimitingHandler.getActuator(rateLimitingPolicy)); + } + } + private Supplier<CompletionStage<Response>> createBusinessCompletionStageSupplier(Invocation invocation) { return () -> { CompletableFuture<Response> result = new CompletableFuture<>(); try { invocation.next(response -> { - result.complete(response); + if (response.isFailed()) { + // For failed response, create a fail to make circuit breaker work. + // Users application maybe much complicated than this simple logic, + // while they need to customize which error will cause circuit breaker. + result.completeExceptionally(response.getResult()); + } else { + result.complete(response); + } }); } catch (Exception e) { result.completeExceptionally(e); diff --git a/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombInvocationContext.java b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombInvocationContext.java new file mode 100644 index 0000000..fdedd0a --- /dev/null +++ b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombInvocationContext.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.handler.governance; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.apache.servicecomb.governance.InvocationContext; +import org.springframework.stereotype.Component; + +@Component +public class ServiceCombInvocationContext implements InvocationContext { + private static final String CONTEXT_KEY = "x-servicecomb-governance-match"; + + private static ThreadLocal<org.apache.servicecomb.swagger.invocation.context.InvocationContext> contextMgr = new ThreadLocal<>(); + + public static void setInvocationContext( + org.apache.servicecomb.swagger.invocation.context.InvocationContext invocationContext) { + contextMgr.set(invocationContext); + } + + public static void removeInvocationContext() { + contextMgr.remove(); + } + + @Override + public Map<String, Boolean> getCalculatedMatches() { + Map<String, Boolean> result = contextMgr.get().getLocalContext(CONTEXT_KEY); + if (result == null) { + return Collections.emptyMap(); + } + return result; + } + + @Override + public void addMatch(String key, Boolean value) { + Map<String, Boolean> result = contextMgr.get().getLocalContext(CONTEXT_KEY); + if (result == null) { + result = new HashMap<>(); + contextMgr.get().addLocalContext(CONTEXT_KEY, result); + } + result.put(key, value); + } +} diff --git a/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombRetryExtension.java b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombRetryExtension.java new file mode 100644 index 0000000..031fc9a --- /dev/null +++ b/handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ServiceCombRetryExtension.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.servicecomb.handler.governance; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.NoRouteToHostException; +import java.net.SocketTimeoutException; +import java.util.List; + +import javax.ws.rs.core.Response.Status; + +import org.apache.servicecomb.governance.handler.ext.RetryExtension; +import org.apache.servicecomb.swagger.invocation.Response; +import org.apache.servicecomb.swagger.invocation.exception.ExceptionFactory; +import org.apache.servicecomb.swagger.invocation.exception.InvocationException; +import org.springframework.stereotype.Component; + +import io.vertx.core.VertxException; + +@Component +public class ServiceCombRetryExtension implements RetryExtension { + @Override + public boolean isRetry(List<Integer> statusList, Object result) { + if (result instanceof Response) { + Response resp = (Response) result; + if (resp.isFailed()) { + if (InvocationException.class.isInstance(resp.getResult())) { + InvocationException e = resp.getResult(); + return e.getStatusCode() == ExceptionFactory.CONSUMER_INNER_STATUS_CODE + || e.getStatusCode() == Status.SERVICE_UNAVAILABLE.getStatusCode() + || e.getStatusCode() == Status.BAD_GATEWAY.getStatusCode(); + } else { + return true; + } + } else { + return false; + } + } + return false; + } + + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + public Class<? extends Throwable>[] retryExceptions() { + return new Class[] { + ConnectException.class, + SocketTimeoutException.class, + IOException.class, + VertxException.class, + NoRouteToHostException.class, + InvocationException.class}; + } +} diff --git a/handlers/handler-governance/src/main/resources/config/cse.handler.xml b/handlers/handler-governance/src/main/resources/config/cse.handler.xml index f99f63d..c7383f0 100644 --- a/handlers/handler-governance/src/main/resources/config/cse.handler.xml +++ b/handlers/handler-governance/src/main/resources/config/cse.handler.xml @@ -18,4 +18,6 @@ <config> <handler id="governance-provider" class="org.apache.servicecomb.handler.governance.ProviderGovernanceHandler"/> + <handler id="governance-consumer" + class="org.apache.servicecomb.handler.governance.ConsumerGovernanceHandler"/> </config> diff --git a/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java b/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java index dc507c6..58dd4c0 100644 --- a/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java +++ b/handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/LoadbalanceHandler.java @@ -298,10 +298,10 @@ public class LoadbalanceHandler implements Handler { // retry in loadbalance, 2.0 feature int currentHandler = invocation.getHandlerIndex(); - SyncResponseExecutor orginExecutor; + SyncResponseExecutor originalExecutor; Executor newExecutor; if (invocation.getResponseExecutor() instanceof SyncResponseExecutor) { - orginExecutor = (SyncResponseExecutor) invocation.getResponseExecutor(); + originalExecutor = (SyncResponseExecutor) invocation.getResponseExecutor(); newExecutor = new Executor() { @Override public void execute(Runnable command) { @@ -312,7 +312,7 @@ public class LoadbalanceHandler implements Handler { }; invocation.setResponseExecutor(newExecutor); } else { - orginExecutor = null; + originalExecutor = null; newExecutor = null; } @@ -346,8 +346,8 @@ public class LoadbalanceHandler implements Handler { context.getRequest().getInvocationQualifiedName(), context.getRequest().getEndpoint()); } - if (orginExecutor != null) { - orginExecutor.execute(() -> { + if (originalExecutor != null) { + originalExecutor.execute(() -> { asyncResp.complete(response); }); } else { @@ -361,8 +361,8 @@ public class LoadbalanceHandler implements Handler { context.getRequest().getTraceIdLogger().error(LOGGER, "Invoke all server failed. Operation {}, e={}", context.getRequest().getInvocationQualifiedName(), ExceptionUtils.getExceptionMessageWithoutTrace(finalException)); - if (orginExecutor != null) { - orginExecutor.execute(() -> { + if (originalExecutor != null) { + originalExecutor.execute(() -> { fail(finalException); }); } else { diff --git a/solutions/solution-basic/pom.xml b/solutions/solution-basic/pom.xml index 29e5ce4..0806478 100644 --- a/solutions/solution-basic/pom.xml +++ b/solutions/solution-basic/pom.xml @@ -73,6 +73,10 @@ </dependency> <dependency> <groupId>org.apache.servicecomb</groupId> + <artifactId>handler-governance</artifactId> + </dependency> + <dependency> + <groupId>org.apache.servicecomb</groupId> <artifactId>handler-loadbalance</artifactId> </dependency> <dependency>
