VReplication: Handle vstream stalls more effectively (#18752)

Signed-off-by: Matt Lord <[email protected]>

Author: mattlord

go/vt/vtgate/vstream_manager.go

@@ -70,6 +70,10 @@ const tabletPickerContextTimeout = 90 * time.Second
 // ending the stream from the tablet.
 const stopOnReshardDelay = 500 * time.Millisecond
 
+// livenessTimeout is the point at which we return an error to the client if the stream has received
+// no events, including heartbeats, from any of the shards.
+var livenessTimeout = 10 * time.Minute
+
 // vstream contains the metadata for one VStream request.
 type vstream struct {
 	// mu protects parts of vgtid, the semantics of a send, and journaler.
@@ -136,6 +140,9 @@ type vstream struct {
 
 	tabletPickerOptions discovery.TabletPickerOptions
 
+	// At what point, without any activity in the stream, should we consider it dead.
+	streamLivenessTimer *time.Timer
+
 	flags *vtgatepb.VStreamFlags
 }
 
@@ -319,6 +326,10 @@ func (vsm *vstreamManager) GetTotalStreamDelay() int64 {
 
 func (vs *vstream) stream(ctx context.Context) error {
 	ctx, vs.cancel = context.WithCancel(ctx)
+	if vs.streamLivenessTimer == nil {
+		vs.streamLivenessTimer = time.NewTimer(livenessTimeout)
+		defer vs.streamLivenessTimer.Stop()
+	}
 
 	vs.wg.Add(1)
 	go func() {
@@ -401,6 +412,13 @@ func (vs *vstream) sendEvents(ctx context.Context) {
 				})
 				return
 			}
+		case <-vs.streamLivenessTimer.C:
+			msg := fmt.Sprintf("vstream failed liveness checks as there was no activity, including heartbeats, within the last %v", livenessTimeout)
+			log.Infof("Error in vstream: %s", msg)
+			vs.once.Do(func() {
+				vs.setError(vterrors.New(vtrpcpb.Code_UNAVAILABLE, msg), "vstream is fully throttled or otherwise hung")
+			})
+			return
 		}
 	}
 }
