VReplication: Improve handling of vplayer stalls

go/test/endtoend/vreplication/cluster_test.go

@@ -60,7 +60,7 @@ var (
 		"--buffer_size", "250000", "--buffer_min_time_between_failovers", "1s", "--buffer_max_failover_duration", loadTestBufferingWindowDuration.String(),
 		"--buffer_drain_concurrency", "10"}
 	extraVtctldArgs = []string{"--remote_operation_timeout", "600s", "--topo_etcd_lease_ttl", "120"}
-	// This variable can be used within specific tests to alter vttablet behavior
+	// This variable can be used within specific tests to alter vttablet behavior.
 	extraVTTabletArgs = []string{}
 
 	parallelInsertWorkers = "--vreplication-parallel-insert-workers=4"

go/vt/vttablet/tabletmanager/vreplication/framework_test.go

@@ -32,8 +32,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
-	"vitess.io/vitess/go/vt/sqlparser"
-
 	_flag "vitess.io/vitess/go/internal/flag"
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/mysql/replication"
@@ -46,6 +44,7 @@ import (
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sidecardb"
+	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/vttablet"
 	"vitess.io/vitess/go/vt/vttablet/queryservice"
@@ -74,6 +73,7 @@ var (
 	testForeignKeyQueries    = false
 	testSetForeignKeyQueries = false
 	doNotLogDBQueries        = false
+	recvTimeout              = 5 * time.Second
 )
 
 type LogExpectation struct {
@@ -492,14 +492,14 @@ func (dbc *realDBClient) ExecuteFetch(query string, maxrows int) (*sqltypes.Resu
 	return qr, err
 }
 
-func (dc *realDBClient) ExecuteFetchMulti(query string, maxrows int) ([]*sqltypes.Result, error) {
+func (dbc *realDBClient) ExecuteFetchMulti(query string, maxrows int) ([]*sqltypes.Result, error) {
 	queries, err := sqlparser.NewTestParser().SplitStatementToPieces(query)
 	if err != nil {
 		return nil, err
 	}
 	results := make([]*sqltypes.Result, 0, len(queries))
 	for _, query := range queries {
-		qr, err := dc.ExecuteFetch(query, maxrows)
+		qr, err := dbc.ExecuteFetch(query, maxrows)
 		if err != nil {
 			return nil, err
 		}
@@ -518,7 +518,7 @@ func expectDeleteQueries(t *testing.T) {
 		"/delete from _vt.vreplication",
 		"/delete from _vt.copy_state",
 		"/delete from _vt.post_copy_action",
-	))
+	), recvTimeout)
 }
 
 func deleteAllVReplicationStreams(t *testing.T) {
@@ -635,7 +635,7 @@ func expectDBClientQueries(t *testing.T, expectations qh.ExpectationSequence, sk
 				))
 			}
 		case <-time.After(5 * time.Second):
-			t.Fatalf("no query received")
+			require.FailNow(t, "no query received")
 			failed = true
 		}
 	}
@@ -656,7 +656,7 @@ func expectDBClientQueries(t *testing.T, expectations qh.ExpectationSequence, sk
 
 // expectNontxQueries disregards transactional statements like begin and commit.
 // It also disregards updates to _vt.vreplication.
-func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence) {
+func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence, recvTimeout time.Duration) {
 	t.Helper()
 	if doNotLogDBQueries {
 		return
@@ -684,8 +684,8 @@ func expectNontxQueries(t *testing.T, expectations qh.ExpectationSequence) {
 				"query:%q\nmessage:%s\nexpectation:%s\nmatched:%t\nerror:%v\nhistory:%s",
 				got, result.Message, result.Expectation, result.Matched, result.Error, validator.History(),
 			))
-		case <-time.After(5 * time.Second):
-			t.Fatalf("no query received")
+		case <-time.After(recvTimeout):
+			require.FailNow(t, "no query received")
 			failed = true
 		}
 	}

go/vt/vttablet/tabletmanager/vreplication/relaylog.go

