feature: add etcd distributed lock

conf/gittar/gittar.yaml

@@ -1 +1,9 @@
 gittar:
+
+etcd:
+  endpoints: "${ETCD_ENDPOINTS:https://localhost:2379}"
+  tls:
+    cert_file: "${ETCD_CERT_FILE:/certs/etcd-client.pem}"
+    cert_key_file: "${ETCD_CERT_KEY_FILE:/certs/etcd-client-key.pem}"
+    ca_file: "${ETCD_CA_FILE:/certs/etcd-ca.pem}"
+

modules/gittar/api/repo.go

@@ -21,7 +21,6 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
-	"path"
 	"regexp"
 	"sort"
 	"strconv"
@@ -675,22 +674,12 @@ func CreateCommit(context *webcontext.Context) {
 	}
 	createCommitRequest.Signature = context.User.ToGitSignature()
 
-	commit, err := repository.CreateCommit(&createCommitRequest)
+	commit, err := repository.CreateCommit(&createCommitRequest, context.EtcdClient)
 	if err != nil {
 		context.Abort(err)
 		return
 	}
 
-	// 外置仓库推送代码过去
-	if repository.IsExternal {
-		repoPath := path.Join(conf.RepoRoot(), repository.Path)
-		err = gitmodule.PushExternalRepository(repoPath)
-		if err != nil {
-			context.Abort(err)
-			return
-		}
-	}
-
 	pushEvent := &models.PayloadPushEvent{
 		Before: beforeCommitID,
 		After:  commit.ID,

modules/gittar/auth/auth.go

@@ -15,16 +15,19 @@
 package auth
 
 import (
+	"context"
 	"errors"
+	"fmt"
+
 	"net/http"
 	"path"
 	"path/filepath"
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
+	"github.com/coreos/etcd/clientv3"
 	"github.com/jinzhu/gorm"
 	"github.com/patrickmn/go-cache"
 	"github.com/sirupsen/logrus"
@@ -198,7 +201,7 @@ func doAuth(c *webcontext.Context, repo *models.Repo, repoName string) {
 	for _, skipUrl := range conf.SkipAuthUrls() {
 		if skipUrl != "" && strings.HasSuffix(host, skipUrl) {
 			logrus.Debugf("skip authenticate host: %s", host)
-			gitRepository, err = openRepository(repo)
+			gitRepository, err = openRepository(c, repo)
 			if err != nil {
 				c.AbortWithStatus(500, err)
 				return
@@ -246,7 +249,7 @@ func doAuth(c *webcontext.Context, repo *models.Repo, repoName string) {
 		if validateError != nil {
 			c.AbortWithString(403, validateError.Error()+" 403")
 		} else {
-			gitRepository, err = openRepository(repo)
+			gitRepository, err = openRepository(c, repo)
 			if err != nil {
 				c.AbortWithStatus(500, err)
 				return
@@ -386,11 +389,7 @@ func ValidaUserRepo(c *webcontext.Context, userId string, repo *models.Repo) (*A
 	}, nil
 }
 
-// 用于外置仓库的锁定，同一时间只有一个线程在同步外置仓库
-var lockMap = sync.Map{}
-
-func openRepository(repo *models.Repo) (*gitmodule.Repository, error) {
-	repo.RwMutex = &sync.RWMutex{}
+func openRepository(ctx *webcontext.Context, repo *models.Repo) (*gitmodule.Repository, error) {
 	gitRepository, err := gitmodule.OpenRepositoryWithInit(conf.RepoRoot(), repo.Path)
 	if err != nil {
 		return nil, err
@@ -401,46 +400,53 @@ func openRepository(repo *models.Repo) (*gitmodule.Repository, error) {
 	gitRepository.OrgId = repo.OrgID
 	gitRepository.Size = repo.Size
 	gitRepository.Url = conf.GittarUrl() + "/" + repo.Path
+	gitRepository.IsExternal = repo.IsExternal
 	if repo.IsExternal {
-		repoPath := path.Join(conf.RepoRoot(), repo.Path)
-
-		// 判定是否锁定
-		lock, ok := lockMap.Load(repoPath)
-		if ok {
-			if lock != nil && !lock.(bool) {
-				// 假如没有锁定，就开始锁定
-				lockMap.Store(repoPath, true)
-				// 结束后取消锁定
-				repo.RwMutex.Lock()
-				go func() {
-					defer func() {
-						lockMap.Store(repoPath, false)
-						repo.RwMutex.Unlock()
-					}()
-					err = gitmodule.SyncExternalRepository(repoPath)
-					if err != nil {
-						logrus.Errorf(" SyncExternalRepository error: %v ", err)
-					}
-				}()
+		// check the key is exist or not
+		key := fmt.Sprintf("/gittar/repo/%d", repo.ID)
+		resp, err := ctx.EtcdClient.Get(context.Background(), key)
+		if err != nil {
+			return nil, err
+		}
+		// if key exist,and the request url's suffix is "/commits" will return err
+		// else return without SyncExternalRepository
+		if len(resp.Kvs) > 0 {
+			if strings.HasSuffix(ctx.EchoContext.Request().URL.String(), "/commits") {
+				return nil, errors.New("the repo is locked, please wait for a moment")
 			}
-		} else {
-			lockMap.Store(repoPath, true)
-			repo.RwMutex.Lock()
-			go func() {
-				defer func() {
-					lockMap.Store(repoPath, false)
-					repo.RwMutex.Unlock()
-				}()
-				err = gitmodule.SyncExternalRepository(repoPath)
-				if err != nil {
-					logrus.Errorf(" SyncExternalRepository error: %v ", err)
-				}
-			}()
+			return gitRepository, nil
+		}
+
+		// minimum lease TTL is 5-second
+		grantResp, err := ctx.EtcdClient.Grant(context.Background(), 5)
+		if err != nil {
+			return nil, err
 		}
 
+		// put key with lease
+		_, err = ctx.EtcdClient.Put(context.Background(), key, "lock", clientv3.WithLease(grantResp.ID))
+		if err != nil {
+			return nil, err
+		}
+
+		// keep alive
+		_, err = ctx.EtcdClient.KeepAlive(context.Background(), grantResp.ID)
+		if err != nil {
+			return nil, err
+		}
+
+		defer func() {
+			_, err = ctx.EtcdClient.Revoke(context.Background(), grantResp.ID)
+			if err != nil {
+				logrus.Errorf("failed to revoke etcd, err: %v ", err)
+			}
+		}()
+
+		err = gitmodule.SyncExternalRepository(path.Join(conf.RepoRoot(), repo.Path))
+		if err != nil {
+			return nil, err
+		}
 	}
-	gitRepository.IsExternal = repo.IsExternal
-	gitRepository.RwLock = repo.RwMutex
 
 	return gitRepository, nil
 }

modules/gittar/initialize.go

@@ -37,7 +37,7 @@ import (
 )
 
 // Initialize 初始化应用启动服务.
-func Initialize() error {
+func (p *provider) Initialize() error {
 	conf.Load()
 	if conf.Debug() {
 		logrus.SetLevel(logrus.DebugLevel)
@@ -76,6 +76,7 @@ func Initialize() error {
 	webcontext.WithDB(dbClient)
 	webcontext.WithBundle(diceBundle)
 	webcontext.WithUCAuth(ucUserAuth)
+	webcontext.WithEtcdClient(p.EtcdClient)
 
 	e := echo.New()
 	systemGroup := e.Group("/_system")

modules/gittar/models/repo.go

@@ -20,7 +20,6 @@ import (
 	"os"
 	"path"
 	"strconv"
-	"sync"
 	"time"
 
 	"github.com/patrickmn/go-cache"
@@ -46,9 +45,6 @@ type Repo struct {
 	Size        int64
 	IsExternal  bool
 	Config      string
-
-	// to ensure sync operation precedes commit
-	RwMutex *sync.RWMutex
 }
 
 func (Repo) TableName() string {

modules/gittar/pkg/gitmodule/repo.go

@@ -25,7 +25,6 @@ import (
 	"path"
 	"path/filepath"
 	"strings"
-	"sync"
 	"time"
 
 	git "github.com/libgit2/git2go/v30"
@@ -56,8 +55,6 @@ type Repository struct {
 	tree       *Tree   `json:"-"` //只有RefType是tree才有值
 	IsLocked   bool    `json:"-"`
 	IsExternal bool    // 是否是外置仓库
-	// to ensure sync operation precedes commit
-	RwLock *sync.RWMutex `json:"-"`
 }
 
 const (

modules/gittar/pkg/gitmodule/repo_edit.go

@@ -17,10 +17,18 @@
 package gitmodule
 
 import (
+	"context"
 	"errors"
+	"fmt"
+	"path"
 	"strings"
 
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/mvcc/mvccpb"
 	git "github.com/libgit2/git2go/v30"
+	"github.com/sirupsen/logrus"
+
+	"github.com/erda-project/erda/modules/gittar/conf"
 )
 
 type EditAction string
@@ -72,9 +80,56 @@ func (req *CreateCommit) Validate() error {
 	return nil
 }
 
-func (repo *Repository) CreateCommit(request *CreateCommit) (*Commit, error) {
-	repo.RwLock.RLock()
-	defer repo.RwLock.RUnlock()
+// waitDelete wait for the etcd key delete
+func waitDelete(key string, cli *clientv3.Client) {
+	wch := cli.Watch(context.Background(), key)
+	for wr := range wch {
+		for _, ev := range wr.Events {
+			switch ev.Type {
+			case mvccpb.DELETE:
+				return
+			}
+		}
+	}
+}
+
+func (repo *Repository) CreateCommit(request *CreateCommit, etcdClient *clientv3.Client) (*Commit, error) {
+	// check the key is exist or not
+	key := fmt.Sprintf("/gittar/repo/%d", repo.ID)
+	resp, err := etcdClient.Get(context.Background(), key)
+	if err != nil {
+		return nil, err
+	}
+	// if exist should wait
+	if len(resp.Kvs) > 0 {
+		waitDelete(key, etcdClient)
+	}
+
+	// minimum lease TTL is 5-second
+	grantResp, err := etcdClient.Grant(context.Background(), 5)
+	if err != nil {
+		return nil, err
+	}
+
+	// put key with lease
+	_, err = etcdClient.Put(context.Background(), key, "lock", clientv3.WithLease(grantResp.ID))
+	if err != nil {
+		return nil, err
+	}
+
+	// keep alive
+	_, err = etcdClient.KeepAlive(context.Background(), grantResp.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	defer func() {
+		_, err = etcdClient.Revoke(context.Background(), grantResp.ID)
+		if err != nil {
+			logrus.Errorf("failed to revoke etcd, err: %v ", err)
+		}
+	}()
+
 	branch := request.Branch
 	message := request.Message
 	isInitCommit := false
@@ -225,6 +280,16 @@ func (repo *Repository) CreateCommit(request *CreateCommit) (*Commit, error) {
 		return nil, err
 	}
 	commit.ParentDirPath = parentDirPath
+
+	// 外置仓库推送代码过去
+	if repo.IsExternal {
+		repoPath := path.Join(conf.RepoRoot(), repo.Path)
+		err = PushExternalRepository(repoPath)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return commit, nil
 }
 

modules/gittar/provider.go

@@ -17,16 +17,23 @@ package gittar
 import (
 	"context"
 
+	"github.com/coreos/etcd/clientv3"
+
 	"github.com/erda-project/erda-infra/base/servicehub"
+	"github.com/erda-project/erda-infra/providers/etcd"
 )
 
-type provider struct{}
+type provider struct {
+	ETCD       etcd.Interface   // autowired
+	EtcdClient *clientv3.Client // autowired
+}
 
-func (p *provider) Run(ctx context.Context) error { return Initialize() }
+func (p *provider) Run(ctx context.Context) error { return p.Initialize() }
 
 func init() {
 	servicehub.Register("gittar", &servicehub.Spec{
-		Services: []string{"gittar"},
-		Creator:  func() servicehub.Provider { return &provider{} },
+		Services:     []string{"gittar"},
+		Dependencies: []string{"etcd"},
+		Creator:      func() servicehub.Provider { return &provider{} },
 	})
 }