Merge pull request #149 from github/meiji163/skip-host

Skip hosts API

Author: meiji163

build.sh

@@ -72,8 +72,8 @@ function precheck() {
     ok=1
   fi
 
-  if ! go version | egrep -q 'go(1\.1[234])' ; then
-    echo "go version must be 1.12 or above"
+  if ! go version | egrep -q 'go(1\.1[6789])' ; then
+    echo "go version must be 1.16 or above"
     ok=1
   fi
 

doc/http.md

@@ -102,6 +102,11 @@ Notes:
 
 - `/check-read-if-exists/<app>/<store-type>/<store-name>/<threshold>`: like `/check-read`, but if the metric is unknown (e.g. `<store-name>` not in `freno`'s configuration), return `200 OK`. This is useful for hybrid systems where some metrics need to be strictly controlled, and some not. `freno` would probe the important stores, and still can serve requests for all stores.
 
+- `/skip-host/<hostname>/ttl/<ttl-minutes>`: skip host when aggregating metrics for specified number of minutes. If host is already skipped, update the TTL.
+- `/skip-host/<hostname>`: same as `/skip-host/<hostname>/ttl/60`
+- `/recover-host/<hostname>`: recover a previously skipped host.
+- `/skipped-hosts`:  list currently skipped hosts
+
 ### Other requests
 
 - `/help`: show all supported request paths

doc/mysql-backend.md

@@ -82,6 +82,14 @@ CREATE TABLE throttled_apps (
 	ratio DOUBLE,
   PRIMARY KEY (app_name)
 );
+
+CREATE TABLE skipped_hosts (
+   host_name varchar(128) NOT NULL,
+   skipped_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+   expires_at TIMESTAMP NOT NULL,
+
+   PRIMARY KEY (host_name)
+);
 ```
 
 The `BackendMySQLUser` account must have `SELECT, INSERT, DELETE, UPDATE` privileges on those tables.

pkg/group/consensus_service.go

@@ -28,6 +28,10 @@ type ConsensusService interface {
 	UnthrottleApp(appName string) error
 	RecentAppsMap() (result map[string](*base.RecentApp))
 
+	SkipHost(hostName string, ttlMinutes int64, expireAt time.Time) error
+	RecoverHost(hostName string) error
+	SkippedHostsMap() (result map[string]time.Time)
+
 	IsHealthy() bool
 	IsLeader() bool
 	GetLeader() string

pkg/group/fsm.go

@@ -26,6 +26,10 @@ func (f *fsm) Apply(l *raft.Log) interface{} {
 		return f.applyThrottleApp(c.Key, c.ExpireAt, c.Ratio)
 	case "unthrottle":
 		return f.applyUnthrottleApp(c.Key)
+	case "skip":
+		return f.applySkipHost(c.Key, c.ExpireAt)
+	case "recover":
+		return f.applyRecoverHost(c.Key)
 	}
 	return log.Errorf("unrecognized command operation: %s", c.Operation)
 }
@@ -38,6 +42,11 @@ func (f *fsm) Snapshot() (raft.FSMSnapshot, error) {
 	for appName, appThrottle := range f.throttler.ThrottledAppsMap() {
 		snapshot.data.throttledApps[appName] = *appThrottle
 	}
+
+	for hostName, expireAt := range f.throttler.SkippedHostsMap() {
+		snapshot.data.skippedHosts[hostName] = expireAt
+	}
+
 	return snapshot, nil
 }
 
@@ -52,7 +61,12 @@ func (f *fsm) Restore(rc io.ReadCloser) error {
 	for appName, appThrottle := range data.throttledApps {
 		f.throttler.ThrottleApp(appName, appThrottle.ExpireAt, appThrottle.Ratio)
 	}
-	log.Debugf("freno/raft: restored from snapshot: %d elements restored", len(data.throttledApps))
+	log.Debugf("freno/raft: restored from snapshot: %d throttled apps", len(data.throttledApps))
+
+	for hostName, expireAt := range data.skippedHosts {
+		f.throttler.SkipHost(hostName, expireAt)
+	}
+	log.Debugf("freno/raft: restored from snapshot: %d skipped hosts", len(data.skippedHosts))
 	return nil
 }
 
@@ -67,3 +81,13 @@ func (f *fsm) applyUnthrottleApp(appName string) interface{} {
 	f.throttler.UnthrottleApp(appName)
 	return nil
 }
+
+func (f *fsm) applySkipHost(hostName string, expireAt time.Time) interface{} {
+	f.throttler.SkipHost(hostName, expireAt)
+	return nil
+}
+
+func (f *fsm) applyRecoverHost(hostName string) interface{} {
+	f.throttler.RecoverHost(hostName)
+	return nil
+}

pkg/group/fsm_snapshot.go

@@ -2,6 +2,7 @@ package group
 
 import (
 	"encoding/json"
+	"time"
 
 	"github.com/github/freno/pkg/base"
 
@@ -12,11 +13,13 @@ import (
 // it will mostly duplicate data stored in `throttler`.
 type snapshotData struct {
 	throttledApps map[string](base.AppThrottle)
+	skippedHosts  map[string]time.Time
 }
 
 func newSnapshotData() *snapshotData {
 	return &snapshotData{
 		throttledApps: make(map[string](base.AppThrottle)),
+		skippedHosts:  make(map[string]time.Time),
 	}
 }
 

pkg/group/mysql.go

@@ -28,6 +28,13 @@ CREATE TABLE throttled_apps (
 	ratio DOUBLE,
   PRIMARY KEY (app_name)
 );
+
+CREATE TABLE skipped_hosts (
+   host_name varchar(128) NOT NULL,
+   skipped_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+   expires_at TIMESTAMP NOT NULL,
+   PRIMARY KEY (host_name)
+);
 */
 
 package group
@@ -145,6 +152,7 @@ func (backend *MySQLBackend) onLeaderStateChange(newLeaderState int64) error {
 	if newLeaderState > 0 {
 		log.Infof("Transitioned into leader state")
 		backend.readThrottledApps()
+		backend.readSkippedHosts()
 	} else {
 		log.Infof("Transitioned out of leader state")
 	}
@@ -343,6 +351,24 @@ func (backend *MySQLBackend) readThrottledApps() error {
 	return err
 }
 
+func (backend *MySQLBackend) readSkippedHosts() error {
+	query := `
+        select
+            host_name,
+            timestampdiff(second, now(), expires_at) as ttl_seconds
+        from
+            skipped_hosts
+            `
+	err := sqlutils.QueryRowsMap(backend.db, query, func(m sqlutils.RowMap) error {
+		hostName := m.GetString("host_name")
+		ttlSeconds := m.GetInt64("ttl_seconds")
+		expireAt := time.Now().Add(time.Duration(ttlSeconds) * time.Second)
+		go backend.throttler.SkipHost(hostName, expireAt)
+		return nil
+	})
+	return err
+}
+
 func (backend *MySQLBackend) ThrottleApp(appName string, ttlMinutes int64, expireAt time.Time, ratio float64) error {
 	log.Debugf("throttle-app: app=%s, ttlMinutes=%+v, expireAt=%+v, ratio=%+v", appName, ttlMinutes, expireAt, ratio)
 	var query string
@@ -389,6 +415,36 @@ func (backend *MySQLBackend) UnthrottleApp(appName string) error {
 	return err
 }
 
+func (backend *MySQLBackend) SkipHost(hostName string, ttlMinutes int64, expireAt time.Time) error {
+	backend.throttler.SkipHost(hostName, expireAt)
+	query := `
+	    replace into skipped_hosts (
+	        host_name, skipped_at, expires_at
+	      ) values (
+	        ?, now(), now() + interval ? minute
+	      )`
+
+	ttl := ttlMinutes
+	if ttlMinutes == 0 {
+		ttl = throttle.DefaultSkipTTLMinutes
+	}
+	args := sqlutils.Args(hostName, ttl)
+	_, err := sqlutils.ExecNoPrepare(backend.db, query, args...)
+	return err
+}
+
+func (backend *MySQLBackend) RecoverHost(hostName string) error {
+	backend.throttler.RecoverHost(hostName)
+	query := `update skipped_hosts set expires_at=now() where host_name=?`
+	args := sqlutils.Args(hostName)
+	_, err := sqlutils.ExecNoPrepare(backend.db, query, args...)
+	return err
+}
+
+func (backend *MySQLBackend) SkippedHostsMap() map[string]time.Time {
+	return backend.throttler.SkippedHostsMap()
+}
+
 func (backend *MySQLBackend) RecentAppsMap() (result map[string](*base.RecentApp)) {
 	return backend.throttler.RecentAppsMap()
 }

pkg/group/store.go

@@ -19,7 +19,7 @@ import (
 	"github.com/github/freno/pkg/throttle"
 
 	"github.com/github/freno/internal/raft"
-	"github.com/github/freno/internal/raft-boltdb"
+	raftboltdb "github.com/github/freno/internal/raft-boltdb"
 	"github.com/outbrain/golib/log"
 	metrics "github.com/rcrowley/go-metrics"
 )
@@ -156,6 +156,27 @@ func (store *Store) UnthrottleApp(appName string) error {
 	return store.genericCommand(c)
 }
 
+func (store *Store) SkipHost(hostName string, ttlMinutes int64, expireAt time.Time) error {
+	c := &command{
+		Operation: "skip",
+		Key:       hostName,
+		ExpireAt:  expireAt,
+	}
+	return store.genericCommand(c)
+}
+
+func (store *Store) RecoverHost(hostName string) error {
+	c := &command{
+		Operation: "recover",
+		Key:       hostName,
+	}
+	return store.genericCommand(c)
+}
+
+func (store *Store) SkippedHostsMap() map[string]time.Time {
+	return store.throttler.SkippedHostsMap()
+}
+
 func (store *Store) ThrottledAppsMap() (result map[string](*base.AppThrottle)) {
 	return store.throttler.ThrottledAppsMap()
 }

pkg/http/api.go

@@ -36,6 +36,9 @@ type API interface {
 	ThrottleApp(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	UnthrottleApp(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	ThrottledApps(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
+	SkipHost(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
+	SkippedHosts(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
+	RecoverHost(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	RecentApps(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	Help(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
 	MemcacheConfig(w http.ResponseWriter, r *http.Request, _ httprouter.Params)
@@ -196,7 +199,7 @@ func (api *APIImpl) ReadCheck(w http.ResponseWriter, r *http.Request, ps httprou
 	api.readCheck(w, r, ps, &throttle.CheckFlags{})
 }
 
-// WriteCheckIfExists checks for a metric, but reports an OK if the metric does not exist.
+// ReadCheckIfExists checks for a metric, but reports an OK if the metric does not exist.
 // If the metric does exist, then all usual checks are made.
 func (api *APIImpl) ReadCheckIfExists(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
 	api.readCheck(w, r, ps, &throttle.CheckFlags{OKIfNotExists: true})
@@ -354,6 +357,39 @@ func metricsHandle(w http.ResponseWriter, r *http.Request, p httprouter.Params)
 	handler.ServeHTTP(w, r)
 }
 
+func (api *APIImpl) SkipHost(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	var ttlMinutes int64
+	var expireAt time.Time
+	var err error
+
+	hostName := ps.ByName("hostName")
+	ttlString := ps.ByName("ttlMinutes")
+	if ttlString == "" {
+		ttlMinutes = 0
+	} else if ttlMinutes, err = strconv.ParseInt(ttlString, 10, 64); err != nil {
+		api.respondGeneric(w, r, err)
+	}
+	if ttlMinutes != 0 {
+		expireAt = time.Now().Add(time.Duration(ttlMinutes) * time.Minute)
+	}
+
+	err = api.consensusService.SkipHost(hostName, ttlMinutes, expireAt)
+
+	api.respondGeneric(w, r, err)
+}
+
+func (api *APIImpl) RecoverHost(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	hostName := ps.ByName("hostName")
+	err := api.consensusService.RecoverHost(hostName)
+
+	api.respondGeneric(w, r, err)
+}
+
+func (api *APIImpl) SkippedHosts(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+	w.Header().Set("Content-Type", "application/json")
+	json.NewEncoder(w).Encode(api.consensusService.SkippedHostsMap())
+}
+
 // ConfigureRoutes configures a set of HTTP routes to be actions dispatched by the
 // given api's methods.
 func ConfigureRoutes(api API) *httprouter.Router {
@@ -387,6 +423,11 @@ func ConfigureRoutes(api API) *httprouter.Router {
 	register(router, "/recent-apps", api.RecentApps)
 	register(router, "/recent-apps/:lastMinutes", api.RecentApps)
 
+	register(router, "/skip-host/:hostName", api.SkipHost)
+	register(router, "/skip-host/:hostName/ttl/:ttlMinutes", api.SkipHost)
+	register(router, "/skipped-hosts", api.SkippedHosts)
+	register(router, "/recover-host/:hostName", api.RecoverHost)
+
 	register(router, "/debug/vars", metricsHandle)
 	register(router, "/debug/metrics", metricsHandle)