This is an automated email from the ASF dual-hosted git repository.
alexstocks pushed a commit to branch 1.5
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/1.5 by this push:
new 1f75d2d Feature: Add some log when the router.Route return empty
invokers (#1158)
1f75d2d is described below
commit 1f75d2d0107d3801e50f38ed1dcd0ce6e1689012
Author: cvictory <[email protected]>
AuthorDate: Sat May 15 11:41:57 2021 +0800
Feature: Add some log when the router.Route return empty invokers (#1158)
* fix sentinel cannot transport Context issue
* optimize SentinelLogging reset
* support print route info when there is no provider
* revert to original version
---
cluster/router/chain.go | 13 +++++
cluster/router/chain/chain.go | 62 +++++++++++++++++++++++
cluster/router/conncheck/conn_check_route.go | 16 ++++++
cluster/router/conncheck/conn_check_route_test.go | 50 ++++++++++++++++++
cluster/router/router.go | 5 ++
5 files changed, 146 insertions(+)
diff --git a/cluster/router/chain.go b/cluster/router/chain.go
index cb33cf9..3c6da4d 100644
--- a/cluster/router/chain.go
+++ b/cluster/router/chain.go
@@ -31,4 +31,17 @@ type Chain interface {
AddRouters([]PriorityRouter)
// GetNotifyChan get notify channel of this chain
GetNotifyChan() chan struct{}
+ // Detect Route State
+ DetectRoute() (RouteSnapshot, error)
+}
+
+// RouteSnapshot is the snapshot of Route
+type RouteSnapshot struct {
+ Invokers []protocol.Invoker
+ RouteSnapshots []string
+}
+
+// nolint
+func (rs *RouteSnapshot) AddRouteSnapshot(msg string) {
+ rs.RouteSnapshots = append(rs.RouteSnapshots, msg)
}
diff --git a/cluster/router/chain/chain.go b/cluster/router/chain/chain.go
index 13cb1ff..a44ef10 100644
--- a/cluster/router/chain/chain.go
+++ b/cluster/router/chain/chain.go
@@ -18,7 +18,10 @@
package chain
import (
+ "reflect"
"sort"
+ "strconv"
+ "strings"
"sync"
"time"
)
@@ -73,6 +76,7 @@ func (c *RouterChain) GetNotifyChan() chan struct{} {
func (c *RouterChain) Route(url *common.URL, invocation protocol.Invocation)
[]protocol.Invoker {
cache := c.loadCache()
if cache == nil {
+ logger.Warnf("there is no cache invoker.")
c.mutex.RLock()
defer c.mutex.RUnlock()
return c.invokers
@@ -84,6 +88,10 @@ func (c *RouterChain) Route(url *common.URL, invocation
protocol.Invocation) []p
}
indexes := bitmap.ToArray()
+ // if the indexes is empty, print the routeSnapshot
+ if len(indexes) == 0 {
+ go c.printRouteSnapshot(cache, url, invocation)
+ }
finalInvokers := make([]protocol.Invoker, len(indexes))
for i, index := range indexes {
finalInvokers[i] = cache.invokers[index]
@@ -121,6 +129,60 @@ func (c *RouterChain) SetInvokers(invokers
[]protocol.Invoker) {
}()
}
+// Detect Route State
+func (c *RouterChain) DetectRoute() (router.RouteSnapshot, error) {
+ defer func() {
+ if err := recover(); err != nil {
+ logger.Warnf("Detect Route fail. %+v", err)
+ }
+ }()
+ cache := c.loadCache()
+ if cache == nil {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ return router.RouteSnapshot{Invokers: c.invokers}, nil
+ }
+ routers := c.copyRouters()
+ routeSnapshots := make([]string, 0, len(routers))
+ for _, r := range routers {
+ if v, ok := r.(router.PriorityRouterDetecter); ok {
+ routeSnapshots = append(routeSnapshots,
v.RouteSnapshot(cache))
+ }
+ }
+
+ return router.RouteSnapshot{Invokers: cache.invokers, RouteSnapshots:
routeSnapshots}, nil
+}
+
+// nolint
+func (c *RouterChain) printRouteSnapshot(cache *InvokerCache, url *common.URL,
invocation protocol.Invocation) {
+ defer func() {
+ if err := recover(); err != nil {
+ logger.Warnf("print Route Snapshot fail. %+v", err)
+ }
+ }()
+
+ bitmap := cache.bitmap
+
+ logger.Warnf("start:print the route info:%s", url.ServiceKey())
+ for _, r := range c.copyRouters() {
+ bitmap = r.Route(bitmap, cache, url, invocation)
+ routeSnapshotSb := strings.Builder{}
+ routeSnapshotSb.WriteString(reflect.TypeOf(r).String())
+ routeSnapshotSb.WriteString(", count:")
+
routeSnapshotSb.WriteString(strconv.FormatUint(bitmap.GetCardinality(), 10))
+ routeSnapshotSb.WriteString(bitmap.String())
+ logger.Warn(routeSnapshotSb.String())
+ }
+
+ if routerSnapshot, err := c.DetectRoute(); err == nil {
+ logger.Warnf("the size of invokers:%d",
len(routerSnapshot.Invokers))
+ for _, item := range routerSnapshot.RouteSnapshots {
+ logger.Warn(item)
+ }
+ }
+ logger.Warnf("end: print the route info:%s", url.ServiceKey())
+}
+
// loop listens on events to update the address cache when it receives
notification
// from address update,
func (c *RouterChain) loop() {
diff --git a/cluster/router/conncheck/conn_check_route.go
b/cluster/router/conncheck/conn_check_route.go
index 97f049d..6518dec 100644
--- a/cluster/router/conncheck/conn_check_route.go
+++ b/cluster/router/conncheck/conn_check_route.go
@@ -19,6 +19,8 @@ package conncheck
import (
"github.com/RoaringBitmap/roaring"
+ "strconv"
+ "strings"
)
import (
@@ -67,6 +69,20 @@ func (r *ConnCheckRouter) Route(invokers *roaring.Bitmap,
cache router.Cache, ur
return healthyInvokers
}
+func (r *ConnCheckRouter) RouteSnapshot(cache router.Cache) string {
+ addrPool := cache.FindAddrPool(r)
+ // Add healthy invoker to the list
+ healthBit := addrPool[connHealthy]
+ sb := strings.Builder{}
+ sb.WriteString(r.Name())
+ sb.WriteString(" -> ")
+ sb.WriteString("Count:")
+ sb.WriteString(strconv.FormatUint(healthBit.GetCardinality(), 10))
+ sb.WriteString(" ")
+ sb.WriteString(healthBit.String())
+ return sb.String()
+}
+
// Pool separates healthy invokers from others.
func (r *ConnCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool,
router.AddrMetadata) {
rb := make(router.AddrPool, 8)
diff --git a/cluster/router/conncheck/conn_check_route_test.go
b/cluster/router/conncheck/conn_check_route_test.go
index fec7331..5f78722 100644
--- a/cluster/router/conncheck/conn_check_route_test.go
+++ b/cluster/router/conncheck/conn_check_route_test.go
@@ -102,6 +102,56 @@ func TestRecovery(t *testing.T) {
assert.Equal(t, len(protocol.GetBlackListInvokers(16)), 0)
}
+func TestPrintlnConnCheckRouterRoute(t *testing.T) {
+
+ connCheck1001URL2 :=
"dubbo://192.168.10.1/com.ikurento.user.UserNoProvider"
+ connCheckRouteUrl2Format :=
"dubbo://%s:20000/com.ikurento.user.UserNoProvider"
+
+ defer protocol.CleanAllStatus()
+ notify := make(chan struct{})
+ go func() {
+ for range notify {
+ }
+ }()
+ consumerURL, _ := common.NewURL(connCheck1001URL2)
+ url1, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format,
connCheckRoute1010IP))
+ url2, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format,
connCheckRoute1011IP))
+ url3, _ := common.NewURL(fmt.Sprintf(connCheckRouteUrl2Format,
connCheckRoute1012IP))
+ hcr, _ := NewConnCheckRouter(consumerURL, notify)
+
+ var invokers []protocol.Invoker
+ invoker1 := NewMockInvoker(url1)
+ invoker2 := NewMockInvoker(url2)
+ invoker3 := NewMockInvoker(url3)
+ protocol.SetInvokerUnhealthyStatus(invoker1)
+ protocol.SetInvokerUnhealthyStatus(invoker2)
+ protocol.SetInvokerUnhealthyStatus(invoker3)
+
+ invokers = append(invokers, invoker1, invoker2, invoker3)
+ inv := invocation.NewRPCInvocation(connCheckRouteMethodNameTest, nil,
nil)
+ cache := setUpAddrCache(hcr.(*ConnCheckRouter), invokers)
+ res := hcr.Route(utils.ToBitmap(invokers), cache, consumerURL, inv)
+
+ // now invoker3 is healthy
+ assert.True(t, len(res.ToArray()) == 3)
+ var (
+ router *ConnCheckRouter
+ ok bool
+ )
+ router, ok = hcr.(*ConnCheckRouter)
+ assert.True(t, ok)
+ assert.Equal(t, router.RouteSnapshot(cache), "conn-check-router ->
Count:0 {}")
+
+ // check blacklist remove
+ protocol.RemoveInvokerUnhealthyStatus(invoker1)
+ protocol.RemoveInvokerUnhealthyStatus(invoker3)
+ cache = setUpAddrCache(hcr.(*ConnCheckRouter), invokers)
+ res = hcr.Route(utils.ToBitmap(invokers), cache, consumerURL, inv)
+ // now invoker3 invoker1 is healthy
+ assert.True(t, len(res.ToArray()) == 2)
+ assert.Equal(t, router.RouteSnapshot(cache), "conn-check-router ->
Count:2 {0,2}")
+}
+
func setUpAddrCache(r router.Poolable, addrs []protocol.Invoker) router.Cache {
pool, info := r.Pool(addrs)
cache := chain.BuildCache(addrs)
diff --git a/cluster/router/router.go b/cluster/router/router.go
index 1d71554..93b0484 100644
--- a/cluster/router/router.go
+++ b/cluster/router/router.go
@@ -56,6 +56,11 @@ type PriorityRouter interface {
Priority() int64
}
+// PriorityRouterDetecter detect the router
+type PriorityRouterDetecter interface {
+ RouteSnapshot(cache Cache) string
+}
+
// Poolable caches address pool and address metadata for a router instance
which will be used later in Router's Route.
type Poolable interface {
// Pool created address pool and address metadata from the invokers.