@@ -17,15 +17,18 @@ limitations under the License.
 package vreplication
 
 import (
+	"context"
 	"io"
 	"sync"
 	"time"
 
-	"context"
+	"vitess.io/vitess/go/vt/vterrors"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+const relayLogIOStalledMsg = "relay log I/O stalled"
+
 type relayLog struct {
 	ctx      context.Context
 	maxItems int
@@ -72,12 +75,18 @@ func (rl *relayLog) Send(events []*binlogdatapb.VEvent) error {
 	if err := rl.checkDone(); err != nil {
 		return err
 	}
+	cancelTimer := rl.startSendTimer()
+	defer cancelTimer()
 	for rl.curSize > rl.maxSize || len(rl.items) >= rl.maxItems {
 		rl.canAccept.Wait()
+		if rl.timedout {
+			return vterrors.Wrap(errVPlayerStalled, relayLogIOStalledMsg)
+		}
 		if err := rl.checkDone(); err != nil {
 			return err
 		}
 	}
+	rl.timedout = false
 	rl.items = append(rl.items, events)
 	rl.curSize += eventsSize(events)
 	rl.hasItems.Broadcast()
@@ -92,7 +101,7 @@ func (rl *relayLog) Fetch() ([][]*binlogdatapb.VEvent, error) {
 	if err := rl.checkDone(); err != nil {
 		return nil, err
 	}
-	cancelTimer := rl.startTimer()
+	cancelTimer := rl.startFetchTimer()
 	defer cancelTimer()
 	for len(rl.items) == 0 && !rl.timedout {
 		rl.hasItems.Wait()
@@ -117,7 +126,33 @@ func (rl *relayLog) checkDone() error {
 	return nil
 }
 
-func (rl *relayLog) startTimer() (cancel func()) {
+// startSendTimer starts a timer that will wake up the sender if we hit
+// the vplayerProgressDeadline timeout. This ensures that we don't
+// block forever if the vplayer cannot process the previous relay log
+// contents in a timely manner; allowing us to provide the user with a
+// helpful error message.
+func (rl *relayLog) startSendTimer() (cancel func()) {
+	timer := time.NewTimer(vplayerProgressDeadline)
+	timerDone := make(chan struct{})
+	go func() {
+		select {
+		case <-timer.C:
+			rl.mu.Lock()
+			defer rl.mu.Unlock()
+			rl.timedout = true
+			rl.canAccept.Broadcast()
+		case <-timerDone:
+		}
+	}()
+	return func() {
+		timer.Stop()
+		close(timerDone)
+	}
+}
+
+// startFetchTimer starts a timer that will wake up the fetcher after
+// idleTimeout to be sure that we're regularly checking for new events.
+func (rl *relayLog) startFetchTimer() (cancel func()) {
 	timer := time.NewTimer(idleTimeout)
 	timerDone := make(chan struct{})
 	go func() {

go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go

@@ -182,7 +182,7 @@ func testPlayerCopyCharPK(t *testing.T) {
 		`/insert into _vt.copy_state \(lastpk, vrepl_id, table_name\) values \('fields:{name:\\"idc\\" type:BINARY charset:63 flags:20611} rows:{lengths:2 values:\\"c\\\\x00\\"}'.*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"a\000", "3"},
@@ -304,7 +304,7 @@ func testPlayerCopyVarcharPKCaseInsensitive(t *testing.T) {
 	}).Then(qh.Immediately(
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"a", "1"},
@@ -415,7 +415,7 @@ func testPlayerCopyVarcharCompositePKCaseSensitiveCollation(t *testing.T) {
 		// Wrap-up.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"1", "B", "B", "3"},
@@ -790,7 +790,7 @@ func testPlayerCopyBigTable(t *testing.T) {
 		// Copy is done. Go into running state.
 		// All tables copied. Final catch up followed by Running state.
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"1", "aaa"},
@@ -918,7 +918,7 @@ func testPlayerCopyWildcardRule(t *testing.T) {
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*src",
 		// Copy is done. Go into running state.
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "src", [][]string{
 		{"1", "aaa"},
@@ -1078,7 +1078,7 @@ func testPlayerCopyTableContinuation(t *testing.T) {
 	)).Then(qh.Immediately(
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*not_copied",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "insert in"},
@@ -1188,7 +1188,7 @@ func testPlayerCopyWildcardTableContinuation(t *testing.T) {
 		`/insert into _vt.copy_state .*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	)))
+	)), recvTimeout)
 
 	expectData(t, "dst", [][]string{
 		{"2", "copied"},
@@ -1279,7 +1279,7 @@ func TestPlayerCopyWildcardTableContinuationWithOptimizeInserts(t *testing.T) {
 		`/insert into _vt.copy_state .*`,
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 	expectData(t, "dst", [][]string{
 		{"2", "copied"},
 		{"3", "uncopied"},
@@ -1659,7 +1659,7 @@ func testPlayerCopyTablesWithGeneratedColumn(t *testing.T) {
 		// copy of dst2 is done: delete from copy_state.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst2",
 		"/update _vt.vreplication set state",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "aaa", "1aaa", "aaa1", "10"},
@@ -1826,7 +1826,7 @@ func testCopyInvisibleColumns(t *testing.T) {
 		// copy of dst1 is done: delete from copy_state.
 		"/delete cs, pca from _vt.copy_state as cs left join _vt.post_copy_action as pca on cs.vrepl_id=pca.vrepl_id and cs.table_name=pca.table_name.*dst1",
 		"/update _vt.vreplication set state='Running'",
-	))
+	), recvTimeout)
 
 	expectData(t, "dst1", [][]string{
 		{"1", "10"},

go/vt/vttablet/tabletmanager/vreplication/vplayer.go

@@ -30,12 +30,25 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/throttlerapp"
 
 	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
+const failedToRecordHeartbeatMsg = "failed to record heartbeat"
+
+var (
+	// At what point should we consider the vplayer to be stalled and return an error.
+	// 5 minutes is well beyond a reasonable amount of time for a transaction to be
+	// replicated.
+	vplayerProgressDeadline = time.Duration(5 * time.Minute)
+
+	// The error to return when we have detected a stall in the vplayer.
+	errVPlayerStalled = errors.New("progress stalled; vplayer was unable to replicate the transaction in a timely manner; examine the target mysqld instance health and the replicated queries' EXPLAIN output to see why queries are taking unusually long")
+)
+
 // vplayer replays binlog events by pulling them from a vstreamer.
 type vplayer struct {
 	vr        *vreplicator
@@ -367,12 +380,13 @@ func (vp *vplayer) applyRowEvent(ctx context.Context, rowEvent *binlogdatapb.Row
 	return nil
 }
 
+// updatePos should get called at a minimum of vreplicationMinimumHeartbeatUpdateInterval.
 func (vp *vplayer) updatePos(ctx context.Context, ts int64) (posReached bool, err error) {
-	vp.numAccumulatedHeartbeats = 0
 	update := binlogplayer.GenerateUpdatePos(vp.vr.id, vp.pos, time.Now().Unix(), ts, vp.vr.stats.CopyRowCount.Get(), vreplicationStoreCompressedGTID)
 	if _, err := vp.query(ctx, update); err != nil {
 		return false, fmt.Errorf("error %v updating position", err)
 	}
+	vp.numAccumulatedHeartbeats = 0
 	vp.unsavedEvent = nil
 	vp.timeLastSaved = time.Now()
 	vp.vr.stats.SetLastPosition(vp.pos)
@@ -399,8 +413,16 @@ func (vp *vplayer) recordHeartbeat() error {
 	if !vp.mustUpdateHeartbeat() {
 		return nil
 	}
+	if err := vp.vr.updateHeartbeatTime(tm); err != nil {
+		return vterrors.Wrapf(errVPlayerStalled, fmt.Sprintf("%s: %v", failedToRecordHeartbeatMsg, err))
+	}
+	// Only reset the pending heartbeat count if the update was successful.
+	// Otherwise the vplayer may not actually be making progress and nobody
+	// is aware of it -- resulting in the com_binlog_dump connection on the
+	// source that is managed by the binlog_player getting closed by mysqld
+	// when the source_net_timeout is hit.
 	vp.numAccumulatedHeartbeats = 0
-	return vp.vr.updateHeartbeatTime(tm)
+	return nil
 }
 
 // applyEvents is the main thread that applies the events. It has the following use
@@ -438,7 +460,7 @@ func (vp *vplayer) recordHeartbeat() error {
 // current position to be saved.
 //
 // In order to handle the above use cases, we use an implicit transaction scheme:
-// A BEGIN does not really start a transaction. Ony a ROW event does. With this
+// A BEGIN does not really start a transaction. Only a ROW event does. With this
 // approach, no transaction gets started if an empty one arrives. If a we receive
 // a commit, and a we are not in a transaction, we infer that the transaction was
 // empty, and remember it as an unsaved event. The next GTID event will reset the
@@ -497,6 +519,7 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				return nil
 			}
 		}
+
 		for i, events := range items {
 			for j, event := range events {
 				if event.Timestamp != 0 {
@@ -526,7 +549,17 @@ func (vp *vplayer) applyEvents(ctx context.Context, relay *relayLog) error {
 				if err := vp.applyEvent(ctx, event, mustSave); err != nil {
 					if err != io.EOF {
 						vp.vr.stats.ErrorCounts.Add([]string{"Apply"}, 1)
-						log.Errorf("Error applying event: %s", err.Error())
+						var table, tableLogMsg string
+						switch {
+						case event.GetFieldEvent() != nil:
+							table = event.GetFieldEvent().TableName
+						case event.GetRowEvent() != nil:
+							table = event.GetRowEvent().TableName
+						}
+						if table != "" {
+							tableLogMsg = fmt.Sprintf(" for table %s", table)
+						}
+						log.Errorf("Error applying event%s: %s", tableLogMsg, err.Error())
 					}
 					return err
 				}
@@ -635,7 +668,8 @@ func (vp *vplayer) applyEvent(ctx context.Context, event *binlogdatapb.VEvent, m
 			log.Infof("Error applying row event: %s", err.Error())
 			return err
 		}
-		//Row event is logged AFTER RowChanges are applied so as to calculate the total elapsed time for the Row event
+		// Row event is logged AFTER RowChanges are applied so as to calculate the total elapsed
+		// time for the Row event.
 		stats.Send(fmt.Sprintf("%v", event.RowEvent))
 	case binlogdatapb.VEventType_OTHER:
 		if vp.vr.dbClient.InTransaction {