This is an automated email from the ASF dual-hosted git repository.
littlecui 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 3147e0b [SCB-2094] Move the service/util package (#727)
3147e0b is described below
commit 3147e0b600f59e51c685d99e8aa335e246fd1047
Author: robotLJW <[email protected]>
AuthorDate: Fri Oct 23 14:47:31 2020 +0800
[SCB-2094] Move the service/util package (#727)
1. move the service/util package to the datasource/etcd/util directory
2. adjust the reference to service/util package
---
datasource/etcd/account.go | 2 +-
datasource/etcd/dep.go | 2 +-
datasource/etcd/ms.go | 5 +-
datasource/etcd/util.go | 2 +-
datasource/etcd/util/dependency.go | 132 ++++++
datasource/etcd/util/dependency_query.go | 366 +++++++++++++++
datasource/etcd/util/dependency_test.go | 294 ++++++++++++
datasource/etcd/util/dependency_util.go | 514 +++++++++++++++++++++
datasource/etcd/util/domain_util.go | 127 +++++
datasource/etcd/util/domain_util_test.go | 84 ++++
datasource/etcd/util/heartbeat_util.go | 53 +++
.../etcd/util/heartbeat_util_test.go | 25 +-
datasource/etcd/util/instance_util.go | 276 +++++++++++
datasource/etcd/util/instance_util_test.go | 170 +++++++
datasource/etcd/util/microservice_util.go | 310 +++++++++++++
datasource/etcd/util/rule_util.go | 300 ++++++++++++
datasource/etcd/util/rule_util_test.go | 306 ++++++++++++
.../etcd/util/schema_util.go | 30 +-
.../etcd/util/schema_util_test.go | 13 +-
datasource/etcd/util/tag_util.go | 67 +++
.../etcd/util/tag_util_test.go | 18 +-
.../tracing.go => datasource/etcd/util/util.go | 31 +-
datasource/etcd/util/util_suite_test.go | 140 ++++++
datasource/etcd/util/versionrule.go | 264 +++++++++++
datasource/etcd/util/versionrule_test.go | 413 +++++++++++++++++
pkg/chain/chain_test.go | 16 +-
pkg/chain/invocation.go | 2 +-
pkg/rest/common.go | 10 +-
pkg/util/common.go | 14 +-
pkg/util/context.go | 10 +-
pkg/util/context_grpc.go | 4 +-
server/handler/cache/cache.go | 12 +-
server/plugin/tracing/pzipkin/buildin.go | 2 +-
server/plugin/tracing/tracing.go | 3 +-
server/plugin/uuid/uuid.go | 4 +-
server/rest/handler.go | 2 +-
36 files changed, 3934 insertions(+), 89 deletions(-)
diff --git a/datasource/etcd/account.go b/datasource/etcd/account.go
index 68fbada..486b6b4 100644
--- a/datasource/etcd/account.go
+++ b/datasource/etcd/account.go
@@ -18,11 +18,11 @@ package etcd
import (
"context"
"encoding/json"
+ utils
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/rbacframe"
"github.com/apache/servicecomb-service-center/server/core/backend"
"github.com/apache/servicecomb-service-center/server/plugin/registry"
- utils "github.com/apache/servicecomb-service-center/server/service/util"
)
func (ds *DataSource) AccountExist(ctx context.Context, key string) (bool,
error) {
diff --git a/datasource/etcd/dep.go b/datasource/etcd/dep.go
index 5e04fd4..36c6666 100644
--- a/datasource/etcd/dep.go
+++ b/datasource/etcd/dep.go
@@ -21,6 +21,7 @@ import (
"context"
"encoding/json"
"fmt"
+ serviceUtil
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
"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"
@@ -29,7 +30,6 @@ import (
"github.com/apache/servicecomb-service-center/server/core/proto"
"github.com/apache/servicecomb-service-center/server/plugin/registry"
scerr "github.com/apache/servicecomb-service-center/server/scerror"
- serviceUtil
"github.com/apache/servicecomb-service-center/server/service/util"
)
func (ds *DataSource) SearchProviderDependency(ctx context.Context, request
*pb.GetDependenciesRequest) ([]*pb.MicroService, error) {
diff --git a/datasource/etcd/ms.go b/datasource/etcd/ms.go
index 7f26e45..b7d29f3 100644
--- a/datasource/etcd/ms.go
+++ b/datasource/etcd/ms.go
@@ -22,6 +22,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ serviceUtil
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
pb "github.com/apache/servicecomb-service-center/pkg/registry"
@@ -35,7 +36,6 @@ import (
"github.com/apache/servicecomb-service-center/server/plugin/registry"
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"
@@ -928,8 +928,7 @@ func (ds *DataSource) reshapeProviderKey(ctx
context.Context, provider *pb.Micro
return provider, nil
}
-func (ds *DataSource) UpdateInstanceStatus(ctx context.Context, request
*pb.UpdateInstanceStatusRequest) (*pb.
- UpdateInstanceStatusResponse, error) {
+func (ds *DataSource) UpdateInstanceStatus(ctx context.Context, request
*pb.UpdateInstanceStatusRequest) (*pb.UpdateInstanceStatusResponse, error) {
domainProject := util.ParseDomainProject(ctx)
updateStatusFlag := util.StringJoin([]string{request.ServiceId,
request.InstanceId, request.Status}, "/")
diff --git a/datasource/etcd/util.go b/datasource/etcd/util.go
index a9f1c3e..d4eb76e 100644
--- a/datasource/etcd/util.go
+++ b/datasource/etcd/util.go
@@ -20,6 +20,7 @@ package etcd
import (
"context"
"encoding/json"
+ serviceUtil
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
errorsEx "github.com/apache/servicecomb-service-center/pkg/errors"
"github.com/apache/servicecomb-service-center/pkg/gopool"
"github.com/apache/servicecomb-service-center/pkg/log"
@@ -32,7 +33,6 @@ import (
"github.com/apache/servicecomb-service-center/server/plugin/registry"
"github.com/apache/servicecomb-service-center/server/plugin/uuid"
scerr "github.com/apache/servicecomb-service-center/server/scerror"
- serviceUtil
"github.com/apache/servicecomb-service-center/server/service/util"
"strconv"
"strings"
"time"
diff --git a/datasource/etcd/util/dependency.go
b/datasource/etcd/util/dependency.go
new file mode 100644
index 0000000..0b13d99
--- /dev/null
+++ b/datasource/etcd/util/dependency.go
@@ -0,0 +1,132 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "encoding/json"
+
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ rmodel "github.com/apache/servicecomb-service-center/pkg/registry"
+ 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/plugin/registry"
+)
+
+// Dependency contains dependency rules
+type Dependency struct {
+ DomainProject string
+ // store the consumer Dependency from dep-queue object
+ Consumer *rmodel.MicroServiceKey
+ ProvidersRule []*rmodel.MicroServiceKey
+ // store the parsed rules from Dependency object
+ DeleteDependencyRuleList []*rmodel.MicroServiceKey
+ CreateDependencyRuleList []*rmodel.MicroServiceKey
+}
+
+func (dep *Dependency) removeConsumerOfProviderRule(ctx context.Context)
([]registry.PluginOp, error) {
+ opts := make([]registry.PluginOp, 0, len(dep.DeleteDependencyRuleList))
+ for _, providerRule := range dep.DeleteDependencyRuleList {
+ proProkey :=
apt.GenerateProviderDependencyRuleKey(providerRule.Tenant, providerRule)
+ log.Debugf("This proProkey is %s", proProkey)
+ consumerValue, err := TransferToMicroServiceDependency(ctx,
proProkey)
+ if err != nil {
+ return nil, err
+ }
+ for key, tmp := range consumerValue.Dependency {
+ if ok := equalServiceDependency(tmp, dep.Consumer); ok {
+ consumerValue.Dependency =
append(consumerValue.Dependency[:key], consumerValue.Dependency[key+1:]...)
+ break
+ }
+ log.Debugf("tmp and dep.Consumer not equal, tmp %v,
consumer %v", tmp, dep.Consumer)
+ }
+ //删除后，如果不存在依赖规则了，就删除该provider的依赖规则，如果有，则更新该依赖规则
+ if len(consumerValue.Dependency) == 0 {
+ opts = append(opts,
registry.OpDel(registry.WithStrKey(proProkey)))
+ continue
+ }
+ data, err := json.Marshal(consumerValue)
+ if err != nil {
+ log.Errorf(err, "Marshal MicroServiceDependency failed")
+ return nil, err
+ }
+ opts = append(opts, registry.OpPut(
+ registry.WithStrKey(proProkey),
+ registry.WithValue(data)))
+ }
+ return opts, nil
+}
+
+func (dep *Dependency) addConsumerOfProviderRule(ctx context.Context)
([]registry.PluginOp, error) {
+ opts := make([]registry.PluginOp, 0, len(dep.CreateDependencyRuleList))
+ for _, providerRule := range dep.CreateDependencyRuleList {
+ proProkey :=
apt.GenerateProviderDependencyRuleKey(providerRule.Tenant, providerRule)
+ tmpValue, err := TransferToMicroServiceDependency(ctx,
proProkey)
+ if err != nil {
+ return nil, err
+ }
+ tmpValue.Dependency = append(tmpValue.Dependency, dep.Consumer)
+
+ data, errMarshal := json.Marshal(tmpValue)
+ if errMarshal != nil {
+ log.Errorf(errMarshal, "Marshal MicroServiceDependency
failed")
+ return nil, errMarshal
+ }
+ opts = append(opts, registry.OpPut(
+ registry.WithStrKey(proProkey),
+ registry.WithValue(data)))
+ if providerRule.ServiceName == "*" {
+ break
+ }
+ }
+ return opts, nil
+}
+
+func (dep *Dependency) updateProvidersRuleOfConsumer(_ context.Context)
([]registry.PluginOp, error) {
+ conKey := apt.GenerateConsumerDependencyRuleKey(dep.DomainProject,
dep.Consumer)
+ if len(dep.ProvidersRule) == 0 {
+ return
[]registry.PluginOp{registry.OpDel(registry.WithStrKey(conKey))}, nil
+ }
+
+ dependency := &rmodel.MicroServiceDependency{
+ Dependency: dep.ProvidersRule,
+ }
+ data, err := json.Marshal(dependency)
+ if err != nil {
+ log.Errorf(err, "Marshal MicroServiceDependency failed")
+ return nil, err
+ }
+ return []registry.PluginOp{registry.OpPut(registry.WithStrKey(conKey),
registry.WithValue(data))}, nil
+}
+
+// Commit is dependent rule operations
+func (dep *Dependency) Commit(ctx context.Context) error {
+ dopts, err := dep.removeConsumerOfProviderRule(ctx)
+ if err != nil {
+ return err
+ }
+ copts, err := dep.addConsumerOfProviderRule(ctx)
+ if err != nil {
+ return err
+ }
+ uopts, err := dep.updateProvidersRuleOfConsumer(ctx)
+ if err != nil {
+ return err
+ }
+ return backend.BatchCommit(ctx, append(append(dopts, copts...),
uopts...))
+}
diff --git a/datasource/etcd/util/dependency_query.go
b/datasource/etcd/util/dependency_query.go
new file mode 100644
index 0000000..0bdf296
--- /dev/null
+++ b/datasource/etcd/util/dependency_query.go
@@ -0,0 +1,366 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "fmt"
+ "strings"
+
+ "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/registry"
+)
+
+// DependencyRelationFilterOpt contains SameDomainProject and NonSelf flag
+type DependencyRelationFilterOpt struct {
+ SameDomainProject bool
+ NonSelf bool
+}
+
+type DependencyRelationFilterOption func(opt DependencyRelationFilterOpt)
DependencyRelationFilterOpt
+
+func WithSameDomainProject() DependencyRelationFilterOption {
+ return func(opt DependencyRelationFilterOpt)
DependencyRelationFilterOpt {
+ opt.SameDomainProject = true
+ return opt
+ }
+}
+func WithoutSelfDependency() DependencyRelationFilterOption {
+ return func(opt DependencyRelationFilterOpt)
DependencyRelationFilterOpt {
+ opt.NonSelf = true
+ return opt
+ }
+}
+
+func toDependencyRelationFilterOpt(opts ...DependencyRelationFilterOption) (op
DependencyRelationFilterOpt) {
+ for _, opt := range opts {
+ op = opt(op)
+ }
+ return
+}
+
+type DependencyRelation struct {
+ ctx context.Context
+ domainProject string
+ consumer *pb.MicroService
+ provider *pb.MicroService
+}
+
+func (dr *DependencyRelation) GetDependencyProviders(opts
...DependencyRelationFilterOption) ([]*pb.MicroService, error) {
+ keys, err := dr.getProviderKeys()
+ if err != nil {
+ return nil, err
+ }
+ services := make([]*pb.MicroService, 0, len(keys))
+ op := toDependencyRelationFilterOpt(opts...)
+ for _, key := range keys {
+ if op.SameDomainProject && key.Tenant != dr.domainProject {
+ continue
+ }
+
+ providerIDs, err := dr.parseDependencyRule(key)
+ if err != nil {
+ return nil, err
+ }
+
+ if key.ServiceName == "*" {
+ services = services[:0]
+ }
+
+ for _, providerID := range providerIDs {
+ provider, err := GetService(dr.ctx, key.Tenant,
providerID)
+ if err != nil {
+ log.Warnf("get provider[%s/%s/%s/%s] failed",
+ key.Environment, key.AppId,
key.ServiceName, key.Version)
+ continue
+ }
+ if provider == nil {
+ log.Warnf("provider[%s/%s/%s/%s] does not
exist",
+ key.Environment, key.AppId,
key.ServiceName, key.Version)
+ continue
+ }
+ if op.NonSelf && providerID == dr.consumer.ServiceId {
+ continue
+ }
+ services = append(services, provider)
+ }
+
+ if key.ServiceName == "*" {
+ break
+ }
+ }
+ return services, nil
+}
+
+func (dr *DependencyRelation) GetDependencyProviderIds() ([]string, error) {
+ keys, err := dr.getProviderKeys()
+ if err != nil {
+ return nil, err
+ }
+ return dr.getDependencyProviderIds(keys)
+}
+
+func (dr *DependencyRelation) getProviderKeys() ([]*pb.MicroServiceKey, error)
{
+ if dr.consumer == nil {
+ return nil, fmt.Errorf("Invalid consumer")
+ }
+ consumerMicroServiceKey := proto.MicroServiceToKey(dr.domainProject,
dr.consumer)
+
+ conKey := apt.GenerateConsumerDependencyRuleKey(dr.domainProject,
consumerMicroServiceKey)
+ consumerDependency, err := TransferToMicroServiceDependency(dr.ctx,
conKey)
+ if err != nil {
+ return nil, err
+ }
+ return consumerDependency.Dependency, nil
+}
+
+func (dr *DependencyRelation) getDependencyProviderIds(providerRules
[]*pb.MicroServiceKey) ([]string, error) {
+ provideServiceIds := make([]string, 0, len(providerRules))
+ for _, provider := range providerRules {
+ serviceIDs, err := dr.parseDependencyRule(provider)
+ switch {
+ case provider.ServiceName == "*":
+ if err != nil {
+ log.Errorf(err, "get all serviceIDs failed")
+ return provideServiceIds, err
+ }
+ return serviceIDs, nil
+ default:
+ if err != nil {
+ log.Errorf(err, "get service[%s/%s/%s/%s]'s
providerIDs failed",
+ provider.Environment, provider.AppId,
provider.ServiceName, provider.Version)
+ return provideServiceIds, err
+ }
+ if len(serviceIDs) == 0 {
+ log.Warnf("get service[%s/%s/%s/%s]'s
providerIDs is empty",
+ provider.Environment, provider.AppId,
provider.ServiceName, provider.Version)
+ continue
+ }
+ provideServiceIds = append(provideServiceIds,
serviceIDs...)
+ }
+ }
+ return provideServiceIds, nil
+}
+
+func (dr *DependencyRelation) parseDependencyRule(dependencyRule
*pb.MicroServiceKey) (serviceIDs []string, err error) {
+ opts := FromContext(dr.ctx)
+ switch {
+ case dependencyRule.ServiceName == "*":
+ log.Infof("service[%s/%s/%s/%s] rely all service",
+ dr.consumer.Environment, dr.consumer.AppId,
dr.consumer.ServiceName, dr.consumer.Version)
+ splited :=
strings.Split(apt.GenerateServiceIndexKey(dependencyRule), "/")
+ allServiceKey := util.StringJoin(splited[:len(splited)-3], "/")
+ "/"
+ sopts := append(opts,
+ registry.WithStrKey(allServiceKey),
+ registry.WithPrefix())
+ resp, err := backend.Store().ServiceIndex().Search(dr.ctx,
sopts...)
+ if err != nil {
+ return nil, err
+ }
+
+ for _, kv := range resp.Kvs {
+ serviceIDs = append(serviceIDs, kv.Value.(string))
+ }
+ default:
+ serviceIDs, _, err = FindServiceIds(dr.ctx,
dependencyRule.Version, dependencyRule)
+ }
+ return
+}
+
+func (dr *DependencyRelation) GetDependencyConsumers(opts
...DependencyRelationFilterOption) ([]*pb.MicroService, error) {
+ consumerDependAllList, err := dr.getDependencyConsumersOfProvider()
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s consumers failed",
dr.provider.ServiceId)
+ return nil, err
+ }
+ consumers := make([]*pb.MicroService, 0)
+ op := toDependencyRelationFilterOpt(opts...)
+ for _, consumer := range consumerDependAllList {
+ if op.SameDomainProject && consumer.Tenant != dr.domainProject {
+ continue
+ }
+
+ service, err := dr.getServiceByMicroServiceKey(consumer)
+ if err != nil {
+ return nil, err
+ }
+ if service == nil {
+ log.Warnf("consumer[%s/%s/%s/%s] does not exist",
+ consumer.Environment, consumer.AppId,
consumer.ServiceName, consumer.Version)
+ continue
+ }
+
+ if op.NonSelf && service.ServiceId == dr.provider.ServiceId {
+ continue
+ }
+
+ consumers = append(consumers, service)
+ }
+ return consumers, nil
+}
+
+func (dr *DependencyRelation) getServiceByMicroServiceKey(service
*pb.MicroServiceKey) (*pb.MicroService, error) {
+ serviceID, err := GetServiceID(dr.ctx, service)
+ if err != nil {
+ return nil, err
+ }
+ if len(serviceID) == 0 {
+ log.Warnf("service[%s/%s/%s/%s] not exist",
+ service.Environment, service.AppId,
service.ServiceName, service.Version)
+ return nil, nil
+ }
+ return GetService(dr.ctx, service.Tenant, serviceID)
+}
+
+func (dr *DependencyRelation) GetDependencyConsumerIds() ([]string, error) {
+ consumerDependAllList, err := dr.getDependencyConsumersOfProvider()
+ if err != nil {
+ return nil, err
+ }
+ consumerIDs := make([]string, 0, len(consumerDependAllList))
+ for _, consumer := range consumerDependAllList {
+ consumerID, err := GetServiceID(dr.ctx, consumer)
+ if err != nil {
+ log.Errorf(err, "get consumer[%s/%s/%s/%s] failed",
+ consumer.Environment, consumer.AppId,
consumer.ServiceName, consumer.Version)
+ return nil, err
+ }
+ if len(consumerID) == 0 {
+ log.Warnf("get consumer[%s/%s/%s/%s] not exist",
+ consumer.Environment, consumer.AppId,
consumer.ServiceName, consumer.Version)
+ continue
+ }
+ consumerIDs = append(consumerIDs, consumerID)
+ }
+ return consumerIDs, nil
+
+}
+
+func (dr *DependencyRelation) getDependencyConsumersOfProvider()
([]*pb.MicroServiceKey, error) {
+ if dr.provider == nil {
+ return nil, fmt.Errorf("Invalid provider")
+ }
+ providerService := proto.MicroServiceToKey(dr.domainProject,
dr.provider)
+ consumerDependAllList, err := dr.getConsumerOfDependAllServices()
+ if err != nil {
+ log.Errorf(err, "get consumers that depend on all services
failed, %s", dr.provider.ServiceId)
+ return nil, err
+ }
+
+ consumerDependList, err :=
dr.getConsumerOfSameServiceNameAndAppID(providerService)
+ if err != nil {
+ log.Errorf(err, "get consumers that depend on rule[%s/%s/%s/%s]
failed",
+ dr.provider.Environment, dr.provider.AppId,
dr.provider.ServiceName, dr.provider.Version)
+ return nil, err
+ }
+ consumerDependAllList = append(consumerDependAllList,
consumerDependList...)
+ return consumerDependAllList, nil
+}
+
+func (dr *DependencyRelation) getConsumerOfDependAllServices()
([]*pb.MicroServiceKey, error) {
+ providerService := proto.MicroServiceToKey(dr.domainProject,
dr.provider)
+ providerService.ServiceName = "*"
+ relyAllKey := apt.GenerateProviderDependencyRuleKey(dr.domainProject,
providerService)
+ opts := append(FromContext(dr.ctx), registry.WithStrKey(relyAllKey))
+ rsp, err := backend.Store().DependencyRule().Search(dr.ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get consumers that rely all service failed,
%s/%s/%s/%s",
+ dr.provider.Environment, dr.provider.AppId,
dr.provider.ServiceName, dr.provider.Version)
+ return nil, err
+ }
+ if len(rsp.Kvs) != 0 {
+ log.Infof("exist consumers that rely all service, %s/%s/%s/%s",
+ dr.provider.Environment, dr.provider.AppId,
dr.provider.ServiceName, dr.provider.Version)
+ return
rsp.Kvs[0].Value.(*pb.MicroServiceDependency).Dependency, nil
+ }
+ return nil, nil
+}
+
+func (dr *DependencyRelation) getConsumerOfSameServiceNameAndAppID(provider
*pb.MicroServiceKey) ([]*pb.MicroServiceKey, error) {
+ providerVersion := provider.Version
+ provider.Version = ""
+ prefix := apt.GenerateProviderDependencyRuleKey(dr.domainProject,
provider)
+ provider.Version = providerVersion
+
+ opts := append(FromContext(dr.ctx),
+ registry.WithStrKey(prefix),
+ registry.WithPrefix())
+ rsp, err := backend.Store().DependencyRule().Search(dr.ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get service[%s/%s/%s]'s dependency rules
failed",
+ provider.Environment, provider.AppId,
provider.ServiceName)
+ return nil, err
+ }
+
+ allConsumers := make([]*pb.MicroServiceKey, 0, len(rsp.Kvs))
+ var latestServiceID []string
+
+ for _, kv := range rsp.Kvs {
+ providerVersionRuleArr :=
strings.Split(util.BytesToStringWithNoCopy(kv.Key), "/")
+ providerVersionRule :=
providerVersionRuleArr[len(providerVersionRuleArr)-1]
+ if providerVersionRule == "latest" {
+ if latestServiceID == nil {
+ latestServiceID, _, err =
FindServiceIds(dr.ctx, providerVersionRule, provider)
+ if err != nil {
+ log.Errorf(err, "get
service[%s/%s/%s/%s]'s serviceID failed",
+ provider.Environment,
provider.AppId, provider.ServiceName, providerVersionRule)
+ return nil, err
+ }
+ }
+ if len(latestServiceID) == 0 {
+ log.Infof("service[%s/%s/%s/%s] does not exist",
+ provider.Environment, provider.AppId,
provider.ServiceName, providerVersionRule)
+ continue
+ }
+ if dr.provider.ServiceId != latestServiceID[0] {
+ continue
+ }
+
+ } else {
+ if !VersionMatchRule(providerVersion,
providerVersionRule) {
+ continue
+ }
+ }
+
+ log.Debugf("providerETCD is %s", providerVersionRuleArr)
+ allConsumers = append(allConsumers,
kv.Value.(*pb.MicroServiceDependency).Dependency...)
+ }
+ return allConsumers, nil
+}
+
+func NewProviderDependencyRelation(ctx context.Context, domainProject string,
provider *pb.MicroService) *DependencyRelation {
+ return NewDependencyRelation(ctx, domainProject, nil, provider)
+}
+
+func NewConsumerDependencyRelation(ctx context.Context, domainProject string,
consumer *pb.MicroService) *DependencyRelation {
+ return NewDependencyRelation(ctx, domainProject, consumer, nil)
+}
+
+func NewDependencyRelation(ctx context.Context, domainProject string, consumer
*pb.MicroService, provider *pb.MicroService) *DependencyRelation {
+ return &DependencyRelation{
+ ctx: ctx,
+ domainProject: domainProject,
+ consumer: consumer,
+ provider: provider,
+ }
+}
diff --git a/datasource/etcd/util/dependency_test.go
b/datasource/etcd/util/dependency_test.go
new file mode 100644
index 0000000..acf50a0
--- /dev/null
+++ b/datasource/etcd/util/dependency_test.go
@@ -0,0 +1,294 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "testing"
+
+ "github.com/apache/servicecomb-service-center/pkg/registry"
+)
+
+func TestDeleteDependencyForService(t *testing.T) {
+ _, err := DeleteDependencyForDeleteService("", "",
®istry.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf(`DeleteDependencyForDeleteService failed`)
+ }
+}
+
+func TestTransferToMicroServiceDependency(t *testing.T) {
+ _, err := TransferToMicroServiceDependency(context.Background(), "")
+ if err == nil {
+ t.Fatalf(`TransferToMicroServiceDependency failed`)
+ }
+}
+
+func TestEqualServiceDependency(t *testing.T) {
+ b := equalServiceDependency(®istry.MicroServiceKey{},
®istry.MicroServiceKey{})
+ if !b {
+ t.Fatalf(`equalServiceDependency failed`)
+ }
+
+ b = equalServiceDependency(®istry.MicroServiceKey{
+ AppId: "a",
+ }, ®istry.MicroServiceKey{
+ AppId: "b",
+ })
+ if b {
+ t.Fatalf(`equalServiceDependency failed`)
+ }
+}
+
+func TestCreateDependencyRule(t *testing.T) {
+ err := CreateDependencyRule(context.Background(), &Dependency{
+ Consumer: ®istry.MicroServiceKey{},
+ })
+ if err != nil {
+ t.Fatalf(`CreateDependencyRule failed`)
+ }
+
+ err = AddDependencyRule(context.Background(), &Dependency{
+ Consumer: ®istry.MicroServiceKey{},
+ })
+ if err != nil {
+ t.Fatalf(`AddDependencyRule failed`)
+ }
+
+ err = AddServiceVersionRule(context.Background(), "",
®istry.MicroService{}, ®istry.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf(`AddServiceVersionRule failed`)
+ }
+
+ b, err := containServiceDependency([]*registry.MicroServiceKey{
+ {AppId: "a"},
+ }, ®istry.MicroServiceKey{
+ AppId: "b",
+ })
+ if b {
+ t.Fatalf(`containServiceDependency contain failed`)
+ }
+
+ b, err = containServiceDependency([]*registry.MicroServiceKey{
+ {AppId: "a"},
+ }, ®istry.MicroServiceKey{
+ AppId: "a",
+ })
+ if !b {
+ t.Fatalf(`containServiceDependency not contain failed`)
+ }
+
+ _, err = containServiceDependency(nil, nil)
+ if err == nil {
+ t.Fatalf(`containServiceDependency invalid failed`)
+ }
+
+ ok := diffServiceVersion(®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "a",
+ Version: "1",
+ }, ®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "a",
+ Version: "2",
+ })
+ if !ok {
+ t.Fatalf(`diffServiceVersion failed`)
+ }
+}
+
+func TestBadParamsResponse(t *testing.T) {
+ p := BadParamsResponse("a")
+ if p == nil {
+ t.Fatalf(`BadParamsResponse failed`)
+ }
+}
+
+func TestDependencyRuleExistUtil(t *testing.T) {
+ _, err := dependencyRuleExistUtil(context.Background(), "",
®istry.MicroServiceKey{})
+ if err == nil {
+ t.Fatalf(`dependencyRuleExistUtil failed`)
+ }
+}
+
+func TestParamsChecker(t *testing.T) {
+ p := ParamsChecker(nil, nil)
+ if p != nil {
+ t.Fatalf(`ParamsChecker invalid failed`)
+ }
+
+ p = ParamsChecker(®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "b",
+ Version: "1.0.0",
+ }, nil)
+ if p != nil {
+ t.Fatalf(`ParamsChecker invalid failed`)
+ }
+
+ p = ParamsChecker(®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "b",
+ Version: "1.0.0",
+ }, []*registry.MicroServiceKey{
+ {ServiceName: "*"},
+ })
+ if p != nil {
+ t.Fatalf(`ParamsChecker * failed`)
+ }
+
+ p = ParamsChecker(®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "b",
+ Version: "1.0.0",
+ }, []*registry.MicroServiceKey{
+ {},
+ })
+ if p == nil {
+ t.Fatalf(`ParamsChecker invalid provider key failed`)
+ }
+
+ p = ParamsChecker(®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "b",
+ Version: "1.0.0",
+ }, []*registry.MicroServiceKey{
+ {ServiceName: "a", Version: "1"},
+ {ServiceName: "a", Version: "1"},
+ })
+ if p == nil {
+ t.Fatalf(`ParamsChecker duplicate provider key failed`)
+ }
+}
+
+func TestServiceDependencyRuleExist(t *testing.T) {
+ _, err := DependencyRuleExist(context.Background(),
®istry.MicroServiceKey{}, ®istry.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf(`ServiceDependencyRuleExist failed`)
+ }
+}
+
+func TestUpdateServiceForAddDependency(t *testing.T) {
+ old := isNeedUpdate([]*registry.MicroServiceKey{
+ {
+ AppId: "a",
+ ServiceName: "a",
+ Version: "1",
+ },
+ }, ®istry.MicroServiceKey{
+ AppId: "a",
+ ServiceName: "a",
+ Version: "2",
+ })
+ if old == nil {
+ t.Fatalf(`isNeedUpdate failed`)
+ }
+}
+
+func TestDependency(t *testing.T) {
+ d := &Dependency{
+ DeleteDependencyRuleList: []*registry.MicroServiceKey{
+ {ServiceName: "a", Version: "1.0.0"},
+ },
+ CreateDependencyRuleList: []*registry.MicroServiceKey{
+ {ServiceName: "a", Version: "1.0.0"},
+ },
+ }
+ err := d.Commit(context.Background())
+ if err != nil {
+ t.Fatalf(`Dependency_UpdateProvidersRuleOfConsumer failed`)
+ }
+
+ dr := &DependencyRelation{
+ provider: ®istry.MicroService{},
+ consumer: ®istry.MicroService{},
+ ctx: context.Background(),
+ }
+ _, err = dr.getDependencyProviderIds([]*registry.MicroServiceKey{
+ {ServiceName: "*"},
+ })
+ if err != nil {
+ t.Fatalf(`DependencyRelation_getDependencyProviderIds * failed`)
+ }
+ _, err = dr.getDependencyProviderIds([]*registry.MicroServiceKey{
+ {ServiceName: "a", Version: "1.0.0"},
+ {ServiceName: "b", Version: "latest"},
+ })
+ if err != nil {
+ t.Fatalf(`DependencyRelation_getDependencyProviderIds failed`)
+ }
+
+ _, err = dr.GetDependencyConsumers()
+ if err != nil {
+ t.Fatalf(`DependencyRelation_GetDependencyConsumers failed`)
+ }
+
+ _, err = dr.getServiceByMicroServiceKey(®istry.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf(`DependencyRelation_getServiceByMicroServiceKey
failed`)
+ }
+
+ _, err =
dr.getConsumerOfSameServiceNameAndAppID(®istry.MicroServiceKey{})
+ if err != nil {
+
t.Fatalf(`DependencyRelation_getConsumerOfSameServiceNameAndAppId failed`)
+ }
+
+ dr = &DependencyRelation{
+ consumer: ®istry.MicroService{},
+ ctx: context.Background(),
+ }
+ _, err = dr.getDependencyConsumersOfProvider()
+ if err == nil {
+ t.Fatalf(`DependencyRelation_getDependencyConsumersOfProvider
failed`)
+ }
+ _, err = dr.GetDependencyProviders()
+ if err != nil {
+ t.Fatalf(`DependencyRelation_GetDependencyProviders failed`)
+ }
+
+ err = CleanUpDependencyRules(context.Background(), "")
+ if err == nil {
+ t.Fatalf(`DependencyRelation_CleanUpDependencyRules failed`)
+ }
+
+ err = CleanUpDependencyRules(context.Background(), "a/b")
+ if err != nil {
+ t.Fatalf(`DependencyRelation_CleanUpDependencyRules failed`)
+ }
+
+ _, err = removeProviderRuleKeys(context.Background(), "a/b", nil)
+ if err != nil {
+ t.Fatalf(`DependencyRelation_removeProviderRuleKeys failed`)
+ }
+}
+
+func TestDependencyRelationFilterOpt(t *testing.T) {
+ op := toDependencyRelationFilterOpt(
+ WithSameDomainProject(),
+ WithoutSelfDependency(),
+ )
+ if !op.NonSelf || !op.SameDomainProject {
+ t.Fatalf(`toDependencyRelationFilterOpt failed`)
+ }
+}
+
+func TestGetConsumerIdsWithFilter(t *testing.T) {
+ _, _, err := getConsumerIdsWithFilter(context.Background(), "",
®istry.MicroService{}, nil)
+ if err != nil {
+ t.Fatalf(`TestGetConsumerIdsWithFilter failed`)
+ }
+}
diff --git a/datasource/etcd/util/dependency_util.go
b/datasource/etcd/util/dependency_util.go
new file mode 100644
index 0000000..ac509d9
--- /dev/null
+++ b/datasource/etcd/util/dependency_util.go
@@ -0,0 +1,514 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "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/registry"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+ "strings"
+)
+
+func GetConsumerIds(ctx context.Context, domainProject string, provider
*pb.MicroService) ([]string, error) {
+ // 查询所有consumer
+ dr := NewProviderDependencyRelation(ctx, domainProject, provider)
+ consumerIds, err := dr.GetDependencyConsumerIds()
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s consumerIds failed",
provider.ServiceId)
+ return nil, err
+ }
+ return consumerIds, nil
+}
+
+func GetProviderIds(ctx context.Context, domainProject string, consumer
*pb.MicroService) ([]string, error) {
+ // 查询所有provider
+ dr := NewConsumerDependencyRelation(ctx, domainProject, consumer)
+ providerIDs, err := dr.GetDependencyProviderIds()
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s providerIDs failed",
consumer.ServiceId)
+ return nil, err
+ }
+ return providerIDs, nil
+}
+
+// GetAllConsumerIds is the function get from dependency rule and filter with
service rules
+func GetAllConsumerIds(ctx context.Context, domainProject string, provider
*pb.MicroService) (allow []string, deny []string, _ error) {
+ if provider == nil || len(provider.ServiceId) == 0 {
+ return nil, nil, fmt.Errorf("invalid provider")
+ }
+
+ //todo 删除服务，最后实例推送有误差
+ providerRules, err := GetRulesUtil(ctx, domainProject,
provider.ServiceId)
+ if err != nil {
+ return nil, nil, err
+ }
+
+ rf := &RuleFilter{
+ DomainProject: domainProject,
+ ProviderRules: providerRules,
+ }
+
+ allow, deny, err = getConsumerIdsWithFilter(ctx, domainProject,
provider, rf)
+ if err != nil {
+ return nil, nil, err
+ }
+ return allow, deny, nil
+}
+
+func getConsumerIdsWithFilter(ctx context.Context, domainProject string,
provider *pb.MicroService, rf *RuleFilter) (allow []string, deny []string, err
error) {
+ consumerIds, err := GetConsumerIds(ctx, domainProject, provider)
+ if err != nil {
+ return nil, nil, err
+ }
+ return rf.FilterAll(ctx, consumerIds)
+}
+
+func GetAllProviderIds(ctx context.Context, domainProject string, service
*pb.MicroService) (allow []string, deny []string, _ error) {
+ providerIDsInCache, err := GetProviderIds(ctx, domainProject, service)
+ if err != nil {
+ return nil, nil, err
+ }
+ l := len(providerIDsInCache)
+ rf := RuleFilter{
+ DomainProject: domainProject,
+ }
+ allowIdx, denyIdx := 0, l
+ providerIDs := make([]string, l)
+ copyCtx := util.SetContext(util.CloneContext(ctx), util.CtxCacheOnly,
"1")
+ for _, providerID := range providerIDsInCache {
+ providerRules, err := GetRulesUtil(copyCtx, domainProject,
providerID)
+ if err != nil {
+ return nil, nil, err
+ }
+ if len(providerRules) == 0 {
+ providerIDs[allowIdx] = providerID
+ allowIdx++
+ continue
+ }
+ rf.ProviderRules = providerRules
+ ok, err := rf.Filter(ctx, service.ServiceId)
+ if err != nil {
+ return nil, nil, err
+ }
+ if ok {
+ providerIDs[allowIdx] = providerID
+ allowIdx++
+ } else {
+ denyIdx--
+ providerIDs[denyIdx] = providerID
+ }
+ }
+ return providerIDs[:allowIdx], providerIDs[denyIdx:], nil
+}
+
+func DependencyRuleExist(ctx context.Context, provider *pb.MicroServiceKey,
consumer *pb.MicroServiceKey) (bool, error) {
+ targetDomainProject := provider.Tenant
+ if len(targetDomainProject) == 0 {
+ targetDomainProject = consumer.Tenant
+ }
+
+ consumerKey := apt.GenerateConsumerDependencyRuleKey(consumer.Tenant,
consumer)
+ existed, err := dependencyRuleExistUtil(ctx, consumerKey, provider)
+ if err != nil || existed {
+ return existed, err
+ }
+
+ providerKey :=
apt.GenerateProviderDependencyRuleKey(targetDomainProject, provider)
+ return dependencyRuleExistUtil(ctx, providerKey, consumer)
+}
+
+func dependencyRuleExistUtil(ctx context.Context, key string, target
*pb.MicroServiceKey) (bool, error) {
+ compareData, err := TransferToMicroServiceDependency(ctx, key)
+ if err != nil {
+ return false, err
+ }
+ if len(compareData.Dependency) != 0 {
+ isEqual, err :=
containServiceDependency(compareData.Dependency, target)
+ if err != nil {
+ return false, err
+ }
+ if isEqual {
+ //删除之前的依赖
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
+func AddServiceVersionRule(ctx context.Context, domainProject string, consumer
*pb.MicroService, provider *pb.MicroServiceKey) error {
+ //创建依赖一致
+ consumerKey := proto.MicroServiceToKey(domainProject, consumer)
+ exist, err := DependencyRuleExist(ctx, provider, consumerKey)
+ if exist || err != nil {
+ return err
+ }
+
+ r := &pb.ConsumerDependency{
+ Consumer: consumerKey,
+ Providers: []*pb.MicroServiceKey{provider},
+ Override: false,
+ }
+ data, err := json.Marshal(r)
+ if err != nil {
+ return err
+ }
+
+ id := util.StringJoin([]string{provider.AppId, provider.ServiceName},
"_")
+ key := apt.GenerateConsumerDependencyQueueKey(domainProject,
consumer.ServiceId, id)
+ resp, err := backend.Registry().TxnWithCmp(ctx,
+ nil,
+ []registry.CompareOp{registry.OpCmp(registry.CmpStrVal(key),
registry.CmpEqual, util.BytesToStringWithNoCopy(data))},
+ []registry.PluginOp{registry.OpPut(registry.WithStrKey(key),
registry.WithValue(data))})
+ if err != nil {
+ return err
+ }
+ if !resp.Succeeded {
+ log.Infof("put in queue[%s/%s]: consumer[%s/%s/%s/%s] ->
provider[%s/%s/%s/%s]", consumer.ServiceId, id,
+ consumer.Environment, consumer.AppId,
consumer.ServiceName, consumer.Version,
+ provider.Environment, provider.AppId,
provider.ServiceName, provider.Version)
+ }
+ return nil
+}
+
+func TransferToMicroServiceDependency(ctx context.Context, key string)
(*pb.MicroServiceDependency, error) {
+ microServiceDependency := &pb.MicroServiceDependency{
+ Dependency: []*pb.MicroServiceKey{},
+ }
+
+ opts := append(FromContext(ctx), registry.WithStrKey(key))
+ res, err := backend.Store().DependencyRule().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(nil, "get dependency rule[%s] failed", key)
+ return nil, err
+ }
+ if len(res.Kvs) != 0 {
+ return res.Kvs[0].Value.(*pb.MicroServiceDependency), nil
+ }
+ return microServiceDependency, nil
+}
+
+func equalServiceDependency(serviceA *pb.MicroServiceKey, serviceB
*pb.MicroServiceKey) bool {
+ stringA := toString(serviceA)
+ stringB := toString(serviceB)
+ return stringA == stringB
+}
+
+func diffServiceVersion(serviceA *pb.MicroServiceKey, serviceB
*pb.MicroServiceKey) bool {
+ stringA := toString(serviceA)
+ stringB := toString(serviceB)
+ if stringA != stringB &&
+ stringA[:strings.LastIndex(stringA, "/")+1] ==
stringB[:strings.LastIndex(stringB, "/")+1] {
+ return true
+ }
+ return false
+}
+
+func toString(in *pb.MicroServiceKey) string {
+ return apt.GenerateProviderDependencyRuleKey(in.Tenant, in)
+}
+
+func parseAddOrUpdateRules(ctx context.Context, dep *Dependency)
(createDependencyRuleList, existDependencyRuleList, deleteDependencyRuleList
[]*pb.MicroServiceKey) {
+ conKey := apt.GenerateConsumerDependencyRuleKey(dep.DomainProject,
dep.Consumer)
+
+ oldProviderRules, err := TransferToMicroServiceDependency(ctx, conKey)
+ if err != nil {
+ log.Errorf(err, "update dependency rule failed, get
consumer[%s/%s/%s/%s]'s dependency rule failed",
+ dep.Consumer.Environment, dep.Consumer.AppId,
dep.Consumer.ServiceName, dep.Consumer.Version)
+ return
+ }
+
+ deleteDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(oldProviderRules.Dependency))
+ createDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(dep.ProvidersRule))
+ existDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(oldProviderRules.Dependency))
+ for _, tmpProviderRule := range dep.ProvidersRule {
+ if ok, _ :=
containServiceDependency(oldProviderRules.Dependency, tmpProviderRule); ok {
+ continue
+ }
+
+ if tmpProviderRule.ServiceName == "*" {
+ createDependencyRuleList =
append([]*pb.MicroServiceKey{}, tmpProviderRule)
+ deleteDependencyRuleList = oldProviderRules.Dependency
+ break
+ }
+
+ createDependencyRuleList = append(createDependencyRuleList,
tmpProviderRule)
+ old := isNeedUpdate(oldProviderRules.Dependency,
tmpProviderRule)
+ if old != nil {
+ deleteDependencyRuleList =
append(deleteDependencyRuleList, old)
+ }
+ }
+ for _, oldProviderRule := range oldProviderRules.Dependency {
+ if oldProviderRule.ServiceName == "*" {
+ createDependencyRuleList = nil
+ deleteDependencyRuleList = nil
+ return
+ }
+ if ok, _ := containServiceDependency(deleteDependencyRuleList,
oldProviderRule); !ok {
+ existDependencyRuleList =
append(existDependencyRuleList, oldProviderRule)
+ }
+ }
+
+ dep.ProvidersRule = append(createDependencyRuleList,
existDependencyRuleList...)
+ return
+}
+
+func parseOverrideRules(ctx context.Context, dep *Dependency)
(createDependencyRuleList, existDependencyRuleList, deleteDependencyRuleList
[]*pb.MicroServiceKey) {
+ conKey := apt.GenerateConsumerDependencyRuleKey(dep.DomainProject,
dep.Consumer)
+
+ oldProviderRules, err := TransferToMicroServiceDependency(ctx, conKey)
+ if err != nil {
+ log.Errorf(err, "override dependency rule failed, get
consumer[%s/%s/%s/%s]'s dependency rule failed",
+ dep.Consumer.Environment, dep.Consumer.AppId,
dep.Consumer.ServiceName, dep.Consumer.Version)
+ return
+ }
+
+ deleteDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(oldProviderRules.Dependency))
+ createDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(dep.ProvidersRule))
+ existDependencyRuleList = make([]*pb.MicroServiceKey, 0,
len(oldProviderRules.Dependency))
+ for _, oldProviderRule := range oldProviderRules.Dependency {
+ if ok, _ := containServiceDependency(dep.ProvidersRule,
oldProviderRule); !ok {
+ deleteDependencyRuleList =
append(deleteDependencyRuleList, oldProviderRule)
+ } else {
+ existDependencyRuleList =
append(existDependencyRuleList, oldProviderRule)
+ }
+ }
+ for _, tmpProviderRule := range dep.ProvidersRule {
+ if ok, _ := containServiceDependency(existDependencyRuleList,
tmpProviderRule); !ok {
+ createDependencyRuleList =
append(createDependencyRuleList, tmpProviderRule)
+ }
+ }
+ return
+}
+
+func syncDependencyRule(ctx context.Context, dep *Dependency, filter
func(context.Context, *Dependency) (_, _, _ []*pb.MicroServiceKey)) error {
+ //更新consumer的providers的值,consumer的版本是确定的
+ consumerFlag := strings.Join([]string{dep.Consumer.Environment,
dep.Consumer.AppId, dep.Consumer.ServiceName, dep.Consumer.Version}, "/")
+
+ createDependencyRuleList, existDependencyRuleList,
deleteDependencyRuleList := filter(ctx, dep)
+ if len(createDependencyRuleList) == 0 && len(existDependencyRuleList)
== 0 && len(deleteDependencyRuleList) == 0 {
+ return nil
+ }
+
+ if len(deleteDependencyRuleList) != 0 {
+ log.Infof("delete consumer[%s]'s dependency rule %v",
consumerFlag, deleteDependencyRuleList)
+ dep.DeleteDependencyRuleList = deleteDependencyRuleList
+ }
+
+ if len(createDependencyRuleList) != 0 {
+ log.Infof("create consumer[%s]'s dependency rule %v",
consumerFlag, createDependencyRuleList)
+ dep.CreateDependencyRuleList = createDependencyRuleList
+ }
+
+ return dep.Commit(ctx)
+}
+
+func AddDependencyRule(ctx context.Context, dep *Dependency) error {
+ return syncDependencyRule(ctx, dep, parseAddOrUpdateRules)
+}
+
+func CreateDependencyRule(ctx context.Context, dep *Dependency) error {
+ return syncDependencyRule(ctx, dep, parseOverrideRules)
+}
+
+func isNeedUpdate(services []*pb.MicroServiceKey, service *pb.MicroServiceKey)
*pb.MicroServiceKey {
+ for _, tmp := range services {
+ if diffServiceVersion(tmp, service) {
+ return tmp
+ }
+ }
+ return nil
+}
+
+func containServiceDependency(services []*pb.MicroServiceKey, service
*pb.MicroServiceKey) (bool, error) {
+ if services == nil || service == nil {
+ return false, errors.New("invalid params input")
+ }
+ for _, value := range services {
+ rst := equalServiceDependency(service, value)
+ if rst {
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
+func BadParamsResponse(detailErr string) *pb.CreateDependenciesResponse {
+ log.Errorf(nil, "request params is invalid. %s", detailErr)
+ if len(detailErr) == 0 {
+ detailErr = "Request params is invalid."
+ }
+ return &pb.CreateDependenciesResponse{
+ Response: proto.CreateResponse(scerr.ErrInvalidParams,
detailErr),
+ }
+}
+
+func ParamsChecker(consumerInfo *pb.MicroServiceKey, providersInfo
[]*pb.MicroServiceKey) *pb.CreateDependenciesResponse {
+ flag := make(map[string]bool, len(providersInfo))
+ for _, providerInfo := range providersInfo {
+ //存在带*的情况，后面的数据就不校验了
+ if providerInfo.ServiceName == "*" {
+ break
+ }
+ if len(providerInfo.AppId) == 0 {
+ providerInfo.AppId = consumerInfo.AppId
+ }
+
+ version := providerInfo.Version
+ if len(version) == 0 {
+ return BadParamsResponse("Required provider version")
+ }
+
+ providerInfo.Version = ""
+ if _, ok := flag[toString(providerInfo)]; ok {
+ return BadParamsResponse("Invalid request body for
provider info.Duplicate provider or (serviceName and appId is same).")
+ }
+ flag[toString(providerInfo)] = true
+ providerInfo.Version = version
+ }
+ return nil
+}
+
+func DeleteDependencyForDeleteService(domainProject string, serviceID string,
service *pb.MicroServiceKey) (registry.PluginOp, error) {
+ key := apt.GenerateConsumerDependencyQueueKey(domainProject, serviceID,
apt.DepsQueueUUID)
+ conDep := new(pb.ConsumerDependency)
+ conDep.Consumer = service
+ conDep.Providers = []*pb.MicroServiceKey{}
+ conDep.Override = true
+ data, err := json.Marshal(conDep)
+ if err != nil {
+ return registry.PluginOp{}, err
+ }
+ return registry.OpPut(registry.WithStrKey(key),
registry.WithValue(data)), nil
+}
+
+func removeProviderRuleOfConsumer(ctx context.Context, domainProject string,
cache map[string]bool) ([]registry.PluginOp, error) {
+ key := apt.GenerateConsumerDependencyRuleKey(domainProject, nil) +
apt.SPLIT
+ resp, err := backend.Store().DependencyRule().Search(ctx,
+ registry.WithStrKey(key), registry.WithPrefix())
+ if err != nil {
+ return nil, err
+ }
+
+ var ops []registry.PluginOp
+loop:
+ for _, kv := range resp.Kvs {
+ var left []*pb.MicroServiceKey
+ all := kv.Value.(*pb.MicroServiceDependency).Dependency
+ for _, key := range all {
+ if key.ServiceName == "*" {
+ continue loop
+ }
+
+ id := apt.GenerateProviderDependencyRuleKey(key.Tenant,
key)
+ exist, ok := cache[id]
+ if !ok {
+ _, exist, err = FindServiceIds(ctx,
key.Version, key)
+ if err != nil {
+ return nil, fmt.Errorf("%v, find
service %s/%s/%s/%s",
+ err, key.Tenant, key.AppId,
key.ServiceName, key.Version)
+ }
+ cache[id] = exist
+ }
+
+ if exist {
+ left = append(left, key)
+ }
+ }
+
+ if len(all) == len(left) {
+ continue
+ }
+
+ if len(left) == 0 {
+ ops = append(ops,
registry.OpDel(registry.WithKey(kv.Key)))
+ } else {
+ val, err :=
json.Marshal(&pb.MicroServiceDependency{Dependency: left})
+ if err != nil {
+ return nil, fmt.Errorf("%v, marshal %v", err,
left)
+ }
+ ops = append(ops,
registry.OpPut(registry.WithKey(kv.Key), registry.WithValue(val)))
+ }
+ }
+ return ops, nil
+}
+
+func removeProviderRuleKeys(ctx context.Context, domainProject string, cache
map[string]bool) ([]registry.PluginOp, error) {
+ key := apt.GenerateProviderDependencyRuleKey(domainProject, nil) +
apt.SPLIT
+ resp, err := backend.Store().DependencyRule().Search(ctx,
+ registry.WithStrKey(key), registry.WithPrefix(),
registry.WithKeyOnly())
+ if err != nil {
+ return nil, err
+ }
+
+ var ops []registry.PluginOp
+ for _, kv := range resp.Kvs {
+ id := util.BytesToStringWithNoCopy(kv.Key)
+ exist, ok := cache[id]
+ if !ok {
+ _, key := apt.GetInfoFromDependencyRuleKV(kv.Key)
+ if key == nil || key.ServiceName == "*" {
+ continue
+ }
+
+ _, exist, err = FindServiceIds(ctx, key.Version, key)
+ if err != nil {
+ return nil, fmt.Errorf("find service
%s/%s/%s/%s, %v",
+ key.Tenant, key.AppId, key.ServiceName,
key.Version, err)
+ }
+ cache[id] = exist
+ }
+
+ if !exist {
+ ops = append(ops,
registry.OpDel(registry.WithKey(kv.Key)))
+ }
+ }
+ return ops, nil
+}
+
+func CleanUpDependencyRules(ctx context.Context, domainProject string) error {
+ if len(domainProject) == 0 {
+ return errors.New("required domainProject")
+ }
+
+ cache := make(map[string]bool)
+ pOps, err := removeProviderRuleOfConsumer(ctx, domainProject, cache)
+ if err != nil {
+ return err
+ }
+
+ kOps, err := removeProviderRuleKeys(ctx, domainProject, cache)
+ if err != nil {
+ return err
+ }
+
+ ops := append(append([]registry.PluginOp(nil), pOps...), kOps...)
+ if len(ops) == 0 {
+ return nil
+ }
+
+ return backend.BatchCommit(ctx, ops)
+}
diff --git a/datasource/etcd/util/domain_util.go
b/datasource/etcd/util/domain_util.go
new file mode 100644
index 0000000..ff0b1c0
--- /dev/null
+++ b/datasource/etcd/util/domain_util.go
@@ -0,0 +1,127 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "strings"
+
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ "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/plugin/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/registry"
+)
+
+func GetAllDomainRawData(ctx context.Context) ([]*discovery.KeyValue, error) {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateDomainKey("")),
+ registry.WithPrefix())
+ rsp, err := backend.Store().Domain().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return rsp.Kvs, nil
+
+}
+
+func GetAllDomain(ctx context.Context) ([]string, error) {
+ insWatherByDomainKeys := []string{}
+ kvs, err := GetAllDomainRawData(ctx)
+ if err != nil {
+ return nil, err
+ }
+
+ if len(kvs) == 0 {
+ return insWatherByDomainKeys, err
+ }
+
+ domain := ""
+ instByDomain := ""
+ var arrTmp []string
+ for _, kv := range kvs {
+ arrTmp = strings.Split(util.BytesToStringWithNoCopy(kv.Key),
"/")
+ domain = arrTmp[len(arrTmp)-1]
+ instByDomain = apt.GetInstanceRootKey(domain)
+ insWatherByDomainKeys = append(insWatherByDomainKeys,
instByDomain)
+ }
+ return insWatherByDomainKeys, err
+}
+
+func DomainExist(ctx context.Context, domain string) (bool, error) {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateDomainKey(domain)),
+ registry.WithCountOnly())
+ rsp, err := backend.Store().Domain().Search(ctx, opts...)
+ if err != nil {
+ return false, err
+ }
+ return rsp.Count > 0, nil
+}
+
+func ProjectExist(ctx context.Context, domain, project string) (bool, error) {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateProjectKey(domain, project)),
+ registry.WithCountOnly())
+ rsp, err := backend.Store().Project().Search(ctx, opts...)
+ if err != nil {
+ return false, err
+ }
+ return rsp.Count > 0, nil
+}
+
+func NewDomain(ctx context.Context, domain string) (bool, error) {
+ ok, err := backend.Registry().PutNoOverride(ctx,
+ registry.WithStrKey(apt.GenerateDomainKey(domain)))
+ if err != nil {
+ return false, err
+ }
+ return ok, nil
+}
+
+func NewProject(ctx context.Context, domain, project string) (bool, error) {
+ ok, err := backend.Registry().PutNoOverride(ctx,
+ registry.WithStrKey(apt.GenerateProjectKey(domain, project)))
+ if err != nil {
+ return ok, err
+ }
+ return ok, nil
+}
+
+func NewDomainProject(ctx context.Context, domain, project string) error {
+ copyCtx := util.SetContext(util.CloneContext(ctx), util.CtxCacheOnly,
"1")
+ ok, err := DomainExist(copyCtx, domain)
+ if !ok && err == nil {
+ ok, err = NewDomain(ctx, domain)
+ if ok {
+ log.Infof("new domain(%s)", domain)
+ }
+ }
+ if err != nil {
+ return err
+ }
+ ok, err = ProjectExist(copyCtx, domain, project)
+ if !ok && err == nil {
+ ok, err = NewProject(ctx, domain, project)
+ if ok {
+ log.Infof("new project(%s/%s)", domain, project)
+ }
+ }
+ return err
+}
diff --git a/datasource/etcd/util/domain_util_test.go
b/datasource/etcd/util/domain_util_test.go
new file mode 100644
index 0000000..d18a810
--- /dev/null
+++ b/datasource/etcd/util/domain_util_test.go
@@ -0,0 +1,84 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "testing"
+)
+
+func TestGetOneDomainProjectServiceCount(t *testing.T) {
+ _, err := GetOneDomainProjectServiceCount(context.Background(), "")
+ if err != nil {
+ t.Fatalf("GetOneDomainProjectServiceCount failed")
+ }
+}
+
+func TestGetOneDomainProjectInstanceCount(t *testing.T) {
+ _, err := GetOneDomainProjectInstanceCount(context.Background(), "")
+ if err != nil {
+ t.Fatalf("GetOneDomainProjectInstanceCount failed")
+ }
+}
+
+func TestGetDomain(t *testing.T) {
+ _, err := GetAllDomainRawData(context.Background())
+ if err != nil {
+ t.Fatalf("GetAllDomainRawData failed")
+ }
+
+ _, err = GetAllDomain(context.Background())
+ if err != nil {
+ t.Fatalf("GetAllDomain failed")
+ }
+}
+
+func TestDomainExist(t *testing.T) {
+ _, err := DomainExist(context.Background(), "")
+ if err != nil {
+ t.Fatalf("DomainExist failed")
+ }
+}
+
+func TestNewDomain(t *testing.T) {
+ _, err := NewDomain(context.Background(), "")
+ if err != nil {
+ t.Fatalf("NewDomain failed")
+ }
+}
+
+func TestProjectExist(t *testing.T) {
+ _, err := ProjectExist(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("DomainExist failed")
+ }
+}
+
+func TestNewProject(t *testing.T) {
+ _, err := NewProject(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("NewProject failed")
+ }
+}
+
+func TestNewDomainProject(t *testing.T) {
+ err := NewDomainProject(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("NewDomainProject failed")
+ }
+}
diff --git a/datasource/etcd/util/heartbeat_util.go
b/datasource/etcd/util/heartbeat_util.go
new file mode 100644
index 0000000..8db60f8
--- /dev/null
+++ b/datasource/etcd/util/heartbeat_util.go
@@ -0,0 +1,53 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "errors"
+
+ 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/plugin/registry"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+)
+
+func HeartbeatUtil(ctx context.Context, domainProject string, serviceID
string, instanceID string) (leaseID int64, ttl int64, _ *scerr.Error) {
+ leaseID, err := GetLeaseID(ctx, domainProject, serviceID, instanceID)
+ if err != nil {
+ return leaseID, ttl,
scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+ ttl, err = KeepAliveLease(ctx, domainProject, serviceID, instanceID,
leaseID)
+ if err != nil {
+ return leaseID, ttl, scerr.NewError(scerr.ErrInstanceNotExists,
err.Error())
+ }
+ return leaseID, ttl, nil
+}
+
+func KeepAliveLease(ctx context.Context, domainProject, serviceID, instanceID
string, leaseID int64) (ttl int64, err error) {
+ if leaseID == -1 {
+ return ttl, errors.New("leaseId not exist, instance not exist")
+ }
+ ttl, err = backend.Store().KeepAlive(ctx,
+ registry.WithStrKey(apt.GenerateInstanceLeaseKey(domainProject,
serviceID, instanceID)),
+ registry.WithLease(leaseID))
+ if err != nil {
+ return ttl, err
+ }
+ return ttl, nil
+}
diff --git a/server/plugin/uuid/uuid.go
b/datasource/etcd/util/heartbeat_util_test.go
similarity index 62%
copy from server/plugin/uuid/uuid.go
copy to datasource/etcd/util/heartbeat_util_test.go
index c8aaa7c..38c91b0 100644
--- a/server/plugin/uuid/uuid.go
+++ b/datasource/etcd/util/heartbeat_util_test.go
@@ -15,15 +15,30 @@
* limitations under the License.
*/
-package uuid
+package util
import (
"context"
+ "testing"
)
-const ContextKey = "_uuid_key"
+func TestHeartbeatUtil(t *testing.T) {
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("TestHeartbeatUtil failed")
+ }
+ }()
+ HeartbeatUtil(context.Background(), "", "", "")
+}
+
+func TestKeepAliveLease(t *testing.T) {
+ _, err := KeepAliveLease(context.Background(), "", "", "", -1)
+ if err == nil {
+ t.Fatalf("KeepAliveLease -1 failed")
+ }
-type UUID interface {
- GetServiceID(ctx context.Context) string
- GetInstanceID(ctx context.Context) string
+ _, err = KeepAliveLease(context.Background(), "", "", "", 0)
+ if err != nil {
+ t.Fatalf("KeepAliveLease failed")
+ }
}
diff --git a/datasource/etcd/util/instance_util.go
b/datasource/etcd/util/instance_util.go
new file mode 100644
index 0000000..d8f833e
--- /dev/null
+++ b/datasource/etcd/util/instance_util.go
@@ -0,0 +1,276 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "crypto/sha1"
+ "encoding/json"
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ pb "github.com/apache/servicecomb-service-center/pkg/registry"
+ rmodel "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/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/registry"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+)
+
+func GetLeaseID(ctx context.Context, domainProject string, serviceID string,
instanceID string) (int64, error) {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateInstanceLeaseKey(domainProject,
serviceID, instanceID)))
+ resp, err := backend.Store().Lease().Search(ctx, opts...)
+ if err != nil {
+ return -1, err
+ }
+ if len(resp.Kvs) <= 0 {
+ return -1, nil
+ }
+ leaseID, _ := strconv.ParseInt(resp.Kvs[0].Value.(string), 10, 64)
+ return leaseID, nil
+}
+
+func GetInstance(ctx context.Context, domainProject string, serviceID string,
instanceID string) (*pb.MicroServiceInstance, error) {
+ key := apt.GenerateInstanceKey(domainProject, serviceID, instanceID)
+ opts := append(FromContext(ctx), registry.WithStrKey(key))
+
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+ if len(resp.Kvs) == 0 {
+ return nil, nil
+ }
+
+ return resp.Kvs[0].Value.(*pb.MicroServiceInstance), nil
+}
+
+func FormatRevision(revs, counts []int64) (s string) {
+ for i, rev := range revs {
+ s += fmt.Sprintf("%d.%d,", rev, counts[i])
+ }
+ return fmt.Sprintf("%x", sha1.Sum(util.StringToBytesWithNoCopy(s)))
+}
+
+func GetAllInstancesOfOneService(ctx context.Context, domainProject string,
serviceID string) ([]*pb.MicroServiceInstance, error) {
+ key := apt.GenerateInstanceKey(domainProject, serviceID, "")
+ opts := append(FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s instances failed", serviceID)
+ return nil, err
+ }
+
+ instances := make([]*pb.MicroServiceInstance, 0, len(resp.Kvs))
+ for _, kv := range resp.Kvs {
+ instances = append(instances,
kv.Value.(*pb.MicroServiceInstance))
+ }
+ return instances, nil
+}
+
+func GetInstanceCountOfOneService(ctx context.Context, domainProject string,
serviceID string) (int64, error) {
+ key := apt.GenerateInstanceKey(domainProject, serviceID, "")
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithPrefix(),
+ registry.WithCountOnly())
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get number of service[%s]'s instances failed",
serviceID)
+ return 0, err
+ }
+ return resp.Count, nil
+}
+
+type EndpointIndexValue struct {
+ serviceID string
+ instanceID string
+}
+
+func ParseEndpointIndexValue(value []byte) EndpointIndexValue {
+ endpointValue := EndpointIndexValue{}
+ tmp := util.BytesToStringWithNoCopy(value)
+ splitedTmp := strings.Split(tmp, "/")
+ endpointValue.serviceID = splitedTmp[0]
+ endpointValue.instanceID = splitedTmp[1]
+ return endpointValue
+}
+
+func DeleteServiceAllInstances(ctx context.Context, serviceID string) error {
+ domainProject := util.ParseDomainProject(ctx)
+
+ instanceLeaseKey := apt.GenerateInstanceLeaseKey(domainProject,
serviceID, "")
+ resp, err := backend.Store().Lease().Search(ctx,
+ registry.WithStrKey(instanceLeaseKey),
+ registry.WithPrefix(),
+ registry.WithNoCache())
+ if err != nil {
+ log.Errorf(err, "delete all of service[%s]'s instances failed:
get instance lease failed", serviceID)
+ return err
+ }
+ if resp.Count <= 0 {
+ log.Warnf("service[%s] has no deployment of instance.",
serviceID)
+ return nil
+ }
+ for _, v := range resp.Kvs {
+ leaseID, _ := strconv.ParseInt(v.Value.(string), 10, 64)
+ err := backend.Registry().LeaseRevoke(ctx, leaseID)
+ if err != nil {
+ log.Error("", err)
+ }
+ }
+ return nil
+}
+
+func QueryAllProvidersInstances(ctx context.Context, selfServiceID string)
(results []*pb.WatchInstanceResponse, rev int64) {
+ results = []*pb.WatchInstanceResponse{}
+
+ domainProject := util.ParseDomainProject(ctx)
+
+ service, err := GetService(ctx, domainProject, selfServiceID)
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s file failed", selfServiceID)
+ return
+ }
+ if service == nil {
+ log.Errorf(nil, "service[%s] does not exist", selfServiceID)
+ return
+ }
+ providerIDs, _, err := GetAllProviderIds(ctx, domainProject, service)
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s providerIDs failed",
selfServiceID)
+ return
+ }
+
+ rev = backend.Revision()
+
+ for _, providerID := range providerIDs {
+ service, err := GetServiceWithRev(ctx, domainProject,
providerID, rev)
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s provider[%s] file
with revision %d failed",
+ selfServiceID, providerID, rev)
+ return
+ }
+ if service == nil {
+ continue
+ }
+
+ kvs, err := queryServiceInstancesKvs(ctx, providerID, rev)
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s provider[%s]
instances with revision %d failed",
+ selfServiceID, providerID, rev)
+ return
+ }
+
+ for _, kv := range kvs {
+ results = append(results, &pb.WatchInstanceResponse{
+ Response:
proto.CreateResponse(proto.Response_SUCCESS, "List instance successfully."),
+ Action: string(rmodel.EVT_INIT),
+ Key: &pb.MicroServiceKey{
+ Environment: service.Environment,
+ AppId: service.AppId,
+ ServiceName: service.ServiceName,
+ Version: service.Version,
+ },
+ Instance: kv.Value.(*pb.MicroServiceInstance),
+ })
+ }
+ }
+ return
+}
+
+func queryServiceInstancesKvs(ctx context.Context, serviceID string, rev
int64) ([]*discovery.KeyValue, error) {
+ domainProject := util.ParseDomainProject(ctx)
+ key := apt.GenerateInstanceKey(domainProject, serviceID, "")
+ resp, err := backend.Store().Instance().Search(ctx,
+ registry.WithStrKey(key),
+ registry.WithPrefix(),
+ registry.WithRev(rev))
+ if err != nil {
+ log.Errorf(err, "get service[%s]'s instances with revision %d
failed",
+ serviceID, rev)
+ return nil, err
+ }
+ return resp.Kvs, nil
+}
+
+func UpdateInstance(ctx context.Context, domainProject string, instance
*pb.MicroServiceInstance) *scerr.Error {
+ leaseID, err := GetLeaseID(ctx, domainProject, instance.ServiceId,
instance.InstanceId)
+ if err != nil {
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+ if leaseID == -1 {
+ return scerr.NewError(scerr.ErrInstanceNotExists, "Instance's
leaseId not exist.")
+ }
+
+ instance.ModTimestamp = strconv.FormatInt(time.Now().Unix(), 10)
+ data, err := json.Marshal(instance)
+ if err != nil {
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+
+ key := apt.GenerateInstanceKey(domainProject, instance.ServiceId,
instance.InstanceId)
+
+ resp, err := backend.Registry().TxnWithCmp(ctx,
+ []registry.PluginOp{registry.OpPut(
+ registry.WithStrKey(key),
+ registry.WithValue(data),
+ registry.WithLease(leaseID))},
+ []registry.CompareOp{registry.OpCmp(
+
registry.CmpVer(util.StringToBytesWithNoCopy(apt.GenerateServiceKey(domainProject,
instance.ServiceId))),
+ registry.CmpNotEqual, 0)},
+ nil)
+ if err != nil {
+ return scerr.NewError(scerr.ErrUnavailableBackend, err.Error())
+ }
+ if !resp.Succeeded {
+ return scerr.NewError(scerr.ErrInstanceNotExists, "Instance
does not exist.")
+ }
+ return nil
+}
+
+func AppendFindResponse(ctx context.Context, index int64, resp *pb.Response,
instances []*pb.MicroServiceInstance,
+ updatedResult *[]*pb.FindResult, notModifiedResult *[]int64,
failedResult **pb.FindFailedResult) {
+ if code := resp.GetCode(); code != proto.Response_SUCCESS {
+ if *failedResult == nil {
+ *failedResult = &pb.FindFailedResult{
+ Error: scerr.NewError(code, resp.GetMessage()),
+ }
+ }
+ (*failedResult).Indexes = append((*failedResult).Indexes, index)
+ return
+ }
+ iv, _ := ctx.Value(util.CtxRequestRevision).(string)
+ ov, _ := ctx.Value(util.CtxResponseRevision).(string)
+ if len(iv) > 0 && iv == ov {
+ *notModifiedResult = append(*notModifiedResult, index)
+ return
+ }
+ *updatedResult = append(*updatedResult, &pb.FindResult{
+ Index: index,
+ Instances: instances,
+ Rev: ov,
+ })
+}
diff --git a/datasource/etcd/util/instance_util_test.go
b/datasource/etcd/util/instance_util_test.go
new file mode 100644
index 0000000..d20b34a
--- /dev/null
+++ b/datasource/etcd/util/instance_util_test.go
@@ -0,0 +1,170 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ pb "github.com/apache/servicecomb-service-center/pkg/registry"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/core/proto"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+ "testing"
+)
+
+func TestFormatRevision(t *testing.T) {
+ // null
+ if x := FormatRevision(nil, nil);
"da39a3ee5e6b4b0d3255bfef95601890afd80709" != x {
+ t.Fatalf("TestFormatRevision failed, %s", x)
+ }
+ // 1.1,11.1,
+ if x := FormatRevision([]int64{1, 11}, []int64{1, 1});
"87aa7d310290ff4f93248c0aed6870b928edf45a" != x {
+ t.Fatalf("TestFormatRevision failed, %s", x)
+ }
+ // 1.11,1.1,
+ if x := FormatRevision([]int64{1, 1}, []int64{11, 1});
"24675d196e3dea5be0c774cab281366640fc99ef" != x {
+ t.Fatalf("TestFormatRevision failed, %s", x)
+ }
+}
+
+func TestGetLeaseId(t *testing.T) {
+ _, err := GetLeaseID(context.Background(), "", "", "")
+ if err != nil {
+ t.Fatalf(`GetLeaseID failed`)
+ }
+}
+
+func TestGetInstance(t *testing.T) {
+ _, err := GetInstance(context.Background(), "", "", "")
+ if err != nil {
+ t.Fatalf(`GetInstance failed`)
+ }
+
+ _, err = GetAllInstancesOfOneService(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf(`GetAllInstancesOfOneService failed`)
+ }
+
+ QueryAllProvidersInstances(context.Background(), "")
+
+ _, err = queryServiceInstancesKvs(context.Background(), "", 0)
+ if err != nil {
+ t.Fatalf(`queryServiceInstancesKvs failed`)
+ }
+}
+
+func TestDeleteServiceAllInstances(t *testing.T) {
+ err := DeleteServiceAllInstances(context.Background(), "")
+ if err != nil {
+ t.Fatalf(`DeleteServiceAllInstances failed`)
+ }
+}
+
+func TestParseEndpointValue(t *testing.T) {
+ epv := ParseEndpointIndexValue([]byte("x/y"))
+ if epv.serviceID != "x" || epv.instanceID != "y" {
+ t.Fatalf(`ParseEndpointIndexValue failed`)
+ }
+}
+
+func TestGetInstanceCountOfOneService(t *testing.T) {
+ _, err := GetInstanceCountOfOneService(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf(`GetInstanceCountOfOneService failed`)
+ }
+}
+
+func TestUpdateInstance(t *testing.T) {
+ err := UpdateInstance(util.SetContext(context.Background(),
util.CtxNocache, "1"), "", &pb.MicroServiceInstance{})
+ if err == nil {
+ 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.Response, find.Instances,
&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, util.CtxResponseRevision, "1")
+ AppendFindResponse(cloneCtx, 1, find.Response, find.Instances,
&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, util.CtxRequestRevision, "1")
+ cloneCtx = context.WithValue(cloneCtx, util.CtxResponseRevision, "1")
+ AppendFindResponse(cloneCtx, 1, find.Response, find.Instances,
&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 = proto.CreateResponse(scerr.ErrInternal, "test")
+ AppendFindResponse(ctx, 1, find.Response, find.Instances,
&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 = proto.CreateResponse(scerr.ErrInvalidParams, "test")
+ AppendFindResponse(ctx, 2, find.Response, find.Instances,
&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.Response, find.Instances,
&updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(ctx, 2, find.Response, find.Instances,
&updatedResult, ¬ModifiedResult, &failedResult)
+ cloneCtx = context.WithValue(ctx, util.CtxRequestRevision, "1")
+ cloneCtx = context.WithValue(cloneCtx, util.CtxResponseRevision, "1")
+ AppendFindResponse(cloneCtx, 3, find.Response, find.Instances,
&updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(cloneCtx, 4, find.Response, find.Instances,
&updatedResult, ¬ModifiedResult, &failedResult)
+ find.Response = proto.CreateResponse(scerr.ErrInternal, "test")
+ AppendFindResponse(ctx, 5, find.Response, find.Instances,
&updatedResult, ¬ModifiedResult, &failedResult)
+ AppendFindResponse(ctx, 6, find.Response, find.Instances,
&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/datasource/etcd/util/microservice_util.go
b/datasource/etcd/util/microservice_util.go
new file mode 100644
index 0000000..98f1911
--- /dev/null
+++ b/datasource/etcd/util/microservice_util.go
@@ -0,0 +1,310 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "encoding/json"
+ "strings"
+
+ "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/discovery"
+ "github.com/apache/servicecomb-service-center/server/plugin/quota"
+ "github.com/apache/servicecomb-service-center/server/plugin/registry"
+)
+
+const defaultCacheValue = "1"
+
+/*
+ get Service by service id
+*/
+func GetServiceWithRev(ctx context.Context, domain string, id string, rev
int64) (*pb.MicroService, error) {
+ key := apt.GenerateServiceKey(domain, id)
+ serviceResp, err := backend.Store().Service().Search(ctx,
+ registry.WithStrKey(key),
+ registry.WithRev(rev))
+ if err != nil {
+ return nil, err
+ }
+ if len(serviceResp.Kvs) == 0 {
+ return nil, nil
+ }
+ return serviceResp.Kvs[0].Value.(*pb.MicroService), nil
+}
+
+func GetService(ctx context.Context, domainProject string, serviceID string)
(*pb.MicroService, error) {
+ key := apt.GenerateServiceKey(domainProject, serviceID)
+ opts := append(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
+}
+
+// GetServiceFromCache gets service from cache
+func GetServiceFromCache(domainProject string, serviceID string)
*pb.MicroService {
+ ctx := context.WithValue(context.WithValue(context.Background(),
+ util.CtxCacheOnly, defaultCacheValue),
+ util.CtxGlobal, defaultCacheValue)
+ svc, _ := GetService(ctx, domainProject, serviceID)
+ return svc
+}
+
+func getServicesRawData(ctx context.Context, domainProject string)
([]*discovery.KeyValue, error) {
+ key := apt.GenerateServiceKey(domainProject, "")
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithPrefix())
+ resp, err := backend.Store().Service().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return resp.Kvs, err
+}
+
+//GetAllServicesAcrossDomainProject get services of all domains, projects
+//the map's key is domainProject
+func GetAllServicesAcrossDomainProject(ctx context.Context)
(map[string][]*pb.MicroService, error) {
+ key := apt.GetServiceRootKey("")
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithPrefix())
+ serviceResp, err := backend.Store().Service().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+
+ services := make(map[string][]*pb.MicroService)
+ if len(serviceResp.Kvs) == 0 {
+ return services, nil
+ }
+
+ for _, value := range serviceResp.Kvs {
+ prefix := util.BytesToStringWithNoCopy(value.Key)
+ parts := strings.Split(prefix, apt.SPLIT)
+ if len(parts) != 7 {
+ continue
+ }
+ domainProject := parts[4] + apt.SPLIT + parts[5]
+ microService, ok := value.Value.(*pb.MicroService)
+ if !ok {
+ log.Errorf(nil, "backend key[%s]'s value is not type
*pb.MicroService", prefix)
+ continue
+ }
+ services[domainProject] = append(services[domainProject],
microService)
+ }
+ return services, nil
+}
+
+func GetServicesByDomainProject(ctx context.Context, domainProject string)
([]*pb.MicroService, error) {
+ kvs, err := getServicesRawData(ctx, domainProject)
+ if err != nil {
+ return nil, err
+ }
+ services := []*pb.MicroService{}
+ for _, kv := range kvs {
+ services = append(services, kv.Value.(*pb.MicroService))
+ }
+ return services, 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(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(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) {
+ copy := *key
+ copy.Version = ""
+ var (
+ prefix string
+ indexer discovery.Indexer
+ )
+ if alias {
+ prefix = apt.GenerateServiceAliasKey(©)
+ indexer = backend.Store().ServiceAlias()
+ } else {
+ prefix = apt.GenerateServiceIndexKey(©)
+ indexer = backend.Store().ServiceIndex()
+ }
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(prefix),
+ registry.WithPrefix(),
+ registry.WithDescendOrder())
+ resp, err := indexer.Search(ctx, opts...)
+ return resp, err
+}
+
+func FindServiceIds(ctx context.Context, versionRule string, key
*pb.MicroServiceKey) ([]string, bool, error) {
+ // 版本规则
+ match := ParseVersionRule(versionRule)
+ if match == nil {
+ copy := *key
+ copy.Version = versionRule
+ serviceID, err := GetServiceID(ctx, ©)
+ 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
+
+FIND_RULE:
+ 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 FIND_RULE
+ }
+ return match(resp.Kvs), true, nil
+}
+
+func ServiceExist(ctx context.Context, domainProject string, serviceID string)
bool {
+ opts := append(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 GetAllServiceUtil(ctx context.Context) ([]*pb.MicroService, error) {
+ domainProject := util.ParseDomainProject(ctx)
+ services, err := GetServicesByDomainProject(ctx, domainProject)
+ if err != nil {
+ return nil, err
+ }
+ return services, nil
+}
+
+func RemandServiceQuota(ctx context.Context) {
+ plugin.Plugins().Quota().RemandQuotas(ctx, quota.MicroServiceQuotaType)
+}
+
+func RemandInstanceQuota(ctx context.Context) {
+ plugin.Plugins().Quota().RemandQuotas(ctx,
quota.MicroServiceInstanceQuotaType)
+}
+
+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
+}
+
+func GetOneDomainProjectServiceCount(ctx context.Context, domainProject
string) (int64, error) {
+ key := apt.GenerateServiceKey(domainProject, "")
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithCountOnly(),
+ registry.WithPrefix())
+ resp, err := backend.Store().Service().Search(ctx, opts...)
+ if err != nil {
+ return 0, err
+ }
+ return resp.Count, nil
+}
+
+func GetOneDomainProjectInstanceCount(ctx context.Context, domainProject
string) (int64, error) {
+ key := apt.GetInstanceRootKey(domainProject) + "/"
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithCountOnly(),
+ registry.WithPrefix())
+ resp, err := backend.Store().Instance().Search(ctx, opts...)
+ if err != nil {
+ return 0, err
+ }
+ return resp.Count, nil
+}
+
+func SetServiceDefaultValue(service *pb.MicroService) {
+ if len(service.AppId) == 0 {
+ service.AppId = proto.APP_ID
+ }
+ if len(service.Version) == 0 {
+ service.Version = proto.VERSION
+ }
+ if len(service.Level) == 0 {
+ service.Level = "BACK"
+ }
+ if len(service.Status) == 0 {
+ service.Status = pb.MS_UP
+ }
+}
diff --git a/datasource/etcd/util/rule_util.go
b/datasource/etcd/util/rule_util.go
new file mode 100644
index 0000000..a7eba03
--- /dev/null
+++ b/datasource/etcd/util/rule_util.go
@@ -0,0 +1,300 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "fmt"
+ "reflect"
+ "regexp"
+ "strings"
+
+ "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/registry"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+)
+
+type RuleFilter struct {
+ DomainProject string
+ ProviderRules []*pb.ServiceRule
+}
+
+func (rf *RuleFilter) Filter(ctx context.Context, consumerID string) (bool,
error) {
+ copyCtx := util.SetContext(util.CloneContext(ctx), util.CtxCacheOnly,
"1")
+ consumer, err := GetService(copyCtx, rf.DomainProject, consumerID)
+ if consumer == nil {
+ return false, err
+ }
+
+ if len(rf.ProviderRules) == 0 {
+ return true, nil
+ }
+
+ tags, err := GetTagsUtils(copyCtx, rf.DomainProject, consumerID)
+ if err != nil {
+ return false, err
+ }
+ matchErr := MatchRules(rf.ProviderRules, consumer, tags)
+ if matchErr != nil {
+ if matchErr.Code == scerr.ErrPermissionDeny {
+ return false, nil
+ }
+ return false, matchErr
+ }
+ return true, nil
+}
+
+func (rf *RuleFilter) FilterAll(ctx context.Context, consumerIDs []string)
(allow []string, deny []string, err error) {
+ l := len(consumerIDs)
+ if l == 0 || len(rf.ProviderRules) == 0 {
+ return consumerIDs, nil, nil
+ }
+
+ allowIdx, denyIdx := 0, l
+ consumers := make([]string, l)
+ for _, consumerID := range consumerIDs {
+ ok, err := rf.Filter(ctx, consumerID)
+ if err != nil {
+ return nil, nil, err
+ }
+ if ok {
+ consumers[allowIdx] = consumerID
+ allowIdx++
+ } else {
+ denyIdx--
+ consumers[denyIdx] = consumerID
+ }
+ }
+ return consumers[:allowIdx], consumers[denyIdx:], nil
+}
+
+func GetRulesUtil(ctx context.Context, domainProject string, serviceID string)
([]*pb.ServiceRule, error) {
+ key := util.StringJoin([]string{
+ apt.GetServiceRuleRootKey(domainProject),
+ serviceID,
+ "",
+ }, "/")
+
+ opts := append(FromContext(ctx), registry.WithStrKey(key),
registry.WithPrefix())
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
+ if err != nil {
+ return nil, err
+ }
+
+ rules := []*pb.ServiceRule{}
+ for _, kv := range resp.Kvs {
+ rules = append(rules, kv.Value.(*pb.ServiceRule))
+ }
+ return rules, nil
+}
+
+func RuleExist(ctx context.Context, domainProject string, serviceID string,
attr string, pattern string) bool {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateRuleIndexKey(domainProject,
serviceID, attr, pattern)),
+ registry.WithCountOnly())
+ resp, err := backend.Store().RuleIndex().Search(ctx, opts...)
+ if err != nil || resp.Count == 0 {
+ return false
+ }
+ return true
+}
+
+func GetServiceRuleType(ctx context.Context, domainProject string, serviceID
string) (string, int, error) {
+ key := apt.GenerateServiceRuleKey(domainProject, serviceID, "")
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(key),
+ registry.WithPrefix())
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get service[%s] rule failed", serviceID)
+ return "", 0, err
+ }
+ if len(resp.Kvs) == 0 {
+ return "", 0, nil
+ }
+ return resp.Kvs[0].Value.(*pb.ServiceRule).RuleType, len(resp.Kvs), nil
+}
+
+func GetOneRule(ctx context.Context, domainProject, serviceID, ruleID string)
(*pb.ServiceRule, error) {
+ opts := append(FromContext(ctx),
+ registry.WithStrKey(apt.GenerateServiceRuleKey(domainProject,
serviceID, ruleID)))
+ resp, err := backend.Store().Rule().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get service rule[%s/%s]", serviceID, ruleID)
+ return nil, err
+ }
+ if len(resp.Kvs) == 0 {
+ log.Errorf(nil, "get service rule[%s/%s] failed", serviceID,
ruleID)
+ return nil, nil
+ }
+ return resp.Kvs[0].Value.(*pb.ServiceRule), nil
+}
+
+func AllowAcrossDimension(ctx context.Context, providerService
*pb.MicroService, consumerService *pb.MicroService) error {
+ if providerService.AppId != consumerService.AppId {
+ if len(providerService.Properties) == 0 {
+ return fmt.Errorf("not allow across app access")
+ }
+
+ if allowCrossApp, ok :=
providerService.Properties[proto.PROP_ALLOW_CROSS_APP]; !ok ||
strings.ToLower(allowCrossApp) != "true" {
+ return fmt.Errorf("not allow across app access")
+ }
+ }
+
+ if
!apt.IsShared(proto.MicroServiceToKey(util.ParseTargetDomainProject(ctx),
providerService)) &&
+ providerService.Environment != consumerService.Environment {
+ return fmt.Errorf("not allow across environment access")
+ }
+
+ return nil
+}
+
+func MatchRules(rulesOfProvider []*pb.ServiceRule, consumer *pb.MicroService,
tagsOfConsumer map[string]string) *scerr.Error {
+ if consumer == nil {
+ return scerr.NewError(scerr.ErrInvalidParams, "consumer is nil")
+ }
+
+ if len(rulesOfProvider) <= 0 {
+ return nil
+ }
+ if rulesOfProvider[0].RuleType == "WHITE" {
+ return patternWhiteList(rulesOfProvider, tagsOfConsumer,
consumer)
+ }
+ return patternBlackList(rulesOfProvider, tagsOfConsumer, consumer)
+}
+
+func patternWhiteList(rulesOfProvider []*pb.ServiceRule, tagsOfConsumer
map[string]string, consumer *pb.MicroService) *scerr.Error {
+ v := reflect.Indirect(reflect.ValueOf(consumer))
+ consumerID := consumer.ServiceId
+ for _, rule := range rulesOfProvider {
+ value, err := parsePattern(v, rule, tagsOfConsumer, consumerID)
+ if err != nil {
+ return err
+ }
+ if len(value) == 0 {
+ continue
+ }
+
+ match, _ := regexp.MatchString(rule.Pattern, value)
+ if match {
+ log.Infof("consumer[%s][%s/%s/%s/%s] match white list,
rule.Pattern is %s, value is %s",
+ consumerID, consumer.Environment,
consumer.AppId, consumer.ServiceName, consumer.Version,
+ rule.Pattern, value)
+ return nil
+ }
+ }
+ return scerr.NewError(scerr.ErrPermissionDeny, "Not found in white
list")
+}
+
+func parsePattern(v reflect.Value, rule *pb.ServiceRule, tagsOfConsumer
map[string]string, consumerID string) (string, *scerr.Error) {
+ if strings.HasPrefix(rule.Attribute, "tag_") {
+ key := rule.Attribute[4:]
+ value := tagsOfConsumer[key]
+ if len(value) == 0 {
+ log.Infof("can not find service[%s] tag[%s]",
consumerID, key)
+ }
+ return value, nil
+ }
+ key := v.FieldByName(rule.Attribute)
+ if !key.IsValid() {
+ log.Errorf(nil, "can not find service[%] field[%s], ruleID is
%s",
+ consumerID, rule.Attribute, rule.RuleId)
+ return "", scerr.NewErrorf(scerr.ErrInternal, "Can not find
field '%s'", rule.Attribute)
+ }
+ return key.String(), nil
+
+}
+
+func patternBlackList(rulesOfProvider []*pb.ServiceRule, tagsOfConsumer
map[string]string, consumer *pb.MicroService) *scerr.Error {
+ v := reflect.Indirect(reflect.ValueOf(consumer))
+ consumerID := consumer.ServiceId
+ for _, rule := range rulesOfProvider {
+ var value string
+ value, err := parsePattern(v, rule, tagsOfConsumer, consumerID)
+ if err != nil {
+ return err
+ }
+ if len(value) == 0 {
+ continue
+ }
+
+ match, _ := regexp.MatchString(rule.Pattern, value)
+ if match {
+ log.Warnf("no permission to access,
consumer[%s][%s/%s/%s/%s] match black list, rule.Pattern is %s, value is %s",
+ consumerID, consumer.Environment,
consumer.AppId, consumer.ServiceName, consumer.Version,
+ rule.Pattern, value)
+ return scerr.NewError(scerr.ErrPermissionDeny, "Found
in black list")
+ }
+ }
+ return nil
+}
+
+func Accessible(ctx context.Context, consumerID string, providerID string)
*scerr.Error {
+ if len(consumerID) == 0 {
+ return nil
+ }
+
+ domainProject := util.ParseDomainProject(ctx)
+ targetDomainProject := util.ParseTargetDomainProject(ctx)
+
+ consumerService, err := GetService(ctx, domainProject, consumerID)
+ if err != nil {
+ return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in
query consumer(%s)", err.Error())
+ }
+ if consumerService == nil {
+ return scerr.NewError(scerr.ErrServiceNotExists, "consumer
serviceID is invalid")
+ }
+
+ // 跨应用权限
+ providerService, err := GetService(ctx, targetDomainProject, providerID)
+ if err != nil {
+ return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in
query provider(%s)", err.Error())
+ }
+ if providerService == nil {
+ return scerr.NewError(scerr.ErrServiceNotExists, "provider
serviceID is invalid")
+ }
+
+ err = AllowAcrossDimension(ctx, providerService, consumerService)
+ if err != nil {
+ return scerr.NewError(scerr.ErrPermissionDeny, err.Error())
+ }
+
+ ctx = util.SetContext(util.CloneContext(ctx), util.CtxCacheOnly, "1")
+
+ // 黑白名单
+ rules, err := GetRulesUtil(ctx, targetDomainProject, providerID)
+ if err != nil {
+ return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in
query provider rules(%s)", err.Error())
+ }
+
+ if len(rules) == 0 {
+ return nil
+ }
+
+ validateTags, err := GetTagsUtils(ctx, domainProject,
consumerService.ServiceId)
+ if err != nil {
+ return scerr.NewErrorf(scerr.ErrInternal, "An error occurred in
query consumer tags(%s)", err.Error())
+ }
+
+ return MatchRules(rules, consumerService, validateTags)
+}
diff --git a/datasource/etcd/util/rule_util_test.go
b/datasource/etcd/util/rule_util_test.go
new file mode 100644
index 0000000..7bdc853
--- /dev/null
+++ b/datasource/etcd/util/rule_util_test.go
@@ -0,0 +1,306 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "net/http"
+ "testing"
+
+ "github.com/apache/servicecomb-service-center/pkg/registry"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/core/proto"
+)
+
+func TestRuleFilter_Filter(t *testing.T) {
+ rf := RuleFilter{
+ DomainProject: "",
+ ProviderRules: []*registry.ServiceRule{},
+ }
+ _, err := rf.Filter(context.Background(), "")
+ if err != nil {
+ t.Fatalf("RuleFilter Filter failed")
+ }
+ _, _, err = rf.FilterAll(context.Background(), []string{""})
+ if err != nil {
+ t.Fatalf("RuleFilter FilterAll failed")
+ }
+ rf.ProviderRules = []*registry.ServiceRule{
+ {},
+ }
+ _, _, err = rf.FilterAll(context.Background(), []string{""})
+ if err != nil {
+ t.Fatalf("RuleFilter FilterAll failed")
+ }
+}
+
+func TestGetRulesUtil(t *testing.T) {
+ _, err := GetRulesUtil(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("GetRulesUtil failed")
+ }
+
+ _, err = GetOneRule(context.Background(), "", "", "")
+ if err != nil {
+ t.Fatalf("GetOneRule failed")
+ }
+}
+
+func TestRuleExist(t *testing.T) {
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("TestRuleExist panic")
+ }
+ }()
+ RuleExist(util.SetContext(context.Background(), util.CtxCacheOnly,
"1"), "", "", "", "")
+ RuleExist(context.Background(), "", "", "", "")
+}
+
+func TestGetServiceRuleType(t *testing.T) {
+ _, _, err := GetServiceRuleType(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("GetServiceRuleType failed")
+ }
+}
+
+func TestAllowAcrossApp(t *testing.T) {
+ err := AllowAcrossDimension(context.Background(),
®istry.MicroService{
+ AppId: "a",
+ }, ®istry.MicroService{
+ AppId: "a",
+ })
+ if err != nil {
+ t.Fatalf("AllowAcrossApp with the same appId and no property
failed")
+ }
+
+ err = AllowAcrossDimension(context.Background(), ®istry.MicroService{
+ AppId: "a",
+ }, ®istry.MicroService{
+ AppId: "c",
+ })
+ if err == nil {
+ t.Fatalf("AllowAcrossApp with the diff appId and no property
failed")
+ }
+
+ err = AllowAcrossDimension(context.Background(), ®istry.MicroService{
+ AppId: "a",
+ Properties: map[string]string{
+ proto.PROP_ALLOW_CROSS_APP: "true",
+ },
+ }, ®istry.MicroService{
+ AppId: "a",
+ })
+ if err != nil {
+ t.Fatalf("AllowAcrossApp with the same appId and allow property
failed")
+ }
+
+ err = AllowAcrossDimension(context.Background(), ®istry.MicroService{
+ AppId: "a",
+ Properties: map[string]string{
+ proto.PROP_ALLOW_CROSS_APP: "true",
+ },
+ }, ®istry.MicroService{
+ AppId: "b",
+ })
+ if err != nil {
+ t.Fatalf("AllowAcrossApp with the diff appId and allow property
failed")
+ }
+
+ err = AllowAcrossDimension(context.Background(), ®istry.MicroService{
+ AppId: "a",
+ Properties: map[string]string{
+ proto.PROP_ALLOW_CROSS_APP: "false",
+ },
+ }, ®istry.MicroService{
+ AppId: "b",
+ })
+ if err == nil {
+ t.Fatalf("AllowAcrossApp with the diff appId and deny property
failed")
+ }
+
+ err = AllowAcrossDimension(context.Background(), ®istry.MicroService{
+ AppId: "a",
+ Properties: map[string]string{
+ proto.PROP_ALLOW_CROSS_APP: "",
+ },
+ }, ®istry.MicroService{
+ AppId: "b",
+ })
+ if err == nil {
+ t.Fatalf("AllowAcrossApp with the diff appId and empty property
failed")
+ }
+}
+
+func TestMatchRules(t *testing.T) {
+ err := MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "WHITE",
+ Attribute: "",
+ Pattern: "",
+ },
+ }, nil, nil)
+ if err == nil {
+ t.Fatalf("MatchRules nil failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "WHITE",
+ Attribute: "",
+ Pattern: "",
+ },
+ }, ®istry.MicroService{}, nil)
+ if err == nil {
+ t.Fatalf("MatchRules invalid WHITE failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "WHITE",
+ Attribute: "ServiceName",
+ Pattern: "^a.*",
+ },
+ }, ®istry.MicroService{
+ ServiceName: "a",
+ }, nil)
+ if err != nil {
+ t.Fatalf("MatchRules WHITE with field ServiceName failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "WHITE",
+ Attribute: "tag_a",
+ Pattern: "^b.*",
+ },
+ }, ®istry.MicroService{}, map[string]string{
+ "a": "b",
+ })
+ if err != nil {
+ t.Fatalf("MatchRules WHITE with tag b failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "WHITE",
+ Attribute: "tag_a",
+ Pattern: "^b.*",
+ },
+ }, ®istry.MicroService{}, map[string]string{
+ "a": "c",
+ })
+ if err == nil {
+ t.Fatalf("MatchRules WHITE with tag c failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "BLACK",
+ Attribute: "tag_a",
+ Pattern: "^b.*",
+ },
+ }, ®istry.MicroService{}, map[string]string{
+ "a": "b",
+ })
+ if err == nil {
+ t.Fatalf("MatchRules BLACK with tag b failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "BLACK",
+ Attribute: "ServiceName",
+ Pattern: "^a.*",
+ },
+ }, ®istry.MicroService{
+ ServiceName: "a",
+ }, nil)
+ if err == nil {
+ t.Fatalf("MatchRules BLACK with field ServiceName failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "BLACK",
+ Attribute: "tag_a",
+ Pattern: "^b.*",
+ },
+ }, ®istry.MicroService{}, map[string]string{
+ "a": "c",
+ })
+ if err != nil {
+ t.Fatalf("MatchRules BLACK with tag c failed")
+ }
+
+ err = MatchRules([]*registry.ServiceRule{
+ {
+ RuleType: "BLACK",
+ Attribute: "tag_a",
+ Pattern: "^b.*",
+ },
+ }, ®istry.MicroService{}, map[string]string{
+ "b": "b",
+ })
+ if err != nil {
+ t.Fatalf("MatchRules with not exist tag failed")
+ }
+}
+
+func TestGetConsumer(t *testing.T) {
+ _, _, err := GetAllProviderIds(context.Background(), "",
®istry.MicroService{})
+ if err != nil {
+ t.Fatalf("GetConsumerIdsByProvider invalid failed")
+ }
+
+ _, _, err = GetAllConsumerIds(context.Background(), "",
®istry.MicroService{
+ ServiceId: "a",
+ })
+ if err != nil {
+ t.Fatalf("GetConsumerIdsByProvider WithCacheOnly not exist
service failed")
+ }
+
+ _, err = GetConsumerIds(context.Background(), "",
+ ®istry.MicroService{
+ ServiceId: "a",
+ })
+ if err != nil {
+ t.Fatalf("GetConsumerIds WithCacheOnly failed")
+ }
+}
+
+func TestGetProvider(t *testing.T) {
+ _, err := GetProviderIds(context.Background(), "",
+ ®istry.MicroService{
+ ServiceId: "a",
+ })
+ if err != nil {
+ t.Fatalf("GetProviderIds WithCacheOnly failed")
+ }
+
+ _, _, err = GetAllProviderIds(context.Background(), "",
®istry.MicroService{})
+ if err != nil {
+ t.Fatalf("GetAllProviderIds WithCacheOnly failed")
+ }
+}
+
+func TestAccessible(t *testing.T) {
+ err := Accessible(context.Background(), "xxx", "")
+ if err.StatusCode() != http.StatusBadRequest {
+ t.Fatalf("Accessible invalid failed")
+ }
+}
diff --git a/server/plugin/tracing/tracing.go
b/datasource/etcd/util/schema_util.go
similarity index 67%
copy from server/plugin/tracing/tracing.go
copy to datasource/etcd/util/schema_util.go
index 309eeb4..298ef99 100644
--- a/server/plugin/tracing/tracing.go
+++ b/datasource/etcd/util/schema_util.go
@@ -15,27 +15,23 @@
* limitations under the License.
*/
-package tracing
+package util
import (
"context"
+
+ "github.com/apache/servicecomb-service-center/server/core/backend"
"github.com/apache/servicecomb-service-center/server/plugin/registry"
)
-const CtxTraceSpan = "x-trace-span"
-
-type Request interface{}
-type Span interface{}
-
-type Tracing interface {
- ServerBegin(operationName string, r Request) Span
- ServerEnd(span Span, code int, message string)
- ClientBegin(operationName string, r Request) Span
- ClientEnd(span Span, code int, message string)
-}
-
-type RegistryRequest struct {
- Ctx context.Context
- Endpoint string
- Options registry.PluginOp
+func CheckSchemaInfoExist(ctx context.Context, key string) (bool, error) {
+ opts := append(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
}
diff --git a/server/plugin/uuid/uuid.go
b/datasource/etcd/util/schema_util_test.go
similarity index 81%
copy from server/plugin/uuid/uuid.go
copy to datasource/etcd/util/schema_util_test.go
index c8aaa7c..f67f3cb 100644
--- a/server/plugin/uuid/uuid.go
+++ b/datasource/etcd/util/schema_util_test.go
@@ -15,15 +15,16 @@
* limitations under the License.
*/
-package uuid
+package util
import (
"context"
+ "testing"
)
-const ContextKey = "_uuid_key"
-
-type UUID interface {
- GetServiceID(ctx context.Context) string
- GetInstanceID(ctx context.Context) string
+func TestCheckSchemaInfoExist(t *testing.T) {
+ _, err := CheckSchemaInfoExist(context.Background(), "")
+ if err == nil {
+ t.Fatalf("TestCheckSchemaInfoExist failed")
+ }
}
diff --git a/datasource/etcd/util/tag_util.go b/datasource/etcd/util/tag_util.go
new file mode 100644
index 0000000..e63fb7c
--- /dev/null
+++ b/datasource/etcd/util/tag_util.go
@@ -0,0 +1,67 @@
+/*
+ * 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 util
+
+import (
+ "context"
+ "encoding/json"
+
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ "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/plugin/registry"
+ scerr "github.com/apache/servicecomb-service-center/server/scerror"
+)
+
+func AddTagIntoETCD(ctx context.Context, domainProject string, serviceID
string, dataTags map[string]string) *scerr.Error {
+ key := apt.GenerateServiceTagKey(domainProject, serviceID)
+ data, err := json.Marshal(dataTags)
+ if err != nil {
+ return scerr.NewError(scerr.ErrInternal, err.Error())
+ }
+
+ resp, err := backend.Registry().TxnWithCmp(ctx,
+ []registry.PluginOp{registry.OpPut(registry.WithStrKey(key),
registry.WithValue(data))},
+ []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 GetTagsUtils(ctx context.Context, domainProject, serviceID string) (tags
map[string]string, err error) {
+ key := apt.GenerateServiceTagKey(domainProject, serviceID)
+ opts := append(FromContext(ctx), registry.WithStrKey(key))
+ resp, err := backend.Store().ServiceTag().Search(ctx, opts...)
+ if err != nil {
+ log.Errorf(err, "get service[%s] tags file failed", serviceID)
+ return tags, err
+ }
+
+ if len(resp.Kvs) != 0 {
+ tags = resp.Kvs[0].Value.(map[string]string)
+ }
+ return tags, nil
+}
diff --git a/server/plugin/uuid/uuid.go b/datasource/etcd/util/tag_util_test.go
similarity index 69%
copy from server/plugin/uuid/uuid.go
copy to datasource/etcd/util/tag_util_test.go
index c8aaa7c..61afb76 100644
--- a/server/plugin/uuid/uuid.go
+++ b/datasource/etcd/util/tag_util_test.go
@@ -15,15 +15,23 @@
* limitations under the License.
*/
-package uuid
+package util
import (
"context"
+ "testing"
)
-const ContextKey = "_uuid_key"
+func TestAddTagIntoETCD(t *testing.T) {
+ err := AddTagIntoETCD(context.Background(), "", "",
map[string]string{"a": "1"})
+ if err == nil {
+ t.Fatalf(`AddTagIntoETCD with {"a": "1"} tags failed`)
+ }
+}
-type UUID interface {
- GetServiceID(ctx context.Context) string
- GetInstanceID(ctx context.Context) string
+func TestGetTagsUtils(t *testing.T) {
+ _, err := GetTagsUtils(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf(`GetTagsUtils failed`)
+ }
}
diff --git a/server/plugin/tracing/tracing.go b/datasource/etcd/util/util.go
similarity index 65%
copy from server/plugin/tracing/tracing.go
copy to datasource/etcd/util/util.go
index 309eeb4..aeaab00 100644
--- a/server/plugin/tracing/tracing.go
+++ b/datasource/etcd/util/util.go
@@ -15,27 +15,24 @@
* limitations under the License.
*/
-package tracing
+package util
import (
"context"
+ "github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/server/plugin/registry"
)
-const CtxTraceSpan = "x-trace-span"
-
-type Request interface{}
-type Span interface{}
-
-type Tracing interface {
- ServerBegin(operationName string, r Request) Span
- ServerEnd(span Span, code int, message string)
- ClientBegin(operationName string, r Request) Span
- ClientEnd(span Span, code int, message string)
-}
-
-type RegistryRequest struct {
- Ctx context.Context
- Endpoint string
- Options registry.PluginOp
+func FromContext(ctx context.Context) []registry.PluginOpOption {
+ opts := make([]registry.PluginOpOption, 0, 5)
+ switch {
+ case ctx.Value(util.CtxNocache) == "1":
+ opts = append(opts, registry.WithNoCache())
+ case ctx.Value(util.CtxCacheOnly) == "1":
+ opts = append(opts, registry.WithCacheOnly())
+ }
+ if ctx.Value(util.CtxGlobal) == "1" {
+ opts = append(opts, registry.WithGlobal())
+ }
+ return opts
}
diff --git a/datasource/etcd/util/util_suite_test.go
b/datasource/etcd/util/util_suite_test.go
new file mode 100644
index 0000000..716ff72
--- /dev/null
+++ b/datasource/etcd/util/util_suite_test.go
@@ -0,0 +1,140 @@
+/*
+ * 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 util_test
+
+import (
+ _
"github.com/apache/servicecomb-service-center/server/plugin/discovery/etcd"
+ _
"github.com/apache/servicecomb-service-center/server/plugin/quota/buildin"
+ _
"github.com/apache/servicecomb-service-center/server/plugin/registry/buildin"
+)
+
+import (
+ "context"
+ "testing"
+
+ serviceUtil
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
+ proto "github.com/apache/servicecomb-service-center/pkg/registry"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/plugin/registry"
+ . "github.com/onsi/ginkgo"
+ "github.com/onsi/ginkgo/reporters"
+ . "github.com/onsi/gomega"
+)
+
+func init() {
+}
+
+func TestMicroservice(t *testing.T) {
+ RegisterFailHandler(Fail)
+ junitReporter := reporters.NewJUnitReporter("model.junit.xml")
+ RunSpecsWithDefaultAndCustomReporters(t, "model Suite",
[]Reporter{junitReporter})
+}
+
+func TestFindServiceIds(t *testing.T) {
+ _, _, err := serviceUtil.FindServiceIds(context.Background(),
+ "latest", &proto.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf("TestFindServiceIds failed")
+ }
+
+ _, _, err = serviceUtil.FindServiceIds(context.Background(),
+ "1.0.0", &proto.MicroServiceKey{})
+ if err != nil {
+ t.Fatalf("TestFindServiceIds failed")
+ }
+
+ _, _, err = serviceUtil.FindServiceIds(context.Background(),
+ "1.0+", &proto.MicroServiceKey{Alias: "test"})
+ if err != nil {
+ t.Fatalf("TestFindServiceIds failed")
+ }
+}
+
+func TestGetService(t *testing.T) {
+ _, err := serviceUtil.GetService(context.Background(), "", "")
+ if err != nil {
+ t.Fatalf("TestGetService failed")
+ }
+
+ _, err = serviceUtil.GetServicesByDomainProject(context.Background(),
"")
+ if err != nil {
+ t.Fatalf("TestGetService failed")
+ }
+
+ _, err = serviceUtil.GetAllServiceUtil(context.Background())
+ if err != nil {
+ t.Fatalf("TestGetService failed")
+ }
+
+ _, err = serviceUtil.GetServiceWithRev(context.Background(), "", "", 0)
+ if err != nil {
+ t.Fatalf("TestGetService failed")
+ }
+
+ _, err = serviceUtil.GetServiceWithRev(context.Background(), "", "", 1)
+ if err != nil {
+ t.Fatalf("TestGetService failed")
+ }
+}
+
+func TestServiceExist(t *testing.T) {
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("TestServiceExist failed")
+ }
+ }()
+ serviceUtil.ServiceExist(util.SetContext(context.Background(),
util.CtxCacheOnly, "1"), "", "")
+}
+
+func TestFromContext(t *testing.T) {
+ ctx := context.WithValue(context.Background(), util.CtxNocache, "1")
+ opts := serviceUtil.FromContext(ctx)
+ if len(opts) == 0 {
+ t.Fatalf("TestFromContext failed")
+ }
+
+ op := registry.OptionsToOp(opts...)
+ if op.Mode != registry.ModeNoCache {
+ t.Fatalf("TestFromContext failed")
+ }
+
+ ctx = context.WithValue(context.Background(), util.CtxCacheOnly, "1")
+ opts = serviceUtil.FromContext(ctx)
+ if len(opts) == 0 {
+ t.Fatalf("TestFromContext failed")
+ }
+
+ op = registry.OptionsToOp(opts...)
+ if op.Mode != registry.ModeCache {
+ t.Fatalf("TestFromContext failed")
+ }
+}
+
+func TestRemandQuota(t *testing.T) {
+ serviceUtil.RemandServiceQuota(context.Background())
+ serviceUtil.RemandInstanceQuota(context.Background())
+}
+
+func TestSetDefault(t *testing.T) {
+ service := &proto.MicroService{}
+ serviceUtil.SetServiceDefaultValue(service)
+ if len(service.Level) == 0 ||
+ len(service.Status) == 0 {
+ t.Fatalf(`TestSetDefault failed`)
+ }
+}
diff --git a/datasource/etcd/util/versionrule.go
b/datasource/etcd/util/versionrule.go
new file mode 100644
index 0000000..1b3be1e
--- /dev/null
+++ b/datasource/etcd/util/versionrule.go
@@ -0,0 +1,264 @@
+/*
+ * 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 util
+
+import (
+ "regexp"
+ "sort"
+ "strconv"
+ "strings"
+
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ "github.com/apache/servicecomb-service-center/server/plugin/discovery"
+)
+
+type VersionRule func(sorted []string, kvs map[string]*discovery.KeyValue,
start, end string) []string
+
+func (vr VersionRule) Match(kvs []*discovery.KeyValue, ops ...string) []string
{
+ sorter := &serviceKeySorter{
+ sortArr: make([]string, len(kvs)),
+ kvs: make(map[string]*discovery.KeyValue, len(kvs)),
+ cmp: Larger,
+ }
+ for i, kv := range kvs {
+ key := util.BytesToStringWithNoCopy(kv.Key)
+ ver := key[strings.LastIndex(key, "/")+1:]
+ sorter.sortArr[i] = ver
+ sorter.kvs[ver] = kv
+ }
+ sort.Sort(sorter)
+
+ args := [2]string{}
+ switch {
+ case len(ops) > 1:
+ args[1] = ops[1]
+ fallthrough
+ case len(ops) > 0:
+ args[0] = ops[0]
+ }
+ return vr(sorter.sortArr, sorter.kvs, args[0], args[1])
+}
+
+type serviceKeySorter struct {
+ sortArr []string
+ kvs map[string]*discovery.KeyValue
+ cmp func(i, j string) bool
+}
+
+func (sks *serviceKeySorter) Len() int {
+ return len(sks.sortArr)
+}
+
+func (sks *serviceKeySorter) Swap(i, j int) {
+ sks.sortArr[i], sks.sortArr[j] = sks.sortArr[j], sks.sortArr[i]
+}
+
+func (sks *serviceKeySorter) Less(i, j int) bool {
+ return sks.cmp(sks.sortArr[i], sks.sortArr[j])
+}
+
+func VersionToInt64(versionStr string) (ret int64, err error) {
+ verBytes := [4]int16{}
+ idx := 0
+ for i := 0; i < 4 && idx < len(versionStr); i++ {
+ f := strings.IndexRune(versionStr[idx:], '.')
+ if f < 0 {
+ f = len(versionStr) - idx
+ }
+ integer, err := strconv.ParseInt(versionStr[idx:idx+f], 10, 16)
+ if err != nil {
+ return 0, err
+ }
+ verBytes[i] = int16(integer)
+ idx += f + 1
+ }
+ ret = util.Int16ToInt64(verBytes[:])
+ return
+}
+
+func Larger(start, end string) bool {
+ s, _ := VersionToInt64(start)
+ e, _ := VersionToInt64(end)
+ return s > e
+}
+
+func LessEqual(start, end string) bool {
+ return !Larger(start, end)
+}
+
+func Latest(sorted []string, kvs map[string]*discovery.KeyValue, start, end
string) []string {
+ if len(sorted) == 0 {
+ return []string{}
+ }
+ return []string{kvs[sorted[0]].Value.(string)}
+}
+
+func Range(sorted []string, kvs map[string]*discovery.KeyValue, start, end
string) []string {
+ result := make([]string, len(sorted))
+ i, flag := 0, 0
+
+ if Larger(start, end) {
+ start, end = end, start
+ }
+
+ l := len(sorted)
+ if l == 0 || Larger(start, sorted[0]) || LessEqual(end, sorted[l-1]) {
+ return []string{}
+ }
+
+ for _, k := range sorted {
+ // end >= k >= start
+ switch flag {
+ case 0:
+ if LessEqual(end, k) {
+ continue
+ }
+ flag = 1
+ case 1:
+ if Larger(start, k) {
+ return result[:i]
+ }
+ }
+
+ result[i] = kvs[k].Value.(string)
+ i++
+ }
+ return result[:i]
+}
+
+func AtLess(sorted []string, kvs map[string]*discovery.KeyValue, start, end
string) []string {
+ result := make([]string, len(sorted))
+
+ if len(sorted) == 0 || Larger(start, sorted[0]) {
+ return []string{}
+ }
+
+ for i, k := range sorted {
+ if Larger(start, k) {
+ return result[:i]
+ }
+ result[i] = kvs[k].Value.(string)
+ }
+ return result[:]
+}
+
+func ParseVersionRule(versionRule string) func(kvs []*discovery.KeyValue)
[]string {
+ if len(versionRule) == 0 {
+ return nil
+ }
+
+ rangeIdx := strings.Index(versionRule, "-")
+ switch {
+ case versionRule == "latest":
+ return func(kvs []*discovery.KeyValue) []string {
+ return VersionRule(Latest).Match(kvs)
+ }
+ case versionRule[len(versionRule)-1:] == "+":
+ // 取最低版本及高版本集合
+ start := versionRule[:len(versionRule)-1]
+ return func(kvs []*discovery.KeyValue) []string {
+ return VersionRule(AtLess).Match(kvs, start)
+ }
+ case rangeIdx > 0:
+ // 取版本范围集合
+ start := versionRule[:rangeIdx]
+ end := versionRule[rangeIdx+1:]
+ return func(kvs []*discovery.KeyValue) []string {
+ return VersionRule(Range).Match(kvs, start, end)
+ }
+ default:
+ // 精确匹配
+ return nil
+ }
+}
+
+func VersionMatchRule(version string, versionRule string) bool {
+ match := ParseVersionRule(versionRule)
+ if match == nil {
+ return version == versionRule
+ }
+
+ return len(match([]*discovery.KeyValue{
+ {
+ Key: util.StringToBytesWithNoCopy("/" + version),
+ Value: "",
+ },
+ })) > 0
+}
+
+type VersionRegexp struct {
+ Regex *regexp.Regexp
+ Fuzzy bool
+}
+
+func (vr *VersionRegexp) MatchString(s string) bool {
+ if vr.Regex != nil && !vr.Regex.MatchString(s) {
+ return false
+ }
+ return vr.validateVersionRule(s) == nil
+}
+
+func (vr *VersionRegexp) String() string {
+ if vr.Fuzzy {
+ return "the form x[.y[.z]] or x[.y[.z]]+ or x[.y[.z]]-x[.y[.z]]
or 'latest' where x y and z are 0-32767 range"
+ }
+ return "the form x[.y[.z]] where x y and z are 0-32767 range"
+}
+
+func (vr *VersionRegexp) validateVersionRule(versionRule string) (err error) {
+ if len(versionRule) == 0 {
+ return
+ }
+
+ if !vr.Fuzzy {
+ _, err = VersionToInt64(versionRule)
+ return
+ }
+
+ rangeIdx := strings.Index(versionRule, "-")
+ switch {
+ case versionRule == "latest":
+ return
+ case versionRule[len(versionRule)-1:] == "+":
+ // 取最低版本及高版本集合
+ start := versionRule[:len(versionRule)-1]
+ _, err = VersionToInt64(start)
+ case rangeIdx > 0:
+ // 取版本范围集合
+ start := versionRule[:rangeIdx]
+ end := versionRule[rangeIdx+1:]
+ _, err = VersionToInt64(start)
+ if err == nil {
+ _, err = VersionToInt64(end)
+ }
+ default:
+ // 精确匹配
+ _, err = VersionToInt64(versionRule)
+ }
+ return
+}
+
+func NewVersionRegexp(fuzzy bool) (vr *VersionRegexp) {
+ vr = &VersionRegexp{Fuzzy: fuzzy}
+ if fuzzy {
+ vr.Regex, _ =
regexp.Compile(`^\d+(\.\d+){0,3}\+?$|^\d+(\.\d+){0,3}-\d+(\.\d+){0,3}$|^latest$`)
+ return
+ }
+ vr.Regex, _ = regexp.Compile(`^\d+(\.\d+){0,3}$`)
+ return
+}
diff --git a/datasource/etcd/util/versionrule_test.go
b/datasource/etcd/util/versionrule_test.go
new file mode 100644
index 0000000..bd47d08
--- /dev/null
+++ b/datasource/etcd/util/versionrule_test.go
@@ -0,0 +1,413 @@
+/*
+ * 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 util
+
+import (
+ "fmt"
+ "reflect"
+ "sort"
+ "testing"
+
+ "github.com/apache/servicecomb-service-center/pkg/log"
+ "github.com/apache/servicecomb-service-center/server/plugin/discovery"
+ "github.com/stretchr/testify/assert"
+)
+
+const VERSIONRULE_BASE = 5000
+
+func BenchmarkVersionRule_Latest_GetServicesIds(b *testing.B) {
+ var kvs = make([]*discovery.KeyValue, VERSIONRULE_BASE)
+ for i := 1; i <= VERSIONRULE_BASE; i++ {
+ kvs[i-1] = &discovery.KeyValue{
+ Key: []byte(fmt.Sprintf("/service/ver/1.%d", i)),
+ Value: []byte(fmt.Sprintf("%d", i)),
+ }
+ }
+ b.N = VERSIONRULE_BASE
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ VersionRule(Latest).Match(kvs)
+ }
+ b.ReportAllocs()
+ // 5000 7105020 ns/op 2180198 B/op 39068 allocs/op
+}
+
+func BenchmarkVersionRule_Range_GetServicesIds(b *testing.B) {
+ var kvs = make([]*discovery.KeyValue, VERSIONRULE_BASE)
+ for i := 1; i <= VERSIONRULE_BASE; i++ {
+ kvs[i-1] = &discovery.KeyValue{
+ Key: []byte(fmt.Sprintf("/service/ver/1.%d", i)),
+ Value: []byte(fmt.Sprintf("%d", i)),
+ }
+ }
+ b.N = VERSIONRULE_BASE
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ VersionRule(Range).Match(kvs, fmt.Sprintf("1.%d", i),
fmt.Sprintf("1.%d", i+VERSIONRULE_BASE/10))
+ }
+ b.ReportAllocs()
+ // 5000 7244029 ns/op 2287389 B/op 39584 allocs/op
+}
+
+func BenchmarkVersionRule_AtLess_GetServicesIds(b *testing.B) {
+ var kvs = make([]*discovery.KeyValue, VERSIONRULE_BASE)
+ for i := 1; i <= VERSIONRULE_BASE; i++ {
+ kvs[i-1] = &discovery.KeyValue{
+ Key: []byte(fmt.Sprintf("/service/ver/1.%d", i)),
+ Value: []byte(fmt.Sprintf("%d", i)),
+ }
+ }
+ b.N = VERSIONRULE_BASE
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ VersionRule(AtLess).Match(kvs, fmt.Sprintf("1.%d", i))
+ }
+ b.ReportAllocs()
+ // 5000 11221098 ns/op 3174720 B/op 58064 allocs/op
+}
+
+func BenchmarkParseVersionRule(b *testing.B) {
+ f := ParseVersionRule("latest")
+ kvs := []*discovery.KeyValue{
+ {
+ Key: []byte("/service/ver/1.0.300"),
+ Value: "1.0.300",
+ },
+ {
+ Key: []byte("/service/ver/1.0.303"),
+ Value: "1.0.303",
+ },
+ {
+ Key: []byte("/service/ver/1.0.304"),
+ Value: "1.0.304",
+ },
+ }
+ b.ResetTimer()
+ b.RunParallel(func(pb *testing.PB) {
+ for pb.Next() {
+ f(kvs)
+ }
+ })
+ b.ReportAllocs()
+}
+
+func TestSorter(t *testing.T) {
+ log.Info("normal")
+
+ t.Run("version asc", func(t *testing.T) {
+ kvs := []string{"1.0.0", "1.0.1"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0.1", kvs[0])
+ assert.Equal(t, "1.0.0", kvs[1])
+ })
+
+ t.Run("version desc", func(t *testing.T) {
+ kvs := []string{"1.0.1", "1.0.0"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0.1", kvs[0])
+ assert.Equal(t, "1.0.0", kvs[1])
+ })
+
+ t.Run("len(v1) != len(v2)", func(t *testing.T) {
+ kvs := []string{"1.0.0.0", "1.0.1"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0.1", kvs[0])
+ assert.Equal(t, "1.0.0.0", kvs[1])
+ })
+
+ t.Run("1.0.9 vs 1.0.10", func(t *testing.T) {
+ kvs := []string{"1.0.9", "1.0.10"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0.10", kvs[0])
+ assert.Equal(t, "1.0.9", kvs[1])
+ })
+
+ t.Run("1.10 vs 4", func(t *testing.T) {
+ kvs := []string{"1.10", "4"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "4", kvs[0])
+ assert.Equal(t, "1.10", kvs[1])
+ })
+
+ log.Info("exception")
+
+ t.Run("invalid version1", func(t *testing.T) {
+ kvs := []string{"1.a", "1.0.1.a", ""}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.a", kvs[0])
+ assert.Equal(t, "1.0.1.a", kvs[1])
+ assert.Equal(t, "", kvs[2])
+ })
+
+ t.Run("invalid version2 > 32767", func(t *testing.T) {
+ kvs := []string{"1.0", "1.0.1.32768"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0", kvs[0])
+ assert.Equal(t, "1.0.1.32768", kvs[1])
+ kvs = []string{"1.0", "1.0.1.32767"}
+ sort.Sort(&serviceKeySorter{
+ sortArr: kvs,
+ kvs: make(map[string]*discovery.KeyValue),
+ cmp: Larger,
+ })
+ assert.Equal(t, "1.0.1.32767", kvs[0])
+ assert.Equal(t, "1.0", kvs[1])
+ })
+}
+
+func TestVersionRule(t *testing.T) {
+ const count = 10
+ var kvs = [count]*discovery.KeyValue{}
+ for i := 1; i <= count; i++ {
+ kvs[i-1] = &discovery.KeyValue{
+ Key: []byte(fmt.Sprintf("/service/ver/1.%d", i)),
+ Value: fmt.Sprintf("%d", i),
+ }
+ }
+
+ log.Info("normal")
+
+ t.Run("latest", func(t *testing.T) {
+ results := VersionRule(Latest).Match(kvs[:])
+ assert.Equal(t, 1, len(results))
+ assert.Equal(t, fmt.Sprintf("%d", count), results[0])
+ })
+
+ t.Run("range1.1 ver in [1.4, 1.8)", func(t *testing.T) {
+ results := VersionRule(Range).Match(kvs[:], "1.4", "1.8")
+ assert.Equal(t, 4, len(results))
+ assert.Equal(t, "7", results[0])
+ assert.Equal(t, "4", results[3])
+ })
+
+ t.Run("range1.2 ver in (1.8, 1.4]", func(t *testing.T) {
+ results := VersionRule(Range).Match(kvs[:], "1.8", "1.4")
+ assert.Equal(t, 4, len(results))
+ assert.Equal(t, "7", results[0])
+ assert.Equal(t, "4", results[3])
+ })
+
+ t.Run("range2 ver in [1, 2]", func(t *testing.T) {
+ results := VersionRule(Range).Match(kvs[:], "1", "2")
+ assert.Equal(t, 10, len(results))
+ assert.Equal(t, "10", results[0])
+ assert.Equal(t, "1", results[9])
+ })
+
+ t.Run("range3 ver in [1.4.1, 1.9.1]", func(t *testing.T) {
+ results := VersionRule(Range).Match(kvs[:], "1.4.1", "1.9.1")
+ assert.Equal(t, 5, len(results))
+ assert.Equal(t, "9", results[0])
+ assert.Equal(t, "5", results[4])
+ })
+
+ t.Run("range4 ver in [2, 4]", func(t *testing.T) {
+ results := VersionRule(Range).Match(kvs[:], "2", "4")
+ assert.Equal(t, len(results), 0)
+ })
+
+ t.Run("atLess1 ver >= 1.6", func(t *testing.T) {
+ results := VersionRule(AtLess).Match(kvs[:], "1.6")
+ assert.Equal(t, len(results), 5)
+ assert.Equal(t, "10", results[0])
+ assert.Equal(t, "6", results[4])
+ })
+
+ t.Run("atLess2 ver >= 1", func(t *testing.T) {
+ results := VersionRule(AtLess).Match(kvs[:], "1")
+ assert.Equal(t, len(results), 10)
+ assert.Equal(t, "10", results[0])
+ assert.Equal(t, "1", results[9])
+ })
+
+ t.Run("atLess3 ver >= 1.5.1", func(t *testing.T) {
+ results := VersionRule(AtLess).Match(kvs[:], "1.5.1")
+ assert.Equal(t, 5, len(results))
+ assert.Equal(t, "10", results[0])
+ assert.Equal(t, "6", results[4])
+ })
+
+ t.Run("atLess4 ver >= 2", func(t *testing.T) {
+ results := VersionRule(AtLess).Match(kvs[:], "2")
+ assert.Equal(t, 0, len(results))
+ })
+
+ log.Info("exception")
+
+ t.Run("nil", func(t *testing.T) {
+ results := VersionRule(Latest).Match(nil)
+ assert.Equal(t, 0, len(results))
+ results = VersionRule(AtLess).Match(nil)
+ assert.Equal(t, 0, len(results))
+ results = VersionRule(Range).Match(nil)
+ assert.Equal(t, 0, len(results))
+ rule := ParseVersionRule("")
+ assert.Equal(t, true, reflect.ValueOf(rule).IsNil())
+ rule = ParseVersionRule("abc")
+ assert.Equal(t, true, reflect.ValueOf(rule).IsNil())
+ assert.Equal(t, true, VersionMatchRule("1.0", "1.0"))
+ assert.Equal(t, false, VersionMatchRule("1.0", "1.2"))
+ })
+
+ log.Info("parse")
+
+ t.Run("latest", func(t *testing.T) {
+ match := ParseVersionRule("latest")
+ results := match(kvs[:])
+ assert.Equal(t, 1, len(results))
+ assert.Equal(t, fmt.Sprintf("%d", count), results[0])
+ })
+
+ t.Run("range ver in[1.4, 1.8]", func(t *testing.T) {
+ match := ParseVersionRule("1.4-1.8")
+ results := match(kvs[:])
+ assert.Equal(t, 4, len(results))
+ assert.Equal(t, "7", results[0])
+ assert.Equal(t, "4", results[3])
+ })
+
+ t.Run("atLess ver >= 1.6", func(t *testing.T) {
+ match := ParseVersionRule("1.6+")
+ results := match(kvs[:])
+ assert.Equal(t, 5, len(results))
+ assert.Equal(t, "10", results[0])
+ assert.Equal(t, "6", results[4])
+ })
+
+ log.Info("version match rule")
+
+ t.Run("latest", func(t *testing.T) {
+ assert.Equal(t, true, VersionMatchRule("1.0", "latest"))
+ })
+
+ t.Run("range ver in [1.4, 1.8]", func(t *testing.T) {
+ assert.Equal(t, true, VersionMatchRule("1.4", "1.4-1.8"))
+ assert.Equal(t, true, VersionMatchRule("1.6", "1.4-1.8"))
+ assert.Equal(t, false, VersionMatchRule("1.8", "1.4-1.8"))
+ assert.Equal(t, false, VersionMatchRule("1.0", "1.4-1.8"))
+ assert.Equal(t, false, VersionMatchRule("1.9", "1.4-1.8"))
+ })
+
+ t.Run("atLess ver >= 1.6", func(t *testing.T) {
+ assert.Equal(t, true, VersionMatchRule("1.6", "1.6+"))
+ assert.Equal(t, true, VersionMatchRule("1.9", "1.6+"))
+ assert.Equal(t, false, VersionMatchRule("1.0", "1.6+"))
+ })
+}
+
+func TestNewVersionRegexp(t *testing.T) {
+ log.Info("normal")
+
+ t.Run("latest", func(t *testing.T) {
+ vr := NewVersionRegexp(false)
+ assert.Equal(t, false, vr.MatchString("latest"))
+ vr = NewVersionRegexp(true)
+ assert.Equal(t, true, vr.MatchString("latest"))
+ })
+
+ t.Run("range", func(t *testing.T) {
+ vr := NewVersionRegexp(false)
+ assert.Equal(t, false, vr.MatchString("1.1-2.2"))
+ vr = NewVersionRegexp(true)
+ assert.Equal(t, false, vr.MatchString("-"))
+ assert.Equal(t, false, vr.MatchString("1.1-"))
+ assert.Equal(t, false, vr.MatchString("-1.1"))
+ assert.Equal(t, false, vr.MatchString("1.a-2.b"))
+ assert.Equal(t, false, vr.MatchString("1.-.2"))
+ assert.Equal(t, false, vr.MatchString("60000-1"))
+ assert.Equal(t, true, vr.MatchString("1.1-2.2"))
+ assert.Equal(t, true, vr.MatchString("1.1.1.1-2.2.2.2"))
+ assert.Equal(t, false, vr.MatchString("1.1.1.1.1-2.2.2.2"))
+ assert.Equal(t, false, vr.MatchString("1.1.1.1-2.2.2.2.2"))
+ })
+
+ t.Run("atLess", func(t *testing.T) {
+ vr := NewVersionRegexp(false)
+ assert.Equal(t, false, vr.MatchString("1.0+"))
+ vr = NewVersionRegexp(true)
+ assert.Equal(t, false, vr.MatchString("+"))
+ assert.Equal(t, false, vr.MatchString("+1.0"))
+ assert.Equal(t, false, vr.MatchString("1.a+"))
+ assert.Equal(t, false, vr.MatchString(".1+"))
+ assert.Equal(t, false, vr.MatchString("1.+"))
+ assert.Equal(t, false, vr.MatchString(".+"))
+ assert.Equal(t, false, vr.MatchString("60000+"))
+ assert.Equal(t, true, vr.MatchString("1.0+"))
+ assert.Equal(t, true, vr.MatchString("1.0.0.0+"))
+ assert.Equal(t, false, vr.MatchString("1.0.0.0.0+"))
+ })
+
+ t.Run("explicit", func(t *testing.T) {
+ vr := NewVersionRegexp(false)
+ assert.Equal(t, false, vr.MatchString(""))
+ assert.Equal(t, false, vr.MatchString("a"))
+ assert.Equal(t, false, vr.MatchString("60000"))
+ assert.Equal(t, false, vr.MatchString("."))
+ assert.Equal(t, false, vr.MatchString("1."))
+ assert.Equal(t, false, vr.MatchString(".1"))
+ assert.Equal(t, true, vr.MatchString("1.4"))
+ vr = NewVersionRegexp(true)
+ assert.Equal(t, false, vr.MatchString(""))
+ assert.Equal(t, false, vr.MatchString("a"))
+ assert.Equal(t, false, vr.MatchString("60000"))
+ assert.Equal(t, false, vr.MatchString("."))
+ assert.Equal(t, false, vr.MatchString("1."))
+ assert.Equal(t, false, vr.MatchString(".1"))
+ assert.Equal(t, true, vr.MatchString("1.4"))
+ assert.Equal(t, true, vr.MatchString("1.4.0.0"))
+ assert.Equal(t, false, vr.MatchString("1.4.0.0.0"))
+ })
+
+ log.Info("exception")
+
+ t.Run("MatchString & String", func(t *testing.T) {
+ vr := VersionRegexp{}
+ assert.Equal(t, true, vr.MatchString(""))
+ assert.NotEqual(t, "", vr.String())
+ vr = VersionRegexp{Fuzzy: true}
+ assert.Equal(t, true, vr.MatchString(""))
+ assert.NotEqual(t, "", vr.String())
+ })
+}
diff --git a/pkg/chain/chain_test.go b/pkg/chain/chain_test.go
index c21ad3a..ad6dd4c 100644
--- a/pkg/chain/chain_test.go
+++ b/pkg/chain/chain_test.go
@@ -87,7 +87,7 @@ type mockHandler struct {
}
func (h *mockHandler) Handle(i *Invocation) {
- x := i.Context().Value("x").(int)
+ x := i.Context().Value(util.CtxKey("x")).(int)
switch x {
case 1:
i.Success(x)
@@ -100,7 +100,7 @@ func (h *mockHandler) Handle(i *Invocation) {
case 4:
i.Next(WithAsyncFunc(func(r Result) {
i.WithContext("x", x*x)
- ch, _ := i.Context().Value("ch").(chan struct{})
+ ch, _ := i.Context().Value(util.CtxKey("ch")).(chan
struct{})
ch <- struct{}{}
}))
case 5:
@@ -132,7 +132,7 @@ func TestChain_Next(t *testing.T) {
i := NewInvocation(context.Background(), ch)
i.WithContext("x", x)
i.Invoke(func(r Result) {
- if !r.OK || i.Context().Value("x").(int) != -len(hs) {
+ if !r.OK || i.Context().Value(util.CtxKey("x")).(int) !=
-len(hs) {
t.Fatalf("TestChain_Next")
}
})
@@ -162,11 +162,11 @@ func TestChain_Next(t *testing.T) {
i = NewInvocation(context.Background(), ch)
i.WithContext("x", x)
i.Invoke(func(r Result) {
- if !r.OK || i.Context().Value("x").(int) != x {
+ if !r.OK || i.Context().Value(util.CtxKey("x")).(int) != x {
t.Fatalf("TestChain_Next")
}
})
- if i.Context().Value("x").(int) != x*x {
+ if i.Context().Value(util.CtxKey("x")).(int) != x*x {
t.Fatalf("TestChain_Next")
}
@@ -176,12 +176,12 @@ func TestChain_Next(t *testing.T) {
i.WithContext("x", x)
i.WithContext("ch", make(chan struct{}))
i.Invoke(func(r Result) {
- if !r.OK || i.Context().Value("x").(int) != x {
+ if !r.OK || i.Context().Value(util.CtxKey("x")).(int) != x {
t.Fatalf("TestChain_Next")
}
})
- <-i.Context().Value("ch").(chan struct{})
- if i.Context().Value("x").(int) != x*x {
+ <-i.Context().Value(util.CtxKey("ch")).(chan struct{})
+ if i.Context().Value(util.CtxKey("x")).(int) != x*x {
t.Fatalf("TestChain_Next")
}
diff --git a/pkg/chain/invocation.go b/pkg/chain/invocation.go
index 6d87500..9701ebd 100644
--- a/pkg/chain/invocation.go
+++ b/pkg/chain/invocation.go
@@ -53,7 +53,7 @@ func (i *Invocation) Context() context.Context {
return i.context
}
-func (i *Invocation) WithContext(key string, val interface{}) *Invocation {
+func (i *Invocation) WithContext(key util.CtxKey, val interface{}) *Invocation
{
i.context.SetKV(key, val)
return i
}
diff --git a/pkg/rest/common.go b/pkg/rest/common.go
index 0e364e8..919ebff 100644
--- a/pkg/rest/common.go
+++ b/pkg/rest/common.go
@@ -19,6 +19,8 @@ package rest
import (
"net/http"
+
+ "github.com/apache/servicecomb-service-center/pkg/util"
)
const (
@@ -27,10 +29,10 @@ const (
HTTPMethodPost = http.MethodPost
HTTPMethodDelete = http.MethodDelete
- CtxResponse = "_server_response"
- CtxRequest = "_server_request"
- CtxMatchPattern = "_server_match_pattern"
- CtxMatchFunc = "_server_match_func"
+ CtxResponse util.CtxKey = "_server_response"
+ CtxRequest util.CtxKey = "_server_request"
+ CtxMatchPattern util.CtxKey = "_server_match_pattern"
+ CtxMatchFunc util.CtxKey = "_server_match_func"
ServerChainName = "_server_chain"
diff --git a/pkg/util/common.go b/pkg/util/common.go
index a876bc9..93ed11a 100644
--- a/pkg/util/common.go
+++ b/pkg/util/common.go
@@ -17,11 +17,13 @@
package util
+type CtxKey string
+
const (
- HeaderRev = "X-Resource-Revision"
- CtxGlobal = "global"
- CtxNocache = "noCache"
- CtxCacheOnly = "cacheOnly"
- CtxRequestRevision = "requestRev"
- CtxResponseRevision = "responseRev"
+ HeaderRev = "X-Resource-Revision"
+ CtxGlobal CtxKey = "global"
+ CtxNocache CtxKey = "noCache"
+ CtxCacheOnly CtxKey = "cacheOnly"
+ CtxRequestRevision CtxKey = "requestRev"
+ CtxResponseRevision CtxKey = "responseRev"
)
diff --git a/pkg/util/context.go b/pkg/util/context.go
index d1d4ceb..7fbec0c 100644
--- a/pkg/util/context.go
+++ b/pkg/util/context.go
@@ -48,7 +48,7 @@ func (c *StringContext) Err() error {
}
func (c *StringContext) Value(key interface{}) interface{} {
- k, ok := key.(string)
+ k, ok := key.(CtxKey)
if !ok {
return c.parentCtx.Value(key)
}
@@ -59,7 +59,7 @@ func (c *StringContext) Value(key interface{}) interface{} {
return v
}
-func (c *StringContext) SetKV(key string, val interface{}) {
+func (c *StringContext) SetKV(key CtxKey, val interface{}) {
c.kv.Put(key, val)
}
@@ -74,7 +74,7 @@ func NewStringContext(ctx context.Context) *StringContext {
return strCtx
}
-func SetContext(ctx context.Context, key string, val interface{})
context.Context {
+func SetContext(ctx context.Context, key CtxKey, val interface{})
context.Context {
strCtx := NewStringContext(ctx)
strCtx.SetKV(key, val)
return strCtx
@@ -101,14 +101,14 @@ func CloneContext(ctx context.Context) context.Context {
return strCtx
}
-func FromContext(ctx context.Context, key string) interface{} {
+func FromContext(ctx context.Context, key CtxKey) interface{} {
if v := ctx.Value(key); v != nil {
return v
}
return FromMetadata(ctx, key)
}
-func SetRequestContext(r *http.Request, key string, val interface{})
*http.Request {
+func SetRequestContext(r *http.Request, key CtxKey, val interface{})
*http.Request {
ctx := r.Context()
ctx = SetContext(ctx, key, val)
if ctx != r.Context() {
diff --git a/pkg/util/context_grpc.go b/pkg/util/context_grpc.go
index bbb10f8..53f68b0 100644
--- a/pkg/util/context_grpc.go
+++ b/pkg/util/context_grpc.go
@@ -20,12 +20,12 @@ import (
"google.golang.org/grpc/metadata"
)
-func FromMetadata(ctx context.Context, key string) string {
+func FromMetadata(ctx context.Context, key CtxKey) string {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return ""
}
- if values, ok := md[key]; ok && len(values) > 0 {
+ if values, ok := md[string(key)]; ok && len(values) > 0 {
return values[0]
}
return ""
diff --git a/server/handler/cache/cache.go b/server/handler/cache/cache.go
index 8d3072f..6d407ca 100644
--- a/server/handler/cache/cache.go
+++ b/server/handler/cache/cache.go
@@ -24,6 +24,12 @@ import (
"net/http"
)
+const (
+ queryGlobal = "global"
+ queryNoCache = "noCache"
+ queryCacheOnly = "cacheOnly"
+)
+
type Handler struct {
}
@@ -33,18 +39,18 @@ func (l *Handler) Handle(i *chain.Invocation) {
r := i.Context().Value(rest.CtxRequest).(*http.Request)
query := r.URL.Query()
- global := util.StringTRUE(query.Get(string(util.CtxGlobal)))
+ global := util.StringTRUE(query.Get(queryGlobal))
if global && r.Method == http.MethodGet {
i.WithContext(util.CtxGlobal, "1")
}
- noCache := util.StringTRUE(query.Get(util.CtxNocache))
+ noCache := util.StringTRUE(query.Get(queryNoCache))
if noCache {
i.WithContext(util.CtxNocache, "1")
return
}
- cacheOnly := util.StringTRUE(query.Get(string(util.CtxCacheOnly)))
+ cacheOnly := util.StringTRUE(query.Get(queryCacheOnly))
if cacheOnly {
i.WithContext(util.CtxCacheOnly, "1")
return
diff --git a/server/plugin/tracing/pzipkin/buildin.go
b/server/plugin/tracing/pzipkin/buildin.go
index 1cfbc00..c2530da 100644
--- a/server/plugin/tracing/pzipkin/buildin.go
+++ b/server/plugin/tracing/pzipkin/buildin.go
@@ -150,7 +150,7 @@ func (zp *Zipkin) ClientBegin(operationName string, itf
tracing.Request) tracing
}
// inject context
err := carrier.ForeachKey(func(key, val string) error {
- ctx = util.SetContext(ctx, key, val)
+ ctx = util.SetContext(ctx, util.CtxKey(key), val)
return nil
})
if err != nil {
diff --git a/server/plugin/tracing/tracing.go b/server/plugin/tracing/tracing.go
index 309eeb4..9171005 100644
--- a/server/plugin/tracing/tracing.go
+++ b/server/plugin/tracing/tracing.go
@@ -19,10 +19,11 @@ package tracing
import (
"context"
+ "github.com/apache/servicecomb-service-center/pkg/util"
"github.com/apache/servicecomb-service-center/server/plugin/registry"
)
-const CtxTraceSpan = "x-trace-span"
+const CtxTraceSpan util.CtxKey = "x-trace-span"
type Request interface{}
type Span interface{}
diff --git a/server/plugin/uuid/uuid.go b/server/plugin/uuid/uuid.go
index c8aaa7c..1f0a6c5 100644
--- a/server/plugin/uuid/uuid.go
+++ b/server/plugin/uuid/uuid.go
@@ -19,9 +19,11 @@ package uuid
import (
"context"
+
+ "github.com/apache/servicecomb-service-center/pkg/util"
)
-const ContextKey = "_uuid_key"
+const ContextKey util.CtxKey = "_uuid_key"
type UUID interface {
GetServiceID(ctx context.Context) string
diff --git a/server/rest/handler.go b/server/rest/handler.go
index dc22b10..26280d1 100644
--- a/server/rest/handler.go
+++ b/server/rest/handler.go
@@ -26,7 +26,7 @@ import (
"github.com/apache/servicecomb-service-center/server/interceptor"
)
-const CtxStartTimestamp = "x-start-timestamp"
+const CtxStartTimestamp util.CtxKey = "x-start-timestamp"
func init() {
// api
