store kv: readWriteLock optimization

.github/workflows/local_storage.yml

@@ -0,0 +1,24 @@
+name: Merge check for local
+on: [push, pull_request]
+jobs:
+  etcd-with-localstorage:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Go
+        uses: actions/setup-go@v1
+        with:
+          go-version: 1.18
+        id: go
+      - name: Check out code into the Go module directory
+        uses: actions/checkout@v1
+      - name: UT for etcd with local storage
+        run: |
+          time docker run -d -p 2379:2379 --name etcd quay.io/coreos/etcd etcd -name etcd --advertise-client-urls http://0.0.0.0:2379 --listen-client-urls http://0.0.0.0:2379
+          while ! nc -z 127.0.0.1 2379; do
+            sleep 1
+          done
+          export TEST_DB_KIND=etcd_with_localstorage
+          export TEST_DB_URI=127.0.0.1:2379
+          export TEST_KVS_ROOT_PATH=/data/kvs
+          sudo rm -rf /data/kvs
+          sudo time go test $(go list ./... | grep -v mongo | grep -v third_party | grep -v examples)

scripts/start.sh

@@ -73,6 +73,7 @@ EOM
 db:
   kind: ${db_type}
   uri: ${uri}
+  localFilePath: ${KVS_ROOT_PATH}
 EOM
 }
 

server/datasource/dao.go

@@ -22,7 +22,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-
 	"github.com/apache/servicecomb-kie/server/datasource/rbac"
 	"github.com/go-chassis/openlog"
 
