Gateway applications need to support a variety of load balancing strategies, including random,Hashing, RoundRobin and so on. In Apache Shenyu gateway, it not only realizes such traditional algorithms, but also makes smoother traffic processing for the entry of server nodes through detailed processing such as traffic warm-up, so as to obtain better overall stability. In this article, let's walk through how Apache Shenyu is designed and implemented this part of the function.

This article based on shenyu-2.5.0 version of the source code analysis.

[TOC]

LoadBalancer SPI#

The implementation of LoadBalancer is in shenyu-loadbalancer module. It has based on its SPI creation mechanism. The core interface code is shown as follows. This interface well explains the concept: load balancing is to select the most appropriate node from a series of server nodes. Routing, traffic processing and load balancing is the basic function of LoadBalancer SPI.

@SPIpublic interface LoadBalancer {
    /**     * this is select one for upstream list.     *     * @param upstreamList upstream list     * @param ip ip     * @return upstream     */    Upstream select(List<Upstream> upstreamList, String ip);}

Where upstreamList represents the server nodes list available for routing. Upstream is the data structure of server node, the important elements including protocol, upstreamUrl , weight, timestamp, warmup、healthy.

public class Upstream {    /**     * protocol.     */    private final String protocol;
    /**     * url.     */    private String url;
    /**     * weight.     */    private final int weight;
    /**     * false close, true open.     */    private boolean status;
    /**     * startup time.     */    private final long timestamp;
    /**     * warmup.     */    private final int warmup;
    /**     * healthy.     */    private boolean healthy;
    /**     * lastHealthTimestamp.     */    private long lastHealthTimestamp;
    /**     * lastUnhealthyTimestamp.     */    private long lastUnhealthyTimestamp;
    /**     * group.     */    private String group;
    /**     * version.     */    private String version;}

Design of LoadBalancer module#

The class diagram of LoadBalancer moduleisshown as follows.

We can draw the outline of LoadBalancer module from the class diagram:

1. The abstract class AbstractLoadBalancer implements the SPI LoadBalancer interface，and supplies the template methods for selection related, such as select(), selector()，and gives the calculation of weight.

2. Three implementation classes which inherit AbstractLoadBalancer to realize their own logic:

   • RandomLoadBalancer - Weight Random
   • HashLoadBalancer - Consistent Hashing
   • RoundRobinLoadBalancer -Weight Round Robin per-packet

3. The factory class LoadBalancerFactory provides public static method to be called.

   The implementation classes and algorithms are configurable. According to its specification, by adding profile in SHENYU_DIERECTORY directory, the data in profile should be key=value-class format, where the value-class will be load by the Apache Shenyu SPI class loader, and key value should be an name defined in LoadBalanceEnum.

random=org.apache.shenyu.loadbalancer.spi.RandomLoadBalancerroundRobin=org.apache.shenyu.loadbalancer.spi.RoundRobinLoadBalancerhash=org.apache.shenyu.loadbalancer.spi.HashLoadBalancer

The code of LoadBalanceEnum is as follows：

public enum LoadBalanceEnum {    /**     * Hash load balance enum.     */    HASH(1, "hash", true),
    /**     * Random load balance enum.     */    RANDOM(2, "random", true),
    /**     * Round robin load balance enum.     */    ROUND_ROBIN(3, "roundRobin", true);
    private final int code;    private final String name;    private final boolean support;}

AbstractLoadBalancer#

This abstract class implements the LoadBalancer interface and define the abstract method doSelect() to be processed by the implementation classes. In the template method select(), It will do validation first then call the doSelect() method.

public abstract class AbstractLoadBalancer implements LoadBalancer {    /**     * Do select divide upstream.     *     * @param upstreamList the upstream list     * @param ip           the ip     * @return the divide upstream     */    protected abstract Upstream doSelect(List<Upstream> upstreamList, String ip);
    @Override    public Upstream select(final List<Upstream> upstreamList, final String ip) {        if (CollectionUtils.isEmpty(upstreamList)) {            return null;        }        if (upstreamList.size() == 1) {            return upstreamList.get(0);        }        return doSelect(upstreamList, ip);    }}

When the timestamp of server node is not null, and the interval between current time and timestamp is within the traffic warm-up time, the formula for weight calculation is. $$ {1-1} ww = min(1,uptime/(warmup/weight)) $$ It can be seen from the formula that the final weight(ww) is proportional to the original-weight value. The closer the time interval is to the warmup time, the greater the final ww. That is, the longer the waiting time of the request, the higher the final weight. When there is no timestamp or other conditions, the ww is equal to the weight value of Upstream object.

The central of thinking about warm-upis to avoid bad performance when adding new server and the new JVMs starting up.

Let's see how the load balancing with Random, Hashing and RoundRobin strategy is implemented.

RandomLoadBalancer#

The RandomLoadBalancer can handle two situations:

1. Each node without weight, or every node has the same weight, randomly choose one.
2. Server Nodes with different weight, choose one randomly by weight.

