TEMP: revert estimate vreplication when fully throttled

Author: mhamza15

go/cmd/vtadmin/main.go

@@ -221,6 +221,8 @@ func main() {
 	if err := rootCmd.Execute(); err != nil {
 		log.Fatal(err)
 	}
+
+	log.Flush()
 }
 
 type noopCloser struct{}

go/test/endtoend/vreplication/fk_config_test.go

@@ -67,15 +67,15 @@ insert into t2 values(1, 1, 't21'), (2, 1, 't22'), (3, 2, 't23');
         }
       ]
     },
-    "t1": {
+	"t1": {
       "column_vindexes": [
         {
           "column": "id",
           "name": "reverse_bits"
         }
       ]
     },
-    "t2": {
+	"t2": {
       "column_vindexes": [
         {
           "column": "t1id",

go/test/endtoend/vreplication/fk_test.go

@@ -285,10 +285,10 @@ func (ls *fkLoadSimulator) exec(query string) *sqltypes.Result {
 // constraints, where the parent table is lexicographically sorted before the child table and
 // thus may be dropped first, can be successfully cancelled.
 func testFKCancel(t *testing.T, vc *VitessCluster) {
-	targetKeyspace := "fktarget"
-	sourceKeyspace := "fksource"
-	workflowName := "wf2"
-	ksWorkflow := fmt.Sprintf("%s.%s", targetKeyspace, workflowName)
+	var targetKeyspace = "fktarget"
+	var sourceKeyspace = "fksource"
+	var workflowName = "wf2"
+	var ksWorkflow = fmt.Sprintf("%s.%s", targetKeyspace, workflowName)
 	mt := newMoveTables(vc, &moveTablesWorkflow{
 		workflowInfo: &workflowInfo{
 			vc:             vc,

go/test/endtoend/vreplication/vreplication_test.go

@@ -22,7 +22,6 @@ import (
 	"io"
 	"net/http"
 	"runtime"
-	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -58,7 +57,7 @@ var (
 	targetKsOpts           = make(map[string]string)
 	httpClient             = throttlebase.SetupHTTPClient(time.Second)
 	sourceThrottlerAppName = throttlerapp.VStreamerName
-	targetThrottlerAppName = throttlerapp.VPlayerName
+	targetThrottlerAppName = throttlerapp.VReplicationName
 )
 
 const (
@@ -1168,7 +1167,7 @@ func materialize(t *testing.T, spec string, useVtctldClient bool) {
 
 func materializeProduct(t *testing.T, useVtctldClient bool) {
 	t.Run("materializeProduct", func(t *testing.T) {
-		// Materializing from "product" keyspace to "customer" keyspace.
+		// materializing from "product" keyspace to "customer" keyspace
 		workflow := "cproduct"
 		keyspace := "customer"
 		defaultCell := vc.Cells[vc.CellNames[0]]
@@ -1182,7 +1181,7 @@ func materializeProduct(t *testing.T, useVtctldClient bool) {
 
 		productTablets := vc.getVttabletsInKeyspace(t, defaultCell, "product", "primary")
 		t.Run("throttle-app-product", func(t *testing.T) {
-			// Now, throttle the source side component (vstreamer), and insert some rows.
+			// Now, throttle the streamer on source tablets, insert some rows
 			for _, tab := range productTablets {
 				body, err := throttleApp(tab, sourceThrottlerAppName)
 				assert.NoError(t, err)
@@ -1193,22 +1192,19 @@ func materializeProduct(t *testing.T, useVtctldClient bool) {
 				waitForTabletThrottlingStatus(t, tab, targetThrottlerAppName, throttlerStatusNotThrottled)
 			}
 			insertMoreProductsForSourceThrottler(t)
-			// To be fair to the test, we give the target time to apply the new changes. We
-			// expect it to NOT get them in the first place, we expect the additional rows
-			// to **not appear** in the materialized view.
+			// To be fair to the test, we give the target time to apply the new changes. We expect it to NOT get them in the first place,
+			// we expect the additional rows to **not appear** in the materialized view
 			for _, tab := range customerTablets {
 				waitForRowCountInTablet(t, tab, keyspace, workflow, 5)
-				// Confirm that we updated the stats on the target tablets as expected.
-				confirmVReplicationThrottling(t, tab, sourceKs, workflow, sourceThrottlerAppName)
 			}
 		})
 		t.Run("unthrottle-app-product", func(t *testing.T) {
-			// Unthrottle the vstreamer component, and expect the rows to show up.
+			// unthrottle on source tablets, and expect the rows to show up
 			for _, tab := range productTablets {
 				body, err := unthrottleApp(tab, sourceThrottlerAppName)
 				assert.NoError(t, err)
 				assert.Contains(t, body, sourceThrottlerAppName)
-				// Give time for unthrottling to take effect and for targets to fetch data.
+				// give time for unthrottling to take effect and for target to fetch data
 				waitForTabletThrottlingStatus(t, tab, sourceThrottlerAppName, throttlerStatusNotThrottled)
 			}
 			for _, tab := range customerTablets {
@@ -1217,8 +1213,8 @@ func materializeProduct(t *testing.T, useVtctldClient bool) {
 		})
 
 		t.Run("throttle-app-customer", func(t *testing.T) {
-			// Now, throttle vreplication on the target side (vplayer), and insert some
-			// more rows.
+			// Now, throttle vreplication (vcopier/vapplier) on target tablets, and
+			// insert some more rows.
 			for _, tab := range customerTablets {
 				body, err := throttleApp(tab, targetThrottlerAppName)
 				assert.NoError(t, err)
@@ -1233,8 +1229,6 @@ func materializeProduct(t *testing.T, useVtctldClient bool) {
 			// rows to **not appear** in the materialized view.
 			for _, tab := range customerTablets {
 				waitForRowCountInTablet(t, tab, keyspace, workflow, 8)
-				// Confirm that we updated the stats on the target tablets as expected.
-				confirmVReplicationThrottling(t, tab, sourceKs, workflow, targetThrottlerAppName)
 			}
 		})
 		t.Run("unthrottle-app-customer", func(t *testing.T) {
@@ -1790,52 +1784,3 @@ func waitForInnoDBHistoryLength(t *testing.T, tablet *cluster.VttabletProcess, e
 func releaseInnoDBRowHistory(t *testing.T, dbConn *mysql.Conn) {
 	execQuery(t, dbConn, "rollback")
 }
-
-// confirmVReplicationThrottling confirms that the throttling related metrics reflect that
-// the workflow is being throttled as expected, via the expected app name, and that this
-// is impacting the lag as expected.
-// The tablet passed should be a target tablet for the given workflow while the keyspace
-// name provided should be the source keyspace as the target tablet stats note the stream's
-// source keyspace and shard.
-func confirmVReplicationThrottling(t *testing.T, tab *cluster.VttabletProcess, keyspace, workflow string, appname throttlerapp.Name) {
-	const (
-		sleepTime = 5 * time.Second
-		zv        = int64(0)
-	)
-	time.Sleep(sleepTime) // To be sure that we accrue some lag
-
-	jsVal, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCounts"})
-	require.NoError(t, err)
-	require.NotEqual(t, "{}", jsVal)
-	// The JSON value looks like this: {"cproduct.4.tablet.vstreamer": 2, "cproduct.4.tablet.vplayer": 4}
-	throttledCount := gjson.Get(jsVal, fmt.Sprintf(`%s\.*\.tablet\.%s`, workflow, appname)).Int()
-	require.Greater(t, throttledCount, zv, "JSON value: %s", jsVal)
-
-	val, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCountTotal"})
-	require.NoError(t, err)
-	require.NotEqual(t, "", val)
-	throttledCountTotal, err := strconv.ParseInt(val, 10, 64)
-	require.NoError(t, err)
-	require.GreaterOrEqual(t, throttledCountTotal, throttledCount, "Value: %s", val)
-
-	// We do not calculate replication lag for the vcopier as it's not replicating
-	// events.
-	if appname != throttlerapp.VCopierName {
-		jsVal, err = getDebugVar(t, tab.Port, []string{"VReplicationLagSeconds"})
-		require.NoError(t, err)
-		require.NotEqual(t, "{}", jsVal)
-		// The JSON value looks like this: {"product.0.cproduct.4": 6}
-		vreplLagSeconds := gjson.Get(jsVal, fmt.Sprintf(`%s\.*\.%s\.*`, keyspace, workflow)).Int()
-		require.NoError(t, err)
-		// Take off 1 second to deal with timing issues in the test.
-		minLagSecs := int64(int64(sleepTime.Seconds()) - 1)
-		require.GreaterOrEqual(t, vreplLagSeconds, minLagSecs, "JSON value: %s", jsVal)
-
-		val, err = getDebugVar(t, tab.Port, []string{"VReplicationLagSecondsMax"})
-		require.NoError(t, err)
-		require.NotEqual(t, "", val)
-		vreplLagSecondsMax, err := strconv.ParseInt(val, 10, 64)
-		require.NoError(t, err)
-		require.GreaterOrEqual(t, vreplLagSecondsMax, vreplLagSeconds, "Value: %s", val)
-	}
-}

go/vt/binlog/binlogplayer/binlog_player.go

@@ -106,8 +106,6 @@ type Stats struct {
 
 	PartialQueryCount     *stats.CountersWithMultiLabels
 	PartialQueryCacheSize *stats.CountersWithMultiLabels
-
-	ThrottledCounts *stats.CountersWithMultiLabels // By throttler and component
 }
 
 // RecordHeartbeat updates the time the last heartbeat from vstreamer was seen
@@ -177,7 +175,6 @@ func NewStats() *Stats {
 	bps.TableCopyTimings = stats.NewTimings("", "", "Table")
 	bps.PartialQueryCacheSize = stats.NewCountersWithMultiLabels("", "", []string{"type"})
 	bps.PartialQueryCount = stats.NewCountersWithMultiLabels("", "", []string{"type"})
-	bps.ThrottledCounts = stats.NewCountersWithMultiLabels("", "", []string{"throttler", "component"})
 	return bps
 }
 
@@ -373,14 +370,13 @@ func (blp *BinlogPlayer) applyEvents(ctx context.Context) error {
 			if backoff == throttler.NotThrottled {
 				break
 			}
-			blp.blplStats.ThrottledCounts.Add([]string{"trx", "binlogplayer"}, 1)
 			// We don't bother checking for context cancellation here because the
 			// sleep will block only up to 1 second. (Usually, backoff is 1s / rate
 			// e.g. a rate of 1000 TPS results into a backoff of 1 ms.)
 			time.Sleep(backoff)
 		}
 
-		// Get the response.
+		// get the response
 		response, err := stream.Recv()
 		// Check context before checking error, because canceled
 		// contexts could be wrapped as regular errors.

go/vt/vtgate/executor.go

@@ -30,12 +30,15 @@ import (
 
 	"github.com/spf13/pflag"
 
-	"vitess.io/vitess/go/acl"
 	"vitess.io/vitess/go/cache/theine"
+	"vitess.io/vitess/go/streamlog"
+	"vitess.io/vitess/go/vt/vtenv"
+	"vitess.io/vitess/go/vt/vthash"
+
+	"vitess.io/vitess/go/acl"
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/stats"
-	"vitess.io/vitess/go/streamlog"
 	"vitess.io/vitess/go/trace"
 	"vitess.io/vitess/go/vt/callerid"
 	"vitess.io/vitess/go/vt/key"
@@ -50,7 +53,6 @@ import (
 	"vitess.io/vitess/go/vt/srvtopo"
 	"vitess.io/vitess/go/vt/sysvars"
 	"vitess.io/vitess/go/vt/topo/topoproto"
-	"vitess.io/vitess/go/vt/vtenv"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
@@ -60,7 +62,6 @@ import (
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 	"vitess.io/vitess/go/vt/vtgate/vschemaacl"
 	"vitess.io/vitess/go/vt/vtgate/vtgateservice"
-	"vitess.io/vitess/go/vt/vthash"
 )
 
 var (

go/vt/vtgate/sandbox_test.go

@@ -281,10 +281,7 @@ func (sct *sandboxTopo) WatchSrvVSchema(ctx context.Context, cell string, callba
 	}
 
 	sct.topoServer.UpdateSrvVSchema(ctx, cell, srvVSchema)
-	current, updateChan, err := sct.topoServer.WatchSrvVSchema(ctx, cell)
-	if err != nil {
-		panic(fmt.Sprintf("sandboxTopo WatchSrvVSchema returned an error: %v", err))
-	}
+	current, updateChan, _ := sct.topoServer.WatchSrvVSchema(ctx, cell)
 	if !callback(current.Value, nil) {
 		panic("sandboxTopo callback returned false")
 	}

go/vt/vttablet/tabletmanager/vreplication/stats.go

@@ -59,12 +59,10 @@ type vrStats struct {
 	mu          sync.Mutex
 	isOpen      bool
 	controllers map[int32]*controller
-
-	ThrottledCount *stats.Counter
 }
 
 func (st *vrStats) register() {
-	st.ThrottledCount = stats.NewCounter("", "")
+
 	stats.NewGaugeFunc("VReplicationStreamCount", "Number of vreplication streams", st.numControllers)
 	stats.NewGaugeFunc("VReplicationLagSecondsMax", "Max vreplication seconds behind primary", st.maxReplicationLagSeconds)
 	stats.NewStringMapFuncWithMultiLabels(
@@ -504,29 +502,6 @@ func (st *vrStats) register() {
 			return result
 		})
 
-	stats.NewCounterFunc(
-		"VReplicationThrottledCountTotal",
-		"The total number of times that vreplication has been throttled",
-		func() int64 {
-			st.mu.Lock()
-			defer st.mu.Unlock()
-			return st.ThrottledCount.Get()
-		})
-	stats.NewCountersFuncWithMultiLabels(
-		"VReplicationThrottledCounts",
-		"The number of times vreplication was throttled by workflow, id, throttler (trx or tablet), and the sub-component that was throttled",
-		[]string{"workflow", "id", "throttler", "component"},
-		func() map[string]int64 {
-			st.mu.Lock()
-			defer st.mu.Unlock()
-			result := make(map[string]int64)
-			for _, ct := range st.controllers {
-				for key, val := range ct.blpStats.ThrottledCounts.Counts() {
-					result[fmt.Sprintf("%s.%d.%s", ct.workflow, ct.id, key)] = val
-				}
-			}
-			return result
-		})
 }
 
 func (st *vrStats) numControllers() int64 {

go/vt/vttablet/tabletmanager/vreplication/stats_test.go

@@ -26,7 +26,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"vitess.io/vitess/go/mysql/replication"
-	"vitess.io/vitess/go/stats"
 
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
 	"vitess.io/vitess/go/vt/proto/binlogdata"
@@ -133,9 +132,7 @@ func TestStatusHtml(t *testing.T) {
 func TestVReplicationStats(t *testing.T) {
 	blpStats := binlogplayer.NewStats()
 	defer blpStats.Stop()
-	testStats := &vrStats{
-		ThrottledCount: stats.NewCounter("", ""),
-	}
+	testStats := &vrStats{}
 	testStats.isOpen = true
 	testStats.controllers = map[int32]*controller{
 		1: {
@@ -187,14 +184,6 @@ func TestVReplicationStats(t *testing.T) {
 	require.Equal(t, int64(100), testStats.status().Controllers[0].CopyLoopCount)
 	require.Equal(t, int64(200), testStats.status().Controllers[0].CopyRowCount)
 
-	testStats.ThrottledCount.Add(99)
-	require.Equal(t, int64(99), testStats.ThrottledCount.Get())
-
-	blpStats.ThrottledCounts.Add([]string{"tablet", "vcopier"}, 10)
-	blpStats.ThrottledCounts.Add([]string{"tablet", "vplayer"}, 80)
-	require.Equal(t, int64(10), testStats.controllers[1].blpStats.ThrottledCounts.Counts()["tablet.vcopier"])
-	require.Equal(t, int64(80), testStats.controllers[1].blpStats.ThrottledCounts.Counts()["tablet.vplayer"])
-
 	var tm int64 = 1234567890
 	blpStats.RecordHeartbeat(tm)
 	require.Equal(t, tm, blpStats.Heartbeat())