This is an automated email from the ASF dual-hosted git repository.
tianxiaoliang pushed a commit to branch v1.x
in repository https://gitbox.apache.org/repos/asf/servicecomb-service-center.git
The following commit(s) were added to refs/heads/v1.x by this push:
new ad6b8e6 SCB-2176 Fix: Inconsistent cache (#822)
ad6b8e6 is described below
commit ad6b8e6c3ed522e7e3a7f3305d11e262345cc0da
Author: little-cui <[email protected]>
AuthorDate: Mon Jan 11 09:25:57 2021 +0800
SCB-2176 Fix: Inconsistent cache (#822)
---
pkg/notify/notification_service.go | 3 ++-
server/plugin/discovery/aggregate/indexer.go | 21 +++++++++------------
server/service/cache/common.go | 2 ++
server/service/cache/dependency.go | 3 +--
server/service/cache/filter_consistency.go | 13 +++++++++----
server/service/cache/instance.go | 3 +--
6 files changed, 24 insertions(+), 21 deletions(-)
diff --git a/pkg/notify/notification_service.go
b/pkg/notify/notification_service.go
index 445e428..658d6ab 100644
--- a/pkg/notify/notification_service.go
+++ b/pkg/notify/notification_service.go
@@ -19,6 +19,7 @@ package notify
import (
"errors"
+ "fmt"
"github.com/apache/servicecomb-service-center/pkg/log"
"sync"
)
@@ -123,7 +124,7 @@ func (s *Service) Publish(job Event) error {
p, ok := s.processors[job.Type()]
if !ok {
s.mux.RUnlock()
- return errors.New("Unknown job type")
+ return fmt.Errorf("unknown job type[%s]", job.Type())
}
s.mux.RUnlock()
p.Accept(job)
diff --git a/server/plugin/discovery/aggregate/indexer.go
b/server/plugin/discovery/aggregate/indexer.go
index 3d7a72f..3d89cd9 100644
--- a/server/plugin/discovery/aggregate/indexer.go
+++ b/server/plugin/discovery/aggregate/indexer.go
@@ -77,7 +77,7 @@ type AggregatorIndexer struct {
*discovery.CacheIndexer
// AdaptorsIndexer searches data from all the adaptors.
AdaptorsIndexer discovery.Indexer
- // LocalIndexer data from local adaptor.
+ // LocalIndexer data from registry indexer.
LocalIndexer discovery.Indexer
}
@@ -85,8 +85,14 @@ type AggregatorIndexer struct {
func (i *AggregatorIndexer) Search(ctx context.Context, opts
...registry.PluginOpOption) (resp *discovery.Response, err error) {
op := registry.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 ...registry.PluginO
return resp, nil
}
- return i.search(ctx, opts...)
-}
-
-func (i *AggregatorIndexer) search(ctx context.Context, opts
...registry.PluginOpOption) (*discovery.Response, error) {
- op := registry.OptionsToOp(opts...)
- if !op.Global {
- return i.LocalIndexer.Search(ctx, opts...)
- }
-
- return i.AdaptorsIndexer.Search(ctx, opts...)
+ return indexer.Search(ctx, opts...)
}
// Creditable implements discovery.Indexer.Creditable.
diff --git a/server/service/cache/common.go b/server/service/cache/common.go
index 2dc6231..ac5202c 100644
--- a/server/service/cache/common.go
+++ b/server/service/cache/common.go
@@ -28,4 +28,6 @@ const (
Find = "_find"
Dep = "_dep"
+
+ DefaultCacheMaxSize = 10000
)
diff --git a/server/service/cache/dependency.go
b/server/service/cache/dependency.go
index 09cf479..b717617 100644
--- a/server/service/cache/dependency.go
+++ b/server/service/cache/dependency.go
@@ -22,12 +22,11 @@ import (
"github.com/apache/servicecomb-service-center/pkg/cache"
pb "github.com/apache/servicecomb-service-center/pkg/registry"
"github.com/apache/servicecomb-service-center/pkg/util"
- "math"
)
var DependencyRule = &DependencyRuleCache{
Tree: cache.NewTree(cache.Configure().
- WithMaxSize(math.MaxInt64))}
+ WithMaxSize(DefaultCacheMaxSize))}
func init() {
DependencyRule.AddFilter(
diff --git a/server/service/cache/filter_consistency.go
b/server/service/cache/filter_consistency.go
index 35a63a5..94725d1 100644
--- a/server/service/cache/filter_consistency.go
+++ b/server/service/cache/filter_consistency.go
@@ -19,6 +19,7 @@ package cache
import (
"context"
+ "fmt"
"github.com/apache/servicecomb-service-center/pkg/cache"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
@@ -50,6 +51,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 ||
!(backend.Store().Instance().Creditable()) {
node = cache.NewNode()
@@ -65,15 +70,15 @@ func (f *ConsistencyFilter) Init(ctx context.Context,
parent *cache.Node) (node
cloneCtx := util.CloneContext(ctx)
cloneCtx = util.SetContext(cloneCtx, util.CtxNocache, "1")
- 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/server/service/cache/instance.go b/server/service/cache/instance.go
index 7ce1d66..259738e 100644
--- a/server/service/cache/instance.go
+++ b/server/service/cache/instance.go
@@ -22,14 +22,13 @@ import (
"github.com/apache/servicecomb-service-center/pkg/cache"
pb "github.com/apache/servicecomb-service-center/pkg/registry"
"github.com/apache/servicecomb-service-center/pkg/util"
- "math"
"time"
)
var FindInstances = &FindInstancesCache{
Tree: cache.NewTree(cache.Configure().
WithTTL(2 * time.Minute).
- WithMaxSize(math.MaxInt64))}
+ WithMaxSize(DefaultCacheMaxSize))}
func init() {
FindInstances.AddFilter(