Following is the random() method of RandomLoadBalancer. When traversing server node list, if the randomly generated value is less than the weight of node, then the current node will be chosen. If after one round traversing, there is no server node match, then it will choose one randomly. The getWeight(final Upstream upstream) is defined in AbstractLoadBalancer class.

    @Override    public Upstream doSelect(final List<Upstream> upstreamList, final String ip) {        int length = upstreamList.size();        // every upstream has the same weight?        boolean sameWeight = true;        // the weight of every upstream        int[] weights = new int[length];        int firstUpstreamWeight = getWeight(upstreamList.get(0));        weights[0] = firstUpstreamWeight;        // init the totalWeight        int totalWeight = firstUpstreamWeight;        int halfLengthTotalWeight = 0;        for (int i = 1; i < length; i++) {            int currentUpstreamWeight = getWeight(upstreamList.get(i));            if (i <= (length + 1) / 2) {                halfLengthTotalWeight = totalWeight;            }            weights[i] = currentUpstreamWeight;            totalWeight += currentUpstreamWeight;            if (sameWeight && currentUpstreamWeight != firstUpstreamWeight) {                // Calculate whether the weight of ownership is the same.                sameWeight = false;            }        }        if (totalWeight > 0 && !sameWeight) {            return random(totalWeight, halfLengthTotalWeight, weights, upstreamList);        }        return random(upstreamList);    }
    private Upstream random(final int totalWeight, final int halfLengthTotalWeight, final int[] weights, final List<Upstream> upstreamList) {        // If the weights are not the same and the weights are greater than 0, then random by the total number of weights.        int offset = RANDOM.nextInt(totalWeight);        int index = 0;        int end = weights.length;        if (offset >= halfLengthTotalWeight) {            index = (weights.length + 1) / 2;            offset -= halfLengthTotalWeight;        } else {            end = (weights.length + 1) / 2;        }        // Determine which segment the random value falls on        for (; index < end; index++) {            offset -= weights[index];            if (offset < 0) {                return upstreamList.get(index);            }        }        return random(upstreamList);    }

HashLoadBalancer#

In HashLoadBalancer, it takes the advantages of consistent hashing , that maps both the input traffic and the servers to a unit circle, or name as hash ring. For the requestedip address, with its hash value to find the node closest in clockwise order as the node to be routed. Let's see how consistent hashing is implemented in HashLoadBalancer.

As to the hash algorithms, HashLoadBalancer uses MD5 hash, which has the advantage of mixing the input in an unpredictable but deterministic way. The output is a 32-bit integer. the code is shown as follows:

private static long hash(final String key) {    // md5 byte    MessageDigest md5;    try {        md5 = MessageDigest.getInstance("MD5");    } catch (NoSuchAlgorithmException e) {        throw new ShenyuException("MD5 not supported", e);    }    md5.reset();    byte[] keyBytes;    keyBytes = key.getBytes(StandardCharsets.UTF_8);    md5.update(keyBytes);    byte[] digest = md5.digest();    // hash code, Truncate to 32-bits    long hashCode = (long) (digest[3] & 0xFF) << 24            | ((long) (digest[2] & 0xFF) << 16)            | ((long) (digest[1] & 0xFF) << 8)            | (digest[0] & 0xFF);    return hashCode & 0xffffffffL;}

Importantly, how to generate the hash ring and avoid skewness? Let's thedoSelect() method inHashLoadBalancer as follows:

    private static final int VIRTUAL_NODE_NUM = 5;
    @Override    public Upstream doSelect(final List<Upstream> upstreamList, final String ip) {        final ConcurrentSkipListMap<Long, Upstream> treeMap = new ConcurrentSkipListMap<>();        upstreamList.forEach(upstream -> IntStream.range(0, VIRTUAL_NODE_NUM).forEach(i -> {            long addressHash = hash("SHENYU-" + upstream.getUrl() + "-HASH-" + i);            treeMap.put(addressHash, upstream);        }));        long hash = hash(ip);        SortedMap<Long, Upstream> lastRing = treeMap.tailMap(hash);        if (!lastRing.isEmpty()) {            return lastRing.get(lastRing.firstKey());        }        return treeMap.firstEntry().getValue();    }

In this method, duplicated labels are used which are called "virtual nodes" (i.e. 5 virtual nodes point to a single "real" server). It will make the distribution in hash ring more evenly, and reduce the occurrence of data skewness.

In order to rescue the data sorted in the hash ring, and can be accessed quickly, we use ConcurrentSkipListMap of Java to store the server node lists ( with virtual nodes) and its hash value as key. This class a member of Java Collections Framework, providing expected average log(n) time cost for retrieve and access operations safely execute concurrent by multiple threads.

Furthermore, the method tailMap(K fromKey) of ConcurrentSkipListMap can return a view of portion of the map whose keys are greater or equal to the fromKey, and not need to navigate the whole map.

In the above code section, after the hash ring is generated, it uses tailMap(K fromKey) of ConcurrentSkipListMap to find the subset that the elements greater, or equal to the hash value of the requested ip, its first element is just the node to be routed. With the suitable data structure, the code looks particularly clear and concise.

Consistent hashing resolved the poor scalability of the traditional hashing by modular operation.

RoundRobinLoadBalancer#

The original Round-robin selection is to select server nodes one by one from the candidate list. Whenever some nodes has crash ( ex, cannot be connected after 1 minute), it will be removed from the candidate list, and do not attend the next round, until the server node is recovered and it will be add to the candidate list again. In RoundRobinLoadBalancer,the weight Round Robin per-packet schema is implemented.

In order to work in concurrent system, it provides an inner static class WeigthRoundRobin to store and calculate the rolling selections of each server node. Following is the main section of this class( removed remark )

protected static class WeightedRoundRobin {
    private int weight;
    private final AtomicLong current = new AtomicLong(0);
    private long lastUpdate;
    void setWeight(final int weight) {        this.weight = weight;        current.set(0);    }    long increaseCurrent() {        return current.addAndGet(weight);    }
    void sel(final int total) {        current.addAndGet(-1 * total);    }    void setLastUpdate(final long lastUpdate) {        this.lastUpdate = lastUpdate;    }}

Please focus on the these method:

• setWeight(final int weight) : set the current value by weight

• increaseCurrent(): Increment the current value by weight, and current set to 0.

• sel(final int total): decrement the current value by total

  Let's see how the weight factor being used in this round-robin selection?

  First it defines a two-level ConcurrentMap type variable named as methodWeightMap , to cache the server node lists and the rolling selection data about each server node.

private final ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<>(16);

In this map, the key of first level is set to upstreamUrl of first element in server node list. The type of second object is ConcurrentMap<String, WeightedRoundRobin>, the key of this inner Map is the value upstreamUrlvariable of each server node in this server list, the value object is WeightedRoundRobin, used to trace the rolling selection data about each server node. As to the implementation class for the Map object, we use ConcurrentHashMap of JUC, a hash table supporting full concurrency of retrievals and high expected concurrency for updates.

In the second level of the map, the embedded static class - WeighedRoundRobin of each node is thread-safe, implementing the weighted RoundRobin per bucket. The following is the code of the doselect() method of this class.

@Overridepublic Upstream doSelect(final List<Upstream> upstreamList, final String ip) {    String key = upstreamList.get(0).getUrl();    ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);    if (Objects.isNull(map)) {        methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<>(16));        map = methodWeightMap.get(key);    }    int totalWeight = 0;    long maxCurrent = Long.MIN_VALUE;    long now = System.currentTimeMillis();    Upstream selectedInvoker = null;    WeightedRoundRobin selectedWeightedRoundRobin = null;    for (Upstream upstream : upstreamList) {        String rKey = upstream.getUrl();        WeightedRoundRobin weightedRoundRobin = map.get(rKey);        int weight = getWeight(upstream);        if (Objects.isNull(weightedRoundRobin)) {            weightedRoundRobin = new WeightedRoundRobin();            weightedRoundRobin.setWeight(weight);            map.putIfAbsent(rKey, weightedRoundRobin);        }        if (weight != weightedRoundRobin.getWeight()) {            // weight changed.            weightedRoundRobin.setWeight(weight);        }        long cur = weightedRoundRobin.increaseCurrent();        weightedRoundRobin.setLastUpdate(now);        if (cur > maxCurrent) {            maxCurrent = cur;            selectedInvoker = upstream;            selectedWeightedRoundRobin = weightedRoundRobin;        }        totalWeight += weight;    }    ......  //erase the section which handles the time-out upstreams.     if (selectedInvoker != null) {        selectedWeightedRoundRobin.sel(totalWeight);        return selectedInvoker;    }    // should not happen here    return upstreamList.get(0);}

For example we assume upstreamUrl values of three server nodes is: LIST = [upstream-20, upstream-50, upstream-30]. After a round of execution, the data in newly created methodWeightMap is as follows:

For the above example LIST, assumes the weight array is [20,50,30]. the following figure shows the value change and polling selection process of the current array in WeighedRoundRobin object.

In each round, it will choose the server node with max current value.

• Round1:
  • Traverse the server node list, initialize the weightedRoundRobin instance of each server node or update the weight value of server nodes object Upstream
  • Traverse the server node list, initialize the weightedRoundRobin instance of each server node or update the weight value of server nodes object Upstream
  • say, in this case, after traverse, the current array of the node list changes to [20, 50,30]，so according to rule, the node Stream-50 would be chosen, and then the static object WeightedRoundRobin of Stream-50 executes sel(-total) , the current array is now [20,-50, 30].
• Round 2: after traverse, the current array should be [40,0,60], so the Stream-30 node would be chosen， current array is now [40,0,-40].
• Round 3: after traverse, current array changes to [60,50,-10], Stream-20 would be chosen，and current array is now [-40,50,-10].

When there is any inconsistence or some server crashed, for example, the lists size does not match with the elements in map, it would copy and modify the element with lock mechanism, and remove the timeout server node, the data in Map updated. Following is the fault tolerance code segment.

    if (!updateLock.get() && upstreamList.size() != map.size() && updateLock.compareAndSet(false, true)) {        try {            // copy -> modify -> update reference.            ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<>(map);            newMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > recyclePeriod);            methodWeightMap.put(key, newMap);        } finally {            updateLock.set(false);        }    }    if (Objects.nonNull(selectedInvoker)) {        selectedWeightedRoundRobin.sel(totalWeight);        return selectedInvoker;    }    // should not happen here.    return upstreamList.get(0);

LoadBalancerFactory#

In this class, a static method calling LoadBalancer is provided, whereExtensionLoader is the entry point of Apache Shenyu SPI. That is to say, LoadBalancer module is configurable and extensible. The algorithm variable in this static method is the name enumeration type defined in LoadBalanceEnum.

    /**     * Selector upstream.     *     * @param upstreamList the upstream list     * @param algorithm    the loadBalance algorithm     * @param ip           the ip     * @return the upstream     */    public static Upstream selector(final List<Upstream> upstreamList, final String algorithm, final String ip) {        LoadBalancer loadBalance = ExtensionLoader.getExtensionLoader(LoadBalancer.class).getJoin(algorithm);        return loadBalance.select(upstreamList, ip);    }

Using of LoadBalancer module#

In the above section, we describe the LoadBalancer SPI and three implementation classes. Let's take a look at how the LoadBalancer to be used in Apache Shenyu. DividePlugin is a plugin in Apache Shenyu responsible for routing http request. when enable to use this plugin, it will transfer traffic according to selection data and rule data, and deliver to next plugin downstream.

@Overrideprotected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {   ......}

The type of second parameter of doExecute() is ShenyuPluginChain, which represents the execution chain of plugins. For details, see the mechanism of Apache Shenyu Plugins. The third one is SelectorData type, and the fourth is RuleData type working as the rule data.

In doExecute() of DividePlugin, first verify the size of header, content length, etc, then preparing for load balancing.

Following is a code fragment usingLoadBalancer in the doExecute() method:

    // find the routing server node list    List<Upstream> upstreamList = UpstreamCacheManager.getInstance().findUpstreamListBySelectorId(selector.getId());    ...     // the requested ip    String ip = Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();
    //calling the Utility class and invoke the LoadBalance processing.    Upstream upstream = LoadBalancerFactory.selector(upstreamList, ruleHandle.getLoadBalance(), ip);

In the above code, the output ofruleHandle.getLoadBalance() is the name variable defined in LoadBalanceEnum, that is random, hash, roundRobin, etc. It is very convenient to use LoadBalancer by LoadBalancerFactory. When adding more LoadBalancer implementing classes, the interface in plugin module will not be effect at all.

Summary#

After reading through the code of LoadBalancer module, from the design perspective, it is concluded that this module has the following characteristics:

1. Extensibility: Interface oriented design and implemented on Apache Shenyu SPI mechanism, it can be easily extended to other dynamic load balancing algorithms (for example, least connection, fastest mode, etc), and supports cluster processing.
2. Scalability： Every load balancing implementation, weighted Random, consistency Hashing and weighted RoundRobin can well support increase or decrease cluster overall capacity.
3. More detailed design such as warm-up can bring better performance and obtain better overall stability.

In most of the plugins ( such as Dubbo, gRPC,Spring-cloud, etc) of Apache Shenyu, the routingparameters are designed to support the combination of multiple conditions. In order to realize such requirements, the parameters and behaviors are abstracted to three parts according to its SPI mechanism, and implemented in shenyu-plugin-base module.

• ParameterData-parameters
• PredictJudge-predicate
• MatchStrategy-matching strategy

Relatively speaking, the MatchStrategy is the part that needs the least extension points. For the combined judgement of multiple conditions, the common selection rules are: All conditions are matched, at least one is matched, at least the first is met, or most of conditions satisfied. As we will need to handle various types of parameters, for example: IP, header, uri, etc.

How to make the MatchStrategy to be simple to use and extensible?

MatchStrategy#

The implementation of MatchStrategy is in shenyu-plugin-base module. It is based on the SPI creation mechanism, and has used factory pattern and strategy design pattern. The class diagram of MatchStrategy is showed as follows.

Based on the interface MatchStrategy we design the implementation classes, and the abstract class AbstractMatchStrategy supplies common method, while the factory class MatchStrategyFactory provides creation functions.

MatchStrategy Interface#

First, let's look at the MatchStrategy SPI interface

@SPIpublic interface MatchStrategy {
    Boolean match(List<ConditionData> conditionDataList, ServerWebExchange exchange);}

The annotation @SPI means that this is an SPI interface. Where ServerWebExchange is org.springframework.web.server.ServerWebExchange, represents the request-response interactive content of HTTP. Following is the code of ConditionData, the more detail about this class can refer to code analysis of PredicteJudge

public class ConditionData {
    private String paramType;    private String operator;
    private String paramName;    private String paramValue;}

AbstractMatchStrategy#

Second, let's look at the abstract class AbstractMatchStrategy，it has defined a buildRealData method，In this method it wraps various parameters to a unified interface through the functionality of ParameterDataFactory, which is the factory class of ParameterData. It supports a variety of types of parameters , such as Ip, Cookie, Header,uri, etc. Modifications of such parameters will not impact the calling of matching rules of MatchStrategy.

public abstract class AbstractMatchStrategy {
    public String buildRealData(final ConditionData condition, final ServerWebExchange exchange) {        return ParameterDataFactory.builderData(condition.getParamType(), condition.getParamName(), exchange);    }}

Implementation class and profile#

Now, let's look at the two implementation class based on the above interface in shenyu-plugin-base module , that is:

• AndMatchStrategy- AND -All conditions are matched

• OrMatchStrategy- OR -at least one is match

  The properties file containing the SPI implementation is shown as follows, which located at the SHENYU_DIRECTORYdirectory. When starting up, the top-level SPI classes will read the key-value and load the classes and cache them.

and=org.apache.shenyu.plugin.base.condition.strategy.AndMatchStrategyor=org.apache.shenyu.plugin.base.condition.strategy.OrMatchStrategy

These two implementation classes inherit AbstractMatchStrategy class and implement MatchStrategy interface.

AndMatchStrategy- “AND” relation#

Since the PredicateJudge interface can encapsulate different variety of Predicates , for example EqualsPredicateJudge, EndsWithPredicateJudge and so on, the ConditionData and ParamData passed to it can present with variety of parameters, for treating of multiple conditions. So usingstream and lambda expression, it can be very simple and efficient to process "AND" logic (all conditions must be matched).

@Joinpublic class AndMatchStrategy extends AbstractMatchStrategy implements MatchStrategy {
    @Override    public Boolean match(final List<ConditionData> conditionDataList, final ServerWebExchange exchange) {        return conditionDataList                .stream()                .allMatch(condition -> PredicateJudgeFactory.judge(condition, buildRealData(condition, exchange)));    }}

The OrMatchStrategy similarly implements the "OR" logic- at least one is match.

MatchStrategyFactory#

This is the factory class of MatchStrategy，there are two methods, one is newInstance(), which will return the MatchStrategy implementation class instance cached by the SPI ExtensionLoader indexed by the key-value.

    public static MatchStrategy newInstance(final Integer strategy) {        String matchMode = MatchModeEnum.getMatchModeByCode(strategy);        return ExtensionLoader.getExtensionLoader(MatchStrategy.class).getJoin(matchMode);    }

the matchMode will be the name of strategy, the value will be "and" or "or". The MatchModeEnum defines the code and name of match strategy as follows.

AND(0, "and"), OR(1, "or");

Another method is match() method, which will invoke the match() method of implementation class.

    public static boolean match(final Integer strategy, final List<ConditionData> conditionDataList, final ServerWebExchange exchange) {        return newInstance(strategy).match(conditionDataList, exchange);    }

How it works#

AbstractShenyuPlugin is the base class of plugins in shenyu-plugin module. In this class two selection method are defined: filterSelector() and filterRule() , Both of them call the match() method of MatchStrategyFactory. The code of filterSelector() is shown as follows.

    private Boolean filterSelector(final SelectorData selector, final ServerWebExchange exchange) {        if (selector.getType() == SelectorTypeEnum.CUSTOM_FLOW.getCode()) {            if (CollectionUtils.isEmpty(selector.getConditionList())) {                return false;            }            return MatchStrategyFactory.match(selector.getMatchMode(), selector.getConditionList(), exchange);        }        return true;    }

In filterSelector() method, after validation of the SelectorData, calls the match method of MatchStrategyFactory, and then this factory class will invokes the match method of corresponding implementation class.

    private Boolean filterRule(final RuleData ruleData, final ServerWebExchange exchange) {        return ruleData.getEnabled() && MatchStrategyFactory.match(ruleData.getMatchMode(), ruleData.getConditionDataList(), exchange);    }

In filterRule() it is also calls the match() method of MatchStrategyFactory. Does it look particularly concise or even simple? In the code analysis of PredicteJudge , you can see more detail about parameter processing in shenyu-plugin.

Summary#

Due to the use of SPI mechanism of Apache Shenyu, the parameter selection module has the characteristic of loose coupling and extensibility. In terms of the combination of multiple conditions, MatchStrategy provides a good design. Although currently only two implementation classes are present, it can be easily used to develop more complex MatchStrategy rules in the future, such as "firstOf"-first condition must matched, or "mostOf"- most of the conditions must be matched, etc.

Interested readers can read the source code of 'shenyu-plugin' to learn more.

Apache Shenyu has been identified as a gateway application which supports a variety of protocols and microservice frameworks such as Dubbo, gRPC, Spring-Cloud, etc. To do this, the product has accomplished an elegant SPI (Service Provider Interface) as its foundation, and make the Rule data parsing and predicting program very simple , resiliency and security. As to rule data parsing processing, the SPI design increases the product's scalability. When appending new plugin, in most cases, the existing module is enough for rule data parsing , otherwise it can be rapidly carry out with tiny effort.

Top level design of SPI#

In Apache Shenyu, the SPI archtecure is defined in shenyu-spi module and composed of three parts: SPI interface, factory design pattern, and configuration file. There is two interface defined as annotation: @SPI and @Join. When class file with @Join annotation, it means that it will join as an SPI extension class, in other words, it is an application or registration. The @SPI denotes that the class is an SPI extension class.

Fig 1 classes in the shenyu-spi

The SPI configuration directory is META-INF/shenyu/. that is specified:

SHENYU_DIRECTORY = "META-INF/shenyu/";

When starting the gateway system , the ExtensionLoader will scan the profiles under SHENYU_DIRECTORY, in turn, load and validate and then initialize each configed class. The configuration file uses "Key = class-file" format. During operation of the system, the corresponding SPI implementation class will be invoked through the factory mechanism.

Implementation of shenyu-plugin SPI#

In shenyu-plugin module, various plugins for HTTP routing are implemented according to the plugin mechanism, including request, redirect, response and rewrite, etc. Plugins for microservice frameworks such as Dubbo, gRPC , Spring-Cloud and Tars have been developed in the gateway product. And plugins are still increasing. If no such dependent module fo parsing and judge routing parameters and data, each plugin is necessary to implement the parsing functions, and has to frequently modify to support their matching rules, such as wildcard, regular expression, SpEL expression, etc. Therefore, they made a high level abstraction for routing parameter data following the SPI framework in shenyu-plugin module. The rule analysis consists of three parts:

• ParameterData- parameter data

• PredicatJudge- predicate whether the actural data match the rule

• MatchStrategy- combine multiple conditions, the final used strategy

These implementation classes are defined in shenyu-plugin-base module. In each plugin, resolution and predication of the routing parameter can be realized through AbstractShenyuPlugin using the above SPIs. That is dedicated and easy to extend, in line with SOLID principle.

​ This section analyzes the PredictJudge in detail. You can find the dependency to shenyu-spi in the pom.xml of this module.

<dependency>    <groupId>org.apache.shenyu</groupId>    <artifactId>shenyu-spi</artifactId>    <version>${project.version}</version></dependency>

Design of PredicateJudge SPI#

PredicateJudge SPI is used to analyze and judge various routing rules configed in Apache Shenyu gateway. The name and functions of this SPI are similar to Predicate in Java, but the acceptance behavior is further abstracted applying for routing aspect. This SPI is implemented through the Factory pattern. Let's look at the Predictejudge SPI interface:

@SPI@FunctionalInterfacepublic interface PredicateJudge {
    /**     * judge conditionData and realData is match.     *     * @param conditionData {@linkplain ConditionData}     * @param realData       realData     * @return true is pass  false is not pass.     */    Boolean judge(ConditionData conditionData, String realData);}

The class diagram is as follows:

Fig 2-Predicate class diagram

The important methods of PredicateJudgeFactory are shown as follows:

Whenever need to parsing and matching routing data, you can use

    public static PredicateJudge newInstance(final String operator) {        return ExtensionLoader.getExtensionLoader(PredicateJudge.class).getJoin(processSpecialOperator(operator));    }

    public static Boolean judge(final ConditionData conditionData, final String realData) {        if (Objects.isNull(conditionData) || StringUtils.isBlank(realData)) {            return false;        }        return newInstance(conditionData.getOperator()).judge(conditionData, realData);    }

ConditionData contains of four attributes of String type: paramType, operator,paramName,paramValue

ParamTypeEnum#

Where paramType must be the enumeration type ParamTypeEnum. The default supported paramType are:

post, uri,query, host, ip,header, cookie,req_method

OperatorEnum#

operator must be the enumeration type OperatorEnum, currently supported operators are:

   match, =,regex, >,<, contains, SpEL,  Groovy, TimeBefore,TimeAfter

Base on the above defination , the plugin module provides the following eight PredicateJudge implemetation classes to realize the logic of these operators respectively.

How to use PredicateJudge#

When you want to parse parameters, you only need to call PredicateJudgeFactory as follows.

PredicateJudgeFactory.judge(final ConditionData conditionData, final String realData);

SPI profile#

The implementation class is configed in the file under directory SHENYU_DIRECTORY . It will be loaded and cached at startup.

equals=org.apache.shenyu.plugin.base.condition.judge.EqualsPredicateJudge
contains=org.apache.shenyu.plugin.base.condition.judge.ContainsPredicateJudgeGroovy=org.apache.shenyu.plugin.base.condition.judge.GroovyPredicateJudgematch=org.apache.shenyu.plugin.base.condition.judge.MatchPredicateJudgeregex=org.apache.shenyu.plugin.base.condition.judge.RegexPredicateJudgeSpEL=org.apache.shenyu.plugin.base.condition.judge.SpELPredicateJudgeTimeAfter=org.apache.shenyu.plugin.base.condition.judge.TimerAfterPredicateJudgeTimeBefore=org.apache.shenyu.plugin.base.condition.judge.TimerBeforePredicateJudge

The usage of PredicateJudge SPI in Shenyu gateway#

Most plugins in Apache Shenyu are inherited from AbstractShenyuPlugin. In this abstract class, the filter functions (selection and matching) are achieved through MatchStrategy SPI, and PredicateJudge will be invoked from MatchStrategy to predicate each condition data.

Fig 3- class diagram of plugins with PredicateJudge and MatchStrategy SPI

The process from client request calling the routing parsing moodule is showed as following chart.

Fig 4- flow chart for Shenyu gateway filter with parameter processing

• When startup, the system will load SPI classes from profile and cache them.
• When the client sends a new request to the Shenyu gateway, will call the corresponding plugin within the gateway.
• When analyzing real data with routing rules, the PredicateJudge implementation class will be invoked according to the contained operator.

Others#

Examples of PredicateJudge judgement#

ContainsPredicateJudge- " contains“ rule#

For example, giving a ConditionData with: paramType="uri", paramValue 是 "/http/**", when using the "contains" relation: ContainsPredicateJudge, the matching result is as follows.

About other PredicateJudge implemetantion classes, you can refer to the code and test classes.

Rate limiter is a very important integral of gateway application, to deal with high traffic. When the system is attacked abnormally by a large number of traffic gathered in a short time; When there are a large number of lower priority request need to be slow down or else it will effect your high priority transactions; Or sometimes your system can not afford the regular traffic; in these scenarios, we need to start rate limiter component to protect our system, through rejection, wait, load shedding,etc, limit the requests to an acceptable quantities, or only certain domains (or services) requests can get through.

Facing above scenarios, following need to be considered when designing the rate limiter component of an gateway.

1. Supports a variety of rate limiter algorithms and easy to extends.
2. Resilient resolvers which can distinguish traffic by different way, such as ip, url, even user group etc.
3. High availability, can quickly get allow or reject result from rate limiter
4. With fault tolerance against when rate limiter is down, the gateway can continue work.

This article will first introduce the overall architecture of the rate limiter module in Apache Shenyu, and then focus on the code analysis of rate limiter SPI.

This article based on shenyu-2.4.0 version of the source code analysis.

Overall design of RateLimiter#

Spring WebFlux is reactive and non-blocking web framework, which can benefit throughput and make applications more resilient. The plugin of Apache Shenyu is based on WebFlux，its rate limiter component is implemented in ratelimiter-plugin. In rate limiter process, the commonly used algorithms are token bucket, leaky bucket, etc. To speed up concurrency performance, the counting and calculation logic is treated in Redis, and Java code is responsible for the transmission of parameters. When applying Redis, the Lua script can be resident memory, and be executed as a whole, so it is atomic. Let alone the reducing of network overhead. Redis commands abstraction and automatic serialization/deserialization with Redis store is provided in Spring Data Redis. Because of based on reactive framework, the Spring Redis Reactive is used in ratelimiter-plugin.

The class diagram of this plugin is as follows, highlighting two packages related to RateLimiter SPI: resolver 和algorithm.

Design of RateLimiter SPI#

High performance issue is achieved through the architecture of Spring data+ Redis+Lua , two SPI are supplied in ratelimiter-plugin for the extension of algorithm and key resolver。

• RateLimiterAlgorithm：used for algorithms expansion.
• RateLimiterKeyResolver： used for resolver expansion, to distinguish requests by various information, including ip, url, ect.

The profile of SPI is located at directory of SHENYU_DIRECTORY (default/META-INF/shenyu).

RateLimiterKeyResolver#

Obtain the critical info of the request used for packet rate limiter，the interface of RateLimiterKeyResolver is follows：

@SPIpublic interface RateLimiterKeyResolver {
    /**     * get Key resolver's name.     *     * @return Key resolver's name     */    String getKeyResolverName();
    /**     * resolve.     *     * @param exchange exchange the current server exchange {@linkplain ServerWebExchange}     * @return rate limiter key     */    String resolve(ServerWebExchange exchange);}

@SPI registers the current interface as Apache Shenyu SPI. Method resolve(ServerWebExchange exchange) is used to provide the resolution way. Currently there are two key resolvers in RateLimiterKeyResolver SPI:WholeKeyResolve and RemoteAddrKeyResolver. The resolve method of RemoteAddrKeyResolveris as follows：

    @Override    public String resolve(final ServerWebExchange exchange) {        return Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();    }

Where the resolved key is ip of request. Based on SPI mechanism and its factory pattern, new resolver can be easily developed.

RateLimiterAlgorithm SPI#

RateLimiterAlgorithm SPI is used to identify and define different rate limiter algorithms, following is the class diagram of this module.

In this module, factory pattern is used , providing interface, abstract class and factory class, and four implementation classes. The lua script corresponding to the implementation class is enumerated in RateLimitEnum and located in /META-INF/scripts.

@SPIpublic interface RateLimiterAlgorithm<T> {        RedisScript<T> getScript();    List<String> getKeys(String id);        /**     * Callback string.     *     * @param script the script     * @param keys the keys     * @param scriptArgs the script args     */    default void callback(final RedisScript<?> script, final List<String> keys, final List<String> scriptArgs) {    }}

@SPI registers the current interface as Apache Shenyu SPI. There are three methods:

• getScript() returns a RedisScript object, which will be passed to Redis.
• getKeys(String id) returns a List contains with keys.
• callback() the callback function which will be executed asynchronously later on, and default is an empty method.

AbstractRateLimiterAlgorithm#

The template method is implemented in this abstract class, and the reified generics used is List<Long>. Two abstract methods getScriptName() and getKeyName() are left for the implementation class. Following is the code to load lua script.

    public RedisScript<List<Long>> getScript() {        if (!this.initialized.get()) {            DefaultRedisScript redisScript = new DefaultRedisScript<>();            String scriptPath = "/META-INF/scripts/" + getScriptName();            redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(scriptPath)));            redisScript.setResultType(List.class);            this.script = redisScript;            initialized.compareAndSet(false, true);            return redisScript;        }        return script;    }

initialized is an AtomicBoolean type variable used to indicate whether the lua script is loaded. If has not been loaded, the system will read specified scripts form META-INF/scripts; After reading, specify the result with List type, and set initialized=true, then returning RedisScriptobject.

The code of getKeys() in AbstractRateLimiterAlgorithm is as follows:

    @Override    public List<String> getKeys(final String id) {        String prefix = getKeyName() + ".{" + id;        String tokenKey = prefix + "}.tokens";        String timestampKey = prefix + "}.timestamp";        return Arrays.asList(tokenKey, timestampKey);    }

Two strings are generated in this template method, where the tokenKey will work as Key to a Sorted map in Redis.

We can observe from above class diagram that ConcurrentRateLimiterAlgorithm and SlidingWindowRateLimiterAlgorithm override getKeys(String id) method but another two implementation classes not, and use template method in AbstractRateLimiterAlgorithm. Only in ConcurrentRateLimiterAlgorithm has override callback() method, the others not. We will do further analysis in the following.

RateLimiterAlgorithmFactory#

The method getsRateLimiterAlgorithm instance by name in RateLimiterAlgorithmFactory is as follows:

public static RateLimiterAlgorithm<?> newInstance(final String name) {    return Optional.ofNullable(ExtensionLoader.getExtensionLoader(RateLimiterAlgorithm.class).getJoin(name)).orElse(new TokenBucketRateLimiterAlgorithm());}

ExtensionLoader of SPI is responsible for loading SPI classes by "name", if cannot find the specified algorithm class, it will return TokenBucketRateLimiterAlgorithm by default.

Data access with Redis#

Above detailed the extension interface in RateLimiter SPI. In Apache Shenyu, we use ReactiveRedisTemplate to perform Redis processing reactively, which is implemented inisAllowed() method of RedisRateLimiter class.

    public Mono<RateLimiterResponse> isAllowed(final String id, final RateLimiterHandle limiterHandle) {        // get parameters that will pass to redis from RateLimiterHandle Object        double replenishRate = limiterHandle.getReplenishRate();        double burstCapacity = limiterHandle.getBurstCapacity();        double requestCount = limiterHandle.getRequestCount();        // get the current used RateLimiterAlgorithm        RateLimiterAlgorithm<?> rateLimiterAlgorithm = RateLimiterAlgorithmFactory.newInstance(limiterHandle.getAlgorithmName());                ........        Flux<List<Long>> resultFlux = Singleton.INST.get(ReactiveRedisTemplate.class).execute(script, keys, scriptArgs);        return resultFlux.onErrorResume(throwable -> Flux.just(Arrays.asList(1L, -1L)))                .reduce(new ArrayList<Long>(), (longs, l) -> {                    longs.addAll(l);                    return longs;                }).map(results -> {                    boolean allowed = results.get(0) == 1L;                    Long tokensLeft = results.get(1);                    return new RateLimiterResponse(allowed, tokensLeft);                })                .doOnError(throwable -> log.error("Error occurred while judging if user is allowed by RedisRateLimiter:{}", throwable.getMessage()))                .doFinally(signalType -> rateLimiterAlgorithm.callback(script, keys, scriptArgs));    }

The POJO class RateLimiterHandle wraps the parameters needed in rate limiter, they are algorithName, replenishRate, burstCapacity, requestCount, etc. First, gets the parameters that need to be passed into Redis from RateLimiterHandle class. Then obtain the current implementation class from RateLimiterAlgorithmFactory.

For convenience, we give an flow image to show the parameters I/O and execution procedure in Java and Redis respectively. On the left is the second half of isAllowed() , and on the right is the processing of Lua script.

Following is the execution process of the JAVA code.

1. Get two keys value in List<String> type from the getKeys() method, the first element will map to a sorted set in Redis.

2. Set four parameters, replenishRate , burstCapacity, timestamp (EpochSecond) and requestcount.

3. Calling ReactiveRedisTemplate with the scripts, keys and parameters, the return a Flux<List<Long>>

4. The return value is converted from Flux<ArrayList<Long>> to Mono<ArrayList<Long>> the through reduce() of Flux ，and then transform it to Mono<RateLimiterResponse> via map() function. Returned two data, one is allowed (1-allow, 0- not allowed), the other is tokensLeft, the number of available remaining request.

5. As for the fault tolerance, due to using of reactor non-blocking model, when an error occurs, the fallback function onErrorResume() will be executed and a new stream (1L, -1L) will generated by Flux.just, which means allow the request getting through, and log the error on the side.

6. After that, performs the doFinally() method, that is to execute the callback() method of the implementation class.

Four rate limiter algorithms#

From above we know that how the java code works with Redis in the gateway. In this chapter we briefly analysis some code of the four rate limiter algorithms, to understand how to develop the interface of RateLimiter SPI and work efficiently with Redis.

Four rate limiter algorithms are supplied in Apache Shenyu Ratelimit SPI:

1. Token bucket rate limiter： Limiting the traffic according to the number of requests. Assuming that N requests can be passed per second, when requests exceeding N will be rejected. In implementing of the algorithm, the requests will be grouped by bucket, the tokens will be generated at an evenly rate. If the number of requests is less than the tokens in the bucket, then it is allowed to pass. The time window is 2* capacity/rate.
2. Slide window rate limiter: Different from token bucket algorithm, its window size is smaller than that of token bucket rate limiter, which is a capacity/rate. And move backward one time window at a time. Other rate limiter principles are similar to token bucket.
3. Concurrent rate limiter： Strictly limit the concurrent requests to N. Each time when there is a new request, it will check whether the number of concurrent requests is greater than N. If it is less than N, it is allowed to pass through, and the count is increased by 1. When the requests call ends, the signal is released (count minus 1).
4. Leaky bucket rate limiter: In contrast with token bucket algorithm, the leaky bucket algorithm can help to smooths the burst of requests and only allows a pre-defined N number of requests. This limiter can force the output flow at a constant rate of N. It is based on a leaky bucket model, the leaky water quantity is time interval*rate. if the leaky water quantity is greater than the number of has used (represented by key_bucket_count), then clear the bucket, that is, set the key_bucket_count to 0. Otherwise, set key_bucket_count minus the leaky water quantity. If the number (requests + key_bucket_count ) is less than the capacity, then allow the requests passing through.

Let's understand the functionality of callback() by reading concurrent rate limiter code, and understand the usage of getKeys() through reading the Lua script of token rate limiter and slide window rate limiter.

callback() used in Concurrent requests limiter#

The getKeys() method of ConcurrentRateLimiterAlgorithm overrides the template method in AbstractRateLimiterAlgorithm ：

    @Override    public List<String> getKeys(final String id) {        String tokenKey = getKeyName() + ".{" + id + "}.tokens";        String requestKey = UUIDUtils.getInstance().generateShortUuid();        return Arrays.asList(tokenKey, requestKey);    }

The second element, requestKey is a long type and non-duplicate value (generated by a distributed ID generator，it is incremented and smaller than the current time Epochsecond value). The corresponding Lua script in concurrent_request_rate_limiter.lua:

local key = KEYS[1]
local capacity = tonumber(ARGV[2])local timestamp = tonumber(ARGV[3])local id = KEYS[2]

Here id is requestKey generated by getKeys() method, it is an uuid(unique value). Subsequent process is as follows:

local count = redis.call("zcard", key)local allowed = 0
if count < capacity then  redis.call("zadd", key, timestamp, id)  allowed = 1  count = count + 1endreturn { allowed, count }

First, using zcard command to obtain the cardinality of the sorted set, and set count equals the cardinality , if the cardinality is less than the capacity, we will add a new member id (it is an uuid) to the sorted set, with the score of current time(in seconds) . then count =count+1, the cardinality is also incremented by 1 in reality.

All of the code above is executed in Redis as an atomic transaction. If there are a large number of concurrent requests from the same key( such as ip) , the cardinality of the sorted set of this key will increasing sharply, when then capacity limit is exceeded, the service will be denied, that is allowed =0。

In concurrent requests limiter, It is required to release the semaphore when the request is completed. However, it is not included in Lua script.

Let's see the callback function of ConcurrentRateLimiterAlgorithm：

    @Override    @SuppressWarnings("unchecked")    public void callback(final RedisScript<?> script, final List<String> keys, final List<String> scriptArgs) {        Singleton.INST.get(ReactiveRedisTemplate.class).opsForZSet().remove(keys.get(0), keys.get(1)).subscribe();    }

Here gives asynchronous subscription, using ReactiveRedisTemplate to delete the elements (key,id) in Redis store. That is once the request operation ends, the semaphore will be released. This remove operation cannot be executed in Lua script. This is just what design intention of callback in RateLimiterAlgorithm SPI .

getKeys() used in token bucket rate limiter#

Following is the corresponding Lua code:

local tokens_key = KEYS[1]local timestamp_key = KEYS[2]

Here we omit the code that get the parameters of rate ,capacity, etc.

local fill_time = capacity/ratelocal ttl = math.floor(fill_time*2)

The window size variable(ttl) is approximately two times of capacity/rate.

local last_tokens = tonumber(redis.call("get", tokens_key))if last_tokens == nil then  last_tokens = capacityend

Get last_tokens from the sorted set, if it not exist, then last_tokens equals capacity.

local last_refreshed = tonumber(redis.call("get", timestamp_key))if last_refreshed == nil then  last_refreshed = 0end

Get the last refreshed time by the key =timestamp_key from the sorted set, and default 0.

local delta = math.max(0, now-last_refreshed)local filled_tokens = math.min(capacity, last_tokens+(delta*rate))local allowed = filled_tokens >= requestedlocal allowed_num = 0if allowed then  new_tokens = filled_tokens - requested  allowed_num = 1end

The filled_tokens is produced evenly by time interval * rate，if the number of tokens greater than requests, then allowed=1, and update new_tokens.

redis.call("setex", tokens_key, ttl, new_tokens)redis.call("setex", timestamp_key, ttl, now)
return { allowed_num, new_tokens }

Here now is current time parameters passed in, set tokens_key to hold the string new_tokens and settokens_key to timeout after ttl of seconds. Set timestamp_key to hold the string value now, and expires after ttl seconds.

getKeys() used in sliding window rate limiter#

The getKeys() in SlidingWindowRateLimiterAlgorithm also overrides the parent class, and the code is consistent with the method in ConcurrentRateLimiterAlgorithm

Following is the Lua code of slide window rate limiter, the receiving of other parameters is omitted.

local timestamp_key = KEYS[2]...... local window_size = tonumber(capacity / rate)local window_time = 1

Here set the window_size to capacity/rate.

local last_requested = 0local exists_key = redis.call('exists', tokens_key)if (exists_key == 1) then    last_requested = redis.call('zcard', tokens_key)end

Obtain the cardinality(last_requested) of the tokens_key in the sorted set.

local remain_request = capacity - last_requestedlocal allowed_num = 0if (last_requested < capacity) then    allowed_num = 1    redis.call('zadd', tokens_key, now, timestamp_key)end

Calculate remaining available remain_request equals capacity minus last_requested . If last_requested less than capacity ,then allow current requests passing through，add element in the sorted set with (key=timestamp_key, value=now) .

redis.call('zremrangebyscore', tokens_key, 0, now - window_size / window_time)redis.call('expire', tokens_key, window_size)
return { allowed_num, remain_request }

Previously has set window_time=1, using zremrangebyscore command of Redis to remove all the elements in the sorted set stored at tokens_key with a score in [0,now - window_size / window_time] , that is, move the window a window size. Set the expire time of tokens_key to window_size.

In the template method getKeys(final String id) of AbstractRateLimiterAlgorithm，the second key ( represented y secondKey) is a fixed string which concat the input parameter{id}. As we can see from the above three algorithm codes, in the token bucket algorithm, secondKey will be updated to the latest time in the Lua code, so it doesn't matter what value is passed in. In the concurrent rate limiter, secondKey will be used as the key to remove Redis data in the java callback method. In the sliding window algorithm, the secondKey will be added to the sorted set as the key of a new element, and will be removed during window sliding.

That's all, when in a new rate limiter algorithm, the getKeys(final String id)method should be carefully designed according to the logic of the algorithm.

How to use RateLimiter SPI#

The three parameters in doExecute() method of RateLimiter plugin， exchange is an web request， chain is the execution chain of the plugins，selector is the selection parameters，rule is the policies or rules of rate limiter setting in the system.

protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {    //get  the `RateLimiterHandle` parameters from cache     RateLimiterHandle limiterHandle = RatelimiterRuleHandleCache.getInstance()        .obtainHandle(CacheKeyUtils.INST.getKey(rule));    //find the resolver name     String resolverKey = Optional.ofNullable(limiterHandle.getKeyResolverName())        .flatMap(name -> Optional.of("-" + RateLimiterKeyResolverFactory.newInstance(name).resolve(exchange)))        .orElse("");    return redisRateLimiter.isAllowed(rule.getId() + resolverKey, limiterHandle)        .flatMap(response -> {            if (!response.isAllowed()) {                exchange.getResponse().setStatusCode(HttpStatus.TOO_MANY_REQUESTS);                Object error = ShenyuResultWrap.error(ShenyuResultEnum.TOO_MANY_REQUESTS.getCode(), ShenyuResultEnum.TOO_MANY_REQUESTS.getMsg(), null);                return WebFluxResultUtils.result(exchange, error);            }            return chain.execute(exchange);        });}

1. Firstly get the RateLimiterHandle parameters from cache.

2. Obtains the corresponding Key resolver by RateLimiterHandle instance.

3. Reactively executes isAllowed() method of RedisRateLimiter.

4. If not allowed, error handling is performed.

5. If the request is allowed, dispatch it to the next process of execution chain.

Summary#

RateLimiter plugin is based on Spring WebFlux,and with Apache Shen SPI, with Redis and Lua script to responsible for the critical algorithm and logic process, make it with characteristics of high concurrency and elastic. As for the RateLimiter SPI.

1. RateLimiter SPI provides two SPI interface, with interface oriented design and various design patterns, it's easy to develop new rate limiter algorithm and key resolver rule.
2. RateLimiterAlgorithm SPI supplies four rate limiter algorithms, token bucket,concurrency rate limiter, leaky bucket and sliding window rate limiter. When designing rate limiter algorithm, the KEY generation need to be carefully designed according to the algorithm characteristic. Using Lua script to realize the logic of the algorithm, and design callback() method for asynchronous processing when needed.
3. Reactive programming, simple and efficient implementation.

Preface#

As a first-time developer in the Shenyu community, I encountered some "Pitfalls" that were not mentioned in the tutorials I followed to start and develop the project. I have documented the detailed steps I took to start shenyu, shenyu-dashboard, shenyu-website in this blog, hoping to help more new contributors in the community.

Environmental Preparation#

• Correct local installation of JDK1.8+
• Properly install Git locally
• Choose a development tool, this article uses IDEA as an example

ShenYu Backend Startup Guide#

Install and Configure Maven#

Maven is a cross-platform project management tool . As the Apache organization's top open source projects , its main service for Java-based platform project creation , dependency management and project information management.

1. Download maven and extract it to a path with no Chinese and no spaces.

   

2. Add the bin directory under the maven directory to the environment variables. For Windows, if the download directory is E:\apache-maven-3.9.1, add E:\apache-maven-3.9.1\bin to the Path system variable.

3. Verify that the installation was successful. Type mvn -v in the cmd window, and if the Maven version and Java version appear, the installation is successful. This is shown below:

   C:\Users\pc>mvn -vApache Maven 3.9.1 (2e178502fcdbffc201671fb2537d0cb4b4cc58f8)Maven home: E:\apache-maven-3.9.1Java version: 18.0.1.1, vendor: Oracle Corporation, runtime: C:\Program Files\Java\jdk-18.0.1.1Default locale: zh_CN, platform encoding: UTF-8OS name: "windows 10", version: "10.0", arch: "amd64", family: "windows"

   Copy

4. To speed up the download of project-related dependencies, you need to change the Maven mirrors, here add Aliyun and other mirrors. Change the <mirrors> </mirrors> tag pair in conf/settings.xml to the following:

   <mirrors>    <mirror>    <id>alimaven</id>    <name>aliyun maven</name>    <url>http://maven.aliyun.com/nexus/content/groups/public/</url>    <mirrorOf>central</mirrorOf>    </mirror>
       <mirror>    <id>alimaven</id>    <mirrorOf>central</mirrorOf>    <name>aliyun maven</name>    <url>http://maven.aliyun.com/nexus/content/repositories/central/</url>    </mirror>
       <mirror>    <id>maven</id>    <mirrorOf>central</mirrorOf>    <name>name_</name>    <url>http://repo1.maven.org/maven2</url>    </mirror> 
       <mirror>    <id>junit</id>    <mirrorOf>central</mirrorOf>    <name>junit address/</name>    <url>http://jcenter.bintray.com/</url>    </mirror></mirrors>

   Copy

   and add <localRepository>E:/maven_local_repository</localRepository> to the next line of </mirrors> to set the location of Maven local repository. You can specify the exact location yourself.

Pull ShenYu Code#

1. Fork ShenYu repository on Github to your own repository, where you can develop and commit PRs in the future

2. Use Git to download the repository from the previous step locally:

   git clone git@github.com:${YOUR_USERNAME}/${TARGET_REPO}.git

   Copy

   If prompted for a long file name, execute the following command via the command line:

   git config --global core.longpaths true

   Copy

ShenYu First Start#

Preparation#

1. Compile with Maven in the shenyu directory:

   mvn clean install -Dmaven.javadoc.skip=true -B -Drat.skip=true -Djacoco.skip=true -DskipITs -DskipTests

   Copy

2. Configure IDEA environment. Open shenyu project with IDEA, click File -> Settings in the top left corner, and configure Maven as shown below. Where User settings file select your settings.xml directory, and then Local repository will automatically load the localRepository path set in settings.xml:

   

3. At this point, IDEA will automatically download the project-related dependencies, you need to wait for a while, when finished, as shown in the following figure:

   

   As you can see, shenyu-e2e, shenyu-examples, shenyu-integrated-test are not marked as Maven projects by IDEA and need to be added manually. Select the pom.xml file in the package and right-click Add as Maven Project. If the shenyu-e2e build fails, then add the <relativePath>. /pom.xml</relativePath> to <relativePath/>.

Start Gateway Service#

1. Start the shenyu-admin console (H2 database is used by default)

   

2. start shenyu-bootstrap

   

By this point, the shenyu gateway has been started.

We can open the browser and access the admin console: http://localhost:9095/

Default account: admin , default password: 123456

Start Application Service#

Apache ShenYu provides samples of Http, Dubbo, SpringCloud and other applications to access the shenyu gateway, located in the shenyu-example module, here the Http service is used as an example.

Start shenyu-examples-http。

At this point, shenyu-examples-http will automatically register the interface methods annotated with @ShenyuSpringMvcClient and the relevant configuration in application.yml to the gateway. We can open the admin console and see the configuration in Client List -> Proxy -> divide.

Test Http Request#

The following uses the IDEA HTTP Client Plugin to mock http to access http services.

• Local access without using shenyu proxy

  

• Use shenyu proxy

  

Use more plugins#

We can refer to the official documentation to the left of Plugins collection to use the required plugins.

Shenyu Front End Startup Guide#

Install Node.js#

Download#

1. Download and install Node.js from official website and select LTS version.

2. When installing, except for setting the installation path, just keep clicking Next.

3. After the installation is complete, verify at the command line:

   C:\Users\pc>node -vv12.22.12
   C:\Users\pc>npm -v6.14.16

   Copy

Pull ShenYu Dashboard Code#

1. Fork ShenYu Dashboard repository

2. Using Git to download locally

   git clone git@github.com:${YOUR_USERNAME}/${TARGET_REPO}.git

   Copy

Front and Back End Co-development#

1. Add enablePrintApiLog: true to the shenyu-admin/src/main/resources/application.yml file in the backend repository shenyu as shown below to show the log of frontend interface calls in the backend console.

   

2. Start ShenyuAdminBootstrap

3. Switch to the front-end repository shenyu-dashboard, open README, click npm install, npm start or enter the above command from cmd to access the front-end interface via http://localhost:8000, and display the log of the front-end interface called in the back-end console. Realize the co-development of front-end and back-end.

   

Package Front-end Code#

Execute the npm build command in README and copy all the generated files from the dist folder to the shenyu-admin/src/main/resources/static/ directory in the backend repository.

Contribute to Shenyu Official Website#

Just follow the README in shenyu-website.

Tips#

1. I recommend downloading the LTS version from the Node website.
2. Windows systems cannot be deployed, if you want to verify your changes, you can deploy on a Linux virtual machine or server.

Environmental preparation#

• Install JDK1.8+ locally
• Install Git locally
• Install Maven locally
• Choose a development tool, such as IDEA

Pull ShenYu code#

Use Git to clone code

git clone https://github.com/apache/shenyu.git

Compile code#

Compile with Maven

cd shenyumvn clean install -Dmaven.javadoc.skip=true -B -Drat.skip=true -Djacoco.skip=true -DskipITs -DskipTests

Start the gateway service#

Use development tools, take IDEA as an example.

Start shenyu-admin (use H2 database by default)

Start shenyu-bootstrap

At this point, shenyu gateway has been activated.

We can open the browser and access the admin console: http://localhost:9095/

Start application service#

Apache ShenYu provides examples for Http, Dubbo, SpringCloud and other applications to access the shenyu gateway, located in the shenyu-example module. Here we take the Http service as an example.

If shenyu-example is not marked as a Maven project by IDEA, you can right-click the pom.xml file in the shenyu-example directory to add it as a Maven project.

Start shenyu-examples-http。

At this time, shenyu-examples-http will automatically register the interface method annotated with @ShenyuSpringMvcClient and the related configuration in application.yml to the gateway. When we open the admin console, you can see the relevant configuration in divide and context-path.

Test Http request#

Now use postman to simulate http to request your http service:

Use more plugins#

We can refer to Official Document to use other plugins.

Here is an example of using the param-mapping plugin.

Edit the param-mapping plugin in BasicConfig -> Plugin and set status.

Configure selectors and rules in PluginList -> http process.

Then use postman to make an http request to /http/test/payment.

Apache ShenYu is an asynchronous, high-performance, cross-language, responsive API gateway.

In ShenYu gateway, data synchronization refers to how to synchronize the updated data to the gateway after the data is sent in the background management system. The Apache ShenYu gateway currently supports data synchronization for ZooKeeper, WebSocket, http long poll, Nacos, etcd and Consul. The main content of this article is based on WebSocket data synchronization source code analysis.

This paper based on shenyu-2.4.0 version of the source code analysis, the official website of the introduction of please refer to the Data Synchronization Design .

1. About ZooKeeper#

Apache ZooKeeper is a software project of the Apache Software Foundation that provides open source distributed configuration services, synchronization services, and naming registries for large-scale distributed computing. ZooKeeper nodes store their data in a hierarchical namespace, much like a file system or a prefix tree structure. Clients can read and write on nodes and thus have a shared configuration service in this way.

2. Admin Data Sync#

We traced the source code from a real case, such as updating a selector data in the Divide plugin to a weight of 90 in a background administration system:

2.1 Accept Data#

• SelectorController.createSelector()

Enter the createSelector() method of the SelectorController class, which validates data, adds or updates data, and returns results.

@Validated@RequiredArgsConstructor@RestController@RequestMapping("/selector")public class SelectorController {        @PutMapping("/{id}")    public ShenyuAdminResult updateSelector(@PathVariable("id") final String id, @Valid @RequestBody final SelectorDTO selectorDTO) {        // set the current selector data ID        selectorDTO.setId(id);        // create or update operation        Integer updateCount = selectorService.createOrUpdate(selectorDTO);        // return result         return ShenyuAdminResult.success(ShenyuResultMessage.UPDATE_SUCCESS, updateCount);    }        // ......}

2.2 Handle Data#

• SelectorServiceImpl.createOrUpdate()

Convert data in the SelectorServiceImpl class using the createOrUpdate() method, save it to the database, publish the event, update upstream.

@RequiredArgsConstructor@Servicepublic class SelectorServiceImpl implements SelectorService {    // eventPublisher    private final ApplicationEventPublisher eventPublisher;        @Override    @Transactional(rollbackFor = Exception.class)    public int createOrUpdate(final SelectorDTO selectorDTO) {        int selectorCount;        // build data DTO --> DO        SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);        List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();        // insert or update ?        if (StringUtils.isEmpty(selectorDTO.getId())) {            //  insert into data            selectorCount = selectorMapper.insertSelective(selectorDO);            // insert into condition data            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));            });            // check selector add            if (dataPermissionMapper.listByUserId(JwtUtils.getUserInfo().getUserId()).size() > 0) {                DataPermissionDTO dataPermissionDTO = new DataPermissionDTO();                dataPermissionDTO.setUserId(JwtUtils.getUserInfo().getUserId());                dataPermissionDTO.setDataId(selectorDO.getId());                dataPermissionDTO.setDataType(AdminConstants.SELECTOR_DATA_TYPE);                dataPermissionMapper.insertSelective(DataPermissionDO.buildPermissionDO(dataPermissionDTO));            }
        } else {            // update data, delete and then insert            selectorCount = selectorMapper.updateSelective(selectorDO);            //delete rule condition then add            selectorConditionMapper.deleteByQuery(new SelectorConditionQuery(selectorDO.getId()));            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                SelectorConditionDO selectorConditionDO = SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO);                selectorConditionMapper.insertSelective(selectorConditionDO);            });        }        // publish event        publishEvent(selectorDO, selectorConditionDTOs);
        // update upstream        updateDivideUpstream(selectorDO);        return selectorCount;    }        // ......    }

In the Service class to persist data, i.e. to the database, this should be familiar, not expand. The update upstream operation is analyzed in the corresponding section below, focusing on the publish event operation, which performs data synchronization.

The logic of the publishEvent() method is to find the plugin corresponding to the selector, build the conditional data, and publish the change data.

       private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {        // find plugin of selector        PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());        // build condition data        List<ConditionData> conditionDataList =                selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());        // publish event        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,                Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));    }

Change data released by eventPublisher.PublishEvent() is complete, the eventPublisher object is a ApplicationEventPublisher class, The fully qualified class name is org.springframework.context.ApplicationEventPublisher. Here we see that publishing data is done through Spring related functionality.

ApplicationEventPublisher：

When a state change, the publisher calls ApplicationEventPublisher of publishEvent method to release an event, Spring container broadcast event for all observers, The observer's onApplicationEvent method is called to pass the event object to the observer. There are two ways to call publishEvent method, one is to implement the interface by the container injection ApplicationEventPublisher object and then call the method, the other is a direct call container, the method of two methods of publishing events not too big difference.

• ApplicationEventPublisher: publish event;
• ApplicationEvent: Spring event, record the event source, time, and data;
• ApplicationListener: event listener, observer.

In Spring event publishing mechanism, there are three objects,

An object is a publish event ApplicationEventPublisher, in ShenYu through the constructor in the injected a eventPublisher.

The other object is ApplicationEvent , inherited from ShenYu through DataChangedEvent, representing the event object.

public class DataChangedEvent extends ApplicationEvent {//......}

The last object is ApplicationListener in ShenYu in through DataChangedEventDispatcher class implements this interface, as the event listener, responsible for handling the event object.

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
    //......    }

2.3 Dispatch Data#

• DataChangedEventDispatcher.onApplicationEvent()

Released when the event is completed, will automatically enter the DataChangedEventDispatcher class onApplicationEvent() method of handling events.

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
  /**     * This method is called when there are data changes   * @param event     */    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // Iterate through the data change listener (usually using a data synchronization approach is fine)      for (DataChangedListener listener : listeners) {            // What kind of data has changed        switch (event.getGroupKey()) {                case APP_AUTH: // app auth data                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());                    break;                case PLUGIN:  // plugin data                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());                    break;                case RULE:    // rule data                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());                    break;                case SELECTOR:   // selector data                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    break;                case META_DATA:  // metadata                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());                    break;                default:  // other types throw exception                  throw new IllegalStateException("Unexpected value: " + event.getGroupKey());            }        }    }    }

When there is a data change, the onApplicationEvent method is called and all the data change listeners are iterated to determine the data type and handed over to the appropriate data listener for processing.

ShenYu groups all the data into five categories: APP_AUTH, PLUGIN, RULE, SELECTOR and META_DATA.

Here the data change listener (DataChangedListener) is an abstraction of the data synchronization policy. Its concrete implementation is:

These implementation classes are the synchronization strategies currently supported by ShenYu:

• WebsocketDataChangedListener: data synchronization based on Websocket;
• ZookeeperDataChangedListener:data synchronization based on Zookeeper;
• ConsulDataChangedListener: data synchronization based on Consul;
• EtcdDataDataChangedListener：data synchronization based on etcd;
• HttpLongPollingDataChangedListener：data synchronization based on http long polling;
• NacosDataChangedListener：data synchronization based on nacos;

Given that there are so many implementation strategies, how do you decide which to use?

Because this paper is based on zookeeper data synchronization source code analysis, so here to ZookeeperDataChangedListener as an example, the analysis of how it is loaded and implemented.

A global search in the source code project shows that its implementation is done in the DataSyncConfiguration class.

/** * Data Sync Configuration * By springboot conditional assembly * The type Data sync configuration. */@Configurationpublic class DataSyncConfiguration {            /**     * zookeeper data sunc     * The type Zookeeper listener.     */    @Configuration    @ConditionalOnProperty(prefix = "shenyu.sync.zookeeper", name = "url")  // The condition property is loaded only when it is met    @Import(ZookeeperConfiguration.class)    static class ZookeeperListener {
        /**         * Config event listener data changed listener.         * @param zkClient the zk client         * @return the data changed listener         */        @Bean        @ConditionalOnMissingBean(ZookeeperDataChangedListener.class)        public DataChangedListener zookeeperDataChangedListener(final ZkClient zkClient) {            return new ZookeeperDataChangedListener(zkClient);        }
        /**         * Zookeeper data init zookeeper data init.         * @param zkClient        the zk client         * @param syncDataService the sync data service         * @return the zookeeper data init         */        @Bean        @ConditionalOnMissingBean(ZookeeperDataInit.class)        public ZookeeperDataInit zookeeperDataInit(final ZkClient zkClient, final SyncDataService syncDataService) {            return new ZookeeperDataInit(zkClient, syncDataService);        }    }        // other code is omitted......}

This configuration class is implemented through the SpringBoot conditional assembly class. The ZookeeperListener class has several annotations:

• @Configuration: Configuration file, application context;

• @ConditionalOnProperty(prefix = "shenyu.sync.zookeeper", name = "url"): attribute condition. The configuration class takes effect only when the condition is met. That is, when we have the following configuration, ZooKeeper is used for data synchronization.

  shenyu:    sync:     zookeeper:          url: localhost:2181          sessionTimeout: 5000          connectionTimeout: 2000

  Copy

• @Import(ZookeeperConfiguration.class)：import ZookeeperConfiguration;

  @EnableConfigurationProperties(ZookeeperProperties.class)  // enable zookeeper properties  public class ZookeeperConfiguration {
    /**     * register zkClient in spring ioc.     * @param zookeeperProp the zookeeper configuration     * @return ZkClient {@linkplain ZkClient}        */      @Bean      @ConditionalOnMissingBean(ZkClient.class)      public ZkClient zkClient(final ZookeeperProperties zookeeperProp) {        return new ZkClient(zookeeperProp.getUrl(), zookeeperProp.getSessionTimeout(), zookeeperProp.getConnectionTimeout()); // 读取zk配置信息，并创建zkClient      }  }

@Data@ConfigurationProperties(prefix = "shenyu.sync.zookeeper") // zookeeper propertiespublic class ZookeeperProperties {
    private String url;
    private Integer sessionTimeout;
    private Integer connectionTimeout;
    private String serializer;}

When we take the initiative to configuration, use the zookeeper data synchronization, zookeeperDataChangedListener is generated. So in the event handler onApplicationEvent(), it goes to the corresponding listener. In our case, it is a selector data update, data synchronization is zookeeper, so, the code will enter the ZookeeperDataChangedListener selector data change process.

    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // Iterate through the data change listener (usually using a data synchronization approach is fine)        for (DataChangedListener listener : listeners) {            // what kind of data has changed         switch (event.getGroupKey()) {                                    // other code logic is omitted                                    case SELECTOR:   // selector data                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());   // In our case, will enter the ZookeeperDataChangedListener selector data change process                    break;         }    }

2.4 Zookeeper Data Changed Listener#

• ZookeeperDataChangedListener.onSelectorChanged()

In the onSelectorChanged() method, determine the type of action, whether to refresh synchronization or update or create synchronization. Determine whether the node is in zk based on the current selector data.

/** * use ZooKeeper to publish change data */public class ZookeeperDataChangedListener implements DataChangedListener {        // The selector information changed    @Override    public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {        // refresh        if (eventType == DataEventTypeEnum.REFRESH && !changed.isEmpty()) {            String selectorParentPath = DefaultPathConstants.buildSelectorParentPath(changed.get(0).getPluginName());            deleteZkPathRecursive(selectorParentPath);        }        // changed data        for (SelectorData data : changed) {            // build selector real path            String selectorRealPath = DefaultPathConstants.buildSelectorRealPath(data.getPluginName(), data.getId());            // delete            if (eventType == DataEventTypeEnum.DELETE) {                deleteZkPath(selectorRealPath);                continue;            }            // selector parent path            String selectorParentPath = DefaultPathConstants.buildSelectorParentPath(data.getPluginName());            // create parent node            createZkNode(selectorParentPath);            // insert or update data            insertZkNode(selectorRealPath, data);        }    }
    // create zk node    private void createZkNode(final String path) {        // create only if it does not exist        if (!zkClient.exists(path)) {            zkClient.createPersistent(path, true);        }    }
    // insert zk node    private void insertZkNode(final String path, final Object data) {        // create zk node        createZkNode(path);        // write data by zkClient         zkClient.writeData(path, null == data ? "" : GsonUtils.getInstance().toJson(data));    }    }

As long as the changed data is correctly written to the zk node, the admin side of the operation is complete. ShenYu uses zk for data synchronization, zk nodes are carefully designed.

In our current case, updating one of the selector data in the Divide plugin with a weight of 90 updates specific nodes in the graph.

We series the above update flow with a sequence diagram.

3. Gateway Data Sync#

Assume that the ShenYu gateway is already running properly, and the data synchronization mode is also Zookeeper. How does the gateway receive and process the selector data after updating it on the admin side and sending the changed data to ZK? Let's continue our source code analysis to find out.

3.1 ZkClient Accept Data#

• ZkClient.subscribeDataChanges()

There is a ZookeeperSyncDataService class on the gateway, which subscribing to the data node through ZkClient and can sense when the data changes.

/** * ZookeeperSyncDataService */public class ZookeeperSyncDataService implements SyncDataService, AutoCloseable {    private void subscribeSelectorDataChanges(final String path) {       // zkClient subscribe data         zkClient.subscribeDataChanges(path, new IZkDataListener() {            @Override            public void handleDataChange(final String dataPath, final Object data) {                cacheSelectorData(GsonUtils.getInstance().fromJson(data.toString(), SelectorData.class)); // zk node data changed            }
            @Override            public void handleDataDeleted(final String dataPath) {                unCacheSelectorData(dataPath);  // zk node data deleted            }        });    }     // ...}

ZooKeeper's Watch mechanism notifies subscribing clients of node changes. In our case, updating the selector information goes to the handleDataChange() method. cacheSelectorData() is used to process data.

3.2 Handle Data#

• ZookeeperSyncDataService.cacheSelectorData()

The data is not null, and caching the selector data is again handled by PluginDataSubscriber.

    private void cacheSelectorData(final SelectorData selectorData) {        Optional.ofNullable(selectorData)                .ifPresent(data -> Optional.ofNullable(pluginDataSubscriber).ifPresent(e -> e.onSelectorSubscribe(data)));    }

PluginDataSubscriber is an interface, it is only a CommonPluginDataSubscriber implementation class, responsible for data processing plugin, selector and rules.

3.3 Common Plugin Data Subscriber#

• PluginDataSubscriber.onSelectorSubscribe()

It has no additional logic and calls the subscribeDataHandler() method directly. Within methods, there are data types (plugins, selectors, or rules) and action types (update or delete) to perform different logic.

/** * The common plugin data subscriber, responsible for handling all plug-in, selector, and rule information */public class CommonPluginDataSubscriber implements PluginDataSubscriber {    //......     // handle selector data    @Override    public void onSelectorSubscribe(final SelectoData selectorData) {        subscribeDataHandler(selectorData, DataEventTypeEnum.UPDATE);    }            // A subscription data handler that handles updates or deletions of data    private <T> void subscribeDataHandler(final T classData, final DataEventTypeEnum dataType) {        Optional.ofNullable(classData).ifPresent(data -> {            // plugin data            if (data instanceof PluginData) {                PluginData pluginData = (PluginData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // update                    // save the data to gateway memory                     BaseDataCache.getInstance().cachePluginData(pluginData);                    // If each plugin has its own processing logic, then do it                    Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.handlerPlugin(pluginData));                } else if (dataType == DataEventTypeEnum.DELETE) {  // delete                    // delete the data from gateway memory                    BaseDataCache.getInstance().removePluginData(pluginData);                    // If each plugin has its own processing logic, then do it                    Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.removePlugin(pluginData));                }            } else if (data instanceof SelectorData) {  // selector data                SelectorData selectorData = (SelectorData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // update                    // save the data to gateway memory                    BaseDataCache.getInstance().cacheSelectData(selectorData);                    // If each plugin has its own processing logic, then do it                     Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));                } else if (dataType == DataEventTypeEnum.DELETE) {  // delete                    // delete the data from gateway memory                    BaseDataCache.getInstance().removeSelectData(selectorData);                    // If each plugin has its own processing logic, then do it                    Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.removeSelector(selectorData));                }            } else if (data instanceof RuleData) {  // rule data                RuleData ruleData = (RuleData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // update                    // save the data to gateway memory                    BaseDataCache.getInstance().cacheRuleData(ruleData);                    // If each plugin has its own processing logic, then do it                    Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.handlerRule(ruleData));                } else if (dataType == DataEventTypeEnum.DELETE) { // delete                    // delete the data from gateway memory                    BaseDataCache.getInstance().removeRuleData(ruleData);                    // If each plugin has its own processing logic, then do it                    Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.removeRule(ruleData));                }            }        });    }    }

3.4 Data cached to Memory#

Adding a selector will enter the following logic:

// save the data to gateway memoryBaseDataCache.getInstance().cacheSelectData(selectorData);// If each plugin has its own processing logic, then do itOptional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));

One is to save the data to the gateway's memory. BaseDataCache is the class that ultimately caches data, implemented in a singleton pattern. The selector data is stored in the SELECTOR_MAP Map. In the subsequent use, also from this data.

public final class BaseDataCache {    // private instance    private static final BaseDataCache INSTANCE = new BaseDataCache();    // private constructor    private BaseDataCache() {    }        /**     * Gets instance.     *  public method     * @return the instance     */    public static BaseDataCache getInstance() {        return INSTANCE;    }        /**      * A Map of the cache selector data     * pluginName -> SelectorData.     */    private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();        public void cacheSelectData(final SelectorData selectorData) {        Optional.ofNullable(selectorData).ifPresent(this::selectorAccept);    }           /**     * cache selector data.     * @param data the selector data     */    private void selectorAccept(final SelectorData data) {        String key = data.getPluginName();        if (SELECTOR_MAP.containsKey(key)) { // Update operation, delete before insert            List<SelectorData> existList = SELECTOR_MAP.get(key);            final List<SelectorData> resultList = existList.stream().filter(r -> !r.getId().equals(data.getId())).collect(Collectors.toList());            resultList.add(data);            final List<SelectorData> collect = resultList.stream().sorted(Comparator.comparing(SelectorData::getSort)).collect(Collectors.toList());            SELECTOR_MAP.put(key, collect);        } else {  // Add new operations directly to Map            SELECTOR_MAP.put(key, Lists.newArrayList(data));        }    }    }

Second, if each plugin has its own processing logic, then do it. Through the IDEA editor, you can see that after adding a selector, there are the following plugins and processing. We're not going to expand it here.

After the above source tracking, and through a practical case, in the admin end to update a selector data, the ZooKeeper data synchronization process analysis is clear.

Let's series the data synchronization process on the gateway side through the sequence diagram:

The data synchronization process has been analyzed. In order to prevent the synchronization process from being interrupted, other logic is ignored during the analysis. We also need to analyze the process of Admin synchronization data initialization and gateway synchronization operation initialization.

4. Admin Data Sync initialization#

When admin starts, the current data will be fully synchronized to zk, the implementation logic is as follows:

/** * Zookeeper data init */public class ZookeeperDataInit implements CommandLineRunner {
    private final ZkClient zkClient;
    private final SyncDataService syncDataService;
    /**     * Instantiates a new Zookeeper data init.     *     * @param zkClient        the zk client     * @param syncDataService the sync data service     */    public ZookeeperDataInit(final ZkClient zkClient, final SyncDataService syncDataService) {        this.zkClient = zkClient;        this.syncDataService = syncDataService;    }
    @Override    public void run(final String... args) {        String pluginPath = DefaultPathConstants.PLUGIN_PARENT;        String authPath = DefaultPathConstants.APP_AUTH_PARENT;        String metaDataPath = DefaultPathConstants.META_DATA;        // Determine whether data exists in zk        if (!zkClient.exists(pluginPath) && !zkClient.exists(authPath) && !zkClient.exists(metaDataPath)) {            syncDataService.syncAll(DataEventTypeEnum.REFRESH);        }    }}

Check whether there is data in zk, if not, then synchronize.

ZookeeperDataInit implements the CommandLineRunner interface. It is an interface provided by SpringBoot that executes the run() method after all Spring Beans initializations and is often used for initialization operations in a project.

• SyncDataService.syncAll()

Query data from the database, and then perform full data synchronization, all authentication information, plugin information, selector information, rule information, and metadata information. Synchronous events are published primarily through eventPublisher. After publishing the event via publishEvent(), the ApplicationListener performs the event change operation. In ShenYu is mentioned in DataChangedEventDispatcher.

@Servicepublic class SyncDataServiceImpl implements SyncDataService {    // eventPublisher    private final ApplicationEventPublisher eventPublisher;         /***     * sync all data     * @param type the type     * @return     */    @Override    public boolean syncAll(final DataEventTypeEnum type) {        // app auth data        appAuthService.syncData();        // plugin data        List<PluginData> pluginDataList = pluginService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, type, pluginDataList));        // selector data        List<SelectorData> selectorDataList = selectorService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, type, selectorDataList));        // rule data        List<RuleData> ruleDataList = ruleService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, type, ruleDataList));        // metadata        metaDataService.syncData();        return true;    }    }

5. Gateway Data Sync Init#

The initial operation of data synchronization on the gateway side is mainly the node in the subscription zk. When there is a data change, the changed data will be received. This relies on the Watch mechanism of ZooKeeper. In ShenYu, the one responsible for zk data synchronization is ZookeeperSyncDataService, also mentioned earlier.

The function logic of ZookeeperSyncDataService is completed in the process of instantiation: the subscription to Shenyu data synchronization node in zk is completed. Subscription here is divided into two kinds, one kind is existing node data updated above, through this zkClient.subscribeDataChanges() method; Another kind is under the current node, add or delete nodes change namely child nodes, it through zkClient.subscribeChildChanges() method.

ZookeeperSyncDataService code is a bit too much, here we use plugin data read and subscribe to track, other types of data operation principle is the same.

/** *  zookeeper sync data service */public class ZookeeperSyncDataService implements SyncDataService, AutoCloseable {    // At instantiation time, the data is read from the ZK and the node is subscribed    public ZookeeperSyncDataService(/* omit the construction argument */ ) {        this.zkClient = zkClient;        this.pluginDataSubscriber = pluginDataSubscriber;        this.metaDataSubscribers = metaDataSubscribers;        this.authDataSubscribers = authDataSubscribers;        // watch plugin, selector and rule data        watcherData();        // watch app auth data        watchAppAuth();        // watch metadata        watchMetaData();    }        private void watcherData() {        // plugin node path        final String pluginParent = DefaultPathConstants.PLUGIN_PARENT;        // all plugin nodes        List<String> pluginZKs = zkClientGetChildren(pluginParent);        for (String pluginName : pluginZKs) {            // watch plugin, selector, rule data node            watcherAll(pluginName);        }        //subscribing to child nodes (adding or removing a plugin)        zkClient.subscribeChildChanges(pluginParent, (parentPath, currentChildren) -> {            if (CollectionUtils.isNotEmpty(currentChildren)) {                for (String pluginName : currentChildren) {                    // you need to subscribe to all plugin, selector, and rule data for the child node                      watcherAll(pluginName);                }            }        });    }        private void watcherAll(final String pluginName) {        // watch plugin        watcherPlugin(pluginName);        // watch selector        watcherSelector(pluginName);        // watch rule        watcherRule(pluginName);    }
    private void watcherPlugin(final String pluginName) {        // plugin path        String pluginPath = DefaultPathConstants.buildPluginPath(pluginName);        // create if not exist        if (!zkClient.exists(pluginPath)) {            zkClient.createPersistent(pluginPath, true);        }        // read the current node data on zk and deserialize it        PluginData pluginData = null == zkClient.readData(pluginPath) ? null                : GsonUtils.getInstance().fromJson((String) zkClient.readData(pluginPath), PluginData.class);        // cached into gateway memory        cachePluginData(pluginData);        // subscribe plugin data        subscribePluginDataChanges(pluginPath, pluginName);    }       private void cachePluginData(final PluginData pluginData) {    //omit implementation logic, is actually the CommonPluginDataSubscriber operation, can connect with the front    }        private void subscribePluginDataChanges(final String pluginPath, final String pluginName) {        // subscribe data changes        zkClient.subscribeDataChanges(pluginPath, new IZkDataListener() {
            @Override            public void handleDataChange(final String dataPath, final Object data) {  // update                 //omit implementation logic, is actually the CommonPluginDataSubscriber operation, can connect with the front            }
            @Override            public void handleDataDeleted(final String dataPath) {   // delete                  // Omit implementation logic, is actually the CommonPluginDataSubscriber operation, can connect with the front
            }        });    }    }    

The above source code is given comments, I believe you can understand. The main logic for subscribing to plug-in data is as follows:

1. Create the current plugin path
2. Create a path if it does not exist
3. Read the current node data on zK and deserialize it
4. The plugin data is cached in the gateway memory
5. Subscribe to the plug-in node

6. Summary#

This paper through a practical case, Zookeeper data synchronization principle source code analysis. The main knowledge points involved are as follows:

• Data synchronization based on ZooKeeper is mainly implemented through watch mechanism;

• Complete event publishing and listening via Spring;

• Support multiple synchronization strategies through abstract DataChangedListener interface, interface oriented programming;

• Use singleton design pattern to cache data class BaseDataCache;

• Loading of configuration classes via conditional assembly of SpringBoot and starter loading mechanism.

This article will conduct an in-depth analysis of Apache ShenYu e2e module.

what is e2e#

e2e (end to end), also known as end-to-end testing, is a method used to test whether the application flow performs as designed from the beginning to the end. The purpose of performing end-to-end testing is to identify system dependencies and ensure that the correct information is passed between various system components and systems. The purpose of end-to-end testing is to test the entire software for dependencies, data integrity, and communication with other systems, interfaces, and databases to simulate a complete production scenario.

Advantages of e2e#

e2e testing can test the integrity and accuracy of software systems in simulated real user scenarios, and can verify whether the entire system works as expected and whether different components can work together. There are several benefits of e2e testing:

1. Help ensure the correctness of system functions.e2e testing can simulate the interaction and operation in real user scenarios, verify whether the entire system can work as expected, and help discover potential problems and defects in the system.
2. Improve test coverage.e2e testing can cover the entire system, including front-end, back-end, database and other different levels and components, thereby improving test coverage and ensuring comprehensiveness and accuracy of testing.
3. Ensure the stability of the system.e2e testing can check the stability and robustness of the system in various situations, including system response time, error handling capabilities, concurrency, etc., to help ensure that the system is in the face of high load and abnormal conditions Still able to maintain stable operation.
4. Reduce testing cost.e2e testing can improve testing efficiency and accuracy, reduce testing cost and time, and thus help enterprises release and deliver high-quality software products more quickly.

In short, e2e testing is a comprehensive testing method that can verify whether the entire system works as expected, improve test coverage and test efficiency, thereby ensuring the stability and correctness of the system, and reducing testing costs and time. And effective testing methods, so we need to improve e2e related codes.

How to implement automated e2e testing#

In Apache ShenYu, the main steps of e2e testing are reflected in the script of the GitHub Action workflow, as shown below, the script is located at ~/.github/workflows directory in the e2e file.

name: e2e
on:  pull_request:  push:    branches:      - masterjobs:  changes:    ...  build-docker-images:    ...  e2e-http:    ...  e2e-case:    runs-on: ubuntu-latest    needs:      - changes      - build-docker-images    if: ${{ needs.changes.outputs.e2e == 'true' }}    strategy:      matrix:        case: [ "shenyu-e2e-case-spring-cloud", "shenyu-e2e-case-apache-dubbo", "shenyu-e2e-case-sofa" ]    steps:      - uses: actions/checkout@v3        with:          submodules: true      - name: Load ShenYu Docker Images        run: |          docker load --input /tmp/apache-shenyu-admin.tar          docker load --input /tmp/apache-shenyu-bootstrap.tar          docker image ls -a      - name: Build examples with Maven        run: ./mvnw -B clean install -Pexample -Dmaven.javadoc.skip=true -Dmaven.test.skip=true -f ./shenyu-examples/pom.xml      - name: Run ShenYu E2E Tests        env:          storage: mysql        run: |          bash ./shenyu-e2e/script/storage_init.sh          ./mvnw -B -f ./shenyu-e2e/pom.xml -pl shenyu-e2e-case/${{ matrix.case }} -Dstorage=mysql test

When the workflow is triggered, use the dockerfile under the shenyu-dist module to build and upload the images of the admin and bootstrap projects. When the e2e test module is running, the admin and bootstrap images can be loaded. Then build the modules in the examples, and finally execute the test method of the corresponding test module.

How to run e2e test locally#

If you need to write e2e test cases, you first need to code and debug locally. Currently e2e supports two startup methods, one is docker startup and the other is host startup. These two modes can be switched in the @ShenYuTest annotation in the test class. The host startup method directly starts the services that need to be started locally to run the test code. Before using docker to start, you need to build the corresponding image first. Because ShenYu currently needs to support e2e testing in the github workflow, it is recommended to use the docker startup method.

Analysis of e2e startup process#

Currently, the e2e module is mainly divided into four parts: case, client, common and engine.

The case module stores the test cases of the plug-in, and the client module writes the clients of admin and gateway to request corresponding interfaces. Common stores some public classes, and the engine module is the core of the framework. Relying on the testcontainer framework, use java code to start the docker container and complete the configuration operations for admin and gatewat.

Next, I will analyze the e2e startup process based on the source code.

When we execute the test method in the case, the @ShenYuTest annotation will take effect and extend the test class. Through @ShenYuTest, we can choose the startup method, configure related parameters for admin and gateway, and choose the docker-compose file to be executed. For admin and gateway, you can configure the user name, password, data synchronization method and modify the content of yaml required for login.

@ShenYuTest(        mode = ShenYuEngineConfigure.Mode.DOCKER,        services = {                @ShenYuTest.ServiceConfigure(                        serviceName = "admin",                        port = 9095,                        baseUrl = "http://{hostname:localhost}:9095",                        parameters = {                                @ShenYuTest.Parameter(key = "username", value = "admin"),                                @ShenYuTest.Parameter(key = "password", value = "123456"),                                @ShenYuTest.Parameter(key = "dataSyn", value = "admin_websocket")                        }                ),                @ShenYuTest.ServiceConfigure(                        serviceName = "gateway",                        port = 9195,                        baseUrl = "http://{hostname:localhost}:9195",                         type = ShenYuEngineConfigure.ServiceType.SHENYU_GATEWAY,                        parameters = {                          @ShenYuTest.Parameter(key = "application", value =  "spring.cloud.discovery.enabled:true,eureka.client.enabled:true"),                           @ShenYuTest.Parameter(key = "dataSyn", value = "gateway_websocket")})},                   dockerComposeFile = "classpath:./docker-compose.mysql.yml")

@ShenYuTest is extended through the ShenYuExtension class, and the configuration of admin and gateway takes effect in beforeAll in ShenYuExtension. The specific effective logic is implemented in the DockerServiceCompose class.

@ShenYuTest configuration items take effect before docker starts, mainly by modifying the yaml file in the resource directory of the test module. Currently, e2e supports testing of different data synchronization methods. The principle is to use the chooseDataSyn method in the DockerServiceCompose class. In the DataSyncHandler, initialize the content that needs to be modified in various data synchronization methods, and finally start the container.

When docker is started, start testing the plug-in function. In the PluginsTest class, there are pre- and post-operations for testing.

    @BeforeAll    static void setup(final AdminClient adminClient, final GatewayClient gatewayClient) throws InterruptedException, JsonProcessingException {        adminClient.login();        Thread.sleep(10000);        List<SelectorDTO> selectorDTOList = adminClient.listAllSelectors();        List<MetaDataDTO> metaDataDTOList = adminClient.listAllMetaData();        List<RuleDTO> ruleDTOList = adminClient.listAllRules();        Assertions.assertEquals(2, selectorDTOList.size());        Assertions.assertEquals(13, metaDataDTOList.size());        Assertions.assertEquals(14, ruleDTOList.size());                for (SelectorDTO selectorDTO : selectorDTOList) {            if (selectorDTO.getHandle() != null && !"".equals(selectorDTO.getHandle())) {                SpringCloudPluginCases.verifierUri(selectorDTO.getHandle());            }        }        List<MetaData> metaDataCacheList = gatewayClient.getMetaDataCache();        List<SelectorCacheData> selectorCacheList = gatewayClient.getSelectorCache();        List<RuleCacheData> ruleCacheList = gatewayClient.getRuleCache();        Assertions.assertEquals(2, selectorCacheList.size());        Assertions.assertEquals(13, metaDataCacheList.size());        Assertions.assertEquals(14, ruleCacheList.size());
        MultiValueMap<String, String> formData = new LinkedMultiValueMap<>();        formData.add("id", "8");        formData.add("name", "springCloud");        formData.add("enabled", "true");        formData.add("role", "Proxy");        formData.add("sort", "200");        adminClient.changePluginStatus("8", formData);        String id = "";        for (SelectorDTO selectorDTO : selectorDTOList) {            if (!"".equals(selectorDTO.getHandle())) {                id = selectorDTO.getId();            }        }        adminClient.deleteSelectors(id);        selectorDTOList = adminClient.listAllSelectors();        Assertions.assertEquals(1, selectorDTOList.size());    }

Taking the springcloud plug-in as an example, you first need to test whether the registration center and data synchronization can work normally, then start the plug-in and delete the existing selector. To test whether the data is successfully registered into the registration center, you can call the interface of the admin client to test, and to test whether the data synchronization is successful, you can obtain the cache of the gateway for testing.

Then run the test case in the case file and get the use case through @ShenYuScenario.

    @ShenYuScenario(provider = SpringCloudPluginCases.class)    void testSpringCloud(GatewayClient gateway, CaseSpec spec) {        spec.getVerifiers().forEach(verifier -> verifier.verify(gateway.getHttpRequesterSupplier().get()));    }

For different plug-ins, we can build a Case class to store the rules to be tested. All test rules are stored in the list and tested in order. Build selectors and rules in beforeEachSpec, caseSpec stores test entities, if they meet the uri rules, they should exist, otherwise they don’t exist. We need to simulate users to add selectors and rules, because the handler rules of the selectors of each plug-in are not necessarily the same, so we need to write its handle class according to the plug-in requirements. And verify that it complies with the rules with the request. Specific test cases are mainly divided into two categories, one is to match uri rules, such as euqal, path_pattern, start_with, end_with, and the other is request types, such as get, put, post, delete.

When all eight matching conditions are tested, it can be judged that the plug-in function is normal. After the test, we need to restore the environment, delete all selectors, set the plug-in to unavailable, and finally close all containers.

    @Override    public List<ScenarioSpec> get() {        return Lists.newArrayList(                testWithUriEquals(),                testWithUriPathPattern(),                testWithUriStartWith(),                testWithEndWith(),                testWithMethodGet(),                testWithMethodPost(),                testWithMethodPut(),                testWithMethodDelete()        );    }
    private ShenYuScenarioSpec testWithUriEquals() {        return ShenYuScenarioSpec.builder()                .name("single-spring-cloud uri =]")                .beforeEachSpec(                        ShenYuBeforeEachSpec.builder()                                .addSelectorAndRule(                                        newSelectorBuilder("selector", Plugin.SPRING_CLOUD)                                               .handle(SpringCloudSelectorHandle.builder().serviceId("springCloud-test")                                                        .gray(true)                                                        .divideUpstreams(DIVIDE_UPSTREAMS).build())                                                .conditionList(newConditions(Condition.ParamType.URI, Condition.Operator.EQUAL, TEST))                                                .build(),                                        newRuleBuilder("rule")                               .handle(SpringCloudRuleHandle.builder().loadBalance("hash").timeout(3000).build())                                                .conditionList(newConditions(Condition.ParamType.URI, Condition.Operator.EQUAL, TEST))                                                .build()                                )                                .checker(notExists(TEST))                                .waiting(exists(TEST))                                .build()                )                .caseSpec(                        ShenYuCaseSpec.builder()                                .addExists(TEST)                                .addNotExists("/springcloud/te")                                .addNotExists("/put")                                .addNotExists("/get")                                .build()                )                .afterEachSpec(ShenYuAfterEachSpec.DEFAULT)                .build();    }

This article will provide an in-depth analysis of Apache ShenYu's integration tests.

What are integration tests?#

Integration testing is also called E2E (End To End) testing in some projects. It is mainly used to test whether each module can meet expectations after being assembled into a system.

Apache ShenYu puts integration tests in continuous integration, using GitHub Actions to trigger each time a Pull Request or Merge is submitted to the main branch. This can greatly reduce the maintenance cost of the project and improve the stability of Apache ShenYu.

How to automate integration testing?#

In Apache ShenYu, the main steps of integration testing are embodied in the script of the GitHub Action workflow, as shown below, which is located at ~/.github/workflows directory.

name: iton:  pull_request:  push:    branches:      - masterjobs:  build:    strategy:      matrix:        case:          - shenyu-integrated-test-alibaba-dubbo          ...    runs-on: ubuntu-latest    steps:      - uses: actions/checkout@v2        with:          submodules: true      ...

Next, I will start from this yaml file and take you to analyze the entire process of automated integration testing.

Triggering the workflow#

Since we specified pull_request and push.branch: master in on, this workflow will be triggered when we submit pull_request or merge branch to master (push).

For more usage of GitHub Action, you can refer to the documentation of GitHub Action, which will not be introduced in detail here.

Initialize the environment#

• pull code

- uses: actions/checkout@v2  with:        submodules: true

• set skip flag

- name: Set Skip Env Var      uses: ./.github/actions/skip-ci

When something unrelated to functionality occurs (such as changing documentation), integration tests are skipped to save resources.

• Cache maven repos, install Java

- name: Cache Maven Repos...- uses: actions/setup-java@v1

Build the entire project while building the docker image#

./mvnw -B clean install -Prelease,docker -Dmaven.javadoc.skip=true -Dmaven.test.skip=true

In the above command, -P is followed by release,docker, which means that the relevant profile configuration in the pom file will be activated.

The two profiles, release and docker, currently only exist in several submodules under shenyu-dist. The following will take the shenyu-dist-admin module as an example to introduce profiles as release and docker The specific content of the configuration. Also, integration tests only use the shenyu-admin image built in this step.

• First is release

  <profile>    <id>release</id>    <activation>        <activeByDefault>false</activeByDefault>    </activation>    <build>        <finalName>apache-shenyu-incubating-${project.version}</finalName>        <plugins>            <plugin>                <groupId>org.apache.maven.plugins</groupId>                <artifactId>maven-assembly-plugin</artifactId>                <executions>                    <execution>                        <id>admin-bin</id>                        <phase>package</phase>                        <goals>                            <goal>single</goal>                        </goals>                    </execution>                </executions>                <configuration>                    <descriptors>                        <descriptor>${project.basedir}/src/main/assembly/binary.xml</descriptor>                    </descriptors>                    <tarLongFileMode>posix</tarLongFileMode>                </configuration>            </plugin>        </plugins>    </build></profile>

  Copy

  When -P is followed by release, the above maven-assembly-plugin plugin is activated. In executions, the execution timing of the plugin is bound to the maven life cycle package, which means that it will be triggered when we execute mvn install.

  The binary.xml we wrote is specified in the configuration, and the maven-assembly-plugin plugin will copy the required files and package them according to this file. You can click the link to view the file: [shenyu-dist/shenyu-admin-dist/src/main/assembly/binary.xml](https://github.com/apache/shenyu/blob/master/shenyu- dist/shenyu-admin-dist/src/main/assembly/binary.xml)

  According to this file, the plugin will "copy" the packaged jar packages, configuration files, startup scripts, etc. under other modules, and finally make them into a compressed package in tar.gz format.

• then docker

  <profile>    <id>docker</id>    <activation>        <activeByDefault>false</activeByDefault>    </activation>    <build>        <plugins>            <plugin>                <groupId>com.spotify</groupId>                <artifactId>dockerfile-maven-plugin</artifactId>                <version>${dockerfile-maven-plugin.version}</version>                <executions>                    <execution>                        <id>tag-latest</id>                        <goals>                            <goal>build</goal>                        </goals>                        <configuration>                            <tag>latest</tag>                        </configuration>                    </execution>                    <execution>                        <id>tag-version</id>                        <goals>                            <goal>build</goal>                        </goals>                        <configuration>                            <tag>${project.version}</tag>                        </configuration>                    </execution>                </executions>                <configuration>                    <repository>apache/shenyu-admin</repository>                    <buildArgs>                        <APP_NAME>apache-shenyu-incubating-${project.version}-admin-bin</APP_NAME>                    </buildArgs>                </configuration>            </plugin>        </plugins>    </build></profile>

  Copy

  Similar to the release above, here is the activation of the dockerfile-maven-plugin plugin. When mvn install -Pdocker, the plugin will use the dockerfile we wrote to build the docker image.

It should be noted that the dockerfile-maven-plugin currently has limited support for aarch64 architecture devices, and the following error will occur when running the plugin on aarch64 architecture machines. And when I wrote this article, it has not been maintained for a long time, which means that the problem of aarch64 architecture devices using this plugin will not be solved in the short term.

[ERROR] Failed to execute goal com.spotify:dockerfile-maven-plugin:1.4.6:build (tag-latest) on project shenyu-admin-dist: Could not build image: java.util.concurrent.ExecutionException: com.spotify.docker.client.shaded.javax.ws.rs.ProcessingException: java.lang.UnsatisfiedLinkError: could not load FFI provider jnr.ffi.provider.jffi.Provider: ExceptionInInitializerError: Can't overwrite cause with java.lang.UnsatisfiedLinkError: java.lang.UnsatisfiedLinkError: /private/var/folders/w2/j27f16yj7cvf_1cxbgqb89vh0000gn/T/jffi4972193792308935312.dylib: dlopen(/private/var/folders/w2/j27f16yj7cvf_1cxbgqb89vh0000gn/T/jffi4972193792308935312.dylib, 1): no suitable image found.  Did find:[ERROR]         /private/var/folders/w2/j27f16yj7cvf_1cxbgqb89vh0000gn/T/jffi4972193792308935312.dylib: no matching architecture in universal wrapper[ERROR]         /private/var/folders/w2/j27f16yj7cvf_1cxbgqb89vh0000gn/T/jffi4972193792308935312.dylib: no matching architecture in universal wrapper...

Here is a temporary solution:

1. Open a new shell, enter the following command, and use socat to route the unix socket to the tcp port

   socat TCP-LISTEN:2375,range=127.0.0.1/32,reuseaddr,fork UNIX-CLIENT:/var/run/docker.sock

   Copy

2. Set environment variables

   export DOCKER_HOST=tcp://127.0.0.1:2375

   Copy

Build the examples module#

- name: Build examples  if: env.SKIP_CI != 'true'  run: ./mvnw -B clean install -Pexample -Dmaven.javadoc.skip=true -Dmaven.test.skip=true -f ./shenyu-examples/pom.xml

Considering the need for release, the current pom file in the project root directory does not contain the example submodule, so the examples module is additionally built in the above step.

Similar to the above, this line of command will also use the maven plugin to build an image for our subsequent docker orchestration.

Build custom gateways#

- name: Build integrated tests  if: env.SKIP_CI != 'true'  run: ./mvnw -B clean install -Pit -DskipTests -f ./shenyu-integrated-test/pom.xml

In order to subdivide the integration tests of different functions of Apache ShenYu, we will build a gateway customized by the integration test module in this step. The so-called "customization" is to introduce the minimum required dependencies in the pom file, and then replace the default shenyu-bootstrap. Similar to the above two steps, this step will also build the docker image.

It is worth noting that the way of packaging and building here is slightly different from that of the shenyu-dist module, which you can find by comparing the pom file.

Run docker compose#

- name: Start docker compose  if: env.SKIP_CI != 'true'  run: docker-compose -f ./shenyu-integrated-test/${{ matrix.case }}/docker-compose.yml up -d

In this step, docker will be arranged according to the different docker-compose.yml files written under the integration test module.

version: "3.9"services:  shenyu-zk:    container_name: shenyu-zk    image: zookeeper:3.5    ...  shenyu-redis:    image: redis:6.0-alpine    container_name: shenyu-redis    ...
  shenyu-examples-http:    deploy:      resources:        limits:          memory: 2048M    container_name: shenyu-examples-http    image: shenyu-examples-http:latest    ...
  shenyu-admin:    image: apache/shenyu-admin:latest    container_name: shenyu-admin    ...
  shenyu-integrated-test-http:    container_name: shenyu-integrated-test-http    image: apache/shenyu-integrated-test-http:latest    ...    depends_on:      shenyu-admin:        condition: service_healthy    healthcheck:      test: [ "CMD", "wget", "http://shenyu-integrated-test-http:9195/actuator/health" ]      timeout: 2s      retries: 30
networks:  shenyu:    name: shenyu

For example, the docker-compose.yml under the shenyu-integrated-test-http module starts zookeeper, redis, example, admin, gateway and other services in sequence. Among them, the mirrors of example, admin, and gateway are built by us before.

Among them, docker-compose uses depends_on to determine the topological relationship between services, and most services have corresponding health checks, and the next service will not be started until the health check passes.

Run the health check and wait for docker-compose to start#

- name: Wait for docker compose start up completely  if: env.SKIP_CI != 'true'  run: bash ./shenyu-integrated-test/${{ matrix.case }}/script/healthcheck.sh

In this step, the host will run the healthcheck.sh script, and then use the curl command to access the health status interface /actuator/health of each service list (in the services.list file), until the service status is normal. will continue.

run tests#

- name: Run test  id: test  if: env.SKIP_CI != 'true'  run: ./mvnw test -Pit -f ./shenyu-integrated-test/${{ matrix.case }}/pom.xml  continue-on-error: true

This step is to use the maven test command to execute the test classes in the /src/test/ directory one by one.

View Docker Compose logs#

- name: Check test result  if: env.SKIP_CI != 'true'  run: |    docker-compose -f ./shenyu-integrated-test/${{ matrix.case }}/docker-compose.yml logs --tail="all"    if [[ ${{steps.test.outcome}} == "failure" ]]; then      echo "Test Failed"      exit 1    else      echo "Test Successful"      exit 0    fi

When there is an error in the workflow, the log of docker compose can help us to better troubleshoot the problem, so in this step, we will print the log of docker compose.

This article is based on the source code analysis of version 'shenyu-2.6.1'

Shenyu provides a mechanism to customize its own plugins or modify existing plugins, which is implemented internally through the configuration of extPlugin. It needs to meet the following two points：

1. Implement interface ShenyuPlugin or PluginDataHandler.
2. After packaging the implemented package, place it in the corresponding path of 'shenyu. extPlugin. path'.

Entry#

The class that truly implements this logic is' ShenyuLoaderService '. Now let's take a look at how this class handles it.

    public ShenyuLoaderService(final ShenyuWebHandler webHandler, final CommonPluginDataSubscriber subscriber, final ShenyuConfig shenyuConfig) {        // Information subscription for plugin information        this.subscriber = subscriber;        // The WebHandler encapsulated by Shenyu contains all the plugin logic        this.webHandler = webHandler;        // configuration information        this.shenyuConfig = shenyuConfig;        // The configuration information of the extension plugin, such as path, whether it is enabled, how many threads are enabled to process, and the frequency of loading checks        ExtPlugin config = shenyuConfig.getExtPlugin();        // If enabled, create a scheduled task to check and load        if (config.getEnabled()) {            // Create a scheduled task with a specified thread name            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(config.getThreads(), ShenyuThreadFactory.create("plugin-ext-loader", true));            // Create a task to be executed at a fixed frequency, with a default time of 30 seconds and execution every 300 seconds            executor.scheduleAtFixedRate(() -> loadExtOrUploadPlugins(null), config.getScheduleDelay(), config.getScheduleTime(), TimeUnit.SECONDS);        }    }    

This class has the following properties:

WebHandler: This class is the entry point for shenyu to process requests, referencing all plugin data. After the extension plugin is loaded, it needs to be updated.

Subscriber: This class is the entry point for the subscription of plugins, referencing the subscription processing classes of all plugins. After the extension configuration is loaded, synchronous updates are also required.

Executor: A scheduled task will be created inside' ShenyuLoaderService 'to periodically scan and load jar packages under the specified path, facilitating the loading of extended plugins and achieving dynamic discovery By default, it will scan every 300 seconds after 30 seconds of startup.

Meanwhile, the decision to enable extension plugin functionality can be made through the configuration of shenyu. extPlugin. enabled.

The above configurations can be adjusted in the configuration file:

shenyu:  extPlugin:    path:   # Storage directory for extension plugins    enabled: true # Is the extension function enabled    threads: 1 # Number of threads loaded by scanning    scheduleTime: 300 # The frequency of task execution    scheduleDelay: 30 # How long after the task starts to execute

Next, let's take a look at the loading logic：

   public void loadExtOrUploadPlugins(final PluginData uploadedJarResource) {        try {            List<ShenyuLoaderResult> plugins = new ArrayList<>();            // Obtain the holding object of ShenyuPluginClassloader            ShenyuPluginClassloaderHolder singleton = ShenyuPluginClassloaderHolder.getSingleton();            if (Objects.isNull(uploadedJarResource)) {                // If the parameter is empty, load all jar packages from the extended directory                // PluginJar: Data containing the ShenyuPlugin interface and PluginDataHandler interface                List<PluginJarParser.PluginJar> uploadPluginJars = ShenyuExtPathPluginJarLoader.loadExtendPlugins(shenyuConfig.getExtPlugin().getPath());                // Traverse all pending plugins                for (PluginJarParser.PluginJar extPath : uploadPluginJars) {                    LOG.info("shenyu extPlugin find new {} to load", extPath.getAbsolutePath());                    // Use the loader of the extension plugin to load the specified plugin, facilitating subsequent loading and unloading                    ShenyuPluginClassLoader extPathClassLoader = singleton.createPluginClassLoader(extPath);                    // Using ShenyuPluginClassLoader for loading                    // The main logic is to determine whether to implement ShenyuPlugin interface, PluginDataHandler interface, or identify annotations such as @ Component \ @ Service. If so, register as SpringBean                    // Construct ShenyuLoaderResult object                    plugins.addAll(extPathClassLoader.loadUploadedJarPlugins());                }            } else {                // Load the specified jar, with the same logic as loading all                PluginJarParser.PluginJar pluginJar = PluginJarParser.parseJar(Base64.getDecoder().decode(uploadedJarResource.getPluginJar()));                LOG.info("shenyu upload plugin jar find new {} to load", pluginJar.getJarKey());                ShenyuPluginClassLoader uploadPluginClassLoader = singleton.createPluginClassLoader(pluginJar);                plugins.addAll(uploadPluginClassLoader.loadUploadedJarPlugins());            }            // Add the extended plugins to the plugin list of ShenyuWebHandler, and subsequent requests will go through the added plugin content            loaderPlugins(plugins);        } catch (Exception e) {            LOG.error("shenyu plugins load has error ", e);        }    }

The logic processed by this method:

1. Check if the parameter uploadedJarResource has a value. If not, all will be loaded. Otherwise, load the specified resource jar package for processing.

2. Retrieve the specified jar package from shenyu. extPlugin. path and encapsulate it as a PluginJar object, which contains the following information about the jar package:

   • version: version information

   • groupId: The groupId of the package

   • artifactId: The artifactId of the package

   • absolutePath: Absolute path

   • clazzMap: Bytecode corresponding to class

   • resourceMap: Bytecode of jar package

3. Create a corresponding ClassLoader using ShenyuPluginClassloaderHolder, with the corresponding class being 'ShenyuPluginClassLoader', and load the corresponding class accordingly.

   • Call ShenyuPluginClassLoader. loadUploadedJarPlugins to load the corresponding class and register it as a Spring Bean, which can be managed using the Spring container

4. Call the loaderPlugins method to update the extended plugin to'webHandler and subscriber.

Plugin Registration#

For the content in the provided jar package, the loader will only handle classes of the specified interface type, and the implementation logic is in the ShenyuPluginClassLoader.loadUploadedJarPlugins() method.

public List<ShenyuLoaderResult> loadUploadedJarPlugins() {        List<ShenyuLoaderResult> results = new ArrayList<>();        // All class mapping relationships        Set<String> names = pluginJar.getClazzMap().keySet();        // Traverse all classes        names.forEach(className -> {            Object instance;            try {                // Try creating objects and, if possible, add them to the Spring container                instance = getOrCreateSpringBean(className);                if (Objects.nonNull(instance)) {                    // Building the ShenyuLoaderResult object                    results.add(buildResult(instance));                    LOG.info("The class successfully loaded into a upload-Jar-plugin {} is registered as a spring bean", className);                }            } catch (ClassNotFoundException | IllegalAccessException | InstantiationException e) {                LOG.warn("Registering upload-Jar-plugins succeeds spring bean fails:{}", className, e);            }        });        return results;    }

This method is responsible for building all eligible objects and encapsulating them into a ShenyuLoaderResult object. This object is encapsulated for the created object and will be processed in the method buildResult().

    private ShenyuLoaderResult buildResult(final Object instance) {        ShenyuLoaderResult result = new ShenyuLoaderResult();        // Does the created object implement ShenyuPlugin        if (instance instanceof ShenyuPlugin) {            result.setShenyuPlugin((ShenyuPlugin) instance);            // Does the created object implement PluginDataHandler        } else if (instance instanceof PluginDataHandler) {            result.setPluginDataHandler((PluginDataHandler) instance);        }        return result;    }

Simultaneously enter the method getOrCreatSpringBean() for further analysis:

    private <T> T getOrCreateSpringBean(final String className) throws ClassNotFoundException, IllegalAccessException, InstantiationException {        // Confirm if it has been registered. If so, do not process it and return directly        if (SpringBeanUtils.getInstance().existBean(className)) {            return SpringBeanUtils.getInstance().getBeanByClassName(className);        }        lock.lock();        try {            // Double check,            T inst = SpringBeanUtils.getInstance().getBeanByClassName(className);            if (Objects.isNull(inst)) {                // Using ShenyuPluginClassLoader to load classes                Class<?> clazz = Class.forName(className, false, this);                //Exclude ShenyuPlugin subclass and PluginDataHandler subclass                // without adding @Component @Service annotation                // Confirm if it is a subclass of ShenyuPlugin or PluginDataHandler                boolean next = ShenyuPlugin.class.isAssignableFrom(clazz)                        || PluginDataHandler.class.isAssignableFrom(clazz);                if (!next) {                    // If not, confirm if @ Component and @ Service annotations are identified                    Annotation[] annotations = clazz.getAnnotations();                    next = Arrays.stream(annotations).anyMatch(e -> e.annotationType().equals(Component.class)                            || e.annotationType().equals(Service.class));                }                if (next) {                    // If the above content is met, register the bean                    GenericBeanDefinition beanDefinition = new GenericBeanDefinition();                    beanDefinition.setBeanClassName(className);                    beanDefinition.setAutowireCandidate(true);                    beanDefinition.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);                    // Registering beans                    String beanName = SpringBeanUtils.getInstance().registerBean(beanDefinition, this);                    // create object                    inst = SpringBeanUtils.getInstance().getBeanByClassName(beanName);                }            }            return inst;        } finally {            lock.unlock();        }    }

The logic is roughly as follows:

1. Check if the interface ShenyuPlugin or PluginDataHandler has been implemented. If not, check if @Component or @Service has been identified`.
2. If the condition of 1 is met, register the object in the Spring container and return the created object.

Sync Data#

After the plugin registration is successful, the plugin is only instantiated, but it will not take effect yet because it has not been added to Shenyu's plugin chain. The synchronization logic is implemented by the loaderPlugins() method.

    private void loaderPlugins(final List<ShenyuLoaderResult> results) {        if (CollectionUtils.isEmpty(results)) {            return;        }        // Get all objects that implement the interface ShenyuPlugin        List<ShenyuPlugin> shenyuExtendPlugins = results.stream().map(ShenyuLoaderResult::getShenyuPlugin).filter(Objects::nonNull).collect(Collectors.toList());        // Synchronize updating plugins in webHandler        webHandler.putExtPlugins(shenyuExtendPlugins);        // Get all objects that implement the interface PluginDataHandler        List<PluginDataHandler> handlers = results.stream().map(ShenyuLoaderResult::getPluginDataHandler).filter(Objects::nonNull).collect(Collectors.toList());        // Synchronize updating handlers in subscriber        subscriber.putExtendPluginDataHandler(handlers);
    }

The logic of this method processes two data points:

1. Synchronize the data that implements the ShenyuPlugin interface to the plugins list of webHandler.

    public void putExtPlugins(final List<ShenyuPlugin> extPlugins) {        if (CollectionUtils.isEmpty(extPlugins)) {            return;        }        // Filter out newly added plugins        final List<ShenyuPlugin> shenyuAddPlugins = extPlugins.stream()                .filter(e -> plugins.stream().noneMatch(plugin -> plugin.named().equals(e.named())))                .collect(Collectors.toList());        // Filter out updated plugins and determine if they have the same name as the old one, then it is an update        final List<ShenyuPlugin> shenyuUpdatePlugins = extPlugins.stream()                .filter(e -> plugins.stream().anyMatch(plugin -> plugin.named().equals(e.named())))                .collect(Collectors.toList());        // If there is no data, skip        if (CollectionUtils.isEmpty(shenyuAddPlugins) && CollectionUtils.isEmpty(shenyuUpdatePlugins)) {            return;        }        // Copy old data        // copy new list        List<ShenyuPlugin> newPluginList = new ArrayList<>(plugins);        // Add new plugin data        // Add extend plugin from pluginData or shenyu ext-lib        this.sourcePlugins.addAll(shenyuAddPlugins);        // Add new data        if (CollectionUtils.isNotEmpty(shenyuAddPlugins)) {            shenyuAddPlugins.forEach(plugin -> LOG.info("shenyu auto add extends plugins:{}", plugin.named()));            newPluginList.addAll(shenyuAddPlugins);        }        // Modify updated data        if (CollectionUtils.isNotEmpty(shenyuUpdatePlugins)) {            shenyuUpdatePlugins.forEach(plugin -> LOG.info("shenyu auto update extends plugins:{}", plugin.named()));            for (ShenyuPlugin updatePlugin : shenyuUpdatePlugins) {                for (int i = 0; i < newPluginList.size(); i++) {                    if (newPluginList.get(i).named().equals(updatePlugin.named())) {                        newPluginList.set(i, updatePlugin);                    }                }                for (int i = 0; i < this.sourcePlugins.size(); i++) {                    if (this.sourcePlugins.get(i).named().equals(updatePlugin.named())) {                        this.sourcePlugins.set(i, updatePlugin);                    }                }            }        }        // REORDER        plugins = sortPlugins(newPluginList);    }

2. Synchronize the data that implements the PluginDataHandler interface to the handlers list of the subscriber.

    public void putExtendPluginDataHandler(final List<PluginDataHandler> handlers) {        if (CollectionUtils.isEmpty(handlers)) {            return;        }        // Traverse all data        for (PluginDataHandler handler : handlers) {            String pluginNamed = handler.pluginNamed();            // Update existing PluginDataHandler list            MapUtils.computeIfAbsent(handlerMap, pluginNamed, name -> {                LOG.info("shenyu auto add extends plugin data handler name is :{}", pluginNamed);                return handler;            });        }    }

At this point, the analysis of the loading process of the extension plugin is completed.

Before starting, you can refer to this article to start the gateway

Body#

First, look at the ContextPathPlugin#doExecute method, which is the core of this plugin.

protected Mono<Void> doExecute(final ServerWebExchange exchange, final ShenyuPluginChain chain, final SelectorData selector, final RuleData rule) {    ...    // 1. get the contextMappingHandle from the JVM cache    ContextMappingHandle contextMappingHandle = ContextPathPluginDataHandler.CACHED_HANDLE.get().obtainHandle(CacheKeyUtils.INST.getKey(rule));    ...    // 2. set shenyu context according to contextMappingHandle    buildContextPath(shenyuContext, contextMappingHandle);    return chain.execute(exchange);}

1. Get the contextMappingHandle from the JVM cache

   The contextMappingHandle here is an instance of the ContextMappingHandle class, which has two member variables: contextPath and addPrefix

   These two variables have appeared in the Rules form in the Admin before, and they are updated when the data is synchronized.

2. Set shenyu context according to contextMappingHandle

   Below is the source code of the ContextPathPlugin#buildContextPath method

   private void buildContextPath(final ShenyuContext context, final ContextMappingHandle handle) {    String realURI = "";    // 1. set the context path of shenyu, remove the prefix of the real URI according to the length of the contextPath    if (StringUtils.isNoneBlank(handle.getContextPath())) {        context.setContextPath(handle.getContextPath());        context.setModule(handle.getContextPath());        realURI = context.getPath().substring(handle.getContextPath().length());    }    // add prefix    if (StringUtils.isNoneBlank(handle.getAddPrefix())) {        if (StringUtils.isNotBlank(realURI)) {            realURI = handle.getAddPrefix() + realURI;        } else {            realURI = handle.getAddPrefix() + context.getPath();        }    }    context.setRealUrl(realURI);}

   Copy

   • Set the context path of shenyu, remove the prefix of the real URI according to the length of the contextPath

     You may be wondering whether there is a problem with the so-called "according to the length of the contextPath" here?

     In fact, such a judgment is not a problem, because the request will be processed by the plugin only after it is matched by the Selector and Rules. Therefore, under the premise of setting up Selector and Rules, it is completely possible to meet the needs of converting a specific contextPath.

Then, the ContextPathPlugin class has a more important method skip, part of the code is shown below. We can find: If it is a call to the RPC service, the context_path plugin will be skipped directly.

public Boolean skip(final ServerWebExchange exchange) {    ...    return Objects.equals(rpcType, RpcTypeEnum.DUBBO.getName())            || Objects.equals(rpcType, RpcTypeEnum.GRPC.getName())            || Objects.equals(rpcType, RpcTypeEnum.TARS.getName())            || Objects.equals(rpcType, RpcTypeEnum.MOTAN.getName())            || Objects.equals(rpcType, RpcTypeEnum.SOFA.getName());}

Finally, the context-path plugin has another class ContextPathPluginDataHandler. The function of this class is to subscribe to the data of the plug-in. When the plugin configuration is modified, deleted, or added, the data is modified, deleted, or added to the JVM cache.

The ShenYu gateway uses the divide plugin to handle http requests. You can see the official documentation Quick start with Http to learn how to use this plugin.

This article is based on shenyu-2.4.3 version for source code analysis, please refer to Http Proxy for the introduction of the official website.

1. Register Service#

1.1 Declaration of registration interface#

Use the annotation @ShenyuSpringMvcClient to register the service to the gateway. The simple demo is as follows.

@RestController@RequestMapping("/order")@ShenyuSpringMvcClient(path = "/order")  // APIpublic class OrderController {    @GetMapping("/findById")    @ShenyuSpringMvcClient(path = "/findById", desc = "Find by id") // method    public OrderDTO findById(@RequestParam("id") final String id) {        return build(id, "hello world findById");    }}

define annotation:

@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE, ElementType.METHOD})public @interface ShenyuSpringMvcClient {        //path    String path() default "";        //rule name    String ruleName() default "";       //desc info    String desc() default "";
    //is enabled    boolean enabled() default true;        //register MetaData    boolean registerMetaData() default false;}

1.2 Scan annotation#

Annotation scanning is done through SpringMvcClientBeanPostProcessor, which implements the BeanPostProcessor interface and is a post-processor provided by Spring.

During constructor instantiation.

• Read the property configuration
• Add annotations, read path information
• Start the registry and register with shenyu-admin

public class SpringMvcClientBeanPostProcessor implements BeanPostProcessor {    //...    /**     * Constructor instantiation     */    public SpringMvcClientBeanPostProcessor(final PropertiesConfig clientConfig,                                            final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        // 1. read Properties        Properties props = clientConfig.getProps();        this.appName = props.getProperty(ShenyuClientConstants.APP_NAME);        this.contextPath = props.getProperty(ShenyuClientConstants.CONTEXT_PATH, "");        if (StringUtils.isBlank(appName) && StringUtils.isBlank(contextPath)) {            String errorMsg = "http register param must config the appName or contextPath";            LOG.error(errorMsg);            throw new ShenyuClientIllegalArgumentException(errorMsg);        }        this.isFull = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.IS_FULL, Boolean.FALSE.toString()));        // 2. add annotation        mappingAnnotation.add(ShenyuSpringMvcClient.class);        mappingAnnotation.add(PostMapping.class);        mappingAnnotation.add(GetMapping.class);        mappingAnnotation.add(DeleteMapping.class);        mappingAnnotation.add(PutMapping.class);        mappingAnnotation.add(RequestMapping.class);        // 3. start register cneter        publisher.start(shenyuClientRegisterRepository);    }        @Override    public Object postProcessAfterInitialization(@NonNull final Object bean, @NonNull final String beanName) throws BeansException {       // override post process                return bean;    }    

• SpringMvcClientBeanPostProcessor#postProcessAfterInitialization()

Rewrite post-processor logic: read annotation information, construct metadata objects and URI objects, and register them with shenyu-admin.

    @Override    public Object postProcessAfterInitialization(@NonNull final Object bean, @NonNull final String beanName) throws BeansException {        // 1. If the all service is registered or is not a Controller class, it is not handled        if (Boolean.TRUE.equals(isFull) || !hasAnnotation(bean.getClass(), Controller.class)) {            return bean;        }        // 2. Read the annotations on the class ShenyuSpringMvcClient        final ShenyuSpringMvcClient beanShenyuClient = AnnotationUtils.findAnnotation(bean.getClass(), ShenyuSpringMvcClient.class);        // 2.1 build  superPath        final String superPath = buildApiSuperPath(bean.getClass());        // 2.2 whether to register the entire class method        if (Objects.nonNull(beanShenyuClient) && superPath.contains("*")) {            // build the metadata object and register it with shenyu-admin            publisher.publishEvent(buildMetaDataDTO(beanShenyuClient, pathJoin(contextPath, superPath)));            return bean;        }        // 3. read all methods        final Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(bean.getClass());        for (Method method : methods) {            // 3.1 read the annotations on the method ShenyuSpringMvcClient            ShenyuSpringMvcClient methodShenyuClient = AnnotationUtils.findAnnotation(method, ShenyuSpringMvcClient.class);            // If there is no annotation on the method, use the annotation on the class            methodShenyuClient = Objects.isNull(methodShenyuClient) ? beanShenyuClient : methodShenyuClient;            if (Objects.nonNull(methodShenyuClient)) {               // 3.2 Build path information, build metadata objects, register with shenyu-admin                publisher.publishEvent(buildMetaDataDTO(methodShenyuClient, buildApiPath(method, superPath)));            }        }                return bean;    }

• 1. If you are registering the whole service or not Controller class, do not handle it
• 2. read the annotation on the class ShenyuSpringMvcClient, if the whole class is registered, build the metadata object here and register it with shenyu-admin.
• 3. Annotation on the handler method ShenyuSpringMvcClient, build path information for the specific method, build the metadata object and then register it with shenyu-admin

There are two methods here that take path and need special instructions.

• buildApiSuperPath()

Construct SuperPath: first take the path property from the annotation ShenyuSpringMvcClient on the class, if not, take the path information from the RequestMapping annotation on the current class.

    private String buildApiSuperPath(@NonNull final Class<?> method) {        // First take the path property from the annotation ShenyuSpringMvcClient on the class        ShenyuSpringMvcClient shenyuSpringMvcClient = AnnotationUtils.findAnnotation(method, ShenyuSpringMvcClient.class);        if (Objects.nonNull(shenyuSpringMvcClient) && StringUtils.isNotBlank(shenyuSpringMvcClient.path())) {            return shenyuSpringMvcClient.path();        }        // Take the path information from the RequestMapping annotation of the current class        RequestMapping requestMapping = AnnotationUtils.findAnnotation(method, RequestMapping.class);        if (Objects.nonNull(requestMapping) && ArrayUtils.isNotEmpty(requestMapping.path()) && StringUtils.isNotBlank(requestMapping.path()[0])) {            return requestMapping.path()[0];        }        return "";    }

• buildApiPath()

Build path: first read the annotation ShenyuSpringMvcClient on the method and build it if it exists; otherwise get the path information from other annotations on the method; complete path = contextPath(context information) + superPath(class information) + methodPath(method information).

    private String buildApiPath(@NonNull final Method method, @NonNull final String superPath) {        // 1. Read the annotation ShenyuSpringMvcClient on the method        ShenyuSpringMvcClient shenyuSpringMvcClient = AnnotationUtils.findAnnotation(method, ShenyuSpringMvcClient.class);        // 1.1 If path exists, build        if (Objects.nonNull(shenyuSpringMvcClient) && StringUtils.isNotBlank(shenyuSpringMvcClient.path())) {            //1.2  path = contextPath+superPath+methodPath            return pathJoin(contextPath, superPath, shenyuSpringMvcClient.path());        }        // 2. Get path information from other annotations on the method        final String path = getPathByMethod(method);        if (StringUtils.isNotBlank(path)) {             // 2.1 path = contextPath+superPath+methodPath            return pathJoin(contextPath, superPath, path);        }        return pathJoin(contextPath, superPath);    }

• getPathByMethod()

Get path information from other annotations on the method, other annotations include.

• ShenyuSpringMvcClient
• PostMapping
• GetMapping
• DeleteMapping
• PutMapping
• RequestMapping

    private String getPathByMethod(@NonNull final Method method) {        // Iterate through interface annotations to get path information        for (Class<? extends Annotation> mapping : mappingAnnotation) {            final String pathByAnnotation = getPathByAnnotation(AnnotationUtils.findAnnotation(method, mapping), pathAttributeNames);            if (StringUtils.isNotBlank(pathByAnnotation)) {                return pathByAnnotation;            }        }        return null;    }

After the scanning annotation is finished, construct the metadata object and send the object to shenyu-admin to complete the registration.

• Metadata

Includes the rule information of the currently registered method: contextPath, appName, registration path, description information, registration type, whether it is enabled, rule name and whether to register metadata.

 private MetaDataRegisterDTO buildMetaDataDTO(@NonNull final ShenyuSpringMvcClient shenyuSpringMvcClient, final String path) {        return MetaDataRegisterDTO.builder()                .contextPath(contextPath) // contextPath                .appName(appName) // appName                .path(path) // Registered path, used when gateway rules match                .pathDesc(shenyuSpringMvcClient.desc()) // desc info                .rpcType(RpcTypeEnum.HTTP.getName()) // divide plugin, http type when default                .enabled(shenyuSpringMvcClient.enabled()) // is enabled?                .ruleName(StringUtils.defaultIfBlank(shenyuSpringMvcClient.ruleName(), path))//rule name                .registerMetaData(shenyuSpringMvcClient.registerMetaData()) // whether to register metadata information                .build();    }

The specific registration logic is implemented by the registration center, which has been analyzed in the previous articles and will not be analyzed in depth here.

1.3 Register URI Data#

ContextRegisterListener is responsible for registering the client's URI information to shenyu-admin, it implements the ApplicationListener interface, when the context refresh event ContextRefreshedEvent occurs, the onApplicationEvent() method is executed to implement the registration logic.

public class ContextRegisterListener implements ApplicationListener<ContextRefreshedEvent>, BeanFactoryAware {    //......        /**     * Constructor instantiation     */    public ContextRegisterListener(final PropertiesConfig clientConfig) {        // read Properties        final Properties props = clientConfig.getProps();        this.isFull = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.IS_FULL, Boolean.FALSE.toString()));        this.contextPath = props.getProperty(ShenyuClientConstants.CONTEXT_PATH);        if (Boolean.TRUE.equals(isFull)) {            if (StringUtils.isBlank(contextPath)) {                final String errorMsg = "http register param must config the contextPath";                LOG.error(errorMsg);                throw new ShenyuClientIllegalArgumentException(errorMsg);            }        }        this.port = Integer.parseInt(Optional.ofNullable(props.getProperty(ShenyuClientConstants.PORT)).orElseGet(() -> "-1"));        this.appName = props.getProperty(ShenyuClientConstants.APP_NAME);        this.protocol = props.getProperty(ShenyuClientConstants.PROTOCOL, ShenyuClientConstants.HTTP);        this.host = props.getProperty(ShenyuClientConstants.HOST);    }
    @Override    public void setBeanFactory(final BeanFactory beanFactory) throws BeansException {        this.beanFactory = beanFactory;    }
    // Execute application events    @Override    public void onApplicationEvent(@NonNull final ContextRefreshedEvent contextRefreshedEvent) {          // The method is guaranteed to be executed once        if (!registered.compareAndSet(false, true)) {            return;        }        // 1. If you are registering for the entire service        if (Boolean.TRUE.equals(isFull)) {            // Build metadata and register            publisher.publishEvent(buildMetaDataDTO());        }        try {            // get port            final int mergedPort = port <= 0 ? PortUtils.findPort(beanFactory) : port;            // 2. Constructing URI data and registering            publisher.publishEvent(buildURIRegisterDTO(mergedPort));        } catch (ShenyuException e) {            throw new ShenyuException(e.getMessage() + "please config ${shenyu.client.http.props.port} in xml/yml !");        }    }
    // build URI data    private URIRegisterDTO buildURIRegisterDTO(final int port) {        return URIRegisterDTO.builder()            .contextPath(this.contextPath) // contextPath            .appName(appName) // appName            .protocol(protocol) // protocol            .host(IpUtils.isCompleteHost(this.host) ? this.host : IpUtils.getHost(this.host)) //host            .port(port) // port            .rpcType(RpcTypeEnum.HTTP.getName()) // divide plugin, default registration http type            .build();    }
    // build MetaData    private MetaDataRegisterDTO buildMetaDataDTO() {        return MetaDataRegisterDTO.builder()            .contextPath(contextPath)            .appName(appName)            .path(contextPath)            .rpcType(RpcTypeEnum.HTTP.getName())            .enabled(true)            .ruleName(contextPath)            .build();    }}

1.4 Handle registration information#

The metadata and URI data registered by the client through the registry are processed in shenyu-admin, which is responsible for storing to the database and synchronizing to the shenyu gateway. The client registration processing logic of Divide plugin is in ShenyuClientRegisterDivideServiceImpl. The inheritance relationship is as follows.

• ShenyuClientRegisterService: client registration service, top-level interface.
• FallbackShenyuClientRegisterService: registration failure, provides retry operation.
• AbstractShenyuClientRegisterServiceImpl: abstract class, implements part of the public registration logic;
• AbstractContextPathRegisterService: abstract class, responsible for registering ContextPath.
• ShenyuClientRegisterDivideServiceImpl: implementation of the Divide plug-in registration.

1.4.1 Register Service#

• org.apache.shenyu.admin.service.register.AbstractShenyuClientRegisterServiceImpl#register()

The metadata MetaDataRegisterDTO object registered by the client through the registry is picked up and dropped in the register() method of shenyu-admin.

   @Override    public String register(final MetaDataRegisterDTO dto) {        //1. register selector        String selectorHandler = selectorHandler(dto);        String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);        //2. register rule        String ruleHandler = ruleHandler();        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);        ruleService.registerDefault(ruleDTO);        //3. register metadat        registerMetadata(dto);        //4. register ContextPath        String contextPath = dto.getContextPath();        if (StringUtils.isNotEmpty(contextPath)) {            registerContextPath(dto);        }        return ShenyuResultMessage.SUCCESS;    }

1.4.1.1 Register Selector#

• org.apache.shenyu.admin.service.impl.SelectorServiceImpl#registerDefault()

Build contextPath, find if the selector information exists, if it does, return id; if it doesn't, create the default selector information.

    @Override    public String registerDefault(final MetaDataRegisterDTO dto, final String pluginName, final String selectorHandler) {        // build contextPath        String contextPath = ContextPathUtils.buildContextPath(dto.getContextPath(), dto.getAppName());        // Find if selector information exists by name        SelectorDO selectorDO = findByNameAndPluginName(contextPath, pluginName);        if (Objects.isNull(selectorDO)) {            // Create a default selector message if it does not exist            return registerSelector(contextPath, pluginName, selectorHandler);        }        return selectorDO.getId();    }