You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@servicecomb.apache.org by GitBox <gi...@apache.org> on 2018/03/19 14:01:56 UTC
[GitHub] asifdxtreme closed pull request #310: SCB-411 SC can not check the reduplicate endpoints when register with ID
asifdxtreme closed pull request #310: SCB-411 SC can not check the reduplicate endpoints when register with ID
URL: https://github.com/apache/incubator-servicecomb-service-center/pull/310
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/pkg/validate/validate.go b/pkg/validate/validate.go
index 59190e3b..7bbe15e1 100644
--- a/pkg/validate/validate.go
+++ b/pkg/validate/validate.go
@@ -48,10 +48,10 @@ func (v *ValidateRule) String() string {
idx++
}
if v.Regexp != nil {
- arr[idx] = fmt.Sprintf("Length: %s", v.Regexp)
+ arr[idx] = fmt.Sprintf("Regexp: %s", v.Regexp)
idx++
}
- return "rule: {" + util.StringJoin(arr[:idx], ",") + "}"
+ return "{" + util.StringJoin(arr[:idx], ",") + "}"
}
func (v *ValidateRule) Match(s interface{}) bool {
@@ -239,9 +239,9 @@ func (v *Validator) Validate(s interface{}) error {
// TODO null pointer如何校验
if field.Kind() != reflect.Ptr && !validate.Match(fi) {
if filter(fieldName) {
- return fmt.Errorf("invalid field: %s.%s , %s", st.Type.Name(), fieldName, validate)
+ return fmt.Errorf("The field '%s.%s' value does not match rule: %s", st.Type.Name(), fieldName, validate)
}
- return fmt.Errorf("invalid field: %s.%s, invalid value: {%v} , %s", st.Type.Name(), fieldName, fi, validate)
+ return fmt.Errorf("The field '%s.%s' value(%v) does not match rule: %s", st.Type.Name(), fieldName, fi, validate)
}
}
}
@@ -249,9 +249,9 @@ func (v *Validator) Validate(s interface{}) error {
}
var (
- BLACK_LIST_FOR_PRINT = map[string]interface{} {
- "Properties": nil,
- }
+ BLACK_LIST_FOR_PRINT = map[string]struct{}{
+ "Properties": {},
+ }
)
func filter(fieldName string) bool {
diff --git a/server/core/backend/store/lease.go b/server/core/backend/store/lease.go
index 04a2ff98..192ae8c6 100644
--- a/server/core/backend/store/lease.go
+++ b/server/core/backend/store/lease.go
@@ -43,27 +43,27 @@ func (lat *LeaseAsyncTask) Do(ctx context.Context) (err error) {
lat.StartTime = time.Now()
lat.TTL, err = backend.Registry().LeaseRenew(ctx, lat.LeaseID)
lat.EndTime = time.Now()
- if err == nil {
- lat.err = err
- util.LogNilOrWarnf(lat.CreateTime, "renew lease %d(rev: %s, run: %s), key %s",
+ if err != nil {
+ util.Logger().Errorf(err, "[%s]renew lease %d failed(rev: %s, run: %s), key %s",
+ time.Now().Sub(lat.CreateTime),
lat.LeaseID,
lat.CreateTime.Format(TIME_FORMAT),
lat.StartTime.Format(TIME_FORMAT),
lat.Key())
- return
+ if _, ok := err.(errorsEx.InternalError); !ok {
+ // it means lease not found if err is not the InternalError type
+ lat.err = err
+ return
+ }
}
- util.Logger().Errorf(err, "[%s]renew lease %d failed(rev: %s, run: %s), key %s",
- time.Now().Sub(lat.CreateTime),
+ lat.err, err = nil, nil
+ util.LogNilOrWarnf(lat.CreateTime, "renew lease %d(rev: %s, run: %s), key %s",
lat.LeaseID,
lat.CreateTime.Format(TIME_FORMAT),
lat.StartTime.Format(TIME_FORMAT),
lat.Key())
- if _, ok := err.(errorsEx.InternalError); !ok {
- lat.err = err
- return
- }
- return nil
+ return
}
func (lat *LeaseAsyncTask) Err() error {
diff --git a/server/core/common.go b/server/core/common.go
index 7b35285b..2ea5de2c 100644
--- a/server/core/common.go
+++ b/server/core/common.go
@@ -67,7 +67,7 @@ func init() {
pathRegex, _ := regexp.Compile(`^[A-Za-z0-9.,?'\\/+&%$#=~_\-@{}]*$`)
// descriptionRegex, _ := regexp.Compile(`^[\p{Han}\w\s。.:*,\-：”“"]*$`)
levelRegex, _ := regexp.Compile(`^(FRONT|MIDDLE|BACK)$`)
- statusRegex, _ := regexp.Compile("^(" + pb.MS_UP + "|" + pb.MS_DOWN + ")*$")
+ statusRegex, _ := regexp.Compile("^(" + pb.MS_UP + "|" + pb.MS_DOWN + ")?$")
serviceIdRegex, _ := regexp.Compile(`^.*$`)
aliasRegex, _ := regexp.Compile(`^[a-zA-Z0-9_\-.:]*$`)
registerByRegex, _ := regexp.Compile("^(" + util.StringJoin([]string{pb.REGISTERBY_SDK, pb.REGISTERBY_SIDECAR}, "|") + ")*$")
@@ -78,7 +78,7 @@ func init() {
// map/slice的长度由validator中的min/max/length控制
schemaIdRegex, _ := regexp.Compile(`^[a-zA-Z0-9]{1,160}$|^[a-zA-Z0-9][a-zA-Z0-9_\-.]{0,158}[a-zA-Z0-9]$`) //length:{1,160}
instStatusRegex, _ := regexp.Compile("^(" + util.StringJoin([]string{
- pb.MSI_UP, pb.MSI_DOWN, pb.MSI_STARTING, pb.MSI_OUTOFSERVICE}, "|") + ")$")
+ pb.MSI_UP, pb.MSI_DOWN, pb.MSI_STARTING, pb.MSI_OUTOFSERVICE}, "|") + ")?$")
tagRegex, _ := regexp.Compile(`^[a-zA-Z][a-zA-Z0-9_\-.]{0,63}$`)
hbModeRegex, _ := regexp.Compile(`^(push|pull)$`)
numberAllowEmptyRegex, _ := regexp.Compile(`^[0-9]*$`)
@@ -111,7 +111,7 @@ func init() {
MicroServiceValidator.AddRules(MicroServiceKeyValidator.GetRules())
MicroServiceValidator.AddRule("Description", &validate.ValidateRule{Length: 256})
MicroServiceValidator.AddRule("Level", &validate.ValidateRule{Min: 1, Regexp: levelRegex})
- MicroServiceValidator.AddRule("Status", &validate.ValidateRule{Min: 1, Regexp: statusRegex})
+ MicroServiceValidator.AddRule("Status", &validate.ValidateRule{Regexp: statusRegex})
MicroServiceValidator.AddRule("Schemas", SchemaIdRule)
MicroServiceValidator.AddSub("Paths", &ServicePathValidator)
MicroServiceValidator.AddRule("Alias", &validate.ValidateRule{Length: 128, Regexp: aliasRegex})
@@ -124,15 +124,15 @@ func init() {
GetSchemaExistsReqValidator.AddRule("ServiceId", ServiceIdRule)
GetSchemaExistsReqValidator.AddRule("SchemaId", SchemaIdRule)
- var subSchemaValidor validate.Validator
- subSchemaValidor.AddRule("SchemaId", SchemaIdRule)
- subSchemaValidor.AddRule("Summary", &validate.ValidateRule{Min: 1, Max: 512, Regexp: SchemaSummaryRegex})
- subSchemaValidor.AddRule("Schema", &validate.ValidateRule{Min: 1})
+ var subSchemaValidator validate.Validator
+ subSchemaValidator.AddRule("SchemaId", SchemaIdRule)
+ subSchemaValidator.AddRule("Summary", &validate.ValidateRule{Min: 1, Max: 512, Regexp: SchemaSummaryRegex})
+ subSchemaValidator.AddRule("Schema", &validate.ValidateRule{Min: 1})
SchemasValidator.AddRule("ServiceId", ServiceIdRule)
- SchemasValidator.AddSub("Schemas", &subSchemaValidor)
+ SchemasValidator.AddSub("Schemas", &subSchemaValidator)
- SchemaValidator.AddRules(subSchemaValidor.GetRules())
+ SchemaValidator.AddRules(subSchemaValidator.GetRules())
SchemaValidator.AddRule("ServiceId", ServiceIdRule)
SchemaValidator.AddRule("Summary", &validate.ValidateRule{Max: 512, Regexp: SchemaSummaryRegex})
diff --git a/server/core/key_generator.go b/server/core/key_generator.go
index cb012362..67119f1a 100644
--- a/server/core/key_generator.go
+++ b/server/core/key_generator.go
@@ -19,6 +19,7 @@ package core
import (
"github.com/apache/incubator-servicecomb-service-center/pkg/util"
pb "github.com/apache/incubator-servicecomb-service-center/server/core/proto"
+ "sort"
)
const (
@@ -44,6 +45,8 @@ const (
ENDPOINTS_ROOT_KEY = "eps"
)
+const NODE_IP = "nodeIP"
+
func GetRootKey() string {
return util.StringJoin([]string{
"",
@@ -372,12 +375,28 @@ func GetEndpointsRootKey(domainProject string) string {
}, "/")
}
-func GenerateEndpointsIndexKey(domainProject, region, availableZone, nodeIP, endpoints string) string {
+func ParseRegionAndAvailableZone(in *pb.DataCenterInfo) (region string, availableZone string) {
+ if in == nil {
+ return "", ""
+ }
+ region = in.Region
+ availableZone = in.AvailableZone
+ return
+}
+
+func GenerateEndpointsIndexKey(domainProject string, instance *pb.MicroServiceInstance) string {
+ region, availableZone := ParseRegionAndAvailableZone(instance.DataCenterInfo)
+ nodeIP := ""
+ if value, ok := instance.Properties[NODE_IP]; ok {
+ nodeIP = value
+ }
+ sort.Strings(instance.Endpoints)
+ endpointsJoin := util.StringJoin(instance.Endpoints, "/")
return util.StringJoin([]string{
GetEndpointsRootKey(domainProject),
region,
availableZone,
nodeIP,
- endpoints,
+ endpointsJoin,
}, "/")
}
diff --git a/server/core/microservice.go b/server/core/microservice.go
index 0ae21f73..d454a8ed 100644
--- a/server/core/microservice.go
+++ b/server/core/microservice.go
@@ -163,12 +163,3 @@ func HeartbeatRequest() *pb.HeartbeatRequest {
InstanceId: Instance.InstanceId,
}
}
-
-func GetRegionAndAvailableZone(in *pb.DataCenterInfo) (region string, availableZone string) {
- if in == nil {
- return "", ""
- }
- region = in.Region
- availableZone = in.AvailableZone
- return
-}
diff --git a/server/core/swagger/v3.yaml b/server/core/swagger/v3.yaml
index 135294d2..2e644b20 100644
--- a/server/core/swagger/v3.yaml
+++ b/server/core/swagger/v3.yaml
@@ -1452,7 +1452,7 @@ definitions:
instances:
type: array
items:
- $ref: '#/definitions/RegistMicroserviceInstance'
+ $ref: '#/definitions/MicroServiceInstance'
tags:
$ref: '#/definitions/Tags'
@@ -1467,7 +1467,7 @@ definitions:
type: object
properties:
instance:
- $ref: '#/definitions/RegistMicroserviceInstance'
+ $ref: '#/definitions/MicroServiceInstance'
CreateInstanceResponse:
type: object
properties:
@@ -1510,7 +1510,6 @@ definitions:
- appId
- serviceName
- version
- - status
properties:
serviceId:
type: string
@@ -1542,7 +1541,7 @@ definitions:
type: string
status:
type: string
- description: 微服务状态，UP表示上线 DOWN表示下线
+ description: 微服务状态，UP表示上线 DOWN表示下线，默认值UP
enum:
- UP
- DOWN
@@ -1586,41 +1585,11 @@ definitions:
times:
type: integer
description: retry times
- RegistMicroserviceInstance:
- type: object
- required:
- - status
- - hostName
- - environment
- properties:
- instanceId:
- description: 实例ID,不填写会自动填充
- type: string
- hostName:
- description: 机器的hostname
- type: string
- endpoints:
- type: array
- items:
- type: string
- description: 例:rest:127.0.0.1:8080
- status:
- type: string
- description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE
- properties:
- $ref: '#/definitions/Properties'
- healthCheck:
- $ref: '#/definitions/HealthCheck'
- environment:
- type: string
- description: development|testing|acceptance|production
- dataCenterInfo:
- $ref: '#/definitions/DataCenterInfo'
MicroServiceInstance:
type: object
required:
- - status
- hostName
+ - endpoints
properties:
instanceId:
type: string
@@ -1640,7 +1609,7 @@ definitions:
description: 例:rest:127.0.0.1:8080
status:
type: string
- description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE
+ description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE，默认值UP
properties:
$ref: '#/definitions/Properties'
healthCheck:
diff --git a/server/core/swagger/v4.yaml b/server/core/swagger/v4.yaml
index cf927087..43cbf62f 100644
--- a/server/core/swagger/v4.yaml
+++ b/server/core/swagger/v4.yaml
@@ -1760,7 +1760,7 @@ definitions:
instances:
type: array
items:
- $ref: '#/definitions/RegistMicroserviceInstance'
+ $ref: '#/definitions/MicroServiceInstance'
tags:
$ref: '#/definitions/Tags'
@@ -1775,7 +1775,7 @@ definitions:
type: object
properties:
instance:
- $ref: '#/definitions/RegistMicroserviceInstance'
+ $ref: '#/definitions/MicroServiceInstance'
CreateInstanceResponse:
type: object
properties:
@@ -1818,7 +1818,6 @@ definitions:
- appId
- serviceName
- version
- - status
properties:
serviceId:
type: string
@@ -1850,7 +1849,7 @@ definitions:
type: string
status:
type: string
- description: 微服务状态，UP表示上线 DOWN表示下线
+ description: 微服务状态，UP表示上线 DOWN表示下线，默认值UP
enum:
- UP
- DOWN
@@ -1894,41 +1893,11 @@ definitions:
times:
type: integer
description: retry times
- RegistMicroserviceInstance:
- type: object
- required:
- - status
- - hostName
- - environment
- properties:
- instanceId:
- description: 实例ID,不填写会自动填充
- type: string
- hostName:
- description: 机器的hostname
- type: string
- endpoints:
- type: array
- items:
- type: string
- description: 例:rest:127.0.0.1:8080
- status:
- type: string
- description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE
- properties:
- $ref: '#/definitions/Properties'
- healthCheck:
- $ref: '#/definitions/HealthCheck'
- environment:
- type: string
- description: development|testing|acceptance|production
- dataCenterInfo:
- $ref: '#/definitions/DataCenterInfo'
MicroServiceInstance:
type: object
required:
- - status
- hostName
+ - endpoints
properties:
instanceId:
type: string
@@ -1948,7 +1917,7 @@ definitions:
description: 例:rest:127.0.0.1:8080
status:
type: string
- description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE
+ description: 实例状态，UP|DOWN|STARTING|OUTOFSERVICE，默认值UP
properties:
$ref: '#/definitions/Properties'
healthCheck:
diff --git a/server/service/instances.go b/server/service/instances.go
index e387fb01..7e966d63 100644
--- a/server/service/instances.go
+++ b/server/service/instances.go
@@ -40,103 +40,17 @@ import (
type InstanceService struct {
}
-func (s *InstanceService) Register(ctx context.Context, in *pb.RegisterInstanceRequest) (*pb.RegisterInstanceResponse, error) {
- if in == nil || in.Instance == nil {
- util.Logger().Errorf(nil, "register instance failed: invalid params.")
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrInvalidParams, "Request format invalid."),
- }, nil
- }
- instance := in.GetInstance()
+func (s *InstanceService) preProcessRegisterInstance(ctx context.Context, instance *pb.MicroServiceInstance) *scerr.Error {
if len(instance.Status) == 0 {
instance.Status = pb.MSI_UP
}
- remoteIP := util.GetIPFromContext(ctx)
- instanceFlag := util.StringJoin([]string{instance.ServiceId, instance.HostName}, "/")
- err := apt.Validate(instance)
- if err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, operator %s: invalid instance parameters.",
- instanceFlag, remoteIP)
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrInvalidParams, err.Error()),
- }, nil
- }
- //先以domain/project的方式组装
- domainProject := util.ParseDomainProject(ctx)
-
- // service id存在性校验
- service, err := serviceUtil.GetService(ctx, domainProject, instance.ServiceId)
- if service == nil || err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, operator %s: service not exist.", instanceFlag, remoteIP)
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrServiceNotExists, "Service does not exist."),
- }, nil
- }
-
- instanceId := instance.InstanceId
- //允许自定义id
- //如果没填写 并且endpoints沒重復，則产生新的全局instance id
- oldInstanceId := ""
-
- var endpointsIndexKey string
- if instanceId == "" {
- util.Logger().Infof("start register a new instance: service %s", instanceFlag)
- if len(instance.Endpoints) != 0 {
- oldInstanceId, endpointsIndexKey, err = serviceUtil.CheckEndPoints(ctx, in)
- if err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, operator %s: check endpoints failed.", instanceFlag, remoteIP)
- if oldInstanceId != "" {
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrEndpointAlreadyExists, err.Error()),
- }, nil
- }
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrInternal, err.Error()),
- }, err
- }
- if oldInstanceId != "" {
- util.Logger().Infof("register instance successful, reuse service %s instance %s, operator %s",
- instance.ServiceId, oldInstanceId, remoteIP)
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(pb.Response_SUCCESS, "instance more exist."),
- InstanceId: oldInstanceId,
- }, nil
- }
- }
- }
-
- var reporter quota.QuotaReporter
- if len(oldInstanceId) == 0 {
- if !apt.IsSCInstance(ctx) {
- res := quota.NewApplyQuotaResource(quota.MicroServiceInstanceQuotaType, domainProject, in.Instance.ServiceId, 1)
- rst := plugin.Plugins().Quota().Apply4Quotas(ctx, res)
- reporter = rst.Reporter
- err := rst.Err
- if reporter != nil {
- defer reporter.Close()
- }
- if err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, operator %s: no quota apply.", instanceFlag, remoteIP)
- response := &pb.RegisterInstanceResponse{
- Response: pb.CreateResponseWithSCErr(err),
- }
- if err.InternalError() {
- return response, err
- }
- return response, nil
- }
- }
- }
-
- if len(instanceId) == 0 {
- instanceId = plugin.Plugins().UUID().GetInstanceId()
- instance.InstanceId = instanceId
+ if len(instance.InstanceId) == 0 {
+ instance.InstanceId = plugin.Plugins().UUID().GetInstanceId()
}
instance.Timestamp = strconv.FormatInt(time.Now().Unix(), 10)
instance.ModTimestamp = instance.Timestamp
- util.Logger().Debug(fmt.Sprintf("instance ID [%s]", instanceId))
// 这里应该根据租约计时
renewalInterval := apt.REGISTRY_DEFAULT_LEASE_RENEWALINTERVAL
@@ -154,10 +68,7 @@ func (s *InstanceService) Register(ctx context.Context, in *pb.RegisterInstanceR
if instance.HealthCheck.Interval <= 0 || instance.HealthCheck.Interval >= math.MaxInt32 ||
instance.HealthCheck.Times <= 0 || instance.HealthCheck.Times >= math.MaxInt32 ||
instance.HealthCheck.Interval*(instance.HealthCheck.Times+1) >= math.MaxInt32 {
- util.Logger().Errorf(err, "register instance %s(%s) failed for invalid health check settings.", instance.ServiceId, instance.HostName)
- return &pb.RegisterInstanceResponse{
- Response: pb.CreateResponse(scerr.ErrInvalidParams, "Invalid health check settings."),
- }, nil
+ return scerr.NewError(scerr.ErrInvalidParams, "Invalid 'healthCheck' settings in request body.")
}
renewalInterval = instance.HealthCheck.Interval
retryTimes = instance.HealthCheck.Times
@@ -165,68 +76,175 @@ func (s *InstanceService) Register(ctx context.Context, in *pb.RegisterInstanceR
// 默认120s
}
}
- ttl := int64(renewalInterval * (retryTimes + 1))
+ domainProject := util.ParseDomainProject(ctx)
+ service, err := serviceUtil.GetService(ctx, domainProject, instance.ServiceId)
+ if service == nil || err != nil {
+ return scerr.NewError(scerr.ErrServiceNotExists, "Invalid 'serviceId' in request body.")
+ }
instance.Version = service.Version
+ return nil
+}
+func (s *InstanceService) Register(ctx context.Context, in *pb.RegisterInstanceRequest) (*pb.RegisterInstanceResponse, error) {
+ if in == nil || in.Instance == nil {
+ util.Logger().Errorf(nil, "register instance failed: invalid params.")
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponse(scerr.ErrInvalidParams, "Request format invalid."),
+ }, nil
+ }
+
+ instance := in.GetInstance()
+ remoteIP := util.GetIPFromContext(ctx)
+ instanceFlag := util.StringJoin([]string{instance.ServiceId, instance.HostName}, "/")
+
+ if err := apt.Validate(instance); err != nil {
+ util.Logger().Errorf(err, "register instance failed, service %s, operator %s.", instanceFlag, remoteIP)
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponse(scerr.ErrInvalidParams, err.Error()),
+ }, nil
+ }
+ //允许自定义id
+ //如果没填写 并且endpoints沒重復，則产生新的全局instance id
+ oldInstanceId, checkErr := serviceUtil.CheckEndPoints(ctx, in.Instance)
+ if checkErr != nil {
+ util.Logger().Errorf(checkErr, "check endpoints index failed, service %s, operator %s.",
+ instanceFlag, remoteIP)
+ resp := pb.CreateResponseWithSCErr(checkErr)
+ if checkErr.InternalError() {
+ return &pb.RegisterInstanceResponse{Response: resp}, checkErr
+ }
+ return &pb.RegisterInstanceResponse{Response: resp}, nil
+ }
+ if len(oldInstanceId) > 0 {
+ util.Logger().Infof("register instance successful, reuse service %s instance %s, operator %s.",
+ instance.ServiceId, oldInstanceId, remoteIP)
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponse(pb.Response_SUCCESS, "instance more exist."),
+ InstanceId: oldInstanceId,
+ }, nil
+ }
+
+ if err := s.preProcessRegisterInstance(ctx, instance); err != nil {
+ util.Logger().Errorf(err, "register instance failed, service %s, operator %s.", instanceFlag, remoteIP)
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponseWithSCErr(err),
+ }, nil
+ }
+
+ //先以domain/project的方式组装
+ domainProject := util.ParseDomainProject(ctx)
+
+ var reporter quota.QuotaReporter
+ if !apt.IsSCInstance(ctx) {
+ res := quota.NewApplyQuotaResource(quota.MicroServiceInstanceQuotaType,
+ domainProject, in.Instance.ServiceId, 1)
+ rst := plugin.Plugins().Quota().Apply4Quotas(ctx, res)
+ reporter = rst.Reporter
+ err := rst.Err
+ if reporter != nil {
+ defer reporter.Close()
+ }
+ if err != nil {
+ util.Logger().Errorf(err, "register instance failed, service %s, operator %s: no quota apply.",
+ instanceFlag, remoteIP)
+ response := &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponseWithSCErr(err),
+ }
+ if err.InternalError() {
+ return response, err
+ }
+ return response, nil
+ }
+ }
+
+ instanceId := instance.InstanceId
data, err := json.Marshal(instance)
if err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, instanceId %s, operator %s: json marshal data failed.",
+ util.Logger().Errorf(err,
+ "register instance failed, service %s, instanceId %s, operator %s: json marshal data failed.",
instanceFlag, instanceId, remoteIP)
return &pb.RegisterInstanceResponse{
Response: pb.CreateResponse(scerr.ErrInternal, "Instance file marshal error."),
}, err
}
- leaseID, err := grantOrRenewLease(ctx, domainProject, instance.ServiceId, instanceId, ttl)
+ ttl := int64(instance.HealthCheck.Interval * (instance.HealthCheck.Times + 1))
+ leaseID, err := backend.Registry().LeaseGrant(ctx, ttl)
if err != nil {
+ util.Logger().Errorf(err, "grant lease failed, instance %s, operator: %s.", instanceFlag, remoteIP)
return &pb.RegisterInstanceResponse{
Response: pb.CreateResponse(scerr.ErrInternal, "Lease grant or renew failed."),
}, err
}
+ util.Logger().Infof("lease grant %ds successfully, instance %s, operator: %s.", ttl, instanceFlag, remoteIP)
+ // build the request options
key := apt.GenerateInstanceKey(domainProject, instance.ServiceId, instanceId)
hbKey := apt.GenerateInstanceLeaseKey(domainProject, instance.ServiceId, instanceId)
-
- util.Logger().Debugf("start register service instance: %s %v, lease: %s %ds", key, instance, hbKey, ttl)
+ cmpBytes := util.StringToBytesWithNoCopy(apt.GenerateEndpointsIndexKey(domainProject, instance))
opts := []registry.PluginOp{
registry.OpPut(registry.WithStrKey(key), registry.WithValue(data),
- registry.WithLease(leaseID), registry.WithIgnoreLease()),
+ registry.WithLease(leaseID)),
+ registry.OpPut(registry.WithStrKey(hbKey), registry.WithStrValue(fmt.Sprintf("%d", leaseID)),
+ registry.WithLease(leaseID)),
+ registry.OpPut(registry.WithKey(cmpBytes), registry.WithStrValue(instance.ServiceId+"/"+instanceId),
+ registry.WithLease(leaseID)),
}
- if leaseID != 0 {
- opts = append(opts,
- registry.OpPut(registry.WithStrKey(hbKey), registry.WithStrValue(fmt.Sprintf("%d", leaseID)),
- registry.WithLease(leaseID), registry.WithIgnoreLease()))
- }
-
- if endpointsIndexKey != "" {
- value := util.StringJoin([]string{
- instance.ServiceId,
- instanceId,
- }, "/")
- opts = append(opts, registry.OpPut(registry.WithStrKey(endpointsIndexKey),
- registry.WithStrValue(value),
- registry.WithLease(leaseID), registry.WithIgnoreLease()))
+ uniqueCmpOpts := []registry.CompareOp{
+ registry.OpCmp(registry.CmpVer(cmpBytes), registry.CMP_EQUAL, 0),
}
- // Set key file
- _, err = backend.Registry().Txn(ctx, opts)
+ resp, err := backend.Registry().TxnWithCmp(ctx, opts, uniqueCmpOpts, nil)
if err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, instanceId %s, operator %s: commit data into etcd failed.",
+ util.Logger().Errorf(err,
+ "register instance failed, service %s, instanceId %s, operator %s: commit data into etcd failed.",
instanceFlag, instanceId, remoteIP)
return &pb.RegisterInstanceResponse{
Response: pb.CreateResponse(scerr.ErrUnavailableBackend, "Commit operations failed."),
}, err
}
+ if !resp.Succeeded {
+ // revoke the unused lease
+ defer backend.Registry().LeaseRevoke(ctx, leaseID)
+
+ oldInstanceId, checkErr := serviceUtil.CheckEndPoints(ctx, in.Instance)
+ if checkErr != nil {
+ util.Logger().Errorf(checkErr, "register instance failed, service %s, operator %s.",
+ instanceFlag, remoteIP)
+ resp := pb.CreateResponseWithSCErr(checkErr)
+ if checkErr.InternalError() {
+ return &pb.RegisterInstanceResponse{Response: resp}, checkErr
+ }
+ return &pb.RegisterInstanceResponse{Response: resp}, nil
+ }
+ if len(oldInstanceId) == 0 {
+ // re-check and found the older lease was revoked
+ util.Logger().Errorf(errors.New("instance is unregistered at the same time"),
+ "register instance failed, service %s, operator %s.", instanceFlag, remoteIP)
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponse(scerr.ErrInvalidParams, "Instance is unregistered at the same time"),
+ }, nil
+ }
+ util.Logger().Warnf(errors.New("instance was registered by others"),
+ "register instance successful, service %s instance %s, operator %s.",
+ instance.ServiceId, oldInstanceId, remoteIP)
+ return &pb.RegisterInstanceResponse{
+ Response: pb.CreateResponse(pb.Response_SUCCESS, "instance more exist."),
+ InstanceId: oldInstanceId,
+ }, nil
+ }
if reporter != nil {
if err := reporter.ReportUsedQuota(ctx); err != nil {
- util.Logger().Errorf(err, "register instance failed, service %s, instanceId %s, operator %s: report used quota failed.",
+ util.Logger().Errorf(err,
+ "register instance failed, service %s, instanceId %s, operator %s: report used quota failed.",
instanceFlag, instanceId, remoteIP)
}
}
- util.Logger().Infof("register instance successful service %s, instanceId %s, operator %s.", instanceFlag, instanceId, remoteIP)
+ util.Logger().Infof("register instance successful service %s, instanceId %s, operator %s.",
+ instanceFlag, instanceId, remoteIP)
return &pb.RegisterInstanceResponse{
Response: pb.CreateResponse(pb.Response_SUCCESS, "Register service instance successfully."),
InstanceId: instanceId,
@@ -319,42 +337,18 @@ func (s *InstanceService) Heartbeat(ctx context.Context, in *pb.HeartbeatRequest
Response: pb.CreateResponse(scerr.ErrInstanceNotExists, "Service instance does not exist."),
}, nil
}
- util.Logger().Infof("heartbeat successful: %s renew ttl to %d. operator: %s", instanceFlag, ttl, remoteIP)
+
+ if ttl == 0 {
+ util.Logger().Warnf(errors.New("connect backend timed out"),
+ "heartbeat successful, but renew %s failed. operator: %s", instanceFlag, remoteIP)
+ } else {
+ util.Logger().Infof("heartbeat successful: %s renew ttl to %d. operator: %s", instanceFlag, ttl, remoteIP)
+ }
return &pb.HeartbeatResponse{
Response: pb.CreateResponse(pb.Response_SUCCESS, "Update service instance heartbeat successfully."),
}, nil
}
-func grantOrRenewLease(ctx context.Context, domainProject string, serviceId string, instanceId string, ttl int64) (leaseID int64, err error) {
- remoteIP := util.GetIPFromContext(ctx)
- instanceFlag := util.StringJoin([]string{serviceId, instanceId}, "/")
-
- var (
- oldTTL int64
- inner bool
- )
-
- leaseID, oldTTL, err, inner = serviceUtil.HeartbeatUtil(ctx, domainProject, serviceId, instanceId)
- if inner {
- util.Logger().Errorf(err, "grant or renew lease failed, instance %s, operator: %s",
- instanceFlag, remoteIP)
- return
- }
-
- if leaseID < 0 || (oldTTL > 0 && oldTTL != ttl) {
- leaseID, err = backend.Registry().LeaseGrant(ctx, ttl)
- if err != nil {
- util.Logger().Errorf(err, "grant or renew lease failed, instance %s, operator: %s: lease grant failed.",
- instanceFlag, remoteIP)
- return
- }
- util.Logger().Infof("lease grant %d->%d successfully, instance %s, operator: %s.",
- oldTTL, ttl, instanceFlag, remoteIP)
- return
- }
- return
-}
-
func (s *InstanceService) HeartbeatSet(ctx context.Context, in *pb.HeartbeatSetRequest) (*pb.HeartbeatSetResponse, error) {
if in == nil || len(in.Instances) == 0 {
util.Logger().Errorf(nil, "heartbeats failed, invalid request. Body not contain Instances or is empty.")
diff --git a/server/service/instances_test.go b/server/service/instances_test.go
index abbf5fc7..18e2da12 100644
--- a/server/service/instances_test.go
+++ b/server/service/instances_test.go
@@ -206,8 +206,11 @@ var _ = Describe("'Instance' service", func() {
resp, err = instanceResource.Register(getContext(), &pb.RegisterInstanceRequest{
Instance: &pb.MicroServiceInstance{
ServiceId: serviceId1,
- HostName: "UT-HOST",
- Status: pb.MSI_UP,
+ Endpoints: []string{
+ "checkpull:127.0.0.1:8080",
+ },
+ HostName: "UT-HOST",
+ Status: pb.MSI_UP,
HealthCheck: &pb.HealthCheck{
Mode: "push",
Interval: 30,
@@ -222,8 +225,11 @@ var _ = Describe("'Instance' service", func() {
resp, err = instanceResource.Register(getContext(), &pb.RegisterInstanceRequest{
Instance: &pb.MicroServiceInstance{
ServiceId: serviceId1,
- HostName: "UT-HOST",
- Status: pb.MSI_UP,
+ Endpoints: []string{
+ "checkpush:127.0.0.1:8080",
+ },
+ HostName: "UT-HOST",
+ Status: pb.MSI_UP,
HealthCheck: &pb.HealthCheck{
Mode: "pull",
Interval: 30,
@@ -239,8 +245,11 @@ var _ = Describe("'Instance' service", func() {
resp, err = instanceResource.Register(getContext(), &pb.RegisterInstanceRequest{
Instance: &pb.MicroServiceInstance{
ServiceId: serviceId1,
- HostName: "UT-HOST",
- Status: pb.MSI_UP,
+ Endpoints: []string{
+ "checkpush:127.0.0.1:8081",
+ },
+ HostName: "UT-HOST",
+ Status: pb.MSI_UP,
HealthCheck: &pb.HealthCheck{
Mode: "push",
Interval: 30,
@@ -255,8 +264,11 @@ var _ = Describe("'Instance' service", func() {
resp, err = instanceResource.Register(getContext(), &pb.RegisterInstanceRequest{
Instance: &pb.MicroServiceInstance{
ServiceId: serviceId1,
- HostName: "UT-HOST",
- Status: pb.MSI_UP,
+ Endpoints: []string{
+ "checkpull:127.0.0.1:8081",
+ },
+ HostName: "UT-HOST",
+ Status: pb.MSI_UP,
HealthCheck: &pb.HealthCheck{
Mode: "pull",
Interval: 30,
diff --git a/server/service/util/instance_util.go b/server/service/util/instance_util.go
index e40fdaed..9ace072d 100644
--- a/server/service/util/instance_util.go
+++ b/server/service/util/instance_util.go
@@ -24,16 +24,14 @@ import (
"github.com/apache/incubator-servicecomb-service-center/server/core/backend"
"github.com/apache/incubator-servicecomb-service-center/server/core/backend/store"
pb "github.com/apache/incubator-servicecomb-service-center/server/core/proto"
+ scerr "github.com/apache/incubator-servicecomb-service-center/server/error"
"github.com/apache/incubator-servicecomb-service-center/server/infra/registry"
"github.com/coreos/etcd/mvcc/mvccpb"
"golang.org/x/net/context"
- "sort"
"strconv"
"strings"
)
-const NODEIP = "nodeIP"
-
func GetLeaseId(ctx context.Context, domainProject string, serviceId string, instanceId string) (int64, error) {
opts := append(FromContext(ctx),
registry.WithStrKey(apt.GenerateInstanceLeaseKey(domainProject, serviceId, instanceId)))
@@ -118,39 +116,32 @@ func InstanceExist(ctx context.Context, domainProject string, serviceId string,
return true, nil
}
-func CheckEndPoints(ctx context.Context, in *pb.RegisterInstanceRequest) (string, string, error) {
+func CheckEndPoints(ctx context.Context, instance *pb.MicroServiceInstance) (string, *scerr.Error) {
domainProject := util.ParseDomainProject(ctx)
- endpoints := in.Instance.Endpoints
- sort.Strings(endpoints)
- endpointsJoin := util.StringJoin(endpoints, "/")
- region, availableZone := apt.GetRegionAndAvailableZone(in.Instance.DataCenterInfo)
- nodeIP := ""
- if value, ok := in.Instance.Properties[NODEIP]; ok {
- nodeIP = value
- }
- instanceEndpointsIndexKey := apt.GenerateEndpointsIndexKey(domainProject, region, availableZone, nodeIP, endpointsJoin)
resp, err := store.Store().Endpoints().Search(ctx,
- registry.WithStrKey(instanceEndpointsIndexKey))
+ registry.WithStrKey(apt.GenerateEndpointsIndexKey(domainProject, instance)))
if err != nil {
- return "", "", err
+ return "", scerr.NewError(scerr.ErrInternal, err.Error())
}
if resp.Count == 0 {
- return "", instanceEndpointsIndexKey, nil
+ return "", nil
}
- endpointValue := ParseEndpointValue(resp.Kvs[0].Value)
- if in.Instance.ServiceId != endpointValue.serviceId {
- return endpointValue.instanceId, "", fmt.Errorf("Find the same endpoints in service %s", endpointValue.serviceId)
+ endpointValue := ParseEndpointIndexValue(resp.Kvs[0].Value)
+ if instance.ServiceId != endpointValue.serviceId {
+ return endpointValue.instanceId,
+ scerr.NewError(scerr.ErrEndpointAlreadyExists,
+ fmt.Sprintf("Find the same endpoints in service %s", endpointValue.serviceId))
}
- return endpointValue.instanceId, "", nil
+ return endpointValue.instanceId, nil
}
-type EndpointValue struct {
+type EndpointIndexValue struct {
serviceId string
instanceId string
}
-func ParseEndpointValue(value []byte) EndpointValue {
- endpointValue := EndpointValue{}
+func ParseEndpointIndexValue(value []byte) EndpointIndexValue {
+ endpointValue := EndpointIndexValue{}
tmp := util.BytesToStringWithNoCopy(value)
splitedTmp := strings.Split(tmp, "/")
endpointValue.serviceId = splitedTmp[0]
diff --git a/server/service/util/instance_util_test.go b/server/service/util/instance_util_test.go
index 5da480c8..cfe3224f 100644
--- a/server/service/util/instance_util_test.go
+++ b/server/service/util/instance_util_test.go
@@ -88,10 +88,8 @@ func TestInstanceExist(t *testing.T) {
}
func TestCheckEndPoints(t *testing.T) {
- _, _, err := CheckEndPoints(context.Background(), &proto.RegisterInstanceRequest{
- Instance: &proto.MicroServiceInstance{
- ServiceId: "a",
- },
+ _, err := CheckEndPoints(context.Background(), &proto.MicroServiceInstance{
+ ServiceId: "a",
})
if err == nil {
fmt.Printf(`CheckEndPoints failed`)
@@ -108,9 +106,17 @@ func TestDeleteServiceAllInstances(t *testing.T) {
}
func TestParseEndpointValue(t *testing.T) {
- epv := ParseEndpointValue([]byte("x/y"))
+ epv := ParseEndpointIndexValue([]byte("x/y"))
if epv.serviceId != "x" || epv.instanceId != "y" {
- fmt.Printf(`ParseEndpointValue failed`)
+ fmt.Printf(`ParseEndpointIndexValue failed`)
+ t.FailNow()
+ }
+}
+
+func TestGetInstanceCountOfOneService(t *testing.T) {
+ _, err := GetInstanceCountOfOneService(context.Background(), "", "")
+ if err == nil {
+ fmt.Printf(`GetInstanceCountOfOneService failed`)
t.FailNow()
}
}
----------------------------------------------------------------
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