transaction: Replace use of CreateLockFuncs with NewTxnWithLocks

glusterd2/commands/volumes/volume-delete.go

@@ -66,18 +66,16 @@ func volumeDeleteHandler(w http.ResponseWriter, r *http.Request) {
 	logger := gdctx.GetReqLogger(ctx)
 	volname := mux.Vars(r)["volname"]
 
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	// Taking a lock outside the txn as volinfo.Nodes() must also
-	// be populated holding the lock. See issue #510
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	volinfo, err := volume.GetVolume(volname)
 	if err != nil {
@@ -95,9 +93,6 @@ func volumeDeleteHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	txn.Steps = []*transaction.Step{
 		{
 			DoFunc: "vol-delete.DeleteVolfiles",

glusterd2/commands/volumes/volume-edit.go

@@ -29,16 +29,16 @@ func volumeEditHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	//Lock on Volume Name
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	//validate volume name
 	volinfo, err := volume.GetVolume(volname)

glusterd2/commands/volumes/volume-option.go

@@ -158,18 +158,16 @@ func volumeOptionsHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	// Taking a lock outside the txn as volinfo.Nodes() must also
-	// be populated holding the lock.
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	volinfo, err := volume.GetVolume(volname)
 	if err != nil {
@@ -181,9 +179,6 @@ func volumeOptionsHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	allNodes, err := peer.GetPeerIDs()
 	if err != nil {
 		restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)

glusterd2/commands/volumes/volume-start.go

@@ -78,18 +78,16 @@ func volumeStartHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	// Taking a lock outside the txn as volinfo.Nodes() must also
-	// be populated holding the lock. See issue #510
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	volinfo, err := volume.GetVolume(volname)
 	if err != nil {
@@ -106,9 +104,6 @@ func volumeStartHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	txn.Steps = []*transaction.Step{
 		{
 			DoFunc:   "vol-start.StartBricks",

glusterd2/commands/volumes/volume-statedump.go

@@ -103,16 +103,16 @@ func volumeStatedumpHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	volinfo, err := volume.GetVolume(volname)
 	if err != nil {
@@ -129,9 +129,6 @@ func volumeStatedumpHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	txn.Steps = []*transaction.Step{
 		{
 			DoFunc: "vol-statedump.TakeStatedump",

glusterd2/commands/volumes/volume-stop.go

@@ -76,18 +76,16 @@ func volumeStopHandler(w http.ResponseWriter, r *http.Request) {
 	logger := gdctx.GetReqLogger(ctx)
 	volname := mux.Vars(r)["volname"]
 
-	lock, unlock := transaction.CreateLockFuncs(volname)
-	// Taking a lock outside the txn as volinfo.Nodes() must also
-	// be populated holding the lock. See issue #510
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, volname)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	volinfo, err := volume.GetVolume(volname)
 	if err != nil {
@@ -104,9 +102,6 @@ func volumeStopHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	txn.Steps = []*transaction.Step{
 		{
 			DoFunc: "vol-stop.StopBricks",

glusterd2/transaction/lock.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/coreos/etcd/clientv3/concurrency"
 	"github.com/pborman/uuid"
-	log "github.com/sirupsen/logrus"
 )
 
 const (
@@ -90,63 +89,6 @@ func CreateLockSteps(key string) (*Step, *Step, error) {
 	return lockStep, unlockStep, nil
 }
 
-// LockUnlockFunc is signature of functions used for distributed locking
-// and unlocking.
-type LockUnlockFunc func(ctx context.Context) error
-
-// CreateLockFuncs creates and returns functions for distributed lock and
-// unlock. This is similar to CreateLockSteps() but returns normal functions.
-func CreateLockFuncs(key string) (LockUnlockFunc, LockUnlockFunc) {
-
-	key = lockPrefix + key
-	locker := concurrency.NewMutex(store.Store.Session, key)
-
-	// TODO: There is an opportunity for refactor here to re-use code
-	// between CreateLockFunc and CreateLockSteps. This variant doesn't
-	// have registry either.
-
-	lockFunc := func(ctx context.Context) error {
-		logger := gdctx.GetReqLogger(ctx)
-		if logger == nil {
-			logger = log.StandardLogger()
-		}
-
-		ctx, cancel := context.WithTimeout(ctx, lockObtainTimeout)
-		defer cancel()
-
-		logger.WithField("key", key).Debug("attempting to lock")
-		err := locker.Lock(ctx)
-		switch err {
-		case nil:
-			logger.WithField("key", key).Debug("lock obtained")
-		case context.DeadlineExceeded:
-			// Propagate this all the way back to the client as a HTTP 409 response
-			logger.WithField("key", key).Debug("timeout: failed to obtain lock")
-			err = ErrLockTimeout
-		}
-
-		return err
-	}
-
-	unlockFunc := func(ctx context.Context) error {
-		logger := gdctx.GetReqLogger(ctx)
-		if logger == nil {
-			logger = log.StandardLogger()
-		}
-
-		logger.WithField("key", key).Debug("attempting to unlock")
-		if err := locker.Unlock(context.Background()); err != nil {
-			logger.WithField("key", key).WithError(err).Error("unlock failed")
-			return err
-		}
-
-		logger.WithField("key", key).Debug("lock unlocked")
-		return nil
-	}
-
-	return lockFunc, unlockFunc
-}
-
 func (t *Txn) lock(lockID string) error {
 	// Ensure that no prior lock exists for the given lockID in this transaction
 	if _, ok := t.locks[lockID]; ok {

glusterd2/transaction/transaction.go

@@ -59,17 +59,17 @@ func NewTxn(ctx context.Context) *Txn {
 }
 
 // NewTxnWithLocks returns an empty Txn with locks obtained on given lockIDs
-func NewTxnWithLocks(ctx context.Context, lockIDs ...string) *Txn {
+func NewTxnWithLocks(ctx context.Context, lockIDs ...string) (*Txn, error) {
 	t := NewTxn(ctx)
 
 	for _, id := range lockIDs {
 		if err := t.Lock(id); err != nil {
 			t.Done()
-			return nil
+			return nil, err
 		}
 	}
 
-	return t
+	return t, nil
 }
 
 // Done releases any obtained locks and cleans up the transaction namespace

plugins/device/rest.go

@@ -33,16 +33,16 @@ func deviceAddHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	lock, unlock := transaction.CreateLockFuncs(peerID)
-	if err := lock(ctx); err != nil {
+	txn, err := transaction.NewTxnWithLocks(ctx, peerID)
+	if err != nil {
 		if err == transaction.ErrLockTimeout {
 			restutils.SendHTTPError(ctx, w, http.StatusConflict, err)
 		} else {
 			restutils.SendHTTPError(ctx, w, http.StatusInternalServerError, err)
 		}
 		return
 	}
-	defer unlock(ctx)
+	defer txn.Done()
 
 	peerInfo, err := peer.GetPeer(peerID)
 	if err != nil {
@@ -70,9 +70,6 @@ func deviceAddHandler(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	txn := transaction.NewTxn(ctx)
-	defer txn.Done()
-
 	txn.Nodes = []uuid.UUID{peerInfo.ID}
 	txn.Steps = []*transaction.Step{
 		{