Merge pull request #617 from github/gtid-sets

Support for gtid-errant-reset-master command; formalized GTID sets

go/app/cli.go

@@ -572,10 +572,10 @@ func Cli(command string, strict bool, instance string, destination string, owner
 			}
 			fmt.Println(instanceKey.DisplayString())
 		}
-	case registerCliCommand("reset-master-gtid-remove-own-uuid", "Replication, general", `Reset master on instance, remove GTID entries generated by instance`):
+	case registerCliCommand("gtid-errant-reset-master", "Replication, general", `Reset master on instance, remove GTID errant transactions`):
 		{
 			instanceKey, _ = inst.FigureInstanceKey(instanceKey, thisInstanceKey)
-			_, err := inst.ResetMasterGTIDOperation(instanceKey, true, "")
+			_, err := inst.ErrantGTIDResetMaster(instanceKey)
 			if err != nil {
 				log.Fatale(err)
 			}

go/db/generate_patches.go

@@ -520,4 +520,9 @@ var generateSQLPatches = []string{
 			database_instance
 			ADD COLUMN gtid_errant text CHARACTER SET ascii NOT NULL AFTER gtid_purged
 	`,
+	`
+		ALTER TABLE
+			database_instance
+			ADD COLUMN ancestry_uuid text CHARACTER SET ascii NOT NULL AFTER master_uuid
+	`,
 }

go/http/api.go

@@ -722,6 +722,27 @@ func (this *HttpAPI) DisableGTID(params martini.Params, r render.Render, req *ht
 	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Disabled GTID on %+v", instance.Key), Details: instance})
 }
 
+// ErrantGTIDResetMaster removes errant transactions on a server by way of RESET MASTER
+func (this *HttpAPI) ErrantGTIDResetMaster(params martini.Params, r render.Render, req *http.Request, user auth.User) {
+	if !isAuthorizedForAction(req, user) {
+		Respond(r, &APIResponse{Code: ERROR, Message: "Unauthorized"})
+		return
+	}
+	instanceKey, err := this.getInstanceKey(params["host"], params["port"])
+
+	if err != nil {
+		Respond(r, &APIResponse{Code: ERROR, Message: err.Error()})
+		return
+	}
+	instance, err := inst.ErrantGTIDResetMaster(&instanceKey)
+	if err != nil {
+		Respond(r, &APIResponse{Code: ERROR, Message: err.Error()})
+		return
+	}
+
+	Respond(r, &APIResponse{Code: OK, Message: fmt.Sprintf("Removed errant GTID on %+v and issued a RESET MASTER", instance.Key), Details: instance})
+}
+
 // MoveBelow attempts to move an instance below its supposed sibling
 func (this *HttpAPI) MoveBelow(params martini.Params, r render.Render, req *http.Request, user auth.User) {
 	if !isAuthorizedForAction(req, user) {
@@ -3302,6 +3323,7 @@ func (this *HttpAPI) RegisterRequests(m *martini.ClassicMartini) {
 	// Replication, general:
 	this.registerAPIRequest(m, "enable-gtid/:host/:port", this.EnableGTID)
 	this.registerAPIRequest(m, "disable-gtid/:host/:port", this.DisableGTID)
+	this.registerAPIRequest(m, "gtid-errant-reset-master/:host/:port", this.ErrantGTIDResetMaster)
 	this.registerAPIRequest(m, "skip-query/:host/:port", this.SkipQuery)
 	this.registerAPIRequest(m, "start-slave/:host/:port", this.StartSlave)
 	this.registerAPIRequest(m, "restart-slave/:host/:port", this.RestartSlave)

go/inst/instance.go

@@ -48,6 +48,7 @@ type Instance struct {
 	SelfBinlogCoordinates  BinlogCoordinates
 	MasterKey              InstanceKey
 	MasterUUID             string
+	AncestryUUID           string
 	IsDetachedMaster       bool
 	Slave_SQL_Running      bool
 	Slave_IO_Running       bool

go/inst/instance_dao.go

@@ -770,14 +770,27 @@ Cleanup:
 	waitGroup.Wait()
 
 	if instanceFound {
+		instance.AncestryUUID = strings.Trim(fmt.Sprintf("%s,%s", instance.AncestryUUID, instance.ServerUUID), ",")
 		if instance.ExecutedGtidSet != "" && instance.masterExecutedGtidSet != "" {
 			// Compare master & replica GTID sets, but ignore the sets that present the master's UUID.
 			// This is because orchestrator may pool master and replica at an inconvenient timing,
 			// such that the replica may _seems_ to have more entries than the master, when in fact
 			// it's just that the naster's probing is stale.
-			redactedExecutedGtidSet := redactGtidSetUUID(instance.ExecutedGtidSet, instance.MasterUUID)
-			redactedMasterExecutedGtidSet := redactGtidSetUUID(instance.masterExecutedGtidSet, instance.MasterUUID)
-			db.QueryRow("select gtid_subtract(?, ?)", redactedExecutedGtidSet, redactedMasterExecutedGtidSet).Scan(&instance.GtidErrant)
+			// redactedExecutedGtidSet := redactGtidSetUUID(instance.ExecutedGtidSet, instance.MasterUUID)
+			// redactedMasterExecutedGtidSet := redactGtidSetUUID(instance.masterExecutedGtidSet, instance.MasterUUID)
+			redactedExecutedGtidSet, _ := NewOracleGtidSet(instance.ExecutedGtidSet)
+			for _, uuid := range strings.Split(instance.AncestryUUID, ",") {
+				if uuid != instance.ServerUUID {
+					redactedExecutedGtidSet.RemoveUUID(uuid)
+				}
+			}
+			// Avoid querying the database if there's no point:
+			if !redactedExecutedGtidSet.IsEmpty() {
+				redactedMasterExecutedGtidSet, _ := NewOracleGtidSet(instance.masterExecutedGtidSet)
+				redactedMasterExecutedGtidSet.RemoveUUID(instance.MasterUUID)
+
+				db.QueryRow("select gtid_subtract(?, ?)", redactedExecutedGtidSet.String(), redactedMasterExecutedGtidSet.String()).Scan(&instance.GtidErrant)
+			}
 		}
 	}
 
@@ -841,6 +854,7 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	var masterClusterName string
 	var masterSuggestedClusterAlias string
 	var masterReplicationDepth uint
+	var ancestryUUID string
 	var masterExecutedGtidSet string
 	masterDataFound := false
 
@@ -852,6 +866,7 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 					replication_depth,
 					master_host,
 					master_port,
+					ancestry_uuid,
 					executed_gtid_set
 				from database_instance
 				where hostname=? and port=?
@@ -864,6 +879,7 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 		masterReplicationDepth = m.GetUint("replication_depth")
 		masterMasterKey.Hostname = m.GetString("master_host")
 		masterMasterKey.Port = m.GetInt("master_port")
+		ancestryUUID = m.GetString("ancestry_uuid")
 		masterExecutedGtidSet = m.GetString("executed_gtid_set")
 		masterDataFound = true
 		return nil
@@ -903,6 +919,7 @@ func ReadInstanceClusterAttributes(instance *Instance) (err error) {
 	instance.SuggestedClusterAlias = masterSuggestedClusterAlias
 	instance.ReplicationDepth = replicationDepth
 	instance.IsCoMaster = isCoMaster
+	instance.AncestryUUID = ancestryUUID
 	instance.masterExecutedGtidSet = masterExecutedGtidSet
 	return nil
 }
@@ -990,6 +1007,7 @@ func readInstanceRow(m sqlutils.RowMap) *Instance {
 	instance.SupportsOracleGTID = m.GetBool("supports_oracle_gtid")
 	instance.UsingOracleGTID = m.GetBool("oracle_gtid")
 	instance.MasterUUID = m.GetString("master_uuid")
+	instance.AncestryUUID = m.GetString("ancestry_uuid")
 	instance.ExecutedGtidSet = m.GetString("executed_gtid_set")
 	instance.GTIDMode = m.GetString("gtid_mode")
 	instance.GtidPurged = m.GetString("gtid_purged")
@@ -2189,6 +2207,7 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		"supports_oracle_gtid",
 		"oracle_gtid",
 		"master_uuid",
+		"ancestry_uuid",
 		"executed_gtid_set",
 		"gtid_mode",
 		"gtid_purged",
@@ -2265,6 +2284,7 @@ func mkInsertOdkuForInstances(instances []*Instance, instanceWasActuallyFound bo
 		args = append(args, instance.SupportsOracleGTID)
 		args = append(args, instance.UsingOracleGTID)
 		args = append(args, instance.MasterUUID)
+		args = append(args, instance.AncestryUUID)
 		args = append(args, instance.ExecutedGtidSet)
 		args = append(args, instance.GTIDMode)
 		args = append(args, instance.GtidPurged)