This is an automated email from the ASF dual-hosted git repository.
tianxiaoliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git
The following commit(s) were added to refs/heads/master by this push:
new 706fd9c SCB-2176 Fix: Inconsistent cache (#823)
706fd9c is described below
commit 706fd9cebbd958cbd9ccb62a4e9e25c732772406
Author: little-cui <[email protected]>
AuthorDate: Mon Jan 11 09:25:43 2021 +0800
SCB-2176 Fix: Inconsistent cache (#823)
---
datasource/etcd/cache/common.go | 2 ++
datasource/etcd/cache/dependency.go | 2 +-
datasource/etcd/cache/filter_consistency.go | 13 +++++++++----
datasource/etcd/cache/instance.go | 2 +-
datasource/etcd/sd/aggregate/indexer.go | 19 ++++++++-----------
pkg/notify/notification_service.go | 3 ++-
6 files changed, 23 insertions(+), 18 deletions(-)
diff --git a/datasource/etcd/cache/common.go b/datasource/etcd/cache/common.go
index 8e501f8..7963391 100644
--- a/datasource/etcd/cache/common.go
+++ b/datasource/etcd/cache/common.go
@@ -37,6 +37,8 @@ const (
Find = "_find"
Dep = "_dep"
+
+ DefaultCacheMaxSize = 10000
)
var (
diff --git a/datasource/etcd/cache/dependency.go
b/datasource/etcd/cache/dependency.go
index d8cef9a..69ccd31 100644
--- a/datasource/etcd/cache/dependency.go
+++ b/datasource/etcd/cache/dependency.go
@@ -26,7 +26,7 @@ import (
)
var DependencyRule = &DependencyRuleCache{
- Tree: cache.NewTree(cache.Configure().WithMaxSize(10000))}
+ Tree: cache.NewTree(cache.Configure().WithMaxSize(DefaultCacheMaxSize))}
func init() {
DependencyRule.AddFilter(
diff --git a/datasource/etcd/cache/filter_consistency.go
b/datasource/etcd/cache/filter_consistency.go
index 0167944..a33ea05 100644
--- a/datasource/etcd/cache/filter_consistency.go
+++ b/datasource/etcd/cache/filter_consistency.go
@@ -19,6 +19,7 @@ package cache
import (
"context"
+ "fmt"
"github.com/apache/servicecomb-service-center/datasource/etcd/kv"
"github.com/apache/servicecomb-service-center/pkg/cache"
@@ -51,6 +52,10 @@ func (f *ConsistencyFilter) Name(ctx context.Context, parent
*cache.Node) string
func (f *ConsistencyFilter) Init(ctx context.Context, parent *cache.Node)
(node *cache.Node, err error) {
pCache := parent.Cache.Get(Find).(*VersionRuleCacheItem)
requestRev := ctx.Value(CtxFindRequestRev).(string)
+ // do not need to check consistency between sc instances:
+ // 1. request without rev param
+ // 2. request rev is the same as cache current sc instance
+ // 3. datasource has no cache indexer
if len(requestRev) == 0 || requestRev == pCache.Rev ||
!(kv.Store().Instance().Creditable()) {
node = cache.NewNode()
@@ -65,15 +70,15 @@ func (f *ConsistencyFilter) Init(ctx context.Context,
parent *cache.Node) (node
}
cloneCtx := util.WithNoCache(util.CloneContext(ctx))
- insts, _, err := f.Find(cloneCtx, parent)
+ insts, rev, err := f.Find(cloneCtx, parent)
if err != nil {
pCache.InitBrokenQueue()
return nil, err
}
- log.Warnf("the cache of finding instances api is broken,
req[%s]!=cache[%s][%s]",
- requestRev, pCache.Rev, parent.Name)
- pCache.Instances = insts
+ log.Warn(fmt.Sprintf("inconsistent rev! %s, req[%s], cache[%s],
datasource[%s]",
+ parent.Name, requestRev, pCache.Rev, rev))
+ pCache.Instances, pCache.Rev = insts, rev
pCache.Broken()
node = cache.NewNode()
diff --git a/datasource/etcd/cache/instance.go
b/datasource/etcd/cache/instance.go
index 57b255b..117fddf 100644
--- a/datasource/etcd/cache/instance.go
+++ b/datasource/etcd/cache/instance.go
@@ -27,7 +27,7 @@ import (
)
var FindInstances = &FindInstancesCache{
- Tree: cache.NewTree(cache.Configure().WithTTL(2 *
time.Minute).WithMaxSize(10000))}
+ Tree: cache.NewTree(cache.Configure().WithTTL(2 *
time.Minute).WithMaxSize(DefaultCacheMaxSize))}
func init() {
FindInstances.AddFilter(
diff --git a/datasource/etcd/sd/aggregate/indexer.go
b/datasource/etcd/sd/aggregate/indexer.go
index 756e85b..598ecce 100644
--- a/datasource/etcd/sd/aggregate/indexer.go
+++ b/datasource/etcd/sd/aggregate/indexer.go
@@ -85,8 +85,14 @@ type AggregatorIndexer struct {
func (i *AggregatorIndexer) Search(ctx context.Context, opts
...client.PluginOpOption) (resp *sd.Response, err error) {
op := client.OpGet(opts...)
+ indexer := i.LocalIndexer
+ if op.Global {
+ // request with global param then do not use local indexer
+ indexer = i.AdaptorsIndexer
+ }
+
if op.NoCache() || !op.Global {
- return i.search(ctx, opts...)
+ return indexer.Search(ctx, opts...)
}
resp, err = i.CacheIndexer.Search(ctx, opts...)
@@ -98,16 +104,7 @@ func (i *AggregatorIndexer) Search(ctx context.Context,
opts ...client.PluginOpO
return resp, nil
}
- return i.search(ctx, opts...)
-}
-
-func (i *AggregatorIndexer) search(ctx context.Context, opts
...client.PluginOpOption) (*sd.Response, error) {
- op := client.OptionsToOp(opts...)
- if !op.Global {
- return i.LocalIndexer.Search(ctx, opts...)
- }
-
- return i.AdaptorsIndexer.Search(ctx, opts...)
+ return indexer.Search(ctx, opts...)
}
// Creditable implements sd.Indexer#Creditable.
diff --git a/pkg/notify/notification_service.go
b/pkg/notify/notification_service.go
index 9678489..d63cace 100644
--- a/pkg/notify/notification_service.go
+++ b/pkg/notify/notification_service.go
@@ -19,6 +19,7 @@ package notify
import (
"errors"
+ "fmt"
"sync"
"github.com/apache/servicecomb-service-center/pkg/log"
@@ -124,7 +125,7 @@ func (s *Service) Publish(evt Event) error {
p, ok := s.processors[evt.Type()]
if !ok {
s.mux.RUnlock()
- return errors.New("unknown event type")
+ return fmt.Errorf("unknown event type[%s]", evt.Type())
}
s.mux.RUnlock()
p.Accept(evt)
