TxThrottler only throttles if current lag is above threshold.

This uses the lag caches in MaxReplicationLagModule to make the decision
whether to throttle or not, in addition to the result of the underlying
throttler struct.

Signed-off-by: Eduardo J. Ortega U <[email protected]>

go/vt/throttler/throttler.go

@@ -35,6 +35,7 @@ import (
 
 	"vitess.io/vitess/go/vt/discovery"
 	"vitess.io/vitess/go/vt/log"
+	"vitess.io/vitess/go/vt/proto/topodata"
 
 	throttlerdatapb "vitess.io/vitess/go/vt/proto/throttlerdata"
 )
@@ -224,6 +225,28 @@ func (t *Throttler) Throttle(threadID int) time.Duration {
 	return t.threadThrottlers[threadID].throttle(t.nowFunc())
 }
 
+// LastMaxLagNotIgnoredForTabletType returns the max of all the last replication lag values seen across all tablets of
+// the provided type, excluding ignored tablets.
+func (t *Throttler) LastMaxLagNotIgnoredForTabletType(tabletType topodata.TabletType) uint32 {
+	cache := t.maxReplicationLagModule.lagCacheByType(tabletType)
+
+	var maxLag uint32
+	cacheEntries := cache.entries
+
+	for key, _ := range cacheEntries {
+		if cache.isIgnored(key) {
+			continue
+		}
+
+		lag := cache.latest(key).Stats.ReplicationLagSeconds
+		if lag > maxLag {
+			maxLag = lag
+		}
+	}
+
+	return maxLag
+}
+
 // ThreadFinished marks threadID as finished and redistributes the thread's
 // rate allotment across the other threads.
 // After ThreadFinished() is called, Throttle() must not be called anymore.

go/vt/vttablet/tabletserver/txthrottler/tx_throttler.go

@@ -81,6 +81,7 @@ type ThrottlerInterface interface {
 	GetConfiguration() *throttlerdatapb.Configuration
 	UpdateConfiguration(configuration *throttlerdatapb.Configuration, copyZeroValues bool) error
 	ResetConfiguration()
+	LastMaxLagNotIgnoredForTabletType(tabletType topodatapb.TabletType) uint32
 }
 
 // TxThrottlerName is the name the wrapped go/vt/throttler object will be registered with
@@ -356,7 +357,18 @@ func (ts *txThrottlerStateImpl) throttle() bool {
 	// Serialize calls to ts.throttle.Throttle()
 	ts.throttleMu.Lock()
 	defer ts.throttleMu.Unlock()
-	return ts.throttler.Throttle(0 /* threadId */) > 0
+
+	var maxLag uint32
+
+	for tabletType, _ := range ts.tabletTypes {
+		maxLagPerTabletType := ts.throttler.LastMaxLagNotIgnoredForTabletType(tabletType)
+		if maxLagPerTabletType > maxLag {
+			maxLag = maxLagPerTabletType
+		}
+	}
+
+	return ts.throttler.Throttle(0 /* threadId */) > 0 &&
+		int64(maxLag) > ts.config.TxThrottlerConfig.TargetReplicationLagSec
 }
 
 func (ts *txThrottlerStateImpl) deallocateResources() {