[
https://issues.apache.org/jira/browse/SCB-352?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16387920#comment-16387920
]
ASF GitHub Bot commented on SCB-352:
------------------------------------
yhs0092 closed pull request #568: [SCB-352] Support provider operation level
flow control
URL: https://github.com/apache/incubator-servicecomb-java-chassis/pull/568
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/handlers/handler-flowcontrol-qps/pom.xml
b/handlers/handler-flowcontrol-qps/pom.xml
index e9c10bdd1..86a3943ec 100644
--- a/handlers/handler-flowcontrol-qps/pom.xml
+++ b/handlers/handler-flowcontrol-qps/pom.xml
@@ -16,29 +16,34 @@
-->
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <groupId>org.apache.servicecomb</groupId>
- <artifactId>handlers</artifactId>
- <version>1.0.0-m1</version>
- </parent>
- <artifactId>handler-flowcontrol-qps</artifactId>
- <name>Java Chassis::Handlers::Flow Control QPS</name>
- <dependencies>
- <dependency>
- <groupId>org.apache.servicecomb</groupId>
- <artifactId>java-chassis-core</artifactId>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>handlers</artifactId>
+ <version>1.0.0-m1</version>
+ </parent>
+ <artifactId>handler-flowcontrol-qps</artifactId>
+ <name>Java Chassis::Handlers::Flow Control QPS</name>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>java-chassis-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicecomb</groupId>
+ <artifactId>foundation-test-scaffolding</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
</project>
diff --git
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/AbstractQpsControllerManager.java
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/AbstractQpsControllerManager.java
new file mode 100644
index 000000000..b9a0b2c19
--- /dev/null
+++
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/AbstractQpsControllerManager.java
@@ -0,0 +1,176 @@
+/*
+ * 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.qps;
+
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.netflix.config.DynamicProperty;
+
+public class AbstractQpsControllerManager {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AbstractQpsControllerManager.class);
+
+ /**
+ * Describe the relationship between configuration and qpsController.
+ */
+ protected final Map<String, QpsController> configQpsControllerMap = new
ConcurrentHashMapEx<>();
+
+ /**
+ * Describe the relationship between qualifiedKey(format is
"microservice.schema.operation") and qpsController.
+ */
+ protected final Map<String, QpsController> qualifiedNameControllerMap = new
ConcurrentHashMapEx<>();
+
+ protected QpsController globalQpsController;
+
+ public static final String SEPARATOR = ".";
+
+ private String configKeyPrefix;
+
+ public QpsController getOrCreate(String key) {
+ return qualifiedNameControllerMap.computeIfAbsent(key, qualifiedNameKey ->
create(qualifiedNameKey));
+ }
+
+ protected QpsController create(String qualifiedNameKey) {
+ // create "microservice"
+ createQpsControllerIfNotExist(qualifiedNameKey.substring(0,
qualifiedNameKey.indexOf(SEPARATOR)));
+ // create "microservice.schema"
+ createQpsControllerIfNotExist(qualifiedNameKey.substring(0,
qualifiedNameKey.lastIndexOf(SEPARATOR)));
+ // create "microservice.schema.operation"
+ createQpsControllerIfNotExist(qualifiedNameKey);
+
+ return searchQpsController(qualifiedNameKey);
+ }
+
+ /**
+ * Try to find a valid qps controller by qualifiedNameKey.
+ *
+ * @param qualifiedNameKey qualifiedNameKey in {@link
#qualifiedNameControllerMap}
+ * @return a valid qps controller, lower level controllers with valid
qpsLimit have priority.
+ */
+ protected QpsController searchQpsController(String qualifiedNameKey) {
+ QpsController qpsController = searchByKey(qualifiedNameKey);
+
+ if (isValidQpsController(qpsController)) {
+ return qpsController;
+ }
+
+ if (null != globalQpsController) {
+ return globalQpsController;
+ }
+
+ // if null is returned, maybe the operation qps controller is not
initiated correctly.
+ // getOrCreateQpsController() should be invoked before.
+ return qpsController;
+ }
+
+ /**
+ * Use qualifiedNameKey to search {@link QpsController}.
+ * Firstly try to search "microservice.schema.operation". If no valid result
found, then try "microservice.schema",
+ * and then "microservice".
+ *
+ * @param qualifiedNameKey qualifiedNameKey in {@link
#qualifiedNameControllerMap}
+ * @return a qps controller, lower level controllers with valid qpsLimit
have priority.
+ * null will be returned only when {@link #configQpsControllerMap} contains
no matched qualifiedNameKey.
+ */
+ protected QpsController searchByKey(String qualifiedNameKey) {
+ QpsController qpsController = configQpsControllerMap.get(qualifiedNameKey);
+ if (isValidQpsController(qpsController)) {
+ return qpsController;
+ }
+
+ int index = qualifiedNameKey.lastIndexOf(SEPARATOR);
+ while (index > 0) {
+ qpsController = configQpsControllerMap.get(qualifiedNameKey.substring(0,
index));
+ if (isValidQpsController(qpsController)) {
+ return qpsController;
+ }
+
+ index = qualifiedNameKey.lastIndexOf(SEPARATOR, index - 1);
+ }
+
+ return qpsController;
+ }
+
+ private boolean keyMatch(String configKey, Entry<String, QpsController>
controllerEntry) {
+ return controllerEntry.getKey().equals(configKey)
+ || controllerEntry.getKey().startsWith(configKey + SEPARATOR);
+ }
+
+ private boolean isValidQpsController(QpsController qpsController) {
+ return null != qpsController && null != qpsController.getQpsLimit();
+ }
+
+ private void createQpsControllerIfNotExist(String configKey) {
+ if (configQpsControllerMap.keySet().contains(configKey)) {
+ return;
+ }
+
+ LOGGER.info("Create qpsController, configKey = [{}]", configKey);
+ DynamicProperty property = getDynamicProperty(configKey);
+ QpsController qpsController = new QpsController(configKey,
property.getInteger());
+
+ configQpsControllerMap.put(configKey, qpsController);
+
+ property.addCallback(() -> {
+ qpsController.setQpsLimit(property.getInteger());
+ LOGGER.info("Qps limit updated, configKey = [{}], value = [{}]",
configKey, property.getString());
+ updateObjMap(configKey);
+ });
+ }
+
+ protected void updateObjMap(String configKey) {
+ for (Entry<String, QpsController> controllerEntry :
qualifiedNameControllerMap.entrySet()) {
+ if (keyMatch(configKey, controllerEntry)) {
+ QpsController qpsController =
searchQpsController(controllerEntry.getKey());
+ controllerEntry.setValue(qpsController);
+ LOGGER.info("QpsController updated, operationId = [{}], configKey =
[{}], qpsLimit = [{}]",
+ controllerEntry.getKey(), qpsController.getKey(),
qpsController.getQpsLimit());
+ }
+ }
+ }
+
+ public AbstractQpsControllerManager setConfigKeyPrefix(String
configKeyPrefix) {
+ this.configKeyPrefix = configKeyPrefix;
+ return this;
+ }
+
+ public AbstractQpsControllerManager setGlobalQpsController(String
globalConfigKey) {
+ DynamicProperty globalQpsProperty =
DynamicProperty.getInstance(globalConfigKey);
+ QpsController qpsController = new QpsController(globalConfigKey,
globalQpsProperty.getInteger());
+
+ globalQpsProperty.addCallback(() -> {
+ qpsController.setQpsLimit(globalQpsProperty.getInteger());
+ LOGGER.info("Global qps limit update, value = [{}]",
globalQpsProperty.getInteger());
+ });
+
+ this.globalQpsController = qpsController;
+ return this;
+ }
+
+ public QpsController getGlobalQpsController() {
+ return globalQpsController;
+ }
+
+ protected DynamicProperty getDynamicProperty(String configKey) {
+ return DynamicProperty.getInstance(configKeyPrefix + configKey);
+ }
+}
diff --git
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsControllerManager.java
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsControllerManager.java
index 076df492d..6b420fd4e 100644
---
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsControllerManager.java
+++
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsControllerManager.java
@@ -17,141 +17,41 @@
package org.apache.servicecomb.qps;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.servicecomb.core.definition.OperationMeta;
-import org.apache.servicecomb.foundation.common.AbstractObjectManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.netflix.config.DynamicProperty;
/**
- * 以microservice.schema.operation为key
- * 如果配置只到microservice级别，那么该microservice中所有的key都关联到同一个qpsController实例
- * 然后schema、operation级别有独立配置的，单独关联自己的qpsController实例
- *
- * schema级独立的qpsController统计时，并不会导致microservice级别的统计也改变，operation级别规则也相同
- * 即：统计只在qpsController实例内部生效，不会产生实例间的关联
+ * <p>
+ * Use microservice.schema.operation as key.
+ * If there is only configuration of microservice level, then all invocations
are mapped to the qpsController of this configuration.
+ * If there are configurations of schema and operation level, then the
corresponding invocations is mapped to the qpsController
+ * separately.
+ * </p>
*
+ * <p>
+ * When qpsController of schema level counts, the qpsController of
microservice level won't count.
+ * It's the same as operation level qpsController.
+ * i.e. the statistics are only taken in qpsController internally. And between
the qpsControllers there is no relationship.
+ * </p>
*/
-public class ConsumerQpsControllerManager extends
AbstractObjectManager<OperationMeta, String, QpsController> {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ConsumerQpsControllerManager.class);
- // 最终使用的QpsController实例，实际都是从下面的map中引用出来的，不会独立创建
-
- // 3种类型的key都保存在这里，不存在冲突
- // microservice
- // microservice.schema
- // microservice.schema.operation
- private Map<String, QpsController> qpsControllerMap = new
ConcurrentHashMap<>();
-
- // 避免重复watch
- // 只会在create流程中调用，是有锁保护的，不必考虑多线程并发
- private Set<String> watchedKeySet = new HashSet<>();
-
- @Override
- protected String getKey(OperationMeta operationMeta) {
- return operationMeta.getMicroserviceQualifiedName();
- }
-
- private QpsController initQpsLimit(String key, Integer qpsLimit) {
- if (qpsLimit == null) {
- return null;
- }
-
- LOGGER.info("qpsLimit of {} init as {}", key, qpsLimit);
-
- QpsController qpsController = new QpsController(key, qpsLimit);
- qpsControllerMap.put(key, qpsController);
- return qpsController;
- }
-
- private QpsController updateQpsLimit(String key, Integer qpsLimit) {
- QpsController qpsController = qpsControllerMap.get(key);
- if (qpsController == null && qpsLimit != null) {
- qpsController = new QpsController(key, qpsLimit);
- qpsControllerMap.put(key, qpsController);
- }
-
- if (qpsController != null) {
- LOGGER.info("qpsLimit of {} changed from {} to {}", key,
qpsController.getQpsLimit(), qpsLimit);
-
- qpsController.setQpsLimit(qpsLimit);
- }
-
- return qpsController;
- }
+public class ConsumerQpsControllerManager extends AbstractQpsControllerManager
{
+ private static volatile ConsumerQpsControllerManager INSTANCE;
- private QpsController findReference(String key) {
- QpsController qpsController = qpsControllerMap.get(key);
- if (qpsController == null) {
- return null;
- }
-
- if (qpsController.getQpsLimit() == null) {
- return null;
- }
-
- return qpsController;
+ private ConsumerQpsControllerManager() {
}
- private QpsController findReference(OperationMeta operationMeta) {
- QpsController qpsController =
findReference(operationMeta.getMicroserviceQualifiedName());
- if (qpsController != null) {
- return qpsController;
- }
-
- qpsController =
findReference(operationMeta.getSchemaMeta().getMicroserviceQualifiedName());
- if (qpsController != null) {
- return qpsController;
- }
-
- qpsController = findReference(operationMeta.getMicroserviceName());
- if (qpsController != null) {
- return qpsController;
+ public static ConsumerQpsControllerManager getINSTANCE() {
+ if (null == INSTANCE) {
+ synchronized (ConsumerQpsControllerManager.class) {
+ if (null == INSTANCE) {
+ INSTANCE = new ConsumerQpsControllerManager();
+ }
+ }
}
-
- return initQpsLimit(operationMeta.getMicroserviceName(),
Integer.MAX_VALUE);
+ return INSTANCE;
}
@Override
- protected QpsController create(OperationMeta operationMeta) {
- // create在父类中是加了锁的，不存在并发的场景
- initConfig(operationMeta, operationMeta.getMicroserviceQualifiedName());
- initConfig(operationMeta,
operationMeta.getSchemaMeta().getMicroserviceQualifiedName());
- initConfig(operationMeta, operationMeta.getMicroserviceName());
-
- return findReference(operationMeta);
- }
-
- private void initConfig(OperationMeta operationMeta, String key) {
- if (watchedKeySet.contains(key)) {
- return;
- }
-
- watchedKeySet.add(key);
-
- String configKey = Config.CONSUMER_LIMIT_KEY_PREFIX + key;
- DynamicProperty property = DynamicProperty.getInstance(configKey);
- initQpsLimit(key, getIntegerLimitProperty(property));
-
- property.addCallback(() -> {
- updateQpsLimit(key, getIntegerLimitProperty(property));
- QpsController qpsController = findReference(operationMeta);
-
- objMap.put(operationMeta.getMicroserviceQualifiedName(), qpsController);
- });
- }
-
- private Integer getIntegerLimitProperty(DynamicProperty property) {
- try {
- return property.getInteger();
- } catch (IllegalArgumentException e) {
- LOGGER.error(e.getMessage());
- return null;
- }
+ protected DynamicProperty getDynamicProperty(String configKey) {
+ return DynamicProperty.getInstance(Config.CONSUMER_LIMIT_KEY_PREFIX +
configKey);
}
}
diff --git
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsFlowControlHandler.java
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsFlowControlHandler.java
index b1fd2a669..785dfb123 100644
---
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsFlowControlHandler.java
+++
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ConsumerQpsFlowControlHandler.java
@@ -19,18 +19,16 @@
import org.apache.servicecomb.core.Handler;
import org.apache.servicecomb.core.Invocation;
-import org.apache.servicecomb.core.definition.OperationMeta;
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
/**
- * consumer端针对调用目标的qps控制
- * 支持microservice、schema、operation三个级别的控制
- *
+ * For qps flow control on consumer side.
+ * Support 3 levels of microservice/schema/operation.
*/
public class ConsumerQpsFlowControlHandler implements Handler {
- private ConsumerQpsControllerManager qpsControllerMgr = new
ConsumerQpsControllerManager();
+ private ConsumerQpsControllerManager qpsControllerMgr =
ConsumerQpsControllerManager.getINSTANCE();
@Override
public void handle(Invocation invocation, AsyncResponse asyncResp) throws
Exception {
@@ -39,10 +37,10 @@ public void handle(Invocation invocation, AsyncResponse
asyncResp) throws Except
return;
}
- OperationMeta operationMeta = invocation.getOperationMeta();
- QpsController qpsController = qpsControllerMgr.getOrCreate(operationMeta);
+ QpsController qpsController = qpsControllerMgr.getOrCreate(
+ invocation.getOperationMeta().getMicroserviceQualifiedName());
if (qpsController.isLimitNewRequest()) {
- // 429
+ // return http status 429
CommonExceptionData errorData = new CommonExceptionData("rejected by qps
flowcontrol");
asyncResp.consumerFail(
new InvocationException(QpsConst.TOO_MANY_REQUESTS_STATUS,
errorData));
diff --git
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsControllerManager.java
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsControllerManager.java
index d8d6c76ac..36093e4ce 100644
---
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsControllerManager.java
+++
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsControllerManager.java
@@ -17,127 +17,28 @@
package org.apache.servicecomb.qps;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-import org.apache.servicecomb.foundation.common.AbstractObjectManager;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import com.netflix.config.DynamicProperty;
-public class ProviderQpsControllerManager
- extends AbstractObjectManager<String, String, QpsController> {
- private static final Logger LOGGER =
LoggerFactory.getLogger(ProviderQpsControllerManager.class);
-
- private Map<String, QpsController> qpsControllerMap = new
ConcurrentHashMap<>();
-
- // 避免重复watch
- // 只会在create流程中调用，是有锁保护的，不必考虑多线程并发
- private Set<String> watchedKeySet = new HashSet<>();
-
- private QpsController globalQpsController = null;
+public class ProviderQpsControllerManager extends AbstractQpsControllerManager
{
+ private static volatile ProviderQpsControllerManager INSTANCE;
- @Override
- protected String getKey(String microServiceName) {
- return microServiceName;
+ private ProviderQpsControllerManager() {
+ setGlobalQpsController(Config.PROVIDER_LIMIT_KEY_GLOBAL);
}
- @Override
- public QpsController getOrCreate(String keyOwner) {
- if (keyOwner == null) {
- if (globalQpsController == null) {
- synchronized (lockObj) {
- if (globalQpsController == null) {
- DynamicProperty property =
- DynamicProperty.getInstance(Config.PROVIDER_LIMIT_KEY_GLOBAL);
- globalQpsController = new QpsController(keyOwner,
getIntegerLimitProperty(property));
- property.addCallback(() -> {
-
globalQpsController.setQpsLimit(getIntegerLimitProperty(property));
- });
- }
+ public static ProviderQpsControllerManager getINSTANCE() {
+ if (null == INSTANCE) {
+ synchronized (ProviderQpsControllerManager.class) {
+ if (null == INSTANCE) {
+ INSTANCE = new ProviderQpsControllerManager();
}
}
- return globalQpsController;
- } else {
- return super.getOrCreate(keyOwner);
}
- }
-
- private QpsController initQpsLimit(String key, Integer qpsLimit) {
- if (qpsLimit == null) {
- qpsLimit = null;
- }
-
- LOGGER.info("qpsLimit of {} init as {}", key, qpsLimit);
-
- QpsController qpsController = new QpsController(key, qpsLimit);
- qpsControllerMap.put(key, qpsController);
- return qpsController;
- }
-
- private QpsController updateQpsLimit(String key, Integer qpsLimit) {
- QpsController qpsController = qpsControllerMap.get(key);
- if (qpsController == null && qpsLimit != null) {
- qpsController = new QpsController(key, qpsLimit);
- qpsControllerMap.put(key, qpsController);
- }
-
- if (qpsController != null) {
- LOGGER.info("qpsLimit of {} changed from {} to {}",
- key,
- qpsController.getQpsLimit(),
- qpsLimit);
-
- qpsController.setQpsLimit(qpsLimit);
- }
-
- return qpsController;
- }
-
- private QpsController findReference(String key) {
- QpsController qpsController = qpsControllerMap.get(key);
- if (qpsController == null) {
- return initQpsLimit(key, Integer.MAX_VALUE);
- }
- return qpsController;
+ return INSTANCE;
}
@Override
- protected QpsController create(String microServiceName) {
- // create在父类中是加了锁的，不存在并发的场景
- initConfig(microServiceName);
-
- return findReference(microServiceName);
- }
-
- private void initConfig(String key) {
- if (watchedKeySet.contains(key)) {
- return;
- }
-
- watchedKeySet.add(key);
-
- String configKey = Config.PROVIDER_LIMIT_KEY_PREFIX + key;
- DynamicProperty property = DynamicProperty.getInstance(configKey);
- initQpsLimit(key, getIntegerLimitProperty(property));
-
- property.addCallback(() -> {
- updateQpsLimit(key, getIntegerLimitProperty(property));
- QpsController qpsController = findReference(key);
-
- objMap.put(key, qpsController);
- });
- }
-
- private Integer getIntegerLimitProperty(DynamicProperty property) {
- try {
- return property.getInteger();
- } catch (IllegalArgumentException e) {
- LOGGER.error(e.getMessage());
- return null;
- }
+ protected DynamicProperty getDynamicProperty(String configKey) {
+ return DynamicProperty.getInstance(Config.PROVIDER_LIMIT_KEY_PREFIX +
configKey);
}
}
diff --git
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsFlowControlHandler.java
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsFlowControlHandler.java
index 1cf669d81..ae8e03c99 100644
---
a/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsFlowControlHandler.java
+++
b/handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/ProviderQpsFlowControlHandler.java
@@ -23,9 +23,10 @@
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
import org.apache.servicecomb.swagger.invocation.exception.InvocationException;
+import org.springframework.util.StringUtils;
public class ProviderQpsFlowControlHandler implements Handler {
- private ProviderQpsControllerManager qpsControllerMgr = new
ProviderQpsControllerManager();
+ private ProviderQpsControllerManager qpsControllerMgr =
ProviderQpsControllerManager.getINSTANCE();
@Override
public void handle(Invocation invocation, AsyncResponse asyncResp) throws
Exception {
@@ -34,16 +35,12 @@ public void handle(Invocation invocation, AsyncResponse
asyncResp) throws Except
return;
}
- String microServiceName = (String)
invocation.getContext(Const.SRC_MICROSERVICE);
- if (microServiceName != null && !microServiceName.isEmpty()) {
- QpsController qpsController =
qpsControllerMgr.getOrCreate(microServiceName);
- if (isLimitNewRequest(qpsController, asyncResp)) {
- return;
- }
- }
-
- QpsController globalQpsController = qpsControllerMgr.getOrCreate(null);
- if (isLimitNewRequest(globalQpsController, asyncResp)) {
+ String microServiceName = invocation.getContext(Const.SRC_MICROSERVICE);
+ QpsController qpsController =
+ StringUtils.isEmpty(microServiceName)
+ ? qpsControllerMgr.getGlobalQpsController()
+ : qpsControllerMgr.getOrCreate(microServiceName +
invocation.getOperationMeta().getSchemaQualifiedName());
+ if (isLimitNewRequest(qpsController, asyncResp)) {
return;
}
diff --git
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/AbstractQpsControllerManagerTest.java
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/AbstractQpsControllerManagerTest.java
new file mode 100644
index 000000000..9004ecdb4
--- /dev/null
+++
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/AbstractQpsControllerManagerTest.java
@@ -0,0 +1,212 @@
+/*
+ * 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.qps;
+
+import java.util.Map;
+
+import org.apache.servicecomb.core.Invocation;
+import org.apache.servicecomb.core.definition.OperationMeta;
+import org.apache.servicecomb.core.definition.SchemaMeta;
+import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import mockit.Deencapsulation;
+
+public class AbstractQpsControllerManagerTest {
+
+ @Before
+ public void beforeTest() {
+ ArchaiusUtils.resetConfig();
+ }
+
+ @After
+ public void afterTest() {
+ ArchaiusUtils.resetConfig();
+ }
+
+ @Test
+ public void testGetOrCreate() {
+ AbstractQpsControllerManager testQpsControllerManager = new
AbstractQpsControllerManager();
+ initTestQpsControllerManager(testQpsControllerManager);
+
+ // pojo
+ setConfigWithDefaultPrefix("pojo", 100);
+ QpsController qpsController =
testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals("pojo", qpsController.getKey());
+ Assert.assertTrue(100 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo2.server.test");
+ Assert.assertEquals("pojo2", qpsController.getKey());
+ Assert.assertNull(qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("poj.server.test");
+ Assert.assertEquals("poj", qpsController.getKey());
+ Assert.assertNull(qpsController.getQpsLimit());
+
+ testGetOrCreateCommon(testQpsControllerManager);
+ }
+
+ @Test
+ public void testGetOrCreateWithGlobalConfig() {
+ AbstractQpsControllerManager testQpsControllerManager = new
AbstractQpsControllerManager()
+ .setGlobalQpsController(Config.PROVIDER_LIMIT_KEY_GLOBAL);
+
+ // global
+ setConfig(Config.PROVIDER_LIMIT_KEY_GLOBAL, 50);
+ QpsController qpsController =
testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
+ Assert.assertTrue(50 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo2.server.test");
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
+ Assert.assertTrue(50 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("poj.server.test");
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
+ Assert.assertTrue(50 == qpsController.getQpsLimit());
+
+ // pojo
+ setConfigWithDefaultPrefix("pojo", 100);
+ qpsController = testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals("pojo", qpsController.getKey());
+ Assert.assertTrue(100 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo2.server.test");
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
+ Assert.assertTrue(50 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("poj.server.test");
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
+ Assert.assertTrue(50 == qpsController.getQpsLimit());
+
+ testGetOrCreateCommon(testQpsControllerManager);
+ }
+
+ private void testGetOrCreateCommon(AbstractQpsControllerManager
testQpsControllerManager) {
+ // pojo.server
+ setConfigWithDefaultPrefix("pojo.server", 200);
+ QpsController qpsController =
testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals("pojo.server", qpsController.getKey());
+ Assert.assertTrue(200 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo.server2.test");
+ Assert.assertEquals("pojo", qpsController.getKey());
+ Assert.assertTrue(100 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo.serve.test");
+ Assert.assertEquals("pojo", qpsController.getKey());
+ Assert.assertTrue(100 == qpsController.getQpsLimit());
+
+ // pojo.server.test
+ setConfigWithDefaultPrefix("pojo.server.test", 300);
+ qpsController = testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals("pojo.server.test", qpsController.getKey());
+ Assert.assertTrue(300 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo.server.test2");
+ Assert.assertEquals("pojo.server", qpsController.getKey());
+ Assert.assertTrue(200 == qpsController.getQpsLimit());
+ qpsController = testQpsControllerManager.getOrCreate("pojo.server.tes");
+ Assert.assertEquals("pojo.server", qpsController.getKey());
+ Assert.assertTrue(200 == qpsController.getQpsLimit());
+ }
+
+ /**
+ * Init testQpsControllerManager to test search function.
+ */
+ private void initTestQpsControllerManager(AbstractQpsControllerManager
testQpsControllerManager) {
+ // pojo.server.test
+ QpsController qpsController =
testQpsControllerManager.getOrCreate("pojo.server.test");
+ Assert.assertEquals("pojo", qpsController.getKey());
+ Assert.assertNull(qpsController.getQpsLimit());
+
+ // pojo.server.test2
+ testQpsControllerManager.getOrCreate("pojo.server.test2");
+
+ // pojo.server.tes
+ testQpsControllerManager.getOrCreate("pojo.server.tes");
+
+ // pojo.server2.test
+ testQpsControllerManager.getOrCreate("pojo.server2.test");
+
+ // pojo.serve.test
+ testQpsControllerManager.getOrCreate("pojo.serve.test");
+
+ // pojo2.server.test
+ qpsController = testQpsControllerManager.getOrCreate("pojo2.server.test");
+ Assert.assertEquals("pojo2", qpsController.getKey());
+ Assert.assertNull(qpsController.getQpsLimit());
+
+ // poj.server.test
+ qpsController = testQpsControllerManager.getOrCreate("poj.server.test");
+ Assert.assertEquals("poj", qpsController.getKey());
+ Assert.assertNull(qpsController.getQpsLimit());
+ }
+
+ @Test
+ public void testMock() {
+ Invocation invocation = getMockInvocation("service", "schema", "oper");
+ OperationMeta operationMeta = invocation.getOperationMeta();
+ SchemaMeta schemaMeta = operationMeta.getSchemaMeta();
+
+ Assert.assertEquals("service", operationMeta.getMicroserviceName());
+ Assert.assertEquals("service.schema.oper",
operationMeta.getMicroserviceQualifiedName());
+ Assert.assertEquals("schema.oper", operationMeta.getSchemaQualifiedName());
+ Assert.assertEquals("schema", schemaMeta.getSchemaId());
+ }
+
+ public static Invocation getMockInvocation(String microserviceName, String
schemaId, String operationId) {
+ return getMockInvocation(
+ getMockOperationMeta(microserviceName, schemaId, operationId)
+ );
+ }
+
+ private static Invocation getMockInvocation(OperationMeta mockOperationMeta)
{
+ Invocation invocation = Mockito.mock(Invocation.class);
+ Mockito.when(invocation.getOperationMeta()).thenReturn(mockOperationMeta);
+ return invocation;
+ }
+
+ public static OperationMeta getMockOperationMeta(String microserviceName,
String schemaId, String operationId) {
+ OperationMeta operationMeta = Mockito.mock(OperationMeta.class);
+ SchemaMeta schemaMeta = Mockito.mock(SchemaMeta.class);
+
+ Mockito.when(operationMeta.getSchemaMeta()).thenReturn(schemaMeta);
+ Mockito.when(operationMeta.getSchemaQualifiedName())
+ .thenReturn(schemaId + AbstractQpsControllerManager.SEPARATOR +
operationId);
+ Mockito.when(operationMeta.getMicroserviceQualifiedName()).thenReturn(
+ microserviceName + AbstractQpsControllerManager.SEPARATOR + schemaId +
AbstractQpsControllerManager.SEPARATOR
+ + operationId);
+
Mockito.when(operationMeta.getMicroserviceName()).thenReturn(microserviceName);
+ Mockito.when(schemaMeta.getSchemaId()).thenReturn(schemaId);
+
+ return operationMeta;
+ }
+
+ public static void setConfig(String key, int value) {
+ Utils.updateProperty(key, value);
+ }
+
+ public static void setConfigWithDefaultPrefix(String key, int value) {
+ String configKey = Config.CONSUMER_LIMIT_KEY_PREFIX + key;
+ Utils.updateProperty(configKey, value);
+ }
+
+ public static void clearState(AbstractQpsControllerManager
qpsControllerManager) {
+ Map<String, QpsController> objMap =
Deencapsulation.getField(qpsControllerManager, "qualifiedNameControllerMap");
+ objMap.clear();
+ Map<String, QpsController> configQpsControllerMap = Deencapsulation
+ .getField(qpsControllerManager, "configQpsControllerMap");
+ configQpsControllerMap.clear();
+ }
+}
diff --git
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsControllermanager.java
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsControllermanager.java
index 426f7f8db..35aa4706d 100644
---
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsControllermanager.java
+++
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsControllermanager.java
@@ -17,59 +17,89 @@
package org.apache.servicecomb.qps;
-import org.apache.servicecomb.core.definition.OperationMeta;
-import org.apache.servicecomb.core.definition.SchemaMeta;
+import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
+import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
-import mockit.Expectations;
-import mockit.Mocked;
-
-/**
- *
- *
- */
public class TestConsumerQpsControllermanager {
- private static String microserviceName = "pojo";
+ private static final String MICROSERVICE_NAME = "pojo";
+
+ public static final String SCHEMA_ID = "server";
- private static String schemaQualified = microserviceName + ".server";
+ private static final String SCHEMA_QUALIFIED = MICROSERVICE_NAME +
AbstractQpsControllerManager.SEPARATOR + SCHEMA_ID;
- private static String operationQualified = schemaQualified + ".test";
+ public static final String OPERATION_ID = "test";
+
+ private static final String OPERATION_QUALIFIED =
+ SCHEMA_QUALIFIED + AbstractQpsControllerManager.SEPARATOR + OPERATION_ID;
+
+ @Before
+ public void beforeTest() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ConsumerQpsControllerManager.getINSTANCE());
+ }
+
+ @After
+ public void afterTest() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ConsumerQpsControllerManager.getINSTANCE());
+ }
@Test
- public void testQpsLimit(@Mocked SchemaMeta schemaMeta, @Mocked
OperationMeta operationMeta) {
- new Expectations() {
- {
- operationMeta.getMicroserviceQualifiedName();
- result = operationQualified;
-
- schemaMeta.getMicroserviceQualifiedName();
- result = schemaQualified;
-
- operationMeta.getMicroserviceName();
- result = microserviceName;
- }
- };
-
- ConsumerQpsControllerManager mgr = new ConsumerQpsControllerManager();
- QpsController qpsController = mgr.getOrCreate(operationMeta);
- Assert.assertEquals((Integer) Integer.MAX_VALUE,
qpsController.getQpsLimit());
- Assert.assertEquals(microserviceName, qpsController.getKey());
-
- doTestQpsLimit(mgr, operationMeta, microserviceName, 100,
microserviceName, 100);
- doTestQpsLimit(mgr, operationMeta, schemaQualified, 200, schemaQualified,
200);
- doTestQpsLimit(mgr, operationMeta, operationQualified, 300,
operationQualified, 300);
- doTestQpsLimit(mgr, operationMeta, operationQualified, null,
schemaQualified, 200);
- doTestQpsLimit(mgr, operationMeta, schemaQualified, null,
microserviceName, 100);
- doTestQpsLimit(mgr, operationMeta, microserviceName, null,
microserviceName, Integer.MAX_VALUE);
+ public void testQpsLimit() {
+ String qualifiedKey = MICROSERVICE_NAME +
AbstractQpsControllerManager.SEPARATOR + SCHEMA_ID
+ + AbstractQpsControllerManager.SEPARATOR + OPERATION_ID;
+
+ ConsumerQpsControllerManager mgr =
ConsumerQpsControllerManager.getINSTANCE();
+ QpsController qpsController = mgr.getOrCreate(qualifiedKey);
+ Assert.assertNull(qpsController.getQpsLimit());
+ Assert.assertEquals(MICROSERVICE_NAME, qpsController.getKey());
+
+ doTestQpsLimit(mgr, qualifiedKey, MICROSERVICE_NAME, 100,
MICROSERVICE_NAME, 100);
+ doTestQpsLimit(mgr, qualifiedKey, SCHEMA_QUALIFIED, 200, SCHEMA_QUALIFIED,
200);
+ doTestQpsLimit(mgr, qualifiedKey, OPERATION_QUALIFIED, 300,
OPERATION_QUALIFIED, 300);
+ doTestQpsLimit(mgr, qualifiedKey, OPERATION_QUALIFIED, null,
SCHEMA_QUALIFIED, 200);
+ doTestQpsLimit(mgr, qualifiedKey, SCHEMA_QUALIFIED, null,
MICROSERVICE_NAME, 100);
+ doTestQpsLimit(mgr, qualifiedKey, MICROSERVICE_NAME, null,
MICROSERVICE_NAME, null);
}
- private void doTestQpsLimit(ConsumerQpsControllerManager mgr, OperationMeta
operationMeta, String key,
+ private void doTestQpsLimit(ConsumerQpsControllerManager mgr, String
qualifiedKey, String key,
Integer newValue,
String expectKey, Integer expectValue) {
Utils.updateProperty(Config.CONSUMER_LIMIT_KEY_PREFIX + key, newValue);
- QpsController qpsController = mgr.getOrCreate(operationMeta);
- Assert.assertEquals(expectValue, qpsController.getQpsLimit());
+ QpsController qpsController = mgr.getOrCreate(qualifiedKey);
Assert.assertEquals(expectKey, qpsController.getKey());
+ Assert.assertEquals(expectValue, qpsController.getQpsLimit());
+ }
+
+ @Test
+ public void testQpsLimitOn2Operation() {
+ // qualifiedKey0 is pojo.server.test
+ // qualifiedKey1 is pojo.server.test1
+ String qualifiedKey0 = MICROSERVICE_NAME +
AbstractQpsControllerManager.SEPARATOR + SCHEMA_ID
+ + AbstractQpsControllerManager.SEPARATOR + OPERATION_ID;
+ String qualifiedKey1 = MICROSERVICE_NAME +
AbstractQpsControllerManager.SEPARATOR + SCHEMA_ID
+ + AbstractQpsControllerManager.SEPARATOR + OPERATION_ID + "1";
+
+ ConsumerQpsControllerManager mgr =
ConsumerQpsControllerManager.getINSTANCE();
+ QpsController qpsController = mgr.getOrCreate(qualifiedKey0);
+ Assert.assertNull(qpsController.getQpsLimit());
+ Assert.assertEquals(MICROSERVICE_NAME, qpsController.getKey());
+
+ qpsController = mgr.getOrCreate(qualifiedKey1);
+ Assert.assertNull(qpsController.getQpsLimit());
+ Assert.assertEquals(MICROSERVICE_NAME, qpsController.getKey());
+
+ // As operationMeta0 and operationMeta1 belong to the same schema, once
the qps configuration of the schema level
+ // is changed, both of their qpsControllers should be changed.
+ Utils.updateProperty(Config.CONSUMER_LIMIT_KEY_PREFIX + SCHEMA_QUALIFIED,
200);
+ qpsController = mgr.getOrCreate(qualifiedKey0);
+ Assert.assertEquals(Integer.valueOf(200), qpsController.getQpsLimit());
+ Assert.assertEquals(SCHEMA_QUALIFIED, qpsController.getKey());
+ qpsController = mgr.getOrCreate(qualifiedKey1);
+ Assert.assertEquals(Integer.valueOf(200), qpsController.getQpsLimit());
+ Assert.assertEquals(SCHEMA_QUALIFIED, qpsController.getKey());
}
}
diff --git
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsFlowControlHandler.java
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsFlowControlHandler.java
index 368459e19..854854412 100644
---
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsFlowControlHandler.java
+++
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestConsumerQpsFlowControlHandler.java
@@ -17,6 +17,8 @@
package org.apache.servicecomb.qps;
+import java.util.concurrent.ConcurrentHashMap;
+
import org.apache.servicecomb.core.Invocation;
import org.apache.servicecomb.core.definition.OperationMeta;
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
@@ -24,13 +26,10 @@
import org.junit.Test;
import org.mockito.Mockito;
+import mockit.Deencapsulation;
import mockit.Mock;
import mockit.MockUp;
-/**
- *
- *
- */
public class TestConsumerQpsFlowControlHandler {
ConsumerQpsFlowControlHandler handler = new ConsumerQpsFlowControlHandler();
@@ -42,7 +41,7 @@
OperationMeta operationMeta = Mockito.mock(OperationMeta.class);
@Test
- public void testQpsController() throws Exception {
+ public void testQpsController() {
QpsController qpsController = new QpsController("abc", 100);
Assert.assertEquals(false, qpsController.isLimitNewRequest());
@@ -68,9 +67,11 @@ public void testHandle() {
boolean validAssert;
try {
validAssert = true;
+ String key = "MicroserviceQualifiedName";
+ QpsController qpsController = new QpsController("key", 12);
Mockito.when(invocation.getOperationMeta()).thenReturn(operationMeta);
-
Mockito.when(operationMeta.getMicroserviceQualifiedName()).thenReturn("MicroserviceQualifiedName");
-
+
Mockito.when(operationMeta.getMicroserviceQualifiedName()).thenReturn(key);
+ setQpsController(key, qpsController);
new MockUp<QpsController>() {
@Mock
public boolean isLimitNewRequest() {
@@ -82,11 +83,12 @@ public boolean isLimitNewRequest() {
@Mock
protected QpsController create(OperationMeta operationMeta) {
- return new QpsController("key", 12);
+ return qpsController;
}
};
handler.handle(invocation, asyncResp);
} catch (Exception e) {
+ e.printStackTrace();
validAssert = false;
}
Assert.assertTrue(validAssert);
@@ -97,8 +99,11 @@ public void testHandleIsLimitNewRequestAsFalse() {
boolean validAssert;
try {
validAssert = true;
+ String key = "MicroserviceQualifiedName";
+ QpsController qpsController = new QpsController("key", 12);
Mockito.when(invocation.getOperationMeta()).thenReturn(operationMeta);
-
Mockito.when(operationMeta.getMicroserviceQualifiedName()).thenReturn("MicroserviceQualifiedName");
+
Mockito.when(operationMeta.getMicroserviceQualifiedName()).thenReturn(key);
+ setQpsController(key, qpsController);
new MockUp<QpsController>() {
@Mock
@@ -111,13 +116,21 @@ public boolean isLimitNewRequest() {
@Mock
protected QpsController create(OperationMeta operationMeta) {
- return new QpsController("key", 12);
+ return qpsController;
}
};
handler.handle(invocation, asyncResp);
} catch (Exception e) {
+ e.printStackTrace();
validAssert = false;
}
Assert.assertTrue(validAssert);
}
+
+ private void setQpsController(String key, QpsController qpsController) {
+ ConsumerQpsControllerManager qpsControllerManager =
Deencapsulation.getField(handler, "qpsControllerMgr");
+ ConcurrentHashMap<String, QpsController> objMap = Deencapsulation
+ .getField(qpsControllerManager, "qualifiedNameControllerMap");
+ objMap.put(key, qpsController);
+ }
}
diff --git
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsControllermanager.java
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsControllermanager.java
index c16f7092d..137e6d86d 100644
---
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsControllermanager.java
+++
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsControllermanager.java
@@ -17,27 +17,53 @@
package org.apache.servicecomb.qps;
+import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
+import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
public class TestProviderQpsControllermanager {
- private static String microserviceName = "pojo";
+ @Before
+ public void beforeTest() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ProviderQpsControllerManager.getINSTANCE());
+ }
+
+ @After
+ public void afterTest() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ProviderQpsControllerManager.getINSTANCE());
+ }
@Test
public void testQpsLimit() {
- ProviderQpsControllerManager mgr = new ProviderQpsControllerManager();
- QpsController qpsController = mgr.getOrCreate(microserviceName);
+ String microserviceName = "pojo";
+
+ ProviderQpsControllerManager mgr =
ProviderQpsControllerManager.getINSTANCE();
+ String schemaId = "server";
+ String operationId = "test";
+ String qualifiedKey = "pojo.server.test";
+ QpsController qpsController = mgr.getOrCreate(qualifiedKey);
Assert.assertEquals(null, qpsController.getQpsLimit());
- Assert.assertEquals(microserviceName, qpsController.getKey());
+ Assert.assertEquals(Config.PROVIDER_LIMIT_KEY_GLOBAL,
qpsController.getKey());
- doTestQpsLimit(mgr, microserviceName, 100, microserviceName, 100);
- doTestQpsLimit(mgr, microserviceName, null, microserviceName, null);
+ doTestQpsLimit(mgr, microserviceName, qualifiedKey, 100, microserviceName,
100);
+ String schemaKey = microserviceName +
AbstractQpsControllerManager.SEPARATOR + schemaId;
+ doTestQpsLimit(mgr, schemaKey, qualifiedKey, 80, schemaKey, 80);
+ String operationKey =
+ microserviceName + AbstractQpsControllerManager.SEPARATOR + schemaId +
AbstractQpsControllerManager.SEPARATOR
+ + operationId;
+ doTestQpsLimit(mgr, operationKey, qualifiedKey, 50, operationKey, 50);
+ doTestQpsLimit(mgr, operationKey, qualifiedKey, null, schemaKey, 80);
+ doTestQpsLimit(mgr, schemaKey, qualifiedKey, null, microserviceName, 100);
+ doTestQpsLimit(mgr, microserviceName, qualifiedKey, null,
Config.PROVIDER_LIMIT_KEY_GLOBAL, null);
}
- private void doTestQpsLimit(ProviderQpsControllerManager mgr, String key,
Integer newValue,
- String expectKey, Integer expectValue) {
- Utils.updateProperty(Config.PROVIDER_LIMIT_KEY_PREFIX + key, newValue);
- QpsController qpsController = mgr.getOrCreate(key);
+ private void doTestQpsLimit(ProviderQpsControllerManager mgr, String
configKey, String qualifiedKey,
+ Integer newValue, String expectKey, Integer expectValue) {
+ Utils.updateProperty(Config.PROVIDER_LIMIT_KEY_PREFIX + configKey,
newValue);
+ QpsController qpsController = mgr.getOrCreate(qualifiedKey);
Assert.assertEquals(expectValue, qpsController.getQpsLimit());
Assert.assertEquals(expectKey, qpsController.getKey());
}
diff --git
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsFlowControlHandler.java
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsFlowControlHandler.java
index 40303db8e..41d05fe78 100644
---
a/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsFlowControlHandler.java
+++
b/handlers/handler-flowcontrol-qps/src/test/java/org/apache/servicecomb/qps/TestProviderQpsFlowControlHandler.java
@@ -17,10 +17,14 @@
package org.apache.servicecomb.qps;
+import static org.junit.Assert.fail;
+
import org.apache.servicecomb.core.Const;
import org.apache.servicecomb.core.Invocation;
import org.apache.servicecomb.core.definition.OperationMeta;
+import org.apache.servicecomb.foundation.test.scaffolding.config.ArchaiusUtils;
import org.apache.servicecomb.swagger.invocation.AsyncResponse;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -42,9 +46,18 @@
@Before
public void setUP() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ProviderQpsControllerManager.getINSTANCE());
Utils.updateProperty(Config.PROVIDER_LIMIT_KEY_PREFIX + "test", 1);
}
+
+ @After
+ public void afterTest() {
+ ArchaiusUtils.resetConfig();
+
AbstractQpsControllerManagerTest.clearState(ProviderQpsControllerManager.getINSTANCE());
+ }
+
@Test
public void testGlobalQpsControl(final @Injectable Invocation invocation,
final @Injectable AsyncResponse asyncResp) throws Exception {
@@ -52,6 +65,8 @@ public void testGlobalQpsControl(final @Injectable Invocation
invocation,
{
invocation.getContext(Const.SRC_MICROSERVICE);
result = "test";
+ invocation.getOperationMeta();
+ result = AbstractQpsControllerManagerTest.getMockOperationMeta("pojo",
"server", "opr");
asyncResp.producerFail((Throwable) any);
result = new RuntimeException("test error");
}
@@ -67,7 +82,7 @@ public void testGlobalQpsControl(final @Injectable Invocation
invocation,
gHandler.handle(invocation, asyncResp);
count++;
gHandler.handle(invocation, asyncResp);
- Assert.assertTrue(false);
+ fail("An exception is expected!");
} catch (Exception e) {
Assert.assertEquals(2, count);
Assert.assertEquals("test error", e.getMessage());
@@ -75,7 +90,7 @@ public void testGlobalQpsControl(final @Injectable Invocation
invocation,
}
@Test
- public void testQpsController() throws Exception {
+ public void testQpsController() {
QpsController qpsController = new QpsController("abc", 100);
Assert.assertEquals(false, qpsController.isLimitNewRequest());
@@ -93,6 +108,7 @@ public void testHandleWithException() {
handler.handle(invocation, asyncResp);
handler.handle(invocation, asyncResp);
} catch (Exception e) {
+ e.printStackTrace();
validAssert = false;
}
Assert.assertTrue(validAssert);
@@ -104,6 +120,8 @@ public void testHandle() {
try {
validAssert = true;
Mockito.when(invocation.getContext(Const.SRC_MICROSERVICE)).thenReturn("test");
+ OperationMeta mockOperationMeta =
AbstractQpsControllerManagerTest.getMockOperationMeta("pojo", "server", "opr");
+
Mockito.when(invocation.getOperationMeta()).thenReturn(mockOperationMeta);
new MockUp<QpsController>() {
@Mock
@@ -121,6 +139,7 @@ protected QpsController create(String serviceName) {
};
handler.handle(invocation, asyncResp);
} catch (Exception e) {
+ e.printStackTrace();
validAssert = false;
}
Assert.assertTrue(validAssert);
@@ -132,6 +151,9 @@ public void testHandleIsLimitNewRequestAsFalse() {
try {
validAssert = true;
Mockito.when(invocation.getContext(Const.SRC_MICROSERVICE)).thenReturn("test");
+ OperationMeta mockOperationMeta = AbstractQpsControllerManagerTest
+ .getMockOperationMeta("pojo", "server", "opr");
+
Mockito.when(invocation.getOperationMeta()).thenReturn(mockOperationMeta);
new MockUp<QpsController>() {
@Mock
@@ -149,6 +171,7 @@ protected QpsController create(String serviceName) {
};
handler.handle(invocation, asyncResp);
} catch (Exception e) {
+ e.printStackTrace();
validAssert = false;
}
Assert.assertTrue(validAssert);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Support operation level flow control on provider side
> -----------------------------------------------------
>
> Key: SCB-352
> URL: https://issues.apache.org/jira/browse/SCB-352
> Project: Apache ServiceComb
> Issue Type: Task
> Reporter: YaoHaishi
> Assignee: YaoHaishi
> Priority: Major
> Attachments: compile_warning.PNG
>
>
> At present our provider qps flow control only support microservice level.
> We need to support schema/operation level qps flow control like consumer side.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
