Merge pull request #6815 from planetscale/vttablet-throttle-feature-flag

vttablet throttler feature flag: -enable-lag-throttler

docker/mini/vttablet-mini-up.sh

@@ -50,8 +50,6 @@ vttablet \
  -init_shard $shard \
  -init_tablet_type $tablet_type \
  -health_check_interval 5s \
- -heartbeat_enable \
- -heartbeat_interval 250ms \
  -enable_semi_sync \
  -enable_replication_reporter \
  -backup_storage_implementation file \

examples/local/scripts/vttablet-up.sh

@@ -47,8 +47,6 @@ vttablet \
  -health_check_interval 5s \
  -enable_semi_sync \
  -enable_replication_reporter \
- -heartbeat_enable \
- -heartbeat_interval 250ms \
  -backup_storage_implementation file \
  -file_backup_storage_root $VTDATAROOT/backups \
  -restore_from_backup \

go/test/endtoend/tabletmanager/throttler/throttler_test.go

@@ -88,6 +88,7 @@ func TestMain(m *testing.M) {
 			"-lock_tables_timeout", "5s",
 			"-watch_replication_stream",
 			"-enable_replication_reporter",
+			"-enable-lag-throttler",
 			"-heartbeat_enable",
 			"-heartbeat_interval", "250ms",
 		}

go/vt/vttablet/tabletserver/repltracker/repltracker.go

