transaction: Resilient Txn Engine

glusterd2/commands/peers/peer-rpc-svc.go

@@ -8,10 +8,12 @@ import (
 	"github.com/gluster/glusterd2/glusterd2/peer"
 	"github.com/gluster/glusterd2/glusterd2/servers/peerrpc"
 	"github.com/gluster/glusterd2/glusterd2/store"
+	"github.com/gluster/glusterd2/glusterd2/transactionv2"
+	"github.com/gluster/glusterd2/glusterd2/transactionv2/cleanuphandler"
 	"github.com/gluster/glusterd2/glusterd2/volume"
 	"github.com/gluster/glusterd2/pkg/utils"
-	"github.com/pborman/uuid"
 
+	"github.com/pborman/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 )
@@ -169,6 +171,8 @@ func ReconfigureStore(c *StoreConfig) error {
 
 	// Stop events framework
 	events.Stop()
+	transaction.StopTxnEngine()
+	cleanuphandler.StopCleanupLeader()
 
 	// do not delete cluster namespace if this is not a loner node
 	var deleteNamespace bool
@@ -217,7 +221,8 @@ func ReconfigureStore(c *StoreConfig) error {
 
 	// Now that new store is up, start events framework
 	events.Start()
-
+	transaction.StartTxnEngine()
+	cleanuphandler.StartCleanupLeader()
 	return nil
 }
 

glusterd2/commands/snapshot/snapshot-activate.go

@@ -152,6 +152,7 @@ func snapshotActivateHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-activate.StoreSnapshot",
 			UndoFunc: "snap-activate.UndoStoreSnapshot",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 	}
 	if err = txn.Do(); err != nil {

glusterd2/commands/snapshot/snapshot-clone.go

@@ -322,6 +322,7 @@ func snapshotCloneHandler(w http.ResponseWriter, r *http.Request) {
 		{
 			DoFunc: "snap-clone.CreateCloneVolinfo",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
+			Sync:   true,
 		},
 		{
 			DoFunc:   "snap-clone.TakeBrickSnapshots",
@@ -332,6 +333,7 @@ func snapshotCloneHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-clone.StoreSnapshot",
 			UndoFunc: "snap-clone.UndoStoreSnapshotOnClone",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 	}
 	if err = txn.Ctx.Set("snapname", &snapname); err != nil {

glusterd2/commands/snapshot/snapshot-create.go

@@ -763,6 +763,7 @@ func snapshotCreateHandler(w http.ResponseWriter, r *http.Request) {
 		{
 			DoFunc: "snap-create.CreateSnapinfo",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
+			Sync:   true,
 		},
 		{
 			DoFunc:   "snap-create.ActivateBarrier",
@@ -774,6 +775,8 @@ func snapshotCreateHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-create.TakeBrickSnapshots",
 			UndoFunc: "snap-create.UndoBrickSnapshots",
 			Nodes:    txn.Nodes,
+			// All bricks need to be barriered before taking a snapshot
+			Sync: true,
 		},
 		{
 			DoFunc: "snap-create.DeactivateBarrier",
@@ -784,6 +787,7 @@ func snapshotCreateHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-create.StoreSnapshot",
 			UndoFunc: "snap-create.UndoStoreSnapshotOnCreate",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 	}
 

glusterd2/commands/snapshot/snapshot-deactivate.go

@@ -148,6 +148,7 @@ func snapshotDeactivateHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-deactivate.StoreSnapshot",
 			UndoFunc: "snap-deactivate.UndoStoreSnapshot",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 	}
 	if err = txn.Ctx.Set("oldsnapinfo", &snapinfo); err != nil {

glusterd2/commands/snapshot/snapshot-delete.go

@@ -161,6 +161,7 @@ func snapshotDeleteHandler(w http.ResponseWriter, r *http.Request) {
 		{
 			DoFunc: "snap-delete.Store",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
+			Sync:   true,
 		},
 	}
 

glusterd2/commands/snapshot/snapshot-restore.go

@@ -364,6 +364,7 @@ func snapshotRestoreHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "snap-restore.Store",
 			UndoFunc: "snap-restore.UndoStore",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc: "snap-restore.CleanBricks",

glusterd2/commands/volumes/brick-replace.go

@@ -161,6 +161,7 @@ LOOP:
 		{
 			DoFunc: "brick-replace.ReplaceVolinfo",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
+			Sync:   true,
 		},
 		{
 			DoFunc:   "vol-create.InitBricks",
@@ -175,6 +176,7 @@ LOOP:
 			DoFunc:   "vol-create.StoreVolume",
 			UndoFunc: "vol-create.UndoStoreVolume",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc: "vol-expand.NotifyClients",

glusterd2/commands/volumes/volume-create.go

@@ -11,6 +11,7 @@ import (
 	"github.com/gluster/glusterd2/glusterd2/gdctx"
 	restutils "github.com/gluster/glusterd2/glusterd2/servers/rest/utils"
 	"github.com/gluster/glusterd2/glusterd2/transaction"
+	transactionv2 "github.com/gluster/glusterd2/glusterd2/transactionv2"
 	"github.com/gluster/glusterd2/glusterd2/volume"
 	"github.com/gluster/glusterd2/pkg/api"
 	gderrors "github.com/gluster/glusterd2/pkg/errors"
@@ -174,7 +175,7 @@ func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	txn, err := transaction.NewTxnWithLocks(ctx, req.Name)
+	txn, err := transactionv2.NewTxnWithLocks(ctx, req.Name)
 	if err != nil {
 		status, err := restutils.ErrToStatusCode(err)
 		restutils.SendHTTPError(ctx, w, status, err)
@@ -196,6 +197,8 @@ func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
 		{
 			DoFunc: "vol-create.ValidateBricks",
 			Nodes:  nodes,
+			// Need to wait for volinfo to be created first
+			Sync: true,
 		},
 		{
 			DoFunc:   "vol-create.InitBricks",
@@ -206,6 +209,7 @@ func volumeCreateHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "vol-create.StoreVolume",
 			UndoFunc: "vol-create.UndoStoreVolume",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 	}
 

glusterd2/commands/volumes/volume-delete.go

@@ -81,6 +81,7 @@ func volumeDeleteHandler(w http.ResponseWriter, r *http.Request) {
 		{
 			DoFunc: "vol-delete.Store",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
+			Sync:   true,
 		},
 	}
 

glusterd2/commands/volumes/volume-expand.go

@@ -176,6 +176,8 @@ func volumeExpandHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc: "vol-expand.ValidateBricks",
 			Nodes:  nodes,
 			Skip:   lvmResizeOp,
+			// Need to wait for newly selected bricks to be set by the previous step
+			Sync: true,
 		},
 		{
 			DoFunc:   "vol-expand.InitBricks",
@@ -193,11 +195,13 @@ func volumeExpandHandler(w http.ResponseWriter, r *http.Request) {
 			UndoFunc: "vol-create.UndoStoreVolume",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
 			Skip:     !lvmResizeOp,
+			Sync:     true,
 		},
 		{
 			DoFunc: "vol-expand.UpdateVolinfo",
 			Nodes:  []uuid.UUID{gdctx.MyUUID},
 			Skip:   lvmResizeOp,
+			Sync:   true,
 		},
 		{
 			DoFunc:   "vol-expand.GenerateBrickVolfiles",

glusterd2/commands/volumes/volume-option.go

@@ -246,6 +246,7 @@ func volumeOptionsHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "vol-option.UpdateVolinfo",
 			UndoFunc: "vol-option.UpdateVolinfo.Undo",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc:   "vol-option.XlatorActionDoSet",

glusterd2/commands/volumes/volume-reset.go

@@ -167,11 +167,13 @@ func volumeResetHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "vol-option.UpdateVolinfo",
 			UndoFunc: "vol-option.UpdateVolinfo.Undo",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc:   "vol-option.GenerateBrickVolfiles",
 			UndoFunc: "vol-option.GenerateBrickvolfiles.Undo",
 			Nodes:    volinfo.Nodes(),
+			Sync:     true,
 		},
 		{
 			DoFunc: "vol-option.NotifyVolfileChange",

glusterd2/commands/volumes/volume-start.go

@@ -171,6 +171,7 @@ func volumeStartHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "vol-start.UpdateVolinfo",
 			UndoFunc: "vol-start.UpdateVolinfo.Undo",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc:   "vol-start.XlatorActionDoVolumeStart",

glusterd2/commands/volumes/volume-stop.go

@@ -152,6 +152,7 @@ func volumeStopHandler(w http.ResponseWriter, r *http.Request) {
 			DoFunc:   "vol-stop.UpdateVolinfo",
 			UndoFunc: "vol-stop.UpdateVolinfo.Undo",
 			Nodes:    []uuid.UUID{gdctx.MyUUID},
+			Sync:     true,
 		},
 		{
 			DoFunc:   "vol-stop.XlatorActionDoVolumeStop",

glusterd2/main.go

@@ -16,6 +16,8 @@ import (
 	"github.com/gluster/glusterd2/glusterd2/pmap"
 	"github.com/gluster/glusterd2/glusterd2/servers"
 	"github.com/gluster/glusterd2/glusterd2/store"
+	"github.com/gluster/glusterd2/glusterd2/transactionv2"
+	"github.com/gluster/glusterd2/glusterd2/transactionv2/cleanuphandler"
 	gdutils "github.com/gluster/glusterd2/glusterd2/utils"
 	"github.com/gluster/glusterd2/glusterd2/volgen"
 	"github.com/gluster/glusterd2/glusterd2/xlator"
@@ -105,6 +107,8 @@ func main() {
 		log.WithError(err).Fatal("Failed to initialize store (etcd client)")
 	}
 
+	transaction.StartTxnEngine()
+	cleanuphandler.StartCleanupLeader()
 	// Start the events framework after store is up
 	if err := events.Start(); err != nil {
 		log.WithError(err).Fatal("Failed to start internal events framework")
@@ -168,6 +172,8 @@ func main() {
 		case unix.SIGINT:
 			log.Info("Received SIGTERM. Stopping GlusterD")
 			gdctx.IsTerminating = true
+			transaction.StopTxnEngine()
+			cleanuphandler.StopCleanupLeader()
 			super.Stop()
 			events.Stop()
 			store.Close()

glusterd2/transaction/context.go

@@ -33,29 +33,32 @@ type TxnCtx interface {
 	// Logger returns the Logrus logger associated with the context
 	Logger() log.FieldLogger
 
-	// commit writes all locally cached keys and values into the store using
+	// Commit writes all locally cached keys and values into the store using
 	// a single etcd transaction. This is for internal use by the txn framework
 	// and hence isn't exported.
-	commit() error
+	Commit() error
+
+	// SyncCache synchronizes the locally cached keys and values from the store
+	SyncCache() error
 }
 
 // Tctx represents structure for transaction context
 type Tctx struct {
-	config         *txnCtxConfig // this will be marshalled and sent on wire
+	config         *TxnCtxConfig // this will be marshalled and sent on wire
 	logger         log.FieldLogger
 	readSet        map[string][]byte // cached responses from store
 	readCacheDirty bool
 	writeSet       map[string]string // to be written to store
 }
 
-// txnCtxConfig is marshalled and sent on wire and is used to reconstruct Tctx
+// TxnCtxConfig is marshalled and sent on wire and is used to reconstruct Tctx
 // on receiver's end.
-type txnCtxConfig struct {
+type TxnCtxConfig struct {
 	LogFields   log.Fields
 	StorePrefix string
 }
 
-func newCtx(config *txnCtxConfig) *Tctx {
+func newCtx(config *TxnCtxConfig) *Tctx {
 	return &Tctx{
 		config:         config,
 		logger:         log.StandardLogger().WithFields(config.LogFields),
@@ -65,6 +68,11 @@ func newCtx(config *txnCtxConfig) *Tctx {
 	}
 }
 
+// NewCtx returns a transaction context from given config
+func NewCtx(config *TxnCtxConfig) *Tctx {
+	return newCtx(config)
+}
+
 // Set attaches the given key-value pair to the context.
 // If the key exists, the value will be updated.
 func (c *Tctx) Set(key string, value interface{}) error {
@@ -86,9 +94,22 @@ func (c *Tctx) Set(key string, value interface{}) error {
 	return nil
 }
 
-// commit writes all locally cached keys and values into the store using
+// SyncCache synchronizes the locally cached keys and values from the store
+func (c *Tctx) SyncCache() error {
+	resp, err := store.Get(context.TODO(), c.config.StorePrefix, clientv3.WithPrefix())
+	if err != nil {
+		return err
+	}
+
+	for _, kv := range resp.Kvs {
+		c.readSet[string(kv.Key)] = kv.Value
+	}
+	return nil
+}
+
+// Commit writes all locally cached keys and values into the store using
 // a single etcd transaction.
-func (c *Tctx) commit() error {
+func (c *Tctx) Commit() error {
 
 	if len(c.writeSet) == 0 {
 		return nil
@@ -120,6 +141,7 @@ func (c *Tctx) commit() error {
 
 	expTxn.Add("txn_ctx_store_commit", 1)
 
+	c.writeSet = make(map[string]string)
 	c.readCacheDirty = true
 
 	return nil
@@ -139,15 +161,10 @@ func (c *Tctx) Get(key string, value interface{}) error {
 
 	// cache all keys and values from the store on the first call to Get
 	if c.readCacheDirty {
-		resp, err := store.Get(context.TODO(), c.config.StorePrefix, clientv3.WithPrefix())
-		if err != nil {
+		if err := c.SyncCache(); err != nil {
 			c.logger.WithError(err).WithField("key", key).Error("failed to get key from transaction context")
 			return err
 		}
-		expTxn.Add("txn_ctx_store_get", 1)
-		for _, kv := range resp.Kvs {
-			c.readSet[string(kv.Key)] = kv.Value
-		}
 		c.readCacheDirty = false
 	}