Slack is noticing a lot of "stream lagged behind during consolidation"

errors because the "streamBufferSize" is hard-coded at only 8.

This PR adds vttablet cli arg --consolidator-stream-buffer-size
which allows increasing beyond the default (8, for parity)
if the site prefers trading some more memory so large results complete.

Author: nharkins

go/flags/endtoend/vtcombo.txt

@@ -48,6 +48,7 @@ Flags:
       --config-persistence-min-interval duration                         minimum interval between persisting dynamic config changes back to disk (if no change has occurred, nothing is done). (default 1s)
       --config-type string                                               Config file type (omit to infer config type from file extension).
       --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
+      --consolidator-stream-buffer-size int                              Configure the stream consolidator buffer size for follower streams. A larger buffer reduces the chance of a follower being dropped during consolidation at the cost of increased memory usage. (default 8)
       --consolidator-stream-query-size int                               Configure the stream consolidator query size in bytes. Setting to 0 disables the stream consolidator. (default 2097152)
       --consolidator-stream-total-size int                               Configure the stream consolidator total size in bytes. Setting to 0 disables the stream consolidator. (default 134217728)
       --consul-auth-static-file string                                   JSON File to read the topos/tokens from.

go/flags/endtoend/vttablet.txt

@@ -81,6 +81,7 @@ Flags:
       --config-persistence-min-interval duration                         minimum interval between persisting dynamic config changes back to disk (if no change has occurred, nothing is done). (default 1s)
       --config-type string                                               Config file type (omit to infer config type from file extension).
       --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
+      --consolidator-stream-buffer-size int                              Configure the stream consolidator buffer size for follower streams. A larger buffer reduces the chance of a follower being dropped during consolidation at the cost of increased memory usage. (default 8)
       --consolidator-stream-query-size int                               Configure the stream consolidator query size in bytes. Setting to 0 disables the stream consolidator. (default 2097152)
       --consolidator-stream-total-size int                               Configure the stream consolidator total size in bytes. Setting to 0 disables the stream consolidator. (default 134217728)
       --consul-auth-static-file string                                   JSON File to read the topos/tokens from.

go/vt/vttablet/tabletserver/query_engine.go