@@ -120,7 +119,9 @@ type ViewDao interface {
 func Init(kind string) error {
 	var err error
 	f, ok := plugins[kind]
-	if kind != "etcd_with_localstorage" && kind != "embedded_etcd_with_localstorage" && !ok {
+
+	if !ok {
+		openlog.Info(fmt.Sprintf("do not support '%s'", kind))
 		return fmt.Errorf("do not support '%s'", kind)
 	}
 

server/datasource/local/file/fileprocess.go

@@ -14,7 +14,7 @@ var FileRootPath = "/data/kvs"
 
 var NewstKVFile = "newest_version.json"
 
-var MutexMap = make(map[string]*sync.Mutex)
+var MutexMap = make(map[string]*sync.RWMutex)
 var mutexMapLock = &sync.Mutex{}
 var rollbackMutexLock = &sync.Mutex{}
 var createDirMutexLock = &sync.Mutex{}
@@ -26,11 +26,11 @@ type FileDoRecord struct {
 	content  []byte
 }
 
-func GetOrCreateMutex(path string) *sync.Mutex {
+func GetOrCreateMutex(path string) *sync.RWMutex {
 	mutexMapLock.Lock()
 	mutex, ok := MutexMap[path]
 	if !ok {
-		mutex = &sync.Mutex{}
+		mutex = &sync.RWMutex{}
 		MutexMap[path] = mutex
 	}
 	mutexMapLock.Unlock()
@@ -213,8 +213,8 @@ func CleanDir(dir string) error {
 func ReadFile(filepath string) ([]byte, error) {
 	// check the file is empty
 	mutex := GetOrCreateMutex(path.Dir(filepath))
-	mutex.Lock()
-	defer mutex.Unlock()
+	mutex.RLocker()
+	defer mutex.RLocker()
 
 	content, err := os.ReadFile(filepath)
 	if err != nil {
@@ -226,8 +226,8 @@ func ReadFile(filepath string) ([]byte, error) {
 
 func CountInDomain(dir string) (int, error) {
 	mutex := GetOrCreateMutex(dir)
-	mutex.Lock()
-	defer mutex.Unlock()
+	mutex.RLock()
+	defer mutex.RUnlock()
 
 	files, err := os.ReadDir(dir)
 	if err != nil {
@@ -279,8 +279,8 @@ func ReadAllKvsFromProjectFolder(dir string) ([][]byte, error) {
 
 func ReadAllFiles(dir string) ([]string, [][]byte, error) {
 	mutex := GetOrCreateMutex(dir)
-	mutex.Lock()
-	defer mutex.Unlock()
+	mutex.RLock()
+	defer mutex.RUnlock()
 
 	files := []string{}
 	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {

server/datasource/local/kv/kv_cache.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/apache/servicecomb-kie/pkg/model"
+	"github.com/apache/servicecomb-kie/pkg/stringutil"
 	"github.com/apache/servicecomb-kie/server/datasource"
 	"github.com/go-chassis/openlog"
 	goCache "github.com/patrickmn/go-cache"
@@ -94,11 +95,14 @@ func (kc *Cache) GetKvDoc(kv *mvccpb.KeyValue) (*model.KVDoc, error) {
 	return kvDoc, nil
 }
 
-func (kc *Cache) GetCacheKey(domain, project string) string {
+
+func (kc *Cache) GetCacheKey(domain, project string, labels map[string]string) string {
+	labelFormat := stringutil.FormatMap(labels)
 	inputKey := strings.Join([]string{
 		"",
 		domain,
 		project,
+		labelFormat,
 	}, "/")
 	return inputKey
 }
@@ -142,7 +146,7 @@ func (kc *Cache) LoadKvDoc(kvID string) (*model.KVDoc, bool) {
 func (kc *Cache) CachePut(kvs []*model.KVDoc) {
 	for _, kvDoc := range kvs {
 		kc.StoreKvDoc(kvDoc.ID, kvDoc)
-		cacheKey := kc.GetCacheKey(kvDoc.Domain, kvDoc.Project)
+		cacheKey := kc.GetCacheKey(kvDoc.Domain, kvDoc.Project, kvDoc.Labels)
 		m, ok := kc.LoadKvIDSet(cacheKey)
 		if !ok {
 			kc.StoreKvIDSet(cacheKey, IDSet{kvDoc.ID: struct{}{}})
@@ -156,7 +160,7 @@ func (kc *Cache) CachePut(kvs []*model.KVDoc) {
 func (kc *Cache) CacheDelete(kvs []*model.KVDoc) {
 	for _, kvDoc := range kvs {
 		kc.DeleteKvDoc(kvDoc.ID)
-		cacheKey := kc.GetCacheKey(kvDoc.Domain, kvDoc.Project)
+		cacheKey := kc.GetCacheKey(kvDoc.Domain, kvDoc.Project, kvDoc.Labels)
 		m, ok := kc.LoadKvIDSet(cacheKey)
 		if !ok {
 			openlog.Error("cacheKey " + cacheKey + "not exists")
@@ -175,7 +179,7 @@ func Search(req *CacheSearchReq) (*model.KVResponse, bool, []string) {
 	result := &model.KVResponse{
 		Data: []*model.KVDoc{},
 	}
-	cacheKey := kvCache.GetCacheKey(req.Domain, req.Project)
+	cacheKey := kvCache.GetCacheKey(req.Domain, req.Project, req.Opts.Labels)
 	kvIds, ok := kvCache.LoadKvIDSet(cacheKey)
 	if !ok {
 		kvCache.StoreKvIDSet(cacheKey, IDSet{})

server/datasource/local/kv/kv_dao.go

@@ -490,9 +490,6 @@ func pagingResult(result *model.KVResponse, opts datasource.FindOptions) *model.
 }
 
 func filterMatch(doc *model.KVDoc, opts datasource.FindOptions, regex *regexp.Regexp) bool {
-	if opts.Key != "" && doc.Key != opts.Key {
-		return false
-	}
 	if opts.Status != "" && doc.Status != opts.Status {
 		return false
 	}

test/init.go

@@ -25,6 +25,7 @@ import (
 	_ "github.com/go-chassis/cari/db/bootstrap"
 
 	_ "github.com/apache/servicecomb-kie/server/datasource/etcd"
+	_ "github.com/apache/servicecomb-kie/server/datasource/local"
 	_ "github.com/apache/servicecomb-kie/server/datasource/mongo"
 	_ "github.com/apache/servicecomb-kie/server/plugin/qms"
 	_ "github.com/apache/servicecomb-kie/server/pubsub/notifier"
@@ -42,8 +43,9 @@ import (
 )
 
 var (
-	uri  string
-	kind string
+	uri           string
+	kind          string
+	localFilePath string
 )
 
 func init() {
@@ -54,6 +56,8 @@ func init() {
 	}
 	kind = archaius.GetString("TEST_DB_KIND", "etcd")
 	uri = archaius.GetString("TEST_DB_URI", "http://127.0.0.1:2379")
+	localFilePath = archaius.GetString("TEST_KVS_ROOT_PATH", "")
+
 	err = archaius.Init(archaius.WithMemorySource())
 	if err != nil {
 		panic(err)
@@ -71,9 +75,10 @@ func init() {
 		panic(err)
 	}
 	err = db.Init(config.DB{
-		URI:     uri,
-		Timeout: "10s",
-		Kind:    kind,
+		URI:           uri,
+		Timeout:       "10s",
+		Kind:          kind,
+		LocalFilePath: localFilePath,
 	})
 	if err != nil {
 		panic(err)
@@ -82,7 +87,15 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-	err = edatasource.Init(kind)
+
+	edatasourceKind := kind
+	if kind == "etcd_with_localstorage" {
+		edatasourceKind = "etcd"
+	}
+	if kind == "embedded_etcd_with_localstorage" {
+		edatasourceKind = "embedded_etcd"
+	}
+	err = edatasource.Init(edatasourceKind)
 	if err != nil {
 		panic(err)
 	}
@@ -115,5 +128,5 @@ func randomListenAddress() string {
 }
 
 func IsEmbeddedetcdMode() bool {
-	return kind == "embedded_etcd"
+	return kind == "embedded_etcd" || kind == "embedded_etcd_with_localstorage"
 }