Ftr: dynamic tag router

cluster/router/condition/listenable_router.go

@@ -85,7 +85,7 @@ func newListenableRouter(url *common.URL, ruleKey string) (*AppRouter, error) {
 	return l, nil
 }
 
-// Process Process config change event , generate routers and set them to the listenableRouter instance
+// Process Process config change event, generate routers and set them to the listenableRouter instance
 func (l *listenableRouter) Process(event *config_center.ConfigChangeEvent) {
 	logger.Infof("Notification of condition rule, change type is:[%s] , raw rule is:[%v]", event.ConfigType, event.Value)
 	if remoting.EventTypeDel == event.ConfigType {

cluster/router/tag/file.go

@@ -42,7 +42,7 @@ type FileTagRouter struct {
 	force      bool
 }
 
-// NewFileTagRouter Create file tag router instance with content ( from config file)
+// NewFileTagRouter Create file tag router instance with content (from config file)
 func NewFileTagRouter(content []byte) (*FileTagRouter, error) {
 	fileRouter := &FileTagRouter{}
 	rule, err := getRule(string(content))

cluster/router/tag/router_rule.go

@@ -22,9 +22,27 @@ import (
 	"github.com/apache/dubbo-go/common/yaml"
 )
 
+/**
+ * %YAML1.2
+ * ---
+ * force: true
+ * runtime: false
+ * enabled: true
+ * priority: 1
+ * key: demo-provider
+ * tags:
+ * - name: tag1
+ * addresses: [ip1, ip2]
+ * - name: tag2
+ * addresses: [ip3, ip4]
+ * ...
+ */
 // RouterRule RouterRule config read from config file or config center
 type RouterRule struct {
 	router.BaseRouterRule `yaml:",inline""`
+	Tags                  []Tag
+	addressToTagNames     map[string][]string
+	tagNameToAddresses    map[string][]string
 }
 
 func getRule(rawRule string) (*RouterRule, error) {
@@ -34,5 +52,58 @@ func getRule(rawRule string) (*RouterRule, error) {
 		return r, err
 	}
 	r.RawRule = rawRule
+	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 {
+	var result []string
+	for _, tag := range t.Tags {
+		result = append(result, tag.Addresses...)
+	}
+	return result
+}
+
+func (t *RouterRule) getTagNames() []string {
+	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 {
+	return t.addressToTagNames
+}
+
+func (t *RouterRule) getTagNameToAddresses() map[string][]string {
+	return t.tagNameToAddresses
+}
+
+func (t *RouterRule) getTags() []Tag {
+	return t.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

@@ -0,0 +1,39 @@
+/*
+ * 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 tag
+
+type Tag struct {
+	Name      string
+	Addresses []string
+}
+
+func (t *Tag) getName() string {
+	return t.Name
+}
+
+func (t *Tag) setName(name string) {
+	t.Name = name
+}
+
+func (t *Tag) getAddresses() []string {
+	return t.Addresses
+}
+
+func (t *Tag) setAddresses(addresses []string) {
+	t.Addresses = addresses
+}

cluster/router/tag/tag_router.go

@@ -18,6 +18,7 @@
 package tag
 
 import (
+	"fmt"
 	"strconv"
 )
 
@@ -28,13 +29,17 @@ import (
 import (
 	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/config_center"
 	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/remoting"
 )
 
 type tagRouter struct {
-	url      *common.URL
-	enabled  bool
-	priority int64
+	url           *common.URL
+	tagRouterRule *RouterRule
+	enabled       bool
+	priority      int64
 }
 
 func NewTagRouter(url *common.URL) (*tagRouter, error) {
@@ -59,7 +64,94 @@ func (c *tagRouter) Route(invokers []protocol.Invoker, url *common.URL, invocati
 	if len(invokers) == 0 {
 		return invokers
 	}
-	return filterUsingStaticTag(invokers, url, invocation)
+	// since the rule can be changed by config center, we should copy one to use.
+	tagRouterRuleCopy := c.tagRouterRule
+	if tagRouterRuleCopy == nil || !tagRouterRuleCopy.Valid || !tagRouterRuleCopy.Enabled {
+		return filterUsingStaticTag(invokers, url, invocation)
+	}
+	tag, ok := invocation.Attachments()[constant.Tagkey]
+	if !ok {
+		tag = url.GetParam(constant.Tagkey, "")
+	}
+	var (
+		result    []protocol.Invoker
+		addresses []string
+	)
+	if tag != "" {
+		addresses, _ = tagRouterRuleCopy.getTagNameToAddresses()[tag]
+		// filter by dynamic tag group first
+		if len(addresses) > 0 {
+			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
+			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.
+			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 {
+			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.
+		}
+		cond := func(invoker protocol.Invoker) bool {
+			localTag := invoker.GetUrl().GetParam(constant.Tagkey, "")
+			return localTag == "" || !(tagRouterRuleCopy.hasTag(localTag))
+		}
+		return filterCondition(result, cond)
+	}
+}
+
+func (c *tagRouter) Process(event *config_center.ConfigChangeEvent) {
+	logger.Infof("Notification of dynamic tag rule, change type is:[%s] , raw rule is:[%v]", event.ConfigType, event.Value)
+	if remoting.EventTypeDel == event.ConfigType {
+		c.tagRouterRule = nil
+		return
+	} else {
+		content, ok := event.Value.(string)
+		if !ok {
+			msg := fmt.Sprintf("Convert event content fail,raw content:[%s] ", event.Value)
+			logger.Error(msg)
+			return
+		}
+
+		routerRule, err := getRule(content)
+		if err != nil {
+			logger.Errorf("Parse dynamic tag router rule fail,error:[%s] ", err)
+			return
+		}
+		c.tagRouterRule = routerRule
+		return
+	}
 }
 
 func (c *tagRouter) URL() common.URL {
@@ -92,3 +184,54 @@ func isForceUseTag(url *common.URL, invocation protocol.Invocation) bool {
 	}
 	return false
 }
+
+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 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 ip match
+		if address == host+":"+port {
+			return true
+		}
+		if address == constant.ANYHOST_VALUE+":"+port {
+			return true
+		}
+	}
+	return false
+}