Very hacky very minimal circuit breaker in vtgate

Signed-off-by: Brendan Dougherty <[email protected]>

examples/common/scripts/vttablet-up.sh

@@ -45,7 +45,7 @@ vttablet \
  --init_keyspace $keyspace \
  --init_shard $shard \
  --init_tablet_type $tablet_type \
- --health_check_interval 5s \
+ --health_check_interval 1s \
  --backup_storage_implementation file \
  --file_backup_storage_root $VTDATAROOT/backups \
  --restore_from_backup \

go/vt/discovery/circuit_breaker.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2019 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package discovery
+
+import "vitess.io/vitess/go/vt/log"
+
+type CircuitBreakerState int
+
+const (
+	CircuitBreakerState_CLOSED CircuitBreakerState = iota
+	CircuitBreakerState_OPEN
+)
+
+// These would come from command line flags
+const queryTimeoutWindowSeconds = 3
+const queryTimeoutPerSecondThreshold = 1
+
+func FilterStatsByCircuitBreakerState(tabletHealthList []*TabletHealth) []*TabletHealth {
+	aliases := make([]string, 0, len(tabletHealthList))
+	for _, ts := range tabletHealthList {
+		aliases = append(aliases, ts.Tablet.Alias.String())
+	}
+	log.Infof("Filtering tablet health list by circuit state: %v", aliases)
+
+	list := make([]*TabletHealth, 0, len(tabletHealthList))
+	for _, ts := range tabletHealthList {
+		if ts.Stats == nil || len(ts.Stats.QueryTimeoutRates) == 0 {
+			continue
+		}
+
+		timeoutRates := ts.Stats.QueryTimeoutRates
+		windowSize := min(queryTimeoutWindowSeconds, len(timeoutRates))
+		ratesInWindow := timeoutRates[len(timeoutRates)-windowSize:]
+		log.Infof("query timeout rates for tablet %v: %v", ts.Tablet.Alias.String(), ratesInWindow)
+
+		sum := 0.0
+		for _, rate := range ratesInWindow {
+			sum += rate
+		}
+		avgRate := sum / float64(windowSize)
+		log.Infof("average query timeout rate for tablet %v: %v", ts.Tablet.Alias.String(), avgRate)
+
+		if avgRate > queryTimeoutPerSecondThreshold {
+			log.Infof("removing tablet %v from healthcheck list due to high query timeout rate: %v", ts.Tablet.Alias.String(), avgRate)
+			ts.CircuitState = CircuitBreakerState_OPEN
+			continue
+		}
+
+		ts.CircuitState = CircuitBreakerState_CLOSED
+		list = append(list, ts)
+	}
+
+	return list
+}

go/vt/discovery/fake_healthcheck.go

@@ -90,6 +90,19 @@ func (fhc *FakeHealthCheck) GetHealthyTabletStats(target *querypb.Target) []*Tab
 	return result
 }
 
