handlers/handler-flowcontrol-qps/src/main/java/org/apache/servicecomb/qps/QpsControllerManager.java - servicecomb-java-chassis - Git at Google

 /*
  * 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.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;

 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.core.Invocation;
 import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
 import org.apache.servicecomb.foundation.common.exceptions.ServiceCombException;
 import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
 import org.apache.servicecomb.qps.strategy.AbstractQpsStrategy;
 import org.apache.servicecomb.qps.strategy.IStrategyFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.annotations.VisibleForTesting;
 import com.netflix.config.DynamicProperty;

 public class QpsControllerManager {
   private static final Logger LOGGER = LoggerFactory.getLogger(QpsControllerManager.class);

   public static final String SEPARATOR = ".";

   /**
    * Describe the relationship between configuration and qpsController.
    */
   private final Map<String, AbstractQpsStrategy> configQpsControllerMap = new ConcurrentHashMapEx<>();

   /**
    * Describe the relationship between qualifiedKey(format is "microservice.schema.operation") and qpsController.
    */
   private final Map<String, AbstractQpsStrategy> qualifiedNameControllerMap = new ConcurrentHashMapEx<>();

   private AbstractQpsStrategy globalQpsStrategy;

   private final String limitKeyPrefix;

   private final String bucketKeyPrefix;

   private final String globalLimitKey;

   private final String globalBucketKey;

   public QpsControllerManager(boolean isProvider) {
     if (isProvider) {
       limitKeyPrefix = Config.PROVIDER_LIMIT_KEY_PREFIX;
       bucketKeyPrefix = Config.PROVIDER_BUCKET_KEY_PREFIX;
       globalLimitKey = Config.PROVIDER_LIMIT_KEY_GLOBAL;
       globalBucketKey = Config.PROVIDER_BUCKET_KEY_GLOBAL;
     } else {
       limitKeyPrefix = Config.CONSUMER_LIMIT_KEY_PREFIX;
       bucketKeyPrefix = Config.CONSUMER_BUCKET_KEY_PREFIX;
       globalLimitKey = Config.CONSUMER_LIMIT_KEY_GLOBAL;
       globalBucketKey = Config.CONSUMER_BUCKET_KEY_GLOBAL;
     }

     initGlobalQpsController();
   }

   @VisibleForTesting
   public Map<String, AbstractQpsStrategy> getQualifiedNameControllerMap() {
     return qualifiedNameControllerMap;
   }

   public QpsStrategy getOrCreate(String microserviceName, Invocation invocation) {
     final String name = validatedName(microserviceName);
     return qualifiedNameControllerMap
         .computeIfAbsent(
             name + SEPARATOR + invocation.getOperationMeta().getSchemaQualifiedName(),
             key -> create(key, name, invocation));
   }

   private String validatedName(String microserviceName) {
     String name = microserviceName;
     if (StringUtils.isEmpty(microserviceName)) {
       name = Config.ANY_SERVICE;
     }
     return name;
   }

   /**
    * Create relevant qpsLimit dynamicProperty and watch the configuration change.
    * Search and return a valid qpsController.
    */
   @VisibleForTesting
   AbstractQpsStrategy create(String qualifiedNameKey, String microserviceName,
       Invocation invocation) {
     createForService(qualifiedNameKey, microserviceName, invocation);
     String qualifiedAnyServiceName = Config.ANY_SERVICE + qualifiedNameKey.substring(microserviceName.length());
     createForService(qualifiedAnyServiceName, Config.ANY_SERVICE, invocation);

     AbstractQpsStrategy strategy = searchQpsController(qualifiedNameKey);
     if (strategy == null) {
       strategy = searchQpsController(qualifiedAnyServiceName);
     }
     if (strategy == null) {
       return globalQpsStrategy;
     }
     return strategy;
   }

   private void createForService(String qualifiedNameKey, String microserviceName,
       Invocation invocation) {
     // create "microservice"
     createQpsControllerIfNotExist(microserviceName);
     // create "microservice.schema"
     createQpsControllerIfNotExist(
         qualifiedNameKey.substring(0, microserviceName.length() + invocation.getSchemaId().length() + 1));
     // create "microservice.schema.operation"
     createQpsControllerIfNotExist(qualifiedNameKey);
   }

   /**
    * <p> Use qualifiedNameKey to search {@link QpsStrategy}.
    * Firstly try to search "microservice.schema.operation". If no valid result found, then try "microservice.schema",
    * and then "microservice" or global qpsController(If there is a global qpsController).</p>
    * <p> This method ensures that there is always an existing qpsController returned, as the relevant qpsController has
    * been created and stored in {@link #create(String, String, Invocation)}</p>
    *
    * @param qualifiedNameKey qualifiedNameKey in {@link #qualifiedNameControllerMap}
    * @return a qps controller, lower level controllers with valid qpsLimit have priority.
    */
   private AbstractQpsStrategy searchQpsController(String qualifiedNameKey) {
     AbstractQpsStrategy qpsStrategy = configQpsControllerMap.get(qualifiedNameKey);
     if (isValidQpsController(qpsStrategy)) {
       return qpsStrategy;
     }

     int index = qualifiedNameKey.lastIndexOf(SEPARATOR);
     while (index > 0) {
       qpsStrategy = configQpsControllerMap.get(qualifiedNameKey.substring(0, index));
       if (isValidQpsController(qpsStrategy)) {
         return qpsStrategy;
       }

       index = qualifiedNameKey.lastIndexOf(SEPARATOR, index - 1);
     }

     if (isValidQpsController(qpsStrategy)) {
       return qpsStrategy;
     }

     return null;
   }

   private boolean isValidQpsController(AbstractQpsStrategy qpsStrategy) {
     return null != qpsStrategy && null != qpsStrategy.getQpsLimit();
   }

   private void createQpsControllerIfNotExist(String configKey) {
     if (configQpsControllerMap.containsKey(configKey)) {
       return;
     }

     LOGGER.info("Create qpsController, configKey = [{}]", configKey);
     DynamicProperty limitProperty = DynamicProperty.getInstance(limitKeyPrefix + configKey);
     DynamicProperty bucketProperty = DynamicProperty.getInstance(bucketKeyPrefix + configKey);
     DynamicProperty strategyProperty = DynamicProperty.getInstance(Config.STRATEGY_KEY);
     AbstractQpsStrategy qpsStrategy = chooseStrategy(configKey, limitProperty.getLong(),
         bucketProperty.getLong(), strategyProperty.getString());

     strategyProperty.addCallback(() -> {
       AbstractQpsStrategy innerQpsStrategy = chooseStrategy(configKey, limitProperty.getLong(),
           bucketProperty.getLong(), strategyProperty.getString());
       configQpsControllerMap.put(configKey, innerQpsStrategy);
       LOGGER.info("Global flow control strategy update, value = [{}]",
           strategyProperty.getString());
       updateObjMap();
     });
     limitProperty.addCallback(() -> {
       qpsStrategy.setQpsLimit(limitProperty.getLong());
       LOGGER.info("Qps limit updated, configKey = [{}], value = [{}]", configKey,
           limitProperty.getString());
       updateObjMap();
     });
     bucketProperty.addCallback(() -> {
       qpsStrategy.setBucketLimit(bucketProperty.getLong());
       LOGGER.info("bucket limit updated, configKey = [{}], value = [{}]", configKey,
           bucketProperty.getString());
       updateObjMap();
     });

     configQpsControllerMap.put(configKey, qpsStrategy);
   }

   protected void updateObjMap() {
     Iterator<Entry<String, AbstractQpsStrategy>> it = qualifiedNameControllerMap.entrySet().iterator();
     while (it.hasNext()) {
       Map.Entry<String, AbstractQpsStrategy> entry = it.next();
       AbstractQpsStrategy qpsStrategy = searchQpsController(entry.getKey());
       if (qpsStrategy == null) {
         it.remove();
         continue;
       }
       if (qpsStrategy != entry.getValue()) {
         entry.setValue(qpsStrategy);
         LOGGER.info("QpsController updated, operationId = [{}], configKey = [{}], qpsLimit = [{}]",
             entry.getKey(), qpsStrategy.getKey(), qpsStrategy.getQpsLimit());
       }
     }
   }

   private void initGlobalQpsController() {
     DynamicProperty globalLimitProperty = DynamicProperty.getInstance(globalLimitKey);
     DynamicProperty globalBucketProperty = DynamicProperty.getInstance(globalBucketKey);
     DynamicProperty globalStrategyProperty = DynamicProperty
         .getInstance(Config.STRATEGY_KEY);
     globalQpsStrategy = chooseStrategy(globalLimitKey, globalLimitProperty.getLong((long) Integer.MAX_VALUE),
         globalBucketProperty.getLong(), globalStrategyProperty.getString());
     globalStrategyProperty.addCallback(() -> {
       globalQpsStrategy = chooseStrategy(globalLimitKey, globalLimitProperty.getLong((long) Integer.MAX_VALUE),
           globalBucketProperty.getLong(), globalStrategyProperty.getString());
       LOGGER.info("Global flow control strategy update, value = [{}]",
           globalStrategyProperty.getString());
     });
     globalLimitProperty.addCallback(() -> {
       globalQpsStrategy.setQpsLimit(globalLimitProperty.getLong((long) Integer.MAX_VALUE));
       LOGGER.info("Global qps limit update, value = [{}]", globalLimitProperty.getLong());
     });
     globalBucketProperty.addCallback(() -> {
       globalQpsStrategy.setBucketLimit(globalBucketProperty.getLong());
       LOGGER.info("Global bucket limit update, value = [{}]", globalBucketProperty.getLong());
     });
   }

   private AbstractQpsStrategy chooseStrategy(String configKey, Long limit, Long bucket,
       String strategyName) {
     if (StringUtils.isEmpty(strategyName)) {
       strategyName = "FixedWindow";
     }
     AbstractQpsStrategy strategy = null;
     List<IStrategyFactory> strategyFactories = SPIServiceUtils
         .getOrLoadSortedService(IStrategyFactory.class);
     for (IStrategyFactory strategyFactory : strategyFactories) {
       strategy = strategyFactory.createStrategy(strategyName);
       if (strategy != null) {
         break;
       }
     }
     if (strategy == null) {
       throw new ServiceCombException(
           "the qps strategy name " + strategyName + " is not exist , please check.");
     }
     strategy.setKey(configKey);
     strategy.setQpsLimit(limit);
     strategy.setBucketLimit(bucket);
     return strategy;
   }
 }
	/*
	* 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.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.Map.Entry;

	import org.apache.commons.lang3.StringUtils;
	import org.apache.servicecomb.core.Invocation;
	import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
	import org.apache.servicecomb.foundation.common.exceptions.ServiceCombException;
	import org.apache.servicecomb.foundation.common.utils.SPIServiceUtils;
	import org.apache.servicecomb.qps.strategy.AbstractQpsStrategy;
	import org.apache.servicecomb.qps.strategy.IStrategyFactory;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.annotations.VisibleForTesting;
	import com.netflix.config.DynamicProperty;

	public class QpsControllerManager {
	private static final Logger LOGGER = LoggerFactory.getLogger(QpsControllerManager.class);

	public static final String SEPARATOR = ".";

	/**
	* Describe the relationship between configuration and qpsController.
	*/
	private final Map<String, AbstractQpsStrategy> configQpsControllerMap = new ConcurrentHashMapEx<>();

	/**
	* Describe the relationship between qualifiedKey(format is "microservice.schema.operation") and qpsController.
	*/
	private final Map<String, AbstractQpsStrategy> qualifiedNameControllerMap = new ConcurrentHashMapEx<>();

	private AbstractQpsStrategy globalQpsStrategy;

	private final String limitKeyPrefix;

	private final String bucketKeyPrefix;

	private final String globalLimitKey;

	private final String globalBucketKey;

	public QpsControllerManager(boolean isProvider) {
	if (isProvider) {
	limitKeyPrefix = Config.PROVIDER_LIMIT_KEY_PREFIX;
	bucketKeyPrefix = Config.PROVIDER_BUCKET_KEY_PREFIX;
	globalLimitKey = Config.PROVIDER_LIMIT_KEY_GLOBAL;
	globalBucketKey = Config.PROVIDER_BUCKET_KEY_GLOBAL;
	} else {
	limitKeyPrefix = Config.CONSUMER_LIMIT_KEY_PREFIX;
	bucketKeyPrefix = Config.CONSUMER_BUCKET_KEY_PREFIX;
	globalLimitKey = Config.CONSUMER_LIMIT_KEY_GLOBAL;
	globalBucketKey = Config.CONSUMER_BUCKET_KEY_GLOBAL;
	}

	initGlobalQpsController();
	}

	@VisibleForTesting
	public Map<String, AbstractQpsStrategy> getQualifiedNameControllerMap() {
	return qualifiedNameControllerMap;
	}

	public QpsStrategy getOrCreate(String microserviceName, Invocation invocation) {
	final String name = validatedName(microserviceName);
	return qualifiedNameControllerMap
	.computeIfAbsent(
	name + SEPARATOR + invocation.getOperationMeta().getSchemaQualifiedName(),
	key -> create(key, name, invocation));
	}

	private String validatedName(String microserviceName) {
	String name = microserviceName;
	if (StringUtils.isEmpty(microserviceName)) {
	name = Config.ANY_SERVICE;
	}
	return name;
	}

	/**
	* Create relevant qpsLimit dynamicProperty and watch the configuration change.
	* Search and return a valid qpsController.
	*/
	@VisibleForTesting
	AbstractQpsStrategy create(String qualifiedNameKey, String microserviceName,
	Invocation invocation) {
	createForService(qualifiedNameKey, microserviceName, invocation);
	String qualifiedAnyServiceName = Config.ANY_SERVICE + qualifiedNameKey.substring(microserviceName.length());
	createForService(qualifiedAnyServiceName, Config.ANY_SERVICE, invocation);

	AbstractQpsStrategy strategy = searchQpsController(qualifiedNameKey);
	if (strategy == null) {
	strategy = searchQpsController(qualifiedAnyServiceName);
	}
	if (strategy == null) {
	return globalQpsStrategy;
	}
	return strategy;
	}

	private void createForService(String qualifiedNameKey, String microserviceName,
	Invocation invocation) {
	// create "microservice"
	createQpsControllerIfNotExist(microserviceName);
	// create "microservice.schema"
	createQpsControllerIfNotExist(
	qualifiedNameKey.substring(0, microserviceName.length() + invocation.getSchemaId().length() + 1));
	// create "microservice.schema.operation"
	createQpsControllerIfNotExist(qualifiedNameKey);
	}

	/**
	* <p> Use qualifiedNameKey to search {@link QpsStrategy}.
	* Firstly try to search "microservice.schema.operation". If no valid result found, then try "microservice.schema",
	* and then "microservice" or global qpsController(If there is a global qpsController).</p>
	* <p> This method ensures that there is always an existing qpsController returned, as the relevant qpsController has
	* been created and stored in {@link #create(String, String, Invocation)}</p>
	*
	* @param qualifiedNameKey qualifiedNameKey in {@link #qualifiedNameControllerMap}
	* @return a qps controller, lower level controllers with valid qpsLimit have priority.
	*/
	private AbstractQpsStrategy searchQpsController(String qualifiedNameKey) {
	AbstractQpsStrategy qpsStrategy = configQpsControllerMap.get(qualifiedNameKey);
	if (isValidQpsController(qpsStrategy)) {
	return qpsStrategy;
	}

	int index = qualifiedNameKey.lastIndexOf(SEPARATOR);
	while (index > 0) {
	qpsStrategy = configQpsControllerMap.get(qualifiedNameKey.substring(0, index));
	if (isValidQpsController(qpsStrategy)) {
	return qpsStrategy;
	}

	index = qualifiedNameKey.lastIndexOf(SEPARATOR, index - 1);
	}

	if (isValidQpsController(qpsStrategy)) {
	return qpsStrategy;
	}

	return null;
	}

	private boolean isValidQpsController(AbstractQpsStrategy qpsStrategy) {
	return null != qpsStrategy && null != qpsStrategy.getQpsLimit();
	}

	private void createQpsControllerIfNotExist(String configKey) {
	if (configQpsControllerMap.containsKey(configKey)) {
	return;
	}

	LOGGER.info("Create qpsController, configKey = [{}]", configKey);
	DynamicProperty limitProperty = DynamicProperty.getInstance(limitKeyPrefix + configKey);
	DynamicProperty bucketProperty = DynamicProperty.getInstance(bucketKeyPrefix + configKey);
	DynamicProperty strategyProperty = DynamicProperty.getInstance(Config.STRATEGY_KEY);
	AbstractQpsStrategy qpsStrategy = chooseStrategy(configKey, limitProperty.getLong(),
	bucketProperty.getLong(), strategyProperty.getString());

	strategyProperty.addCallback(() -> {
	AbstractQpsStrategy innerQpsStrategy = chooseStrategy(configKey, limitProperty.getLong(),
	bucketProperty.getLong(), strategyProperty.getString());
	configQpsControllerMap.put(configKey, innerQpsStrategy);
	LOGGER.info("Global flow control strategy update, value = [{}]",
	strategyProperty.getString());
	updateObjMap();
	});
	limitProperty.addCallback(() -> {
	qpsStrategy.setQpsLimit(limitProperty.getLong());
	LOGGER.info("Qps limit updated, configKey = [{}], value = [{}]", configKey,
	limitProperty.getString());
	updateObjMap();
	});
	bucketProperty.addCallback(() -> {
	qpsStrategy.setBucketLimit(bucketProperty.getLong());
	LOGGER.info("bucket limit updated, configKey = [{}], value = [{}]", configKey,
	bucketProperty.getString());
	updateObjMap();
	});

	configQpsControllerMap.put(configKey, qpsStrategy);
	}

	protected void updateObjMap() {
	Iterator<Entry<String, AbstractQpsStrategy>> it = qualifiedNameControllerMap.entrySet().iterator();
	while (it.hasNext()) {
	Map.Entry<String, AbstractQpsStrategy> entry = it.next();
	AbstractQpsStrategy qpsStrategy = searchQpsController(entry.getKey());
	if (qpsStrategy == null) {
	it.remove();
	continue;
	}
	if (qpsStrategy != entry.getValue()) {
	entry.setValue(qpsStrategy);
	LOGGER.info("QpsController updated, operationId = [{}], configKey = [{}], qpsLimit = [{}]",
	entry.getKey(), qpsStrategy.getKey(), qpsStrategy.getQpsLimit());
	}
	}
	}

	private void initGlobalQpsController() {
	DynamicProperty globalLimitProperty = DynamicProperty.getInstance(globalLimitKey);
	DynamicProperty globalBucketProperty = DynamicProperty.getInstance(globalBucketKey);
	DynamicProperty globalStrategyProperty = DynamicProperty
	.getInstance(Config.STRATEGY_KEY);
	globalQpsStrategy = chooseStrategy(globalLimitKey, globalLimitProperty.getLong((long) Integer.MAX_VALUE),
	globalBucketProperty.getLong(), globalStrategyProperty.getString());
	globalStrategyProperty.addCallback(() -> {
	globalQpsStrategy = chooseStrategy(globalLimitKey, globalLimitProperty.getLong((long) Integer.MAX_VALUE),
	globalBucketProperty.getLong(), globalStrategyProperty.getString());
	LOGGER.info("Global flow control strategy update, value = [{}]",
	globalStrategyProperty.getString());
	});
	globalLimitProperty.addCallback(() -> {
	globalQpsStrategy.setQpsLimit(globalLimitProperty.getLong((long) Integer.MAX_VALUE));
	LOGGER.info("Global qps limit update, value = [{}]", globalLimitProperty.getLong());
	});
	globalBucketProperty.addCallback(() -> {
	globalQpsStrategy.setBucketLimit(globalBucketProperty.getLong());
	LOGGER.info("Global bucket limit update, value = [{}]", globalBucketProperty.getLong());
	});
	}

	private AbstractQpsStrategy chooseStrategy(String configKey, Long limit, Long bucket,
	String strategyName) {
	if (StringUtils.isEmpty(strategyName)) {
	strategyName = "FixedWindow";
	}
	AbstractQpsStrategy strategy = null;
	List<IStrategyFactory> strategyFactories = SPIServiceUtils
	.getOrLoadSortedService(IStrategyFactory.class);
	for (IStrategyFactory strategyFactory : strategyFactories) {
	strategy = strategyFactory.createStrategy(strategyName);
	if (strategy != null) {
	break;
	}
	}
	if (strategy == null) {
	throw new ServiceCombException(
	"the qps strategy name " + strategyName + " is not exist , please check.");
	}
	strategy.setKey(configKey);
	strategy.setQpsLimit(limit);
	strategy.setBucketLimit(bucket);
	return strategy;
	}
	}