Ftr: dynamic tag router

cluster/router/tag/router_rule.go

@@ -40,7 +40,7 @@ import (
 // RouterRule RouterRule config read from config file or config center
 type RouterRule struct {
 	router.BaseRouterRule `yaml:",inline""`
-	tags                  []tag
+	Tags                  []Tag
 	addressToTagNames     map[string][]string
 	tagNameToAddresses    map[string][]string
 }
@@ -52,18 +52,44 @@ func getRule(rawRule string) (*RouterRule, error) {
 		return r, err
 	}
 	r.RawRule = rawRule
-	// TODO init tags
+	r.init()
 	return r, nil
 }
 
+func (t *RouterRule) init() {
+	t.addressToTagNames = make(map[string][]string)
+	t.tagNameToAddresses = make(map[string][]string)
+	for _, tag := range t.Tags {
+		for _, address := range tag.Addresses {
+			t.addressToTagNames[address] = append(t.addressToTagNames[address], tag.Name)
+		}
+		t.tagNameToAddresses[tag.Name] = tag.Addresses
+	}
+}
+
 func (t *RouterRule) getAddresses() []string {
-	// TODO get all tag addresses
-	return nil
+	var result []string
+	for _, tag := range t.Tags {
+		result = append(result, tag.Addresses...)
+	}
+	return result
 }
 
 func (t *RouterRule) getTagNames() []string {
-	// TODO get all tag names
-	return nil
+	var result []string
+	for _, tag := range t.Tags {
+		result = append(result, tag.Name)
+	}
+	return result
+}
+
+func (t *RouterRule) hasTag(tag string) bool {
+	for _, t := range t.Tags {
+		if tag == t.Name {
+			return true
+		}
+	}
+	return false
 }
 
 func (t *RouterRule) getAddressToTagNames() map[string][]string {
@@ -74,10 +100,10 @@ func (t *RouterRule) getTagNameToAddresses() map[string][]string {
 	return t.tagNameToAddresses
 }
 
-func (t *RouterRule) getTags() []tag {
-	return t.tags
+func (t *RouterRule) getTags() []Tag {
+	return t.Tags
 }
 
-func (t *RouterRule) setTags(tags []tag) {
-	t.tags = tags
+func (t *RouterRule) setTags(tags []Tag) {
+	t.Tags = tags
 }

cluster/router/tag/router_rule_test.go

@@ -22,19 +22,56 @@ import (
 )
 
 import (
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
 )
 
-func TestGetRule(t *testing.T) {
+type RuleTestSuite struct {
+	suite.Suite
+	rule *RouterRule
+}
+
+func (suite *RuleTestSuite) SetupTest() {
+	var err error
 	yml := `
 scope: application
-runtime: true
 force: true
+runtime: false
+enabled: true
+priority: 1
+key: demo-provider
+tags:
+  - name: tag1
+    addresses: [ip1, ip2]
+  - name: tag2
+    addresses: [ip3, ip4]
 `
-	rule, e := getRule(yml)
-	assert.Nil(t, e)
-	assert.NotNil(t, rule)
-	assert.Equal(t, true, rule.Force)
-	assert.Equal(t, true, rule.Runtime)
-	assert.Equal(t, "application", rule.Scope)
+	suite.rule, err = getRule(yml)
+	suite.Nil(err)
+}
+
+func (suite *RuleTestSuite) TestGetRule() {
+	var err error
+	suite.Equal(true, suite.rule.Force)
+	suite.Equal(false, suite.rule.Runtime)
+	suite.Equal("application", suite.rule.Scope)
+	suite.Equal(1, suite.rule.Priority)
+	suite.Equal("demo-provider", suite.rule.Key)
+	suite.Nil(err)
+}
+
+func (suite *RuleTestSuite) TestGetTagNames() {
+	suite.Equal([]string{"tag1", "tag2"}, suite.rule.getTagNames())
+}
+
+func (suite *RuleTestSuite) TestGetAddresses() {
+	suite.Equal([]string{"ip1", "ip2", "ip3", "ip4"}, suite.rule.getAddresses())
+}
+
+func (suite *RuleTestSuite) TestHasTag() {
+	suite.Equal(true, suite.rule.hasTag("tag1"))
+	suite.Equal(false, suite.rule.hasTag("tag404"))
+}
+
+func TestRuleTestSuite(t *testing.T) {
+	suite.Run(t, new(RuleTestSuite))
 }

cluster/router/tag/tag.go

@@ -17,23 +17,23 @@
 
 package tag
 
-type tag struct {
-	name      string
-	addresses []string
+type Tag struct {
+	Name      string
+	Addresses []string
 }
 
-func (t *tag) getName() string {
-	return t.name
+func (t *Tag) getName() string {
+	return t.Name
 }
 
-func (t *tag) setName(name string) {
-	t.name = name
+func (t *Tag) setName(name string) {
+	t.Name = name
 }
 
-func (t *tag) getAddresses() []string {
-	return t.addresses
+func (t *Tag) getAddresses() []string {
+	return t.Addresses
 }
 
-func (t *tag) setAddresses(addresses []string) {
-	t.addresses = addresses
+func (t *Tag) setAddresses(addresses []string) {
+	t.Addresses = addresses
 }

cluster/router/tag/tag_router.go

@@ -81,40 +81,53 @@ func (c *tagRouter) Route(invokers []protocol.Invoker, url *common.URL, invocati
 		addresses, _ = tagRouterRuleCopy.getTagNameToAddresses()[tag]
 		// filter by dynamic tag group first
 		if len(addresses) > 0 {
-			// TODO filter invokers
-			result = nil
+			result = filterAddressMatches(invokers, addresses)
 			if len(result) > 0 || tagRouterRuleCopy.Force {
 				return result
 			}
 		} else {
 			// dynamic tag group doesn't have any item about the requested app OR it's null after filtered by
 			// dynamic tag group but force=false. check static tag
-			// TODO filter invokers
-			return result
+			cond := func(invoker protocol.Invoker) bool {
+				if invoker.GetUrl().GetParam(constant.Tagkey, "") == tag {
+					return true
+				}
+				return false
+			}
+			result = filterCondition(invokers, cond)
 		}
 		// If there's no tagged providers that can match the current tagged request. force.tag is set by default
 		// to false, which means it will invoke any providers without a tag unless it's explicitly disallowed.
 		if len(result) > 0 || isForceUseTag(url, invocation) {
 			return result
 		} else {
 			// FAILOVER: return all Providers without any tags.
-			// TODO filter invokers
-			return result
+			result = filterAddressNotMatches(invokers, tagRouterRuleCopy.getAddresses())
+			cond := func(invoker protocol.Invoker) bool {
+				if invoker.GetUrl().GetParam(constant.Tagkey, "") == "" {
+					return true
+				}
+				return false
+			}
+			return filterCondition(result, cond)
 		}
 	} else {
 		// return all addresses in dynamic tag group.
 		addresses = tagRouterRuleCopy.getAddresses()
 		if len(addresses) > 0 {
-			// TODO filter invokers
+			result = filterAddressNotMatches(invokers, addresses)
 			// 1. all addresses are in dynamic tag group, return empty list.
 			if len(result) == 0 {
 				return result
 			}
 			// 2. if there are some addresses that are not in any dynamic tag group, continue to filter using the
 			// static tag group.
 		}
-		// TODO filter invokers
-		return result
+		cond := func(invoker protocol.Invoker) bool {
+			localTag := invoker.GetUrl().GetParam(constant.Tagkey, "")
+			return localTag == "" || !(tagRouterRuleCopy.hasTag(localTag))
+		}
+		return filterCondition(result, cond)
 	}
 }
 
@@ -172,18 +185,51 @@ func isForceUseTag(url *common.URL, invocation protocol.Invocation) bool {
 	return false
 }
 
-func addressMatches(url *common.URL, addresses []string) bool {
-	return len(addresses) > 0 && checkAddressMatch(addresses, url.Ip, url.Port)
+func filterAddressMatches(invokers []protocol.Invoker, addresses []string) []protocol.Invoker {
+	var idx int
+	for _, invoker := range invokers {
+		url := invoker.GetUrl()
+		if !(len(addresses) > 0 && checkAddressMatch(addresses, url.Ip, url.Port)) {
+			continue
+		}
+		invokers[idx] = invoker
+		idx++
+	}
+	return invokers[:idx]
+}
+
+func filterAddressNotMatches(invokers []protocol.Invoker, addresses []string) []protocol.Invoker {
+	var idx int
+	for _, invoker := range invokers {
+		url := invoker.GetUrl()
+		if !(len(addresses) == 0 || !checkAddressMatch(addresses, url.Ip, url.Port)) {
+			continue
+		}
+		invokers[idx] = invoker
+		idx++
+	}
+	return invokers[:idx]
 }
 
-func addressNotMatches(url *common.URL, addresses []string) bool {
-	return len(addresses) == 0 || !checkAddressMatch(addresses, url.Ip, url.Port)
+func filterCondition(invokers []protocol.Invoker, condition func(protocol.Invoker) bool) []protocol.Invoker {
+	var idx int
+	for _, invoker := range invokers {
+		if !condition(invoker) {
+			continue
+		}
+		invokers[idx] = invoker
+		idx++
+	}
+	return invokers[:idx]
 }
 
 func checkAddressMatch(addresses []string, host, port string) bool {
 	for _, address := range addresses {
-		// TODO address parse
-		if address == (host + port) {
+		// TODO ip match
+		if address == host+":"+port {
+			return true
+		}
+		if address == constant.ANYHOST_VALUE+":"+port {
 			return true
 		}
 	}

cluster/router/tag/tag_router_test.go

@@ -19,6 +19,7 @@ package tag
 
 import (
 	"context"
+	"github.com/apache/dubbo-go/common/constant"
 	"testing"
 )
 
@@ -160,3 +161,25 @@ func TestTagRouterRouteNoForce(t *testing.T) {
 	invRst2 := tagRouter.Route(invokers, &u1, inv)
 	assert.Equal(t, 3, len(invRst2))
 }
+
+func TestFilterCondition(t *testing.T) {
+	u2, e2 := common.NewURL(tagRouterTestHangZhouUrl)
+	u3, e3 := common.NewURL(tagRouterTestShangHaiUrl)
+	u4, e4 := common.NewURL(tagRouterTestBeijingUrl)
+	assert.Nil(t, e2)
+	assert.Nil(t, e3)
+	assert.Nil(t, e4)
+	inv2 := NewMockInvoker(u2)
+	inv3 := NewMockInvoker(u3)
+	inv4 := NewMockInvoker(u4)
+	var invokers []protocol.Invoker
+	invokers = append(invokers, inv2, inv3, inv4)
+	cond := func(invoker protocol.Invoker) bool {
+		if invoker.GetUrl().GetParam(constant.Tagkey, "") == "beijing" {
+			return true
+		}
+		return false
+	}
+	res := filterCondition(invokers, cond)
+	assert.Equal(t, []protocol.Invoker{inv4}, res)
+}