Alanxtl commented on code in PR #3305:
URL: https://github.com/apache/dubbo-go/pull/3305#discussion_r3216256941
##########
cluster/router/tag/router.go:
##########
@@ -157,3 +172,269 @@ func parseRoute(routeContent string)
(*global.RouterConfig, error) {
}
return routerConfig, nil
}
+
+func (p *PriorityRouter) Name() string {
+ return "tag"
+}
+
+func (p *PriorityRouter) ShouldPool() bool {
+ return true
+}
+
+func (p *PriorityRouter) Pool(invokers []base.Invoker) (router.AddrPool,
router.AddrMetadata) {
+ pool := make(router.AddrPool)
+ for i, invoker := range invokers {
+ url := invoker.GetURL()
+ upsertBM(pool, "*", i)
+ tag := url.GetParam(constant.Tagkey, "")
+ upsertBM(pool, "tag\x00"+tag, i)
+ addr := url.Location
+ if addr != "" {
+ upsertBM(pool, "addr\x00"+addr, i)
+ if idx := strings.LastIndex(addr, ":"); idx > 0 {
+ upsertBM(pool, "port\x00"+addr[idx+1:], i)
+ }
+ }
+ for _, key := range []string{"version", "group"} {
+ if v := url.GetParam(key, ""); v != "" {
+ upsertBM(pool, "param\x00"+key+"\x00"+v, i)
+ } else {
+ upsertBM(pool, "param\x00"+key+"\x00\x00", i)
+ }
+ }
+ }
+ return pool, nil
+}
+
+func (p *PriorityRouter) SetCache(cache router.Cache) {
+ p.cache = cache
+}
+
+func (p *PriorityRouter) routeWithPool(invokers []base.Invoker, pool
router.AddrPool, url *common.URL, invocation base.Invocation) []base.Invoker {
+ tag := invocation.GetAttachmentWithDefaultValue(constant.Tagkey,
url.GetParam(constant.Tagkey, ""))
+
+ application := invokers[0].GetURL().GetParam(constant.ApplicationKey,
"")
+ key := strings.Join([]string{application,
constant.TagRouterRuleSuffix}, "")
+ value, ok := p.routerConfigs.Load(key)
+ if !ok {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ routerCfg := value.(global.RouterConfig)
+ enabled := routerCfg.Enabled == nil || *routerCfg.Enabled
+ valid := (routerCfg.Valid != nil && *routerCfg.Valid) ||
(routerCfg.Valid == nil && len(routerCfg.Tags) > 0)
+ if !enabled || !valid {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ if tag == "" {
+ return collectInvokers(invokers, p.emptyTagMatchBM(pool,
routerCfg))
+ }
+ bm := p.requestTagMatchBM(pool, url, invocation, routerCfg, tag)
+ if bm == nil {
+ return requestTag(invokers, url, invocation, routerCfg, tag)
+ }
+ return collectInvokers(invokers, bm)
+}
+
+func (p *PriorityRouter) staticTagMatchBM(pool router.AddrPool, tag string,
url *common.URL, invocation base.Invocation) *roaring.Bitmap {
+ if tag != "" {
+ if bm := pool["tag\x00"+tag]; bm != nil && !bm.IsEmpty() {
+ return bm
+ }
+ if requestIsForce(url, invocation) {
+ return nil
+ }
+ }
+ return pool["tag\x00"]
+}
+
+func (p *PriorityRouter) requestTagMatchBM(pool router.AddrPool, url
*common.URL, invocation base.Invocation,
+ cfg global.RouterConfig, tag string) *roaring.Bitmap {
+ var (
+ addresses []string
+ match []*common.ParamMatch
+ )
+ for _, tagCfg := range cfg.Tags {
+ if tagCfg.Name == tag {
+ addresses = tagCfg.Addresses
+ match = tagCfg.Match
+ break
+ }
+ }
+
+ var resultBM *roaring.Bitmap
+ if len(match) != 0 {
+ resultBM = p.matchBM(pool, match)
+ } else if len(addresses) != 0 {
+ resultBM = p.addressesBM(pool, addresses)
+ } else {
+ resultBM = pool["tag\x00"+tag]
+ }
+
+ if (cfg.Force != nil && *cfg.Force) || requestIsForce(url, invocation) {
+ return resultBM
+ }
+ if resultBM != nil && !resultBM.IsEmpty() {
+ return resultBM
+ }
+
+ emptyBM := pool["tag\x00"]
+ if emptyBM == nil {
+ return nil
+ }
+ if len(addresses) == 0 {
+ return emptyBM
+ }
+ addrBM := p.addressesBM(pool, addresses)
+ if addrBM != nil && !addrBM.IsEmpty() {
+ allBM := pool["*"]
+ if allBM != nil {
+ result := allBM.Clone()
+ result.AndNot(addrBM)
+ return result
+ }
+ }
+ return pool["*"]
+}
+
+func (p *PriorityRouter) emptyTagMatchBM(pool router.AddrPool, cfg
global.RouterConfig) *roaring.Bitmap {
+ bm := pool["tag\x00"]
+ if bm == nil || bm.IsEmpty() {
+ return bm
+ }
+ for _, tagCfg := range cfg.Tags {
+ if len(tagCfg.Addresses) == 0 {
+ continue
+ }
+ addrBM := p.addressesBM(pool, tagCfg.Addresses)
+ if addrBM != nil && !addrBM.IsEmpty() && bm != nil {
+ result := bm.Clone()
+ result.AndNot(addrBM)
+ bm = result
+ }
+ }
+ return bm
+}
+
+func (p *PriorityRouter) addressesBM(pool router.AddrPool, addrs []string)
*roaring.Bitmap {
+ bm := roaring.NewBitmap()
+ for _, addr := range addrs {
+ if ab := pool["addr\x00"+addr]; ab != nil {
+ bm.Or(ab)
+ }
+ if idx := strings.LastIndex(addr, ":"); idx > 0 && addr[:idx]
== constant.AnyHostValue {
+ if pb := pool["port\x00"+addr[idx+1:]]; pb != nil {
+ bm.Or(pb)
+ }
+ }
+ }
+ return bm
+}
+
+func (p *PriorityRouter) matchBM(pool router.AddrPool, matches
[]*common.ParamMatch) *roaring.Bitmap {
+ var result *roaring.Bitmap
+ for _, m := range matches {
+ bm := p.paramMatchBM(pool, m)
+ if bm == nil {
+ return nil
+ }
+ if result == nil {
+ result = bm.Clone()
+ } else {
+ result.And(bm)
+ }
+ if result.IsEmpty() {
+ return result
+ }
+ }
+ return result
+}
+
+func (p *PriorityRouter) paramMatchBM(pool router.AddrPool, pm
*common.ParamMatch) *roaring.Bitmap {
+ prefix := "param\x00" + pm.Key + "\x00"
+ switch {
+ case pm.Value.Exact != "":
+ return pool[prefix+pm.Value.Exact]
+ case pm.Value.Wildcard != "":
+ if pm.Value.Wildcard == constant.AnyValue {
+ return pool["*"]
+ }
+ return pool[prefix+pm.Value.Wildcard]
+ case pm.Value.Prefix != "":
+ return scanPrefixBM(pool, prefix, pm.Value.Prefix)
+ case pm.Value.Regex != "":
+ return scanRegexBM(pool, prefix, pm.Value.Regex)
+ case pm.Value.Empty != "":
+ return pool[prefix+"\x00"]
+ case pm.Value.Noempty != "":
+ emptyBM := pool[prefix+"\x00"]
+ allBM := pool["*"]
+ if allBM == nil {
+ return roaring.NewBitmap()
+ }
+ if emptyBM != nil {
+ result := allBM.Clone()
+ result.AndNot(emptyBM)
+ return result
+ }
+ return allBM
+ }
+ return roaring.NewBitmap()
+}
Review Comment:
bitmap ParamMatch only indexes `version` and `group`, but `Prefix`, `Regex`,
and `Noempty` on other keys don’t fall back to the original matcher.
Is this designed on purpose?
You can verify a Match{Key:"env", Prefix:"prod"} case where fallback returns
one invoker and cached path returns empty.
##########
cluster/router/tag/router.go:
##########
@@ -52,6 +57,16 @@ func (p *PriorityRouter) Route(invokers []base.Invoker, url
*common.URL, invocat
logger.Warnf("[tag router] invokers from previous router is
empty")
return invokers
}
+
+ if p.cache != nil {
+ if
!invocation.GetAttributeWithDefaultValue(constant.RouterCacheDisable,
false).(bool) {
+ pool := p.cache.FindAddrPool(p)
+ if pool != nil {
+ return p.routeWithPool(p.cache.GetInvokers(),
pool, url, invocation)
+ }
+ }
+ }
Review Comment:
cache pool and invoker snapshot are read in two separate calls.
`FindAddrPool()` can return the old bitmap, then `GetInvokers()` can return the
new invoker slice after `routerCache.rebuild()`, so bitmap indexes can select
the wrong provider. The cache API needs to return pool + invokers from one
locked snapshot.
##########
cluster/router/tag/router.go:
##########
@@ -157,3 +172,269 @@ func parseRoute(routeContent string)
(*global.RouterConfig, error) {
}
return routerConfig, nil
}
+
+func (p *PriorityRouter) Name() string {
+ return "tag"
+}
+
+func (p *PriorityRouter) ShouldPool() bool {
+ return true
+}
+
+func (p *PriorityRouter) Pool(invokers []base.Invoker) (router.AddrPool,
router.AddrMetadata) {
+ pool := make(router.AddrPool)
+ for i, invoker := range invokers {
+ url := invoker.GetURL()
+ upsertBM(pool, "*", i)
+ tag := url.GetParam(constant.Tagkey, "")
+ upsertBM(pool, "tag\x00"+tag, i)
+ addr := url.Location
+ if addr != "" {
+ upsertBM(pool, "addr\x00"+addr, i)
+ if idx := strings.LastIndex(addr, ":"); idx > 0 {
+ upsertBM(pool, "port\x00"+addr[idx+1:], i)
+ }
+ }
+ for _, key := range []string{"version", "group"} {
+ if v := url.GetParam(key, ""); v != "" {
+ upsertBM(pool, "param\x00"+key+"\x00"+v, i)
+ } else {
+ upsertBM(pool, "param\x00"+key+"\x00\x00", i)
+ }
+ }
+ }
+ return pool, nil
+}
+
+func (p *PriorityRouter) SetCache(cache router.Cache) {
+ p.cache = cache
+}
+
+func (p *PriorityRouter) routeWithPool(invokers []base.Invoker, pool
router.AddrPool, url *common.URL, invocation base.Invocation) []base.Invoker {
+ tag := invocation.GetAttachmentWithDefaultValue(constant.Tagkey,
url.GetParam(constant.Tagkey, ""))
+
+ application := invokers[0].GetURL().GetParam(constant.ApplicationKey,
"")
+ key := strings.Join([]string{application,
constant.TagRouterRuleSuffix}, "")
+ value, ok := p.routerConfigs.Load(key)
+ if !ok {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ routerCfg := value.(global.RouterConfig)
+ enabled := routerCfg.Enabled == nil || *routerCfg.Enabled
+ valid := (routerCfg.Valid != nil && *routerCfg.Valid) ||
(routerCfg.Valid == nil && len(routerCfg.Tags) > 0)
+ if !enabled || !valid {
+ return collectInvokers(invokers, p.staticTagMatchBM(pool, tag,
url, invocation))
+ }
+ if tag == "" {
+ return collectInvokers(invokers, p.emptyTagMatchBM(pool,
routerCfg))
+ }
+ bm := p.requestTagMatchBM(pool, url, invocation, routerCfg, tag)
+ if bm == nil {
+ return requestTag(invokers, url, invocation, routerCfg, tag)
+ }
+ return collectInvokers(invokers, bm)
+}
+
+func (p *PriorityRouter) staticTagMatchBM(pool router.AddrPool, tag string,
url *common.URL, invocation base.Invocation) *roaring.Bitmap {
+ if tag != "" {
+ if bm := pool["tag\x00"+tag]; bm != nil && !bm.IsEmpty() {
+ return bm
+ }
+ if requestIsForce(url, invocation) {
+ return nil
+ }
+ }
+ return pool["tag\x00"]
+}
+
+func (p *PriorityRouter) requestTagMatchBM(pool router.AddrPool, url
*common.URL, invocation base.Invocation,
+ cfg global.RouterConfig, tag string) *roaring.Bitmap {
+ var (
+ addresses []string
+ match []*common.ParamMatch
+ )
+ for _, tagCfg := range cfg.Tags {
+ if tagCfg.Name == tag {
+ addresses = tagCfg.Addresses
+ match = tagCfg.Match
+ break
+ }
+ }
+
+ var resultBM *roaring.Bitmap
+ if len(match) != 0 {
+ resultBM = p.matchBM(pool, match)
+ } else if len(addresses) != 0 {
+ resultBM = p.addressesBM(pool, addresses)
+ } else {
+ resultBM = pool["tag\x00"+tag]
+ }
+
+ if (cfg.Force != nil && *cfg.Force) || requestIsForce(url, invocation) {
+ return resultBM
+ }
+ if resultBM != nil && !resultBM.IsEmpty() {
+ return resultBM
+ }
+
+ emptyBM := pool["tag\x00"]
+ if emptyBM == nil {
+ return nil
+ }
+ if len(addresses) == 0 {
+ return emptyBM
+ }
+ addrBM := p.addressesBM(pool, addresses)
+ if addrBM != nil && !addrBM.IsEmpty() {
+ allBM := pool["*"]
+ if allBM != nil {
+ result := allBM.Clone()
+ result.AndNot(addrBM)
+ return result
+ }
+ }
+ return pool["*"]
Review Comment:
dynamic tag address failover changes semantics. Original path falls back to
untagged providers, excluding configured addresses; cached path returns
`pool["*"]` when configured addresses miss, so a request for tag `gray` can be
routed to all providers, including other tagged providers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
