Tablet throttler: (feature flagged) get remote tablets metrics from Realtime Stats

go/vt/vttablet/tabletserver/health_streamer.go

@@ -225,7 +225,7 @@ func (hs *healthStreamer) unregister(ch chan *querypb.StreamHealthResponse) {
 	delete(hs.clients, ch)
 }
 
-func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimestamp time.Time, lag time.Duration, err error, serving bool) {
+func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimestamp time.Time, lag time.Duration, throttlerMetric float64, throttlerMetricsErr error, stateErr error, serving bool) {
 	hs.mu.Lock()
 	defer hs.mu.Unlock()
 
@@ -235,8 +235,8 @@ func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimes
 	} else {
 		hs.state.TabletExternallyReparentedTimestamp = 0
 	}
-	if err != nil {
-		hs.state.RealtimeStats.HealthError = err.Error()
+	if stateErr != nil {
+		hs.state.RealtimeStats.HealthError = stateErr.Error()
 	} else {
 		hs.state.RealtimeStats.HealthError = ""
 	}
@@ -246,6 +246,14 @@ func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimes
 	hs.state.RealtimeStats.FilteredReplicationLagSeconds, hs.state.RealtimeStats.BinlogPlayersCount = blpFunc()
 	hs.state.RealtimeStats.Qps = hs.stats.QPSRates.TotalRate()
 
+	hs.state.RealtimeStats.ThrottlerMetric = throttlerMetric
+	if throttlerMetricsErr != nil {
+		hs.state.RealtimeStats.ThrottlerMetricError = throttlerMetricsErr.Error()
+	} else {
+		hs.state.RealtimeStats.ThrottlerMetricError = ""
+	}
+	hs.state.RealtimeStats.ThrottlerMetricCollected = true // the field `ThrottlerMetricCollected` will be removed in v18
+
 	shr := proto.Clone(hs.state).(*querypb.StreamHealthResponse)
 
 	hs.broadCastToClients(shr)
@@ -254,7 +262,7 @@ func (hs *healthStreamer) ChangeState(tabletType topodatapb.TabletType, terTimes
 		serving:    shr.Serving,
 		tabletType: shr.Target.TabletType,
 		lag:        lag,
-		err:        err,
+		err:        stateErr,
 	})
 }
 

go/vt/vttablet/tabletserver/health_streamer_test.go

@@ -93,7 +93,7 @@ func TestHealthStreamerBroadcast(t *testing.T) {
 	}
 	assert.Truef(t, proto.Equal(want, shr), "want: %v, got: %v", want, shr)
 
-	hs.ChangeState(topodatapb.TabletType_REPLICA, time.Time{}, 0, nil, false)
+	hs.ChangeState(topodatapb.TabletType_REPLICA, time.Time{}, 0, 0, nil, nil, false)
 	shr = <-ch
 	want = &querypb.StreamHealthResponse{
 		Target: &querypb.Target{
@@ -109,7 +109,7 @@ func TestHealthStreamerBroadcast(t *testing.T) {
 
 	// Test primary and timestamp.
 	now := time.Now()
-	hs.ChangeState(topodatapb.TabletType_PRIMARY, now, 0, nil, true)
+	hs.ChangeState(topodatapb.TabletType_PRIMARY, now, 0, 0, nil, nil, true)
 	shr = <-ch
 	want = &querypb.StreamHealthResponse{
 		Target: &querypb.Target{
@@ -126,7 +126,7 @@ func TestHealthStreamerBroadcast(t *testing.T) {
 	assert.Truef(t, proto.Equal(want, shr), "want: %v, got: %v", want, shr)
 
 	// Test non-serving, and 0 timestamp for non-primary.
-	hs.ChangeState(topodatapb.TabletType_REPLICA, now, 1*time.Second, nil, false)
+	hs.ChangeState(topodatapb.TabletType_REPLICA, now, 1*time.Second, 1.0, nil, nil, false)
 	shr = <-ch
 	want = &querypb.StreamHealthResponse{
 		Target: &querypb.Target{
@@ -136,13 +136,14 @@ func TestHealthStreamerBroadcast(t *testing.T) {
 		RealtimeStats: &querypb.RealtimeStats{
 			ReplicationLagSeconds:         1,
 			FilteredReplicationLagSeconds: 1,
+			ThrottlerMetric:               1.0,
 			BinlogPlayersCount:            2,
 		},
 	}
 	assert.Truef(t, proto.Equal(want, shr), "want: %v, got: %v", want, shr)
 
 	// Test Health error.
-	hs.ChangeState(topodatapb.TabletType_REPLICA, now, 0, errors.New("repl err"), false)
+	hs.ChangeState(topodatapb.TabletType_REPLICA, now, 0, 0, nil, errors.New("repl err"), false)
 	shr = <-ch
 	want = &querypb.StreamHealthResponse{
 		Target: &querypb.Target{

go/vt/vttablet/tabletserver/state_manager.go

@@ -181,6 +181,7 @@ type (
 	lagThrottler interface {
 		Open() error
 		Close()
+		SelfMetrics() (float64, error)
 	}
 
 	tableGarbageCollector interface {
@@ -663,8 +664,9 @@ func (sm *stateManager) Broadcast() {
 	sm.mu.Lock()
 	defer sm.mu.Unlock()
 
-	lag, err := sm.refreshReplHealthLocked()
-	sm.hs.ChangeState(sm.target.TabletType, sm.terTimestamp, lag, err, sm.isServingLocked())
+	lag, stateErr := sm.refreshReplHealthLocked()
+	throttlerMetric, throttlerMetricsErr := sm.throttler.SelfMetrics()
+	sm.hs.ChangeState(sm.target.TabletType, sm.terTimestamp, lag, throttlerMetric, throttlerMetricsErr, stateErr, sm.isServingLocked())
 }
 
 func (sm *stateManager) refreshReplHealthLocked() (time.Duration, error) {

go/vt/vttablet/tabletserver/state_manager_test.go

@@ -923,6 +923,10 @@ func (te *testLagThrottler) Close() {
 	te.state = testStateClosed
 }
 
+func (te *testLagThrottler) SelfMetrics() (float64, error) {
+	return 0, nil
+}
+
 type testTableGC struct {
 	testOrderState
 }

go/vt/vttablet/tabletserver/tabletserver.go

@@ -176,7 +176,7 @@ func NewTabletServer(name string, config *tabletenv.TabletConfig, topoServer *to
 	tsv.hs = newHealthStreamer(tsv, alias)
 	tsv.se = schema.NewEngine(tsv)
 	tsv.rt = repltracker.NewReplTracker(tsv, alias)
-	tsv.lagThrottler = throttle.NewThrottler(tsv, srvTopoServer, topoServer, alias.Cell, tsv.rt.HeartbeatWriter(), tabletTypeFunc)
+	tsv.lagThrottler = throttle.NewThrottler(tsv, srvTopoServer, topoServer, alias, tsv.rt.HeartbeatWriter(), tabletTypeFunc)
 	tsv.vstreamer = vstreamer.NewEngine(tsv, srvTopoServer, tsv.se, tsv.lagThrottler, alias.Cell)
 	tsv.tracker = schema.NewTracker(tsv, tsv.vstreamer, tsv.se)
 	tsv.watcher = NewBinlogWatcher(tsv, tsv.vstreamer, tsv.config)