Merge pull request #286 from Shopify/allow_super_user_on_read_only

add AllowSuperUserOnReadOnly flag

config.go

@@ -753,6 +753,17 @@ type Config struct {
 	// in the Progress. This behaviour is perfectly okay and doesn't mean there are no rows being written
 	// to the target DB.
 	EnableRowBatchSize bool
+
+	// If the target DB is set to read_only ghostferry will throw an error during the initialization step.
+	// AllowSuperUserOnReadOnly flag allows to run ghostferry even if the target DB is read_only. This is helpful in
+	// scenarios where target DB needs to be restricted from writes made by any other user then the ghostferry user.
+	//
+	// Optional: Defaults to false.
+	//
+	// NOTE:
+	// The ghostferry target user should have SUPER permissions to actually write to the target DB,
+	// if ghostferry is ran with AllowSuperUserOnReadOnly = true and the target DB is set to read_only.
+	AllowSuperUserOnReadOnly bool
 }
 
 func (c *Config) ValidateConfig() error {

ferry.go

@@ -375,13 +375,15 @@ func (f *Ferry) Initialize() (err error) {
 		return err
 	}
 
-	isReplica, err := CheckDbIsAReplica(f.TargetDB)
-	if err != nil {
-		f.logger.WithError(err).Error("cannot check if target db is writable")
-		return err
-	}
-	if isReplica {
-		return fmt.Errorf("@@read_only must be OFF on target db")
+	if !f.Config.AllowSuperUserOnReadOnly {
+		isReplica, err := CheckDbIsAReplica(f.TargetDB)
+		if err != nil {
+			f.logger.WithError(err).Error("cannot check if target db is writable")
+			return err
+		}
+		if isReplica {
+			return fmt.Errorf("@@read_only must be OFF on target db")
+		}
 	}
 
 	// Check if we're running from a replica or not and sanity check

test/go/ferry_test.go

@@ -34,6 +34,20 @@ func (t *FerryTestSuite) TestReadOnlyDatabaseFailsInitialization() {
 	t.Require().Nil(err)
 }
 
+func (t *FerryTestSuite) TestReadOnlyDatabaseDoesNotFailInitializationWithAllowSuperUserOnReadOnlyFlag() {
+	_, err := t.Ferry.TargetDB.Exec("SET GLOBAL read_only = ON")
+	t.Require().Nil(err)
+
+	ferry := testhelpers.NewTestFerry().Ferry
+	ferry.Config.AllowSuperUserOnReadOnly = true
+
+	err = ferry.Initialize()
+	t.Require().Nil(err)
+
+	_, err = t.Ferry.TargetDB.Exec("SET GLOBAL read_only = OFF")
+	t.Require().Nil(err)
+}
+
 func (t *FerryTestSuite) TestSourceDatabaseWithForeignKeyConstraintFailsInitialization() {
 	createTableWithFkConstraint := `
 		CREATE TABLE gftest.test_fk (