WIP: ApplySchema to run online schema changes via gh-ost

config/init_db.sql

@@ -98,6 +98,13 @@ GRANT SUPER, PROCESS, REPLICATION SLAVE, RELOAD
 GRANT SELECT
   ON _vt.* TO 'orc_client_user'@'%';
 
+# User for gh-ost (https://github.com/github/gh-ost).
+CREATE USER 'gh-ost'@'127.0.0.1' IDENTIFIED BY 'gh-ost';
+GRANT SUPER, REPLICATION SLAVE
+  ON *.* TO 'gh-ost'@'127.0.0.1';
+GRANT ALTER, CREATE, DELETE, DROP, INDEX, INSERT, LOCK TABLES, SELECT, TRIGGER, UPDATE
+  ON *.* TO 'gh-ost'@'127.0.0.1';
+
 FLUSH PRIVILEGES;
 
 RESET SLAVE ALL;

docker/local/install_local_dependencies.sh

@@ -18,5 +18,10 @@ rm -f /tmp/etcd-${ETCD_VER}-linux-amd64.tar.gz
 
 mkdir -p /var/run/etcd && chown -R vitess:vitess /var/run/etcd
 
+# Install gh-ost
+curl -k -L https://github.com/github/gh-ost/releases/download/v1.0.49/gh-ost-binary-linux-20200209110835.tar.gz -o /tmp/gh-ost.tar.gz
+(cd /tmp/ && tar xzf gh-ost.tar.gz)
+cp /tmp/gh-ost /usr/bin
+
 # Clean up files we won't need in the final image.
 rm -rf /var/lib/apt/lists/*

go.mod

@@ -40,11 +40,12 @@ require (
 	github.com/hashicorp/serf v0.9.2 // indirect
 	github.com/icrowley/fake v0.0.0-20180203215853-4178557ae428
 	github.com/imdario/mergo v0.3.6 // indirect
+	github.com/julienschmidt/httprouter v1.3.0
 	github.com/klauspost/compress v1.4.1 // indirect
 	github.com/klauspost/cpuid v1.2.0 // indirect
 	github.com/klauspost/pgzip v1.2.4
 	github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
-	github.com/krishicks/yaml-patch v0.0.10
+	github.com/krishicks/yaml-patch v0.0.10 // indirect
 	github.com/magiconair/properties v1.8.1
 	github.com/mattn/go-runewidth v0.0.3 // indirect
 	github.com/minio/minio-go v0.0.0-20190131015406-c8a261de75c1

go.sum

@@ -355,6 +355,8 @@ github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVY
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
+github.com/julienschmidt/httprouter v1.3.0 h1:U0609e9tgbseu3rBINet9P48AI/D3oJs4dN7jwJOQ1U=
+github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/gotool v1.0.0 h1:AV2c/EiW3KqPNT9ZKl07ehoAGi4C5/01Cfbblndcapg=

go/test/endtoend/tabletmanager/tablet_health_test.go

@@ -126,7 +126,7 @@ func TestHealthCheck(t *testing.T) {
 
 	// stop replication, make sure we don't go unhealthy.
 	// TODO: replace with StopReplication once StopSlave has been removed
-	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopSlave", rTablet.Alias)
+	err = clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", rTablet.Alias)
 	require.NoError(t, err)
 	err = clusterInstance.VtctlclientProcess.ExecuteCommand("RunHealthCheck", rTablet.Alias)
 	require.NoError(t, err)

go/vt/mysqlctl/rice-box.go

@@ -19,7 +19,7 @@ func init() {
 		Filename:    "init_db.sql",
 		FileModTime: time.Unix(1593186461, 0),
 
-		Content: string("# This file is executed immediately after mysql_install_db,\n# to initialize a fresh data directory.\n\n###############################################################################\n# WARNING: This sql is *NOT* safe for production use,\n#          as it contains default well-known users and passwords.\n#          Care should be taken to change these users and passwords\n#          for production.\n###############################################################################\n\n###############################################################################\n# Equivalent of mysql_secure_installation\n###############################################################################\n\n# Changes during the init db should not make it to the binlog.\n# They could potentially create errant transactions on replicas.\nSET sql_log_bin = 0;\n# Remove anonymous users.\nDELETE FROM mysql.user WHERE User = '';\n\n# Disable remote root access (only allow UNIX socket).\nDELETE FROM mysql.user WHERE User = 'root' AND Host != 'localhost';\n\n# Remove test database.\nDROP DATABASE IF EXISTS test;\n\n###############################################################################\n# Vitess defaults\n###############################################################################\n\n# Vitess-internal database.\nCREATE DATABASE IF NOT EXISTS _vt;\n# Note that definitions of local_metadata and shard_metadata should be the same\n# as in production which is defined in go/vt/mysqlctl/metadata_tables.go.\nCREATE TABLE IF NOT EXISTS _vt.local_metadata (\n  name VARCHAR(255) NOT NULL,\n  value VARCHAR(255) NOT NULL,\n  db_name VARBINARY(255) NOT NULL,\n  PRIMARY KEY (db_name, name)\n  ) ENGINE=InnoDB;\nCREATE TABLE IF NOT EXISTS _vt.shard_metadata (\n  name VARCHAR(255) NOT NULL,\n  value MEDIUMBLOB NOT NULL,\n  db_name VARBINARY(255) NOT NULL,\n  PRIMARY KEY (db_name, name)\n  ) ENGINE=InnoDB;\n\n# Admin user with all privileges.\nCREATE USER 'vt_dba'@'localhost';\nGRANT ALL ON *.* TO 'vt_dba'@'localhost';\nGRANT GRANT OPTION ON *.* TO 'vt_dba'@'localhost';\n\n# User for app traffic, with global read-write access.\nCREATE USER 'vt_app'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_app'@'localhost';\n\n# User for app debug traffic, with global read access.\nCREATE USER 'vt_appdebug'@'localhost';\nGRANT SELECT, SHOW DATABASES, PROCESS ON *.* TO 'vt_appdebug'@'localhost';\n\n# User for administrative operations that need to be executed as non-SUPER.\n# Same permissions as vt_app here.\nCREATE USER 'vt_allprivs'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_allprivs'@'localhost';\n\n# User for slave replication connections.\nCREATE USER 'vt_repl'@'%';\nGRANT REPLICATION SLAVE ON *.* TO 'vt_repl'@'%';\n\n# User for Vitess filtered replication (binlog player).\n# Same permissions as vt_app.\nCREATE USER 'vt_filtered'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_filtered'@'localhost';\n\n# User for general MySQL monitoring.\nCREATE USER 'vt_monitoring'@'localhost';\nGRANT SELECT, PROCESS, SUPER, REPLICATION CLIENT, RELOAD\n  ON *.* TO 'vt_monitoring'@'localhost';\nGRANT SELECT, UPDATE, DELETE, DROP\n  ON performance_schema.* TO 'vt_monitoring'@'localhost';\n\n# User for Orchestrator (https://github.com/openark/orchestrator).\nCREATE USER 'orc_client_user'@'%' IDENTIFIED BY 'orc_client_user_password';\nGRANT SUPER, PROCESS, REPLICATION SLAVE, RELOAD\n  ON *.* TO 'orc_client_user'@'%';\nGRANT SELECT\n  ON _vt.* TO 'orc_client_user'@'%';\n\nFLUSH PRIVILEGES;\n\nRESET SLAVE ALL;\nRESET MASTER;\n"),
+		Content: string("# This file is executed immediately after mysql_install_db,\n# to initialize a fresh data directory.\n\n###############################################################################\n# WARNING: This sql is *NOT* safe for production use,\n#          as it contains default well-known users and passwords.\n#          Care should be taken to change these users and passwords\n#          for production.\n###############################################################################\n\n###############################################################################\n# Equivalent of mysql_secure_installation\n###############################################################################\n\n# Changes during the init db should not make it to the binlog.\n# They could potentially create errant transactions on replicas.\nSET sql_log_bin = 0;\n# Remove anonymous users.\nDELETE FROM mysql.user WHERE User = '';\n\n# Disable remote root access (only allow UNIX socket).\nDELETE FROM mysql.user WHERE User = 'root' AND Host != 'localhost';\n\n# Remove test database.\nDROP DATABASE IF EXISTS test;\n\n###############################################################################\n# Vitess defaults\n###############################################################################\n\n# Vitess-internal database.\nCREATE DATABASE IF NOT EXISTS _vt;\n# Note that definitions of local_metadata and shard_metadata should be the same\n# as in production which is defined in go/vt/mysqlctl/metadata_tables.go.\nCREATE TABLE IF NOT EXISTS _vt.local_metadata (\n  name VARCHAR(255) NOT NULL,\n  value VARCHAR(255) NOT NULL,\n  db_name VARBINARY(255) NOT NULL,\n  PRIMARY KEY (db_name, name)\n  ) ENGINE=InnoDB;\nCREATE TABLE IF NOT EXISTS _vt.shard_metadata (\n  name VARCHAR(255) NOT NULL,\n  value MEDIUMBLOB NOT NULL,\n  db_name VARBINARY(255) NOT NULL,\n  PRIMARY KEY (db_name, name)\n  ) ENGINE=InnoDB;\n\n# Admin user with all privileges.\nCREATE USER 'vt_dba'@'localhost';\nGRANT ALL ON *.* TO 'vt_dba'@'localhost';\nGRANT GRANT OPTION ON *.* TO 'vt_dba'@'localhost';\n\n# User for app traffic, with global read-write access.\nCREATE USER 'vt_app'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_app'@'localhost';\n\n# User for app debug traffic, with global read access.\nCREATE USER 'vt_appdebug'@'localhost';\nGRANT SELECT, SHOW DATABASES, PROCESS ON *.* TO 'vt_appdebug'@'localhost';\n\n# User for administrative operations that need to be executed as non-SUPER.\n# Same permissions as vt_app here.\nCREATE USER 'vt_allprivs'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_allprivs'@'localhost';\n\n# User for slave replication connections.\nCREATE USER 'vt_repl'@'%';\nGRANT REPLICATION SLAVE ON *.* TO 'vt_repl'@'%';\n\n# User for Vitess filtered replication (binlog player).\n# Same permissions as vt_app.\nCREATE USER 'vt_filtered'@'localhost';\nGRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, RELOAD, PROCESS, FILE,\n  REFERENCES, INDEX, ALTER, SHOW DATABASES, CREATE TEMPORARY TABLES,\n  LOCK TABLES, EXECUTE, REPLICATION SLAVE, REPLICATION CLIENT, CREATE VIEW,\n  SHOW VIEW, CREATE ROUTINE, ALTER ROUTINE, CREATE USER, EVENT, TRIGGER\n  ON *.* TO 'vt_filtered'@'localhost';\n\n# User for general MySQL monitoring.\nCREATE USER 'vt_monitoring'@'localhost';\nGRANT SELECT, PROCESS, SUPER, REPLICATION CLIENT, RELOAD\n  ON *.* TO 'vt_monitoring'@'localhost';\nGRANT SELECT, UPDATE, DELETE, DROP\n  ON performance_schema.* TO 'vt_monitoring'@'localhost';\n\n# User for Orchestrator (https://github.com/openark/orchestrator).\nCREATE USER 'orc_client_user'@'%' IDENTIFIED BY 'orc_client_user_password';\nGRANT SUPER, PROCESS, REPLICATION SLAVE, RELOAD\n  ON *.* TO 'orc_client_user'@'%';\nGRANT SELECT\n  ON _vt.* TO 'orc_client_user'@'%';\n# User for gh-ost (https://github.com/github/gh-ost).\nCREATE USER 'gh-ost'@'127.0.0.1' IDENTIFIED BY 'gh-ost';\nGRANT SUPER, REPLICATION SLAVE ON *.* TO 'gh-ost'@'127.0.0.1';\nGRANT ALTER, CREATE, DELETE, DROP, INDEX, INSERT, LOCK TABLES, SELECT, TRIGGER, UPDATE ON *.* TO 'gh-ost'@'127.0.0.1';\n\nFLUSH PRIVILEGES;\n\nRESET SLAVE ALL;\nRESET MASTER;\n"),
 	}
 	file5 := &embedded.EmbeddedFile{
 		Filename:    "mycnf/default-fast.cnf",

go/vt/mysqlctl/tmutils/schema.go

@@ -327,12 +327,16 @@ type SchemaChange struct {
 	AllowReplication bool
 	BeforeSchema     *tabletmanagerdatapb.SchemaDefinition
 	AfterSchema      *tabletmanagerdatapb.SchemaDefinition
+	Online           bool
+	Hint             string
 }
 
 // Equal compares two SchemaChange objects.
 func (s *SchemaChange) Equal(s2 *SchemaChange) bool {
 	return s.SQL == s2.SQL &&
 		s.Force == s2.Force &&
+		s.Online == s2.Online &&
+		s.Hint == s2.Hint &&
 		s.AllowReplication == s2.AllowReplication &&
 		proto.Equal(s.BeforeSchema, s2.BeforeSchema) &&
 		proto.Equal(s.AfterSchema, s2.AfterSchema)

go/vt/schemamanager/tablet_executor.go

@@ -24,6 +24,7 @@ import (
 	"golang.org/x/net/context"
 
 	"vitess.io/vitess/go/sync2"
+	"vitess.io/vitess/go/vt/mysqlctl/tmutils"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/wrangler"
 
@@ -36,6 +37,7 @@ type TabletExecutor struct {
 	tablets              []*topodatapb.Tablet
 	isClosed             bool
 	allowBigSchemaChange bool
+	onlineSchemaChange   bool
 	keyspace             string
 	waitReplicasTimeout  time.Duration
 }
@@ -46,6 +48,7 @@ func NewTabletExecutor(wr *wrangler.Wrangler, waitReplicasTimeout time.Duration)
 		wr:                   wr,
 		isClosed:             true,
 		allowBigSchemaChange: false,
+		onlineSchemaChange:   false,
 		waitReplicasTimeout:  waitReplicasTimeout,
 	}
 }
@@ -62,6 +65,11 @@ func (exec *TabletExecutor) DisallowBigSchemaChange() {
 	exec.allowBigSchemaChange = false
 }
 
+// SetOnlineSchemaChange sets TabletExecutor such that it initiates online schema change migrations
+func (exec *TabletExecutor) SetOnlineSchemaChange() {
+	exec.onlineSchemaChange = true
+}
+
 // Open opens a connection to the master for every shard.
 func (exec *TabletExecutor) Open(ctx context.Context, keyspace string) error {
 	if !exec.isClosed {
@@ -160,6 +168,11 @@ func (exec *TabletExecutor) detectBigSchemaChanges(ctx context.Context, parsedDD
 		switch ddl.Action {
 		case sqlparser.DropStr, sqlparser.CreateStr, sqlparser.TruncateStr, sqlparser.RenameStr:
 			continue
+		case sqlparser.AlterStr:
+			if exec.onlineSchemaChange {
+				// Seeing that we intend to run an online-schema-change, we can skip the "big change" check.
+				continue
+			}
 		}
 		tableName := ddl.Table.Name.String()
 		if rowCount, ok := tableWithCount[tableName]; ok {
@@ -217,15 +230,32 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 
 	for index, sql := range sqls {
 		execResult.CurSQLIndex = index
-		exec.executeOnAllTablets(ctx, &execResult, sql)
+
+		stat, err := sqlparser.Parse(sql)
+		if err != nil {
+			execResult.ExecutorErr = err.Error()
+			return &execResult
+		}
+		executeOnlineSchemaChange := false
+		switch ddl := stat.(type) {
+		case *sqlparser.DDL:
+			if ddl.Action == sqlparser.AlterStr && exec.onlineSchemaChange {
+				executeOnlineSchemaChange = true
+			}
+		}
+		exec.wr.Logger().Infof("Received DDL request. online schema change = %t", executeOnlineSchemaChange)
+		exec.executeOnAllTablets(ctx, &execResult, sql, executeOnlineSchemaChange)
 		if len(execResult.FailedShards) > 0 {
 			break
 		}
 	}
 	return &execResult
 }
 
-func (exec *TabletExecutor) executeOnAllTablets(ctx context.Context, execResult *ExecuteResult, sql string) {
+func (exec *TabletExecutor) executeOnAllTablets(
+	ctx context.Context, execResult *ExecuteResult, sql string,
+	executeOnlineSchemaChange bool,
+) {
 	var wg sync.WaitGroup
 	numOfMasterTablets := len(exec.tablets)
 	wg.Add(numOfMasterTablets)
@@ -234,7 +264,11 @@ func (exec *TabletExecutor) executeOnAllTablets(ctx context.Context, execResult
 	for _, tablet := range exec.tablets {
 		go func(tablet *topodatapb.Tablet) {
 			defer wg.Done()
-			exec.executeOneTablet(ctx, tablet, sql, errChan, successChan)
+			if executeOnlineSchemaChange {
+				exec.executeOnlineSchemaChangeOneTablet(ctx, tablet, sql, errChan, successChan)
+			} else {
+				exec.executeOneTablet(ctx, tablet, sql, errChan, successChan)
+			}
 		}(tablet)
 	}
 	wg.Wait()
@@ -297,6 +331,29 @@ func (exec *TabletExecutor) executeOneTablet(
 	}
 }
 
+// executeOnlineSchemaChangeOneTablet will request the tablet to run an online schema change
+func (exec *TabletExecutor) executeOnlineSchemaChangeOneTablet(
+	ctx context.Context,
+	tablet *topodatapb.Tablet,
+	sql string,
+	errChan chan ShardWithError,
+	successChan chan ShardResult,
+) {
+	change := &tmutils.SchemaChange{
+		SQL:    sql,
+		Online: true,
+		Hint:   "", // TODO(shlomi) generate and populate hint
+	}
+	_, err := exec.wr.TabletManagerClient().ApplySchema(ctx, tablet, change)
+	if err != nil {
+		errChan <- ShardWithError{Shard: tablet.Shard, Err: err.Error()}
+		return
+	}
+	successChan <- ShardResult{
+		Shard: tablet.Shard,
+	}
+}
+
 // Close clears tablet executor states
 func (exec *TabletExecutor) Close() {
 	if !exec.isClosed {