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 deb2b08 SCB-2094 Add list provider instances APIs (#726)
deb2b08 is described below
commit deb2b086c97fcf41f68619823a96972b8b943a4f
Author: little-cui <[email protected]>
AuthorDate: Wed Oct 21 11:49:29 2020 +0800
SCB-2094 Add list provider instances APIs (#726)
---
datasource/etcd/ms.go | 231 ++++++++++++++++++++++++++++++++-----------------
datasource/ms.go | 10 +++
pkg/registry/struct.go | 8 +-
3 files changed, 168 insertions(+), 81 deletions(-)
diff --git a/datasource/etcd/ms.go b/datasource/etcd/ms.go
index d7793e8..7f26e45 100644
--- a/datasource/etcd/ms.go
+++ b/datasource/etcd/ms.go
@@ -36,10 +36,24 @@ import (
scerr "github.com/apache/servicecomb-service-center/server/scerror"
"github.com/apache/servicecomb-service-center/server/service/cache"
serviceUtil
"github.com/apache/servicecomb-service-center/server/service/util"
+ "sort"
"strconv"
"time"
)
+var clustersIndex = make(map[string]int)
+
+func init() {
+ var clusters []string
+ for name := range registry.Configuration().Clusters {
+ clusters = append(clusters, name)
+ }
+ sort.Strings(clusters)
+ for i, name := range clusters {
+ clustersIndex[name] = i
+ }
+}
+
// RegisterService() implement:
// 1. capsule request to etcd kv format
// 2. invoke etcd client to store data
@@ -616,6 +630,143 @@ func (ds *DataSource) GetInstance(ctx context.Context,
request *pb.GetOneInstanc
}, nil
}
+func (ds *DataSource) GetInstances(ctx context.Context, request
*pb.GetInstancesRequest) (*pb.GetInstancesResponse,
+ error) {
+ domainProject := util.ParseDomainProject(ctx)
+
+ service := &pb.MicroService{}
+ var err error
+ if len(request.ConsumerServiceId) > 0 {
+ service, err = serviceUtil.GetService(ctx, domainProject,
request.ConsumerServiceId)
+ if err != nil {
+ log.Errorf(err, "get consumer failed, consumer[%s] find
provider instances",
+ request.ConsumerServiceId,
request.ProviderServiceId)
+ return &pb.GetInstancesResponse{
+ Response:
proto.CreateResponse(scerr.ErrInternal, err.Error()),
+ }, err
+ }
+ if service == nil {
+ log.Errorf(nil, "consumer does not exist, consumer[%s]
find provider instances",
+ request.ConsumerServiceId,
request.ProviderServiceId)
+ return &pb.GetInstancesResponse{
+ Response:
proto.CreateResponse(scerr.ErrServiceNotExists,
+ fmt.Sprintf("Consumer[%s] does not
exist.", request.ConsumerServiceId)),
+ }, nil
+ }
+ }
+
+ provider, err := serviceUtil.GetService(ctx, domainProject,
request.ProviderServiceId)
+ if err != nil {
+ log.Errorf(err, "get provider failed, consumer[%s] find
provider instances",
+ request.ConsumerServiceId, request.ProviderServiceId)
+ return &pb.GetInstancesResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal,
err.Error()),
+ }, err
+ }
+ if provider == nil {
+ log.Errorf(nil, "provider does not exist, consumer[%s] find
provider instances",
+ request.ConsumerServiceId, request.ProviderServiceId)
+ return &pb.GetInstancesResponse{
+ Response:
proto.CreateResponse(scerr.ErrServiceNotExists,
+ fmt.Sprintf("Provider[%s] does not exist.",
request.ProviderServiceId)),
+ }, nil
+ }
+
+ findFlag := func() string {
+ return fmt.Sprintf("Consumer[%s][%s/%s/%s/%s] find
provider[%s][%s/%s/%s/%s] instances",
+ request.ConsumerServiceId, service.Environment,
service.AppId, service.ServiceName, service.Version,
+ provider.ServiceId, provider.Environment,
provider.AppId, provider.ServiceName, provider.Version)
+ }
+
+ var item *cache.VersionRuleCacheItem
+ rev, _ := ctx.Value(util.CtxRequestRevision).(string)
+ item, err = cache.FindInstances.GetWithProviderID(ctx, service,
proto.MicroServiceToKey(domainProject, provider),
+ &pb.HeartbeatSetElement{
+ ServiceId: request.ProviderServiceId,
+ }, request.Tags, rev)
+ if err != nil {
+ log.Errorf(err, "FindInstances.GetWithProviderID failed, %s
failed", findFlag())
+ return &pb.GetInstancesResponse{
+ Response: proto.CreateResponse(scerr.ErrInternal,
err.Error()),
+ }, err
+ }
+ if item == nil || len(item.ServiceIds) == 0 {
+ mes := fmt.Errorf("%s failed, provider instance does not
exist", findFlag())
+ log.Errorf(mes, "FindInstances.GetWithProviderID failed")
+ return &pb.GetInstancesResponse{
+ Response:
proto.CreateResponse(scerr.ErrServiceNotExists, mes.Error()),
+ }, nil
+ }
+
+ instances := item.Instances
+ if rev == item.Rev {
+ instances = nil // for gRPC
+ }
+ _ = util.SetContext(ctx, util.CtxResponseRevision, item.Rev)
+
+ return &pb.GetInstancesResponse{
+ Response: proto.CreateResponse(proto.Response_SUCCESS, "Query
service instances successfully."),
+ Instances: instances,
+ }, nil
+}
+
+func (ds *DataSource) GetProviderInstances(ctx context.Context, request
*pb.HeartbeatSetElement) (instances []*pb.MicroServiceInstance, rev string, err
error) {
+ var (
+ maxRevs = make([]int64, len(clustersIndex))
+ counts = make([]int64, len(clustersIndex))
+ domainProject = util.ParseTargetDomainProject(ctx)
+ )
+ instances, err = ds.findInstances(ctx, domainProject,
request.ServiceId, request.InstanceId, maxRevs, counts)
+ if err != nil {
+ return
+ }
+ return instances, serviceUtil.FormatRevision(maxRevs, counts), nil
+}
+
+func (ds *DataSource) BatchGetProviderInstances(ctx context.Context, request
*pb.BatchGetInstancesRequest) (instances []*pb.MicroServiceInstance, rev
string, err error) {
+ var (
+ maxRevs = make([]int64, len(clustersIndex))
+ counts = make([]int64, len(clustersIndex))
+ domainProject = util.ParseTargetDomainProject(ctx)
+ )
+ if request == nil || len(request.ServiceIds) == 0 {
+ return nil, "", fmt.Errorf("invalid param
BatchGetInstancesRequest")
+ }
+
+ for _, providerServiceID := range request.ServiceIds {
+ insts, err := ds.findInstances(ctx, domainProject,
providerServiceID, "", maxRevs, counts)
+ if err != nil {
+ return nil, "", err
+ }
+ instances = append(instances, insts...)
+ }
+
+ return instances, serviceUtil.FormatRevision(maxRevs, counts), nil
+}
+
+func (ds *DataSource) findInstances(ctx context.Context, domainProject,
serviceID, instanceID string, maxRevs []int64, counts []int64) (instances
[]*pb.MicroServiceInstance, err error) {
+ key := apt.GenerateInstanceKey(domainProject, serviceID, instanceID)
+ opts := append(serviceUtil.FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+ if len(resp.Kvs) == 0 {
+ return
+ }
+
+ for _, kv := range resp.Kvs {
+ if i, ok := clustersIndex[kv.ClusterName]; ok {
+ if kv.ModRevision > maxRevs[i] {
+ maxRevs[i] = kv.ModRevision
+ }
+ counts[i]++
+ }
+ instances = append(instances,
kv.Value.(*pb.MicroServiceInstance))
+ }
+ return
+}
+
func (ds *DataSource) FindInstances(ctx context.Context, request
*pb.FindInstancesRequest) (*pb.FindInstancesResponse,
error) {
provider := &pb.MicroServiceKey{
@@ -903,86 +1054,6 @@ func (ds *DataSource) HeartbeatSet(ctx context.Context,
request *pb.HeartbeatSet
}, nil
}
-func (ds *DataSource) GetInstances(ctx context.Context, request
*pb.GetInstancesRequest) (*pb.GetInstancesResponse,
- error) {
- domainProject := util.ParseDomainProject(ctx)
-
- service := &pb.MicroService{}
- var err error
- if len(request.ConsumerServiceId) > 0 {
- service, err = serviceUtil.GetService(ctx, domainProject,
request.ConsumerServiceId)
- if err != nil {
- log.Errorf(err, "get consumer failed, consumer[%s] find
provider instances",
- request.ConsumerServiceId,
request.ProviderServiceId)
- return &pb.GetInstancesResponse{
- Response:
proto.CreateResponse(scerr.ErrInternal, err.Error()),
- }, err
- }
- if service == nil {
- log.Errorf(nil, "consumer does not exist, consumer[%s]
find provider instances",
- request.ConsumerServiceId,
request.ProviderServiceId)
- return &pb.GetInstancesResponse{
- Response:
proto.CreateResponse(scerr.ErrServiceNotExists,
- fmt.Sprintf("Consumer[%s] does not
exist.", request.ConsumerServiceId)),
- }, nil
- }
- }
-
- provider, err := serviceUtil.GetService(ctx, domainProject,
request.ProviderServiceId)
- if err != nil {
- log.Errorf(err, "get provider failed, consumer[%s] find
provider instances",
- request.ConsumerServiceId, request.ProviderServiceId)
- return &pb.GetInstancesResponse{
- Response: proto.CreateResponse(scerr.ErrInternal,
err.Error()),
- }, err
- }
- if provider == nil {
- log.Errorf(nil, "provider does not exist, consumer[%s] find
provider instances",
- request.ConsumerServiceId, request.ProviderServiceId)
- return &pb.GetInstancesResponse{
- Response:
proto.CreateResponse(scerr.ErrServiceNotExists,
- fmt.Sprintf("Provider[%s] does not exist.",
request.ProviderServiceId)),
- }, nil
- }
-
- findFlag := func() string {
- return fmt.Sprintf("Consumer[%s][%s/%s/%s/%s] find
provider[%s][%s/%s/%s/%s] instances",
- request.ConsumerServiceId, service.Environment,
service.AppId, service.ServiceName, service.Version,
- provider.ServiceId, provider.Environment,
provider.AppId, provider.ServiceName, provider.Version)
- }
-
- var item *cache.VersionRuleCacheItem
- rev, _ := ctx.Value(util.CtxRequestRevision).(string)
- item, err = cache.FindInstances.GetWithProviderID(ctx, service,
proto.MicroServiceToKey(domainProject, provider),
- &pb.HeartbeatSetElement{
- ServiceId: request.ProviderServiceId,
- }, request.Tags, rev)
- if err != nil {
- log.Errorf(err, "FindInstances.GetWithProviderID failed, %s
failed", findFlag())
- return &pb.GetInstancesResponse{
- Response: proto.CreateResponse(scerr.ErrInternal,
err.Error()),
- }, err
- }
- if item == nil || len(item.ServiceIds) == 0 {
- mes := fmt.Errorf("%s failed, provider instance does not
exist", findFlag())
- log.Errorf(mes, "FindInstances.GetWithProviderID failed")
- return &pb.GetInstancesResponse{
- Response:
proto.CreateResponse(scerr.ErrServiceNotExists, mes.Error()),
- }, nil
- }
-
- instances := item.Instances
- if rev == item.Rev {
- instances = nil // for gRPC
- }
- _ = util.SetContext(ctx, util.CtxResponseRevision, item.Rev)
-
- return &pb.GetInstancesResponse{
- Response: proto.CreateResponse(proto.Response_SUCCESS, "Query
service instances successfully."),
- Instances: instances,
- }, nil
-}
-
func (ds *DataSource) BatchFind(ctx context.Context, request
*pb.BatchFindInstancesRequest) (
*pb.BatchFindInstancesResponse, error) {
response := &pb.BatchFindInstancesResponse{
diff --git a/datasource/ms.go b/datasource/ms.go
index 424530b..ad18b3c 100644
--- a/datasource/ms.go
+++ b/datasource/ms.go
@@ -25,6 +25,7 @@ import (
// Attention: request validation must be finished before the following
interface being invoked!!!
// MetadataManager contains the CRUD of registry metadata
type MetadataManager interface {
+ // Microservice management
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)
@@ -39,9 +40,15 @@ type MetadataManager interface {
GetDeleteServiceFunc(ctx context.Context, serviceID string, force bool,
serviceRespChan chan<- *pb.DelServicesRspInfo)
func(context.Context)
+ // Instance management
RegisterInstance(ctx context.Context, request
*pb.RegisterInstanceRequest) (*pb.RegisterInstanceResponse, error)
+ // GetInstances returns instances under the current domain
GetInstance(ctx context.Context, request *pb.GetOneInstanceRequest)
(*pb.GetOneInstanceResponse, error)
GetInstances(ctx context.Context, request *pb.GetInstancesRequest)
(*pb.GetInstancesResponse, error)
+ // GetProviderInstances returns instances under the specified domain
+ GetProviderInstances(ctx context.Context, request
*pb.HeartbeatSetElement) (instances []*pb.MicroServiceInstance, rev string, err
error)
+ BatchGetProviderInstances(ctx context.Context, request
*pb.BatchGetInstancesRequest) (instances []*pb.MicroServiceInstance, rev
string, err error)
+ // FindInstances returns instances under the specified domain
FindInstances(ctx context.Context, request *pb.FindInstancesRequest)
(*pb.FindInstancesResponse, error)
UpdateInstanceStatus(ctx context.Context, request
*pb.UpdateInstanceStatusRequest) (
*pb.UpdateInstanceStatusResponse, error)
@@ -53,6 +60,7 @@ type MetadataManager interface {
HeartbeatSet(ctx context.Context, request *pb.HeartbeatSetRequest)
(*pb.HeartbeatSetResponse, error)
BatchFind(ctx context.Context, request *pb.BatchFindInstancesRequest)
(*pb.BatchFindInstancesResponse, error)
+ // Schema management
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)
@@ -60,11 +68,13 @@ type MetadataManager interface {
GetAllSchemas(ctx context.Context, request *pb.GetAllSchemaRequest)
(*pb.GetAllSchemaResponse, error)
DeleteSchema(ctx context.Context, request *pb.DeleteSchemaRequest)
(*pb.DeleteSchemaResponse, error)
+ // Tag management
AddTags(ctx context.Context, request *pb.AddServiceTagsRequest)
(*pb.AddServiceTagsResponse, error)
GetTags(ctx context.Context, request *pb.GetServiceTagsRequest)
(*pb.GetServiceTagsResponse, error)
UpdateTag(ctx context.Context, request *pb.UpdateServiceTagRequest)
(*pb.UpdateServiceTagResponse, error)
DeleteTags(ctx context.Context, request *pb.DeleteServiceTagsRequest)
(*pb.DeleteServiceTagsResponse, error)
+ // White/black list management
AddRule(ctx context.Context, request *pb.AddServiceRulesRequest)
(*pb.AddServiceRulesResponse, error)
GetRule(ctx context.Context, request *pb.GetServiceRulesRequest)
(*pb.GetServiceRulesResponse, error)
UpdateRule(ctx context.Context, request *pb.UpdateServiceRuleRequest)
(*pb.UpdateServiceRuleResponse, error)
diff --git a/pkg/registry/struct.go b/pkg/registry/struct.go
index 453ce57..02b8eb9 100644
--- a/pkg/registry/struct.go
+++ b/pkg/registry/struct.go
@@ -104,7 +104,9 @@ type GetServicesInfoResponse struct {
}
type MicroServiceKey struct {
- Tenant string `protobuf:"bytes,1,opt,name=tenant"
json:"tenant,omitempty"`
+ // Tenant: The format is "{domain}/{project}"
+ Tenant string `protobuf:"bytes,1,opt,name=tenant"
json:"tenant,omitempty"`
+ // Deprecated: Use Tenant instead
Project string `protobuf:"bytes,2,opt,name=project"
json:"project,omitempty"`
AppId string `protobuf:"bytes,3,opt,name=appId"
json:"appId,omitempty"`
ServiceName string `protobuf:"bytes,4,opt,name=serviceName"
json:"serviceName,omitempty"`
@@ -415,3 +417,7 @@ type GetAppsResponse struct {
type MicroServiceDependency struct {
Dependency []*MicroServiceKey `json:"Dependency,omitempty"`
}
+
+type BatchGetInstancesRequest struct {
+ ServiceIds []string `json:"serviceIds,omitempty"`
+}
