You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by ma...@apache.org on 2025/06/25 04:24:36 UTC
(dubbo-go) branch revert-2920-delete_polaris created (now 1c3668e94)
This is an automated email from the ASF dual-hosted git repository.
marsevilspirit pushed a change to branch revert-2920-delete_polaris
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
at 1c3668e94 Revert "delete all polaris (#2920)"
This branch includes the following new commits:
new 1c3668e94 Revert "delete all polaris (#2920)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(dubbo-go) 01/01: Revert "delete all polaris (#2920)"
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
marsevilspirit pushed a commit to branch revert-2920-delete_polaris
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
commit 1c3668e94602b9e1ece8fb35dba92a57a182c1fe
Author: marsevilspirit <ma...@gmail.com>
AuthorDate: Wed Jun 25 12:24:28 2025 +0800
Revert "delete all polaris (#2920)"
This reverts commit 941fde77c099f99728baa4aa17a1124decaf6753.
---
README.md | 2 +-
cluster/router/polaris/default.go | 27 +++
cluster/router/polaris/factory.go | 35 ++++
cluster/router/polaris/router.go | 331 +++++++++++++++++++++++++++++++
common/constant/polaris_key.go | 42 ++++
filter/filter_impl/import.go | 1 +
filter/polaris/limit/default.go | 30 +++
filter/polaris/limit/limiter.go | 173 ++++++++++++++++
go.mod | 5 +
go.sum | 24 +++
imports/imports.go | 3 +
registry/options.go | 6 +
registry/polaris/core.go | 131 +++++++++++++
registry/polaris/doc.go | 19 ++
registry/polaris/listener.go | 123 ++++++++++++
registry/polaris/registry.go | 287 +++++++++++++++++++++++++++
registry/polaris/service_discovery.go | 348 +++++++++++++++++++++++++++++++++
registry/polaris/utils.go | 45 +++++
remoting/polaris/builder.go | 160 +++++++++++++++
remoting/polaris/builder_test.go | 45 +++++
remoting/polaris/parser/parser.go | 125 ++++++++++++
remoting/polaris/parser/parser_test.go | 168 ++++++++++++++++
remoting/polaris/polaris.yaml | 84 ++++++++
tools/dubbo-go-schema/dubbo-go.json | 1 +
24 files changed, 2214 insertions(+), 1 deletion(-)
diff --git a/README.md b/README.md
index 4a448da6c..c2966f506 100644
--- a/README.md
+++ b/README.md
@@ -76,7 +76,7 @@ See the [samples](https://github.com/apache/dubbo-go-samples) for detailed infor
- **RPC Protocols**: Triple, gRPC compatible and HTTP-friendly
-- **Service Discovery**: Nacos, Zookeeper, Etcd, Consul.
+- **Service Discovery**: Nacos, Zookeeper, Etcd, Polaris-mesh, Consul.
- **Load Balance**: Adaptive, Random, RoundRobin, LeastActive, ConsistentHash
- **Traffic Management**: traffic split, timeout, rate limiting, canary release
- **Configuration**: yaml file, dynamic configuration(Nacos, Zookeeper, etc.).
diff --git a/cluster/router/polaris/default.go b/cluster/router/polaris/default.go
new file mode 100644
index 000000000..2797a648b
--- /dev/null
+++ b/cluster/router/polaris/default.go
@@ -0,0 +1,27 @@
+/*
+ * 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 polaris
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+)
+
+func init() {
+ extension.SetRouterFactory(constant.PluginPolarisRouterFactory, NewPolarisRouterFactory)
+}
diff --git a/cluster/router/polaris/factory.go b/cluster/router/polaris/factory.go
new file mode 100644
index 000000000..76d9cae36
--- /dev/null
+++ b/cluster/router/polaris/factory.go
@@ -0,0 +1,35 @@
+/*
+ * 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 polaris
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
+)
+
+// RouteFactory router factory
+type RouteFactory struct{}
+
+// NewPolarisRouterFactory constructs a new PriorityRouterFactory
+func NewPolarisRouterFactory() router.PriorityRouterFactory {
+ return &RouteFactory{}
+}
+
+// NewPriorityRouter construct a new PriorityRouter
+func (f *RouteFactory) NewPriorityRouter() (router.PriorityRouter, error) {
+ return newPolarisRouter()
+}
diff --git a/cluster/router/polaris/router.go b/cluster/router/polaris/router.go
new file mode 100644
index 000000000..6bcb1f9ce
--- /dev/null
+++ b/cluster/router/polaris/router.go
@@ -0,0 +1,331 @@
+/*
+ * 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 polaris
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ "github.com/polarismesh/polaris-go"
+ "github.com/polarismesh/polaris-go/pkg/model"
+ v1 "github.com/polarismesh/polaris-go/pkg/model/pb/v1"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/router"
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/config"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ remotingpolaris "dubbo.apache.org/dubbo-go/v3/remoting/polaris"
+ "dubbo.apache.org/dubbo-go/v3/remoting/polaris/parser"
+)
+
+var (
+ _ router.PriorityRouter = (*polarisRouter)(nil)
+)
+
+var (
+ ErrorPolarisServiceRouteRuleEmpty = errors.New("service route rule is empty")
+)
+
+func newPolarisRouter() (*polarisRouter, error) {
+ if err := remotingpolaris.Check(); errors.Is(err, remotingpolaris.ErrorNoOpenPolarisAbility) {
+ return &polarisRouter{
+ openRoute: false,
+ }, nil
+ }
+
+ routerAPI, err := remotingpolaris.GetRouterAPI()
+ if err != nil {
+ return nil, err
+ }
+ consumerAPI, err := remotingpolaris.GetConsumerAPI()
+ if err != nil {
+ return nil, err
+ }
+
+ return &polarisRouter{
+ openRoute: true,
+ routerAPI: routerAPI,
+ consumerAPI: consumerAPI,
+ }, nil
+}
+
+type polarisRouter struct {
+ openRoute bool
+
+ routerAPI polaris.RouterAPI
+ consumerAPI polaris.ConsumerAPI
+
+ cancel context.CancelFunc
+
+ lock sync.RWMutex
+ instances map[string]model.Instance
+}
+
+// Route Determine the target invokers list.
+func (p *polarisRouter) Route(invokers []base.Invoker, url *common.URL,
+ invoaction base.Invocation) []base.Invoker {
+
+ if !p.openRoute {
+ logger.Debug("[Router][Polaris] not open polaris route ability")
+ return invokers
+ }
+
+ if len(invokers) == 0 {
+ logger.Warn("[Router][Polaris] invokers from previous router is empty")
+ return invokers
+ }
+
+ service := getService(url)
+ instanceMap := p.buildInstanceMap(service)
+ if len(instanceMap) == 0 {
+ return invokers
+ }
+
+ invokersMap := make(map[string]base.Invoker, len(invokers))
+ targetIns := make([]model.Instance, 0, len(invokers))
+ for i := range invokers {
+ invoker := invokers[i]
+ instanceID := invoker.GetURL().GetParam(constant.PolarisInstanceID, "")
+ if len(instanceID) == 0 {
+ continue
+ }
+ invokersMap[instanceID] = invoker
+ if val, ok := instanceMap[instanceID]; ok {
+ targetIns = append(targetIns, val)
+ }
+ }
+
+ req, err := p.buildRouteRequest(service, url, invoaction)
+ if err != nil {
+ return invokers
+ }
+ req.DstInstances = model.NewDefaultServiceInstances(model.ServiceInfo{
+ Service: service,
+ Namespace: remotingpolaris.GetNamespace(),
+ }, targetIns)
+
+ resp, err := p.routerAPI.ProcessRouters(&req)
+ if err != nil {
+ return invokers
+ }
+
+ ret := make([]base.Invoker, 0, len(resp.GetInstances()))
+ for i := range resp.GetInstances() {
+ if val, ok := invokersMap[resp.GetInstances()[i].GetId()]; ok {
+ ret = append(ret, val)
+ }
+ }
+
+ return ret
+}
+
+func getService(url *common.URL) string {
+ applicationMode := false
+ for _, item := range config.GetRootConfig().Registries {
+ if item.Protocol == constant.PolarisKey {
+ applicationMode = item.RegistryType == constant.ServiceKey
+ }
+ }
+
+ service := url.Interface()
+ if applicationMode {
+ service = config.GetApplicationConfig().Name
+ }
+
+ return service
+}
+
+func (p *polarisRouter) buildRouteRequest(svc string, url *common.URL,
+ invocation base.Invocation) (polaris.ProcessRoutersRequest, error) {
+
+ routeReq := polaris.ProcessRoutersRequest{
+ ProcessRoutersRequest: model.ProcessRoutersRequest{
+ SourceService: model.ServiceInfo{
+ Metadata: map[string]string{},
+ },
+ },
+ }
+
+ attachement := invocation.Attachments()
+ arguments := invocation.Arguments()
+
+ labels, err := p.buildTrafficLabels(svc)
+ if err != nil {
+ return polaris.ProcessRoutersRequest{}, err
+ }
+
+ for i := range labels {
+ label := labels[i]
+ if strings.Compare(label, model.LabelKeyPath) == 0 {
+ routeReq.AddArguments(model.BuildPathArgument(getInvokeMethod(url, invocation)))
+ continue
+ }
+ if strings.HasPrefix(label, model.LabelKeyHeader) {
+ if val, ok := attachement[strings.TrimPrefix(label, model.LabelKeyHeader)]; ok {
+ routeReq.SourceService.Metadata[label] = fmt.Sprintf("%+v", val)
+ routeReq.AddArguments(model.BuildArgumentFromLabel(label, fmt.Sprintf("%+v", val)))
+ }
+ }
+ if strings.HasPrefix(label, model.LabelKeyQuery) {
+ if val := parser.ParseArgumentsByExpression(label, arguments); val != nil {
+ routeReq.AddArguments(model.BuildArgumentFromLabel(label, fmt.Sprintf("%+v", val)))
+ }
+ }
+ }
+
+ return routeReq, nil
+}
+
+func (p *polarisRouter) buildTrafficLabels(svc string) ([]string, error) {
+ req := &model.GetServiceRuleRequest{}
+ req.Namespace = remotingpolaris.GetNamespace()
+ req.Service = svc
+ req.SetTimeout(time.Second)
+ engine := p.routerAPI.SDKContext().GetEngine()
+ resp, err := engine.SyncGetServiceRule(model.EventRouting, req)
+ if err != nil {
+ logger.Errorf("[Router][Polaris] ns:%s svc:%s get route rule fail : %+v", req.GetNamespace(), req.GetService(), err)
+ return nil, err
+ }
+
+ if resp == nil || resp.GetValue() == nil {
+ logger.Errorf("[Router][Polaris] ns:%s svc:%s get route rule empty", req.GetNamespace(), req.GetService())
+ return nil, ErrorPolarisServiceRouteRuleEmpty
+ }
+
+ routeRule := resp.GetValue().(*v1.Routing)
+ labels := make([]string, 0, 4)
+ labels = append(labels, collectRouteLabels(routeRule.GetInbounds())...)
+ labels = append(labels, collectRouteLabels(routeRule.GetInbounds())...)
+
+ return labels, nil
+}
+
+func getInvokeMethod(url *common.URL, invoaction base.Invocation) string {
+ applicationMode := false
+ for _, item := range config.GetRootConfig().Registries {
+ if item.Protocol == constant.PolarisKey {
+ applicationMode = item.RegistryType == constant.ServiceKey
+ }
+ }
+
+ method := invoaction.MethodName()
+ if applicationMode {
+ method = url.Interface() + "/" + invoaction.MethodName()
+ }
+
+ return method
+}
+
+func collectRouteLabels(routings []*v1.Route) []string {
+ ret := make([]string, 0, 4)
+
+ for i := range routings {
+ route := routings[i]
+ sources := route.GetSources()
+ for p := range sources {
+ source := sources[p]
+ for k := range source.GetMetadata() {
+ ret = append(ret, k)
+ }
+ }
+ }
+
+ return ret
+}
+
+func (p *polarisRouter) buildInstanceMap(svc string) map[string]model.Instance {
+ resp, err := p.consumerAPI.GetAllInstances(&polaris.GetAllInstancesRequest{
+ GetAllInstancesRequest: model.GetAllInstancesRequest{
+ Service: svc,
+ Namespace: remotingpolaris.GetNamespace(),
+ },
+ })
+ if err != nil {
+ logger.Errorf("[Router][Polaris] ns:%s svc:%s get all instances fail : %+v", remotingpolaris.GetNamespace(), svc, err)
+ return nil
+ }
+
+ ret := make(map[string]model.Instance, len(resp.GetInstances()))
+
+ for i := range resp.GetInstances() {
+ ret[resp.GetInstances()[i].GetId()] = resp.GetInstances()[i]
+ }
+
+ return ret
+}
+
+// URL Return URL in router
+func (p *polarisRouter) URL() *common.URL {
+ return nil
+}
+
+// Priority Return Priority in router
+// 0 to ^int(0) is better
+func (p *polarisRouter) Priority() int64 {
+ return 0
+}
+
+// Notify the router the invoker list
+func (p *polarisRouter) Notify(invokers []base.Invoker) {
+ if !p.openRoute {
+ return
+ }
+ if len(invokers) == 0 {
+ return
+ }
+ service := getService(invokers[0].GetURL())
+ if service == "" {
+ logger.Error("url service is empty")
+ return
+ }
+
+ req := &model.GetServiceRuleRequest{}
+ req.Namespace = remotingpolaris.GetNamespace()
+ req.Service = service
+ req.SetTimeout(time.Second)
+
+ engine := p.routerAPI.SDKContext().GetEngine()
+ _, err := engine.SyncGetServiceRule(model.EventRouting, req)
+ if err != nil {
+ logger.Errorf("[Router][Polaris] ns:%s svc:%s get route rule fail : %+v", req.GetNamespace(), req.GetService(), err)
+ return
+ }
+
+ _, err = p.consumerAPI.GetAllInstances(&polaris.GetAllInstancesRequest{
+ GetAllInstancesRequest: model.GetAllInstancesRequest{
+ Service: service,
+ Namespace: remotingpolaris.GetNamespace(),
+ },
+ })
+ if err != nil {
+ logger.Errorf("[Router][Polaris] ns:%s svc:%s get all instances fail : %+v", req.GetNamespace(), req.GetService(), err)
+ return
+ }
+}
diff --git a/common/constant/polaris_key.go b/common/constant/polaris_key.go
new file mode 100644
index 000000000..ea7d0fa23
--- /dev/null
+++ b/common/constant/polaris_key.go
@@ -0,0 +1,42 @@
+/*
+ * 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 constant
+
+const (
+ PolarisKey = "polaris"
+ PolarisDefaultRoleType = 3
+ PolarisServiceToken = "token"
+ PolarisServiceNameSeparator = ":"
+ PolarisDubboPath = "DUBBOPATH"
+ PolarisInstanceID = "polaris.instanceID"
+ PolarisDefaultNamespace = "default"
+ PolarisDubboGroup = "dubbo.group"
+ PolarisClientName = "polaris-client"
+)
+
+const (
+ PolarisInstanceHealthStatus = "healthstatus"
+ PolarisInstanceIsolatedStatus = "isolated"
+ PolarisCIrcuirbreakerStatus = "circuitbreaker"
+)
+
+const (
+ PluginPolarisTpsLimiter = "polaris-limit"
+ PluginPolarisRouterFactory = "polaris-router"
+ PluginPolarisReportFilter = "polaris-report"
+)
diff --git a/filter/filter_impl/import.go b/filter/filter_impl/import.go
index a647242db..277b5f56e 100644
--- a/filter/filter_impl/import.go
+++ b/filter/filter_impl/import.go
@@ -31,6 +31,7 @@ import (
_ "dubbo.apache.org/dubbo-go/v3/filter/graceful_shutdown"
_ "dubbo.apache.org/dubbo-go/v3/filter/hystrix"
_ "dubbo.apache.org/dubbo-go/v3/filter/metrics"
+ _ "dubbo.apache.org/dubbo-go/v3/filter/polaris/limit"
_ "dubbo.apache.org/dubbo-go/v3/filter/seata"
_ "dubbo.apache.org/dubbo-go/v3/filter/sentinel"
_ "dubbo.apache.org/dubbo-go/v3/filter/token"
diff --git a/filter/polaris/limit/default.go b/filter/polaris/limit/default.go
new file mode 100644
index 000000000..5f40295c4
--- /dev/null
+++ b/filter/polaris/limit/default.go
@@ -0,0 +1,30 @@
+/*
+ * 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 limit
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/filter"
+)
+
+func init() {
+ extension.SetTpsLimiter(constant.PluginPolarisTpsLimiter, func() filter.TpsLimiter {
+ return &polarisTpsLimiter{}
+ })
+}
diff --git a/filter/polaris/limit/limiter.go b/filter/polaris/limit/limiter.go
new file mode 100644
index 000000000..d482aa11a
--- /dev/null
+++ b/filter/polaris/limit/limiter.go
@@ -0,0 +1,173 @@
+/*
+ * 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 limit
+
+import (
+ "errors"
+ "fmt"
+ "time"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ "github.com/polarismesh/polaris-go"
+ "github.com/polarismesh/polaris-go/pkg/flow/data"
+ "github.com/polarismesh/polaris-go/pkg/model"
+ v1 "github.com/polarismesh/polaris-go/pkg/model/pb/v1"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/config"
+ "dubbo.apache.org/dubbo-go/v3/protocol/base"
+ remotingpolaris "dubbo.apache.org/dubbo-go/v3/remoting/polaris"
+ "dubbo.apache.org/dubbo-go/v3/remoting/polaris/parser"
+)
+
+type polarisTpsLimiter struct {
+ limitAPI polaris.LimitAPI
+}
+
+func (pl *polarisTpsLimiter) IsAllowable(url *common.URL, invocation base.Invocation) bool {
+ if err := remotingpolaris.Check(); errors.Is(err, remotingpolaris.ErrorNoOpenPolarisAbility) {
+ logger.Debug("[TpsLimiter][Polaris] not open polaris ratelimit ability")
+ return true
+ }
+
+ var err error
+
+ pl.limitAPI, err = remotingpolaris.GetLimiterAPI()
+ if err != nil {
+ logger.Error("[TpsLimiter][Polaris] create polaris LimitAPI fail : %+v", err)
+ return true
+ }
+
+ req := pl.buildQuotaRequest(url, invocation)
+ if req == nil {
+ return true
+ }
+ logger.Debugf("[TpsLimiter][Polaris] quota req : %+v", req)
+
+ resp, err := pl.limitAPI.GetQuota(req)
+ if err != nil {
+ logger.Error("[TpsLimiter][Polaris] ns:%s svc:%s get quota fail : %+v", remotingpolaris.GetNamespace(), url.Service(), err)
+ return true
+ }
+
+ return resp.Get().Code == model.QuotaResultOk
+}
+
+func (pl *polarisTpsLimiter) buildQuotaRequest(url *common.URL, invoaction base.Invocation) polaris.QuotaRequest {
+ ns := remotingpolaris.GetNamespace()
+ applicationMode := false
+ for _, item := range config.GetRootConfig().Registries {
+ if item.Protocol == constant.PolarisKey {
+ applicationMode = item.RegistryType == constant.ServiceKey
+ }
+ }
+
+ svc := url.Interface()
+ method := invoaction.MethodName()
+ if applicationMode {
+ svc = config.GetApplicationConfig().Name
+ method = url.Interface() + "/" + invoaction.MethodName()
+ }
+
+ req := polaris.NewQuotaRequest()
+ req.SetNamespace(ns)
+ req.SetService(svc)
+ req.SetMethod(method)
+
+ matchs, ok := pl.buildArguments(req.(*model.QuotaRequestImpl))
+ if !ok {
+ return nil
+ }
+
+ attachement := invoaction.Attachments()
+ arguments := invoaction.Arguments()
+
+ for i := range matchs {
+ item := matchs[i]
+ switch item.GetType() {
+ case v1.MatchArgument_HEADER:
+ if val, ok := attachement[item.GetKey()]; ok {
+ req.AddArgument(model.BuildHeaderArgument(item.GetKey(), fmt.Sprintf("%+v", val)))
+ }
+ case v1.MatchArgument_QUERY:
+ if val := parser.ParseArgumentsByExpression(item.GetKey(), arguments); val != nil {
+ req.AddArgument(model.BuildQueryArgument(item.GetKey(), fmt.Sprintf("%+v", val)))
+ }
+ case v1.MatchArgument_CALLER_IP:
+ callerIp := url.GetParam(constant.RemoteAddr, "")
+ if len(callerIp) != 0 {
+ req.AddArgument(model.BuildCallerIPArgument(callerIp))
+ }
+ case model.ArgumentTypeCallerService:
+ }
+ }
+
+ return req
+}
+
+func (pl *polarisTpsLimiter) buildArguments(req *model.QuotaRequestImpl) ([]*v1.MatchArgument, bool) {
+ engine := pl.limitAPI.SDKContext().GetEngine()
+
+ getRuleReq := &data.CommonRateLimitRequest{
+ DstService: model.ServiceKey{
+ Namespace: req.GetNamespace(),
+ Service: req.GetService(),
+ },
+ Trigger: model.NotifyTrigger{
+ EnableDstRateLimit: true,
+ },
+ ControlParam: model.ControlParam{
+ Timeout: time.Millisecond * 500,
+ },
+ }
+
+ if err := engine.SyncGetResources(getRuleReq); err != nil {
+ logger.Error("[TpsLimiter][Polaris] ns:%s svc:%s get RateLimit Rule fail : %+v", req.GetNamespace(), req.GetService(), err)
+ return nil, false
+ }
+
+ svcRule := getRuleReq.RateLimitRule
+ if svcRule == nil || svcRule.GetValue() == nil {
+ logger.Warnf("[TpsLimiter][Polaris] ns:%s svc:%s get RateLimit Rule is nil", req.GetNamespace(), req.GetService())
+ return nil, false
+ }
+
+ rules, ok := svcRule.GetValue().(*v1.RateLimit)
+ if !ok {
+ logger.Error("[TpsLimiter][Polaris] ns:%s svc:%s get RateLimit Rule invalid", req.GetNamespace(), req.GetService())
+ return nil, false
+ }
+
+ ret := make([]*v1.MatchArgument, 0, 4)
+ for i := range rules.GetRules() {
+ rule := rules.GetRules()[i]
+ if len(rule.GetArguments()) == 0 {
+ continue
+ }
+
+ ret = append(ret, rule.Arguments...)
+ }
+
+ return ret, true
+}
diff --git a/go.mod b/go.mod
index 342562487..dbc74af64 100644
--- a/go.mod
+++ b/go.mod
@@ -37,8 +37,10 @@ require (
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd
github.com/nacos-group/nacos-sdk-go/v2 v2.2.2
github.com/natefinch/lumberjack v2.0.0+incompatible
+ github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
github.com/opentracing/opentracing-go v1.2.0
github.com/pkg/errors v0.9.1
+ github.com/polarismesh/polaris-go v1.3.0
github.com/prometheus/client_golang v1.13.0
github.com/prometheus/common v0.37.0
github.com/sirupsen/logrus v1.8.1
@@ -89,6 +91,7 @@ require (
github.com/gorilla/websocket v1.4.2 // indirect
github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
github.com/hashicorp/errwrap v1.1.0 // indirect
+ github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/k0kubun/pp v3.0.1+incompatible // indirect
@@ -97,6 +100,7 @@ require (
github.com/mattn/go-isatty v0.0.16 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
github.com/mitchellh/copystructure v1.2.0 // indirect
+ github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/mitchellh/reflectwalk v1.0.2 // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/mschoch/smat v0.2.0 // indirect
@@ -108,6 +112,7 @@ require (
github.com/prometheus/client_model v0.3.0 // indirect
github.com/prometheus/procfs v0.8.0 // indirect
github.com/shirou/gopsutil/v3 v3.22.2 // indirect
+ github.com/spaolacci/murmur3 v1.1.0 // indirect
github.com/stretchr/objx v0.5.0 // indirect
github.com/tklauser/go-sysconf v0.3.10 // indirect
github.com/tklauser/numcpus v0.4.0 // indirect
diff --git a/go.sum b/go.sum
index 661bb6fec..c3bfc201f 100644
--- a/go.sum
+++ b/go.sum
@@ -49,6 +49,8 @@ github.com/Workiva/go-datastructures v1.0.52 h1:PLSK6pwn8mYdaoaCZEMsXBpBotr4HHn9
github.com/Workiva/go-datastructures v1.0.52/go.mod h1:Z+F2Rca0qCsVYDS8z7bAGm8f3UkzuWYS/oBZz5a7VVA=
github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5 h1:rFw4nCn9iMW+Vajsk51NtYIcwSTkXr+JGrMd36kTDJw=
github.com/afex/hystrix-go v0.0.0-20180502004556-fa1af6a1f4f5/go.mod h1:SkGFH1ia65gfNATL8TAiHDNxPzPdmEL5uirI2Uyuz6c=
+github.com/agiledragon/gomonkey v2.0.2+incompatible h1:eXKi9/piiC3cjJD1658mEE2o3NjkJ5vDLgYjCQu0Xlw=
+github.com/agiledragon/gomonkey v2.0.2+incompatible/go.mod h1:2NGfXu1a80LLr2cmWXGBDaHEjb1idR6+FVlX5T3D9hw=
github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -306,6 +308,13 @@ github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8l
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gonum/blas v0.0.0-20181208220705-f22b278b28ac/go.mod h1:P32wAyui1PQ58Oce/KYkOqQv8cVw1zAapXOl+dRFGbc=
+github.com/gonum/floats v0.0.0-20181209220543-c233463c7e82/go.mod h1:PxC8OnwL11+aosOB5+iEPoV3picfs8tUpkVd0pDo+Kg=
+github.com/gonum/integrate v0.0.0-20181209220457-a422b5c0fdf2/go.mod h1:pDgmNM6seYpwvPos3q+zxlXMsbve6mOIPucUnUOrI7Y=
+github.com/gonum/internal v0.0.0-20181124074243-f884aa714029/go.mod h1:Pu4dmpkhSyOzRwuXkOgAvijx4o+4YMUJJo9OvPYMkks=
+github.com/gonum/lapack v0.0.0-20181123203213-e4cdc5a0bff9/go.mod h1:XA3DeT6rxh2EAE789SSiSJNqxPaC0aE9J8NTOI0Jo/A=
+github.com/gonum/matrix v0.0.0-20181209220409-c518dec07be9/go.mod h1:0EXg4mc1CNP0HCqCz+K4ts155PXIlUywf0wqN+GfPZw=
+github.com/gonum/stat v0.0.0-20181125101827-41a0da705a5b/go.mod h1:Z4GIJBJO3Wa4gD4vbwQxXXZ+WHmW6E9ixmNrwvs0iZs=
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/btree v1.0.0 h1:0udJVsspx3VBr5FwtLhQQtuAsVc79tTq0ocGIPAU6qo=
github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
@@ -385,6 +394,8 @@ github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjh
github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
github.com/hashicorp/go-plugin v1.0.1/go.mod h1:++UyYGoz3o5w9ZzAdZxtQKrWWP+iqPBn3cQptSMzBuY=
github.com/hashicorp/go-retryablehttp v0.5.4/go.mod h1:9B5zBasrRhHXnJnui7y6sL7es7NDiJgTc6Er0maI1Xs=
github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
@@ -521,6 +532,7 @@ github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFW
github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
github.com/mitchellh/go-testing-interface v0.0.0-20171004221916-a61a99592b77/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
github.com/mitchellh/go-testing-interface v1.0.0/go.mod h1:kRemZodwjscx+RGhAo8eIhFbs2+BFgRtFPeD/KE+zxI=
@@ -564,6 +576,8 @@ github.com/oklog/oklog v0.3.2/go.mod h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtb
github.com/oklog/run v1.0.0/go.mod h1:dlhp/R75TPv97u0XWUtDeV/lRKWPKSdTuV0TZvrmrQA=
github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
github.com/olekukonko/tablewriter v0.0.0-20170122224234-a0225b3f23b5/go.mod h1:vsDQFd/mU46D+Z4whnwzcISnGGzXWMclvtLoiIKAKIo=
+github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852 h1:Yl0tPBa8QPjGmesFh1D0rDy+q1Twx6FyU7VWHi8wZbI=
+github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852/go.mod h1:eqOVx5Vwu4gd2mmMZvVZsgIqNSaW3xxRThUJ0k/TPk4=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
@@ -600,6 +614,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/polarismesh/polaris-go v1.3.0 h1:KZKX//ow4OPPoS5+s7h07ptprg+2AcNVGrN6WakC9QM=
+github.com/polarismesh/polaris-go v1.3.0/go.mod h1:HsN0ierETIujHpmnnYJ3qkwQw4QGAECuHvBZTDaw1tI=
github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
github.com/posener/complete v1.2.3/go.mod h1:WZIdtGGp+qx0sLrYKtIRAruyNpv6hFCicSgv7Sy7s/s=
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw=
@@ -686,6 +702,8 @@ github.com/soheilhy/cmux v0.1.5-0.20210205191134-5ec6847320e5 h1:GJTW+uNMIV1RKwo
github.com/soheilhy/cmux v0.1.5-0.20210205191134-5ec6847320e5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
github.com/sony/gobreaker v0.4.1/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
+github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
github.com/spf13/cobra v0.0.3/go.mod h1:1l0Ry5zgKvJasoi3XT1TypsSe7PqH0Sj9dhYf7v3XqQ=
@@ -934,6 +952,7 @@ golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod h1:9nx3DQGgdP8bBQD5qx
golang.org/x/net v0.0.0-20211105192438-b53810dc28af/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
+golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
@@ -1033,6 +1052,7 @@ golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -1191,6 +1211,7 @@ google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6D
google.golang.org/genproto v0.0.0-20210106152847-07624b53cd92/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
google.golang.org/genproto v0.0.0-20211104193956-4c6863e31247/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
+google.golang.org/genproto v0.0.0-20220504150022-98cd25cafc72/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4=
google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d h1:VBu5YqKPv6XiJ199exd8Br+Aetz+o08F+PLMnwJQHAY=
google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d/go.mod h1:yZTlhN0tQnXo3h00fuXNCxJdLdIdnVFVBaRJ5LWBbw4=
google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d h1:DoPTO70H+bcDXcd39vOqb2viZxgqeBeSGtZ55yZU4/Q=
@@ -1221,6 +1242,8 @@ google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTp
google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
+google.golang.org/grpc v1.46.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
+google.golang.org/grpc v1.46.2/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
google.golang.org/grpc v1.48.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
@@ -1237,6 +1260,7 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
diff --git a/imports/imports.go b/imports/imports.go
index 61fd206dc..35743fd88 100644
--- a/imports/imports.go
+++ b/imports/imports.go
@@ -37,6 +37,7 @@ import (
_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/random"
_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/roundrobin"
_ "dubbo.apache.org/dubbo-go/v3/cluster/router/condition"
+ _ "dubbo.apache.org/dubbo-go/v3/cluster/router/polaris"
_ "dubbo.apache.org/dubbo-go/v3/cluster/router/script"
_ "dubbo.apache.org/dubbo-go/v3/cluster/router/tag"
_ "dubbo.apache.org/dubbo-go/v3/config_center/nacos"
@@ -53,6 +54,7 @@ import (
_ "dubbo.apache.org/dubbo-go/v3/filter/hystrix"
_ "dubbo.apache.org/dubbo-go/v3/filter/metrics"
_ "dubbo.apache.org/dubbo-go/v3/filter/otel/trace"
+ _ "dubbo.apache.org/dubbo-go/v3/filter/polaris/limit"
_ "dubbo.apache.org/dubbo-go/v3/filter/seata"
_ "dubbo.apache.org/dubbo-go/v3/filter/sentinel"
_ "dubbo.apache.org/dubbo-go/v3/filter/token"
@@ -80,6 +82,7 @@ import (
_ "dubbo.apache.org/dubbo-go/v3/registry/directory"
_ "dubbo.apache.org/dubbo-go/v3/registry/etcdv3"
_ "dubbo.apache.org/dubbo-go/v3/registry/nacos"
+ _ "dubbo.apache.org/dubbo-go/v3/registry/polaris"
_ "dubbo.apache.org/dubbo-go/v3/registry/protocol"
_ "dubbo.apache.org/dubbo-go/v3/registry/servicediscovery"
_ "dubbo.apache.org/dubbo-go/v3/registry/zookeeper"
diff --git a/registry/options.go b/registry/options.go
index eac366886..e9f1fb285 100644
--- a/registry/options.go
+++ b/registry/options.go
@@ -70,6 +70,12 @@ func WithNacos() Option {
}
}
+func WithPolaris() Option {
+ return func(opts *Options) {
+ opts.Registry.Protocol = constant.PolarisKey
+ }
+}
+
func WithXDS() Option {
return func(opts *Options) {
opts.Registry.Protocol = constant.XDSRegistryKey
diff --git a/registry/polaris/core.go b/registry/polaris/core.go
new file mode 100644
index 000000000..1db3ef04f
--- /dev/null
+++ b/registry/polaris/core.go
@@ -0,0 +1,131 @@
+/*
+ * 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 polaris
+
+import (
+ "sync"
+ "time"
+)
+
+import (
+ api "github.com/polarismesh/polaris-go"
+ internalapi "github.com/polarismesh/polaris-go/api"
+ "github.com/polarismesh/polaris-go/pkg/model"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/config_center"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+)
+
+type item func(remoting.EventType, []model.Instance)
+
+type PolarisServiceWatcher struct {
+ consumer api.ConsumerAPI
+ subscribeParam *api.WatchServiceRequest
+ lock *sync.RWMutex
+ subscribers []item
+ execOnce *sync.Once
+}
+
+// newPolarisWatcher create PolarisServiceWatcher to do watch service action
+func newPolarisWatcher(param *api.WatchServiceRequest, consumer api.ConsumerAPI) (*PolarisServiceWatcher, error) {
+ watcher := &PolarisServiceWatcher{
+ subscribeParam: param,
+ consumer: consumer,
+ lock: &sync.RWMutex{},
+ subscribers: make([]item, 0),
+ execOnce: &sync.Once{},
+ }
+ return watcher, nil
+}
+
+// AddSubscriber add subscriber into watcher's subscribers
+func (watcher *PolarisServiceWatcher) AddSubscriber(subscriber func(remoting.EventType, []model.Instance)) {
+
+ watcher.lock.Lock()
+ watcher.lazyRun()
+ defer watcher.lock.Unlock()
+
+ watcher.subscribers = append(watcher.subscribers, subscriber)
+}
+
+// lazyRun Delayed execution, only triggered when AddSubscriber is called, and will only be executed once
+func (watcher *PolarisServiceWatcher) lazyRun() {
+ watcher.execOnce.Do(func() {
+ go watcher.startWatch()
+ })
+}
+
+// startWatch start run work to watch target service by polaris
+func (watcher *PolarisServiceWatcher) startWatch() {
+ for {
+ resp, err := watcher.consumer.WatchService(watcher.subscribeParam)
+ if err != nil {
+ time.Sleep(time.Duration(500 * time.Millisecond))
+ continue
+ }
+ watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
+ Value: resp.GetAllInstancesResp.Instances,
+ ConfigType: remoting.EventTypeAdd,
+ })
+
+ for event := range resp.EventChannel {
+ eType := event.GetSubScribeEventType()
+ if eType == internalapi.EventInstance {
+ insEvent := event.(*model.InstanceEvent)
+
+ if insEvent.AddEvent != nil {
+ watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
+ Value: insEvent.AddEvent.Instances,
+ ConfigType: remoting.EventTypeAdd,
+ })
+ }
+ if insEvent.UpdateEvent != nil {
+ instances := make([]model.Instance, len(insEvent.UpdateEvent.UpdateList))
+ for i := range insEvent.UpdateEvent.UpdateList {
+ instances[i] = insEvent.UpdateEvent.UpdateList[i].After
+ }
+ watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
+ Value: instances,
+ ConfigType: remoting.EventTypeUpdate,
+ })
+ }
+ if insEvent.DeleteEvent != nil {
+ watcher.notifyAllSubscriber(&config_center.ConfigChangeEvent{
+ Value: insEvent.DeleteEvent.Instances,
+ ConfigType: remoting.EventTypeDel,
+ })
+ }
+ }
+ }
+
+ }
+}
+
+// notifyAllSubscriber notify config_center.ConfigChangeEvent to all subscriber
+func (watcher *PolarisServiceWatcher) notifyAllSubscriber(event *config_center.ConfigChangeEvent) {
+ watcher.lock.RLock()
+ defer watcher.lock.RUnlock()
+
+ for i := 0; i < len(watcher.subscribers); i++ {
+ subscriber := watcher.subscribers[i]
+ subscriber(event.ConfigType, event.Value.([]model.Instance))
+ }
+
+}
diff --git a/registry/polaris/doc.go b/registry/polaris/doc.go
new file mode 100644
index 000000000..4645a2776
--- /dev/null
+++ b/registry/polaris/doc.go
@@ -0,0 +1,19 @@
+/*
+ * 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 polaris implements registry around polaris.
+package polaris
diff --git a/registry/polaris/listener.go b/registry/polaris/listener.go
new file mode 100644
index 000000000..e39ca995d
--- /dev/null
+++ b/registry/polaris/listener.go
@@ -0,0 +1,123 @@
+/*
+ * 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 polaris
+
+import (
+ "fmt"
+ "net/url"
+ "strconv"
+)
+
+import (
+ gxchan "github.com/dubbogo/gost/container/chan"
+ "github.com/dubbogo/gost/log/logger"
+
+ perrors "github.com/pkg/errors"
+
+ "github.com/polarismesh/polaris-go/pkg/model"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/config_center"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+)
+
+type polarisListener struct {
+ watcher *PolarisServiceWatcher
+ events *gxchan.UnboundedChan
+ closeCh chan struct{}
+}
+
+// NewPolarisListener new polaris listener
+func NewPolarisListener(watcher *PolarisServiceWatcher) (*polarisListener, error) {
+ listener := &polarisListener{
+ watcher: watcher,
+ events: gxchan.NewUnboundedChan(32),
+ closeCh: make(chan struct{}),
+ }
+ listener.startListen()
+ return listener, nil
+}
+func (pl *polarisListener) startListen() {
+ pl.watcher.AddSubscriber(func(et remoting.EventType, ins []model.Instance) {
+ for i := range ins {
+ pl.events.In() <- &config_center.ConfigChangeEvent{Value: ins[i], ConfigType: et}
+ }
+ })
+}
+
+// Next returns next service event once received
+func (pl *polarisListener) Next() (*registry.ServiceEvent, error) {
+ for {
+ select {
+ case <-pl.closeCh:
+ logger.Warnf("polaris listener is close")
+ return nil, perrors.New("listener stopped")
+ case val := <-pl.events.Out():
+ e, _ := val.(*config_center.ConfigChangeEvent)
+ logger.Debugf("got polaris event %s", e)
+ instance := e.Value.(model.Instance)
+ return ®istry.ServiceEvent{Action: e.ConfigType, Service: generateUrl(instance)}, nil
+ }
+ }
+}
+
+// Close closes this listener
+func (pl *polarisListener) Close() {
+ // TODO need to add UnWatch in polaris
+ close(pl.closeCh)
+}
+
+func generateUrl(instance model.Instance) *common.URL {
+ if instance.GetMetadata() == nil {
+ logger.Errorf("polaris instance metadata is empty,instance:%+v", instance)
+ return nil
+ }
+ path := instance.GetMetadata()["path"]
+ myInterface := instance.GetMetadata()["interface"]
+ if len(path) == 0 && len(myInterface) == 0 {
+ logger.Errorf("polaris instance metadata does not have both path key and interface key,instance:%+v", instance)
+ return nil
+ }
+ if len(path) == 0 && len(myInterface) != 0 {
+ path = "/" + myInterface
+ }
+ protocol := instance.GetProtocol()
+ if len(protocol) == 0 {
+ logger.Errorf("polaris instance metadata does not have protocol key,instance:%+v", instance)
+ return nil
+ }
+ urlMap := url.Values{}
+ for k, v := range instance.GetMetadata() {
+ urlMap.Set(k, v)
+ }
+ urlMap.Set(constant.PolarisInstanceID, instance.GetId())
+ urlMap.Set(constant.PolarisInstanceHealthStatus, fmt.Sprintf("%+v", instance.IsHealthy()))
+ urlMap.Set(constant.PolarisInstanceIsolatedStatus, fmt.Sprintf("%+v", instance.IsIsolated()))
+ instance.GetCircuitBreakerStatus()
+ return common.NewURLWithOptions(
+ common.WithIp(instance.GetHost()),
+ common.WithPort(strconv.Itoa(int(instance.GetPort()))),
+ common.WithProtocol(protocol),
+ common.WithParams(urlMap),
+ common.WithPath(path),
+ )
+}
diff --git a/registry/polaris/registry.go b/registry/polaris/registry.go
new file mode 100644
index 000000000..41f754f10
--- /dev/null
+++ b/registry/polaris/registry.go
@@ -0,0 +1,287 @@
+/*
+ * 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 polaris
+
+import (
+ "fmt"
+ "strconv"
+ "sync"
+ "time"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ perrors "github.com/pkg/errors"
+
+ api "github.com/polarismesh/polaris-go"
+ "github.com/polarismesh/polaris-go/pkg/model"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+ "dubbo.apache.org/dubbo-go/v3/remoting/polaris"
+)
+
+const (
+ RegistryConnDelay = 3
+ defaultHeartbeatIntervalSec = 5
+)
+
+func init() {
+ extension.SetRegistry(constant.PolarisKey, newPolarisRegistry)
+}
+
+// newPolarisRegistry will create new instance
+func newPolarisRegistry(url *common.URL) (registry.Registry, error) {
+ if err := polaris.InitSDKContext(url); err != nil {
+ return &polarisRegistry{}, err
+ }
+
+ providerApi, err := polaris.GetProviderAPI()
+ if err != nil {
+ return nil, err
+ }
+
+ consumerApi, err := polaris.GetConsumerAPI()
+ if err != nil {
+ return nil, err
+ }
+
+ pRegistry := &polarisRegistry{
+ url: url,
+ namespace: url.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace),
+ provider: providerApi,
+ consumer: consumerApi,
+ registryUrls: make([]*common.URL, 0, 4),
+ watchers: map[string]*PolarisServiceWatcher{},
+ }
+
+ return pRegistry, nil
+}
+
+type polarisRegistry struct {
+ namespace string
+ url *common.URL
+ consumer api.ConsumerAPI
+ provider api.ProviderAPI
+ lock sync.RWMutex
+ registryUrls []*common.URL
+ listenerLock sync.RWMutex
+ watchers map[string]*PolarisServiceWatcher
+}
+
+// Register will register the service @url to its polaris registry center.
+func (pr *polarisRegistry) Register(url *common.URL) error {
+ if getCategory(url) != "providers" {
+ return nil
+ }
+
+ serviceName := url.Interface()
+ request := createRegisterParam(url, serviceName)
+ request.Namespace = pr.namespace
+ resp, err := pr.provider.RegisterInstance(request)
+ if err != nil {
+ return err
+ }
+
+ if resp.Existed {
+ logger.Warnf("instance already regist, namespace:%+v, service:%+v, host:%+v, port:%+v",
+ request.Namespace, request.Service, request.Host, request.Port)
+ }
+ url.SetParam(constant.PolarisInstanceID, resp.InstanceID)
+
+ pr.lock.Lock()
+ pr.registryUrls = append(pr.registryUrls, url)
+ pr.lock.Unlock()
+
+ return nil
+}
+
+// UnRegister returns nil if unregister successfully. If not, returns an error.
+func (pr *polarisRegistry) UnRegister(url *common.URL) error {
+ request := createDeregisterParam(url, url.Interface())
+ request.Namespace = pr.namespace
+ if err := pr.provider.Deregister(request); err != nil {
+ return perrors.WithMessagef(err, "fail to deregister(conf:%+v)", url)
+ }
+ return nil
+}
+
+// Subscribe returns nil if subscribing registry successfully. If not returns an error.
+func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.NotifyListener) error {
+
+ role, _ := strconv.Atoi(url.GetParam(constant.RegistryRoleKey, ""))
+ if role != common.CONSUMER {
+ return nil
+ }
+ timer := time.NewTimer(time.Duration(RegistryConnDelay) * time.Second)
+ defer timer.Stop()
+
+ for {
+ serviceName := url.Interface()
+ watcher, err := pr.createPolarisWatcher(serviceName)
+ if err != nil {
+ logger.Warnf("getwatcher() = err:%v", perrors.WithStack(err))
+ <-timer.C
+ timer.Reset(time.Duration(RegistryConnDelay) * time.Second)
+ continue
+ }
+
+ listener, err := NewPolarisListener(watcher)
+ if err != nil {
+ logger.Warnf("getListener() = err:%v", perrors.WithStack(err))
+ <-timer.C
+ timer.Reset(time.Duration(RegistryConnDelay) * time.Second)
+ continue
+ }
+
+ for {
+ serviceEvent, err := listener.Next()
+
+ if err != nil {
+ logger.Warnf("Selector.watch() = error{%v}", perrors.WithStack(err))
+ listener.Close()
+ return err
+ }
+ logger.Infof("update begin, service event: %v", serviceEvent.String())
+ notifyListener.Notify(serviceEvent)
+ }
+ }
+}
+
+// UnSubscribe returns nil if unsubscribing registry successfully. If not returns an error.
+func (pr *polarisRegistry) UnSubscribe(url *common.URL, notifyListener registry.NotifyListener) error {
+ // TODO wait polaris support it
+ return perrors.New("UnSubscribe not support in polarisRegistry")
+}
+
+// LoadSubscribeInstances load subscribe instance
+func (pr *polarisRegistry) LoadSubscribeInstances(url *common.URL, notify registry.NotifyListener) error {
+ serviceName := url.Interface()
+ resp, err := pr.consumer.GetInstances(&api.GetInstancesRequest{
+ GetInstancesRequest: model.GetInstancesRequest{
+ Service: serviceName,
+ Namespace: pr.namespace,
+ },
+ })
+ if err != nil {
+ return perrors.New(fmt.Sprintf("could not query the instances for serviceName=%s,namespace=%s,error=%v",
+ serviceName, pr.namespace, err))
+ }
+ for i := range resp.Instances {
+ if newUrl := generateUrl(resp.Instances[i]); newUrl != nil {
+ notify.Notify(®istry.ServiceEvent{Action: remoting.EventTypeAdd, Service: newUrl})
+ }
+ }
+ return nil
+}
+
+// GetURL returns polaris registry's url.
+func (pr *polarisRegistry) GetURL() *common.URL {
+ return pr.url
+}
+
+func (pr *polarisRegistry) createPolarisWatcher(serviceName string) (*PolarisServiceWatcher, error) {
+
+ pr.listenerLock.Lock()
+ defer pr.listenerLock.Unlock()
+
+ if _, exist := pr.watchers[serviceName]; !exist {
+ subscribeParam := &api.WatchServiceRequest{
+ WatchServiceRequest: model.WatchServiceRequest{
+ Key: model.ServiceKey{
+ Namespace: pr.namespace,
+ Service: serviceName,
+ },
+ },
+ }
+
+ watcher, err := newPolarisWatcher(subscribeParam, pr.consumer)
+ if err != nil {
+ return nil, err
+ }
+ pr.watchers[serviceName] = watcher
+ }
+
+ return pr.watchers[serviceName], nil
+}
+
+// Destroy stop polaris registry.
+func (pr *polarisRegistry) Destroy() {
+ for i := range pr.registryUrls {
+ url := pr.registryUrls[i]
+ err := pr.UnRegister(url)
+ logger.Infof("DeRegister Polaris URL:%+v", url)
+ if err != nil {
+ logger.Errorf("Deregister Polaris URL:%+v err:%v", url, err.Error())
+ }
+ }
+}
+
+// IsAvailable always return true when use polaris
+func (pr *polarisRegistry) IsAvailable() bool {
+ return true
+}
+
+// createRegisterParam convert dubbo url to polaris instance register request
+func createRegisterParam(url *common.URL, serviceName string) *api.InstanceRegisterRequest {
+ common.HandleRegisterIPAndPort(url)
+ port, _ := strconv.Atoi(url.Port)
+
+ metadata := make(map[string]string, len(url.GetParams()))
+ url.RangeParams(func(key, value string) bool {
+ metadata[key] = value
+ return true
+ })
+ metadata[constant.PolarisDubboPath] = url.Path
+
+ ver := url.GetParam("version", "")
+
+ req := &api.InstanceRegisterRequest{
+ InstanceRegisterRequest: model.InstanceRegisterRequest{
+ Service: serviceName,
+ Host: url.Ip,
+ Port: port,
+ Protocol: &url.Protocol,
+ Version: &ver,
+ Metadata: metadata,
+ },
+ }
+
+ req.SetTTL(defaultHeartbeatIntervalSec)
+
+ return req
+}
+
+// createDeregisterParam convert dubbo url to polaris instance deregister request
+func createDeregisterParam(url *common.URL, serviceName string) *api.InstanceDeRegisterRequest {
+ common.HandleRegisterIPAndPort(url)
+ port, _ := strconv.Atoi(url.Port)
+ return &api.InstanceDeRegisterRequest{
+ InstanceDeRegisterRequest: model.InstanceDeRegisterRequest{
+ Service: serviceName,
+ Host: url.Ip,
+ Port: port,
+ },
+ }
+}
diff --git a/registry/polaris/service_discovery.go b/registry/polaris/service_discovery.go
new file mode 100644
index 000000000..f294b0b8b
--- /dev/null
+++ b/registry/polaris/service_discovery.go
@@ -0,0 +1,348 @@
+/*
+ * 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 polaris
+
+import (
+ "fmt"
+ "sync"
+)
+
+import (
+ gxset "github.com/dubbogo/gost/container/set"
+ gxpage "github.com/dubbogo/gost/hash/page"
+ "github.com/dubbogo/gost/log/logger"
+
+ perrors "github.com/pkg/errors"
+
+ api "github.com/polarismesh/polaris-go"
+ "github.com/polarismesh/polaris-go/pkg/model"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/common/extension"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+ "dubbo.apache.org/dubbo-go/v3/remoting"
+ "dubbo.apache.org/dubbo-go/v3/remoting/polaris"
+)
+
+func init() {
+ extension.SetServiceDiscovery(constant.PolarisKey, newPolarisServiceDiscovery)
+}
+
+// newPolarisServiceDiscovery will create new service discovery instance
+func newPolarisServiceDiscovery(url *common.URL) (registry.ServiceDiscovery, error) {
+ discoveryURL := common.NewURLWithOptions(
+ common.WithParams(url.GetParams()),
+ common.WithParamsValue(constant.TimeoutKey, url.GetParam(constant.RegistryTimeoutKey, constant.DefaultRegTimeout)),
+ common.WithParamsValue(constant.PolarisServiceToken, url.Password),
+ common.WithParamsValue(constant.RegistryNamespaceKey, url.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace)))
+ discoveryURL.Location = url.Location
+ discoveryURL.Password = url.Password
+
+ if err := polaris.InitSDKContext(url); err != nil {
+ return nil, err
+ }
+
+ providerApi, err := polaris.GetProviderAPI()
+ if err != nil {
+ return nil, err
+ }
+
+ consumerApi, err := polaris.GetConsumerAPI()
+ if err != nil {
+ return nil, err
+ }
+
+ if err != nil {
+ return nil, perrors.WithMessage(err, "create polaris namingClient failed.")
+ }
+
+ descriptor := fmt.Sprintf("polaris-service-discovery[%s]", discoveryURL.Location)
+
+ newInstance := &polarisServiceDiscovery{
+ namespace: discoveryURL.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace),
+ descriptor: descriptor,
+ consumer: consumerApi,
+ provider: providerApi,
+ services: gxset.NewSet(),
+ registryInstances: make(map[string]*PolarisInstanceInfo),
+ watchers: make(map[string]*PolarisServiceWatcher),
+ }
+ return newInstance, nil
+}
+
+type polarisServiceDiscovery struct {
+ namespace string
+ descriptor string
+ provider api.ProviderAPI
+ consumer api.ConsumerAPI
+ services *gxset.HashSet
+ instanceLock sync.RWMutex
+ registryInstances map[string]*PolarisInstanceInfo
+ watchers map[string]*PolarisServiceWatcher
+ listenerLock sync.RWMutex
+}
+
+// Destroy destroy polarisServiceDiscovery, will do unregister all ServiceInstance
+// and close polaris.ConsumerAPI and polaris.ProviderAPI
+func (polaris *polarisServiceDiscovery) Destroy() error {
+ for _, inst := range polaris.registryInstances {
+ err := polaris.Unregister(inst.instance)
+ logger.Infof("Unregister polaris instance:%+v", inst)
+ if err != nil {
+ logger.Errorf("Unregister polaris instance:%+v, err:%+v", inst, err)
+ }
+ }
+ polaris.provider.Destroy()
+ polaris.consumer.Destroy()
+ return nil
+}
+
+// Register do register for ServiceInstance
+func (polaris *polarisServiceDiscovery) Register(instance registry.ServiceInstance) error {
+
+ ins := convertToRegisterInstance(polaris.namespace, instance)
+ resp, err := polaris.provider.RegisterInstance(ins)
+ if err != nil {
+ return perrors.WithMessage(err, "could not register the instance. "+instance.GetServiceName())
+ }
+
+ if resp.Existed {
+ logger.Warnf("instance already regist, namespace:%+v, service:%+v, host:%+v, port:%+v",
+ polaris.namespace, instance.GetServiceName(), instance.GetHost(), instance.GetPort())
+ }
+
+ polaris.instanceLock.Lock()
+ defer polaris.instanceLock.Unlock()
+
+ polaris.registryInstances[getInstanceKey(polaris.namespace, instance)] = &PolarisInstanceInfo{
+ instance: instance,
+ }
+ polaris.services.Add(instance.GetServiceName())
+
+ return nil
+}
+
+// Update update ServiceInstance info
+func (polaris *polarisServiceDiscovery) Update(instance registry.ServiceInstance) error {
+ err := polaris.Unregister(instance)
+ if err != nil {
+ return perrors.WithStack(err)
+ }
+ polaris.services.Add(instance.GetServiceName())
+ return polaris.Register(instance)
+}
+
+// Unregister do Unregister for ServiceInstance
+func (polaris *polarisServiceDiscovery) Unregister(instance registry.ServiceInstance) error {
+
+ func() {
+ polaris.instanceLock.Lock()
+ defer polaris.instanceLock.Unlock()
+ key := getInstanceKey(polaris.namespace, instance)
+ delete(polaris.registryInstances, key)
+ }()
+
+ err := polaris.provider.Deregister(convertToDeregisterInstance(polaris.namespace, instance))
+ if err != nil {
+ return perrors.WithMessage(err, "Could not unregister the instance. "+instance.GetServiceName())
+ }
+
+ polaris.services.Remove(instance.GetServiceName())
+ return nil
+}
+
+func (polaris *polarisServiceDiscovery) GetDefaultPageSize() int {
+ return registry.DefaultPageSize
+}
+
+func (polaris *polarisServiceDiscovery) GetServices() *gxset.HashSet {
+ return polaris.services
+}
+
+// GetInstances will return all service instances with serviceName
+func (polaris *polarisServiceDiscovery) GetInstances(serviceName string) []registry.ServiceInstance {
+ resp, err := polaris.consumer.GetAllInstances(&api.GetAllInstancesRequest{
+ GetAllInstancesRequest: model.GetAllInstancesRequest{
+ Service: serviceName,
+ Namespace: polaris.namespace,
+ },
+ })
+
+ if err != nil {
+ logger.Errorf("Could not query the instances for service: %+v . It happened err %+v", serviceName, err)
+ return make([]registry.ServiceInstance, 0)
+ }
+ res := make([]registry.ServiceInstance, 0, len(resp.Instances))
+ for _, ins := range resp.Instances {
+ metadata := ins.GetMetadata()
+ res = append(res, ®istry.DefaultServiceInstance{
+ ID: ins.GetId(),
+ ServiceName: serviceName,
+ Host: ins.GetHost(),
+ Port: int(ins.GetPort()),
+ Enable: !ins.IsIsolated(),
+ Healthy: ins.IsHealthy(),
+ Metadata: metadata,
+ })
+ }
+ return res
+}
+
+// GetInstancesByPage will return a page containing instances of ServiceInstance with the serviceName
+// the page will start at offset
+func (polaris *polarisServiceDiscovery) GetInstancesByPage(serviceName string, offset int, pageSize int) gxpage.Pager {
+ all := polaris.GetInstances(serviceName)
+ res := make([]any, 0, pageSize)
+ for i := offset; i < len(all) && i < offset+pageSize; i++ {
+ res = append(res, all[i])
+ }
+ return gxpage.NewPage(offset, pageSize, res, len(all))
+}
+
+// GetHealthyInstancesByPage will return a page containing instances of ServiceInstance.
+// The param healthy indices that the instance should be healthy or not.
+// The page will start at offset
+func (polaris *polarisServiceDiscovery) GetHealthyInstancesByPage(serviceName string, offset int, pageSize int, healthy bool) gxpage.Pager {
+ all := polaris.GetInstances(serviceName)
+ res := make([]any, 0, pageSize)
+ // could not use res = all[a:b] here because the res should be []any, not []ServiceInstance
+ var (
+ i = offset
+ count = 0
+ )
+ for i < len(all) && count < pageSize {
+ ins := all[i]
+ if ins.IsHealthy() == healthy {
+ res = append(res, all[i])
+ count++
+ }
+ i++
+ }
+ return gxpage.NewPage(offset, pageSize, res, len(all))
+}
+
+// GetRequestInstances get all instances by the specified service names
+func (polaris *polarisServiceDiscovery) GetRequestInstances(serviceNames []string, offset int, requestedSize int) map[string]gxpage.Pager {
+ res := make(map[string]gxpage.Pager, len(serviceNames))
+ for _, name := range serviceNames {
+ res[name] = polaris.GetInstancesByPage(name, offset, requestedSize)
+ }
+ return res
+}
+
+// AddListener add listener for ServiceInstancesChangedListener
+func (polaris *polarisServiceDiscovery) AddListener(listener registry.ServiceInstancesChangedListener) error {
+
+ for _, val := range listener.GetServiceNames().Values() {
+ serviceName := val.(string)
+ watcher, err := polaris.createPolarisWatcherIfAbsent(serviceName)
+ if err != nil {
+ return err
+ }
+
+ watcher.AddSubscriber(func(et remoting.EventType, instances []model.Instance) {
+ dubboInstances := make([]registry.ServiceInstance, 0, len(instances))
+ for _, instance := range instances {
+ dubboInstances = append(dubboInstances, ®istry.DefaultServiceInstance{
+ ID: instance.GetId(),
+ ServiceName: instance.GetService(),
+ Host: instance.GetHost(),
+ Port: int(instance.GetPort()),
+ Enable: !instance.IsIsolated(),
+ Healthy: instance.IsHealthy(),
+ Metadata: instance.GetMetadata(),
+ GroupName: instance.GetMetadata()[constant.PolarisDubboGroup],
+ })
+ }
+
+ listener.OnEvent(registry.NewServiceInstancesChangedEvent(serviceName, dubboInstances))
+ })
+ }
+
+ return nil
+}
+
+// createPolarisWatcherIfAbsent Calculate whether the corresponding PolarisWatcher needs to be created,
+// if it does not exist, create one, otherwise return the existing one
+func (polaris *polarisServiceDiscovery) createPolarisWatcherIfAbsent(serviceName string) (*PolarisServiceWatcher, error) {
+
+ polaris.listenerLock.Lock()
+ defer polaris.listenerLock.Unlock()
+
+ if _, exist := polaris.watchers[serviceName]; !exist {
+ subscribeParam := &api.WatchServiceRequest{
+ WatchServiceRequest: model.WatchServiceRequest{
+ Key: model.ServiceKey{
+ Namespace: polaris.namespace,
+ Service: serviceName,
+ },
+ },
+ }
+
+ watcher, err := newPolarisWatcher(subscribeParam, polaris.consumer)
+ if err != nil {
+ return nil, err
+ }
+ polaris.watchers[serviceName] = watcher
+ }
+
+ return polaris.watchers[serviceName], nil
+}
+
+// String retuen descriptor
+func (polaris *polarisServiceDiscovery) String() string {
+ return polaris.descriptor
+}
+
+func convertToRegisterInstance(namespace string, instance registry.ServiceInstance) *api.InstanceRegisterRequest {
+
+ var (
+ health = instance.IsHealthy()
+ isolate = instance.IsEnable()
+ ttl = 5
+ protocolVal = string(constant.DubboProtocol)
+ )
+
+ return &api.InstanceRegisterRequest{
+ InstanceRegisterRequest: model.InstanceRegisterRequest{
+ Service: instance.GetServiceName(),
+ Namespace: namespace,
+ Host: instance.GetHost(),
+ Port: instance.GetPort(),
+ Protocol: &protocolVal,
+ Metadata: instance.GetMetadata(),
+ Healthy: &health,
+ Isolate: &isolate,
+ TTL: &ttl,
+ },
+ }
+}
+
+func convertToDeregisterInstance(namespace string, instance registry.ServiceInstance) *api.InstanceDeRegisterRequest {
+ return &api.InstanceDeRegisterRequest{
+ InstanceDeRegisterRequest: model.InstanceDeRegisterRequest{
+ Service: instance.GetServiceName(),
+ Namespace: namespace,
+ Host: instance.GetHost(),
+ Port: instance.GetPort(),
+ },
+ }
+}
diff --git a/registry/polaris/utils.go b/registry/polaris/utils.go
new file mode 100644
index 000000000..cdd6d97d1
--- /dev/null
+++ b/registry/polaris/utils.go
@@ -0,0 +1,45 @@
+/*
+ * 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 polaris
+
+import (
+ "fmt"
+ "strconv"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/registry"
+)
+
+type PolarisInstanceInfo struct {
+ instance registry.ServiceInstance
+ url *common.URL
+}
+
+func getInstanceKey(namespace string, instance registry.ServiceInstance) string {
+ return fmt.Sprintf("%s-%s-%s-%d", namespace, instance.GetServiceName(), instance.GetHost(), instance.GetPort())
+}
+
+// just copy from dubbo-go for nacos
+func getCategory(url *common.URL) string {
+ role, _ := strconv.Atoi(url.GetParam(constant.RegistryRoleKey, strconv.Itoa(constant.PolarisDefaultRoleType)))
+ category := common.DubboNodes[role]
+ return category
+}
diff --git a/remoting/polaris/builder.go b/remoting/polaris/builder.go
new file mode 100644
index 000000000..30ab31d9e
--- /dev/null
+++ b/remoting/polaris/builder.go
@@ -0,0 +1,160 @@
+/*
+ * 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 polaris
+
+import (
+ "errors"
+ "fmt"
+ "net"
+ "strconv"
+ "strings"
+ "sync"
+)
+
+import (
+ perrors "github.com/pkg/errors"
+
+ "github.com/polarismesh/polaris-go"
+ "github.com/polarismesh/polaris-go/api"
+ "github.com/polarismesh/polaris-go/pkg/config"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+)
+
+var (
+ once sync.Once
+ namesapce string
+ sdkCtx api.SDKContext
+ openPolarisAbility bool
+)
+
+var (
+ ErrorNoOpenPolarisAbility = errors.New("polaris ability not open")
+ ErrorSDKContextNotInit = errors.New("polaris SDKContext not init")
+)
+
+// GetConsumerAPI creates one polaris ConsumerAPI instance
+func GetConsumerAPI() (polaris.ConsumerAPI, error) {
+ if err := Check(); err != nil {
+ return nil, err
+ }
+
+ return polaris.NewConsumerAPIByContext(sdkCtx), nil
+}
+
+// GetProviderAPI creates one polaris ProviderAPI instance
+func GetProviderAPI() (polaris.ProviderAPI, error) {
+ if err := Check(); err != nil {
+ return nil, err
+ }
+
+ return polaris.NewProviderAPIByContext(sdkCtx), nil
+}
+
+// GetRouterAPI create one polaris RouterAPI instance
+func GetRouterAPI() (polaris.RouterAPI, error) {
+ if err := Check(); err != nil {
+ return nil, err
+ }
+
+ return polaris.NewRouterAPIByContext(sdkCtx), nil
+}
+
+// GetLimiterAPI creates one polaris LimiterAPI instance
+func GetLimiterAPI() (polaris.LimitAPI, error) {
+ if err := Check(); err != nil {
+ return nil, err
+ }
+
+ return polaris.NewLimitAPIByContext(sdkCtx), nil
+}
+
+func Check() error {
+ if !openPolarisAbility {
+ return ErrorNoOpenPolarisAbility
+ }
+ if sdkCtx == nil {
+ return ErrorSDKContextNotInit
+ }
+ return nil
+}
+
+// GetNamespace gets user defined namespace info
+func GetNamespace() string {
+ return namesapce
+}
+
+// InitSDKContext inits polaris SDKContext by URL
+func InitSDKContext(url *common.URL) error {
+ if url == nil {
+ return errors.New("url is empty")
+ }
+
+ openPolarisAbility = true
+
+ var rerr error
+ once.Do(func() {
+ addresses := strings.Split(url.Location, ",")
+ serverConfigs := make([]string, 0, len(addresses))
+ for _, addr := range addresses {
+ ip, portStr, err := net.SplitHostPort(addr)
+ if err != nil {
+ rerr = perrors.WithMessagef(err, "split [%s] ", addr)
+ }
+ port, _ := strconv.Atoi(portStr)
+ serverConfigs = append(serverConfigs, fmt.Sprintf("%s:%d", ip, uint64(port)))
+ }
+
+ polarisConf := config.NewDefaultConfiguration(serverConfigs)
+ _sdkCtx, err := api.InitContextByConfig(polarisConf)
+ rerr = err
+ sdkCtx = _sdkCtx
+ namesapce = url.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace)
+ })
+
+ return rerr
+}
+
+func mergePolarisConfiguration(easy, complexConf config.Configuration) {
+
+ easySvrList := easy.GetGlobal().GetServerConnector().GetAddresses()
+
+ complexSvrList := complexConf.GetGlobal().GetServerConnector().GetAddresses()
+
+ result := make(map[string]bool)
+
+ for i := range complexSvrList {
+ result[complexSvrList[i]] = true
+ }
+
+ for i := range easySvrList {
+ if _, exist := result[easySvrList[i]]; !exist {
+ result[easySvrList[i]] = true
+ }
+ }
+
+ finalSvrList := make([]string, 0)
+ for k := range result {
+ finalSvrList = append(finalSvrList, k)
+ }
+
+ complexConf.GetGlobal().GetServerConnector().SetAddresses(finalSvrList)
+}
diff --git a/remoting/polaris/builder_test.go b/remoting/polaris/builder_test.go
new file mode 100644
index 000000000..6a27f3307
--- /dev/null
+++ b/remoting/polaris/builder_test.go
@@ -0,0 +1,45 @@
+/*
+ * 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 polaris
+
+import (
+ "net/url"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common"
+)
+
+func TestGetPolarisConfigByUrl(t *testing.T) {
+ regurl := getRegUrl()
+ err := InitSDKContext(regurl)
+
+ assert.Nil(t, err)
+ assert.ElementsMatch(t, []string{"127.0.0.1:8091"}, sdkCtx.GetConfig().GetGlobal().GetServerConnector().GetAddresses(), "server address")
+}
+
+func getRegUrl() *common.URL {
+ regurlMap := url.Values{}
+ regurl, _ := common.NewURL("registry://127.0.0.1:8091", common.WithParams(regurlMap))
+ return regurl
+}
diff --git a/remoting/polaris/parser/parser.go b/remoting/polaris/parser/parser.go
new file mode 100644
index 000000000..c8bec14f9
--- /dev/null
+++ b/remoting/polaris/parser/parser.go
@@ -0,0 +1,125 @@
+/*
+ * 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 parser
+
+import (
+ "encoding/json"
+ "regexp"
+ "strconv"
+ "strings"
+)
+
+import (
+ "github.com/dubbogo/gost/log/logger"
+
+ "github.com/oliveagle/jsonpath"
+)
+
+const (
+ _pefixParam = "param"
+ _prefixParamArr = "param["
+)
+
+var (
+ _arrayRegx, _ = regexp.Compile(`"^.+\\[[0-9]+\\]"`)
+)
+
+// ParseArgumentsByExpression follow https://goessner.net/articles/JsonPath/
+//
+// {
+// "store":{
+// "book":[
+// {
+// "category":"reference",
+// "author":"Nigel Rees",
+// "title":"Sayings of the Century",
+// "price":8.95
+// },
+// {
+// "category":"fiction",
+// "author":"Evelyn Waugh",
+// "title":"Sword of Honor",
+// "price":12.99
+// },
+// {
+// "category":"fiction",
+// "author":"Herman Melville",
+// "title":"Moby Dick",
+// "isbn":"0-553-21311-3",
+// "price":8.99
+// },
+// {
+// "category":"fiction",
+// "author":"J. R. R. Tolkien",
+// "title":"The Lord of the Rings",
+// "isbn":"0-395-19395-8",
+// "price":22.99
+// }
+// ],
+// "bicycle":{
+// "color":"red",
+// "price":19.95
+// }
+// }
+// }
+//
+// examples
+// - case 1: param.$.store.book[*].author
+func ParseArgumentsByExpression(key string, parameters []any) any {
+ index, key := resolveIndex(key)
+ if index == -1 || index >= len(parameters) {
+ logger.Errorf("[Parser][Polaris] invalid expression for : %s", key)
+ return nil
+ }
+
+ data, err := json.Marshal(parameters[index])
+ if err != nil {
+ logger.Errorf("[Parser][Polaris] marshal parameter %+v fail : %+v", parameters[index], err)
+ return nil
+ }
+ var searchVal any
+ _ = json.Unmarshal(data, &searchVal)
+ res, err := jsonpath.JsonPathLookup(searchVal, key)
+ if err != nil {
+ logger.Errorf("[Parser][Polaris] invalid do json path lookup by key : %s, err : %+v", key, err)
+ }
+
+ return res
+}
+
+func resolveIndex(key string) (int, string) {
+ if strings.HasPrefix(key, _prefixParamArr) {
+ // param[0].$.
+ endIndex := strings.Index(key, "]")
+ indexStr := key[len(_prefixParamArr):endIndex]
+ index, err := strconv.ParseInt(indexStr, 10, 32)
+ if err != nil {
+ return -1, ""
+ }
+ startIndex := endIndex + 2
+ if rune(key[endIndex+1]) != rune('.') {
+ startIndex = endIndex + 1
+ }
+ return int(index), key[startIndex:]
+ } else if strings.HasPrefix(key, _pefixParam) {
+ key = strings.TrimPrefix(key, _pefixParam+".")
+ return 0, strings.TrimPrefix(key, _pefixParam+".")
+ }
+
+ return -1, ""
+}
diff --git a/remoting/polaris/parser/parser_test.go b/remoting/polaris/parser/parser_test.go
new file mode 100644
index 000000000..bdb4615af
--- /dev/null
+++ b/remoting/polaris/parser/parser_test.go
@@ -0,0 +1,168 @@
+/*
+ * 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 parser
+
+import (
+ "encoding/json"
+ "testing"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+)
+
+var (
+ testDataStore = `
+ {
+ "book":[
+ {
+ "category":"reference",
+ "author":"Nigel Rees",
+ "title":"Sayings of the Century",
+ "price":8.95
+ },
+ {
+ "category":"fiction",
+ "author":"Evelyn Waugh",
+ "title":"Sword of Honor",
+ "price":12.99
+ },
+ {
+ "category":"fiction",
+ "author":"Herman Melville",
+ "title":"Moby Dick",
+ "isbn":"0-553-21311-3",
+ "price":8.99
+ },
+ {
+ "category":"fiction",
+ "author":"J. R. R. Tolkien",
+ "title":"The Lord of the Rings",
+ "isbn":"0-395-19395-8",
+ "price":22.99
+ }
+ ],
+ "bicycle":{
+ "color":"red",
+ "price":19.95
+ }
+ }
+ `
+
+ testDataBicyle = `
+ {
+ "color":"red",
+ "price":19.95
+ }
+ `
+)
+
+func TestParseArgumentsByExpression(t *testing.T) {
+
+ var (
+ argStore, argBicyle any
+ )
+
+ json.Unmarshal([]byte(testDataStore), &argStore)
+ json.Unmarshal([]byte(testDataBicyle), &argBicyle)
+
+ t.Run("test-case-1", func(t *testing.T) {
+ ret := ParseArgumentsByExpression("param.$.book[0].category", []any{argStore})
+ assert.Equal(t, "reference", ret)
+ })
+
+ t.Run("test-case-2", func(t *testing.T) {
+ ret := ParseArgumentsByExpression("param[0].$.book[0].category", []any{argStore, argBicyle})
+ assert.Equal(t, "reference", ret)
+ })
+
+ t.Run("test-case-2", func(t *testing.T) {
+ ret := ParseArgumentsByExpression("param[1].$.color", []any{argStore, argBicyle})
+ assert.Equal(t, "red", ret)
+ })
+
+ t.Run("test-case-3", func(t *testing.T) {
+ ret := ParseArgumentsByExpression("param.$.color", []any{argBicyle})
+ assert.Equal(t, "red", ret)
+ })
+
+}
+
+func Test_resolveIndex(t *testing.T) {
+ type args struct {
+ key string
+ }
+ tests := []struct {
+ name string
+ args args
+ want int
+ want1 string
+ }{
+ {
+ name: "case-1",
+ args: args{
+ key: "param.$.key",
+ },
+ want: 0,
+ want1: "$.key",
+ },
+ {
+ name: "case-2",
+ args: args{
+ key: "param[1].$.key",
+ },
+ want: 1,
+ want1: "$.key",
+ },
+ {
+ name: "case-3",
+ args: args{
+ key: "param[10].$.key",
+ },
+ want: 10,
+ want1: "$.key",
+ },
+ {
+ name: "case-4",
+ args: args{
+ key: "param[11]$.key",
+ },
+ want: 11,
+ want1: "$.key",
+ },
+ {
+ name: "case-5",
+ args: args{
+ key: "param[11]key",
+ },
+ want: 11,
+ want1: "key",
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, got1 := resolveIndex(tt.args.key)
+ if got != tt.want {
+ t.Errorf("resolveIndex() got = %v, want %v", got, tt.want)
+ }
+ if got1 != tt.want1 {
+ t.Errorf("resolveIndex() got1 = %v, want %v", got1, tt.want1)
+ }
+ })
+ }
+}
diff --git a/remoting/polaris/polaris.yaml b/remoting/polaris/polaris.yaml
new file mode 100644
index 000000000..2347aad37
--- /dev/null
+++ b/remoting/polaris/polaris.yaml
@@ -0,0 +1,84 @@
+global:
+ system:
+ mode: 0
+ discoverCluster:
+ namespace: Polaris
+ service: polaris.discover
+ refreshInterval: 10m
+ healthCheckCluster:
+ namespace: Polaris
+ service: polaris.healthcheck
+ refreshInterval: 10m
+ monitorCluster:
+ namespace: Polaris
+ service: polaris.monitor
+ refreshInterval: 10m
+ api:
+ timeout: 1s
+ reportInterval: 10m
+ maxRetryTimes: 5
+ retryInterval: 1s
+ serverConnector:
+ protocol: grpc
+ connectTimeout: 500ms
+ messageTimeout: 1s
+ connectionIdleTimeout: 1s
+ requestQueueSize: 1000
+ serverSwitchInterval: 10m
+ plugin:
+ grpc:
+ maxCallRecvMsgSize: 52428800
+ statReporter:
+ enable: false
+consumer:
+ localCache:
+ type: inmemory
+ serviceExpireTime: 24h
+ serviceRefreshInterval: 2s
+ persistDir: $HOME/polaris/backup
+ persistMaxWriteRetry: 5
+ persistMaxReadRetry: 1
+ persistRetryInterval: 1s
+ persistAvailableInterval: 60s
+ startUseFileCache: true
+ serviceRouter:
+ chain:
+ - ruleBasedRouter
+ - nearbyBasedRouter
+ plugin:
+ nearbyBasedRouter:
+ matchLevel: zone
+ ruleBasedRouter: {}
+ percentOfMinInstances: 0
+ enableRecoverAll: true
+ loadbalancer:
+ type: weightedRandom
+ plugin:
+ ringHash:
+ vnodeCount: 500
+ circuitBreaker:
+ enable: true
+ checkPeriod: 30s
+ requestCountAfterHalfOpen: 10
+ sleepWindow: 30s
+ successCountAfterHalfOpen: 8
+ recoverWindow: 60s
+ recoverNumBuckets: 10
+ chain:
+ - errorCount
+ - errorRate
+ plugin:
+ errorCount:
+ continuousErrorThreshold: 10
+ metricNumBuckets: 10
+ metricStatTimeWindow: 1m0s
+ errorRate:
+ errorRateThreshold: 0.5
+ metricNumBuckets: 5
+ metricStatTimeWindow: 1m0s
+ requestVolumeThreshold: 10
+ subscribe:
+ type: subscribeLocalChannel
+ plugin:
+ subscribeLocalChannel:
+ channelBufferSize: 50
\ No newline at end of file
diff --git a/tools/dubbo-go-schema/dubbo-go.json b/tools/dubbo-go-schema/dubbo-go.json
index 292d201ef..870a2ed3a 100644
--- a/tools/dubbo-go-schema/dubbo-go.json
+++ b/tools/dubbo-go-schema/dubbo-go.json
@@ -93,6 +93,7 @@
"enum": [
"nacos",
"etcdv3",
+ "polaris",
"xds",
"zookeeper",
"service-discovery-registry"