From 222aff07d8db065b5b9872c9265bdfd6629c0a4d Mon Sep 17 00:00:00 2001 From: little-cui Date: Mon, 18 Sep 2023 19:22:26 +0800 Subject: [PATCH] Revert "[fix] fix inconsistency bug between cache layer and etcd. (#287)" (#298) (#299) This reverts commit e005d80afeab09163289feac644dfd6a2f12fa41. Co-authored-by: tornado-ssy <64736788+tornado-ssy@users.noreply.github.com> Co-authored-by: songshiyuan 00649746 --- examples/dev/kie-conf.yaml | 8 +- server/config/struct.go | 11 +- server/datasource/etcd/kv/kv_cache.go | 152 +++++------- server/datasource/etcd/kv/kv_cache_test.go | 270 ++------------------- 4 files changed, 95 insertions(+), 346 deletions(-) diff --git a/examples/dev/kie-conf.yaml b/examples/dev/kie-conf.yaml index d4aabb1e..11b91067 100644 --- a/examples/dev/kie-conf.yaml +++ b/examples/dev/kie-conf.yaml @@ -18,10 +18,4 @@ db: # rsaPublicKeyFile: ./examples/dev/public.key sync: # turn on the synchronization switch related operations will be written to the task in the db - enabled: false -#cache: -# labels: -# - environment -# - service -# - app -# - version + enabled: false \ No newline at end of file diff --git a/server/config/struct.go b/server/config/struct.go index 44669def..83e91f08 100644 --- a/server/config/struct.go +++ b/server/config/struct.go @@ -19,10 +19,9 @@ package config // Config is yaml file struct type Config struct { - DB DB `yaml:"db"` - RBAC RBAC `yaml:"rbac"` - Sync Sync `yaml:"sync"` - Cache Cache `yaml:"cache"` + DB DB `yaml:"db"` + RBAC RBAC `yaml:"rbac"` + Sync Sync `yaml:"sync"` //config from cli ConfigFile string NodeName string @@ -60,7 +59,3 @@ type RBAC struct { type Sync struct { Enabled bool `yaml:"enabled"` } - -type Cache struct { - Labels []string `yaml:"labels"` -} diff --git a/server/datasource/etcd/kv/kv_cache.go b/server/datasource/etcd/kv/kv_cache.go index 2930a99d..5776bf73 100644 --- a/server/datasource/etcd/kv/kv_cache.go +++ b/server/datasource/etcd/kv/kv_cache.go @@ -11,7 +11,6 @@ import ( "github.com/apache/servicecomb-kie/pkg/model" "github.com/apache/servicecomb-kie/pkg/stringutil" - "github.com/apache/servicecomb-kie/server/config" "github.com/apache/servicecomb-kie/server/datasource" "github.com/apache/servicecomb-kie/server/datasource/etcd/key" "github.com/go-chassis/foundation/backoff" @@ -38,27 +37,21 @@ const ( type IDSet map[string]struct{} -type LabelsSet map[string]struct{} - -type CacheSearchReq struct { - Domain string - Project string - Opts *datasource.FindOptions - Regex *regexp.Regexp +type Cache struct { + timeOut time.Duration + client etcdadpt.Client + revision int64 + kvIDCache sync.Map + kvDocCache *goCache.Cache } func NewKvCache() *Cache { kvDocCache := goCache.New(cacheExpirationTime, cacheCleanupInterval) - labelsSet := LabelsSet{} - for _, label := range config.Configurations.Cache.Labels { - labelsSet[label] = struct{}{} - } return &Cache{ timeOut: etcdWatchTimeout, client: etcdadpt.Instance(), revision: 0, kvDocCache: kvDocCache, - labelsSet: labelsSet, } } @@ -66,13 +59,11 @@ func Enabled() bool { return kvCache != nil } -type Cache struct { - timeOut time.Duration - client etcdadpt.Client - revision int64 - kvIDCache sync.Map - kvDocCache *goCache.Cache - labelsSet LabelsSet +type CacheSearchReq struct { + Domain string + Project string + Opts *datasource.FindOptions + Regex *regexp.Regexp } func (kc *Cache) Refresh(ctx context.Context) { @@ -139,7 +130,7 @@ func (kc *Cache) list(ctx context.Context) (*etcdadpt.Response, error) { return rsp, nil } -func (kc *Cache) watchCallBack(_ string, rsp *etcdadpt.Response) error { +func (kc *Cache) watchCallBack(message string, rsp *etcdadpt.Response) error { if rsp == nil || len(rsp.Kvs) == 0 { return fmt.Errorf("unknown event") } @@ -163,9 +154,6 @@ func (kc *Cache) cachePut(rsp *etcdadpt.Response) { openlog.Error(fmt.Sprintf("failed to unmarshal kv, err %v", err)) continue } - if !kc.isInLabelsSet(kvDoc.Labels) { - continue - } kc.StoreKvDoc(kvDoc.ID, kvDoc) cacheKey := kc.GetCacheKey(kvDoc.Domain, kvDoc.Project, kvDoc.Labels) m, ok := kc.LoadKvIDSet(cacheKey) @@ -232,6 +220,46 @@ func (kc *Cache) DeleteKvDoc(kvID string) { kc.kvDocCache.Delete(kvID) } +func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool, error) { + if !req.Opts.ExactLabels { + return nil, false, nil + } + + openlog.Debug(fmt.Sprintf("using cache to search kv, domain %v, project %v, opts %+v", req.Domain, req.Project, *req.Opts)) + result := &model.KVResponse{ + Data: []*model.KVDoc{}, + } + cacheKey := kvCache.GetCacheKey(req.Domain, req.Project, req.Opts.Labels) + kvIds, ok := kvCache.LoadKvIDSet(cacheKey) + if !ok { + kvCache.StoreKvIDSet(cacheKey, IDSet{}) + return result, true, nil + } + + var docs []*model.KVDoc + + var kvIdsLeft []string + for kvID := range kvIds { + if doc, ok := kvCache.LoadKvDoc(kvID); ok { + docs = append(docs, doc) + continue + } + kvIdsLeft = append(kvIdsLeft, kvID) + } + + tpData := kvCache.getKvFromEtcd(ctx, req, kvIdsLeft) + docs = append(docs, tpData...) + + for _, doc := range docs { + if isMatch(req, doc) { + datasource.ClearPart(doc) + result.Data = append(result.Data, doc) + } + } + result.Total = len(result.Data) + return result, true, nil +} + func (kc *Cache) getKvFromEtcd(ctx context.Context, req *CacheSearchReq, kvIdsLeft []string) []*model.KVDoc { if len(kvIdsLeft) == 0 { return nil @@ -266,6 +294,19 @@ func (kc *Cache) getKvFromEtcd(ctx context.Context, req *CacheSearchReq, kvIdsLe return docs } +func isMatch(req *CacheSearchReq, doc *model.KVDoc) bool { + if doc == nil { + return false + } + if req.Opts.Status != "" && doc.Status != req.Opts.Status { + return false + } + if req.Regex != nil && !req.Regex.MatchString(doc.Key) { + return false + } + return true +} + func (kc *Cache) GetKvDoc(kv *mvccpb.KeyValue) (*model.KVDoc, error) { kvDoc := &model.KVDoc{} err := json.Unmarshal(kv.Value, kvDoc) @@ -285,66 +326,3 @@ func (kc *Cache) GetCacheKey(domain, project string, labels map[string]string) s }, "/") return inputKey } - -func (kc *Cache) isInLabelsSet(Labels map[string]string) bool { - for label := range Labels { - if _, ok := kc.labelsSet[label]; !ok { - return false - } - } - return true -} - -func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool, error) { - result := &model.KVResponse{ - Data: []*model.KVDoc{}, - } - if !req.Opts.ExactLabels || !kvCache.isInLabelsSet(req.Opts.Labels) { - return result, false, nil - } - - openlog.Debug(fmt.Sprintf("using cache to search kv, domain %v, project %v, opts %+v", req.Domain, req.Project, *req.Opts)) - cacheKey := kvCache.GetCacheKey(req.Domain, req.Project, req.Opts.Labels) - - kvIds, ok := kvCache.LoadKvIDSet(cacheKey) - if !ok { - kvCache.StoreKvIDSet(cacheKey, IDSet{}) - return result, true, nil - } - - var docs []*model.KVDoc - - var kvIdsLeft []string - for kvID := range kvIds { - if doc, ok := kvCache.LoadKvDoc(kvID); ok { - docs = append(docs, doc) - continue - } - kvIdsLeft = append(kvIdsLeft, kvID) - } - - tpData := kvCache.getKvFromEtcd(ctx, req, kvIdsLeft) - docs = append(docs, tpData...) - - for _, doc := range docs { - if isMatch(req, doc) { - datasource.ClearPart(doc) - result.Data = append(result.Data, doc) - } - } - result.Total = len(result.Data) - return result, true, nil -} - -func isMatch(req *CacheSearchReq, doc *model.KVDoc) bool { - if doc == nil { - return false - } - if req.Opts.Status != "" && doc.Status != req.Opts.Status { - return false - } - if req.Regex != nil && !req.Regex.MatchString(doc.Key) { - return false - } - return true -} diff --git a/server/datasource/etcd/kv/kv_cache_test.go b/server/datasource/etcd/kv/kv_cache_test.go index c128b5ac..22866996 100644 --- a/server/datasource/etcd/kv/kv_cache_test.go +++ b/server/datasource/etcd/kv/kv_cache_test.go @@ -1,67 +1,46 @@ package kv import ( - "fmt" - "reflect" "testing" - "time" - "github.com/apache/servicecomb-kie/pkg/model" - "github.com/apache/servicecomb-kie/server/config" "github.com/little-cui/etcdadpt" - goCache "github.com/patrickmn/go-cache" "github.com/stretchr/testify/assert" "go.etcd.io/etcd/api/v3/mvccpb" ) -func init() { - config.Configurations.Cache.Labels = []string{"environment"} +type args struct { + rsp *etcdadpt.Response } func TestCachePut(t *testing.T) { - type args struct { - rsp *etcdadpt.Response - } tests := []struct { name string args args want int }{ - { - name: "put 0 kvDoc, cache should store 0 kvDoc", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{}}}, - want: 0, + {"put 0 kvDoc, cache should store 0 kvDoc", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{}}}, + 0, }, - { - name: "put 1 kvDoc, cache should store 1 kvDoc", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ + {"put 1 kvDoc, cache should store 1 kvDoc", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, }}}, - want: 1, + 1, }, - { - name: "put 2 kvDocs with different kvIds, cache should store 2 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ + {"put 2 kvDocs with different kvIds, cache should store 2 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, {Value: []byte(`{"id":"2", "key":"withToys", "value":"yes", "labels":{"environment":"testing"}}`)}, }}}, - want: 2, + 2, }, - { - name: "put 2 kvDocs with same kvId, cache should store 1 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ + {"put 2 kvDocs with same kvId, cache should store 1 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, {Value: []byte(`{"id":"1", "key":"withToys", "value":"yes", "labels":{"environment":"testing"}}`)}, }}}, - want: 1, - }, - { - name: "put 2 kvDoc, but labels are not cached, cache should store 0 kvDoc", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ - {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"env":"testing"}}`)}, - {Value: []byte(`{"id":"1", "key":"withToys", "value":"yes", "labels":{"env":"testing"}}`)}, - }}}, - want: 0, + 1, }, } for _, tt := range tests { @@ -75,40 +54,33 @@ func TestCachePut(t *testing.T) { } func TestCacheDelete(t *testing.T) { - type args struct { - rsp *etcdadpt.Response - } tests := []struct { name string args args want int }{ - { - name: "first put 2 kvDocs, then delete 0 kvDoc, cache should store 2 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{}}}, - want: 2, + {"first put 2 kvDocs, then delete 0 kvDoc, cache should store 2 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{}}}, + 2, }, - { - name: "first put 2 kvDocs, then delete kvId=1, cache should store 1 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{ + {"first put 2 kvDocs, then delete kvId=1, cache should store 1 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, }}}, - want: 1, + 1, }, - { - name: "first put 2 kvDocs, then delete kvId=1 and kvId=2, cache should store 0 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ + {"first put 2 kvDocs, then delete kvId=1 and kvId=2, cache should store 0 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, {Value: []byte(`{"id":"2", "key":"withToys", "value":"yes", "labels":{"environment":"testing"}}`)}, }}}, - want: 0, + 0, }, - { - name: "first put 2 kvDocs, then delete non-exist kvId=0, cache should store 2 kvDocs", - args: args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ + {"first put 2 kvDocs, then delete non-exist kvId=0, cache should store 2 kvDocs", + args{&etcdadpt.Response{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{ {Value: []byte(`{"id":"0", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}, }}}, - want: 2, + 2, }, } for _, tt := range tests { @@ -124,193 +96,3 @@ func TestCacheDelete(t *testing.T) { }) } } - -func TestWatchCallBack(t *testing.T) { - type args struct { - rsp []*etcdadpt.Response - } - type want struct { - kvNum int - err error - } - tests := []struct { - name string - args args - want want - }{ - { - name: "receive 2 messages without kvs, expected: error is not nil, cache should store 0 kvDoc", - args: args{ - rsp: []*etcdadpt.Response{ - { - Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{}, - }, - - { - Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{}, - }, - }, - }, - want: want{ - kvNum: 0, - err: fmt.Errorf("unknown event"), - }, - }, - { - name: "receive 1 put message, put 0 kvDoc, expected: error is not nil, cache should store 0 kvDoc", - args: args{ - rsp: []*etcdadpt.Response{ - { - Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{}, - }, - }, - }, - want: want{ - kvNum: 0, - err: fmt.Errorf("unknown event"), - }, - }, - { - name: "receive 1 delete message, delete 0 kvDoc, expected: error is not nil, cache should store 0 kvDoc", - args: args{ - rsp: []*etcdadpt.Response{{Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{}}}, - }, - want: want{ - kvNum: 0, - err: fmt.Errorf("unknown event"), - }, - }, - { - name: "receive put message, put 1 kvDocs, expected: error is nil, cache should store 1 kvDocs", - args: args{ - rsp: []*etcdadpt.Response{ - { - Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{{Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}}}, - }, - }, - want: want{ - kvNum: 1, - err: nil, - }, - }, - { - name: "receive 1 put message, 1 delete message, first put 1 kvDoc, then delete it, expected: error is nil, cache should store 0 kvDoc", - args: args{ - rsp: []*etcdadpt.Response{ - { - Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{{Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}}, - }, - { - Action: etcdadpt.ActionDelete, Kvs: []*mvccpb.KeyValue{{Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"environment":"testing"}}`)}}, - }, - }, - }, - want: want{ - kvNum: 0, - err: nil, - }, - }, - { - name: "receive put message put 1 kvDoc, but labels are not cached, cache should store 0 kvDoc", - args: args{ - []*etcdadpt.Response{ - { - Action: etcdadpt.ActionPut, Kvs: []*mvccpb.KeyValue{{Value: []byte(`{"id":"1", "key":"withFruit", "value":"no", "labels":{"env":"testing"}}`)}}, - }, - }, - }, - want: want{ - kvNum: 0, - err: nil, - }, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - kc := NewKvCache() - for _, rsp := range tt.args.rsp { - err := kc.watchCallBack("", rsp) - assert.Equal(t, tt.want.err, err) - } - num := kc.kvDocCache.ItemCount() - assert.Equal(t, tt.want.kvNum, num) - }) - } -} - -func TestStoreAndLoadKvDoc(t *testing.T) { - type want struct { - kvDoc *model.KVDoc - exist bool - } - type args struct { - kvID string - kvDoc *model.KVDoc - expireTime time.Duration - waitTimeAfterStore time.Duration - } - tests := []struct { - name string - args args - want want - }{ - { - name: "store 1 kv and the expire time is 1 seconds, then load the kv with no wait time, expect: load kv successfully", - args: args{ - kvID: "", - kvDoc: &model.KVDoc{ - ID: "1", - Key: "withFood", - Value: "yes", - Labels: map[string]string{ - "environment": "testing", - }, - }, - expireTime: 1 * time.Second, - waitTimeAfterStore: 0, - }, - want: want{ - kvDoc: &model.KVDoc{ - ID: "1", - Key: "withFood", - Value: "yes", - Labels: map[string]string{ - "environment": "testing", - }, - }, - exist: true, - }, - }, - { - name: "store 1 kv and the expire time is 1 seconds, after waiting 2 seconds, then load the kv, expect: unable to load the kv", - args: args{ - kvID: "", - kvDoc: &model.KVDoc{ - ID: "1", - Key: "withFood", - Value: "yes", - Labels: map[string]string{ - "environment": "testing", - }, - }, - expireTime: 1 * time.Second, - waitTimeAfterStore: 2 * time.Second, - }, - want: want{ - kvDoc: nil, - exist: false, - }, - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - kc := NewKvCache() - kc.kvDocCache = goCache.New(tt.args.expireTime, tt.args.expireTime) - kc.StoreKvDoc(tt.args.kvID, tt.args.kvDoc) - time.Sleep(tt.args.waitTimeAfterStore) - doc, exist := kc.LoadKvDoc(tt.args.kvID) - assert.Equal(t, tt.want.exist, exist) - reflect.DeepEqual(tt.want.kvDoc, doc) - }) - } -}