@@ -50,7 +50,8 @@ var (
 
 // ReplTracker tracks replication lag.
 type ReplTracker struct {
-	mode string
+	mode           string
+	forceHeartbeat bool
 
 	mu       sync.Mutex
 	isMaster bool
@@ -63,10 +64,11 @@ type ReplTracker struct {
 // NewReplTracker creates a new ReplTracker.
 func NewReplTracker(env tabletenv.Env, alias topodatapb.TabletAlias) *ReplTracker {
 	return &ReplTracker{
-		mode:   env.Config().ReplicationTracker.Mode,
-		hw:     newHeartbeatWriter(env, alias),
-		hr:     newHeartbeatReader(env),
-		poller: &poller{},
+		mode:           env.Config().ReplicationTracker.Mode,
+		forceHeartbeat: env.Config().EnableLagThrottler,
+		hw:             newHeartbeatWriter(env, alias),
+		hr:             newHeartbeatReader(env),
+		poller:         &poller{},
 	}
 }
 
@@ -88,6 +90,9 @@ func (rt *ReplTracker) MakeMaster() {
 		rt.hr.Close()
 		rt.hw.Open()
 	}
+	if rt.forceHeartbeat {
+		rt.hw.Open()
+	}
 }
 
 // MakeNonMaster must be called if the tablet type becomes non-MASTER.
@@ -105,6 +110,9 @@ func (rt *ReplTracker) MakeNonMaster() {
 		// Run the status once to pre-initialize values.
 		rt.poller.Status()
 	}
+	if rt.forceHeartbeat {
+		rt.hw.Close()
+	}
 }
 
 // Close closes ReplTracker.

go/vt/vttablet/tabletserver/repltracker/repltracker_test.go

@@ -82,7 +82,7 @@ func TestReplTracker(t *testing.T) {
 	rt.InitDBConfig(target, mysqld)
 	assert.Equal(t, tabletenv.Polling, rt.mode)
 	assert.Equal(t, mysqld, rt.poller.mysqld)
-	assert.True(t, rt.hw.enabled)
+	assert.False(t, rt.hw.enabled)
 	assert.False(t, rt.hr.enabled)
 
 	rt.MakeNonMaster()

go/vt/vttablet/tabletserver/repltracker/writer.go

@@ -73,6 +73,11 @@ type heartbeatWriter struct {
 // newHeartbeatWriter creates a new heartbeatWriter.
 func newHeartbeatWriter(env tabletenv.Env, alias topodatapb.TabletAlias) *heartbeatWriter {
 	config := env.Config()
+
+	// config.EnableLagThrottler is a feature flag for the throttler; if throttler runs, then heartbeat must also run
+	if config.ReplicationTracker.Mode != tabletenv.Heartbeat && !config.EnableLagThrottler {
+		return &heartbeatWriter{}
+	}
 	heartbeatInterval := config.ReplicationTracker.HeartbeatIntervalSeconds.Get()
 	return &heartbeatWriter{
 		env:         env,
@@ -182,6 +187,9 @@ func (w *heartbeatWriter) recordError(err error) {
 
 // enableWrites actives or deactives heartbeat writes
 func (w *heartbeatWriter) enableWrites(enable bool) {
+	if w.ticks == nil {
+		return
+	}
 	if enable {
 		w.ticks.Start(w.writeHeartbeat)
 	} else {

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

@@ -142,6 +142,7 @@ func init() {
 
 	flag.BoolVar(&enableHeartbeat, "heartbeat_enable", false, "If true, vttablet records (if master) or checks (if replica) the current time of a replication heartbeat in the table _vt.heartbeat. The result is used to inform the serving state of the vttablet via healthchecks.")
 	flag.DurationVar(&heartbeatInterval, "heartbeat_interval", 1*time.Second, "How frequently to read and write replication heartbeat.")
+	flag.BoolVar(&currentConfig.EnableLagThrottler, "enable-lag-throttler", defaultConfig.EnableLagThrottler, "If true, vttablet will run a throttler service, and will implicitly enable heartbeats")
 
 	flag.BoolVar(&currentConfig.EnforceStrictTransTables, "enforce_strict_trans_tables", defaultConfig.EnforceStrictTransTables, "If true, vttablet requires MySQL to run with STRICT_TRANS_TABLES or STRICT_ALL_TABLES on. It is recommended to not turn this flag off. Otherwise MySQL may alter your supplied values before saving them to the database.")
 	flag.BoolVar(&enableConsolidator, "enable-consolidator", true, "This option enables the query consolidator.")
@@ -261,6 +262,8 @@ type TabletConfig struct {
 	TxThrottlerConfig           string   `json:"-"`
 	TxThrottlerHealthCheckCells []string `json:"-"`
 
+	EnableLagThrottler bool `json:"-"`
+
 	TransactionLimitConfig `json:"-"`
 
 	EnforceStrictTransTables bool `json:"-"`
@@ -452,6 +455,8 @@ var defaultConfig = TabletConfig{
 	TxThrottlerConfig:           defaultTxThrottlerConfig(),
 	TxThrottlerHealthCheckCells: []string{},
 
+	EnableLagThrottler: false, // Feature flag; to switch to 'true' at some stage in the future
+
 	TransactionLimitConfig: defaultTransactionLimitConfig(),
 
 	EnforceStrictTransTables: true,

go/vt/vttablet/tabletserver/throttle/check_result.go

@@ -42,3 +42,5 @@ func NewErrorCheckResult(statusCode int, err error) *CheckResult {
 
 // NoSuchMetricCheckResult is a result returns when a metric is unknown
 var NoSuchMetricCheckResult = NewErrorCheckResult(http.StatusNotFound, base.ErrNoSuchMetric)
+
+var okMetricCheckResult = NewCheckResult(http.StatusOK, 0, 0, nil)

go/vt/vttablet/tabletserver/throttle/throttler.go

@@ -186,7 +186,9 @@ func (throttler *Throttler) initThrottleTabletTypes() {
 func (throttler *Throttler) InitDBConfig(keyspace, shard string) {
 	throttler.keyspace = keyspace
 	throttler.shard = shard
-	go throttler.Operate(context.Background())
+	if throttler.env.Config().EnableLagThrottler {
+		go throttler.Operate(context.Background())
+	}
 }
 
 // initThrottler initializes config
@@ -692,6 +694,9 @@ func (throttler *Throttler) AppRequestMetricResult(ctx context.Context, appName
 
 // Check is the main serving function of the throttler, and returns a check result for this cluster's lag
 func (throttler *Throttler) Check(ctx context.Context, appName string, remoteAddr string, flags *CheckFlags) (checkResult *CheckResult) {
+	if !throttler.env.Config().EnableLagThrottler {
+		return okMetricCheckResult
+	}
 	return throttler.check.Check(ctx, appName, "mysql", localStoreName, remoteAddr, flags)
 }