Add global throttled count and e2e tests

Signed-off-by: Matt Lord <mattalord@gmail.com>

Author: mattlord

go/test/endtoend/vreplication/vreplication_test.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"net/http"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -1184,6 +1185,32 @@ func materializeProduct(t *testing.T, useVtctldClient bool) {
 			// 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 throttled target tablets as expected.
+				jsVal, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCounts"})
+				require.NoError(t, err)
+				require.NotEqual(t, "{}", jsVal)
+				// Value looks like this: {"cproduct.4.tablet.vstreamer": 2}
+				require.Greater(t, gjson.Get(jsVal, fmt.Sprintf(`%s\.*\.tablet\.vstreamer`, workflow)).Int(), int64(0))
+				// We only need to do this once.
+				val, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCountTotal"})
+				require.NoError(t, err)
+				throttledCount, err := strconv.Atoi(val)
+				require.NoError(t, err)
+				require.Greater(t, throttledCount, 0)
+			}
+			// Now confirm that we updated the stats on the throttled source tablets as expected.
+			for _, tab := range productTablets {
+				jsVal, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCounts"})
+				require.NoError(t, err)
+				require.NotEqual(t, "{}", jsVal)
+				// Value looks like this: {"rollup.1.tablet.vstreamer": 2, "rollup.1.tablet.rowstreamer": 1}
+				require.Greater(t, gjson.Get(jsVal, `*\.*\.tablet\.vstreamer`).Int(), int64(0))
+				// We only need to do this once.
+				val, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCountTotal"})
+				require.NoError(t, err)
+				throttledCount, err := strconv.Atoi(val)
+				require.NoError(t, err)
+				require.Greater(t, throttledCount, 0)
 			}
 		})
 		t.Run("unthrottle-app-product", func(t *testing.T) {
@@ -1217,6 +1244,12 @@ 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 throttled target tablets as expected.
+				jsVal, err := getDebugVar(t, tab.Port, []string{"VReplicationThrottledCounts"})
+				require.NoError(t, err)
+				require.NotEqual(t, "{}", jsVal)
+				// Value looks like this: {"cproduct.4.tablet.vstreamer": 2, "cproduct.4.tablet.vplayer": 4}
+				require.Greater(t, gjson.Get(jsVal, fmt.Sprintf(`%s\.*\.tablet\.vplayer`, workflow)).Int(), int64(0))
 			}
 		})
 		t.Run("unthrottle-app-customer", func(t *testing.T) {

go/vt/binlog/binlogplayer/binlog_player.go

@@ -106,7 +106,7 @@ type Stats struct {
 	PartialQueryCount     *stats.CountersWithMultiLabels
 	PartialQueryCacheSize *stats.CountersWithMultiLabels
 
-	ThrottledCounts *stats.CountersWithMultiLabels
+	ThrottledCounts *stats.CountersWithMultiLabels // By throttler and component
 }
 
 // RecordHeartbeat updates the time the last heartbeat from vstreamer was seen

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

@@ -59,10 +59,12 @@ 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(
@@ -502,6 +504,14 @@ 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",
@@ -511,7 +521,7 @@ func (st *vrStats) register() {
 			defer st.mu.Unlock()
 			result := make(map[string]int64)
 			for _, ct := range st.controllers {
-				for key, val := range ct.blpStats.ErrorCounts.Counts() {
+				for key, val := range ct.blpStats.ThrottledCounts.Counts() {
 					result[fmt.Sprintf("%s.%d.%s", ct.workflow, ct.id, key)] = val
 				}
 			}

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

@@ -26,6 +26,7 @@ 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"
@@ -132,7 +133,9 @@ func TestStatusHtml(t *testing.T) {
 func TestVReplicationStats(t *testing.T) {
 	blpStats := binlogplayer.NewStats()
 	defer blpStats.Stop()
-	testStats := &vrStats{}
+	testStats := &vrStats{
+		ThrottledCount: stats.NewCounter("", ""),
+	}
 	testStats.isOpen = true
 	testStats.controllers = map[int32]*controller{
 		1: {
@@ -184,6 +187,9 @@ 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"])

go/vt/vttablet/tabletmanager/vreplication/vreplicator.go

@@ -589,6 +589,7 @@ func (vr *vreplicator) throttlerAppName() string {
 func (vr *vreplicator) updateTimeThrottled(appThrottled throttlerapp.Name) error {
 	appName := appThrottled.String()
 	vr.stats.ThrottledCounts.Add([]string{"tablet", appName}, 1)
+	globalStats.ThrottledCount.Add(1)
 	err := vr.throttleUpdatesRateLimiter.Do(func() error {
 		tm := time.Now().Unix()
 		update, err := binlogplayer.GenerateUpdateTimeThrottled(vr.id, tm, appName)