vstreamer to throttle on source endpoint

.github/workflows/cluster_endtoend_vreplication_basic.yml

@@ -0,0 +1,42 @@
+name: Cluster (vreplication_basic)
+on: [push, pull_request]
+jobs:
+
+  build:
+    name: Run endtoend tests on Cluster (29)
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Set up Go
+      uses: actions/setup-go@v1
+      with:
+        go-version: 1.15
+
+    - name: Check out code
+      uses: actions/checkout@v2
+
+    - name: Get dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y mysql-server mysql-client make unzip g++ etcd curl git wget eatmydata
+        sudo service mysql stop
+        sudo service etcd stop
+        sudo ln -s /etc/apparmor.d/usr.sbin.mysqld /etc/apparmor.d/disable/
+        sudo apparmor_parser -R /etc/apparmor.d/usr.sbin.mysqld
+        go mod download
+
+        wget https://repo.percona.com/apt/percona-release_latest.$(lsb_release -sc)_all.deb
+        sudo apt-get install -y gnupg2
+        sudo dpkg -i percona-release_latest.$(lsb_release -sc)_all.deb
+        sudo apt-get update
+        sudo apt-get install percona-xtrabackup-24
+
+    - name: Installing zookeeper and consul
+      run: |
+          make tools
+
+    - name: Run cluster endtoend test
+      timeout-minutes: 30
+      run: |
+        source build.env
+        eatmydata -- go run test.go -docker=false -print-log -follow -shard 29

go/sqltypes/named_result.go

@@ -78,6 +78,22 @@ func (r RowNamedValues) AsUint64(fieldName string, def uint64) uint64 {
 	return def
 }
 
+// AsFloat64 returns the named field as float64, or default value if nonexistent/error
+func (r RowNamedValues) AsFloat64(fieldName string, def float64) float64 {
+	if v, err := r.ToFloat64(fieldName); err == nil {
+		return v
+	}
+	return def
+}
+
+// ToFloat64 returns the named field as float64
+func (r RowNamedValues) ToFloat64(fieldName string) (float64, error) {
+	if v, ok := r[fieldName]; ok {
+		return v.ToFloat64()
+	}
+	return 0, ErrNoSuchField
+}
+
 // ToBool returns the named field as bool
 func (r RowNamedValues) ToBool(fieldName string) (bool, error) {
 	if v, ok := r[fieldName]; ok {

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

@@ -32,8 +32,8 @@ import (
 
 var (
 	clusterInstance *cluster.LocalProcessCluster
-	masterTablet    cluster.Vttablet
-	replicaTablet   cluster.Vttablet
+	primaryTablet   *cluster.Vttablet
+	replicaTablet   *cluster.Vttablet
 	hostname        = "localhost"
 	keyspaceName    = "ks"
 	cell            = "zone1"
@@ -65,8 +65,9 @@ var (
     }
 	}`
 
-	httpClient   = base.SetupHTTPClient(time.Second)
-	checkAPIPath = "throttler/check"
+	httpClient       = base.SetupHTTPClient(time.Second)
+	checkAPIPath     = "throttler/check"
+	checkSelfAPIPath = "throttler/check-self"
 )
 
 func TestMain(m *testing.M) {
@@ -110,9 +111,9 @@ func TestMain(m *testing.M) {
 		tablets := clusterInstance.Keyspaces[0].Shards[0].Vttablets
 		for _, tablet := range tablets {
 			if tablet.Type == "master" {
-				masterTablet = *tablet
+				primaryTablet = tablet
 			} else if tablet.Type != "rdonly" {
-				replicaTablet = *tablet
+				replicaTablet = tablet
 			}
 		}
 
@@ -121,18 +122,24 @@ func TestMain(m *testing.M) {
 	os.Exit(exitCode)
 }
 
-func throttleCheck() (*http.Response, error) {
-	return httpClient.Head(fmt.Sprintf("http://localhost:%d/%s", masterTablet.HTTPPort, checkAPIPath))
+func throttleCheck(tablet *cluster.Vttablet) (*http.Response, error) {
+	return httpClient.Head(fmt.Sprintf("http://localhost:%d/%s", tablet.HTTPPort, checkAPIPath))
+}
+
+func throttleCheckSelf(tablet *cluster.Vttablet) (*http.Response, error) {
+	return httpClient.Head(fmt.Sprintf("http://localhost:%d/%s", tablet.HTTPPort, checkSelfAPIPath))
 }
 
 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)
+	{
+		resp, err := throttleCheck(primaryTablet)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusNotFound, resp.StatusCode)
+	}
 }
 
 func TestThrottlerAfterMetricsCollected(t *testing.T) {
@@ -141,9 +148,21 @@ func TestThrottlerAfterMetricsCollected(t *testing.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)
+	{
+		resp, err := throttleCheck(primaryTablet)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+	}
+	{
+		resp, err := throttleCheckSelf(primaryTablet)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+	}
+	{
+		resp, err := throttleCheckSelf(replicaTablet)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.StatusCode)
+	}
 }
 
 func TestLag(t *testing.T) {
@@ -156,19 +175,44 @@ func TestLag(t *testing.T) {
 		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)
+		{
+			resp, err := throttleCheck(primaryTablet)
+			assert.NoError(t, err)
+			assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode)
+		}
+		{
+			resp, err := throttleCheckSelf(primaryTablet)
+			assert.NoError(t, err)
+			// self (on primary) is unaffected by replication lag
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+		}
+		{
+			resp, err := throttleCheckSelf(replicaTablet)
+			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)
+		{
+			resp, err := throttleCheck(primaryTablet)
+			assert.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+		}
+		{
+			resp, err := throttleCheckSelf(primaryTablet)
+			assert.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+		}
+		{
+			resp, err := throttleCheckSelf(replicaTablet)
+			assert.NoError(t, err)
+			assert.Equal(t, http.StatusOK, resp.StatusCode)
+		}
 	}
 }
 
@@ -181,7 +225,7 @@ func TestNoReplicas(t *testing.T) {
 		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()
+		resp, err := throttleCheck(primaryTablet)
 		assert.NoError(t, err)
 		assert.Equal(t, http.StatusOK, resp.StatusCode)
 	}
@@ -191,7 +235,7 @@ func TestNoReplicas(t *testing.T) {
 
 		time.Sleep(10 * time.Second)
 		// Restore valid replica
-		resp, err := throttleCheck()
+		resp, err := throttleCheck(primaryTablet)
 		assert.NoError(t, err)
 		assert.Equal(t, http.StatusOK, resp.StatusCode)
 	}

go/test/endtoend/vreplication/cluster.go

@@ -205,8 +205,14 @@ func (vc *VitessCluster) AddTablet(t *testing.T, cell *Cell, keyspace *Keyspace,
 		vc.Topo.Port,
 		globalConfig.hostname,
 		globalConfig.tmpDir,
-		[]string{"-queryserver-config-schema-reload-time", "5"}, //FIXME: for multi-cell initial schema doesn't seem to load without this
+		[]string{
+			"-queryserver-config-schema-reload-time", "5",
+			"-enable-lag-throttler",
+			"-heartbeat_enable",
+			"-heartbeat_interval", "250ms",
+		}, //FIXME: for multi-cell initial schema doesn't seem to load without "-queryserver-config-schema-reload-time"
 		false)
+
 	require.NotNil(t, vttablet)
 	vttablet.SupportsBackup = false