You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ti...@apache.org on 2020/09/30 06:09:01 UTC
[servicecomb-service-center] branch master updated: [Enhancement]
Implement service/ms exist interface (#701)
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 e3a9443 [Enhancement] Implement service/ms exist interface (#701)
e3a9443 is described below
commit e3a94432616248cc76a15d0e041d4b2af1342b34
Author: popozy <18...@163.com>
AuthorDate: Wed Sep 30 14:08:52 2020 +0800
[Enhancement] Implement service/ms exist interface (#701)
1. implement service/ms/etcd interface
+ ExistService
+ ModifySchema
+ ModifySchemas
+ ExistSchema
2. finish unit test
---
server/service/ms/datasource.go | 6 +-
server/service/ms/etcd/etcd.go | 386 +++++++++++++-
server/service/ms/etcd/etcd_test.go | 989 +++++++++++++++++++++++++++++++++++-
server/service/ms/etcd/util.go | 306 ++++++++++-
4 files changed, 1670 insertions(+), 17 deletions(-)
diff --git a/server/service/ms/datasource.go b/server/service/ms/datasource.go
index 61e81bb..62ff6d2 100644
--- a/server/service/ms/datasource.go
+++ b/server/service/ms/datasource.go
@@ -25,6 +25,7 @@ type DataSource interface {
RegisterService(ctx context.Context, request *pb.CreateServiceRequest) (*pb.CreateServiceResponse, error)
GetServices(ctx context.Context, request *pb.GetServicesRequest) (*pb.GetServicesResponse, error)
GetService(ctx context.Context, request *pb.GetServiceRequest) (*pb.GetServiceResponse, error)
+ ExistService(ctx context.Context, request *pb.GetExistenceRequest) (*pb.GetExistenceResponse, error)
UpdateService(ctx context.Context, request *pb.UpdateServicePropsRequest) (*pb.UpdateServicePropsResponse, error)
UnregisterService(ctx context.Context, request *pb.DeleteServiceRequest) (*pb.DeleteServiceResponse, error)
@@ -33,9 +34,10 @@ type DataSource interface {
UpdateInstance()
UnRegisterInstance()
- AddSchema()
+ ModifySchemas(ctx context.Context, request *pb.ModifySchemasRequest) (*pb.ModifySchemasResponse, error)
+ ModifySchema(ctx context.Context, request *pb.ModifySchemaRequest) (*pb.ModifySchemaResponse, error)
+ ExistSchema(ctx context.Context, request *pb.GetExistenceRequest) (*pb.GetExistenceResponse, error)
GetSchema()
- UpdateSchema()
DeleteSchema()
AddTag()
diff --git a/server/service/ms/etcd/etcd.go b/server/service/ms/etcd/etcd.go
index 4a77b8c..49e3b3a 100644
--- a/server/service/ms/etcd/etcd.go
+++ b/server/service/ms/etcd/etcd.go
@@ -17,22 +17,36 @@ package etcd
import (
"context"
+ "fmt"
"github.com/apache/servicecomb-service-center/pkg/log"
pb "github.com/apache/servicecomb-service-center/pkg/registry"
"github.com/apache/servicecomb-service-center/pkg/util"
+ apt "github.com/apache/servicecomb-service-center/server/core"
"github.com/apache/servicecomb-service-center/server/core/backend"
"github.com/apache/servicecomb-service-center/server/core/proto"
+ "github.com/apache/servicecomb-service-center/server/plugin"
+ "github.com/apache/servicecomb-service-center/server/plugin/quota"
+ "github.com/apache/servicecomb-service-center/server/plugin/registry"
scerr "github.com/apache/servicecomb-service-center/server/scerror"
+ "os"
+ "strings"
)
// TODO: define error with names here
+var schemaEditable bool
+
func init() {
// TODO: set logger
// TODO: register storage plugin to plugin manager
+ schemaEditableConfig := strings.ToLower(os.Getenv("SCHEMA_EDITABLE"))
+ schemaEditable = strings.Compare(schemaEditableConfig, "true") == 0
}
-type DataSource struct{}
+type DataSource struct {
+ // schemaEditable determines whether schema modification is allowed for
+ SchemaEditable bool
+}
func NewDataSource() *DataSource {
// TODO: construct a reasonable DataSource instance
@@ -47,6 +61,7 @@ func NewDataSource() *DataSource {
func (ds *DataSource) initialize() error {
// TODO: init DataSource members
+ ds.SchemaEditable = schemaEditable
return nil
}
@@ -60,7 +75,7 @@ func (ds *DataSource) RegisterService(ctx context.Context, request *pb.CreateSer
serviceBody := request.Service
// construct data to invoke etcd client
- opts, uniqueCmpOpts, failOpts, err := CapRegisterData(ctx, request)
+ opts, uniqueCmpOpts, failOpts, err := capRegisterData(ctx, request)
if err != nil {
return &pb.CreateServiceResponse{
Response: proto.CreateResponse(scerr.ErrInternal, err.Error()),
@@ -68,14 +83,14 @@ func (ds *DataSource) RegisterService(ctx context.Context, request *pb.CreateSer
}
resp, err := backend.Registry().TxnWithCmp(ctx, opts, uniqueCmpOpts, failOpts)
- return NewRegisterServiceResp(ctx, serviceBody, resp, err)
+ return newRegisterServiceResp(ctx, serviceBody, resp, err)
}
func (ds *DataSource) GetServices(ctx context.Context, request *pb.GetServicesRequest) (
*pb.GetServicesResponse, error) {
domainProject := util.ParseDomainProject(ctx)
- services, err := GetAllServiceUtil(ctx, domainProject)
+ services, err := getAllServiceUtil(ctx, domainProject)
if err != nil {
log.Errorf(err, "get all services by domain failed")
return &pb.GetServicesResponse{
@@ -92,7 +107,7 @@ func (ds *DataSource) GetServices(ctx context.Context, request *pb.GetServicesRe
func (ds *DataSource) GetService(ctx context.Context, request *pb.GetServiceRequest) (
*pb.GetServiceResponse, error) {
domainProject := util.ParseDomainProject(ctx)
- singleService, err := GetSingleService(ctx, domainProject, request.ServiceId)
+ singleService, err := getSingleService(ctx, domainProject, request.ServiceId)
if err != nil {
log.Errorf(err, "get micro-service[%s] failed, get service file failed", request.ServiceId)
@@ -112,6 +127,44 @@ func (ds *DataSource) GetService(ctx context.Context, request *pb.GetServiceRequ
}, nil
}
+func (ds *DataSource) ExistService(ctx context.Context, request *pb.GetExistenceRequest) (*pb.GetExistenceResponse,
+ error) {
+ domainProject := util.ParseDomainProject(ctx)
+ serviceFlag := util.StringJoin([]string{
+ request.Environment, request.AppId, request.ServiceName, request.Version}, "/")
+
+ ids, exist, err := findServiceIds(ctx, request.Version, &pb.MicroServiceKey{
+ Environment: request.Environment,
+ AppId: request.AppId,
+ ServiceName: request.ServiceName,
+ Alias: request.ServiceName,
+ Version: request.Version,
+ Tenant: domainProject,
+ })
+ if err != nil {
+ log.Errorf(err, "micro-service[%s] exist failed, find serviceIDs failed", serviceFlag)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal, err.Error()),
+ }, err
+ }
+ if !exist {
+ log.Infof("micro-service[%s] exist failed, service does not exist", serviceFlag)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrServiceNotExists, serviceFlag+" does not exist."),
+ }, nil
+ }
+ if len(ids) == 0 {
+ log.Infof("micro-service[%s] exist failed, version mismatch", serviceFlag)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrServiceVersionNotExists, serviceFlag+" version mismatch."),
+ }, nil
+ }
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(proto.Response_SUCCESS, "get service id successfully."),
+ ServiceId: ids[0], // 约定多个时，取较新版本
+ }, nil
+}
+
func (ds *DataSource) UpdateService(ctx context.Context, service *pb.UpdateServicePropsRequest) (
*pb.UpdateServicePropsResponse, error) {
panic("implement me")
@@ -138,8 +191,71 @@ func (ds *DataSource) UnRegisterInstance() {
panic("implement me")
}
-func (ds *DataSource) AddSchema() {
- panic("implement me")
+func (ds *DataSource) ModifySchemas(ctx context.Context, request *pb.ModifySchemasRequest) (
+ *pb.ModifySchemasResponse, error) {
+ remoteIP := util.GetIPFromContext(ctx)
+ serviceID := request.ServiceId
+ domainProject := util.ParseDomainProject(ctx)
+
+ serviceInfo, err := getService(ctx, domainProject, serviceID)
+ if err != nil {
+ log.Errorf(err, "modify service[%s] schemas failed, get service failed, operator: %s", serviceID, remoteIP)
+ return &pb.ModifySchemasResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal, err.Error()),
+ }, err
+ }
+ if serviceInfo == nil {
+ log.Errorf(nil, "modify service[%s] schemas failed, service does not exist, operator: %s",
+ serviceID, remoteIP)
+ return &pb.ModifySchemasResponse{
+ Response: proto.CreateResponse(scerr.ErrServiceNotExists, "Service does not exist."),
+ }, nil
+ }
+
+ respErr := ds.modifySchemas(ctx, domainProject, serviceInfo, request.Schemas)
+ if respErr != nil {
+ log.Errorf(nil, "modify service[%s] schemas failed, operator: %s", serviceID, remoteIP)
+ resp := &pb.ModifySchemasResponse{
+ Response: proto.CreateResponseWithSCErr(respErr),
+ }
+ if respErr.InternalError() {
+ return resp, respErr
+ }
+ return resp, nil
+ }
+
+ return &pb.ModifySchemasResponse{
+ Response: proto.CreateResponse(proto.Response_SUCCESS, "modify schemas info successfully."),
+ }, nil
+}
+
+func (ds *DataSource) ModifySchema(ctx context.Context, request *pb.ModifySchemaRequest) (
+ *pb.ModifySchemaResponse, error) {
+ remoteIP := util.GetIPFromContext(ctx)
+ serviceID := request.ServiceId
+ schemaID := request.SchemaId
+
+ schema := pb.Schema{
+ SchemaId: schemaID,
+ Summary: request.Summary,
+ Schema: request.Schema,
+ }
+ err := ds.modifySchema(ctx, serviceID, &schema)
+ if err != nil {
+ log.Errorf(err, "modify schema[%s/%s] failed, operator: %s", serviceID, schemaID, remoteIP)
+ resp := &pb.ModifySchemaResponse{
+ Response: proto.CreateResponseWithSCErr(err),
+ }
+ if err.InternalError() {
+ return resp, err
+ }
+ return resp, nil
+ }
+
+ log.Infof("modify schema[%s/%s] successfully, operator: %s", serviceID, schemaID, remoteIP)
+ return &pb.ModifySchemaResponse{
+ Response: proto.CreateResponse(proto.Response_SUCCESS, "modify schema info success"),
+ }, nil
}
func (ds *DataSource) GetSchema() {
@@ -154,6 +270,46 @@ func (ds *DataSource) DeleteSchema() {
panic("implement me")
}
+func (ds *DataSource) ExistSchema(ctx context.Context, request *pb.GetExistenceRequest) (
+ *pb.GetExistenceResponse, error) {
+ domainProject := util.ParseDomainProject(ctx)
+
+ if !serviceExist(ctx, domainProject, request.ServiceId) {
+ log.Warnf("schema[%s/%s] exist failed, service does not exist", request.ServiceId, request.SchemaId)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrServiceNotExists, "service does not exist."),
+ }, nil
+ }
+
+ key := apt.GenerateServiceSchemaKey(domainProject, request.ServiceId, request.SchemaId)
+ exist, err := checkSchemaInfoExist(ctx, key)
+ if err != nil {
+ log.Errorf(err, "schema[%s/%s] exist failed, get schema failed", request.ServiceId, request.SchemaId)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal, err.Error()),
+ }, err
+ }
+ if !exist {
+ log.Infof("schema[%s/%s] exist failed, schema does not exist", request.ServiceId, request.SchemaId)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrSchemaNotExists, "schema does not exist."),
+ }, nil
+ }
+ schemaSummary, err := getSchemaSummary(ctx, domainProject, request.ServiceId, request.SchemaId)
+ if err != nil {
+ log.Errorf(err, "schema[%s/%s] exist failed, get schema summary failed",
+ request.ServiceId, request.SchemaId)
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal, err.Error()),
+ }, err
+ }
+ return &pb.GetExistenceResponse{
+ Response: proto.CreateResponse(proto.Response_SUCCESS, "Schema exist."),
+ SchemaId: request.SchemaId,
+ Summary: schemaSummary,
+ }, nil
+}
+
func (ds *DataSource) AddTag() {
panic("implement me")
}
@@ -185,3 +341,219 @@ func (ds *DataSource) UpdateRule() {
func (ds *DataSource) DeleteRule() {
panic("implement me")
}
+
+func (ds *DataSource) modifySchemas(ctx context.Context, domainProject string, service *pb.MicroService,
+ schemas []*pb.Schema) *scerr.Error {
+ remoteIP := util.GetIPFromContext(ctx)
+ serviceID := service.ServiceId
+ schemasFromDatabase, err := getSchemasFromDatabase(ctx, domainProject, serviceID)
+ if err != nil {
+ log.Errorf(nil, "modify service[%s] schemas failed, get schemas failed, operator: %s",
+ serviceID, remoteIP)
+ return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+
+ needUpdateSchemas, needAddSchemas, needDeleteSchemas, nonExistSchemaIds :=
+ schemasAnalysis(schemas, schemasFromDatabase, service.Schemas)
+
+ pluginOps := make([]registry.PluginOp, 0)
+ if !ds.isSchemaEditable(service) {
+ if len(service.Schemas) == 0 {
+ res := quota.NewApplyQuotaResource(quota.SchemaQuotaType, domainProject, serviceID, int64(len(nonExistSchemaIds)))
+ rst := plugin.Plugins().Quota().Apply4Quotas(ctx, res)
+ errQuota := rst.Err
+ if errQuota != nil {
+ log.Errorf(errQuota, "modify service[%s] schemas failed, operator: %s", serviceID, remoteIP)
+ return errQuota
+ }
+
+ service.Schemas = nonExistSchemaIds
+ opt, err := updateService(domainProject, serviceID, service)
+ if err != nil {
+ log.Errorf(err, "modify service[%s] schemas failed, update service.Schemas failed, operator: %s",
+ serviceID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ pluginOps = append(pluginOps, opt)
+ } else {
+ if len(nonExistSchemaIds) != 0 {
+ errInfo := fmt.Errorf("non-existent schemaIDs %v", nonExistSchemaIds)
+ log.Errorf(errInfo, "modify service[%s] schemas failed, operator: %s", serviceID, remoteIP)
+ return scerr.NewError(scerr.ErrUndefinedSchemaID, errInfo.Error())
+ }
+ for _, needUpdateSchema := range needUpdateSchemas {
+ exist, err := isExistSchemaSummary(ctx, domainProject, serviceID, needUpdateSchema.SchemaId)
+ if err != nil {
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ if !exist {
+ opts := schemaWithDatabaseOpera(registry.OpPut, domainProject, serviceID, needUpdateSchema)
+ pluginOps = append(pluginOps, opts...)
+ } else {
+ log.Warnf("schema[%s/%s] and it's summary already exist, skip to update, operator: %s",
+ serviceID, needUpdateSchema.SchemaId, remoteIP)
+ }
+ }
+ }
+
+ for _, schema := range needAddSchemas {
+ log.Infof("add new schema[%s/%s], operator: %s", serviceID, schema.SchemaId, remoteIP)
+ opts := schemaWithDatabaseOpera(registry.OpPut, domainProject, service.ServiceId, schema)
+ pluginOps = append(pluginOps, opts...)
+ }
+ } else {
+ quotaSize := len(needAddSchemas) - len(needDeleteSchemas)
+ if quotaSize > 0 {
+ res := quota.NewApplyQuotaResource(quota.SchemaQuotaType, domainProject, serviceID, int64(quotaSize))
+ rst := plugin.Plugins().Quota().Apply4Quotas(ctx, res)
+ err := rst.Err
+ if err != nil {
+ log.Errorf(err, "modify service[%s] schemas failed, operator: %s", serviceID, remoteIP)
+ return err
+ }
+ }
+
+ var schemaIDs []string
+ for _, schema := range needAddSchemas {
+ log.Infof("add new schema[%s/%s], operator: %s", serviceID, schema.SchemaId, remoteIP)
+ opts := schemaWithDatabaseOpera(registry.OpPut, domainProject, service.ServiceId, schema)
+ pluginOps = append(pluginOps, opts...)
+ schemaIDs = append(schemaIDs, schema.SchemaId)
+ }
+
+ for _, schema := range needUpdateSchemas {
+ log.Infof("update schema[%s/%s], operator: %s", serviceID, schema.SchemaId, remoteIP)
+ opts := schemaWithDatabaseOpera(registry.OpPut, domainProject, serviceID, schema)
+ pluginOps = append(pluginOps, opts...)
+ schemaIDs = append(schemaIDs, schema.SchemaId)
+ }
+
+ for _, schema := range needDeleteSchemas {
+ log.Infof("delete non-existent schema[%s/%s], operator: %s", serviceID, schema.SchemaId, remoteIP)
+ opts := schemaWithDatabaseOpera(registry.OpDel, domainProject, serviceID, schema)
+ pluginOps = append(pluginOps, opts...)
+ }
+
+ service.Schemas = schemaIDs
+ opt, err := updateService(domainProject, serviceID, service)
+ if err != nil {
+ log.Errorf(err, "modify service[%s] schemas failed, update service.Schemas failed, operator: %s",
+ serviceID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ pluginOps = append(pluginOps, opt)
+ }
+
+ if len(pluginOps) != 0 {
+ resp, err := backend.BatchCommitWithCmp(ctx, pluginOps,
+ []registry.CompareOp{registry.OpCmp(
+ registry.CmpVer(util.StringToBytesWithNoCopy(apt.GenerateServiceKey(domainProject, serviceID))),
+ registry.CmpNotEqual, 0)},
+ nil)
+ if err != nil {
+ return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+ if !resp.Succeeded {
+ return scerr.NewError(scerr.ErrServiceNotExists, "Service does not exist.")
+ }
+ }
+ return nil
+}
+
+func (ds *DataSource) isSchemaEditable(service *pb.MicroService) bool {
+ return (len(service.Environment) != 0 && service.Environment != pb.ENV_PROD) || ds.SchemaEditable
+}
+
+func (ds *DataSource) modifySchema(ctx context.Context, serviceID string, schema *pb.Schema) *scerr.Error {
+ remoteIP := util.GetIPFromContext(ctx)
+ domainProject := util.ParseDomainProject(ctx)
+ schemaID := schema.SchemaId
+
+ microService, err := getService(ctx, domainProject, serviceID)
+ if err != nil {
+ log.Errorf(err, "modify schema[%s/%s] failed, get `microService failed, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ if microService == nil {
+ log.Errorf(nil, "modify schema[%s/%s] failed, microService does not exist, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrServiceNotExists, "Service does not exist")
+ }
+
+ var pluginOps []registry.PluginOp
+ isExist := isExistSchemaID(microService, []*pb.Schema{schema})
+
+ if !ds.isSchemaEditable(microService) {
+ if len(microService.Schemas) != 0 && !isExist {
+ return scerr.NewError(scerr.ErrUndefinedSchemaID, "Non-existent schemaID can't be added in "+pb.ENV_PROD)
+ }
+
+ key := apt.GenerateServiceSchemaKey(domainProject, serviceID, schemaID)
+ respSchema, err := backend.Store().Schema().Search(ctx, registry.WithStrKey(key), registry.WithCountOnly())
+ if err != nil {
+ log.Errorf(err, "modify schema[%s/%s] failed, get schema summary failed, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+
+ if respSchema.Count != 0 {
+ if len(schema.Summary) == 0 {
+ log.Errorf(err, "%s mode, schema[%s/%s] already exists, can not be changed, operator: %s",
+ pb.ENV_PROD, serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrModifySchemaNotAllow,
+ "schema already exist, can not be changed in "+pb.ENV_PROD)
+ }
+
+ exist, err := isExistSchemaSummary(ctx, domainProject, serviceID, schemaID)
+ if err != nil {
+ log.Errorf(err, "check schema[%s/%s] summary existence failed, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ if exist {
+ log.Errorf(err, "%s mode, schema[%s/%s] already exist, can not be changed, operator: %s",
+ pb.ENV_PROD, serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrModifySchemaNotAllow, "schema already exist, can not be changed in "+pb.ENV_PROD)
+ }
+ }
+
+ if len(microService.Schemas) == 0 {
+ microService.Schemas = append(microService.Schemas, schemaID)
+ opt, err := updateService(domainProject, serviceID, microService)
+ if err != nil {
+ log.Errorf(err, "modify schema[%s/%s] failed, update microService.Schemas failed, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ pluginOps = append(pluginOps, opt)
+ }
+ } else {
+ if !isExist {
+ microService.Schemas = append(microService.Schemas, schemaID)
+ opt, err := updateService(domainProject, serviceID, microService)
+ if err != nil {
+ log.Errorf(err, "modify schema[%s/%s] failed, update microService.Schemas failed, operator: %s",
+ serviceID, schemaID, remoteIP)
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ pluginOps = append(pluginOps, opt)
+ }
+ }
+
+ opts := commitSchemaInfo(domainProject, serviceID, schema)
+ pluginOps = append(pluginOps, opts...)
+
+ resp, err := backend.Registry().TxnWithCmp(ctx, pluginOps,
+ []registry.CompareOp{registry.OpCmp(
+ registry.CmpVer(util.StringToBytesWithNoCopy(apt.GenerateServiceKey(domainProject, serviceID))),
+ registry.CmpNotEqual, 0)},
+ nil)
+ if err != nil {
+ return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+ if !resp.Succeeded {
+ return scerr.NewError(scerr.ErrServiceNotExists, "Service does not exist.")
+ }
+ return nil
+}
diff --git a/server/service/ms/etcd/etcd_test.go b/server/service/ms/etcd/etcd_test.go
index 0bf73e1..605ee98 100644
--- a/server/service/ms/etcd/etcd_test.go
+++ b/server/service/ms/etcd/etcd_test.go
@@ -16,10 +16,12 @@
package etcd_test
import (
+ "github.com/apache/servicecomb-service-center/pkg/log"
pb "github.com/apache/servicecomb-service-center/pkg/registry"
"github.com/apache/servicecomb-service-center/server/core/proto"
"github.com/apache/servicecomb-service-center/server/plugin/quota"
scerr "github.com/apache/servicecomb-service-center/server/scerror"
+ "github.com/apache/servicecomb-service-center/server/service"
"github.com/apache/servicecomb-service-center/server/service/ms"
"github.com/apache/servicecomb-service-center/server/service/ms/etcd"
"github.com/go-chassis/go-archaius"
@@ -35,6 +37,7 @@ var (
TooLongAppId = strings.Repeat("x", 162)
TooLongSchemaId = strings.Repeat("x", 162)
TooLongServiceName = strings.Repeat("x", 130)
+ TooLongExistence = strings.Repeat("x", 128+160+2)
TooLongAlias = strings.Repeat("x", 130)
TooLongFramework = strings.Repeat("x", 66)
)
@@ -44,7 +47,7 @@ func TestInit(t *testing.T) {
_ = archaius.Set("servicecomb.ms.name", "etcd")
}
-func TestEtcd_RegisterService(t *testing.T) {
+func TestService_Register(t *testing.T) {
t.Run("Register service after init & install, should pass", func(t *testing.T) {
ms.Install("etcd", func(opts ms.Options) (ms.DataSource, error) {
return etcd.NewDataSource(), nil
@@ -292,7 +295,7 @@ func TestEtcd_RegisterService(t *testing.T) {
})
}
-func TestEtcd_GetService(t *testing.T) {
+func TestService_Get(t *testing.T) {
// get service test
t.Run("query all services, should pass", func(t *testing.T) {
ms.Install("etcd", func(opts ms.Options) (ms.DataSource, error) {
@@ -359,3 +362,985 @@ func TestEtcd_GetService(t *testing.T) {
assert.Equal(t, resp.Response.GetCode(), scerr.ErrServiceNotExists)
})
}
+
+func TestService_Exist(t *testing.T) {
+ var (
+ serviceId1 string
+ serviceId2 string
+ )
+
+ ms.Install("etcd", func(opts ms.Options) (ms.DataSource, error) {
+ return etcd.NewDataSource(), nil
+ })
+
+ err := ms.Init(ms.Options{
+ Endpoint: "",
+ PluginImplName: ms.ImplName(archaius.GetString("servicecomb.ms.name", "etcd")),
+ })
+ assert.NoError(t, err)
+
+ t.Run("create service", func(t *testing.T) {
+ svc := &pb.MicroService{
+ Alias: "es_service_ms",
+ ServiceName: "exist_service_service_ms",
+ AppId: "exist_appId_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ },
+ Status: "UP",
+ }
+ resp, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: svc,
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, "", resp.ServiceId)
+ serviceId1 = resp.ServiceId
+
+ svc.ServiceId = ""
+ svc.Environment = pb.ENV_PROD
+ resp, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: svc,
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, "", resp.ServiceId)
+ serviceId2 = resp.ServiceId
+ })
+
+ t.Run("check exist when service does not exist", func(t *testing.T) {
+ log.Info("check by querying a not exist serviceName")
+ resp, err := ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId",
+ ServiceName: "notExistService_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrServiceNotExists, resp.Response.GetCode())
+
+ log.Info("check by querying a not exist env")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ Environment: pb.ENV_TEST,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "exist_service_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrServiceNotExists, resp.Response.GetCode())
+
+ log.Info("check by querying a not exist env with alias")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ Environment: pb.ENV_TEST,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrServiceNotExists, resp.Response.GetCode())
+
+ log.Info("check by querying with a mismatching version")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "exist_service_service_ms",
+ Version: "2.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrServiceNotExists, resp.Response.GetCode())
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "exist_service_service_ms",
+ Version: "0.0.0-1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrServiceVersionNotExists, resp.Response.GetCode())
+ })
+
+ t.Run("check exist when service exists", func(t *testing.T) {
+ log.Info("search with serviceName")
+ resp, err := ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "exist_service_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId1, resp.ServiceId)
+
+ log.Info("check with serviceName and env")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ Environment: pb.ENV_PROD,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "exist_service_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId2, resp.ServiceId)
+
+ log.Info("check with alias")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId1, resp.ServiceId)
+
+ log.Info("check with alias and env")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ Environment: pb.ENV_PROD,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "1.0.0",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId2, resp.ServiceId)
+
+ log.Info("check with latest versionRule")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "latest",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId1, resp.ServiceId)
+
+ log.Info("check with 1.0.0+ versionRule")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "1.0.0+",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId1, resp.ServiceId)
+
+ log.Info("check with range versionRule")
+ resp, err = ms.MicroService().ExistService(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeMicroservice,
+ AppId: "exist_appId_service_ms",
+ ServiceName: "es_service_ms",
+ Version: "0.9.1-1.0.1",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, serviceId1, resp.ServiceId)
+ })
+}
+
+func TestSchema_Create(t *testing.T) {
+ ms.Install("etcd", func(opts ms.Options) (ms.DataSource, error) {
+ return etcd.NewDataSource(), nil
+ })
+ err := ms.Init(ms.Options{
+ Endpoint: "",
+ PluginImplName: ms.ImplName(archaius.GetString("servicecomb.ms.name", "etcd")),
+ })
+ assert.NoError(t, err)
+
+ var (
+ serviceIdDev string
+ )
+
+ t.Run("create service, should pass", func(t *testing.T) {
+ svc := &pb.MicroService{
+ Alias: "create_schema_group_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ }
+ resp, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: svc,
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, "", resp.ServiceId)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ serviceIdDev = resp.ServiceId
+
+ resp, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_group_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, "", resp.ServiceId)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ })
+
+ t.Run("create schemas out of gauge", func(t *testing.T) {
+ log.Info("create schemas out of gauge")
+ size := quota.DefaultSchemaQuota + 1
+ schemaIds := make([]string, 0, size)
+ schemas := make([]*pb.Schema, 0, size)
+ for i := 0; i < size; i++ {
+ s := "ServiceCombTestTheLimitOfSchemasServiceMS" + strconv.Itoa(i)
+
+ schemaIds = append(schemaIds, s)
+ schemas = append(schemas, &pb.Schema{
+ SchemaId: s,
+ Schema: s,
+ Summary: s,
+ })
+ }
+
+ log.Info("batch modify schemas 1, should failed")
+ resp, err := ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrNotEnoughQuota, resp.Response.GetCode())
+
+ log.Info("batch modify schemas 2")
+ resp, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev,
+ Schemas: schemas[:quota.DefaultSchemaQuota],
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+
+ log.Info("should be failed in production env")
+ resp, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrNotEnoughQuota, resp.Response.GetCode())
+ })
+
+ t.Run("batch create schemas in dev env", func(t *testing.T) {
+ var (
+ serviceIdDev1 string
+ serviceIdDev2 string
+ )
+
+ log.Info("register service, should pass")
+ resp, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schemas_dev_service_ms",
+ ServiceName: "create_schemas_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ serviceIdDev1 = resp.ServiceId
+
+ resp, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schemas_dev_service_ms",
+ ServiceName: "create_schemas_service_service_ms",
+ Version: "1.0.1",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ },
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ serviceIdDev2 = resp.ServiceId
+
+ log.Info("create schemas with service schemaId set is empty")
+ schemas := []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ },
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ },
+ }
+ respCreateSchema, err := ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateSchema.Response.GetCode())
+
+ // todo: test GetAllSchemaInfo interface refers to schema_test line 342
+
+ log.Info("modify schemas")
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_change_service_ms",
+ Summary: "first0summary1change_service_ms",
+ },
+ }
+ respCreateSchema, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev1,
+ Schemas: schemas,
+ })
+
+ log.Info("add schemas")
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ Summary: "second0summary_service_ms",
+ },
+ }
+ respCreateSchema, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateSchema.Response.GetCode())
+
+ log.Info("query service by serviceID to obtain schema info")
+ respGetService, err := ms.MicroService().GetService(getContext(), &pb.GetServiceRequest{
+ ServiceId: serviceIdDev1,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respGetService.Response.GetCode())
+ assert.Equal(t, []string{"second_schemaId_service_ms"}, respGetService.Service.Schemas)
+
+ log.Info("add new schemaId not exist in service's schemaId list")
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ Summary: "second0summary_service_ms",
+ },
+ }
+ respCreateSchema, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdDev2,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateSchema.Response.GetCode())
+
+ respGetService, err = ms.MicroService().GetService(getContext(), &pb.GetServiceRequest{
+ ServiceId: serviceIdDev2,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respGetService.Response.GetCode())
+ assert.Equal(t, []string{"second_schemaId_service_ms"}, respGetService.Service.Schemas)
+ })
+
+ t.Run("batch create schemas in production env", func(t *testing.T) {
+ var (
+ serviceIdPro1 string
+ serviceIdPro2 string
+ )
+
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schemas_prod_service_ms",
+ ServiceName: "create_schemas_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro1 = respCreateService.ServiceId
+
+ respCreateService, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schemas_prod_service_ms",
+ ServiceName: "create_schemas_service_service_ms",
+ Version: "1.0.1",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ "second_schemaId_service_ms",
+ },
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro2 = respCreateService.ServiceId
+
+ log.Info("add schemas to service whose schemaId set is empty")
+ schemas := []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ },
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ },
+ }
+ respModifySchemas, err := ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+ respGetService, err := ms.MicroService().GetService(getContext(), &pb.GetServiceRequest{
+ ServiceId: serviceIdPro1,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respGetService.Response.GetCode())
+ assert.Equal(t, []string{"first_schemaId_service_ms"}, respGetService.Service.Schemas)
+
+ // todo: finish ut after implementing GetAllSchemaInfo, refer to schema_test.go line. 496
+
+ log.Info("modify schemas content already exists, will skip more exist schema")
+ respModifySchemas, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+
+ log.Info("add schemas, non-exist schemaId")
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ Summary: "second0summary_service_ms",
+ },
+ }
+ respModifySchemas, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrUndefinedSchemaID, respModifySchemas.Response.GetCode())
+
+ log.Info("add schema when summary is empty")
+ respModifySchema, err := ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro2,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("add schemas when summary in database is empty")
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ },
+ }
+ respModifySchemas, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro2,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+ respExist, err := ms.MicroService().ExistSchema(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeSchema,
+ ServiceId: serviceIdPro2,
+ SchemaId: "first_schemaId_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, "first0summary_service_ms", respExist.Summary)
+
+ respModifySchemas, err = ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro2,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+ })
+
+ t.Run("create a schema in dev env", func(t *testing.T) {
+ var (
+ serviceIdDev1 string
+ serviceIdDev2 string
+ )
+
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_dev_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdDev1 = respCreateService.ServiceId
+
+ respCreateService, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_dev_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.1",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ },
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdDev2 = respCreateService.ServiceId
+
+ log.Info("create a schema for service whose schemaID is empty")
+ respModifySchema, err := ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdDev1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("create schema for the service whose schemaId already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdDev2,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("create schema for the service whose schema summary is empty")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdDev1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_change_service_ms",
+ Summary: "first0summary1change_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("create schema for the service whose schema summary already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdDev1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("add schema")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdDev1,
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ })
+
+ t.Run("create a schema in production env", func(t *testing.T) {
+ var (
+ serviceIdPro1 string
+ serviceIdPro2 string
+ )
+
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_prod_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro1 = respCreateService.ServiceId
+
+ respCreateService, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_prod_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.1",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ "second_schemaId_service_ms",
+ },
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro2 = respCreateService.ServiceId
+
+ log.Info("create a schema for service whose schemaID is empty")
+ respModifySchema, err := ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schema summary is empty")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_change_service_ms",
+ Summary: "first0summary1change_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schema summary already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("add schema")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schemaId already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro2,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ })
+
+ t.Run("create a schema in empty env", func(t *testing.T) {
+ var (
+ serviceIdPro1 string
+ serviceIdPro2 string
+ )
+
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_empty_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro1 = respCreateService.ServiceId
+
+ respCreateService, err = ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "create_schema_empty_service_ms",
+ ServiceName: "create_schema_service_service_ms",
+ Version: "1.0.1",
+ Level: "FRONT",
+ Schemas: []string{
+ "first_schemaId_service_ms",
+ "second_schemaId_service_ms",
+ },
+ Status: pb.MS_UP,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro2 = respCreateService.ServiceId
+
+ log.Info("create a schema for service whose schemaID is empty")
+ respModifySchema, err := ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schema summary is empty")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_change_service_ms",
+ Summary: "first0summary1change_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schema summary already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ Summary: "first0summary_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("add schema")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: "second_schemaId_service_ms",
+ Schema: "second_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.NotEqual(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+
+ log.Info("modify schema for the service whose schemaId already exist")
+ respModifySchema, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro2,
+ SchemaId: "first_schemaId_service_ms",
+ Schema: "first_schema_service_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+ })
+
+ t.Run("add a schemaId in production env while schema editable is set", func(t *testing.T) {
+ var (
+ serviceIdPro1 string
+ )
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "add_a_schemaId_prod_schema_lock_ms",
+ ServiceName: "add_a_schemaId_prod_schema_lock_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro1 = respCreateService.ServiceId
+
+ log.Info("add a schema with new schemaId, should pass")
+ schemas := []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_ms",
+ Schema: "first_schema_ms",
+ Summary: "first0summary_ms",
+ },
+ }
+ respModifySchemas, err := ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+
+ respService, err := ms.MicroService().GetService(getContext(), &pb.GetServiceRequest{
+ ServiceId: serviceIdPro1,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respService.Response.GetCode())
+ assert.Equal(t, []string{"first_schemaId_ms"}, respService.Service.Schemas)
+
+ schemas = []*pb.Schema{
+ {
+ SchemaId: "second_schemaId_ms",
+ Schema: "second_schema_ms",
+ Summary: "second0summary_ms",
+ },
+ }
+ log.Info("schema edit not allowed, add a schema with new schemaId should fail")
+
+ localMicroServiceDs := &etcd.DataSource{SchemaEditable: false}
+ respModifySchemas, err = localMicroServiceDs.ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrUndefinedSchemaID, respModifySchemas.Response.GetCode())
+
+ log.Info("schema edit allowed, add a schema with new schemaId, should pass")
+ localMicroServiceDs = &etcd.DataSource{SchemaEditable: true}
+ respModifySchemas, err = localMicroServiceDs.ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+ })
+
+ t.Run("modify a schema in production env while schema editable is set", func(t *testing.T) {
+ var (
+ serviceIdPro1 string
+ )
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "modify_a_schema_prod_schema_lock_ms",
+ ServiceName: "modify_a_schema_prod_schema_lock_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_PROD,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceIdPro1 = respCreateService.ServiceId
+
+ log.Info("add schemas, should pass")
+ schemas := []*pb.Schema{
+ {
+ SchemaId: "first_schemaId_ms",
+ Schema: "first_schema_ms",
+ Summary: "first0summary_ms",
+ },
+ }
+ respModifySchemas, err := ms.MicroService().ModifySchemas(getContext(), &pb.ModifySchemasRequest{
+ ServiceId: serviceIdPro1,
+ Schemas: schemas,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchemas.Response.GetCode())
+
+ respService, err := ms.MicroService().GetService(getContext(), &pb.GetServiceRequest{
+ ServiceId: serviceIdPro1,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, []string{"first_schemaId_ms"}, respService.Service.Schemas)
+
+ log.Info("schema edit not allowed, modify schema should fail")
+ localMicroServiceDs := &etcd.DataSource{SchemaEditable: false}
+ respModifySchema, err := localMicroServiceDs.ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: schemas[0].SchemaId,
+ Summary: schemas[0].Summary,
+ Schema: schemas[0].SchemaId,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, scerr.ErrModifySchemaNotAllow, respModifySchema.Response.GetCode())
+
+ log.Info("schema edit allowed, add a schema with new schemaId, should pass")
+ localMicroServiceDs = &etcd.DataSource{SchemaEditable: true}
+ respModifySchema, err = localMicroServiceDs.ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceIdPro1,
+ SchemaId: schemas[0].SchemaId,
+ Summary: schemas[0].Summary,
+ Schema: schemas[0].SchemaId,
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respModifySchema.Response.GetCode())
+ })
+}
+
+func TestSchema_Exist(t *testing.T) {
+ ms.Install("etcd", func(opts ms.Options) (ms.DataSource, error) {
+ return etcd.NewDataSource(), nil
+ })
+ err := ms.Init(ms.Options{
+ Endpoint: "",
+ PluginImplName: ms.ImplName(archaius.GetString("servicecomb.ms.name", "etcd")),
+ })
+ assert.NoError(t, err)
+
+ var (
+ serviceId string
+ )
+
+ t.Run("register service and add schema", func(t *testing.T) {
+ log.Info("register service")
+ respCreateService, err := ms.MicroService().RegisterService(getContext(), &pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "query_schema_group_ms",
+ ServiceName: "query_schema_service_ms",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ Environment: pb.ENV_DEV,
+ },
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, respCreateService.Response.GetCode())
+ serviceId = respCreateService.ServiceId
+
+ log.Info("add schemas, should pass")
+ resp, err := ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceId,
+ SchemaId: "com.huawei.test.ms",
+ Schema: "query schema ms",
+ Summary: "summary_ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+
+ resp, err = ms.MicroService().ModifySchema(getContext(), &pb.ModifySchemaRequest{
+ ServiceId: serviceId,
+ SchemaId: "com.huawei.test.no.summary.ms",
+ Schema: "query schema ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ })
+
+ t.Run("check exists", func(t *testing.T) {
+ log.Info("check schema exist, should pass")
+ resp, err := ms.MicroService().ExistSchema(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeSchema,
+ ServiceId: serviceId,
+ SchemaId: "com.huawei.test.ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ assert.Equal(t, "summary_ms", resp.Summary)
+
+ resp, err = ms.MicroService().ExistSchema(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeSchema,
+ ServiceId: serviceId,
+ SchemaId: "com.huawei.test.ms",
+ AppId: "()",
+ ServiceName: "",
+ Version: "()",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+
+ resp, err = ms.MicroService().ExistSchema(getContext(), &pb.GetExistenceRequest{
+ Type: service.ExistTypeSchema,
+ ServiceId: serviceId,
+ SchemaId: "com.huawei.test.no.summary.ms",
+ })
+ assert.NoError(t, err)
+ assert.Equal(t, proto.Response_SUCCESS, resp.Response.GetCode())
+ assert.Equal(t, "com.huawei.test.no.summary.ms", resp.SchemaId)
+ assert.Equal(t, "", resp.Summary)
+ })
+}
diff --git a/server/service/ms/etcd/util.go b/server/service/ms/etcd/util.go
index a0d6011..be422e4 100644
--- a/server/service/ms/etcd/util.go
+++ b/server/service/ms/etcd/util.go
@@ -31,10 +31,12 @@ import (
scerr "github.com/apache/servicecomb-service-center/server/scerror"
serviceUtil "github.com/apache/servicecomb-service-center/server/service/util"
"strconv"
+ "strings"
"time"
)
-func GetSingleService(ctx context.Context, domainProject string, serviceID string) (*pb.MicroService, error) {
+// service
+func getSingleService(ctx context.Context, domainProject string, serviceID string) (*pb.MicroService, error) {
key := apt.GenerateServiceKey(domainProject, serviceID)
opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key))
serviceResp, err := backend.Store().Service().Search(ctx, opts...)
@@ -47,15 +49,15 @@ func GetSingleService(ctx context.Context, domainProject string, serviceID strin
return serviceResp.Kvs[0].Value.(*pb.MicroService), nil
}
-func GetAllServiceUtil(ctx context.Context, domainProject string) ([]*pb.MicroService, error) {
- services, err := GetServicesByDomainProject(ctx, domainProject)
+func getAllServiceUtil(ctx context.Context, domainProject string) ([]*pb.MicroService, error) {
+ services, err := getServicesByDomainProject(ctx, domainProject)
if err != nil {
return nil, err
}
return services, nil
}
-func GetServicesByDomainProject(ctx context.Context, domainProject string) ([]*pb.MicroService, error) {
+func getServicesByDomainProject(ctx context.Context, domainProject string) ([]*pb.MicroService, error) {
kvs, err := getServicesRawData(ctx, domainProject)
if err != nil {
return nil, err
@@ -79,7 +81,7 @@ func getServicesRawData(ctx context.Context, domainProject string) ([]*discovery
return resp.Kvs, err
}
-func CapRegisterData(ctx context.Context, request *pb.CreateServiceRequest) (
+func capRegisterData(ctx context.Context, request *pb.CreateServiceRequest) (
[]registry.PluginOp, []registry.CompareOp, []registry.PluginOp, error) {
remoteIP := util.GetIPFromContext(ctx)
serviceBody := request.Service
@@ -140,7 +142,7 @@ func CapRegisterData(ctx context.Context, request *pb.CreateServiceRequest) (
return opts, uniqueCmpOpts, failOpts, nil
}
-func NewRegisterServiceResp(ctx context.Context, reqService *pb.MicroService, resp *registry.PluginResponse,
+func newRegisterServiceResp(ctx context.Context, reqService *pb.MicroService, resp *registry.PluginResponse,
err error) (*pb.CreateServiceResponse, error) {
remoteIP := util.GetIPFromContext(ctx)
serviceFlag := util.StringJoin([]string{
@@ -192,3 +194,295 @@ func NewRegisterServiceResp(ctx context.Context, reqService *pb.MicroService, re
ServiceId: reqService.ServiceId,
}, nil
}
+
+func serviceExist(ctx context.Context, domainProject string, serviceID string) bool {
+ opts := append(serviceUtil.FromContext(ctx),
+ registry.WithStrKey(apt.GenerateServiceKey(domainProject, serviceID)),
+ registry.WithCountOnly())
+ resp, err := backend.Store().Service().Search(ctx, opts...)
+ if err != nil || resp.Count == 0 {
+ return false
+ }
+ return true
+}
+
+func findServiceIds(ctx context.Context, versionRule string, key *pb.MicroServiceKey) ([]string, bool, error) {
+ // 版本规则
+ match := serviceUtil.ParseVersionRule(versionRule)
+ if match == nil {
+ copyData := *key
+ copyData.Version = versionRule
+ serviceID, err := getServiceID(ctx, ©Data)
+ if err != nil {
+ return nil, false, err
+ }
+ if len(serviceID) > 0 {
+ return []string{serviceID}, true, nil
+ }
+ return nil, false, nil
+ }
+
+ searchAlias := false
+ alsoFindAlias := len(key.Alias) > 0
+
+FindRule:
+ resp, err := getServiceAllVersions(ctx, key, searchAlias)
+ if err != nil {
+ return nil, false, err
+ }
+ if len(resp.Kvs) == 0 {
+ if !alsoFindAlias {
+ return nil, false, nil
+ }
+ searchAlias = true
+ alsoFindAlias = false
+ goto FindRule
+ }
+ return match(resp.Kvs), true, nil
+}
+
+func getServiceID(ctx context.Context, key *pb.MicroServiceKey) (serviceID string, err error) {
+ serviceID, err = searchServiceID(ctx, key)
+ if err != nil {
+ return
+ }
+ if len(serviceID) == 0 {
+ // 别名查询
+ log.Debugf("could not search microservice[%s/%s/%s/%s] id by 'serviceName', now try 'alias'",
+ key.Environment, key.AppId, key.ServiceName, key.Version)
+ return searchServiceIDFromAlias(ctx, key)
+ }
+ return
+}
+
+func searchServiceID(ctx context.Context, key *pb.MicroServiceKey) (string, error) {
+ opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(apt.GenerateServiceIndexKey(key)))
+ resp, err := backend.Store().ServiceIndex().Search(ctx, opts...)
+ if err != nil {
+ return "", err
+ }
+ if len(resp.Kvs) == 0 {
+ return "", nil
+ }
+ return resp.Kvs[0].Value.(string), nil
+}
+
+func searchServiceIDFromAlias(ctx context.Context, key *pb.MicroServiceKey) (string, error) {
+ opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(apt.GenerateServiceAliasKey(key)))
+ resp, err := backend.Store().ServiceAlias().Search(ctx, opts...)
+ if err != nil {
+ return "", err
+ }
+ if len(resp.Kvs) == 0 {
+ return "", nil
+ }
+ return resp.Kvs[0].Value.(string), nil
+}
+
+func getServiceAllVersions(ctx context.Context, key *pb.MicroServiceKey, alias bool) (*discovery.Response, error) {
+ copyData := *key
+ copyData.Version = ""
+ var (
+ prefix string
+ indexer discovery.Indexer
+ )
+ if alias {
+ prefix = apt.GenerateServiceAliasKey(©Data)
+ indexer = backend.Store().ServiceAlias()
+ } else {
+ prefix = apt.GenerateServiceIndexKey(©Data)
+ indexer = backend.Store().ServiceIndex()
+ }
+ opts := append(serviceUtil.FromContext(ctx),
+ registry.WithStrKey(prefix),
+ registry.WithPrefix(),
+ registry.WithDescendOrder())
+ resp, err := indexer.Search(ctx, opts...)
+ return resp, err
+}
+
+func updateService(domainProject string, serviceID string, service *pb.MicroService) (opt registry.PluginOp,
+ err error) {
+ opt = registry.PluginOp{}
+ key := apt.GenerateServiceKey(domainProject, serviceID)
+ data, err := json.Marshal(service)
+ if err != nil {
+ log.Errorf(err, "marshal service file failed")
+ return
+ }
+ opt = registry.OpPut(registry.WithStrKey(key), registry.WithValue(data))
+ return
+}
+
+// schema
+func getSchemaSummary(ctx context.Context, domainProject string, serviceID string, schemaID string) (string, error) {
+ key := apt.GenerateServiceSchemaSummaryKey(domainProject, serviceID, schemaID)
+ resp, err := backend.Store().SchemaSummary().Search(ctx,
+ registry.WithStrKey(key),
+ )
+ if err != nil {
+ log.Errorf(err, "get schema[%s/%s] summary failed", serviceID, schemaID)
+ return "", err
+ }
+ if len(resp.Kvs) == 0 {
+ return "", nil
+ }
+ return resp.Kvs[0].Value.(string), nil
+}
+
+func getService(ctx context.Context, domainProject string, serviceID string) (*pb.MicroService, error) {
+ key := apt.GenerateServiceKey(domainProject, serviceID)
+ opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key))
+ serviceResp, err := backend.Store().Service().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+ if len(serviceResp.Kvs) == 0 {
+ return nil, nil
+ }
+ return serviceResp.Kvs[0].Value.(*pb.MicroService), nil
+}
+
+func getSchemasFromDatabase(ctx context.Context, domainProject string, serviceID string) ([]*pb.Schema, error) {
+ key := apt.GenerateServiceSchemaKey(domainProject, serviceID, "")
+ resp, err := backend.Store().Schema().Search(ctx,
+ registry.WithPrefix(),
+ registry.WithStrKey(key))
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s schema failed", serviceID)
+ return nil, err
+ }
+ schemas := make([]*pb.Schema, 0, len(resp.Kvs))
+ for _, kv := range resp.Kvs {
+ key := util.BytesToStringWithNoCopy(kv.Key)
+ tmp := strings.Split(key, "/")
+ schemaID := tmp[len(tmp)-1]
+ schema := util.BytesToStringWithNoCopy(kv.Value.([]byte))
+ schemaStruct := &pb.Schema{
+ SchemaId: schemaID,
+ Schema: schema,
+ }
+ schemas = append(schemas, schemaStruct)
+ }
+ return schemas, nil
+}
+
+func schemasAnalysis(schemas []*pb.Schema, schemasFromDb []*pb.Schema, schemaIDsInService []string) (
+ []*pb.Schema, []*pb.Schema, []*pb.Schema, []string) {
+ needUpdateSchemas := make([]*pb.Schema, 0, len(schemas))
+ needAddSchemas := make([]*pb.Schema, 0, len(schemas))
+ needDeleteSchemas := make([]*pb.Schema, 0, len(schemasFromDb))
+ nonExistSchemaIds := make([]string, 0, len(schemas))
+
+ duplicate := make(map[string]struct{})
+ for _, schema := range schemas {
+ if _, ok := duplicate[schema.SchemaId]; ok {
+ continue
+ }
+ duplicate[schema.SchemaId] = struct{}{}
+
+ exist := false
+ for _, schemaFromDb := range schemasFromDb {
+ if schema.SchemaId == schemaFromDb.SchemaId {
+ needUpdateSchemas = append(needUpdateSchemas, schema)
+ exist = true
+ break
+ }
+ }
+ if !exist {
+ needAddSchemas = append(needAddSchemas, schema)
+ }
+
+ exist = false
+ for _, schemaID := range schemaIDsInService {
+ if schema.SchemaId == schemaID {
+ exist = true
+ }
+ }
+ if !exist {
+ nonExistSchemaIds = append(nonExistSchemaIds, schema.SchemaId)
+ }
+ }
+
+ for _, schemaFromDb := range schemasFromDb {
+ exist := false
+ for _, schema := range schemas {
+ if schema.SchemaId == schemaFromDb.SchemaId {
+ exist = true
+ break
+ }
+ }
+ if !exist {
+ needDeleteSchemas = append(needDeleteSchemas, schemaFromDb)
+ }
+ }
+
+ return needUpdateSchemas, needAddSchemas, needDeleteSchemas, nonExistSchemaIds
+}
+
+func checkSchemaInfoExist(ctx context.Context, key string) (bool, error) {
+ opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key), registry.WithCountOnly())
+ resp, errDo := backend.Store().Schema().Search(ctx, opts...)
+ if errDo != nil {
+ return false, errDo
+ }
+ if resp.Count == 0 {
+ return false, nil
+ }
+ return true, nil
+}
+
+func isExistSchemaSummary(ctx context.Context, domainProject, serviceID, schemaID string) (bool, error) {
+ key := apt.GenerateServiceSchemaSummaryKey(domainProject, serviceID, schemaID)
+ resp, err := backend.Store().SchemaSummary().Search(ctx, registry.WithStrKey(key), registry.WithCountOnly())
+ if err != nil {
+ return true, err
+ }
+ if resp.Count == 0 {
+ return false, nil
+ }
+ return true, nil
+}
+
+func schemaWithDatabaseOpera(invoke registry.Operation, domainProject string, serviceID string, schema *pb.Schema) []registry.PluginOp {
+ pluginOps := make([]registry.PluginOp, 0)
+ key := apt.GenerateServiceSchemaKey(domainProject, serviceID, schema.SchemaId)
+ opt := invoke(registry.WithStrKey(key), registry.WithStrValue(schema.Schema))
+ pluginOps = append(pluginOps, opt)
+ keySummary := apt.GenerateServiceSchemaSummaryKey(domainProject, serviceID, schema.SchemaId)
+ opt = invoke(registry.WithStrKey(keySummary), registry.WithStrValue(schema.Summary))
+ pluginOps = append(pluginOps, opt)
+ return pluginOps
+}
+
+func isExistSchemaID(service *pb.MicroService, schemas []*pb.Schema) bool {
+ serviceSchemaIds := service.Schemas
+ for _, schema := range schemas {
+ if !containsValueInSlice(serviceSchemaIds, schema.SchemaId) {
+ log.Errorf(nil, "schema[%s/%s] does not exist schemaID", service.ServiceId, schema.SchemaId)
+ return false
+ }
+ }
+ return true
+}
+
+func containsValueInSlice(in []string, value string) bool {
+ if in == nil || len(value) == 0 {
+ return false
+ }
+ for _, i := range in {
+ if i == value {
+ return true
+ }
+ }
+ return false
+}
+
+func commitSchemaInfo(domainProject string, serviceID string, schema *pb.Schema) []registry.PluginOp {
+ if len(schema.Summary) != 0 {
+ return schemaWithDatabaseOpera(registry.OpPut, domainProject, serviceID, schema)
+ }
+ key := apt.GenerateServiceSchemaKey(domainProject, serviceID, schema.SchemaId)
+ opt := registry.OpPut(registry.WithStrKey(key), registry.WithStrValue(schema.Schema))
+ return []registry.PluginOp{opt}
+}