add local storage test

.github/workflows/eventbase-ci.yml

@@ -38,25 +38,4 @@ jobs:
           export TEST_DB_MODE=etcd
           export TEST_DB_URI=http://127.0.0.1:2379
           cd eventbase
-          time go test -short -covermode=atomic $(go list ./... | grep -v mongo | grep -v third_party | grep -v examples)
-  etcd-storage-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 source code
-        uses: actions/checkout@v1
-      - name: UT for etcd
-        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_MODE=etcd
-          export TEST_DB_URI=http://127.0.0.1:2379
-          export SCHEMA_ROOT_PATH=/data/kvs
-          cd eventbase
-          sudo time go test -short -covermode=atomic $(go list ./... | grep -v mongo | grep -v third_party | grep -v examples)
+          time go test -short -covermode=atomic $(go list ./... | grep -v mongo | grep -v third_party | grep -v examples)

.github/workflows/static_check.yml

@@ -43,6 +43,20 @@ jobs:
           sudo docker-compose -f ./scripts/docker-compose.yaml up -d
           sleep 20
           bash -x scripts/ut_test_in_docker.sh mongo
+  local:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Go 1.18
+        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-LOCAL_STORAGE
+        run: |
+          rm -rf /data/schemas
+          bash -x scripts/ut_test_in_docker.sh local
   integration-test:
     runs-on: ubuntu-latest
     steps:

datasource/etcd/util/microservice_util.go

@@ -21,6 +21,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"path/filepath"
 	"strings"
 
 	pb "github.com/go-chassis/cari/discovery"