+// GetTabletStats returns all the tablets
+func (fhc *FakeHealthCheck) GetTabletStats(target *querypb.Target) []*TabletHealth {
+	result := make([]*TabletHealth, 0)
+	fhc.mu.Lock()
+	defer fhc.mu.Unlock()
+	for _, item := range fhc.items {
+		if proto.Equal(item.ts.Target, target) {
+			result = append(result, item.ts)
+		}
+	}
+	return result
+}
+
 // GetTabletHealthByAlias results the TabletHealth of the tablet that matches the given alias
 func (fhc *FakeHealthCheck) GetTabletHealthByAlias(alias *topodatapb.TabletAlias) (*TabletHealth, error) {
 	return fhc.GetTabletHealth("", alias)

go/vt/discovery/healthcheck.go

@@ -222,6 +222,12 @@ type HealthCheck interface {
 	// synchronization
 	GetHealthyTabletStats(target *query.Target) []*TabletHealth
 
+	// GetTabletStats returns all tablets for the given target.
+	// The returned array is owned by the caller.
+	// For TabletType_PRIMARY, this will only return at most one entry,
+	// the most recent tablet of type primary.
+	GetTabletStats(target *query.Target) []*TabletHealth
+
 	// GetTabletHealth results the TabletHealth of the tablet that matches the given alias
 	GetTabletHealth(kst KeyspaceShardTabletType, alias *topodata.TabletAlias) (*TabletHealth, error)
 
@@ -560,6 +566,17 @@ func (hc *HealthCheckImpl) updateHealth(th *TabletHealth, prevTarget *query.Targ
 		}
 	}
 
+	// We recompute the circuit breaker health whenever we get a health update
+	// We want this to happen for primary tablets as well, hence why we're not including it in `recomputeHealthy`
+	if hc.isIncluded(th.Target.TabletType, th.Tablet.Alias) {
+		hc.recomputeCircuitBreakerHealth(targetKey)
+	}
+	// Not sure if this is necessary
+	if targetChanged && hc.isIncluded(th.Target.TabletType, th.Tablet.Alias) {
+		oldTargetKey := KeyFromTarget(prevTarget)
+		hc.recomputeCircuitBreakerHealth(oldTargetKey)
+	}
+
 	isNewPrimary := isPrimary && prevTarget.TabletType != topodata.TabletType_PRIMARY
 	if isNewPrimary {
 		log.Errorf("Adding 1 to PrimaryPromoted counter for target: %v, tablet: %v, tabletType: %v", prevTarget, topoproto.TabletAliasString(th.Tablet.Alias), th.Target.TabletType)
@@ -582,6 +599,18 @@ func (hc *HealthCheckImpl) recomputeHealthy(key KeyspaceShardTabletType) {
 	hc.healthy[key] = FilterStatsByReplicationLag(allArray)
 }
 
+func (hc *HealthCheckImpl) recomputeCircuitBreakerHealth(key KeyspaceShardTabletType) {
+	all := hc.healthData[key]
+	allArray := make([]*TabletHealth, 0, len(all))
+	for _, s := range all {
+		// Only tablets in same cell / cellAlias are included in healthy list.
+		if hc.isIncluded(s.Tablet.Type, s.Tablet.Alias) {
+			allArray = append(allArray, s)
+		}
+	}
+	hc.healthy[key] = FilterStatsByCircuitBreakerState(allArray)
+}
+
 // Subscribe adds a listener. Used by vtgate buffer to learn about primary changes.
 func (hc *HealthCheckImpl) Subscribe() chan *TabletHealth {
 	hc.subMu.Lock()

go/vt/discovery/tablet_health.go

@@ -40,6 +40,7 @@ type TabletHealth struct {
 	PrimaryTermStartTime int64
 	LastError            error
 	Serving              bool
+	CircuitState         CircuitBreakerState
 }
 
 func (th *TabletHealth) MarshalJSON() ([]byte, error) {

go/vt/vtgate/tabletgateway.go

@@ -290,9 +290,16 @@ func (gw *TabletGateway) withRetry(ctx context.Context, target *querypb.Target,
 					err = vterrors.Errorf(vtrpcpb.Code_CLUSTER_EVENT, buffer.ClusterEventReparentInProgress)
 					continue
 				}
-				// if primary is serving, but we initially found no tablet, we're in an inconsistent state
-				// we then retry the entire loop
 				if primary != nil {
+					thList := gw.hc.GetTabletStats(target)
+					if len(thList) == 1 && thList[0].CircuitState == discovery.CircuitBreakerState_OPEN {
+						// fail fast with a specific error message
+						err = vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "circuit breaker is open for '%s'", target.String())
+						break
+					}
+
+					// if primary is serving, but we initially found no tablet, we're in an inconsistent state
+					// we then retry the entire loop
 					err = vterrors.Errorf(vtrpcpb.Code_UNAVAILABLE, "inconsistent state detected, primary is serving but initially found no available tablet")
 					continue
 				}