add mongo distributed lock

cmd/layotto/main.go

@@ -97,6 +97,7 @@ import (
 	lock_etcd "mosn.io/layotto/components/lock/etcd"
 	lock_redis "mosn.io/layotto/components/lock/redis"
 	lock_zookeeper "mosn.io/layotto/components/lock/zookeeper"
+	lock_mongo "mosn.io/layotto/components/lock/mongo"
 	runtime_lock "mosn.io/layotto/pkg/runtime/lock"
 
 	// Sequencer
@@ -289,6 +290,9 @@ func NewRuntimeGrpcServer(data json.RawMessage, opts ...grpc.ServerOption) (mgrp
 			runtime_lock.NewFactory("etcd", func() lock.LockStore {
 				return lock_etcd.NewEtcdLock(log.DefaultLogger)
 			}),
+			runtime_lock.NewFactory("mongo", func() lock.LockStore {
+				return lock_mongo.NewMongoLock(log.DefaultLogger)
+			}),
 		),
 
 		// bindings

components/go.mod

@@ -26,6 +26,7 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.0
 	go.etcd.io/etcd/client/v3 v3.5.0
 	go.etcd.io/etcd/server/v3 v3.5.0
+	go.mongodb.org/mongo-driver v1.8.0
 	golang.org/x/net v0.0.0-20210614182718-04defd469f4e // indirect
 	golang.org/x/oauth2 v0.0.0-20201208152858-08078c50e5b5 // indirect
 	google.golang.org/grpc v1.38.0

components/lock/mongo/mongo_lock.go

@@ -0,0 +1,214 @@
+package mongo
+
+import (
+	"context"
+	"fmt"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
+	"go.mongodb.org/mongo-driver/x/bsonx"
+	"mosn.io/layotto/components/lock"
+	"mosn.io/layotto/components/pkg/utils"
+	"mosn.io/pkg/log"
+	"time"
+)
+
+// mongo lock store
+type MongoLock struct {
+	client     *mongo.Client
+	collection *mongo.Collection
+	database   *mongo.Database
+	metadata   utils.MongoMetadata
+
+	features []lock.Feature
+	logger   log.ErrorLogger
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+// NewMongoLock returns a new mongo lock
+func NewMongoLock(logger log.ErrorLogger) *MongoLock {
+	s := &MongoLock{
+		features: make([]lock.Feature, 0),
+		logger:   logger,
+	}
+	return s
+}
+
+func (e *MongoLock) Init(metadata lock.Metadata) error {
+	// 1.parse config
+	m, err := utils.ParseMongoMetadata(metadata.Properties)
+	if err != nil {
+		return err
+	}
+	e.metadata = m
+
+	// 2. construct client
+	if e.client, err = utils.NewMongoClient(m); err != nil {
+		return err
+	}
+
+	e.ctx, e.cancel = context.WithCancel(context.Background())
+
+	if err := e.client.Ping(e.ctx, nil); err != nil {
+		return err
+	}
+
+	wc, err := utils.GetWriteConcernObject(e.metadata.WriteConcern)
+	if err != nil {
+		return err
+	}
+
+	rc, err := utils.GetReadConcrenObject(e.metadata.ReadConcern)
+	if err != nil {
+		return err
+	}
+
+	opts := options.Collection().SetWriteConcern(wc).SetReadConcern(rc)
+	database := e.client.Database(e.metadata.DatabaseName)
+	collection := database.Collection(e.metadata.CollectionName, opts)
+	e.database = database
+	e.collection = collection
+
+	// create exprie time index
+	indexModel := mongo.IndexModel{
+		Keys:    bsonx.Doc{{"Expire", bsonx.Int64(1)}},
+		Options: options.Index().SetExpireAfterSeconds(0),
+	}
+	e.collection.Indexes().CreateOne(e.ctx, indexModel)
+
+	return err
+}
+
+// Features is to get MongoLock's features
+func (e *MongoLock) Features() []lock.Feature {
+	return e.features
+}
+
+func (e *MongoLock) TryLock(req *lock.TryLockRequest) (*lock.TryLockResponse, error) {
+	session, err := e.client.StartSession()
+	txnOpts := options.Transaction().SetReadConcern(readconcern.Snapshot()).
+		SetWriteConcern(writeconcern.New(writeconcern.WMajority()))
+
+	defer session.EndSession(e.ctx)
+
+	// check session
+	if err != nil {
+		return &lock.TryLockResponse{}, fmt.Errorf("[mongoLock]: Create session return error: %s ResourceId: %s", err, req.ResourceId)
+	}
+
+	status, err := session.WithTransaction(e.ctx, func(sessionContext mongo.SessionContext) (interface{}, error) {
+		var status int
+		var err error
+		var singleResult *mongo.SingleResult
+		var insertOneResult *mongo.InsertOneResult
+
+		// set exprie date
+		expireTime := time.Now().Add(time.Duration(req.Expire) * time.Second)
+
+		// find mongo lock
+		singleResult = e.collection.FindOne(e.ctx, bson.M{"_id": req.ResourceId})
+
+		// insert mongo lock
+		if singleResult.Err() == mongo.ErrNoDocuments {
+			insertOneResult, err = e.collection.InsertOne(e.ctx, bson.M{"_id": req.ResourceId, "LockOwner": req.LockOwner, "Expire": expireTime})
+		}
+
+		if err != nil {
+			sessionContext.AbortTransaction(sessionContext)
+			return status, err
+		}
+
+		// commit and set status
+		if insertOneResult != nil && insertOneResult.InsertedID == req.ResourceId {
+			if err = sessionContext.CommitTransaction(sessionContext); err == nil {
+				status = 1
+			}
+		}
+
+		return status, err
+	}, txnOpts)
+
+	// check lock
+	if err != nil {
+		return &lock.TryLockResponse{}, fmt.Errorf("[mongoLock]: Create new lock return error: %s ResourceId: %s", err, req.ResourceId)
+	}
+
+	if status == 1 {
+		return &lock.TryLockResponse{
+			Success: true,
+		}, nil
+	} else {
+		return &lock.TryLockResponse{
+			Success: false,
+		}, nil
+	}
+}
+
+func (e *MongoLock) Unlock(req *lock.UnlockRequest) (*lock.UnlockResponse, error) {
+	session, err := e.client.StartSession()
+	txnOpts := options.Transaction().SetReadConcern(readconcern.Snapshot()).
+		SetWriteConcern(writeconcern.New(writeconcern.WMajority()))
+
+	defer session.EndSession(e.ctx)
+
+	if err != nil {
+		return newInternalErrorUnlockResponse(), fmt.Errorf("[mongoLock]: Create Session return error: %s ResourceId: %s", err, req.ResourceId)
+	}
+
+	status, err := session.WithTransaction(e.ctx, func(sessionContext mongo.SessionContext) (interface{}, error) {
+		var status int
+
+		result, err := e.collection.DeleteOne(e.ctx, bson.M{"_id": req.ResourceId, "LockOwner": req.LockOwner})
+
+		if result.DeletedCount == 1 && err == nil {
+			status = 0
+		} else if result.DeletedCount == 0 && err == nil {
+			if cursor, err := e.collection.Find(e.ctx, bson.M{"_id": req.ResourceId}); cursor != nil && cursor.RemainingBatchLength() != 0 && err == nil {
+				status = 2
+			} else if cursor != nil && cursor.RemainingBatchLength() == 0 && err == nil {
+				status = 1
+			}
+		}
+
+		if err != nil {
+			sessionContext.AbortTransaction(sessionContext)
+			return nil, err
+		}
+		err = sessionContext.CommitTransaction(sessionContext)
+
+		return status, err
+	}, txnOpts)
+
+	if err != nil {
+		return newInternalErrorUnlockResponse(), fmt.Errorf("[mongoLock]: Unlock returned error: %s ResourceId: %s", err, req.ResourceId)
+	}
+
+	resp := lock.INTERNAL_ERROR
+
+	if status == 0 {
+		resp = lock.SUCCESS
+	} else if status == 1 {
+		resp = lock.LOCK_UNEXIST
+	} else if status == 2 {
+		resp = lock.LOCK_BELONG_TO_OTHERS
+	}
+	return &lock.UnlockResponse{
+		Status: resp,
+	}, nil
+}
+
+func newInternalErrorUnlockResponse() *lock.UnlockResponse {
+	return &lock.UnlockResponse{
+		Status: lock.INTERNAL_ERROR,
+	}
+}
+
+func (e *MongoLock) Close() error {
+	e.cancel()
+
+	return e.client.Disconnect(e.ctx)
+}

components/lock/mongo/mongo_lock_test.go

@@ -0,0 +1,151 @@
+package mongo
+
+import (
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"mosn.io/layotto/components/lock"
+	"mosn.io/pkg/log"
+	"sync"
+	"testing"
+)
+
+const (
+	resourceId  = "resource_xxx"
+	resourceId2 = "resource_xxx2"
+	resourceId3 = "resource_xxx3"
+	resourceId4 = "resource_xxx4"
+	lockOwner   = "lockOwner1"
+	lockOwner2  = "lockOwner2"
+	lockOwner3  = "lockOwner3"
+)
+
+func TestMongoLock_Init(t *testing.T) {
+	var err error
+	var mongoUrl = "localhost:27017"
+	comp := NewMongoLock(log.DefaultLogger)
+
+	cfg := lock.Metadata{
+		Properties: make(map[string]string),
+	}
+
+	err = comp.Init(cfg)
+	assert.Error(t, err)
+
+	cfg.Properties["mongoHost"] = mongoUrl
+	cfg.Properties["operationTimeout"] = "a"
+	err = comp.Init(cfg)
+	assert.Error(t, err)
+
+	cfg.Properties["operationTimeout"] = "2"
+	err = comp.Init(cfg)
+	assert.Error(t, err)
+}
+
+func TestMongoLock_TryLock(t *testing.T) {
+	var err error
+	var resp *lock.TryLockResponse
+	var mongoUrl = "localhost:27017"
+
+	comp := NewMongoLock(log.DefaultLogger)
+
+	cfg := lock.Metadata{
+		Properties: make(map[string]string),
+	}
+
+	cfg.Properties["mongoHost"] = mongoUrl
+	err = comp.Init(cfg)
+	assert.NoError(t, err)
+
+	ownerId1 := uuid.New().String()
+	resp, err = comp.TryLock(&lock.TryLockRequest{
+		ResourceId: resourceId,
+		LockOwner:  ownerId1,
+		Expire:     10,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, true, resp.Success)
+
+	resp, err = comp.TryLock(&lock.TryLockRequest{
+		ResourceId: resourceId,
+		LockOwner:  ownerId1,
+		Expire:     10,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, false, resp.Success)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	go func() {
+		ownerId2 := uuid.New().String()
+		resp, err = comp.TryLock(&lock.TryLockRequest{
+			ResourceId: resourceId,
+			LockOwner:  ownerId2,
+			Expire:     10,
+		})
+		assert.NoError(t, err)
+		assert.Equal(t, false, resp.Success)
+		wg.Done()
+	}()
+
+	wg.Wait()
+
+	//another resource
+	resp, err = comp.TryLock(&lock.TryLockRequest{
+		ResourceId: resourceId2,
+		LockOwner:  ownerId1,
+		Expire:     10,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, true, resp.Success)
+}
+
+func TestMongoLock_Unlock(t *testing.T) {
+	var err error
+	var resp *lock.UnlockResponse
+	var lockresp *lock.TryLockResponse
+	var mongoUrl = "localhost:27017"
+
+	comp := NewMongoLock(log.DefaultLogger)
+
+	cfg := lock.Metadata{
+		Properties: make(map[string]string),
+	}
+
+	cfg.Properties["mongoHost"] = mongoUrl
+	err = comp.Init(cfg)
+	assert.NoError(t, err)
+
+	ownerId1 := uuid.New().String()
+	lockresp, err = comp.TryLock(&lock.TryLockRequest{
+		ResourceId: resourceId3,
+		LockOwner:  ownerId1,
+		Expire:     10,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, true, lockresp.Success)
+
+	//error ownerid
+	resp, err = comp.Unlock(&lock.UnlockRequest{
+		ResourceId: resourceId3,
+		LockOwner:  uuid.New().String(),
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, lock.LOCK_BELONG_TO_OTHERS, resp.Status)
+
+	//error resourceid
+	resp, err = comp.Unlock(&lock.UnlockRequest{
+		ResourceId: resourceId4,
+		LockOwner:  ownerId1,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, lock.LOCK_UNEXIST, resp.Status)
+
+	//success
+	resp, err = comp.Unlock(&lock.UnlockRequest{
+		ResourceId: resourceId3,
+		LockOwner:  ownerId1,
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, lock.SUCCESS, resp.Status)
+}