@@ -697,6 +715,7 @@ func (vs *vstream) streamFromTablet(ctx context.Context, sgtid *binlogdatapb.Sha
 
 			sendevents := make([]*binlogdatapb.VEvent, 0, len(events))
 			for i, event := range events {
+				vs.streamLivenessTimer.Reset(livenessTimeout) // Any event in the stream demonstrates liveness
 				switch event.Type {
 				case binlogdatapb.VEventType_FIELD:
 					ev := maybeUpdateTableName(event, sgtid.Keyspace, vs.flags.GetExcludeKeyspaceFromTableName(), extractFieldTableName)

go/vt/vtgate/vstream_manager_test.go

@@ -1748,6 +1748,84 @@ func TestVStreamIdleHeartbeat(t *testing.T) {
 	}
 }
 
+func TestVStreamLivenessChecks(t *testing.T) {
+	ctx := utils.LeakCheckContext(t)
+	cell := "aa"
+	ks := "TestVStream"
+	_ = createSandbox(ks)
+	hc := discovery.NewFakeHealthCheck(nil)
+	st := getSandboxTopo(ctx, cell, ks, []string{"-20"})
+	vsm := newTestVStreamManager(ctx, hc, st, cell)
+	origLivenessTimeout := livenessTimeout
+	defer func() {
+		livenessTimeout = origLivenessTimeout
+	}()
+	fakeTablet := hc.AddTestTablet("aa", "1.1.1.1", 1001, ks, "-20", topodatapb.TabletType_PRIMARY, true, 1, nil)
+	addTabletToSandboxTopo(t, ctx, st, fakeTablet.Tablet().Keyspace, fakeTablet.Tablet().Shard, fakeTablet.Tablet())
+	vgtid := &binlogdatapb.VGtid{
+		ShardGtids: []*binlogdatapb.ShardGtid{{
+			Keyspace: fakeTablet.Tablet().Keyspace,
+			Shard:    fakeTablet.Tablet().Shard,
+		}},
+	}
+
+	type testcase struct {
+		name            string
+		livenessTimeout time.Duration
+		// We use simulated tablet vstreamer heartbeats as a substitute for the hardcoded 900ms
+		// heartbeats that come from a real tablet server's vstreamer because we have no real
+		// tablet server here.
+		simulateVstreamerHeartbeats bool
+		wantErr                     string
+	}
+	testcases := []testcase{
+		{
+			name:            "should fail liveness check",
+			livenessTimeout: 100 * time.Millisecond,
+			wantErr:         fmt.Sprintf("vstream is fully throttled or otherwise hung: vstream failed liveness checks as there was no activity, including heartbeats, within the last %v", 100*time.Millisecond),
+		},
+		{
+			name:                        "should not fail liveness check",
+			livenessTimeout:             100 * time.Millisecond,
+			simulateVstreamerHeartbeats: true,
+		},
+	}
+
+	for _, tcase := range testcases {
+		t.Run(tcase.name, func(t *testing.T) {
+			vstreamCtx, vstreamCancel := context.WithTimeout(ctx, tcase.livenessTimeout*2)
+			defer vstreamCancel()
+
+			livenessTimeout = tcase.livenessTimeout
+			if tcase.simulateVstreamerHeartbeats {
+				// We need to ensure that there's a steady stream of vtgate<-vttablet vstream heartbeat
+				// events so that we stay within the defined livenessTimeout and the stream ends when
+				// the vstreamCtx times out.
+				numEvents := tcase.livenessTimeout.Nanoseconds() / 1e5
+				for range numEvents {
+					event := &binlogdatapb.VEvent{Type: binlogdatapb.VEventType_HEARTBEAT}
+					fakeTablet.AddVStreamEvents([]*binlogdatapb.VEvent{event}, nil)
+				}
+				origVStreamEventDelay := fakeTablet.VStreamEventDelay
+				defer func() {
+					fakeTablet.VStreamEventDelay = origVStreamEventDelay
+				}()
+				fakeTablet.VStreamEventDelay = time.Duration((tcase.livenessTimeout.Nanoseconds() / numEvents) * 2)
+			}
+
+			err := vsm.VStream(vstreamCtx, topodatapb.TabletType_PRIMARY, vgtid, nil, &vtgatepb.VStreamFlags{}, func(events []*binlogdatapb.VEvent) error {
+				return nil
+			})
+			if tcase.wantErr == "" {
+				// Then we expect the context ended error, which means that no real errors occurred
+				// in the stream.
+				tcase.wantErr = "context ended while sending events: context deadline exceeded"
+			}
+			require.EqualError(t, err, tcase.wantErr)
+		})
+	}
+}
+
 func TestKeyspaceHasBeenSharded(t *testing.T) {
 	ctx := utils.LeakCheckContext(t)
 

go/vt/vttablet/sandboxconn/sandboxconn.go

@@ -110,10 +110,11 @@ type SandboxConn struct {
 	MessageIDs []*querypb.Value
 
 	// vstream expectations.
-	StartPos      string
-	VStreamEvents [][]*binlogdatapb.VEvent
-	VStreamErrors []error
-	VStreamCh     chan *binlogdatapb.VEvent
+	StartPos          string
+	VStreamEvents     [][]*binlogdatapb.VEvent
+	VStreamErrors     []error
+	VStreamCh         chan *binlogdatapb.VEvent
+	VStreamEventDelay time.Duration // Any sleep that should be introduced before each event is streamed
 
 	// transaction id generator
 	TransactionID atomic.Int64
@@ -625,6 +626,9 @@ func (sbc *SandboxConn) VStream(ctx context.Context, request *binlogdatapb.VStre
 			if ev == nil {
 				return err
 			}
+			if sbc.VStreamEventDelay > 0 {
+				time.Sleep(sbc.VStreamEventDelay)
+			}
 			if err := send(ev); err != nil {
 				return err
 			}

go/vt/vttablet/tabletserver/throttle/client.go

@@ -94,8 +94,9 @@ func (c *Client) ThrottleCheckOK(ctx context.Context, overrideAppName throttlera
 		return emptyCheckResult, true
 	}
 	if c.throttler == nil {
-		// no throttler
-		return emptyCheckResult, true
+		// No throttler, so we only throttle if we're in a test and have set the
+		// always throttle app name.
+		return emptyCheckResult, overrideAppName != throttlerapp.TestingAlwaysThrottledName
 	}
 	checkApp := c.appName
 	if overrideAppName != "" {

go/vt/vttablet/tabletserver/vstreamer/main_test.go

@@ -287,7 +287,13 @@ func expectLog(ctx context.Context, t *testing.T, input any, ch <-chan []*binlog
 	}
 }
 
+func startFullyThrottledStream(ctx context.Context, t *testing.T, filter *binlogdatapb.Filter, position string, tablePKs []*binlogdatapb.TableLastPK) (*sync.WaitGroup, <-chan []*binlogdatapb.VEvent) {
+	return startStreamWithAllOrNothingThrottlingOption(ctx, t, filter, position, tablePKs, true)
+}
 func startStream(ctx context.Context, t *testing.T, filter *binlogdatapb.Filter, position string, tablePKs []*binlogdatapb.TableLastPK) (*sync.WaitGroup, <-chan []*binlogdatapb.VEvent) {
+	return startStreamWithAllOrNothingThrottlingOption(ctx, t, filter, position, tablePKs, false)
+}
+func startStreamWithAllOrNothingThrottlingOption(ctx context.Context, t *testing.T, filter *binlogdatapb.Filter, position string, tablePKs []*binlogdatapb.TableLastPK, alwaysThrottle bool) (*sync.WaitGroup, <-chan []*binlogdatapb.VEvent) {
 	switch position {
 	case "":
 		position = primaryPosition(t)
@@ -302,14 +308,14 @@ func startStream(ctx context.Context, t *testing.T, filter *binlogdatapb.Filter,
 	go func() {
 		defer close(ch)
 		defer wg.Done()
-		if vstream(ctx, t, position, tablePKs, filter, ch) != nil {
+		if vstream(ctx, t, position, tablePKs, filter, ch, alwaysThrottle) != nil {
 			t.Log("vstream returned error")
 		}
 	}()
 	return &wg, ch
 }
 
-func vstream(ctx context.Context, t *testing.T, pos string, tablePKs []*binlogdatapb.TableLastPK, filter *binlogdatapb.Filter, ch chan []*binlogdatapb.VEvent) error {
+func vstream(ctx context.Context, t *testing.T, pos string, tablePKs []*binlogdatapb.TableLastPK, filter *binlogdatapb.Filter, ch chan []*binlogdatapb.VEvent, fullyThrottle bool) error {
 	if filter == nil {
 		filter = &binlogdatapb.Filter{
 			Rules: []*binlogdatapb.Rule{{
@@ -330,7 +336,12 @@ func vstream(ctx context.Context, t *testing.T, pos string, tablePKs []*binlogda
 	utils.SetFlagVariantsForTests(options.ConfigOverrides, "vstream-dynamic-packet-size", dynamicPacketSize)
 	utils.SetFlagVariantsForTests(options.ConfigOverrides, "vstream-packet-size", packetSize)
 
-	return engine.Stream(ctx, pos, tablePKs, filter, throttlerapp.VStreamerName, func(evs []*binlogdatapb.VEvent) error {
+	appName := throttlerapp.VStreamerName
+	if fullyThrottle {
+		appName = throttlerapp.TestingAlwaysThrottledName
+	}
+
+	return engine.Stream(ctx, pos, tablePKs, filter, appName, func(evs []*binlogdatapb.VEvent) error {
 		timer := time.NewTimer(2 * time.Second)
 		defer timer.Stop()
 

go/vt/vttablet/tabletserver/vstreamer/vstreamer.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"google.golang.org/protobuf/encoding/prototext"
@@ -51,17 +52,24 @@ import (
 )
 
 const (
-	trxHistoryLenQuery = `select count as history_len from information_schema.INNODB_METRICS where name = 'trx_rseg_history_len'`
-	replicaLagQuery    = `show replica status`
-	legacyLagQuery     = `show slave status`
-	hostQuery          = `select @@hostname as hostname, @@port as port`
+	trxHistoryLenQuery        = `select count as history_len from information_schema.INNODB_METRICS where name = 'trx_rseg_history_len'`
+	replicaLagQuery           = `show replica status`
+	legacyLagQuery            = `show slave status`
+	hostQuery                 = `select @@hostname as hostname, @@port as port`
+	fullyThrottledMetricLabel = "FullyThrottledTimeout"
 )
 
 // HeartbeatTime is set to slightly below 1s, compared to idleTimeout
 // set by VPlayer at slightly above 1s. This minimizes conflicts
 // between the two timeouts.
 var HeartbeatTime = 900 * time.Millisecond
 
+// How long we can be fully throttled before returning an error.
+// If we hit this then we can surface a metric for operators and we can run the tablet picker again
+// to try and pick another tablet which is perhaps less burdened. Running the tablet picker also gives
+// us a natural backoff period.
+var fullyThrottledTimeout = 10 * time.Minute
+
 // vstreamer is for serving a single vreplication stream on the source side.
 type vstreamer struct {
 	ctx    context.Context
@@ -330,7 +338,10 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 	if vs.filter != nil && vs.filter.WorkflowName != "" {
 		wfNameLog = " in workflow " + vs.filter.WorkflowName
 	}
+	throttlerErrs := make(chan error, 1) // How we share the error when we've been fully throttled too long
+	defer close(throttlerErrs)
 	throttleEvents := func(throttledEvents chan mysql.BinlogEvent) {
+		throttledTime := atomic.Int64{}
 		for {
 			// Check throttler.
 			if checkResult, ok := vs.vse.throttlerClient.ThrottleCheckOKOrWaitAppName(ctx, vs.throttlerApp); !ok {
@@ -341,9 +352,17 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 				default:
 					// Do nothing special.
 				}
+				curtime := time.Now().Unix()
+				if !throttledTime.CompareAndSwap(0, curtime) {
+					if curtime-throttledTime.Load() > int64(fullyThrottledTimeout.Seconds()) {
+						throttlerErrs <- vterrors.Errorf(vtrpcpb.Code_INTERNAL, "vstreamer has been fully throttled for more than %v, giving up so that we can retry", fullyThrottledTimeout)
+						return
+					}
+				}
 				logger.Infof("vstreamer throttled%s: %s.", wfNameLog, checkResult.Summary())
 				continue
 			}
+			throttledTime.Store(0) // We are no longer fully throttled
 			select {
 			case ev, ok := <-events:
 				if ok {
@@ -414,6 +433,9 @@ func (vs *vstreamer) parseEvents(ctx context.Context, events <-chan mysql.Binlog
 			}
 		case err := <-errs:
 			return err
+		case throttlerErr := <-throttlerErrs:
+			vs.vse.errorCounts.Add(fullyThrottledMetricLabel, 1)
+			return throttlerErr
 		case <-ctx.Done():
 			return nil
 		case <-hbTimer.C:

go/vt/vttablet/tabletserver/vstreamer/vstreamer_test.go

@@ -530,7 +530,7 @@ func TestVStreamMissingFieldsInLastPK(t *testing.T) {
 	}
 	ctx := context.Background()
 	ch := make(chan []*binlogdatapb.VEvent)
-	err := vstream(ctx, t, "", tablePKs, filter, ch)
+	err := vstream(ctx, t, "", tablePKs, filter, ch, false)
 	require.ErrorContains(t, err, "lastpk for table t1 has no fields defined")
 }
 
@@ -1483,7 +1483,7 @@ func TestDDLDropColumn(t *testing.T) {
 		}
 	}()
 	defer close(ch)
-	err := vstream(ctx, t, pos, nil, nil, ch)
+	err := vstream(ctx, t, pos, nil, nil, ch, false)
 	want := "cannot determine table columns"
 	if err == nil || !strings.Contains(err.Error(), want) {
 		t.Errorf("err: %v, must contain %s", err, want)
@@ -1962,6 +1962,39 @@ func TestHeartbeat(t *testing.T) {
 	cancel()
 }
 
+func TestFullyThrottledTimeout(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	origTimeout := fullyThrottledTimeout
+	origHeartbeatTime := HeartbeatTime
+	startingMetric := engine.errorCounts.Counts()[fullyThrottledMetricLabel]
+	defer func() {
+		fullyThrottledTimeout = origTimeout
+		HeartbeatTime = origHeartbeatTime
+	}()
+
+	fullyThrottledTimeout = 100 * time.Millisecond
+	HeartbeatTime = fullyThrottledTimeout * 15
+	waitTimer := time.NewTimer(HeartbeatTime)
+	defer waitTimer.Stop()
+	done := make(chan struct{})
+	go func() {
+		wg, evs := startFullyThrottledStream(ctx, t, nil, "", nil) // Fully throttled
+		wg.Wait()
+		require.Zero(t, len(evs))
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		endingMetric := engine.errorCounts.Counts()[fullyThrottledMetricLabel]
+		require.Equal(t, startingMetric+1, endingMetric)
+		return
+	case <-waitTimer.C:
+		require.FailNow(t, "fully throttled stall handler did not fire as expected")
+	}
+}
+
 func TestNoFutureGTID(t *testing.T) {
 	// Execute something to make sure we have ranges in GTIDs.
 	execStatements(t, []string{
@@ -1990,7 +2023,7 @@ func TestNoFutureGTID(t *testing.T) {
 		}
 	}()
 	defer close(ch)
-	err = vstream(ctx, t, future, nil, nil, ch)
+	err = vstream(ctx, t, future, nil, nil, ch, false)
 	want := "GTIDSet Mismatch"
 	if err == nil || !strings.Contains(err.Error(), want) {
 		t.Errorf("err: %v, must contain %s", err, want)