@@ -31,6 +32,8 @@ import (
 	"github.com/apache/servicecomb-service-center/datasource/etcd/sd"
 	"github.com/apache/servicecomb-service-center/datasource/etcd/state/kvstore"
 	"github.com/apache/servicecomb-service-center/datasource/etcd/sync"
+	"github.com/apache/servicecomb-service-center/datasource/local"
+	"github.com/apache/servicecomb-service-center/datasource/schema"
 	"github.com/apache/servicecomb-service-center/pkg/log"
 	"github.com/apache/servicecomb-service-center/pkg/util"
 	"github.com/apache/servicecomb-service-center/server/config"
@@ -247,6 +250,32 @@ func UpdateService(ctx context.Context, domainProject string, serviceID string,
 		log.Error("marshal service file failed", err)
 		return opts, err
 	}
+
+	if schema.StorageType == "local" {
+		contents := make([]*schema.ContentItem, len(service.Schemas))
+		err = schema.Instance().PutManyContent(ctx, &schema.PutManyContentRequest{
+			ServiceID: service.ServiceId,
+			SchemaIDs: service.Schemas,
+			Contents:  contents,
+			Init:      true,
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		serviceMutex := local.GetOrCreateMutex(service.ServiceId)
+		serviceMutex.Lock()
+		defer serviceMutex.Unlock()
+	}
+	defer func() {
+		if schema.StorageType == "local" && err != nil {
+			cleanDirErr := local.CleanDir(filepath.Join(schema.RootFilePath, domainProject, service.ServiceId))
+			if cleanDirErr != nil {
+				log.Error("clean dir error when rollback in RegisterService", cleanDirErr)
+			}
+		}
+	}()
+
 	opt := etcdadpt.OpPut(etcdadpt.WithStrKey(key), etcdadpt.WithValue(data))
 	opts = append(opts, opt)
 	syncOpts, err := sync.GenUpdateOpts(ctx, datasource.ResourceKV, data, sync.WithOpts(map[string]string{"key": key}))

datasource/local/schema.go

@@ -24,6 +24,7 @@ import (
 	"io/fs"
 	"os"
 	pathutil "path"
+	"path/filepath"
 	"strings"
 	"sync"
 
@@ -36,8 +37,6 @@ import (
 	"github.com/go-chassis/cari/discovery"
 	"github.com/go-chassis/openlog"
 	"github.com/little-cui/etcdadpt"
-
-	"path/filepath"
 )
 
 var MutexMap = make(map[string]*sync.RWMutex)
@@ -49,7 +48,7 @@ func init() {
 	schema.Install("local", NewSchemaDAO)
 }
 
-func NewSchemaDAO(opts schema.Options) (schema.DAO, error) {
+func NewSchemaDAO(_ schema.Options) (schema.DAO, error) {
 	return &SchemaDAO{}, nil
 }
 
@@ -356,6 +355,9 @@ func (s *SchemaDAO) ListRef(ctx context.Context, refRequest *schema.RefRequest)
 	schemaIDs, contents, err := ReadAllFiles(dir)
 
 	if err != nil {
+		if os.IsNotExist(err) {
+			return make([]*schema.Ref, 0), nil
+		}
 		log.Error(fmt.Sprintf("read service[%s] schema content files failed ", serviceID), err)
 		return nil, err
 	}
@@ -433,7 +435,7 @@ func (s *SchemaDAO) DeleteRef(ctx context.Context, refRequest *schema.RefRequest
 	return nil
 }
 
-func (s *SchemaDAO) GetContent(ctx context.Context, contentRequest *schema.ContentRequest) (*schema.Content, error) {
+func (s *SchemaDAO) GetContent(_ context.Context, _ *schema.ContentRequest) (*schema.Content, error) {
 	// no usage, should not be called
 	log.Error("Occur error when call SchemaDAO.GetContent, this method should not be called in any condition", schema.ErrSchemaNotFound)
 	return nil, schema.ErrSchemaNotFound
@@ -477,13 +479,13 @@ func (s *SchemaDAO) PutContent(ctx context.Context, contentRequest *schema.PutCo
 		return err
 	}
 
-	var schemaIdValid = false
-	for _, serviceSchemaId := range service.Schemas {
-		if serviceSchemaId == contentRequest.SchemaID {
-			schemaIdValid = true
+	var schemaIDValid = false
+	for _, serviceSchemaID := range service.Schemas {
+		if serviceSchemaID == contentRequest.SchemaID {
+			schemaIDValid = true
 		}
 	}
-	if !schemaIdValid {
+	if !schemaIDValid {
 		err = schema.ErrSchemaNotFound
 		log.Error(fmt.Sprintf("update service[%s] failed when valide schema id", serviceID), err)
 		return err
@@ -571,7 +573,7 @@ func (s *SchemaDAO) PutManyContent(ctx context.Context, contentRequest *schema.P
 
 	// create or update files
 	for i := 0; i < len(contentRequest.SchemaIDs); i++ {
-		schemaId := contentRequest.SchemaIDs[i]
+		schemaID := contentRequest.SchemaIDs[i]
 		schema := contentRequest.Contents[i]
 
 		schemaBytes, marshalErr := json.Marshal(schema)
@@ -580,7 +582,7 @@ func (s *SchemaDAO) PutManyContent(ctx context.Context, contentRequest *schema.P
 			openlog.Error("fail to marshal kv " + err.Error())
 			return err
 		}
-		err = createOrUpdateFile(servicepath+"/"+schemaId+".json", schemaBytes, &rollbackOperations, false)
+		err = createOrUpdateFile(servicepath+"/"+schemaID+".json", schemaBytes, &rollbackOperations, false)
 		if err != nil {
 			break
 		}
@@ -614,13 +616,13 @@ func (s *SchemaDAO) PutManyContent(ctx context.Context, contentRequest *schema.P
 	return err
 }
 
-func (s *SchemaDAO) DeleteContent(ctx context.Context, contentRequest *schema.ContentRequest) error {
+func (s *SchemaDAO) DeleteContent(_ context.Context, _ *schema.ContentRequest) error {
 	// no usage, should not be called
 	log.Error("Occur error when call SchemaDAO.DeleteContent, this method should not be called in any condition", schema.ErrSchemaContentNotFound)
 	return schema.ErrSchemaContentNotFound
 }
 
-func (s *SchemaDAO) DeleteNoRefContents(ctx context.Context) (int, error) {
+func (s *SchemaDAO) DeleteNoRefContents(_ context.Context) (int, error) {
 	// no usage, should not be called
 	log.Error("Occur error when call SchemaDAO.DeleteNoRefContents, this method should not be called in any condition", schema.ErrSchemaNotFound)
 	return 0, schema.ErrSchemaNotFound

datasource/schema/init.go

@@ -47,11 +47,11 @@ func Init(opts Options) error {
 		return nil
 	}
 	kind := opts.Kind
-	if strings.Trim(config.GetString("schema.root.path", "", config.WithStandby("schemaRootPath")), " ") != "" {
-		log.Warn("llllllllllocal")
+	rootFilePath := config.GetString("schema.root.path", "", config.WithStandby("schemaRootPath"))
+	if strings.Trim(rootFilePath, " ") != "" {
 		kind = "local"
 		StorageType = "local"
-		RootFilePath = config.GetRegistry().SchemaRootPath
+		RootFilePath = rootFilePath
 	}
 
 	engineFunc, ok := plugins[kind]

scripts/ut_test_in_docker.sh

@@ -43,6 +43,14 @@ elif [ ${db_name} == "mongo" ];then
     sleep 1
   done
   echo "${green}mongodb is running......${reset}"
+elif [ ${db_name} == "local" ];then
+  echo "${green}Starting etcd in docker${reset}"
+  docker run -d -v /usr/share/ca-certificates/:/etc/ssl/certs -p 40010:40010 -p 23800:23800 -p 2379:2379 --name etcd quay.io/coreos/etcd etcd -name etcd0 -advertise-client-urls http://127.0.0.1:2379,http://127.0.0.1:40010 -listen-client-urls http://0.0.0.0:2379,http://0.0.0.0:40010 -initial-advertise-peer-urls http://127.0.0.1:23800 -listen-peer-urls http://0.0.0.0:23800 -initial-cluster-token etcd-cluster-1 -initial-cluster etcd0=http://127.0.0.1:23800 -initial-cluster-state new
+  while ! nc -z 127.0.0.1 2379; do
+    echo "Waiting Etcd to launch on 2379..."
+    sleep 1
+  done
+  echo "${green}Etcd is running......${reset}"
 else
   echo "${db_name} non-existent"
 	exit 1
@@ -57,6 +65,9 @@ if [ ${db_name} == "etcd" ];then
 elif [ ${db_name} == "mongo" ];then
   export TEST_MODE=mongo
   [ $? == 0 ] && ut_for_dir 'datasource/etcd\|datasource/schema'
+elif [ ${db_name} == "local" ];then
+  export TEST_MODE=local
+  [ $? == 0 ] && ut_for_dir 'datasource/etcd\|datasource/schema'
 else
   echo "${db_name} non-existent"
 	exit 1

server/service/disco/schema.go

@@ -235,7 +235,7 @@ func getOldSchemaIDs(ctx context.Context, serviceID string) ([]string, error) {
 	return schemaIDs, nil
 }
 
-func mergeRequests(ctx context.Context, serviceID string, refs []*schema.Ref, oldSchemaIDs []string) ([]*pb.GetSchemaRequest, error) {
+func mergeRequests(_ context.Context, serviceID string, refs []*schema.Ref, oldSchemaIDs []string) ([]*pb.GetSchemaRequest, error) {
 	set := mapset.NewSet()
 	for _, schemaID := range oldSchemaIDs {
 		set.Add(schemaID)

test/test.go

@@ -49,6 +49,13 @@ func init() {
 		_ = archaius.Set("registry.kind", "etcd")
 		_ = archaius.Set("registry.etcd.cluster.name", "sc-0")
 		_ = archaius.Set("registry.etcd.cluster.endpoints", "sc-0="+uri+",sc-1=http://127.0.0.2:2379")
+	} else if IsLOCAL() {
+		_ = archaius.Set("registry.cache.mode", 0)
+		_ = archaius.Set("discovery.kind", "etcd")
+		_ = archaius.Set("registry.kind", "etcd")
+		_ = archaius.Set("registry.etcd.cluster.name", "sc-0")
+		_ = archaius.Set("registry.etcd.cluster.endpoints", "sc-0="+uri+",sc-1=http://127.0.0.2:2379")
+		_ = archaius.Set("schema.root.path", "/data/schemas")
 	} else {
 		_ = archaius.Set("registry.heartbeat.kind", "checker")
 		kind = "mongo"
@@ -81,3 +88,8 @@ func IsETCD() bool {
 	}
 	return t == "etcd"
 }
+
+func IsLOCAL() bool {
+	t := archaius.Get("TEST_MODE")
+	return t == "local"
+}