You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by GitBox <gi...@apache.org> on 2018/12/03 01:04:53 UTC
[GitHub] little-cui closed pull request #502: SCB-1053 Batch microservices
instances discovery API
little-cui closed pull request #502: SCB-1053 Batch microservices instances discovery API
URL: https://github.com/apache/servicecomb-service-center/pull/502
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/integration/instances_test.go b/integration/instances_test.go
index eb259bda..86976a5e 100644
--- a/integration/instances_test.go
+++ b/integration/instances_test.go
@@ -348,6 +348,52 @@ var _ = Describe("MicroService Api Test", func() {
rev = resp.Header.Get("X-Resource-Revision")
Expect(rev).NotTo(BeEmpty())
})
+
+ It("Batch Find Micro-service Instance", func() {
+ notExistsService := map[string]interface{}{
+ "service": map[string]interface{}{
+ "appId": serviceAppId,
+ "serviceName": "notexisted",
+ "version": serviceVersion,
+ },
+ }
+ provider := map[string]interface{}{
+ "service": map[string]interface{}{
+ "appId": serviceAppId,
+ "serviceName": serviceName,
+ "version": serviceVersion,
+ },
+ }
+ findRequest := map[string]interface{}{
+ "services": []map[string]interface{}{
+ provider,
+ notExistsService,
+ },
+ }
+ body, _ := json.Marshal(findRequest)
+ bodyBuf := bytes.NewReader(body)
+ req, _ := http.NewRequest(POST, SCURL+FINDINSTANCE, bodyBuf)
+ req.Header.Set("X-Domain-Name", "default")
+ req.Header.Set("X-ConsumerId", serviceId)
+ resp, _ := scclient.Do(req)
+ respbody, _ := ioutil.ReadAll(resp.Body)
+ Expect(resp.StatusCode).To(Equal(http.StatusOK))
+ servicesStruct := map[string][]map[string]interface{}{}
+ json.Unmarshal(respbody, &servicesStruct)
+ failed := false
+ for _, services := range servicesStruct["failed"] {
+ a := services["indexes"].([]interface{})[0] == 1.0
+ b := services["error"].(map[string]interface{})["errorCode"] == "400012"
+ if a && b {
+ failed = true
+ break
+ }
+ }
+ Expect(failed).To(Equal(true))
+ Expect(servicesStruct["updated"][0]["index"]).To(Equal(0.0))
+ Expect(len(servicesStruct["updated"][0]["instances"].([]interface{}))).
+ ToNot(Equal(0))
+ })
})
By("Update Micro-Service Instance Information API's", func() {
diff --git a/server/core/proto/batch_find.go b/server/core/proto/batch_find.go
new file mode 100644
index 00000000..68e3883c
--- /dev/null
+++ b/server/core/proto/batch_find.go
@@ -0,0 +1,48 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package proto
+
+import (
+ scerr "github.com/apache/servicecomb-service-center/server/error"
+)
+
+type FindService struct {
+ Service *MicroServiceKey `protobuf:"bytes,1,opt,name=service" json:"service"`
+ Rev string `protobuf:"bytes,2,opt,name=rev" json:"rev,omitempty"`
+}
+
+type FindResult struct {
+ Index int64 `protobuf:"varint,1,opt,name=index" json:"index"`
+ Rev string `protobuf:"bytes,2,opt,name=rev" json:"rev"`
+ Instances []*MicroServiceInstance `protobuf:"bytes,3,rep,name=instances" json:"instances"`
+}
+
+type FindFailedResult struct {
+ Indexes []int64 `protobuf:"varint,1,rep,packed,name=indexes" json:"indexes"`
+ Error *scerr.Error `protobuf:"bytes,2,opt,name=error" json:"error"`
+}
+
+type BatchFindInstancesRequest struct {
+ ConsumerServiceId string `protobuf:"bytes,1,opt,name=consumerServiceId" json:"consumerServiceId,omitempty"`
+ Services []*FindService `protobuf:"bytes,2,rep,name=services" json:"services"`
+}
+
+type BatchFindInstancesResponse struct {
+ Response *Response `protobuf:"bytes,1,opt,name=response" json:"response,omitempty"`
+ Failed []*FindFailedResult `protobuf:"bytes,2,rep,name=failed" json:"failed,omitempty"`
+ NotModified []int64 `protobuf:"varint,3,rep,packed,name=notModified" json:"notModified,omitempty"`
+ Updated []*FindResult `protobuf:"bytes,4,rep,name=updated" json:"updated,omitempty"`
+}
diff --git a/server/core/proto/services.go b/server/core/proto/services.go
index 80b158af..a2e65098 100644
--- a/server/core/proto/services.go
+++ b/server/core/proto/services.go
@@ -24,7 +24,10 @@ import (
type ServiceInstanceCtrlServerEx interface {
ServiceInstanceCtrlServer
+ BatchFind(ctx context.Context, in *BatchFindInstancesRequest) (*BatchFindInstancesResponse, error)
+
WebSocketWatch(ctx context.Context, in *WatchInstanceRequest, conn *websocket.Conn)
WebSocketListAndWatch(ctx context.Context, in *WatchInstanceRequest, conn *websocket.Conn)
+
ClusterHealth(ctx context.Context) (*GetInstancesResponse, error)
}
diff --git a/server/core/proto/types.go b/server/core/proto/types.go
index d3624bc3..36de9a4e 100644
--- a/server/core/proto/types.go
+++ b/server/core/proto/types.go
@@ -47,8 +47,8 @@ type ServerConfig struct {
CompactIndexDelta int64 `json:"-"`
CompactInterval string `json:"-"`
- EnablePProf bool `json:"-"`
- EnableCache bool `json:"-"`
+ EnablePProf bool `json:"enablePProf"`
+ EnableCache bool `json:"enableCache"`
LogRotateSize int64 `json:"-"`
LogBackupCount int64 `json:"-"`
@@ -60,7 +60,7 @@ type ServerConfig struct {
PluginsDir string `json:"-"`
Plugins util.JSONObject `json:"plugins"`
- SelfRegister bool `json:"-"`
+ SelfRegister bool `json:"selfRegister"`
}
type ServerInformation struct {
diff --git a/server/core/swagger/v4.yaml b/server/core/swagger/v4.yaml
index 75ff962d..7d3aaef8 100644
--- a/server/core/swagger/v4.yaml
+++ b/server/core/swagger/v4.yaml
@@ -1370,13 +1370,16 @@ paths:
in: header
type: string
default: default
- - name: project
- in: path
- required: true
+ - name: x-resource-revsion
+ in: header
type: string
+ description: 客户端缓存的版本号,由上一次请求该API返回Header中获得;如请求版本号不为空且与服务端不匹配则服务端返回其最新的实例集合和版本号;如匹配则服务端返回304状态且Body为空。
- name: X-ConsumerId
in: header
description: 微服务消费者的微服务唯一标识。
+ type: string
+ - name: project
+ in: path
required: true
type: string
- name: appId
@@ -1407,6 +1410,10 @@ paths:
responses:
200:
description: 查询成功
+ headers:
+ "X-Resource-Revision":
+ type: "string"
+ description: 返回集合的版本号,当集合内容发生变化,版本号随之变化
schema:
$ref: '#/definitions/GetInstancesResponse'
400:
@@ -1417,6 +1424,44 @@ paths:
description: 内部错误
schema:
type: string
+ post:
+ description: |
+ 批量微服务实例发现接口
+ operationId: batchFind
+ parameters:
+ - name: x-domain-name
+ in: header
+ type: string
+ default: default
+ - name: X-ConsumerId
+ in: header
+ description: 微服务消费者的微服务唯一标识。
+ type: string
+ - name: project
+ in: path
+ required: true
+ type: string
+ - name: services
+ in: body
+ description: 查询微服务的请求结构体
+ required: true
+ schema:
+ $ref: '#/definitions/BatchFindRequest'
+ tags:
+ - instances
+ responses:
+ 200:
+ description: 查询成功
+ schema:
+ $ref: '#/definitions/BatchFindResponse'
+ 400:
+ description: 错误的请求
+ schema:
+ type: string
+ 500:
+ description: 内部错误
+ schema:
+ type: string
/v4/{project}/registry/microservices/{serviceId}/watcher:
get:
description: |
@@ -1692,6 +1737,8 @@ definitions:
type: string
arch:
type: string
+ config:
+ $ref: '#/definitions/Config'
Properties:
type: object
description: 扩展属性
@@ -1907,6 +1954,60 @@ definitions:
modTimestamp:
type: string
description: 更新时间
+ FindService:
+ type: object
+ properties:
+ service:
+ $ref: '#definitions/DependencyKey'
+ rev:
+ type: string
+ description: 客户端缓存的版本号。
+ BatchFindRequest:
+ type: object
+ properties:
+ services:
+ type: array
+ items:
+ $ref: '#/definitions/FindService'
+ FindResult:
+ type: object
+ properties:
+ index:
+ type: integer
+ description: 与请求数组对应的索引。
+ rev:
+ type: string
+ description: 服务端返回集合版本,如跟客户端缓存版本号一致,则instances为空。
+ instances:
+ type: array
+ items:
+ $ref: '#/definitions/MicroServiceInstance'
+ FindFailedResult:
+ type: object
+ properties:
+ indexes:
+ type: array
+ items:
+ type: integer
+ description: 与请求数组对应的索引集合。
+ error:
+ $ref: '#/definitions/Error'
+ BatchFindResponse:
+ type: object
+ properties:
+ failed:
+ type: array
+ items:
+ $ref: '#/definitions/FindFailedResult'
+ notModified:
+ type: array
+ items:
+ type: integer
+ description: 与请求数组对应的索引集合。
+ updated:
+ type: array
+ items:
+ $ref: '#/definitions/FindResult'
CreateDependenciesRequest:
type: object
properties:
@@ -2413,16 +2514,12 @@ definitions:
type: string
sslCiphers:
type: string
- autoSyncInterval:
- type: string
- compactIndexDelta:
- type: integer
- compactInterval:
- type: string
- logRotateSize:
- type: integer
- logBackupCount:
- type: integer
+ enablePProf:
+ type: boolean
+ enableCache:
+ type: boolean
+ selfRegister:
+ type: boolean
DumpResponse:
type: object
properties:
@@ -2446,3 +2543,12 @@ definitions:
properties:
clusters:
$ref: '#/definitions/Clusters'
+ Error:
+ type: object
+ properties:
+ errorCode:
+ type: string
+ errorMessage:
+ type: string
+ detail:
+ type: string
diff --git a/server/plugin/pkg/discovery/aggregate/repo.go b/server/plugin/pkg/discovery/aggregate/repo.go
index cb6d7e36..4393bd56 100644
--- a/server/plugin/pkg/discovery/aggregate/repo.go
+++ b/server/plugin/pkg/discovery/aggregate/repo.go
@@ -18,6 +18,7 @@ package aggregate
import (
mgr "github.com/apache/servicecomb-service-center/server/plugin"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
)
func init() {
@@ -32,5 +33,12 @@ func (r *AggregateRepository) New(t discovery.Type, cfg *discovery.Config) disco
}
func NewRepository() mgr.PluginInstance {
+ InitConfigs()
return &AggregateRepository{}
}
+
+func InitConfigs() {
+ mgr.DISCOVERY.ActiveConfigs().
+ Set("config", registry.Configuration()).
+ Set("aggregateMode", repos)
+}
diff --git a/server/plugin/pkg/discovery/etcd/repo.go b/server/plugin/pkg/discovery/etcd/repo.go
index 3fb614d5..8239fa1f 100644
--- a/server/plugin/pkg/discovery/etcd/repo.go
+++ b/server/plugin/pkg/discovery/etcd/repo.go
@@ -18,6 +18,7 @@ package etcd
import (
mgr "github.com/apache/servicecomb-service-center/server/plugin"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
)
func init() {
@@ -33,5 +34,11 @@ func (r *EtcdRepository) New(t discovery.Type, cfg *discovery.Config) discovery.
}
func NewRepository() mgr.PluginInstance {
+ InitConfigs()
return &EtcdRepository{}
}
+
+func InitConfigs() {
+ mgr.DISCOVERY.ActiveConfigs().
+ Set("config", registry.Configuration())
+}
diff --git a/server/plugin/pkg/discovery/servicecenter/repo.go b/server/plugin/pkg/discovery/servicecenter/repo.go
index 93512bf9..a15a3ea6 100644
--- a/server/plugin/pkg/discovery/servicecenter/repo.go
+++ b/server/plugin/pkg/discovery/servicecenter/repo.go
@@ -18,6 +18,7 @@ package servicecenter
import (
mgr "github.com/apache/servicecomb-service-center/server/plugin"
"github.com/apache/servicecomb-service-center/server/plugin/pkg/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/pkg/registry"
)
func init() {
@@ -32,5 +33,11 @@ func (r *ServiceCenterRepository) New(t discovery.Type, cfg *discovery.Config) d
}
func NewRepository() mgr.PluginInstance {
+ InitConfigs()
return &ServiceCenterRepository{}
}
+
+func InitConfigs() {
+ mgr.DISCOVERY.ActiveConfigs().
+ Set("config", registry.Configuration())
+}
diff --git a/server/plugin/pkg/registry/config.go b/server/plugin/pkg/registry/config.go
index 6113cce7..ef36bad8 100644
--- a/server/plugin/pkg/registry/config.go
+++ b/server/plugin/pkg/registry/config.go
@@ -30,15 +30,14 @@ var (
)
type Config struct {
- SslEnabled bool
- EmbedMode string
- ManagerAddress string
- ClusterName string
- ClusterAddresses string // the raw string of cluster configuration
- Clusters Clusters // parsed from ClusterAddresses
- DialTimeout time.Duration
- RequestTimeOut time.Duration
- AutoSyncInterval time.Duration
+ SslEnabled bool `json:"-"`
+ ManagerAddress string `json:"manageAddress,omitempty"`
+ ClusterName string `json:"manageName,omitempty"`
+ ClusterAddresses string `json:"manageClusters,omitempty"` // the raw string of cluster configuration
+ Clusters Clusters `json:"-"` // parsed from ClusterAddresses
+ DialTimeout time.Duration `json:"connectTimeout"`
+ RequestTimeOut time.Duration `json:"registryTimeout"`
+ AutoSyncInterval time.Duration `json:"autoSyncInterval"`
}
func (c *Config) InitClusters() {
diff --git a/server/rest/controller/v4/instance_controller.go b/server/rest/controller/v4/instance_controller.go
index d6e5fdbc..36efd165 100644
--- a/server/rest/controller/v4/instance_controller.go
+++ b/server/rest/controller/v4/instance_controller.go
@@ -38,6 +38,7 @@ type MicroServiceInstanceService struct {
func (this *MicroServiceInstanceService) URLPatterns() []rest.Route {
return []rest.Route{
{rest.HTTP_METHOD_GET, "/v4/:project/registry/instances", this.FindInstances},
+ {rest.HTTP_METHOD_POST, "/v4/:project/registry/instances", this.BatchFindInstances},
{rest.HTTP_METHOD_GET, "/v4/:project/registry/microservices/:serviceId/instances", this.GetInstances},
{rest.HTTP_METHOD_GET, "/v4/:project/registry/microservices/:serviceId/instances/:instanceId", this.GetOneInstance},
{rest.HTTP_METHOD_POST, "/v4/:project/registry/microservices/:serviceId/instances", this.RegisterInstance},
@@ -154,6 +155,29 @@ func (this *MicroServiceInstanceService) FindInstances(w http.ResponseWriter, r
controller.WriteResponse(w, respInternal, resp)
}
+func (this *MicroServiceInstanceService) BatchFindInstances(w http.ResponseWriter, r *http.Request) {
+ message, err := ioutil.ReadAll(r.Body)
+ if err != nil {
+ log.Error("read body failed", err)
+ controller.WriteError(w, scerr.ErrInvalidParams, err.Error())
+ return
+ }
+
+ request := &pb.BatchFindInstancesRequest{}
+ err = json.Unmarshal(message, request)
+ if err != nil {
+ log.Errorf(err, "Invalid json: %s", util.BytesToStringWithNoCopy(message))
+ controller.WriteError(w, scerr.ErrInvalidParams, "Unmarshal error")
+ return
+ }
+ request.ConsumerServiceId = r.Header.Get("X-ConsumerId")
+ ctx := util.SetTargetDomainProject(r.Context(), r.Header.Get("X-Domain-Name"), r.URL.Query().Get(":project"))
+ resp, _ := core.InstanceAPI.BatchFind(ctx, request)
+ respInternal := resp.Response
+ resp.Response = nil
+ controller.WriteResponse(w, respInternal, resp)
+}
+
func (this *MicroServiceInstanceService) GetOneInstance(w http.ResponseWriter, r *http.Request) {
var ids []string
query := r.URL.Query()
diff --git a/server/service/instance.go b/server/service/instance.go
index 667d6757..53eca314 100644
--- a/server/service/instance.go
+++ b/server/service/instance.go
@@ -616,6 +616,46 @@ func (s *InstanceService) Find(ctx context.Context, in *pb.FindInstancesRequest)
}, nil
}
+func (s *InstanceService) BatchFind(ctx context.Context, in *pb.BatchFindInstancesRequest) (*pb.BatchFindInstancesResponse, error) {
+ err := Validate(in)
+ if err != nil {
+ log.Errorf(err, "batch find instance failed: invalid parameters")
+ return &pb.BatchFindInstancesResponse{
+ Response: pb.CreateResponse(scerr.ErrInvalidParams, err.Error()),
+ }, nil
+ }
+
+ response := &pb.BatchFindInstancesResponse{
+ Response: pb.CreateResponse(pb.Response_SUCCESS, "Batch query service instances successfully."),
+ }
+ failedResult := make(map[int32]*pb.FindFailedResult)
+ for index, key := range in.Services {
+ cloneCtx := util.SetContext(ctx, serviceUtil.CTX_REQUEST_REVISION, key.Rev)
+ resp, err := s.Find(cloneCtx, &pb.FindInstancesRequest{
+ ConsumerServiceId: in.ConsumerServiceId,
+ AppId: key.Service.AppId,
+ ServiceName: key.Service.ServiceName,
+ VersionRule: key.Service.Version,
+ Environment: key.Service.Environment,
+ })
+ if err != nil {
+ return &pb.BatchFindInstancesResponse{
+ Response: resp.Response,
+ }, err
+ }
+ failed, ok := failedResult[resp.GetResponse().GetCode()]
+ serviceUtil.AppendFindResponse(cloneCtx, int64(index), resp,
+ &response.Updated, &response.NotModified, &failed)
+ if !ok && failed != nil {
+ failedResult[resp.GetResponse().GetCode()] = failed
+ }
+ }
+ for _, result := range failedResult {
+ response.Failed = append(response.Failed, result)
+ }
+ return response, nil
+}
+
func (s *InstanceService) reshapeProviderKey(ctx context.Context, provider *pb.MicroServiceKey, providerId string) (*pb.MicroServiceKey, error) {
//维护version的规则,service name 可能是别名，所以重新获取
providerService, err := serviceUtil.GetService(ctx, provider.Tenant, providerId)
diff --git a/server/service/instance_test.go b/server/service/instance_test.go
index d40dd63d..3b67e018 100644
--- a/server/service/instance_test.go
+++ b/server/service/instance_test.go
@@ -1035,7 +1035,7 @@ var _ = Describe("'Instance' service", func() {
respFind, err = instanceResource.Find(getContext(), &pb.FindInstancesRequest{
ConsumerServiceId: serviceId1,
AppId: "query_instance",
- ServiceName: "",
+ ServiceName: " ",
VersionRule: "1.0.0",
})
Expect(err).To(BeNil())
@@ -1135,7 +1135,248 @@ var _ = Describe("'Instance' service", func() {
VersionRule: "2.0.0",
})
Expect(err).To(BeNil())
- Expect(respFind.Response.Code).ToNot(Equal(pb.Response_SUCCESS))
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrServiceNotExists))
+ })
+ })
+
+ Context("when batch query invalid parameters", func() {
+ It("should be failed", func() {
+ By("invalid services")
+ respFind, err := instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: nil,
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{},
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{{}},
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+
+ By("invalid appId")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: TOO_LONG_APPID,
+ ServiceName: "query_instance_service",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "",
+ ServiceName: "query_instance_service",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: " ",
+ ServiceName: "query_instance_service",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+
+ By("invalid serviceName")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: TOO_LONG_EXISTENCE,
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: " ",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+
+ By("invalid version")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "1.32768.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "1.0.0-1.32768.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: " ",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(scerr.ErrInvalidParams))
+
+ By("consumerId is empty")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "1.0.0+",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+
+ By("provider does not exist")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "noneservice",
+ Version: "latest",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(respFind.Failed[0].Error.Code).To(Equal(scerr.ErrServiceNotExists))
+ Expect(respFind.Failed[0].Indexes[0]).To(Equal(int64(0)))
+
+ By("provider does not contain 3.0.0+ versions")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "3.0.0+",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(0))
+ Expect(respFind.Updated[0].Index).To(Equal(int64(0)))
+ Expect(respFind.Updated[0].Rev).ToNot(Equal(""))
+
+ By("consumer does not exist")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: "notExistServiceId",
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "2.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(respFind.Failed[0].Indexes[0]).To(Equal(int64(0)))
+ Expect(respFind.Failed[0].Error.Code).To(Equal(scerr.ErrServiceNotExists))
})
})
@@ -1322,6 +1563,198 @@ var _ = Describe("'Instance' service", func() {
})
})
+ Context("when batch query instances", func() {
+ It("should be passed", func() {
+ By("find with version rule")
+ respFind, err := instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId1,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "latest",
+ },
+ },
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "1.0.0+",
+ },
+ },
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "0.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(respFind.Updated[0].Index).To(Equal(int64(0)))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId2))
+ Expect(respFind.Updated[1].Index).To(Equal(int64(1)))
+ Expect(respFind.Updated[1].Instances[0].InstanceId).To(Equal(instanceId2))
+ Expect(respFind.Failed[0].Indexes[0]).To(Equal(int64(2)))
+ Expect(respFind.Failed[0].Error.Code).To(Equal(scerr.ErrServiceNotExists))
+
+ By("find with env")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId4,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_diff_env_service",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(1))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId4))
+
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ Environment: pb.ENV_PROD,
+ AppId: "query_instance",
+ ServiceName: "query_instance_diff_env_service",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(1))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId4))
+
+ By("find with rev")
+ ctx := util.SetContext(getContext(), serviceUtil.CTX_NOCACHE, "")
+ respFind, err = instanceResource.BatchFind(ctx, &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId8,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_with_rev",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ rev := respFind.Updated[0].Rev
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId8))
+ Expect(len(rev)).NotTo(Equal(0))
+
+ respFind, err = instanceResource.BatchFind(ctx, &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId8,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_with_rev",
+ Version: "1.0.0",
+ },
+ Rev: "x",
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId8))
+ Expect(respFind.Updated[0].Rev).To(Equal(rev))
+
+ respFind, err = instanceResource.BatchFind(ctx, &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId8,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_with_rev",
+ Version: "1.0.0",
+ },
+ Rev: rev,
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(respFind.NotModified[0]).To(Equal(int64(0)))
+
+ By("find should return 200 even if consumer is diff apps")
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId3,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "query_instance",
+ ServiceName: "query_instance_service",
+ Version: "1.0.5",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(0))
+
+ By("shared service discovery")
+ os.Setenv("CSE_SHARED_SERVICES", "query_instance_shared_provider")
+ core.SetSharedMode()
+ core.Service.Environment = pb.ENV_PROD
+
+ respFind, err = instanceResource.BatchFind(
+ util.SetTargetDomainProject(
+ util.SetDomainProject(util.CloneContext(getContext()), "user", "user"),
+ "default", "default"),
+ &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId6,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "default",
+ ServiceName: "query_instance_shared_provider",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(1))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId5))
+
+ respFind, err = instanceResource.BatchFind(getContext(), &pb.BatchFindInstancesRequest{
+ ConsumerServiceId: serviceId7,
+ Services: []*pb.FindService{
+ {
+ Service: &pb.MicroServiceKey{
+ AppId: "default",
+ ServiceName: "query_instance_shared_provider",
+ Version: "1.0.0",
+ },
+ },
+ },
+ })
+ Expect(err).To(BeNil())
+ Expect(respFind.Response.Code).To(Equal(pb.Response_SUCCESS))
+ Expect(len(respFind.Updated[0].Instances)).To(Equal(1))
+ Expect(respFind.Updated[0].Instances[0].InstanceId).To(Equal(instanceId5))
+
+ core.Service.Environment = pb.ENV_DEV
+ })
+ })
+
Context("when query instances between diff dimensions", func() {
It("should be failed", func() {
By("diff appId")
diff --git a/server/service/instance_validator.go b/server/service/instance_validator.go
index 9668d76a..cf7a1dfb 100644
--- a/server/service/instance_validator.go
+++ b/server/service/instance_validator.go
@@ -26,6 +26,7 @@ import (
var (
findInstanceReqValidator validate.Validator
+ batchFindInstanceReqValidator validate.Validator
getInstanceReqValidator validate.Validator
updateInstanceReqValidator validate.Validator
registerInstanceReqValidator validate.Validator
@@ -56,6 +57,17 @@ func FindInstanceReqValidator() *validate.Validator {
})
}
+func BatchFindInstanceReqValidator() *validate.Validator {
+ return batchFindInstanceReqValidator.Init(func(v *validate.Validator) {
+ var findServiceValidator validate.Validator
+ findServiceValidator.AddRule("Service", &validate.ValidateRule{Min: 1})
+ findServiceValidator.AddSub("Service", ExistenceReqValidator())
+ v.AddRule("ConsumerServiceId", GetInstanceReqValidator().GetRule("ConsumerServiceId"))
+ v.AddRule("Services", &validate.ValidateRule{Min: 1})
+ v.AddSub("Services", &findServiceValidator)
+ })
+}
+
func GetInstanceReqValidator() *validate.Validator {
return getInstanceReqValidator.Init(func(v *validate.Validator) {
v.AddRule("ConsumerServiceId", &validate.ValidateRule{Max: 64, Regexp: serviceIdRegex})
diff --git a/server/service/util/instance_util.go b/server/service/util/instance_util.go
index 302347f2..1e0e7916 100644
--- a/server/service/util/instance_util.go
+++ b/server/service/util/instance_util.go
@@ -272,3 +272,27 @@ func UpdateInstance(ctx context.Context, domainProject string, instance *pb.Micr
}
return nil
}
+
+func AppendFindResponse(ctx context.Context, index int64, find *pb.FindInstancesResponse,
+ updatedResult *[]*pb.FindResult, notModifiedResult *[]int64, failedResult **pb.FindFailedResult) {
+ if code := find.GetResponse().GetCode(); code != pb.Response_SUCCESS {
+ if *failedResult == nil {
+ *failedResult = &pb.FindFailedResult{
+ Error: scerr.NewError(code, find.GetResponse().GetMessage()),
+ }
+ }
+ (*failedResult).Indexes = append((*failedResult).Indexes, index)
+ return
+ }
+ iv, _ := ctx.Value(CTX_REQUEST_REVISION).(string)
+ ov, _ := ctx.Value(CTX_RESPONSE_REVISION).(string)
+ if len(iv) > 0 && iv == ov {
+ *notModifiedResult = append(*notModifiedResult, index)
+ return
+ }
+ *updatedResult = append(*updatedResult, &pb.FindResult{
+ Index: index,
+ Instances: find.Instances,
+ Rev: ov,
+ })
+}
diff --git a/server/service/util/instance_util_test.go b/server/service/util/instance_util_test.go
index abe5182c..50087f2e 100644
--- a/server/service/util/instance_util_test.go
+++ b/server/service/util/instance_util_test.go
@@ -20,6 +20,7 @@ import (
"github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/server/core/proto"
pb "github.com/apache/servicecomb-service-center/server/core/proto"
+ scerr "github.com/apache/servicecomb-service-center/server/error"
"golang.org/x/net/context"
"testing"
)
@@ -115,3 +116,77 @@ func TestUpdateInstance(t *testing.T) {
t.Fatalf(`UpdateInstance CTX_NOCACHE failed`)
}
}
+
+func TestAppendFindResponse(t *testing.T) {
+ ctx := context.Background()
+ var (
+ find pb.FindInstancesResponse
+ updatedResult []*pb.FindResult
+ notModifiedResult []int64
+ failedResult *pb.FindFailedResult
+ )
+ AppendFindResponse(ctx, 1, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult == nil || notModifiedResult != nil || failedResult != nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if updatedResult[0].Index != 1 {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+
+ updatedResult = nil
+ cloneCtx := context.WithValue(ctx, CTX_RESPONSE_REVISION, "1")
+ AppendFindResponse(cloneCtx, 1, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult == nil || notModifiedResult != nil || failedResult != nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if updatedResult[0].Index != 1 || updatedResult[0].Rev != "1" {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+
+ updatedResult = nil
+ cloneCtx = context.WithValue(ctx, CTX_REQUEST_REVISION, "1")
+ cloneCtx = context.WithValue(cloneCtx, CTX_RESPONSE_REVISION, "1")
+ AppendFindResponse(cloneCtx, 1, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult != nil || notModifiedResult == nil || failedResult != nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if notModifiedResult[0] != 1 {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+
+ notModifiedResult = nil
+ find.Response = pb.CreateResponse(scerr.ErrInternal, "test")
+ AppendFindResponse(ctx, 1, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult != nil || notModifiedResult != nil || failedResult == nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if failedResult.Error.Code != scerr.ErrInternal {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ find.Response = pb.CreateResponse(scerr.ErrInvalidParams, "test")
+ AppendFindResponse(ctx, 2, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult != nil || notModifiedResult != nil || failedResult == nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if failedResult.Error.Code != scerr.ErrInternal {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+
+ failedResult = nil
+ find.Response = nil
+ AppendFindResponse(ctx, 1, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(ctx, 2, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ cloneCtx = context.WithValue(ctx, CTX_REQUEST_REVISION, "1")
+ cloneCtx = context.WithValue(cloneCtx, CTX_RESPONSE_REVISION, "1")
+ AppendFindResponse(cloneCtx, 3, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(cloneCtx, 4, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ find.Response = pb.CreateResponse(scerr.ErrInternal, "test")
+ AppendFindResponse(ctx, 5, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(ctx, 6, &find, &updatedResult, ¬ModifiedResult, &failedResult)
+ if updatedResult == nil || notModifiedResult == nil || failedResult == nil {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+ if len(updatedResult) != 2 || len(notModifiedResult) != 2 || len(failedResult.Indexes) != 2 {
+ t.Fatal("TestAppendFindResponse failed")
+ }
+}
diff --git a/server/service/validate.go b/server/service/validate.go
index c13728ad..825703dc 100644
--- a/server/service/validate.go
+++ b/server/service/validate.go
@@ -74,6 +74,8 @@ func Validate(v interface{}) error {
return RegisterInstanceReqValidator().Validate(v)
case *pb.FindInstancesRequest:
return FindInstanceReqValidator().Validate(v)
+ case *pb.BatchFindInstancesRequest:
+ return BatchFindInstanceReqValidator().Validate(v)
case *pb.HeartbeatRequest, *pb.UnregisterInstanceRequest:
return HeartbeatReqValidator().Validate(v)
case *pb.UpdateInstancePropsRequest:
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services