Feature: adding, deleting and modifying operations to add tasks when …

…synchronization is turned on (#228)

.github/workflows/mongo_storage.yml

@@ -13,11 +13,12 @@ jobs:
       uses: actions/checkout@v1
     - name: UT
       run: |
-        cd build
-        time bash build_docker.sh
-        cd ../
-        sudo docker-compose -f ./deployments/docker/docker-compose.yaml up -d
+        sudo docker-compose -f ./examples/dev/docker-compose.yaml up -d
         sleep 20
         export TEST_DB_KIND=mongo
         export TEST_DB_URI=mongodb://kie:[email protected]:27017/kie
-        time go test -v $(go list ./... | grep -v etcd |  grep -v third_party | grep -v examples)
+        time go test -v $(go list ./... | grep -v etcd |  grep -v third_party | grep -v examples)
+    - name: Build kie images
+      run: |
+        cd build
+        time bash build_docker.sh

examples/dev/db.js

@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+db.createUser(
+    {
+        user: "kie",
+        pwd: "123",
+        roles: [
+            {
+                role: "readWrite",
+                db: "kie"
+            }
+        ]
+    }
+);

examples/dev/docker-compose.yaml

@@ -14,22 +14,19 @@
 ## See the License for the specific language governing permissions and
 ## limitations under the License.
 ## ---------------------------------------------------------------------------
-version: '3.1'
+version: '3.3'
 services:
   mongo:
-    image: mongo:4.0
+    image: mongo:4.2
+    command: mongod --replSet my-replica-set --port 27017 --bind_ip_all
     ports:
       - 27017:27017
     environment:
+      MONGO_INITDB_DATABASE: kie
       MONGO_INITDB_ROOT_USERNAME: kie
       MONGO_INITDB_ROOT_PASSWORD: 123
-      MONGO_INITDB_DATABASE: kie
     volumes:
-      - ../../deployments/db.js:/docker-entrypoint-initdb.d/db.js:ro
-  mongo-express:
-    image: mongo-express
-    ports:
-      - 8081:8081
-    environment:
-      ME_CONFIG_MONGODB_ADMINUSERNAME: kie
-      ME_CONFIG_MONGODB_ADMINPASSWORD: 123
+      - ./db.js:/docker-entrypoint-initdb.d/db.js:ro
+    healthcheck:
+      test: test $$(echo "rs.initiate({_id:'my-replica-set',members:[{_id:0,host:\"127.0.0.1:27017\"}]}).ok || rs.status().ok" | mongo -u $${MONGO_INITDB_ROOT_USERNAME} -p $${MONGO_INITDB_ROOT_PASSWORD}  --port 27017 --quiet) -eq 1
+      interval: 10s

examples/dev/kie-conf.yaml

@@ -7,6 +7,9 @@ db:
   #   kind=embedded_etcd, then is the embedded etcd server's advertise-peer-urls, e.g. default=http://127.0.0.1:2380
   #uri: mongodb://kie:[email protected]:27017/kie
   uri: http://127.0.0.1:2379
+sync:
+  # turn on the synchronization switch related operations will be written to the task in the db
+  enabled: false
 #  poolSize: 10
 #  timeout: 5m
 #  sslEnabled: false

go.mod

@@ -2,12 +2,12 @@ module github.com/apache/servicecomb-kie
 
 require (
 	github.com/emicklei/go-restful v2.12.0+incompatible
-	github.com/go-chassis/cari v0.5.0
+	github.com/go-chassis/cari v0.5.1-0.20211127060616-302dc831a28f
 	github.com/go-chassis/foundation v0.3.1-0.20210811025651-7f4d2b2b906c
 	github.com/go-chassis/go-archaius v1.5.2-0.20210301074935-e4694f6b077b
-	github.com/go-chassis/go-chassis/v2 v2.3.0
+	github.com/go-chassis/go-chassis/v2 v2.3.1-0.20211122011324-0a97efcf095b
 	github.com/go-chassis/openlog v1.1.3
-	github.com/go-chassis/seclog v1.3.0
+	github.com/go-chassis/seclog v1.3.1-0.20210917082355-52c40864f240
 	github.com/gofrs/uuid v4.0.0+incompatible
 	github.com/hashicorp/serf v0.9.5
 	github.com/little-cui/etcdadpt v0.2.1

server/config/config.go

@@ -47,3 +47,8 @@ func GetDB() DB {
 func GetRBAC() RBAC {
 	return Configurations.RBAC
 }
+
+// GetSync return sync config
+func GetSync() Sync {
+	return Configurations.Sync
+}

server/config/struct.go

@@ -21,6 +21,7 @@ package config
 type Config struct {
 	DB   DB   `yaml:"db"`
 	RBAC RBAC `yaml:"rbac"`
+	Sync Sync `yaml:"sync"`
 	//config from cli
 	ConfigFile     string
 	NodeName       string
@@ -48,3 +49,8 @@ type RBAC struct {
 	Enabled    bool   `yaml:"enabled"`
 	PubKeyFile string `yaml:"rsaPublicKeyFile"`
 }
+
+// Sync is sync config
+type Sync struct {
+	Enabled bool `yaml:"enabled"`
+}

server/datasource/dao.go

@@ -24,9 +24,12 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/go-chassis/cari/sync"
+	"github.com/go-chassis/openlog"
+	"github.com/gofrs/uuid"
+
 	"github.com/apache/servicecomb-kie/pkg/model"
 	"github.com/apache/servicecomb-kie/server/config"
-	"github.com/go-chassis/openlog"
 )
 
 var (
@@ -38,14 +41,15 @@ var (
 	ErrKeyNotExists     = errors.New("can not find any key value")
 	ErrRecordNotExists  = errors.New("can not find any polling data")
 	ErrRevisionNotExist = errors.New("revision does not exist")
-	ErrAliasNotGiven    = errors.New("label alias not given")
 	ErrKVAlreadyExists  = errors.New("kv already exists")
 	ErrTooMany          = errors.New("key with labels should be only one")
 )
 
 const (
 	DefaultValueType = "text"
 	MaxHistoryNum    = 100
+
+	ConfigResource = "config"
 )
 
 //New init db session
@@ -70,13 +74,14 @@ func GetBroker() Broker {
 //KVDao provide api of KV entity
 type KVDao interface {
 	// Create Update List are usually for admin console
-	Create(ctx context.Context, kv *model.KVDoc) (*model.KVDoc, error)
-	Update(ctx context.Context, kv *model.KVDoc) error
+	Create(ctx context.Context, kv *model.KVDoc, options ...WriteOption) (*model.KVDoc, error)
+	Update(ctx context.Context, kv *model.KVDoc, options ...WriteOption) error
 	List(ctx context.Context, project, domain string, options ...FindOption) (*model.KVResponse, error)
 	//FindOneAndDelete deletes one kv by id and return the deleted kv as these appeared before deletion
-	FindOneAndDelete(ctx context.Context, kvID string, project, domain string) (*model.KVDoc, error)
+	FindOneAndDelete(ctx context.Context, kvID string, project, domain string, options ...WriteOption) (*model.KVDoc, error)
 	//FindManyAndDelete deletes multiple kvs and return the deleted kv list as these appeared before deletion
-	FindManyAndDelete(ctx context.Context, kvIDs []string, project, domain string) ([]*model.KVDoc, int64, error)
+	FindManyAndDelete(ctx context.Context, kvIDs []string, project, domain string, options ...WriteOption) ([]*model.KVDoc, int64, error)
+
 	//Get return kv by id
 	Get(ctx context.Context, req *model.GetKVRequest) (*model.KVDoc, error)
 	Exist(ctx context.Context, key, project, domain string, options ...FindOption) (bool, error)
@@ -157,3 +162,29 @@ func ClearPart(kv *model.KVDoc) {
 	kv.Project = ""
 	kv.LabelFormat = ""
 }
+
+// NewTask return task with action and datatype
+func NewTask(domain, project, action, dataType string) (*sync.Task, error) {
+	taskId, err := uuid.NewV4()
+	if err != nil {
+		return nil, err
+	}
+	return &sync.Task{
+		TaskID:    taskId.String(),
+		Action:    action,
+		DataType:  dataType,
+		Domain:    domain,
+		Project:   project,
+		Timestamp: time.Now().Unix(),
+	}, nil
+}
+
+// NewTombstone return tombstone with resourceType ,domain and project
+func NewTombstone(domain, project, resourceType string) *sync.Tombstone {
+	return &sync.Tombstone{
+		ResourceType: resourceType,
+		Domain:       domain,
+		Project:      project,
+		Timestamp:    time.Now().Unix(),
+	}
+}

server/datasource/etcd/history/history_dao.go

@@ -21,11 +21,12 @@ import (
 	"context"
 	"encoding/json"
 
+	"github.com/go-chassis/openlog"
+	"github.com/little-cui/etcdadpt"
+
 	"github.com/apache/servicecomb-kie/pkg/model"
 	"github.com/apache/servicecomb-kie/server/datasource"
 	"github.com/apache/servicecomb-kie/server/datasource/etcd/key"
-	"github.com/go-chassis/openlog"
-	"github.com/little-cui/etcdadpt"
 )
 
 //Dao is the implementation

server/datasource/etcd/init.go

@@ -69,6 +69,7 @@ func (*Broker) GetHistoryDao() datasource.HistoryDao {
 func (*Broker) GetTrackDao() datasource.TrackDao {
 	return &track.Dao{}
 }
+
 func init() {
 	datasource.RegisterPlugin("etcd", NewFrom)
 	datasource.RegisterPlugin("embedded_etcd", NewFrom)

server/datasource/etcd/key/key.go

@@ -28,8 +28,28 @@ const (
 	keyCounter = "counter"
 	keyHistory = "kv-history"
 	keyTrack   = "track"
+	syncer     = "syncer"
+	task       = "task"
+	tombstone  = "tombstone"
 )
 
+func getSyncRootKey() string {
+	return split + syncer + split + task
+}
+
+func getTombstoneRootKey() string {
+	return split + tombstone
+}
+
+func TaskKey(domain, project, taskID string, timestamp int64) string {
+	strTimestamp := strconv.FormatInt(timestamp, 10)
+	return strings.Join([]string{getSyncRootKey(), domain, project, strTimestamp, taskID}, split)
+}
+
+func TombstoneKey(domain, project, resourceType, resourceID string) string {
+	return strings.Join([]string{getTombstoneRootKey(), domain, project, resourceType, resourceID}, split)
+}
+
 func KV(domain, project, kvID string) string {
 	return strings.Join([]string{keyKV, domain, project, kvID}, split)
 }