vttablet throttling

docker/mini/vttablet-mini-up.sh

@@ -51,7 +51,7 @@ vttablet \
  -init_tablet_type $tablet_type \
  -health_check_interval 5s \
  -heartbeat_enable \
- -heartbeat_interval 500ms \
+ -heartbeat_interval 250ms \
  -enable_semi_sync \
  -enable_replication_reporter \
  -backup_storage_implementation file \

examples/compose/vttablet-up.sh

@@ -164,6 +164,8 @@ exec $VTROOT/bin/vttablet \
  -health_check_interval 5s \
  -enable_semi_sync \
  -enable_replication_reporter \
+ -heartbeat_enable \
+ -heartbeat_interval 250ms \
  -port $web_port \
  -grpc_port $grpc_port \
  -binlog_use_v3_resharding_mode=true \

examples/local/scripts/vttablet-up.sh

@@ -47,6 +47,8 @@ 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

@@ -0,0 +1,197 @@
+/*
+Copyright 2020 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package master
+
+import (
+ "flag"
+ "fmt"
+ "net/http"
+ "os"
+ "testing"
+ "time"
+
+ "vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/base"
+
+ "vitess.io/vitess/go/test/endtoend/cluster"
+
+ "github.com/stretchr/testify/assert"
+)
+
+var (
+ clusterInstance *cluster.LocalProcessCluster
+ masterTablet cluster.Vttablet
+ replicaTablet cluster.Vttablet
+ hostname = "localhost"
+ keyspaceName = "ks"
+ cell = "zone1"
+ sqlSchema = `
+ create table t1(
+ id bigint,
+ value varchar(16),
+ primary key(id)
+ ) Engine=InnoDB;
+`
+
+ vSchema = `
+ {
+ "sharded": true,
+ "vindexes": {
+ "hash": {
+ "type": "hash"
+ }
+ },
+ "tables": {
+ "t1": {
+ "column_vindexes": [
+ {
+ "column": "id",
+ "name": "hash"
+ }
+ ]
+ }
+ }
+ }`
+
+ httpClient = base.SetupHTTPClient(time.Second)
+ checkAPIPath = "throttler/check"
+)
+
+func TestMain(m *testing.M) {
+ defer cluster.PanicHandler(nil)
+ flag.Parse()
+
+ exitCode := func() int {
+ clusterInstance = cluster.NewCluster(cell, hostname)
+ defer clusterInstance.Teardown()
+
+ // Start topo server
+ err := clusterInstance.StartTopo()
+ if err != nil {
+ return 1
+ }
+
+ // Set extra tablet args for lock timeout
+ clusterInstance.VtTabletExtraArgs = []string{
+ "-lock_tables_timeout", "5s",
+ "-watch_replication_stream",
+ "-enable_replication_reporter",
+ "-heartbeat_enable",
+ "-heartbeat_interval", "250ms",
+ }
+ // We do not need semiSync for this test case.
+ clusterInstance.EnableSemiSync = false
+
+ // Start keyspace
+ keyspace := &cluster.Keyspace{
+ Name: keyspaceName,
+ SchemaSQL: sqlSchema,
+ VSchema: vSchema,
+ }
+
+ if err = clusterInstance.StartUnshardedKeyspace(*keyspace, 1, false); err != nil {
+ return 1
+ }
+
+ // Collect table paths and ports
+ tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
+ for _, tablet := range tablets {
+ if tablet.Type == "master" {
+ masterTablet = *tablet
+ } else if tablet.Type != "rdonly" {
+ replicaTablet = *tablet
+ }
+ }
+
+ return m.Run()
+ }()
+ os.Exit(exitCode)
+}
+
+func throttleCheck() (*http.Response, error) {
+ return httpClient.Head(fmt.Sprintf("http://localhost:%d/%s", masterTablet.HTTPPort, checkAPIPath))
+}
+
+func TestThrottlerBeforeMetricsCollected(t *testing.T) {
+ defer cluster.PanicHandler(t)
+
+ // Immediately after startup, we expect this response:
+ // {"StatusCode":404,"Value":0,"Threshold":0,"Message":"No such metric"}
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+}
+
+func TestThrottlerAfterMetricsCollected(t *testing.T) {
+ defer cluster.PanicHandler(t)
+
+ time.Sleep(10 * time.Second)
+ // By this time metrics will have been collected. We expect no lag, and something like:
+ // {"StatusCode":200,"Value":0.282278,"Threshold":1,"Message":""}
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusOK, resp.StatusCode)
+}
+
+func TestLag(t *testing.T) {
+ defer cluster.PanicHandler(t)
+
+ {
+ err := clusterInstance.VtctlclientProcess.ExecuteCommand("StopReplication", replicaTablet.Alias)
+ assert.NoError(t, err)
+
+ time.Sleep(2 * time.Second)
+ // Lag will have accumulated
+ // {"StatusCode":429,"Value":4.864921,"Threshold":1,"Message":"Threshold exceeded"}
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode)
+ }
+ {
+ err := clusterInstance.VtctlclientProcess.ExecuteCommand("StartReplication", replicaTablet.Alias)
+ assert.NoError(t, err)
+
+ time.Sleep(5 * time.Second)
+ // Restore
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusOK, resp.StatusCode)
+ }
+}
+
+func TestNoReplicas(t *testing.T) {
+ defer cluster.PanicHandler(t)
+ {
+ err := clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", replicaTablet.Alias, "RDONLY")
+ assert.NoError(t, err)
+
+ time.Sleep(10 * time.Second)
+ // This makes no REPLICA servers available. We expect something like:
+ // {"StatusCode":200,"Value":0,"Threshold":1,"Message":""}
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusOK, resp.StatusCode)
+ }
+ {
+ err := clusterInstance.VtctlclientProcess.ExecuteCommand("ChangeTabletType", replicaTablet.Alias, "REPLICA")
+ assert.NoError(t, err)
+
+ time.Sleep(10 * time.Second)
+ // Restore valid replica
+ resp, err := throttleCheck()
+ assert.NoError(t, err)
+ assert.Equal(t, http.StatusOK, resp.StatusCode)
+ }
+}

go/vt/throttler/throttler_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2019 The Vitess Authors.
+Copyright 2020 The Vitess Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

go/vt/vttablet/endtoend/vstreamer_test.go

@@ -58,6 +58,8 @@ func TestSchemaVersioning(t *testing.T) {
  tsv := framework.Server
  tsv.EnableHistorian(false)
  tsv.SetTracking(false)
+ tsv.EnableHeartbeat(false)
+ defer tsv.EnableHeartbeat(true)
  defer tsv.EnableHistorian(true)
  defer tsv.SetTracking(true)
 

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

@@ -128,3 +128,9 @@ func (rt *ReplTracker) Status() (time.Duration, error) {
  // rt.mode == tabletenv.Poller
  return rt.poller.Status()
 }
+
+// EnableHeartbeat enables or disables writes of heartbeat. This functionality
+// is only used by tests.
+func (rt *ReplTracker) EnableHeartbeat(enable bool) {
+ rt.hw.enableWrites(enable)
+}

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.False(t, rt.hw.enabled)
+ assert.True(t, rt.hw.enabled)
  assert.False(t, rt.hr.enabled)
 
  rt.MakeNonMaster()

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

@@ -73,9 +73,6 @@ type heartbeatWriter struct {
 // newHeartbeatWriter creates a new heartbeatWriter.
 func newHeartbeatWriter(env tabletenv.Env, alias topodatapb.TabletAlias) *heartbeatWriter {
  config := env.Config()
- if config.ReplicationTracker.Mode != tabletenv.Heartbeat {
- return &heartbeatWriter{}
- }
  heartbeatInterval := config.ReplicationTracker.HeartbeatIntervalSeconds.Get()
  return &heartbeatWriter{
  env: env,
@@ -111,7 +108,7 @@ func (w *heartbeatWriter) Open() {
  log.Info("Hearbeat Writer: opening")
 
  w.pool.Open(w.env.Config().DB.AppWithDB(), w.env.Config().DB.DbaWithDB(), w.env.Config().DB.AppDebugWithDB())
- w.ticks.Start(w.writeHeartbeat)
+ w.enableWrites(true)
  w.isOpen = true
 }
 
@@ -126,7 +123,7 @@ func (w *heartbeatWriter) Close() {
  return
  }
 
- w.ticks.Stop()
+ w.enableWrites(false)
  w.pool.Close()
  w.isOpen = false
  log.Info("Hearbeat Writer: closed")
@@ -182,3 +179,12 @@ func (w *heartbeatWriter) recordError(err error) {
  w.errorLog.Errorf("%v", err)
  writeErrors.Add(1)
 }
+
+// enableWrites actives or deactives heartbeat writes
+func (w *heartbeatWriter) enableWrites(enable bool) {
+ if enable {
+ w.ticks.Start(w.writeHeartbeat)
+ } else {
+ w.ticks.Stop()
+ }
+}

go/vt/vttablet/tabletserver/state_manager.go

@@ -105,6 +105,7 @@ type stateManager struct {
  txThrottler txThrottler
  te txEngine
  messager subComponent
+ throttler lagThrottler
 
  // hcticks starts on initialiazation and runs forever.
  hcticks *timer.Timer
@@ -155,6 +156,11 @@ type (
  Open() error
  Close()
  }
+
+ lagThrottler interface {
+ Open() error
+ Close()
+ }
 )
 
 // Init performs the second phase of initialization.
@@ -403,6 +409,7 @@ func (sm *stateManager) serveMaster() error {
  return err
  }
  sm.messager.Open()
+ sm.throttler.Open()
  sm.setState(topodatapb.TabletType_MASTER, StateServing)
  return nil
 }
@@ -422,6 +429,7 @@ func (sm *stateManager) unserveMaster() error {
 }
 
 func (sm *stateManager) serveNonMaster(wantTabletType topodatapb.TabletType) error {
+ sm.throttler.Close()
  sm.messager.Close()
  sm.tracker.Close()
  sm.se.MakeNonMaster()
@@ -469,6 +477,7 @@ func (sm *stateManager) connect(tabletType topodatapb.TabletType) error {
 }
 
 func (sm *stateManager) unserveCommon() {
+ sm.throttler.Close()
  sm.messager.Close()
  sm.te.Close()
  sm.qe.StopServing()