Consider repl NOT healthy with IO_Thread connecting and last attempt failed

go/cmd/vtbackup/vtbackup.go

@@ -378,7 +378,7 @@ func takeBackup(ctx context.Context, topoServer *topo.Server, backupStorage back
 			log.Infof("Replication caught up to %v after %v", status.Position, time.Since(waitStartTime))
 			break
 		}
-		if !status.ReplicationRunning() {
+		if !status.Healthy() {
 			log.Warning("Replication has stopped before backup could be taken. Trying to restart replication.")
 			if err := startReplication(ctx, mysqld, topoServer); err != nil {
 				log.Warningf("Failed to restart replication: %v", err)

go/mysql/constants.go

@@ -680,3 +680,33 @@ func IsSchemaApplyError(err error) bool {
 	}
 	return false
 }
+
+type ReplicationState int
+
+const (
+	ReplicationStateUnknown ReplicationState = iota
+	ReplicationStateStopped
+	ReplicationStateConnecting
+	ReplicationStateRunning
+)
+
+// ReplicationStatusToState converts a value you have for the IO thread(s) or SQL
+// thread(s) or Group Replication applier thread(s) from MySQL or intermediate
+// layers to a mysql.ReplicationState.
+// on,yes,true == ReplicationStateRunning
+// off,no,false == ReplicationStateStopped
+// connecting == ReplicationStateConnecting
+// anything else == ReplicationStateUnknown
+func ReplicationStatusToState(s string) ReplicationState {
+	// Group Replication uses ON instead of Yes
+	switch strings.ToLower(s) {
+	case "yes", "on", "true":
+		return ReplicationStateRunning
+	case "no", "off", "false":
+		return ReplicationStateStopped
+	case "connecting":
+		return ReplicationStateConnecting
+	default:
+		return ReplicationStateUnknown
+	}
+}

go/mysql/flavor.go

@@ -350,8 +350,10 @@ func parseReplicationStatus(fields map[string]string) ReplicationStatus {
 	status := ReplicationStatus{
 		SourceHost: fields["Master_Host"],
 		// These fields are returned from the underlying DB and cannot be renamed
-		IOThreadRunning:  fields["Slave_IO_Running"] == "Yes" || fields["Slave_IO_Running"] == "Connecting",
-		SQLThreadRunning: fields["Slave_SQL_Running"] == "Yes",
+		IOState:      ReplicationStatusToState(fields["Slave_IO_Running"]),
+		LastIOError:  fields["Last_IO_Error"],
+		SQLState:     ReplicationStatusToState(fields["Slave_SQL_Running"]),
+		LastSQLError: fields["Last_SQL_Error"],
 	}
 	parseInt, _ := strconv.ParseInt(fields["Master_Port"], 10, 0)
 	status.SourcePort = int(parseInt)

go/mysql/flavor_mysqlgr.go

@@ -141,32 +141,32 @@ func (mysqlGRFlavor) status(c *Conn) (ReplicationStatus, error) {
 		return res, nil
 	}
 
-	// Populate IOThreadRunning from replication_connection_status
+	// Populate IOState from replication_connection_status
 	query = fmt.Sprintf(`SELECT SERVICE_STATE
 		FROM performance_schema.replication_connection_status
 		WHERE CHANNEL_NAME='%s'`, chanel)
-	var ioThreadRunning bool
+	var connectionState ReplicationState
 	err = fetchStatusForGroupReplication(c, query, func(values []sqltypes.Value) error {
-		ioThreadRunning = values[0].ToString() == "ON"
+		connectionState = ReplicationStatusToState(values[0].ToString())
 		return nil
 	})
 	if err != nil {
 		return ReplicationStatus{}, err
 	}
-	res.IOThreadRunning = ioThreadRunning
-	// Populate SQLThreadRunning from replication_connection_status
-	var sqlThreadRunning bool
+	res.IOState = connectionState
+	// Populate SQLState from replication_connection_status
+	var applierState ReplicationState
 	query = fmt.Sprintf(`SELECT SERVICE_STATE
 		FROM performance_schema.replication_applier_status_by_coordinator
 		WHERE CHANNEL_NAME='%s'`, chanel)
 	err = fetchStatusForGroupReplication(c, query, func(values []sqltypes.Value) error {
-		sqlThreadRunning = values[0].ToString() == "ON"
+		applierState = ReplicationStatusToState(values[0].ToString())
 		return nil
 	})
 	if err != nil {
 		return ReplicationStatus{}, err
 	}
-	res.SQLThreadRunning = sqlThreadRunning
+	res.SQLState = applierState
 
 	// Collect lag information
 	// we use the difference between the last processed transaction's commit time

go/mysql/flavor_mysqlgr_test.go

@@ -33,8 +33,8 @@ func TestMysqlGRParsePrimaryGroupMember(t *testing.T) {
 	parsePrimaryGroupMember(&res, rows)
 	assert.Equal(t, "host1", res.SourceHost)
 	assert.Equal(t, 10, res.SourcePort)
-	assert.Equal(t, false, res.IOThreadRunning)
-	assert.Equal(t, false, res.SQLThreadRunning)
+	assert.Equal(t, ReplicationStateUnknown, res.IOState)
+	assert.Equal(t, ReplicationStateUnknown, res.SQLState)
 }
 
 func TestMysqlGRReplicationApplierLagParse(t *testing.T) {

go/mysql/replication_status.go

@@ -37,8 +37,10 @@ type ReplicationStatus struct {
 	FilePosition          Position
 	FileRelayLogPosition  Position
 	SourceServerID        uint
-	IOThreadRunning       bool
-	SQLThreadRunning      bool
+	IOState               ReplicationState
+	LastIOError           string
+	SQLState              ReplicationState
+	LastSQLError          string
 	ReplicationLagSeconds uint
 	ReplicationLagUnknown bool
 	SourceHost            string
@@ -47,10 +49,28 @@ type ReplicationStatus struct {
 	SourceUUID            SID
 }
 
-// ReplicationRunning returns true iff both the IO and SQL threads are
-// running.
-func (s *ReplicationStatus) ReplicationRunning() bool {
-	return s.IOThreadRunning && s.SQLThreadRunning
+// Running returns true if both the IO and SQL threads are running.
+func (s *ReplicationStatus) Running() bool {
+	return s.IOState == ReplicationStateRunning && s.SQLState == ReplicationStateRunning
+}
+
+// Healthy returns true if both the SQL IO components are healthy
+func (s *ReplicationStatus) Healthy() bool {
+	return s.SQLHealthy() && s.IOHealthy()
+}
+
+// IOHealthy returns true if the IO thread is running OR, the
+// IO thread is connecting AND there's no IO error from the last
+// attempt to connect to the source.
+func (s *ReplicationStatus) IOHealthy() bool {
+	return s.IOState == ReplicationStateRunning ||
+		(s.IOState == ReplicationStateConnecting && s.LastIOError == "")
+}
+
+// SQLHealthy returns true if the SQLState is running.
+// For consistency and to support altering this calculation in the future.
+func (s *ReplicationStatus) SQLHealthy() bool {
+	return s.SQLState == ReplicationStateRunning
 }
 
 // ReplicationStatusToProto translates a Status to proto3.
@@ -61,13 +81,15 @@ func ReplicationStatusToProto(s ReplicationStatus) *replicationdatapb.Status {
 		FilePosition:          EncodePosition(s.FilePosition),
 		FileRelayLogPosition:  EncodePosition(s.FileRelayLogPosition),
 		SourceServerId:        uint32(s.SourceServerID),
-		IoThreadRunning:       s.IOThreadRunning,
-		SqlThreadRunning:      s.SQLThreadRunning,
 		ReplicationLagSeconds: uint32(s.ReplicationLagSeconds),
 		SourceHost:            s.SourceHost,
 		SourcePort:            int32(s.SourcePort),
 		ConnectRetry:          int32(s.ConnectRetry),
 		SourceUuid:            s.SourceUUID.String(),
+		IoState:               int32(s.IOState),
+		LastIoError:           s.LastIOError,
+		SqlState:              int32(s.SQLState),
+		LastSqlError:          s.LastSQLError,
 	}
 }
 
@@ -102,13 +124,15 @@ func ProtoToReplicationStatus(s *replicationdatapb.Status) ReplicationStatus {
 		FilePosition:          filePos,
 		FileRelayLogPosition:  fileRelayPos,
 		SourceServerID:        uint(s.SourceServerId),
-		IOThreadRunning:       s.IoThreadRunning,
-		SQLThreadRunning:      s.SqlThreadRunning,
 		ReplicationLagSeconds: uint(s.ReplicationLagSeconds),
 		SourceHost:            s.SourceHost,
 		SourcePort:            int(s.SourcePort),
 		ConnectRetry:          int(s.ConnectRetry),
 		SourceUUID:            sid,
+		IOState:               ReplicationState(s.IoState),
+		LastIOError:           s.LastIoError,
+		SQLState:              ReplicationState(s.SqlState),
+		LastSQLError:          s.LastSqlError,
 	}
 }
 

go/mysql/replication_status_test.go

@@ -24,34 +24,34 @@ import (
 
 func TestStatusReplicationRunning(t *testing.T) {
 	input := &ReplicationStatus{
-		IOThreadRunning:  true,
-		SQLThreadRunning: true,
+		IOState:  ReplicationStatusToState("yes"),
+		SQLState: ReplicationStatusToState("yes"),
 	}
 	want := true
-	if got := input.ReplicationRunning(); got != want {
-		t.Errorf("%#v.ReplicationRunning() = %v, want %v", input, got, want)
+	if got := input.Running(); got != want {
+		t.Errorf("%#v.Running() = %v, want %v", input, got, want)
 	}
 }
 
 func TestStatusIOThreadNotRunning(t *testing.T) {
 	input := &ReplicationStatus{
-		IOThreadRunning:  false,
-		SQLThreadRunning: true,
+		IOState:  ReplicationStatusToState("no"),
+		SQLState: ReplicationStatusToState("yes"),
 	}
 	want := false
-	if got := input.ReplicationRunning(); got != want {
-		t.Errorf("%#v.ReplicationRunning() = %v, want %v", input, got, want)
+	if got := input.Running(); got != want {
+		t.Errorf("%#v.Running() = %v, want %v", input, got, want)
 	}
 }
 
 func TestStatusSQLThreadNotRunning(t *testing.T) {
 	input := &ReplicationStatus{
-		IOThreadRunning:  true,
-		SQLThreadRunning: false,
+		IOState:  ReplicationStatusToState("yes"),
+		SQLState: ReplicationStatusToState("no"),
 	}
 	want := false
-	if got := input.ReplicationRunning(); got != want {
-		t.Errorf("%#v.ReplicationRunning() = %v, want %v", input, got, want)
+	if got := input.Running(); got != want {
+		t.Errorf("%#v.Running() = %v, want %v", input, got, want)
 	}
 }
 

go/vt/mysqlctl/builtinbackupengine.go

@@ -159,7 +159,7 @@ func (be *BuiltinBackupEngine) ExecuteBackup(ctx context.Context, params BackupP
 	replicaStatus, err := params.Mysqld.ReplicationStatus()
 	switch err {
 	case nil:
-		replicaStartRequired = replicaStatus.ReplicationRunning() && !*DisableActiveReparents
+		replicaStartRequired = replicaStatus.Healthy() && !*DisableActiveReparents
 	case mysql.ErrNotReplica:
 		// keep going if we're the primary, might be a degenerate case
 		sourceIsPrimary = true

go/vt/mysqlctl/fakemysqldaemon/fakemysqldaemon.go

@@ -271,10 +271,10 @@ func (fmd *FakeMysqlDaemon) ReplicationStatus() (mysql.ReplicationStatus, error)
 		ReplicationLagSeconds: fmd.ReplicationLagSeconds,
 		// implemented as AND to avoid changing all tests that were
 		// previously using Replicating = false
-		IOThreadRunning:  fmd.Replicating && fmd.IOThreadRunning,
-		SQLThreadRunning: fmd.Replicating,
-		SourceHost:       fmd.CurrentSourceHost,
-		SourcePort:       fmd.CurrentSourcePort,
+		IOState:    mysql.ReplicationStatusToState(fmt.Sprintf("%v", fmd.Replicating && fmd.IOThreadRunning)),
+		SQLState:   mysql.ReplicationStatusToState(fmt.Sprintf("%v", fmd.Replicating)),
+		SourceHost: fmd.CurrentSourceHost,
+		SourcePort: fmd.CurrentSourcePort,
 	}, nil
 }
 

go/vt/mysqlctl/replication.go

@@ -47,7 +47,7 @@ func WaitForReplicationStart(mysqld MysqlDaemon, replicaStartDeadline int) error
 			return err
 		}
 
-		if status.ReplicationRunning() {
+		if status.Running() {
 			return nil
 		}
 		time.Sleep(time.Second)