[dubbo-11297] Dynamic Routing Expansion

dubbo-cluster/pom.xml

@@ -40,6 +40,10 @@
             <groupId>org.yaml</groupId>
             <artifactId>snakeyaml</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-jexl3</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-framework</artifactId>

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Constants.java

@@ -120,6 +120,17 @@ public interface Constants {
      */
     String EXPORT_KEY = "export";
 
+
+    /**
+     * Specify the expression context builder in order for evaluating the expression
+     */
+    String CONTEXT_BUILDER_KEY = "context.builder";
+
+    /**
+     * The default strategy name for context builder
+     */
+    String DEFAULT_CONTEXT_BUILDER = "default";
+
     String PEER_KEY = "peer";
 
     String CONSUMER_URL_KEY = "CONSUMER_URL";

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/ExpressionRouter.java

@@ -0,0 +1,99 @@
+package org.apache.dubbo.rpc.cluster.router.expression;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.configcenter.ConfigChangeType;
+import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.ExtensionLoader;
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.utils.Holder;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.cluster.Constants;
+import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
+import org.apache.dubbo.rpc.cluster.router.expression.context.ContextBuilder;
+import org.apache.dubbo.rpc.cluster.router.expression.model.Rule;
+import org.apache.dubbo.rpc.cluster.router.expression.model.RuleSet;
+import org.apache.dubbo.rpc.cluster.router.expression.model.ExpressionRuleConstructor;
+
+import org.apache.commons.jexl3.JexlBuilder;
+import org.apache.commons.jexl3.JexlContext;
+import org.apache.commons.jexl3.JexlEngine;
+import org.apache.commons.jexl3.MapContext;
+import org.apache.dubbo.rpc.cluster.router.state.BitList;
+import org.yaml.snakeyaml.Yaml;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+public class ExpressionRouter<T> extends ObserverRouter<T> {
+
+    public static final String NAME = "expression";
+
+    private static final Logger logger = LoggerFactory.getLogger(ExpressionRouter.class);
+
+    /**
+     * Store the mapping relations of provider/ruleSet.
+     */
+    private static final Map<String, RuleSet> ruleSets = new ConcurrentHashMap<>();
+
+    private static final JexlEngine engine = new JexlBuilder().create();
+
+    private ContextBuilder contextBuilder;
+
+    public ExpressionRouter(URL url) {
+        super(url, url.getParameter(CommonConstants.APPLICATION_KEY));
+        contextBuilder = ExtensionLoader.getExtensionLoader(ContextBuilder.class)
+            .getExtension(url.getParameter(Constants.CONTEXT_BUILDER_KEY, Constants.DEFAULT_CONTEXT_BUILDER));
+    }
+
+    @Override
+    protected BitList<Invoker<T>> doRoute(BitList<Invoker<T>> invokers, URL url, Invocation invocation, boolean needToPrintMessage, Holder<RouterSnapshotNode<T>> nodeHolder, Holder<String> messageHolder) throws RpcException {
+        String application = url.getParameter(CommonConstants.REMOTE_APPLICATION_KEY);
+        if(application == null){
+            return invokers;
+        }
+        RuleSet ruleSet = ruleSets.get(application);
+        if (logger.isTraceEnabled()) {
+            logger.trace(ruleSet.toString());
+        }
+        if (ruleSet != null && ruleSet.isEnabled()) {
+            JexlContext clientContext = new MapContext();
+            contextBuilder.buildClientContext(url, invocation).forEach(clientContext::set);
+            for (Rule rule : ruleSet.getRules()) {
+                Object clientQualified = engine.createExpression(rule.getClientCondition()).evaluate(clientContext);
+                if (clientQualified instanceof Boolean && (Boolean) clientQualified) {
+                    List<Invoker<T>> result = invokers
+                        .stream()
+                        .filter(invoker -> matches(contextBuilder.buildServerContext(invoker, url, invocation), rule.getServerQuery()))
+                        .collect(Collectors.toList());
+                    if (CollectionUtils.isNotEmpty(result)) {
+                        return result;
+                    }
+                }
+            }
+            if (ruleSet.isDefaultRuleEnabled()) {
+                return invokers;
+            } else {
+                return new BitList<Invoker<T>>(new ArrayList<>());
+            }
+        }
+        return invokers;
+    }
+
+    public boolean matches(Map<String, Object> objects, String expression) {
+        JexlContext context = new MapContext();
+        objects.forEach(context::set);
+        Object qualified = engine.createExpression(expression).evaluate(context);
+        return qualified instanceof Boolean && (Boolean) qualified;
+    }
+
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/ExpressionRouterFactory.java

@@ -0,0 +1,19 @@
+package org.apache.dubbo.rpc.cluster.router.expression;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.rpc.cluster.Router;
+import org.apache.dubbo.rpc.cluster.RouterFactory;
+import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
+import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
+
+@Activate
+public class ExpressionRouterFactory implements StateRouterFactory {
+
+    public static final String NAME = "expression";
+
+    @Override
+    public <T> StateRouter<T> getRouter(Class<T> interfaceClass, URL url) {
+        return new ExpressionRouter(url);
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/ObserverRouter.java

@@ -0,0 +1,29 @@
+package org.apache.dubbo.rpc.cluster.router.expression;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
+import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
+import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
+import org.apache.dubbo.common.utils.StringUtils;
+import org.apache.dubbo.rpc.cluster.router.state.AbstractStateRouter;
+
+public abstract class ObserverRouter<T> extends AbstractStateRouter<T> implements ConfigurationListener {
+    public static final String NAME = "OBSERVER_ROUTER";
+    private static final String RULE_SUFFIX = ".observer-router";
+
+    public ObserverRouter(URL url, String ruleKey) {
+        super(url);
+        this.init(ruleKey);
+    }
+
+    private synchronized void init(String ruleKey) {
+        if (StringUtils.isNotEmpty(ruleKey)) {
+            String routerKey = ruleKey + RULE_SUFFIX;
+            ruleRepository.addListener(routerKey, this);
+            String rule = ruleRepository.getRule(routerKey, DynamicConfiguration.DEFAULT_GROUP);
+            if (StringUtils.isNotEmpty(rule)) {
+                this.process(new ConfigChangedEvent(routerKey, DynamicConfiguration.DEFAULT_GROUP, rule));
+            }
+        }
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/context/ContextBuilder.java

@@ -0,0 +1,18 @@
+package org.apache.dubbo.rpc.cluster.router.expression.context;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.SPI;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+@SPI
+public interface ContextBuilder {
+
+    Map<String, Object> buildClientContext(URL url, Invocation invocation);
+
+    <T> Map<String, Object> buildServerContext(Invoker<T> invoker, URL url, Invocation invocation);
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/context/DefaultContextBuilder.java

@@ -0,0 +1,44 @@
+package org.apache.dubbo.rpc.cluster.router.expression.context;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.extension.Activate;
+import org.apache.dubbo.common.utils.CollectionUtils;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The default context builder used for evaluating the expressions.
+ */
+@Activate
+public class DefaultContextBuilder implements ContextBuilder {
+
+    /**
+     * The object name of Client-Side
+     */
+    private static final String CLIENT_NAME = "c";
+    /**
+     * The object name of Request
+     */
+    private static final String REQUEST_NAME = "r";
+    /**
+     * The object name of Server-Side
+     */
+    private static final String SERVER_NAME = "s";
+
+    @Override
+    public Map<String, Object> buildClientContext(URL url, Invocation invocation) {
+        return CollectionUtils.toMap(REQUEST_NAME, invocation.getAttachments(), CLIENT_NAME, url.getParameters());
+    }
+
+    @Override
+    public <T> Map<String, Object> buildServerContext(Invoker<T> invoker, URL url, Invocation invocation) {
+        Map<String, Object> params = new HashMap<>(invoker.getUrl().getParameters());
+        params.put("port", invoker.getUrl().getPort());
+        params.put("address", invoker.getUrl().getAddress());
+        return Collections.singletonMap(SERVER_NAME, params);
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/model/ExpressionRuleConstructor.java

@@ -0,0 +1,43 @@
+package org.apache.dubbo.rpc.cluster.router.expression.model;
+
+import org.yaml.snakeyaml.TypeDescription;
+import org.yaml.snakeyaml.constructor.Constructor;
+import org.yaml.snakeyaml.nodes.MappingNode;
+import org.yaml.snakeyaml.nodes.Node;
+import org.yaml.snakeyaml.nodes.Tag;
+
+import java.util.stream.Collectors;
+
+/**
+ * A yaml constructor for parsing RuleSets which should be a map.
+ */
+public class ExpressionRuleConstructor extends Constructor {
+
+    private TypeDescription itemType = new TypeDescription(RuleSet.class);
+
+    private static final String ROOT_NAME = "ruleSetRoot";
+
+    public ExpressionRuleConstructor() {
+        this.rootTag = new Tag(ROOT_NAME);
+        this.addTypeDescription(itemType);
+    }
+
+    @Override
+    protected Object constructObject(Node node) {
+        if (ROOT_NAME.equals(node.getTag().getValue()) && node instanceof MappingNode) {
+            MappingNode mNode = (MappingNode) node;
+            return mNode.getValue().stream().collect(
+                Collectors.toMap(
+                    t -> super.constructObject(t.getKeyNode()),
+                    t -> {
+                        Node child = t.getValueNode();
+                        child.setType(itemType.getType());
+                        return super.constructObject(child);
+                    }
+                )
+            );
+        } else {
+            return super.constructObject(node);
+        }
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/model/Rule.java

@@ -0,0 +1,39 @@
+package org.apache.dubbo.rpc.cluster.router.expression.model;
+
+/**
+ * A single rule which client prerequisite and server filter.
+ */
+public class Rule {
+
+    /**
+     * Somewhat like whenCondition in ConditionRouter.
+     * This is acted on client and the result should be true/false after evaluation.
+     */
+    private String clientCondition;
+    /**
+     * Somewhat like thenCondition in ConditionRouter.
+     * This is acted on server and the result should be server list after evaluation.
+     */
+    private String serverQuery;
+
+    public String getClientCondition() {
+        return clientCondition;
+    }
+
+    public void setClientCondition(String clientCondition) {
+        this.clientCondition = clientCondition;
+    }
+
+    public String getServerQuery() {
+        return serverQuery;
+    }
+
+    public void setServerQuery(String serverQuery) {
+        this.serverQuery = serverQuery;
+    }
+
+    public String toString(){
+        return "Rule(clientCondition=" + clientCondition
+            + ", serverQuery=" + serverQuery + ")";
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/expression/model/RuleSet.java

@@ -0,0 +1,56 @@
+package org.apache.dubbo.rpc.cluster.router.expression.model;
+
+import java.util.List;
+
+public class RuleSet {
+
+    /**
+     * Whether the ruleSet is enabled or not, set true as its default value.
+     */
+    private boolean enabled = true;
+
+    /**
+     * Whether default rule is enabled, set false as its default value.
+     * This is useful when none of the provider is found after evaluating all the rules.
+     * If this is set to false, exception of no provider will be thrown.
+     * If this is set to true, all the left providers will be chosen, just like the rule of following:
+     * clientCondition: true
+     * serverQuery: true
+     */
+    private boolean defaultRuleEnabled;
+
+    /**
+     * The rules are in order. The top one will be evaluated in top priority.
+     */
+    private List<Rule> rules;
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public boolean isDefaultRuleEnabled() {
+        return defaultRuleEnabled;
+    }
+
+    public void setDefaultRuleEnabled(boolean defaultRuleEnabled) {
+        this.defaultRuleEnabled = defaultRuleEnabled;
+    }
+
+    public List<Rule> getRules() {
+        return rules;
+    }
+
+    public void setRules(List<Rule> rules) {
+        this.rules = rules;
+    }
+
+    public String toString(){
+        return "RuleSet(enabled=" + enabled
+            + ", defaultRuleEnabled=" + defaultRuleEnabled
+            + ",rules=" + rules + ")";
+    }
+}

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/state/AbstractStateRouter.java

@@ -36,7 +36,7 @@ public abstract class AbstractStateRouter<T> implements StateRouter<T> {
     private volatile URL url;
     private volatile StateRouter<T> nextRouter = null;
 
-    private final GovernanceRuleRepository ruleRepository;
+    protected final GovernanceRuleRepository ruleRepository;
 
     /**
      * Should continue route if current router's result is empty

dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.expression.context.ContextBuilder

@@ -0,0 +1 @@
+default=org.apache.dubbo.rpc.cluster.router.expression.context.DefaultContextBuilder

dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory

@@ -4,3 +4,4 @@ service=org.apache.dubbo.rpc.cluster.router.condition.config.ServiceStateRouterF
 app=org.apache.dubbo.rpc.cluster.router.condition.config.AppStateRouterFactory
 mock=org.apache.dubbo.rpc.cluster.router.mock.MockStateRouterFactory
 standard-mesh-rule=org.apache.dubbo.rpc.cluster.router.mesh.route.StandardMeshRuleRouterFactory
+expression=org.apache.dubbo.rpc.cluster.router.expression.ExpressionRouterFactory