@@ -238,7 +238,7 @@ func NewQueryEngine(env tabletenv.Env, se *schema.Engine) *QueryEngine {
 	qe.consolidator = sync2.NewConsolidator()
 	if config.ConsolidatorStreamTotalSize > 0 && config.ConsolidatorStreamQuerySize > 0 {
 		log.Info(fmt.Sprintf("Stream consolidator is enabled with query size set to %d and total size set to %d.", config.ConsolidatorStreamQuerySize, config.ConsolidatorStreamTotalSize))
-		qe.streamConsolidator = NewStreamConsolidator(config.ConsolidatorStreamTotalSize, config.ConsolidatorStreamQuerySize, returnStreamResult)
+		qe.streamConsolidator = NewStreamConsolidator(config.ConsolidatorStreamTotalSize, config.ConsolidatorStreamQuerySize, config.ConsolidatorStreamBufferSize, returnStreamResult)
 	} else {
 		log.Info("Stream consolidator is not enabled.")
 	}

go/vt/vttablet/tabletserver/stream_consolidator.go

@@ -28,29 +28,30 @@ import (
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 )
 
-const streamBufferSize = 8
-
 // StreamConsolidator is a data structure capable of merging several identical streaming queries so only
 // one query is executed in MySQL and its response is fanned out to all the clients simultaneously.
 type StreamConsolidator struct {
 	mu                             sync.Mutex
 	inflight                       map[string]*streamInFlight
 	memory                         int64
 	maxMemoryTotal, maxMemoryQuery int64
+	streamBufferSize               int
 	blocking                       bool
 	cleanup                        StreamCallback
 }
 
 // NewStreamConsolidator allocates a stream consolidator. The consolidator will use up to maxMemoryTotal
 // bytes in order to allow simultaneous queries to "catch up" to each other. Each individual stream will
 // only use up to maxMemoryQuery bytes of memory as a history buffer to catch up.
-func NewStreamConsolidator(maxMemoryTotal, maxMemoryQuery int64, cleanup StreamCallback) *StreamConsolidator {
+// streamBufferSize controls the channel buffer depth for each follower stream.
+func NewStreamConsolidator(maxMemoryTotal, maxMemoryQuery int64, streamBufferSize int, cleanup StreamCallback) *StreamConsolidator {
 	return &StreamConsolidator{
-		inflight:       make(map[string]*streamInFlight),
-		maxMemoryTotal: maxMemoryTotal,
-		maxMemoryQuery: maxMemoryQuery,
-		blocking:       false,
-		cleanup:        cleanup,
+		inflight:         make(map[string]*streamInFlight),
+		maxMemoryTotal:   maxMemoryTotal,
+		maxMemoryQuery:   maxMemoryQuery,
+		streamBufferSize: streamBufferSize,
+		blocking:         false,
+		cleanup:          cleanup,
 	}
 }
 
@@ -87,7 +88,7 @@ func (sc *StreamConsolidator) Consolidate(waitTimings *servenv.TimingsWrapper, l
 
 	// if there's an existing stream for our query, try to follow it
 	if inflight != nil {
-		catchup, followChan = inflight.follow()
+		catchup, followChan = inflight.follow(sc.streamBufferSize)
 	}
 
 	// if there isn't an existing stream; OR if there is an existing stream but
@@ -194,7 +195,7 @@ type streamInFlight struct {
 // that will receive all the Results in the future.
 // If this stream has been running for too long and we cannot catch up to it, follow
 // returns a nil channel.
-func (s *streamInFlight) follow() ([]*sqltypes.Result, chan *sqltypes.Result) {
+func (s *streamInFlight) follow(streamBufferSize int) ([]*sqltypes.Result, chan *sqltypes.Result) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 

go/vt/vttablet/tabletserver/stream_consolidator_flaky_test.go

@@ -148,7 +148,7 @@ func (ct *consolidationTest) run(workers int, generateCallback func(int) (string
 
 func TestConsolidatorSimple(t *testing.T) {
 	ct := consolidationTest{
-		cc:              NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+		cc:              NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 		streamItemDelay: 10 * time.Millisecond,
 		streamItemCount: 10,
 	}
@@ -175,7 +175,7 @@ func TestConsolidatorSimple(t *testing.T) {
 func TestConsolidatorErrorPropagation(t *testing.T) {
 	t.Run("from mysql", func(t *testing.T) {
 		ct := consolidationTest{
-			cc: NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+			cc: NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 			leaderCallback: func(callback StreamCallback) error {
 				time.Sleep(100 * time.Millisecond)
 				return errors.New("mysqld error")
@@ -195,7 +195,7 @@ func TestConsolidatorErrorPropagation(t *testing.T) {
 
 	t.Run("from leader", func(t *testing.T) {
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 			streamItemDelay: 10 * time.Millisecond,
 			streamItemCount: 10,
 		}
@@ -229,7 +229,7 @@ func TestConsolidatorErrorPropagation(t *testing.T) {
 
 	t.Run("from followers", func(t *testing.T) {
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 			streamItemDelay: 10 * time.Millisecond,
 			streamItemCount: 10,
 		}
@@ -263,7 +263,7 @@ func TestConsolidatorErrorPropagation(t *testing.T) {
 
 func TestConsolidatorDelayedListener(t *testing.T) {
 	ct := consolidationTest{
-		cc:              NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+		cc:              NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 		streamItemDelay: 1 * time.Millisecond,
 		streamItemCount: 100,
 	}
@@ -304,7 +304,7 @@ func TestConsolidatorDelayedListener(t *testing.T) {
 func TestConsolidatorMemoryLimits(t *testing.T) {
 	t.Run("rows too large", func(t *testing.T) {
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, 32, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, 32, 8, nocleanup),
 			streamItemDelay: 1 * time.Millisecond,
 			streamItemCount: 100,
 		}
@@ -326,7 +326,7 @@ func TestConsolidatorMemoryLimits(t *testing.T) {
 
 	t.Run("two-phase consolidation (time)", func(t *testing.T) {
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, 2*1024, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, 2*1024, 8, nocleanup),
 			streamItemDelay: 2 * time.Millisecond,
 			streamItemCount: 10,
 		}
@@ -354,7 +354,7 @@ func TestConsolidatorMemoryLimits(t *testing.T) {
 		rsize := results[0].CachedSize(true)
 
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, rsize*streamsInFirstBatch+1, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, rsize*streamsInFirstBatch+1, 8, nocleanup),
 			streamItemDelay: 1 * time.Millisecond,
 			streamItems:     results,
 		}
@@ -379,7 +379,7 @@ func TestConsolidatorMemoryLimits(t *testing.T) {
 		rsize := results[0].CachedSize(true)
 
 		ct := consolidationTest{
-			cc:              NewStreamConsolidator(128*1024, rsize*2+1, nocleanup),
+			cc:              NewStreamConsolidator(128*1024, rsize*2+1, 8, nocleanup),
 			streamItemDelay: 10 * time.Millisecond,
 			streamItems:     results,
 		}

go/vt/vttablet/tabletserver/tabletenv/config.go

@@ -204,6 +204,7 @@ func registerTabletEnvFlags(fs *pflag.FlagSet) {
 	utils.SetFlagBoolVar(fs, &enableConsolidatorReplicas, "enable-consolidator-replicas", false, "This option enables the query consolidator only on replicas.")
 	fs.Int64Var(&currentConfig.ConsolidatorStreamQuerySize, "consolidator-stream-query-size", defaultConfig.ConsolidatorStreamQuerySize, "Configure the stream consolidator query size in bytes. Setting to 0 disables the stream consolidator.")
 	fs.Int64Var(&currentConfig.ConsolidatorStreamTotalSize, "consolidator-stream-total-size", defaultConfig.ConsolidatorStreamTotalSize, "Configure the stream consolidator total size in bytes. Setting to 0 disables the stream consolidator.")
+	fs.IntVar(&currentConfig.ConsolidatorStreamBufferSize, "consolidator-stream-buffer-size", defaultConfig.ConsolidatorStreamBufferSize, "Configure the stream consolidator buffer size for follower streams. A larger buffer reduces the chance of a follower being dropped during consolidation at the cost of increased memory usage.")
 
 	fs.Int64Var(&currentConfig.ConsolidatorQueryWaiterCap, "consolidator-query-waiter-cap", 0, "Configure the maximum number of clients allowed to wait on the consolidator.")
 	utils.SetFlagDurationVar(fs, &healthCheckInterval, "health-check-interval", defaultConfig.Healthcheck.Interval, "Interval between health checks")
@@ -335,24 +336,25 @@ type TabletConfig struct {
 	ReplicationTracker ReplicationTrackerConfig `json:"replicationTracker"`
 
 	// Consolidator can be enable, disable, or notOnPrimary. Default is enable.
-	Consolidator                string        `json:"consolidator,omitempty"`
-	PassthroughDML              bool          `json:"passthroughDML,omitempty"`
-	StreamBufferSize            int           `json:"streamBufferSize,omitempty"`
-	ConsolidatorStreamTotalSize int64         `json:"consolidatorStreamTotalSize,omitempty"`
-	ConsolidatorStreamQuerySize int64         `json:"consolidatorStreamQuerySize,omitempty"`
-	ConsolidatorQueryWaiterCap  int64         `json:"consolidatorMaxQueryWait,omitempty"`
-	QueryCacheMemory            int64         `json:"queryCacheMemory,omitempty"`
-	QueryCacheDoorkeeper        bool          `json:"queryCacheDoorkeeper,omitempty"`
-	SchemaReloadInterval        time.Duration `json:"schemaReloadIntervalSeconds,omitempty"`
-	SchemaChangeReloadTimeout   time.Duration `json:"schemaChangeReloadTimeout,omitempty"`
-	WatchReplication            bool          `json:"watchReplication,omitempty"` // Ignored and unused, remove in v25
-	TrackSchemaVersions         bool          `json:"trackSchemaVersions,omitempty"`
-	SchemaVersionMaxAgeSeconds  int64         `json:"schemaVersionMaxAgeSeconds,omitempty"`
-	TerseErrors                 bool          `json:"terseErrors,omitempty"`
-	TruncateErrorLen            int           `json:"truncateErrorLen,omitempty"`
-	AnnotateQueries             bool          `json:"annotateQueries,omitempty"`
-	MessagePostponeParallelism  int           `json:"messagePostponeParallelism,omitempty"`
-	SignalWhenSchemaChange      bool          `json:"signalWhenSchemaChange,omitempty"`
+	Consolidator                 string        `json:"consolidator,omitempty"`
+	PassthroughDML               bool          `json:"passthroughDML,omitempty"`
+	StreamBufferSize             int           `json:"streamBufferSize,omitempty"`
+	ConsolidatorStreamTotalSize  int64         `json:"consolidatorStreamTotalSize,omitempty"`
+	ConsolidatorStreamQuerySize  int64         `json:"consolidatorStreamQuerySize,omitempty"`
+	ConsolidatorStreamBufferSize int           `json:"consolidatorStreamBufferSize,omitempty"`
+	ConsolidatorQueryWaiterCap   int64         `json:"consolidatorMaxQueryWait,omitempty"`
+	QueryCacheMemory             int64         `json:"queryCacheMemory,omitempty"`
+	QueryCacheDoorkeeper         bool          `json:"queryCacheDoorkeeper,omitempty"`
+	SchemaReloadInterval         time.Duration `json:"schemaReloadIntervalSeconds,omitempty"`
+	SchemaChangeReloadTimeout    time.Duration `json:"schemaChangeReloadTimeout,omitempty"`
+	WatchReplication             bool          `json:"watchReplication,omitempty"` // Ignored and unused, remove in v25
+	TrackSchemaVersions          bool          `json:"trackSchemaVersions,omitempty"`
+	SchemaVersionMaxAgeSeconds   int64         `json:"schemaVersionMaxAgeSeconds,omitempty"`
+	TerseErrors                  bool          `json:"terseErrors,omitempty"`
+	TruncateErrorLen             int           `json:"truncateErrorLen,omitempty"`
+	AnnotateQueries              bool          `json:"annotateQueries,omitempty"`
+	MessagePostponeParallelism   int           `json:"messagePostponeParallelism,omitempty"`
+	SignalWhenSchemaChange       bool          `json:"signalWhenSchemaChange,omitempty"`
 
 	ExternalConnections map[string]*dbconfigs.DBConfigs `json:"externalConnections,omitempty"`
 
@@ -1093,9 +1095,10 @@ var defaultConfig = TabletConfig{
 		// of them ready in MySQL and profit from a pipelining effect.
 		MaxConcurrency: 5,
 	},
-	Consolidator:                Enable,
-	ConsolidatorStreamTotalSize: 128 * 1024 * 1024,
-	ConsolidatorStreamQuerySize: 2 * 1024 * 1024,
+	Consolidator:                 Enable,
+	ConsolidatorStreamTotalSize:  128 * 1024 * 1024,
+	ConsolidatorStreamQuerySize:  2 * 1024 * 1024,
+	ConsolidatorStreamBufferSize: 8,
 	// The value for StreamBufferSize was chosen after trying out a few of
 	// them. Too small buffers force too many packets to be sent. Too big
 	// buffers force the clients to read them in multiple chunks and make

go/vt/vttablet/tabletserver/tabletenv/config_test.go

@@ -132,6 +132,7 @@ func TestDefaultConfig(t *testing.T) {
 	gotBytes, err := yaml2.Marshal(NewDefaultConfig())
 	require.NoError(t, err)
 	want := `consolidator: enable
+consolidatorStreamBufferSize: 8
 consolidatorStreamQuerySize: 2097152
 consolidatorStreamTotalSize: 134217728
 gracePeriods: