Support slow‑start feature to allows a backend instance gradually recover its weight (bfenetworks#692)

Author: yangshuothtf

bfe_balance/backend/bfe_backend.go

@@ -41,6 +41,8 @@ type BfeBackend struct {
 	succNum      int  // number of consecutive successes of health-check request
 
 	closeChan chan bool // tell health-check to stop
+
+	restarted    bool // indicate if this backend is new bring-up by health-check
 }
 
 func NewBfeBackend() *BfeBackend {
@@ -90,6 +92,19 @@ func (back *BfeBackend) setAvail(avail bool) {
 	}
 }
 
+func (back *BfeBackend) SetRestart(restart bool) {
+	back.Lock()
+	back.restarted = restart
+	back.Unlock()
+}
+
+func (back *BfeBackend) GetRestart() bool {
+	back.RLock()
+	restart := back.restarted
+	back.RUnlock()
+	return restart
+}
+
 func (back *BfeBackend) ConnNum() int {
 	back.RLock()
 	connNum := back.connNum

bfe_balance/backend/health_check.go

@@ -96,6 +96,7 @@ loop:
 		}
 
 		log.Logger.Info("backend %s back to Normal", backend.Name)
+		backend.SetRestart(true)
 		backend.SetAvail(true)
 		break loop
 	}

bfe_balance/bal_gslb/bal_gslb.go

@@ -89,6 +89,16 @@ func (bal *BalanceGslb) SetGslbBasic(gslbBasic cluster_conf.GslbBasicConf) {
 	bal.lock.Unlock()
 }
 
+func (bal *BalanceGslb) SetSlowStart(backendConf cluster_conf.BackendBasic) {
+	bal.lock.Lock()
+
+	for _, sub := range bal.subClusters {
+		sub.setSlowStart(*backendConf.SlowStartTime)
+	}
+
+	bal.lock.Unlock()
+}
+
 // Init inializes gslb cluster with config
 func (bal *BalanceGslb) Init(gslbConf gslb_conf.GslbClusterConf) error {
 	totalWeight := 0

bfe_balance/bal_gslb/bal_gslb_test.go

@@ -133,3 +133,65 @@ func SetReqHeader(req *bfe_basic.Request, key string) {
 		req.HttpRequest.Header.Set(key, "val")
 	}
 }
+
+func TestSlowStart(t *testing.T) {
+	t.Logf("bal_gslb_test: TestSlowStart")
+	var c cluster_table_conf.ClusterBackend
+	var gb cluster_conf.GslbBasicConf
+	var g gslb_conf.GslbClusterConf
+	var err error
+
+	loadJson("testdata/cluster1", &c)
+	loadJson("testdata/gb", &gb)
+	loadJson("testdata/g1", &g)
+	t.Logf("%v %v %v\n", c, gb, g)
+
+	bal := NewBalanceGslb("cluster_dumi")
+	if err := bal.Init(g); err != nil {
+		t.Errorf("init error %s", err)
+	}
+	t.Logf("%+v\n", bal)
+	if bal.totalWeight != 100 || !bal.single || bal.subClusters[bal.avail].Name != "light.example.wt" || bal.retryMax != 3 || bal.crossRetry != 1 {
+		t.Errorf("init error")
+	}
+
+	if len(bal.subClusters) != 3 {
+		t.Errorf("cluster len error")
+	}
+
+	t.Logf("%+v", bal.subClusters[0])
+	t.Logf("%+v", bal.subClusters[1])
+	t.Logf("%+v", bal.subClusters[2])
+
+	var c1 cluster_table_conf.ClusterBackend
+	var gb1 cluster_conf.GslbBasicConf
+	var g1 gslb_conf.GslbClusterConf
+	loadJson("testdata/cluster2", &c1)
+	loadJson("testdata/gb2", &gb1)
+	loadJson("testdata/g2", &g1)
+
+	err = cluster_conf.GslbBasicConfCheck(&gb1)
+	if err != nil {
+		t.Errorf("GslbBasicConfCheck err %s", err)
+	}
+	t.Logf("%v %v %v\n", c1, gb1, g1)
+	if err := bal.Reload(g1); err != nil {
+		t.Errorf("reload error %s", err)
+	}
+
+	bal.SetGslbBasic(gb1)
+
+	var backendConf cluster_conf.BackendBasic
+	err = cluster_conf.BackendBasicCheck(&backendConf)
+	if err != nil {
+		t.Errorf("BackendBasicCheck err %s", err)
+	}
+	var ssTime = 30
+	backendConf.SlowStartTime = &ssTime
+	bal.SetSlowStart(backendConf)
+
+	t.Logf("%+v\n", bal)
+	t.Logf("%+v", bal.subClusters[0])
+	t.Logf("%+v", bal.subClusters[1])
+	t.Logf("%+v", bal.subClusters[2])
+}

