Migrations functionality

gbus/builder/builder.go

@@ -82,6 +82,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 				panic(err)
 			}
 			gb.TxProvider = mysqltx
+			mysql.EnsureSchema(mysqltx.Database, gb.SvcName)
 			//TODO move purge logic into the NewSagaStore factory method
 			sagaStore = mysql.NewSagaStore(gb.SvcName, mysqltx)
 			if builder.purgeOnStartup {

gbus/tx/mysql/migrations.go

@@ -0,0 +1,97 @@
+package mysql
+
+import (
+	"database/sql"
+
+	"fmt"
+	"github.com/lopezator/migrator"
+	"github.com/wework/grabbit/gbus/tx"
+)
+
+func SagaStoreTableMigration(svcName string) *migrator.Migration {
+	tblName := tx.GetSagatableName(svcName)
+
+	createTableQuery := `CREATE TABLE IF NOT EXISTS ` + tblName + ` (
+		rec_id INT PRIMARY KEY AUTO_INCREMENT,
+		saga_id VARCHAR(255) UNIQUE NOT NULL,
+		saga_type VARCHAR(255)  NOT NULL,
+		saga_data LONGBLOB NOT NULL,
+		version integer NOT NULL DEFAULT 0,
+		last_update timestamp  DEFAULT NOW(),
+		INDEX ` + tblName + `_sagatype_idx (saga_type))`
+
+	return &migrator.Migration{
+		Name: "create saga store table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(createTableQuery); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func OutboxMigrations(svcName string) *migrator.Migration {
+
+	query := `CREATE TABLE IF NOT EXISTS ` + getOutboxName(svcName) + ` (
+	rec_id int NOT NULL AUTO_INCREMENT,
+	message_id varchar(50) NOT NULL UNIQUE,
+	message_type varchar(50) NOT NULL,
+	exchange	varchar(50) NOT NULL,
+	routing_key	varchar(50) NOT NULL,
+	publishing	longblob NOT NULL,
+	status	int(11) NOT NULL,
+	relay_id varchar(50)  NULL,
+	delivery_tag	bigint(20) NOT NULL,
+	delivery_attempts int NOT NULL DEFAULT 0,
+	insert_date	timestamp DEFAULT CURRENT_TIMESTAMP,
+	PRIMARY KEY(rec_id),
+	INDEX status_delivery (rec_id, status, delivery_attempts))`
+
+	return &migrator.Migration{
+		Name: "create outbox table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(query); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func TimoutTableMigration(svcName string) *migrator.Migration {
+	tblName := GetTimeoutsTableName(svcName)
+
+	createTableQuery := `CREATE TABLE IF NOT EXISTS ` + tblName + ` (
+      rec_id INT PRIMARY KEY AUTO_INCREMENT,
+      saga_id VARCHAR(255) UNIQUE NOT NULL,
+	  timeout DATETIME NOT NULL,
+	  INDEX (timeout),
+	  INDEX (saga_id)
+	 )`
+
+	return &migrator.Migration{
+		Name: "create timeout table",
+		Func: func(tx *sql.Tx) error {
+			if _, err := tx.Exec(createTableQuery); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+}
+
+func EnsureSchema(db *sql.DB, svcName string) {
+
+	migrationsTable := fmt.Sprintf("grabbitMigrations_%s", svcName)
+
+	migrate := migrator.NewNamed(migrationsTable,
+		OutboxMigrations(svcName),
+		SagaStoreTableMigration(svcName),
+		TimoutTableMigration(svcName),
+	)
+	err := migrate.Migrate(db)
+	if err != nil {
+		panic(err)
+	}
+}

gbus/tx/mysql/sagastore.go

@@ -1,10 +1,6 @@
 package mysql
 
 import (
-	"database/sql"
-	"fmt"
-
-	log "github.com/sirupsen/logrus"
 	"github.com/wework/grabbit/gbus"
 	"github.com/wework/grabbit/gbus/saga"
 	"github.com/wework/grabbit/gbus/tx"
@@ -15,85 +11,6 @@ type SagaStore struct {
 	*tx.SagaStore
 }
 
-func (store *SagaStore) log() *log.Entry {
-	return log.WithField("saga_store", "mysql")
-}
-
-func (store *SagaStore) ensureSchema() {
-	store.log().Info("ensuring saga schema exists")
-	if tablesExists := store.sagaTablesExist(); !tablesExists {
-
-		store.log().Info("could not find saga schema, attempting to creat schema")
-
-		store.createSagaTables()
-	}
-}
-
-func (store *SagaStore) sagaTablesExist() bool {
-
-	tblName := store.GetSagatableName()
-	tx := store.NewTx()
-	defer func() {
-		err := tx.Commit()
-		if err != nil {
-			store.log().WithError(err).Error("could not commit sagaTablesExist")
-		}
-	}()
-
-	selectSQL := `SELECT 1 FROM ` + tblName + ` LIMIT 1;`
-
-	store.log().Info(selectSQL)
-
-	row := tx.QueryRow(selectSQL)
-	var exists int
-	err := row.Scan(&exists)
-	if err != nil && err != sql.ErrNoRows {
-		return false
-	}
-
-	return true
-}
-
-func (store *SagaStore) createSagaTables() {
-	tblName := store.GetSagatableName()
-	tx := store.NewTx()
-
-	createTable := `CREATE TABLE ` + tblName + ` (
-		rec_id INT PRIMARY KEY AUTO_INCREMENT,
-		saga_id VARCHAR(255) UNIQUE NOT NULL,
-		saga_type VARCHAR(255)  NOT NULL,
-		saga_data LONGBLOB NOT NULL,
-		version integer NOT NULL DEFAULT 0,
-		last_update timestamp  DEFAULT NOW()
-		)`
-	createSagaTypeIndex := `CREATE INDEX ` + tblName + `_sagatype_idx ON ` + tblName + ` (saga_type)`
-
-	sqls := [2]string{
-		createTable,
-		createSagaTypeIndex}
-
-	for i, sql := range sqls {
-		_, error := tx.Exec(sql)
-
-		store.log().WithFields(log.Fields{"step": i, "sql": sql}).Info("creating saga tables")
-
-		if error != nil {
-			txErr := fmt.Errorf("failed to create saga tables.\n%v", error)
-			if rollbackErr := tx.Rollback(); rollbackErr != nil {
-				log.WithError(rollbackErr).Error("could not rollback transaction")
-			}
-			panic(txErr)
-		}
-	}
-
-	error := tx.Commit()
-	if error != nil {
-		txErr := fmt.Errorf("failed to create saga tables.\n%v", error)
-		panic(txErr)
-	}
-
-}
-
 //NewSagaStore creates a bew SagaStore
 func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store {
 
@@ -103,7 +20,6 @@ func NewSagaStore(svcName string, txProvider gbus.TxProvider) saga.Store {
 		ParamsMarkers: getParamsMarker()}
 	store := &SagaStore{
 		base}
-	store.ensureSchema()
 	return store
 }
 

gbus/tx/mysql/timeout.go

@@ -24,31 +24,6 @@ type TimeoutManager struct {
 	exit              chan bool
 }
 
-func (tm *TimeoutManager) ensureSchema() error {
-	tblName := tm.timeoutsTableName
-	tx, e := tm.Txp.New()
-	if e != nil {
-		tm.Log().WithError(e).Error("failed to create schema for mysql timeout manager")
-		return e
-	}
-
-	createTableSQL := `CREATE TABLE IF NOT EXISTS ` + tblName + ` (
-      rec_id INT PRIMARY KEY AUTO_INCREMENT,
-      saga_id VARCHAR(255) UNIQUE NOT NULL,
-	  timeout DATETIME NOT NULL,
-	  INDEX (timeout),
-	  INDEX (saga_id)
-	 )`
-
-	if _, e := tx.Exec(createTableSQL); e != nil {
-		if rbkErr := tx.Rollback(); rbkErr != nil {
-			tm.Log().Warn("timeout manager failed to rollback transaction")
-		}
-		return e
-	}
-	return tx.Commit()
-}
-
 func (tm *TimeoutManager) purge() error {
 	purgeSQL := `DELETE FROM ` + tm.timeoutsTableName
 
@@ -197,7 +172,7 @@ func (tm *TimeoutManager) SetTimeoutFunction(timeoutFunc func(tx *sql.Tx, sagaID
 }
 
 //GetTimeoutsTableName returns the table name in which to store timeouts
-func getTimeoutsTableName(svcName string) string {
+func GetTimeoutsTableName(svcName string) string {
 
 	var re = regexp.MustCompile(`-|;|\\|`)
 	sanitized := re.ReplaceAllString(svcName, "")
@@ -208,18 +183,14 @@ func getTimeoutsTableName(svcName string) string {
 //NewTimeoutManager creates a new instance of a mysql based TimeoutManager
 func NewTimeoutManager(bus gbus.Bus, txp gbus.TxProvider, logger func() logrus.FieldLogger, svcName string, purge bool) *TimeoutManager {
 
-	timeoutsTableName := getTimeoutsTableName(svcName)
+	timeoutsTableName := GetTimeoutsTableName(svcName)
 	tm := &TimeoutManager{
 		Log:               logger,
 		Bus:               bus,
 		Txp:               txp,
 		SvcName:           svcName,
 		timeoutsTableName: timeoutsTableName,
 		exit:              make(chan bool)}
-
-	if err := tm.ensureSchema(); err != nil {
-		panic(err)
-	}
 	if purge {
 		if err := tm.purge(); err != nil {
 			panic(err)

gbus/tx/mysql/txoutbox.go

@@ -5,15 +5,14 @@ import (
 	"database/sql"
 	"encoding/gob"
 	"fmt"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
 	"github.com/rs/xid"
 	log "github.com/sirupsen/logrus"
 	"github.com/streadway/amqp"
 	"github.com/wework/grabbit/gbus"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
 )
 
 var (
@@ -55,13 +54,6 @@ func (outbox *TxOutbox) Start(amqpOut *gbus.AMQPOutbox) error {
 	if e != nil {
 		panic(fmt.Sprintf("passed in transaction provider failed with the following error\n%s", e))
 	}
-	if ensureErr := outbox.ensureSchema(tx, outbox.svcName); ensureErr != nil {
-		err := tx.Rollback()
-		if err != nil {
-			outbox.log().WithError(err).Error("could not rollback the transaction for creation of schemas")
-		}
-		return ensureErr
-	}
 	if outbox.purgeOnStartup {
 		if purgeErr := outbox.purge(tx); purgeErr != nil {
 			outbox.log().WithError(purgeErr).Error("failed to purge transactional outbox")
@@ -321,75 +313,6 @@ func (outbox *TxOutbox) sendMessages(recordSelector func(tx *sql.Tx) (*sql.Rows,
 	return nil
 }
 
-func (outbox *TxOutbox) ensureSchema(tx *sql.Tx, svcName string) error {
-
-	schemaExists := outbox.outBoxTablesExists(tx, svcName)
-
-	if schemaExists {
-		/*
-			The following  performs an alter schema to accommodate for breaking change introduced in commit 6a9f5df
-			so that earlier consumers of grabbit will not break once the upgrade to the 1.0.0 release.
-			Once a proper DB migration stratagy will be in place and implemented (post 1.0.0) the following code
-			will be deleted.
-		*/
-
-		return outbox.migrate0_9To1_0(tx, svcName)
-	}
-
-	createTablesSQL := `CREATE TABLE IF NOT EXISTS ` + getOutboxName(svcName) + ` (
-	rec_id int NOT NULL AUTO_INCREMENT,
-	message_id varchar(50) NOT NULL UNIQUE,
-	message_type varchar(50) NOT NULL,
-	exchange	varchar(50) NOT NULL,
-	routing_key	varchar(50) NOT NULL,
-	publishing	longblob NOT NULL,
-	status	int(11) NOT NULL,
-	relay_id varchar(50)  NULL,
-	delivery_tag	bigint(20) NOT NULL,
-	delivery_attempts int NOT NULL DEFAULT 0,
-	insert_date	timestamp DEFAULT CURRENT_TIMESTAMP,
-	PRIMARY KEY(rec_id),
-	INDEX status_delivery (rec_id, status, delivery_attempts))`
-
-	_, createErr := tx.Exec(createTablesSQL)
-
-	return createErr
-
-}
-
-func (outbox *TxOutbox) outBoxTablesExists(tx *sql.Tx, svcName string) bool {
-
-	tblName := getOutboxName(svcName)
-
-	selectSQL := `SELECT 1 FROM ` + tblName + ` LIMIT 1;`
-
-	outbox.log().Info(selectSQL)
-
-	row := tx.QueryRow(selectSQL)
-	var exists int
-	err := row.Scan(&exists)
-	if err != nil && err != sql.ErrNoRows {
-		return false
-	}
-
-	return true
-}
-
-func (outbox *TxOutbox) migrate0_9To1_0(tx *sql.Tx, svcName string) error {
-	tblName := getOutboxName(svcName)
-	alter := `ALTER TABLE ` + tblName + ` CHANGE COLUMN delivery_attemtps delivery_attempts int NOT NULL DEFAULT 0;`
-	_, execErr := tx.Exec(alter)
-	if execErr != nil {
-		outbox.log().WithField("sql_err", execErr).Info("migration:renaming column")
-	}
-	addIndex := `ALTER TABLE ` + tblName + ` ADD INDEX status_delivery (rec_id, status, delivery_attempts);`
-	_, indexErr := tx.Exec(addIndex)
-	if indexErr != nil {
-		outbox.log().WithField("sql_err", execErr).Info("migration:adding index column")
-	}
-	return nil
-}
-
 func getOutboxName(svcName string) string {
 
 	return strings.ToLower("grabbit_" + sanitizeTableName(svcName) + "_outbox")

gbus/tx/mysql/txprovider.go

@@ -5,12 +5,11 @@ import (
 	"time"
 
 	_ "github.com/go-sql-driver/mysql" //blank import
-	"github.com/wework/grabbit/gbus"
 	"github.com/wework/grabbit/gbus/tx"
 )
 
 //NewTxProvider returns a new PgProvider
-func NewTxProvider(connStr string) (gbus.TxProvider, error) {
+func NewTxProvider(connStr string) (*tx.Provider, error) {
 	db, err := sql.Open("mysql", connStr)
 	db.SetConnMaxLifetime(time.Second)
 	if err != nil {