This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/main by this push:
new d7ce4d445 support conditional routing with multiple destinations,
customize conditional routing priorities and operation in route fail (#2685)
d7ce4d445 is described below
commit d7ce4d445b283b3e6725e477971666ec9224ffca
Author: YarBor <[email protected]>
AuthorDate: Thu Jun 6 10:21:35 2024 +0800
support conditional routing with multiple destinations, customize
conditional routing priorities and operation in route fail (#2685)
---
cluster/router/condition/dynamic_router.go | 191 +++++++++++++++++++----
cluster/router/condition/route.go | 75 ++++-----
cluster/router/condition/router_test.go | 240 +++++++++++++++++++++++++++--
cluster/utils/version.go | 103 +++++++++++++
common/constant/key.go | 1 +
common/constant/version.go | 7 +-
config/router_config.go | 16 ++
7 files changed, 547 insertions(+), 86 deletions(-)
diff --git a/cluster/router/condition/dynamic_router.go
b/cluster/router/condition/dynamic_router.go
index 90245b910..38e718736 100644
--- a/cluster/router/condition/dynamic_router.go
+++ b/cluster/router/condition/dynamic_router.go
@@ -18,16 +18,15 @@
package condition
import (
+ "fmt"
+ "sort"
"strconv"
"strings"
"sync"
)
import (
- "github.com/dubbogo/gost/log/logger"
-)
-
-import (
+ "dubbo.apache.org/dubbo-go/v3/cluster/utils"
"dubbo.apache.org/dubbo-go/v3/common"
conf "dubbo.apache.org/dubbo-go/v3/common/config"
"dubbo.apache.org/dubbo-go/v3/common/constant"
@@ -35,22 +34,75 @@ import (
"dubbo.apache.org/dubbo-go/v3/config_center"
"dubbo.apache.org/dubbo-go/v3/protocol"
"dubbo.apache.org/dubbo-go/v3/remoting"
+
+ "github.com/dubbogo/gost/log/logger"
+
+ "gopkg.in/yaml.v2"
)
+type conditionRoute []*StateRouter
+
+func (p conditionRoute) route(invokers []protocol.Invoker, url *common.URL,
invocation protocol.Invocation) []protocol.Invoker {
+ if len(invokers) == 0 || len(p) == 0 {
+ return invokers
+ }
+ for _, router := range p {
+ invokers, _ = router.Route(invokers, url, invocation)
+ if len(invokers) == 0 {
+ break
+ }
+ }
+ return invokers
+}
+
+type multiplyConditionRoute []*StateRouter
+
+func (m multiplyConditionRoute) route(invokers []protocol.Invoker, url
*common.URL, invocation protocol.Invocation) []protocol.Invoker {
+ if len(invokers) == 0 || len(m) == 0 {
+ return invokers
+ }
+ for _, router := range m {
+ matchInvokers, isMatch := router.Route(invokers, url,
invocation)
+ if !isMatch || (len(matchInvokers) == 0 && !router.force) {
+ continue
+ }
+ return matchInvokers
+ }
+ return []protocol.Invoker{}
+}
+
+type condRouter interface {
+ route(invokers []protocol.Invoker, url *common.URL, invocation
protocol.Invocation) []protocol.Invoker
+}
+
type DynamicRouter struct {
- conditionRouters []*StateRouter
- routerConfig *config.RouterConfig
+ mu sync.RWMutex
+ force bool
+ enable bool
+ conditionRouter condRouter
}
func (d *DynamicRouter) Route(invokers []protocol.Invoker, url *common.URL,
invocation protocol.Invocation) []protocol.Invoker {
- if len(invokers) == 0 || len(d.conditionRouters) == 0 {
+ if len(invokers) == 0 {
return invokers
}
- for _, router := range d.conditionRouters {
- invokers = router.Route(invokers, url, invocation)
+ d.mu.RLock()
+ force, enable, cr := d.force, d.enable, d.conditionRouter
+ d.mu.RUnlock()
+
+ if !enable {
+ return invokers
+ }
+ if cr != nil {
+ res := cr.route(invokers, url, invocation)
+ if len(res) == 0 && !force {
+ return invokers
+ }
+ return res
+ } else {
+ return invokers
}
- return invokers
}
func (d *DynamicRouter) URL() *common.URL {
@@ -58,45 +110,126 @@ func (d *DynamicRouter) URL() *common.URL {
}
func (d *DynamicRouter) Process(event *config_center.ConfigChangeEvent) {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+
if event.ConfigType == remoting.EventTypeDel {
- d.routerConfig = nil
- d.conditionRouters = make([]*StateRouter, 0)
+ d.conditionRouter = nil
} else {
- routerConfig, err := parseRoute(event.Value.(string))
+ rc, force, enable, err :=
generateCondition(event.Value.(string))
if err != nil {
- logger.Warnf("[condition router]Build a new condition
route config error, %+v and we will use the original condition rule
configuration.", err)
- return
+ logger.Errorf("generate condition error: %v", err)
+ d.conditionRouter = nil
+ } else {
+ d.force, d.enable, d.conditionRouter = force, enable, rc
+ }
+ }
+}
+
+/*
+to check configVersion, here need decode twice.
+From a performance perspective, decoding from a string and decoding from a
map[string]interface{}
+cost nearly same (a few milliseconds).
+
+To keep the code simpler,
+here use yaml-to-map and yaml-to-struct, not yaml-to-map-to-struct
+
+generateCondition @return(router,force,enable,error)
+*/
+func generateCondition(rawConfig string) (condRouter, bool, bool, error) {
+ m := map[string]interface{}{}
+
+ err := yaml.Unmarshal([]byte(rawConfig), m)
+ if err != nil {
+ return nil, false, false, err
+ }
+
+ rawVersion, ok := m["configVersion"]
+ if !ok {
+ return nil, false, false, fmt.Errorf("miss `ConfigVersion` in
%s", rawConfig)
+ }
+
+ version, ok := rawVersion.(string)
+ if !ok {
+ return nil, false, false, fmt.Errorf("`ConfigVersion` should be
of type `string`, got %T", rawVersion)
+ }
+
+ v, parseErr := utils.ParseVersion(version)
+ if parseErr != nil {
+ return nil, false, false, fmt.Errorf("invalid version %s: %s",
version, parseErr.Error())
+ }
+
+ switch {
+ case v.Equal(utils.V3_1) || v.Greater(utils.V3_1):
+ return generateMultiConditionRoute(rawConfig)
+ case v.Less(utils.V3_1):
+ return generateConditionsRoute(rawConfig)
+ default:
+ panic("invalid version compare return")
+ }
+}
+
+func generateMultiConditionRoute(rawConfig string) (multiplyConditionRoute,
bool, bool, error) {
+ routerConfig, err := parseMultiConditionRoute(rawConfig)
+ if err != nil {
+ logger.Warnf("[condition router]Build a new condition route
config error, %s and we will use the original condition rule configuration.",
err.Error())
+ return nil, false, false, err
+ }
+
+ force, enable := routerConfig.Enabled, routerConfig.Force
+ if !enable {
+ return nil, false, false, nil
+ }
+
+ conditionRouters := make([]*StateRouter, 0,
len(routerConfig.Conditions))
+ for _, conditionRule := range routerConfig.Conditions {
+ url, err := common.NewURL("condition://")
+ if err != nil {
+ return nil, false, false, err
}
- d.routerConfig = routerConfig
- conditions, err := generateConditions(d.routerConfig)
+ url.AddParam(constant.RuleKey, conditionRule.Rule)
+ url.AddParam(constant.ForceKey,
strconv.FormatBool(conditionRule.Force))
+ url.AddParam(constant.PriorityKey,
strconv.FormatInt(int64(conditionRule.Priority), 10))
+ conditionRoute, err := NewConditionStateRouter(url)
if err != nil {
- logger.Warnf("[condition router]Build a new condition
route config error, %+v and we will use the original condition rule
configuration.", err)
- return
+ return nil, false, false, err
}
- d.conditionRouters = conditions
+ conditionRouters = append(conditionRouters, conditionRoute)
}
+
+ sort.Slice(conditionRouters, func(i, j int) bool {
+ return conditionRouters[i].priority >
conditionRouters[j].priority
+ })
+ return conditionRouters, force, enable, nil
}
-func generateConditions(routerConfig *config.RouterConfig) ([]*StateRouter,
error) {
- if routerConfig == nil {
- return make([]*StateRouter, 0), nil
+func generateConditionsRoute(rawConfig string) (conditionRoute, bool, bool,
error) {
+ routerConfig, err := parseConditionRoute(rawConfig)
+ if err != nil {
+ logger.Warnf("[condition router]Build a new condition route
config error, %s and we will use the original condition rule configuration.",
err.Error())
+ return nil, false, false, err
}
+
+ force, enable := *routerConfig.Enabled, *routerConfig.Force
+ if !enable {
+ return nil, false, false, nil
+ }
+
conditionRouters := make([]*StateRouter, 0,
len(routerConfig.Conditions))
for _, conditionRule := range routerConfig.Conditions {
url, err := common.NewURL("condition://")
if err != nil {
- return nil, err
+ return nil, false, false, err
}
url.AddParam(constant.RuleKey, conditionRule)
url.AddParam(constant.ForceKey,
strconv.FormatBool(*routerConfig.Force))
- url.AddParam(constant.EnabledKey,
strconv.FormatBool(*routerConfig.Enabled))
conditionRoute, err := NewConditionStateRouter(url)
if err != nil {
- return nil, err
+ return nil, false, false, err
}
conditionRouters = append(conditionRouters, conditionRoute)
}
- return conditionRouters, nil
+ return conditionRouters, force, enable, nil
}
// ServiceRouter is Service level router
@@ -142,7 +275,6 @@ type ApplicationRouter struct {
DynamicRouter
application string
currentApplication string
- mu sync.Mutex
}
func NewApplicationRouter() *ApplicationRouter {
@@ -184,9 +316,6 @@ func (a *ApplicationRouter) Notify(invokers
[]protocol.Invoker) {
return
}
- a.mu.Lock()
- defer a.mu.Unlock()
-
if providerApplicaton != a.application {
if a.application != "" {
dynamicConfiguration.RemoveListener(strings.Join([]string{a.application,
constant.ConditionRouterRuleSuffix}, ""), a)
diff --git a/cluster/router/condition/route.go
b/cluster/router/condition/route.go
index 26f2c2578..7d3074af2 100644
--- a/cluster/router/condition/route.go
+++ b/cluster/router/condition/route.go
@@ -41,17 +41,14 @@ import (
)
var (
- routePattern = regexp.MustCompile("([&!=,]*)\\s*([^&!=,\\s]+)")
-
- illegalMsg = "Illegal route rule \"%s\", The error char '%s' before
'%s'"
-
+ routePattern = regexp.MustCompile("([&!=,]*)\\s*([^&!=,\\s]+)")
+ illegalMsg = "Illegal route rule \"%s\", The error char '%s'
before '%s'"
matcherFactories = make([]matcher.ConditionMatcherFactory, 0, 8)
-
- once sync.Once
+ once sync.Once
)
type StateRouter struct {
- enable bool
+ priority int64
force bool
url *common.URL
whenCondition map[string]matcher.Matcher
@@ -66,41 +63,38 @@ func NewConditionStateRouter(url *common.URL)
(*StateRouter, error) {
}
force := url.GetParamBool(constant.ForceKey, false)
- enable := url.GetParamBool(constant.EnabledKey, true)
+ priority := url.GetParamInt(constant.PriorityKey,
constant.DefaultPriority)
c := &StateRouter{
- url: url,
- force: force,
- enable: enable,
+ url: url,
+ force: force,
+ priority: priority,
}
- if enable {
- when, then, err := generateMatcher(url)
- if err != nil {
- return nil, err
- }
- c.whenCondition = when
- c.thenCondition = then
+ when, then, err := generateMatcher(url)
+ if err != nil {
+ return nil, err
}
+ c.whenCondition = when
+ c.thenCondition = then
return c, nil
}
// Route Determine the target invokers list.
-func (s *StateRouter) Route(invokers []protocol.Invoker, url *common.URL,
invocation protocol.Invocation) []protocol.Invoker {
- if !s.enable {
- return invokers
- }
-
+// condition rule like `self_condition => peers_condition `
+//
+// @return active_peers_invokers, Is_self_condition_match_success
+func (s *StateRouter) Route(invokers []protocol.Invoker, url *common.URL,
invocation protocol.Invocation) ([]protocol.Invoker, bool) {
if len(invokers) == 0 {
- return invokers
+ return invokers, false
}
if !s.matchWhen(url, invocation) {
- return invokers
+ return invokers, false
}
if len(s.thenCondition) == 0 {
logger.Warn("condition state router thenCondition is empty")
- return []protocol.Invoker{}
+ return []protocol.Invoker{}, true
}
var result = make([]protocol.Invoker, 0, len(invokers))
@@ -110,22 +104,7 @@ func (s *StateRouter) Route(invokers []protocol.Invoker,
url *common.URL, invoca
}
}
- if len(result) != 0 {
- return result
- } else if s.force {
- logger.Warn("execute condition state router result list is
empty. and force=true")
- return result
- }
-
- return invokers
-}
-
-func (s *StateRouter) URL() *common.URL {
- return s.url
-}
-
-func (s *StateRouter) Priority() int64 {
- return 0
+ return result, true
}
func (s *StateRouter) matchWhen(url *common.URL, invocation
protocol.Invocation) bool {
@@ -318,7 +297,7 @@ func (a byPriority) Len() int { return len(a) }
func (a byPriority) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a byPriority) Less(i, j int) bool { return a[i].Priority() <
a[j].Priority() }
-func parseRoute(routeContent string) (*config.RouterConfig, error) {
+func parseConditionRoute(routeContent string) (*config.RouterConfig, error) {
routeDecoder := yaml.NewDecoder(strings.NewReader(routeContent))
routerConfig := &config.RouterConfig{}
err := routeDecoder.Decode(routerConfig)
@@ -327,3 +306,13 @@ func parseRoute(routeContent string)
(*config.RouterConfig, error) {
}
return routerConfig, nil
}
+
+func parseMultiConditionRoute(routeContent string) (*config.ConditionRouter,
error) {
+ routeDecoder := yaml.NewDecoder(strings.NewReader(routeContent))
+ routerConfig := &config.ConditionRouter{}
+ err := routeDecoder.Decode(routerConfig)
+ if err != nil {
+ return nil, err
+ }
+ return routerConfig, nil
+}
diff --git a/cluster/router/condition/router_test.go
b/cluster/router/condition/router_test.go
index 451306bc0..d8de7c374 100644
--- a/cluster/router/condition/router_test.go
+++ b/cluster/router/condition/router_test.go
@@ -18,6 +18,7 @@
package condition
import (
+ "dubbo.apache.org/dubbo-go/v3/remoting"
"testing"
)
@@ -213,7 +214,7 @@ func TestRouteMatchFilter(t *testing.T) {
router, err := NewConditionStateRouter(url)
assert.Nil(t, err)
- filteredInvokers := router.Route(invokerList,
data.comsumerURL, rpcInvocation)
+ filteredInvokers, _ := router.Route(invokerList,
data.comsumerURL, rpcInvocation)
resVal := len(filteredInvokers)
assert.Equal(t, data.wantVal, resVal)
})
@@ -239,7 +240,7 @@ func TestRouterMethodRoute(t *testing.T) {
wantVal: true,
},
{
- name: "Exactly one method, match",
+ name: "Exactly one method, Route",
consumerURL: remoteConsumerAddr + "?methods=getFoo",
rule: "methods=getFoo => host = 1.2.3.4",
@@ -406,7 +407,7 @@ func TestRouteReturn(t *testing.T) {
router, err := NewConditionStateRouter(url)
assert.Nil(t, err)
- filterInvokers := router.Route(invokers, consumerURL,
rpcInvocation)
+ filterInvokers, _ := router.Route(invokers,
consumerURL, rpcInvocation)
resVal := len(filterInvokers)
assert.Equal(t, data.wantVal, resVal)
@@ -478,7 +479,7 @@ func TestRouteArguments(t *testing.T) {
rpcInvocation := invocation.NewRPCInvocation("getBar",
arguments, nil)
- filterInvokers := router.Route(invokerList,
consumerURL, rpcInvocation)
+ filterInvokers, _ := router.Route(invokerList,
consumerURL, rpcInvocation)
resVal := len(filterInvokers)
assert.Equal(t, data.wantVal, resVal)
@@ -558,7 +559,7 @@ func TestRouteAttachments(t *testing.T) {
router, err := NewConditionStateRouter(url)
assert.Nil(t, err)
- filterInvokers := router.Route(invokerList,
consumerURL, rpcInvocation)
+ filterInvokers, _ := router.Route(invokerList,
consumerURL, rpcInvocation)
resVal := len(filterInvokers)
assert.Equal(t, data.wantVal, resVal)
@@ -647,7 +648,7 @@ func TestRouteRangePattern(t *testing.T) {
router, err := NewConditionStateRouter(url)
assert.Nil(t, err)
- filterInvokers := router.Route(invokerList,
consumerURL, rpcInvocation)
+ filterInvokers, _ := router.Route(invokerList,
consumerURL, rpcInvocation)
resVal := len(filterInvokers)
assert.Equal(t, data.wantVal, resVal)
@@ -678,7 +679,7 @@ func TestRouteMultipleConditions(t *testing.T) {
wantVal int
}{
{
- name: "All conditions match",
+ name: "All conditions Route",
argument: "a",
consumerURL: localConsumerAddr +
"?application=consumer_app",
rule: "application=consumer_app&arguments[0]=a
=> host = 127.0.0.1",
@@ -686,7 +687,7 @@ func TestRouteMultipleConditions(t *testing.T) {
wantVal: 2,
},
{
- name: "One of the conditions does not match",
+ name: "One of the conditions does not Route",
argument: "a",
consumerURL: localConsumerAddr +
"?application=another_consumer_app",
rule: "application=consumer_app&arguments[0]=a
=> host = 127.0.0.1",
@@ -711,7 +712,7 @@ func TestRouteMultipleConditions(t *testing.T) {
rpcInvocation := invocation.NewRPCInvocation(method,
arguments, nil)
- filterInvokers := router.Route(invokerList,
consumerUrl, rpcInvocation)
+ filterInvokers, _ := router.Route(invokerList,
consumerUrl, rpcInvocation)
resVal := len(filterInvokers)
assert.Equal(t, data.wantVal, resVal)
})
@@ -739,6 +740,7 @@ func TestServiceRouter(t *testing.T) {
ccURL, _ := common.NewURL("mock://127.0.0.1:1111")
mockFactory := &config_center.MockDynamicConfigurationFactory{
Content: `
+configVersion: v3.0
scope: service
force: true
enabled: true
@@ -783,6 +785,7 @@ func TestApplicationRouter(t *testing.T) {
ccURL, _ := common.NewURL("mock://127.0.0.1:1111")
mockFactory := &config_center.MockDynamicConfigurationFactory{
Content: `
+configVersion: V3.0
scope: application
force: true
enabled: true
@@ -805,3 +808,222 @@ conditions:
invokers = router.Route(invokerList, consumerURL, rpcInvocation)
assert.Equal(t, 3, len(invokers))
}
+
+var providerUrls = []string{
+ "dubbo://127.0.0.1/com.foo.BarService",
+ "dubbo://127.0.0.1/com.foo.BarService",
+ "dubbo://127.0.0.1/com.foo.BarService?env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=beijing&env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou&env=gray",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou&env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou&env=normal",
+ "dubbo://127.0.0.1/com.foo.BarService?region=hangzhou&env=normal",
+ "dubbo://dubbo.apache.org/com.foo.BarService",
+ "dubbo://dubbo.apache.org/com.foo.BarService",
+ "dubbo://dubbo.apache.org/com.foo.BarService?env=normal",
+ "dubbo://dubbo.apache.org/com.foo.BarService?env=normal",
+ "dubbo://dubbo.apache.org/com.foo.BarService?env=normal",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing&env=gray",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=beijing&env=normal",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou&env=gray",
+ "dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou&env=gray",
+
"dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou&env=normal",
+
"dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou&env=normal",
+
"dubbo://dubbo.apache.org/com.foo.BarService?region=hangzhou&env=normal",
+}
+
+func buildInvokers() []protocol.Invoker {
+ res := make([]protocol.Invoker, 0, len(providerUrls))
+ for _, url := range providerUrls {
+ u, err := common.NewURL(url)
+ if err != nil {
+ panic(err)
+ }
+ res = append(res, protocol.NewBaseInvoker(u))
+ }
+ return res
+}
+
+func TestConditionRoutePriority(t *testing.T) {
+ ivks := buildInvokers()
+ ar := NewApplicationRouter()
+ ar.Process(&config_center.ConfigChangeEvent{Key: "", Value: `
+configVersion: v3.1
+scope: service
+force: true
+runtime: true
+enabled: true
+key: org.apache.dubbo.samples.CommentService
+conditionAction : true
+conditions:
+ - rule: method=getComment & env=gray => region=Hangzhou & env=gray
+ priority: 3
+ - rule: method=getComment & env=gray => region=beijing & env=gray
+ priority: 3
+ - rule: method=getComment & env=gray => region=$region & env=gray
+ priority: 3
+ - rule: method=getComment & env=normal => region=beijing
+ priority: 3
+ - rule: method=getComment => region=$region ######### match here
+ priority: 30
+ - rule: method=echo => region=$region
+ - rule: method=echo =>
+ force: true
+`, ConfigType: remoting.EventTypeUpdate})
+ consumerUrl, err :=
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=gray®ion=beijing")
+ if err != nil {
+ panic(err)
+ }
+ got := ar.Route(ivks, consumerUrl,
invocation.NewRPCInvocation("getComment", nil, nil))
+ expLen := 0
+ for _, ivk := range ivks {
+ if ivk.GetURL().GetParam("region", "") == "beijing" {
+ expLen++
+ }
+ }
+ assert.Equal(t, expLen, len(got))
+}
+
+func TestConditionRouteTrafficDisable(t *testing.T) {
+ ivks := buildInvokers()
+ ar := NewApplicationRouter()
+ ar.Process(&config_center.ConfigChangeEvent{Key: "", Value: `
+configVersion: v3.1
+scope: service
+force: true
+runtime: true
+enabled: true
+key: org.apache.dubbo.samples.CommentService
+conditionAction : true
+conditions:
+ - rule: method=getComment & env=gray => region=Hangzhou & env=gray
+ priority: 3
+ - rule: method=getComment & env=gray => region=beijing & env=gray
+ priority: 3
+ - rule: method=getComment & env=gray => region=$region & env=gray
+ priority: 3
+ - rule: method=getComment & env=normal => region=beijing
+ priority: 3
+ - rule: method=getComment => region=$region
+ priority: 30
+ - rule: method=echo =>
+ force: true
+ - rule: method=echo => region=$region
+`, ConfigType: remoting.EventTypeUpdate})
+ consumerUrl, err :=
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=gray®ion=beijing")
+ if err != nil {
+ panic(err)
+ }
+ got := ar.Route(ivks, consumerUrl, invocation.NewRPCInvocation("echo",
nil, nil))
+ assert.Equal(t, 0, len(got))
+}
+
+func TestConditionRouteRegionPriority(t *testing.T) {
+ ivks := buildInvokers()
+ ar := NewApplicationRouter()
+ ar.Process(&config_center.ConfigChangeEvent{Key: "", Value: `
+configVersion: v3.1
+scope: service
+force: true
+runtime: true
+enabled: true
+key: org.apache.dubbo.samples.CommentService
+conditionAction : true
+conditions:
+ - rule: => region=$region & env=$env
+ - rule: method=getComment & env=gray => env=$env
+ - rule: method=getComment & env=gray & region=beijing => region=beijing &
env=gray
+ - rule: method=getComment & env=gray => region=$region & env=gray
+ - rule: method=getComment & env=normal => region=beijing
+ - rule: method=echo =>
+ force: true
+ - rule: method=echo => region=$region
+`, ConfigType: remoting.EventTypeUpdate})
+ consumerUrl, err :=
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=gray®ion=beijing")
+ if err != nil {
+ panic(err)
+ }
+ got := ar.Route(ivks, consumerUrl,
invocation.NewRPCInvocation("getComment", nil, nil))
+ expLen := 0
+ for _, ivk := range ivks {
+ if ivk.GetURL().GetRawParam("env") ==
consumerUrl.GetRawParam("env") &&
+ ivk.GetURL().GetRawParam("region") ==
consumerUrl.GetRawParam("region") {
+ expLen++
+ }
+ }
+ assert.Equal(t, expLen, len(got))
+ consumerUrl, err =
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=gray®ion=hangzhou")
+ if err != nil {
+ panic(err)
+ }
+ got = ar.Route(ivks, consumerUrl,
invocation.NewRPCInvocation("getComment", nil, nil))
+ expLen = 0
+ for _, ivk := range ivks {
+ if ivk.GetURL().GetRawParam("env") ==
consumerUrl.GetRawParam("env") &&
+ ivk.GetURL().GetRawParam("region") ==
consumerUrl.GetRawParam("region") {
+ expLen++
+ }
+ }
+ assert.Equal(t, expLen, len(got))
+ consumerUrl, err =
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=normal®ion=shanghai")
+ if err != nil {
+ panic(err)
+ }
+ got = ar.Route(ivks, consumerUrl,
invocation.NewRPCInvocation("getComment", nil, nil))
+ expLen = 0
+ for _, ivk := range ivks {
+ if ivk.GetURL().GetRawParam("region") == "beijing" {
+ expLen++
+ }
+ }
+ assert.Equal(t, expLen, len(got))
+}
+
+func TestConditionRouteMatchFail(t *testing.T) {
+ ivks := buildInvokers()
+ ar := NewApplicationRouter()
+ ar.Process(&config_center.ConfigChangeEvent{Key: "", Value: `
+configVersion: v3.1
+scope: service
+force: false
+runtime: true
+enabled: true
+key: org.apache.dubbo.samples.CommentService
+conditionAction : true
+conditions:
+ - rule: => region=$region & env=$env & errTag=errTag
+ - rule: method=getComment & env=gray => env=$env
+ - rule: method=getComment & env=gray & region=beijing => region=beijing &
env=gray
+ - rule: method=getComment & env=gray => region=$region & env=gray
+ - rule: method=getComment & env=normal => region=beijing
+ - rule: method=echo =>
+ force: true
+ - rule: method=echo => region=$region
+`, ConfigType: remoting.EventTypeUpdate})
+ consumerUrl, err :=
common.NewURL("consumer://127.0.0.1/com.foo.BarService?env=gray®ion=beijing")
+ if err != nil {
+ panic(err)
+ }
+ got := ar.Route(ivks, consumerUrl,
invocation.NewRPCInvocation("errMethod", nil, nil))
+ assert.Equal(t, len(ivks), len(got))
+}
diff --git a/cluster/utils/version.go b/cluster/utils/version.go
new file mode 100644
index 000000000..a486bbb40
--- /dev/null
+++ b/cluster/utils/version.go
@@ -0,0 +1,103 @@
+/*
+ * 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 utils
+
+import (
+ "fmt"
+ "regexp"
+ "strconv"
+ "strings"
+)
+
+type Version []int
+
+var (
+ parseVersionRe = regexp.MustCompile(`^[Vv](\d+(\.\d+)*)$`)
+ V3_1, _ = ParseVersion("v3.1")
+)
+
+func ParseVersion(versionStr string) (Version, error) {
+ if versionContainsIllegalCharacters(versionStr) {
+ return nil, fmt.Errorf("illegal version string: %s , parse fail
", versionStr)
+ }
+ matches := parseVersionRe.FindStringSubmatch(versionStr)
+ if matches == nil {
+ return nil, fmt.Errorf("invalid version string format")
+ }
+
+ versionParts := strings.Split(matches[1], ".")
+ version := make([]int, len(versionParts))
+
+ for i, part := range versionParts {
+ number, err := strconv.Atoi(part)
+ if err != nil {
+ return nil, fmt.Errorf("invalid version number part:
%s", part)
+ }
+ version[i] = number
+ }
+
+ return version, nil
+}
+
+const (
+ versionEqual = 0
+ versionLess = -1
+ versionGreater = 1
+)
+
+func (v Version) Equal(target Version) bool {
+ return v.compareVersions(target) == versionEqual
+}
+func (v Version) Less(target Version) bool {
+ return v.compareVersions(target) == versionLess
+}
+func (v Version) Greater(target Version) bool {
+ return v.compareVersions(target) == versionGreater
+}
+
+func (v Version) compareVersions(target Version) int {
+ for i := 0; i < len(v) || i < len(target); i++ {
+ v1 := 0
+ if i < len(v) {
+ v1 = v[i]
+ }
+ v2 := 0
+ if i < len(target) {
+ v2 = target[i]
+ }
+ if v1 > v2 {
+ return versionGreater
+ } else if v1 < v2 {
+ return versionLess
+ }
+ }
+
+ if len(v) > len(target) {
+ return versionGreater
+ } else if len(v) < len(target) {
+ return versionLess
+ } else {
+ return versionEqual
+ }
+}
+
+var versionContainsIllegalCharactersRe = regexp.MustCompile(`^[Vv0-9.]+$`)
+
+func versionContainsIllegalCharacters(s string) bool {
+ return !versionContainsIllegalCharactersRe.MatchString(s)
+}
diff --git a/common/constant/key.go b/common/constant/key.go
index 96b3f1382..1e9959947 100644
--- a/common/constant/key.go
+++ b/common/constant/key.go
@@ -322,6 +322,7 @@ const (
ConditionServiceRouterFactoryKey = "service.condition"
ScriptRouterFactoryKey = "consumer.script"
ForceKey = "force"
+ PriorityKey = "priority"
Arguments = "arguments"
Attachments = "attachments"
Param = "param"
diff --git a/common/constant/version.go b/common/constant/version.go
index 883b09810..6ce664d8d 100644
--- a/common/constant/version.go
+++ b/common/constant/version.go
@@ -18,7 +18,8 @@
package constant
const (
- Version = "3.2.0" // apache/dubbo-go version
- Name = "dubbogo" // module name
- DATE = "2024/1/10" // release date
+ Version = "3.2.0" // apache/dubbo-go version
+ Name = "dubbogo" // module name
+ DATE = "2024/1/10" // release date
+ RouteVersion = "v3.1"
)
diff --git a/config/router_config.go b/config/router_config.go
index d4f7827d6..3cfff9cad 100644
--- a/config/router_config.go
+++ b/config/router_config.go
@@ -49,6 +49,22 @@ type Tag struct {
Addresses []string `yaml:"addresses"
json:"addresses,omitempty" property:"addresses"`
}
+type ConditionRule struct {
+ Rule string `validate:"required" yaml:"rule" json:"rule,omitempty"
property:"rule"`
+ Priority int `default:"0" yaml:"priority" json:"priority,omitempty"
property:"priority"`
+ Force bool `default:"false" yaml:"force" json:"force,omitempty"
property:"force"`
+}
+
+// ConditionRouter -- when RouteConfigVersion == v3.1, decode by this
+type ConditionRouter struct {
+ Scope string `validate:"required" yaml:"scope"
json:"scope,omitempty" property:"scope"` // must be chosen from `service` and
`application`.
+ Key string `validate:"required" yaml:"key"
json:"key,omitempty" property:"key"` // specifies which service or
application the rule body acts on.
+ Force bool `default:"false" yaml:"force"
json:"force,omitempty" property:"force"`
+ Runtime bool `default:"false" yaml:"runtime"
json:"runtime,omitempty" property:"runtime"`
+ Enabled bool `default:"true" yaml:"enabled"
json:"enabled,omitempty" property:"enabled"`
+ Conditions []ConditionRule `yaml:"conditions"
json:"conditions,omitempty" property:"conditions"`
+}
+
// Prefix dubbo.router
func (RouterConfig) Prefix() string {
return constant.RouterConfigPrefix
