Merge pull request #150 from github/meiji163/throttle-cluster

Throttle app by cluster

doc/http.md

@@ -49,7 +49,7 @@ Notes:
   - `/throttle-app/archive/ttl/30/ratio/1`: completely refuse `/check/archive/*` requests for a duration of `30` minutes
   - `/throttle-app/archive/ttl/30/ratio/0.9`: _mostly_ refuse `/check/archive/*` requests for a duration of `30` minutes. On average (random dice roll), `9` out of `10` requests (i.e. `90%`) will be denied, and one approved.
   - `/throttle-app/archive/ttl/30/ratio/0.5`: refuse `50%` of `/check/archive/*` requests for a duration of `30` minutes
-
+  
 - `/throttle-app/<app-name>/ttl/<ttlMinutes>`:
 
   - If app is already throttled, modify TTL portion only, without changing the ratio.
@@ -65,7 +65,8 @@ Notes:
 
   Same as calling `/throttle-app/<app-name>/ttl/60/ratio/1`. Provided as convenience endpoint.
 
-- `/unthrottle-app/<app-name>`: remove any imposed throttling constraint from given app. Example:
+- `/throttle-app` can take a query parameter `store_name` to throttle the app only on one store (i.e. MySQL cluster). For example `/throttle-app/archive?store_name=mycluster` refuses `/check/archive/mysql/mycluster` requests for `1` hour.
+
 
   `/unthrottle-app/archive` will re-allow the `archive` app to get valid response from `/check/archive/*` requests.
 

pkg/http/api.go

@@ -235,11 +235,21 @@ func (api *APIImpl) MetricsHealth(w http.ResponseWriter, r *http.Request, ps htt
 
 // ThrottleApp forcibly marks given app as throttled. Future requests by this app may be denied.
 func (api *APIImpl) ThrottleApp(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
-	appName := ps.ByName("app")
+	storeName := r.URL.Query().Get("store_name")
+	var appName string
+	if storeName != "" {
+		// limit throttling to this store
+		appName = fmt.Sprintf("%s/%s", ps.ByName("app"), storeName)
+	} else {
+		// default is throttle app globally
+		appName = ps.ByName("app")
+	}
+
 	var expireAt time.Time // default zero
 	var ttlMinutes int64
 	var ratio float64
 	var err error
+
 	if ps.ByName("ttlMinutes") == "" {
 		ttlMinutes = 0
 	} else if ttlMinutes, err = strconv.ParseInt(ps.ByName("ttlMinutes"), 10, 64); err != nil {
@@ -263,12 +273,21 @@ response:
 	api.respondGeneric(w, r, err)
 }
 
-// ThrottleApp unthrottles given app.
+// UnthrottleApp unthrottles given app.
 func (api *APIImpl) UnthrottleApp(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
 	appName := ps.ByName("app")
-	err := api.consensusService.UnthrottleApp(appName)
-
-	api.respondGeneric(w, r, err)
+	appWithStorePrefix := appName + "/"
+
+	for app := range api.consensusService.ThrottledAppsMap() {
+		if app == appName || strings.HasPrefix(app, appWithStorePrefix) {
+			err := api.consensusService.UnthrottleApp(app)
+			if err != nil {
+				api.respondGeneric(w, r, err)
+				return
+			}
+		}
+	}
+	api.respondGeneric(w, r, nil)
 }
 
 // ThrottledApps returns a snapshot of all currently throttled apps

pkg/throttle/check.go

@@ -48,7 +48,7 @@ func (check *ThrottlerCheck) checkAppMetricResult(appName string, storeType stri
 		}
 	}
 	//
-	metricResult, threshold := check.throttler.AppRequestMetricResult(appName, metricResultFunc, denyApp)
+	metricResult, threshold := check.throttler.AppRequestMetricResult(appName, storeName, metricResultFunc, denyApp)
 	if flags.OverrideThreshold > 0 {
 		threshold = flags.OverrideThreshold
 	}

pkg/throttle/throttler.go

@@ -527,16 +527,21 @@ func (throttler *Throttler) UnthrottleApp(appName string) {
 	throttler.throttledApps.Delete(appName)
 }
 
-func (throttler *Throttler) IsAppThrottled(appName string) bool {
-	if object, found := throttler.throttledApps.Get(appName); found {
-		appThrottle := object.(*base.AppThrottle)
-		if appThrottle.ExpireAt.Before(time.Now()) {
-			// throttling cleanup hasn't purged yet, but it is expired
-			return false
-		}
-		// handle ratio
-		if rand.Float64() < appThrottle.Ratio {
-			return true
+func (throttler *Throttler) IsAppThrottled(appName, storeName string) bool {
+	appWithStore := fmt.Sprintf("%s/%s", appName, storeName)
+	keys := []string{appWithStore, appName}
+	// check if app is throttled for this store or globally
+	for _, key := range keys {
+		if object, found := throttler.throttledApps.Get(key); found {
+			appThrottle := object.(*base.AppThrottle)
+			if appThrottle.ExpireAt.Before(time.Now()) {
+				// throttling cleanup hasn't purged yet, but it is expired
+				continue
+			}
+			// handle ratio
+			if rand.Float64() < appThrottle.Ratio {
+				return true
+			}
 		}
 	}
 	return false
@@ -589,11 +594,11 @@ func (throttler *Throttler) metricsHealthSnapshot() base.MetricHealthMap {
 	return snapshot
 }
 
-func (throttler *Throttler) AppRequestMetricResult(appName string, metricResultFunc base.MetricResultFunc, denyApp bool) (metricResult base.MetricResult, threshold float64) {
+func (throttler *Throttler) AppRequestMetricResult(appName string, storeName string, metricResultFunc base.MetricResultFunc, denyApp bool) (metricResult base.MetricResult, threshold float64) {
 	if denyApp {
 		return base.AppDeniedMetric, 0
 	}
-	if throttler.IsAppThrottled(appName) {
+	if throttler.IsAppThrottled(appName, storeName) {
 		return base.AppDeniedMetric, 0
 	}
 	return metricResultFunc()