feat:support polaris circuitbreaker ability (#212)

* feat:support polaris circuitbreaker ability * feat:support polaris circuitbreaker ability * feat:support polaris circuitbreaker ability * feat:support polaris circuitbreaker ability * feat:support polaris circuitbreaker ability * feat:support polaris circuitbreaker ability
apache · Jul 13, 2023 · 644960b · 644960b
1 parent 3c49199
commit 644960b
Show file tree

Hide file tree

Showing 14 changed files with 375 additions and 271 deletions.
diff --git a/...ter-polaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/InstanceInvoker.java b/...ter-polaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/InstanceInvoker.java
@@ -22,20 +22,22 @@
 import com.tencent.polaris.api.pojo.Instance;
 import com.tencent.polaris.api.pojo.StatusDimension;
 import com.tencent.polaris.common.registry.Consts;
-import com.tencent.polaris.common.registry.ConvertUtils;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Objects;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.constants.CommonConstants;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.StringUtils;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.cluster.Constants;
 
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
 public class InstanceInvoker<T> implements Instance, Invoker<T> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(InstanceInvoker.class);
@@ -57,14 +59,22 @@ public InstanceInvoker(Invoker<T> invoker, String namespace) {
         defaultInstance.setIsolated(Boolean.parseBoolean(url.getParameter(Consts.INSTANCE_KEY_ISOLATED)));
         defaultInstance.setVersion(url.getParameter(CommonConstants.VERSION_KEY));
         defaultInstance.setWeight(url.getParameter(Constants.WEIGHT_KEY, 100));
-        String circuitBreakerStr = url.getParameter(Consts.INSTANCE_KEY_CIRCUIT_BREAKER);
-        Map<StatusDimension, CircuitBreakerStatus> statusDimensionCircuitBreakerStatusMap = ConvertUtils
-                .stringToCircuitBreakers(circuitBreakerStr);
-        defaultInstance.getCircuitBreakerStatuses().putAll(statusDimensionCircuitBreakerStatusMap);
-        defaultInstance.setMetadata(url.getParameters());
+        defaultInstance.setMetadata(convertMetadata(url.getParameters()));
         LOGGER.info(String.format("[POLARIS] construct instance from invoker, url %s, instance %s", url, defaultInstance));
     }
 
+    private Map<String, String> convertMetadata(Map<String, String> parameters) {
+        Map<String, String> ret = new HashMap<>();
+        parameters.forEach((key, value) -> {
+            ret.put(key, value);
+            if (StringUtils.isEquals(key, CommonConstants.REMOTE_APPLICATION_KEY)) {
+                key = "application";
+                ret.put(key, value);
+            }
+        });
+        return ret;
+    }
+
     @Override
     public Class<T> getInterface() {
         return invoker.getInterface();
@@ -200,10 +210,6 @@ public int compareTo(Instance o) {
         return defaultInstance.compareTo(o);
     }
 
-    private static final String SEP_CIRCUIT_BREAKER = ",";
-
-    private static final String SEP_CIRCUIT_BREAKER_VALUE = ":";
-
     @Override
     public boolean equals(Object o) {
         if (this == o) {

diff --git a/...uster-polaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/PolarisRouter.java b/...uster-polaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/PolarisRouter.java
@@ -22,19 +22,11 @@
 import com.tencent.polaris.api.pojo.ServiceEventKey.EventType;
 import com.tencent.polaris.api.pojo.ServiceRule;
 import com.tencent.polaris.api.utils.StringUtils;
-import com.tencent.polaris.client.pb.RoutingProto.Routing;
+import com.tencent.polaris.common.parser.QueryParser;
 import com.tencent.polaris.common.registry.PolarisOperator;
 import com.tencent.polaris.common.registry.PolarisOperators;
-import com.tencent.polaris.common.router.ObjectParser;
 import com.tencent.polaris.common.router.RuleHandler;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicReference;
+import com.tencent.polaris.specification.api.v1.traffic.manage.RoutingProto;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.Logger;
 import org.apache.dubbo.common.logger.LoggerFactory;
@@ -44,6 +36,12 @@
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.cluster.Constants;
 
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
 public class PolarisRouter extends AbstractRouter {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(PolarisRouter.class);
@@ -52,15 +50,17 @@ public class PolarisRouter extends AbstractRouter {
 
     private final PolarisOperator polarisOperator;
 
-    private final AtomicReference<Map<URL, InstanceInvoker<?>>> invokersCache = new AtomicReference<>();
+    private final QueryParser parser;
 
     public PolarisRouter(URL url) {
         super(url);
         LOGGER.info(String.format("[POLARIS] init service router, url is %s, parameters are %s", url,
-                url.getParameters()));
+            url.getParameters()));
+        System.setProperty("dubbo.polaris.query_parser", System.getProperty("dubbo.polaris.query_parser", "JsonPath"));
         this.priority = url.getParameter(Constants.PRIORITY_KEY, 0);
-        routeRuleHandler = new RuleHandler();
-        polarisOperator = PolarisOperators.INSTANCE.getPolarisOperator(url.getHost(), url.getPort());
+        this.routeRuleHandler = new RuleHandler();
+        this.polarisOperator = PolarisOperators.INSTANCE.getPolarisOperator(url.getHost(), url.getPort());
+        this.parser = QueryParser.load();
     }
 
     @Override
@@ -71,25 +71,22 @@ public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation
         if (null == polarisOperator) {
             return invokers;
         }
-        List<Instance> instances = new ArrayList<>(invokers.size());
-        Map<URL, InstanceInvoker<?>> instanceInvokerMap = invokersCache.get();
-        if (null == instanceInvokerMap) {
-            instanceInvokerMap = Collections.emptyMap();
-        }
-        for (Invoker<T> invoker : invokers) {
-                InstanceInvoker<?> instanceInvoker = instanceInvokerMap.get(invoker.getUrl());
-                if (null != instanceInvoker) {
-                    instances.add(instanceInvoker);
-                } else {
-                    instances.add(new InstanceInvoker<>(invoker, polarisOperator.getPolarisConfig().getNamespace()));
-                }
+        List<Instance> instances;
+        if (invokers.get(0) instanceof Instance) {
+            instances = (List<Instance>) ((List<?>) invokers);
+        } else {
+            instances = new ArrayList<>();
+            for (Invoker<T> invoker : invokers) {
+                instances.add(new InstanceInvoker<>(invoker, polarisOperator.getPolarisConfig().getNamespace()));
+            }
         }
+
         String service = url.getServiceInterface();
         ServiceRule serviceRule = polarisOperator.getServiceRule(service, EventType.ROUTING);
         Object ruleObject = serviceRule.getRule();
         Set<RouteArgument> arguments = new HashSet<>();
         if (null != ruleObject) {
-            Routing routing = (Routing) ruleObject;
+            RoutingProto.Routing routing = (RoutingProto.Routing) ruleObject;
             Set<String> routeLabels = routeRuleHandler.getRouteLabels(routing);
             for (String routeLabel : routeLabels) {
                 if (StringUtils.equals(RouteArgument.LABEL_KEY_PATH, routeLabel)) {
@@ -103,29 +100,17 @@ public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation
                 } else if (routeLabel.startsWith(RouteArgument.LABEL_KEY_QUERY)) {
                     String queryName = routeLabel.substring(RouteArgument.LABEL_KEY_QUERY.length());
                     if (!StringUtils.isBlank(queryName)) {
-                        Object value = ObjectParser.parseArgumentsByExpression(queryName, invocation.getArguments());
-                        if (null != value) {
-                            arguments.add(RouteArgument.buildQuery(queryName, String.valueOf(value)));
-                        }
+                        Optional<String> queryValue = parser.parse(queryName, invocation.getArguments());
+                        queryValue.ifPresent(value -> arguments.add(RouteArgument.buildQuery(queryName, value)));
                     }
                 }
             }
         }
-        LOGGER.debug(String.format("[POLARIS] list service %s, method %s, labels %s, url %s", service,
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug(String.format("[POLARIS] list service %s, method %s, labels %s, url %s", service,
                 invocation.getMethodName(), arguments, url));
-        List<Instance> resultInstances = polarisOperator
-                .route(service, invocation.getMethodName(), arguments, instances);
-        return (List<Invoker<T>>) ((List<?>) resultInstances);
-    }
-
-    public <T> void notify(List<Invoker<T>> invokers) {
-        if (null == polarisOperator) {
-            return;
         }
-        Map<URL, InstanceInvoker<?>> instanceInvokers = new HashMap<>(invokers.size());
-        for (Invoker<T> invoker : invokers) {
-            instanceInvokers.put(invoker.getUrl(), new InstanceInvoker<>(invoker, polarisOperator.getPolarisConfig().getNamespace()));
-        }
-        invokersCache.set(instanceInvokers);
+        List<Instance> resultInstances = polarisOperator.route(service, invocation.getMethodName(), arguments, instances);
+        return (List<Invoker<T>>) ((List<?>) resultInstances);
     }
 }
diff --git a/...olaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/PolarisRouterFactory.java b/...olaris-dubbo2/src/main/java/org/apache/dubbo/rpc/cluster/router/PolarisRouterFactory.java
@@ -18,13 +18,17 @@
 package org.apache.dubbo.rpc.cluster.router;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.common.extension.Activate;
 import org.apache.dubbo.rpc.cluster.Router;
 import org.apache.dubbo.rpc.cluster.RouterFactory;
 
+@Activate(group = CommonConstants.CONSUMER)
 public class PolarisRouterFactory implements RouterFactory {
 
     @Override
     public Router getRouter(URL url) {
         return new PolarisRouter(url);
     }
 }
+
diff --git a/dubbo-cluster-extensions/pom.xml b/dubbo-cluster-extensions/pom.xml
@@ -39,5 +39,4 @@
         <module>dubbo-cluster-polaris-dubbo2</module>
     </modules>
 
-
 </project>
diff --git a/dubbo-extensions-dependencies-bom/pom.xml b/dubbo-extensions-dependencies-bom/pom.xml
@@ -133,8 +133,7 @@
         <logback_version>1.2.11</logback_version>
         <rs_api_version>2.0</rs_api_version>
         <resteasy_version>3.0.20.Final</resteasy_version>
-        <polaris_adapter_version>0.1.3</polaris_adapter_version>
-
+        <polaris_adapter_version>0.2.1</polaris_adapter_version>
         <maven_flatten_version>1.2.5</maven_flatten_version>
     </properties>
 

diff --git a/...circuitbreaker-dubbo2/src/main/java/org/apache/dubbo/filter/dubbo2/CallAbortCallback.java b/...circuitbreaker-dubbo2/src/main/java/org/apache/dubbo/filter/dubbo2/CallAbortCallback.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dubbo.filter.dubbo2;
+
+import com.tencent.polaris.circuitbreak.client.exception.CallAbortedException;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+
+public interface CallAbortCallback {
+
+    Result handle(Invoker<?> invoker, Invocation invocation, CallAbortedException ex);
+
+}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -39,5 +39,4 @@
		<module>dubbo-cluster-polaris-dubbo2</module>
		</modules>


		</project>