This is an automated email from the ASF dual-hosted git repository.
robotljw 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 080cdf0 [feat] add dep sync func when db is etcd
new 160431c Merge pull request #1200 from robotLJW/master
080cdf0 is described below
commit 080cdf03a4359b513bde30f078ee6180cb4ba4d8
Author: robotljw <[email protected]>
AuthorDate: Fri Dec 31 17:35:18 2021 +0800
[feat] add dep sync func when db is etcd
---
datasource/common.go | 5 +-
datasource/etcd/dep.go | 14 ++++-
datasource/etcd/dep_test.go | 134 ++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 149 insertions(+), 4 deletions(-)
diff --git a/datasource/common.go b/datasource/common.go
index f783594..d391864 100644
--- a/datasource/common.go
+++ b/datasource/common.go
@@ -32,8 +32,9 @@ const (
RegistryAppID = "default"
Provider = "p"
- ResourceAccount = "account"
- ResourceRole = "role"
+ ResourceAccount = "account"
+ ResourceRole = "role"
+ ResourceDependency = "dependency"
)
// WrapErrResponse is temp func here to wait finish to refact the discosvc pkg
diff --git a/datasource/etcd/dep.go b/datasource/etcd/dep.go
index ba3c837..a9faa86 100644
--- a/datasource/etcd/dep.go
+++ b/datasource/etcd/dep.go
@@ -23,6 +23,10 @@ import (
"errors"
"fmt"
+ pb "github.com/go-chassis/cari/discovery"
+ "github.com/go-chassis/cari/sync"
+ "github.com/little-cui/etcdadpt"
+
"github.com/apache/servicecomb-service-center/datasource"
"github.com/apache/servicecomb-service-center/datasource/etcd/event"
"github.com/apache/servicecomb-service-center/datasource/etcd/path"
@@ -30,8 +34,6 @@ import (
serviceUtil
"github.com/apache/servicecomb-service-center/datasource/etcd/util"
"github.com/apache/servicecomb-service-center/pkg/log"
"github.com/apache/servicecomb-service-center/pkg/util"
- pb "github.com/go-chassis/cari/discovery"
- "github.com/little-cui/etcdadpt"
)
type DepManager struct {
@@ -162,6 +164,14 @@ func (dm *DepManager) AddOrUpdateDependencies(ctx
context.Context, dependencyInf
opts = append(opts, etcdadpt.OpPut(etcdadpt.WithStrKey(key),
etcdadpt.WithValue(data)))
}
+ if datasource.EnableSync {
+ taskOpt, err := GenTaskOpts("", "", sync.UpdateAction,
datasource.ResourceDependency, dependencyInfos)
+ if err != nil {
+ log.Error("", err)
+ return pb.CreateResponse(pb.ErrInternal, err.Error()),
err
+ }
+ opts = append(opts, taskOpt)
+ }
err := etcdadpt.Txn(ctx, opts)
if err != nil {
log.Error(fmt.Sprintf("put request into dependency queue
failed, override: %t, %v",
diff --git a/datasource/etcd/dep_test.go b/datasource/etcd/dep_test.go
new file mode 100644
index 0000000..7dc8c72
--- /dev/null
+++ b/datasource/etcd/dep_test.go
@@ -0,0 +1,134 @@
+/*
+ * 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 etcd_test
+
+import (
+ "context"
+ "testing"
+
+ pb "github.com/go-chassis/cari/discovery"
+ "github.com/stretchr/testify/assert"
+
+ "github.com/apache/servicecomb-service-center/datasource"
+ "github.com/apache/servicecomb-service-center/eventbase/model"
+ "github.com/apache/servicecomb-service-center/eventbase/service/task"
+ "github.com/apache/servicecomb-service-center/pkg/util"
+ _ "github.com/apache/servicecomb-service-center/test"
+)
+
+func depGetContext() context.Context {
+ return util.WithNoCache(util.SetDomainProject(context.Background(),
"sync-dep", "sync-dep"))
+}
+
+func TestSyncAddOrUpdateDependencies(t *testing.T) {
+ datasource.EnableSync = true
+ var (
+ consumerId string
+ providerId string
+ )
+
+ t.Run("register service", func(t *testing.T) {
+ t.Run("create a consumer service should pass", func(t
*testing.T) {
+ resp, err :=
datasource.GetMetadataManager().RegisterService(depGetContext(),
&pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "sync_dep_group",
+ ServiceName: "sync_dep_consumer",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ },
+ })
+ assert.NotNil(t, resp)
+ assert.NoError(t, err)
+ assert.Equal(t, pb.ResponseSuccess,
resp.Response.GetCode())
+ consumerId = resp.ServiceId
+ })
+ t.Run("create a provider service should pass", func(t
*testing.T) {
+ resp, err :=
datasource.GetMetadataManager().RegisterService(depGetContext(),
&pb.CreateServiceRequest{
+ Service: &pb.MicroService{
+ AppId: "sync_dep_group",
+ ServiceName: "sync_dep_provider",
+ Version: "1.0.0",
+ Level: "FRONT",
+ Status: pb.MS_UP,
+ },
+ })
+ assert.NotNil(t, resp)
+ assert.NoError(t, err)
+ assert.Equal(t, pb.ResponseSuccess,
resp.Response.GetCode())
+ providerId = resp.ServiceId
+ })
+ })
+
+ t.Run("AddOrUpdateDependencies", func(t *testing.T) {
+ t.Run("add dependencies should pass", func(t *testing.T) {
+ consumer := &pb.MicroServiceKey{
+ ServiceName: "sync_dep_consumer",
+ AppId: "sync_dep_group",
+ Version: "1.0.0",
+ }
+ resp, err :=
datasource.GetDependencyManager().AddOrUpdateDependencies(depGetContext(),
[]*pb.ConsumerDependency{
+ {
+ Consumer: consumer,
+ Providers: []*pb.MicroServiceKey{
+ {
+ AppId:
"sync_dep_group",
+ ServiceName:
"sync_dep_provider",
+ },
+ },
+ },
+ }, false)
+ assert.NotNil(t, resp)
+ assert.NoError(t, err)
+ assert.Equal(t, pb.ResponseSuccess, resp.GetCode())
+ listTaskReq := model.ListTaskRequest{
+ Domain: "",
+ Project: "",
+ ResourceType: datasource.ResourceDependency,
+ }
+ tasks, err := task.List(context.Background(),
&listTaskReq)
+ assert.NoError(t, err)
+ assert.Equal(t, 1, len(tasks))
+ err = task.Delete(context.Background(), tasks...)
+ assert.NoError(t, err)
+ tasks, err = task.List(context.Background(),
&listTaskReq)
+ assert.NoError(t, err)
+ assert.Equal(t, 0, len(tasks))
+ })
+ })
+
+ t.Run("unregister consumer and provider", func(t *testing.T) {
+ t.Run("unregister consumer and provider should pass", func(t
*testing.T) {
+ respDelP, err :=
datasource.GetMetadataManager().UnregisterService(depGetContext(),
&pb.DeleteServiceRequest{
+ ServiceId: consumerId, Force: true,
+ })
+ assert.NotNil(t, respDelP)
+ assert.NoError(t, err)
+ assert.Equal(t, pb.ResponseSuccess,
respDelP.Response.GetCode())
+
+ respDelP, err =
datasource.GetMetadataManager().UnregisterService(depGetContext(),
&pb.DeleteServiceRequest{
+ ServiceId: providerId, Force: true,
+ })
+ assert.NotNil(t, respDelP)
+ assert.NoError(t, err)
+ assert.Equal(t, pb.ResponseSuccess,
respDelP.Response.GetCode())
+
+ })
+ })
+ datasource.EnableSync = false
+}