bfe_balance/bal_gslb/sub_cluster.go

@@ -85,6 +85,10 @@ func (sub *SubCluster) balance(algor int, key []byte) (*backend.BfeBackend, erro
 	return sub.backends.Balance(algor, key)
 }
 
+func (sub *SubCluster) setSlowStart(slowStartTime int) {
+	sub.backends.SetSlowStart(slowStartTime)
+}
+
 // SubClusterList is a list of subcluster.
 type SubClusterList []*SubCluster
 

bfe_balance/bal_slb/backend_rr.go

@@ -16,15 +16,27 @@
 
 package bal_slb
 
+import (
+	"time"
+)
+
 import (
 	"github.com/bfenetworks/bfe/bfe_balance/backend"
 	"github.com/bfenetworks/bfe/bfe_config/bfe_cluster_conf/cluster_table_conf"
 )
 
+type WeightSS struct {
+	final          int        // final target weight after slow-start
+	slowStartTime  int        // time for backend increases the weight to the full value, in seconds
+	startTime      time.Time  // time of the first request
+}
+
 type BackendRR struct {
-	weight  int                 // weight of this backend
-	current int                 // current weight
-	backend *backend.BfeBackend // point to BfeBackend
+	weight       int                 // weight of this backend
+	current      int                 // current weight
+	backend      *backend.BfeBackend // point to BfeBackend
+	inSlowStart  bool                // indicate if in slow-start phase
+	weightSS     WeightSS            // slow_start related parameters
 }
 
 func NewBackendRR() *BackendRR {
@@ -36,15 +48,17 @@ func NewBackendRR() *BackendRR {
 
 // Init initialize BackendRR with BackendConf
 func (backRR *BackendRR) Init(subClusterName string, conf *cluster_table_conf.BackendConf) {
-	backRR.weight = *conf.Weight
-	backRR.current = *conf.Weight
+	// scale up 100 times from conf file
+	backRR.weight = *conf.Weight * 100
+	backRR.current = backRR.weight
+	backRR.weightSS.final = backRR.weight
 
 	back := backRR.backend
 	back.Init(subClusterName, conf)
 }
 
 func (backRR *BackendRR) UpdateWeight(weight int) {
-	backRR.weight = weight
+	backRR.weight = weight * 100
 
 	// if weight > 0, don't touch backRR.current
 	if weight <= 0 {
@@ -60,3 +74,33 @@ func (backRR *BackendRR) MatchAddrPort(addr string, port int) bool {
 	back := backRR.backend
 	return back.Addr == addr && back.Port == port
 }
+
+func (backRR *BackendRR) initSlowStart(ssTime int) {
+	backRR.weightSS.slowStartTime = ssTime
+	if backRR.weightSS.slowStartTime == 0 {
+		backRR.inSlowStart = false
+	} else {
+		backRR.weightSS.startTime = time.Now()
+		backRR.inSlowStart = true
+
+		// set weight/current to 1, to avoid no traffic allowed at the beginning of start
+		backRR.weight = 1
+		backRR.current = 1
+	}
+}
+
+func (backRR *BackendRR) updateSlowStart() {
+	if backRR.inSlowStart {
+		current := time.Duration(backRR.weightSS.final) * time.Since(backRR.weightSS.startTime)
+		if backRR.weightSS.slowStartTime != 0 {
+			current /= time.Duration(backRR.weightSS.slowStartTime) * time.Second
+			backRR.weight = int(current)
+		} else {
+			backRR.weight = backRR.weightSS.final
+		}
+		if backRR.weight >= backRR.weightSS.final {
+			backRR.weight = backRR.weightSS.final
+			backRR.inSlowStart = false
+		}
+	}
+}

bfe_balance/bal_slb/backend_rr_test.go

@@ -36,12 +36,12 @@ func TestBackendRRInit_case1(t *testing.T) {
 	backendRR := NewBackendRR()
 	backendRR.Init("example.cluster", &conf)
 
-	if backendRR.weight != 10 {
-		t.Error("backend.weight should be 10")
+	if backendRR.weight != 10 * 100 {
+		t.Error("backend.weight should be 10 * 100")
 	}
 
-	if backendRR.current != 10 {
-		t.Error("backend.current should be 10")
+	if backendRR.current != 10 * 100 {
+		t.Error("backend.current should be 10 * 100")
 	}
 
 	backend := backendRR.backend

bfe_balance/bal_slb/bal_rr.go

@@ -89,10 +89,13 @@ func (s BackendListSorter) Less(i, j int) bool {
 
 type BalanceRR struct {
 	sync.Mutex
-	Name     string
-	backends BackendList // list of BackendRR
-	sorted   bool        // list of BackeneRR sorted or not
-	next     int         // next backend to schedule
+	Name          string
+	backends      BackendList // list of BackendRR
+	sorted        bool        // list of BackeneRR sorted or not
+	next          int         // next backend to schedule
+
+	slowStartNum  int         // number of backends in slow_start phase
+	slowStartTime int         // time for backend increases the weight to the full value, in seconds
 }
 
 func NewBalanceRR(name string) *BalanceRR {
@@ -113,6 +116,27 @@ func (brr *BalanceRR) Init(conf cluster_table_conf.SubClusterBackend) {
 	brr.next = 0
 }
 
+func (brr *BalanceRR) SetSlowStart(ssTime int) {
+	brr.Lock()
+	brr.slowStartTime = ssTime
+	brr.Unlock()
+}
+
+func (brr *BalanceRR) checkSlowStart() {
+	brr.Lock()
+	defer brr.Unlock()
+	if brr.slowStartTime > 0 {
+		for _, backendRR := range brr.backends {
+			backend := backendRR.backend
+			if backend.GetRestart() {
+				backend.SetRestart(false)
+				backendRR.initSlowStart(brr.slowStartTime)
+			}
+			backendRR.updateSlowStart()
+		}
+	}
+}
+
 // Release releases backend list.
 func (brr *BalanceRR) Release() {
 	for _, back := range brr.backends {
@@ -162,6 +186,8 @@ func (brr *BalanceRR) Update(conf cluster_table_conf.SubClusterBackend) {
 	for _, bkConf := range confMap {
 		backendRR := NewBackendRR()
 		backendRR.Init(brr.Name, bkConf)
+		backend := backendRR.backend
+		backend.SetRestart(true)
 		// add to backendsNew
 		backendsNew = append(backendsNew, backendRR)
 	}
@@ -195,6 +221,10 @@ func (brr *BalanceRR) ensureSortedUnlocked() {
 
 // Balance select one backend from sub cluster in round robin manner.
 func (brr *BalanceRR) Balance(algor int, key []byte) (*backend.BfeBackend, error) {
+	// Slow start is not supported when session sticky is enabled
+	if algor != WrrSticky {
+		brr.checkSlowStart()
+	}
 	switch algor {
 	case WrrSimple:
 		return brr.simpleBalance()