Support resource level rule replacement in hotspot module (alibaba#373)

Author: luckyxiaoqiang

core/hotspot/rule_manager.go

@@ -35,7 +35,7 @@ var (
 	tcGenFuncMap  = make(map[ControlBehavior]TrafficControllerGenFunc, 4)
 	tcMap         = make(trafficControllerMap)
 	tcMux         = new(sync.RWMutex)
-	currentRules  = make([]*Rule, 0)
+	currentRules  = make(map[string][]*Rule, 0)
 	updateRuleMux = new(sync.Mutex)
 )
 
@@ -82,27 +82,36 @@ func getTrafficControllersFor(res string) []TrafficShapingController {
 	return tcMap[res]
 }
 
-// LoadRules replaces old rules with the given hotspot parameter flow control rules. Return value:
-//
-// bool: indicates whether the internal map has been changed;
-// error: indicates whether occurs the error.
+// LoadRules replaces all old hotspot param flow rules with the given rules.
+// Return value:
+//   bool: indicates whether the internal map has been changed;
+//   error: indicates whether occurs the error.
 func LoadRules(rules []*Rule) (bool, error) {
+	resRulesMap := make(map[string][]*Rule, 16)
+	for _, rule := range rules {
+		resRules, exists := resRulesMap[rule.Resource]
+		if !exists {
+			resRules = make([]*Rule, 0, 1)
+		}
+		resRulesMap[rule.Resource] = append(resRules, rule)
+	}
+
 	updateRuleMux.Lock()
 	defer updateRuleMux.Unlock()
-	isEqual := reflect.DeepEqual(currentRules, rules)
+	isEqual := reflect.DeepEqual(currentRules, resRulesMap)
 	if isEqual {
 		logging.Info("[HotSpot] Load rules is the same with current rules, so ignore load operation.")
 		return false, nil
 	}
 
-	err := onRuleUpdate(rules)
+	err := onRuleUpdate(resRulesMap)
 	return true, err
 }
 
-// GetRules returns all the rules based on copy.
-// It doesn't take effect for hotspot module if user changes the rule.
+// GetRules returns all the hotspot param flow rules based on copy.
+// It doesn't take effect for hotspot module if user changes the returned rules.
 // GetRules need to compete hotspot module's global lock and the high performance losses of copy,
-// 		reduce or do not call GetRules if possible
+// 		reduce or do not call GetRules if possible.
 func GetRules() []Rule {
 	tcMux.RLock()
 	rules := rulesFrom(tcMap)
@@ -115,10 +124,10 @@ func GetRules() []Rule {
 	return ret
 }
 
-// GetRulesOfResource returns specific resource's rules based on copy.
-// It doesn't take effect for hotspot module if user changes the rule.
+// GetRulesOfResource returns specific resource's hotspot parameter flow control rules based on copy.
+// It doesn't take effect for hotspot module if user changes the returned rules.
 // GetRulesOfResource need to compete hotspot module's global lock and the high performance losses of copy,
-// 		reduce or do not call GetRulesOfResource frequently if possible
+// 		reduce or do not call GetRulesOfResource frequently if possible.
 func GetRulesOfResource(res string) []Rule {
 	tcMux.RLock()
 	resTcs := tcMap[res]
@@ -131,13 +140,19 @@ func GetRulesOfResource(res string) []Rule {
 	return ret
 }
 
-// ClearRules clears all parameter flow rules.
+// ClearRules clears all hotspot param flow rules.
 func ClearRules() error {
 	_, err := LoadRules(nil)
 	return err
 }
 
-func onRuleUpdate(rules []*Rule) (err error) {
+// ClearRulesOfResource clears resource level hotspot param flow rules.
+func ClearRulesOfResource(res string) error {
+	_, err := LoadRulesOfResource(res, nil)
+	return err
+}
+
+func onRuleUpdate(rawResRulesMap map[string][]*Rule) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			var ok bool
@@ -148,24 +163,20 @@ func onRuleUpdate(rules []*Rule) (err error) {
 		}
 	}()
 
-	newRuleMap := make(map[string][]*Rule, len(rules))
-	for _, r := range rules {
-		if err := IsValidRule(r); err != nil {
-			logging.Warn("[HotSpot onRuleUpdate] Ignoring invalid hotspot rule when loading new rules", "rule", r, "err", err.Error())
-			continue
+	// ignore invalid rules
+	validResRulesMap := make(map[string][]*Rule, len(rawResRulesMap))
+	for res, rules := range rawResRulesMap {
+		validResRules := make([]*Rule, 0, len(rules))
+		for _, rule := range rules {
+			if err := IsValidRule(rule); err != nil {
+				logging.Warn("[HotSpot onRuleUpdate] Ignoring invalid hotspot param flow rule when loading new rules", "rule", rule, "err", err.Error())
+				continue
+			}
+			validResRules = append(validResRules, rule)
 		}
-		res := r.ResourceName()
-		ruleSet, ok := newRuleMap[res]
-		if !ok {
-			ruleSet = make([]*Rule, 0, 1)
+		if len(validResRules) > 0 {
+			validResRulesMap[res] = validResRules
 		}
-		ruleSet = append(ruleSet, r)
-		newRuleMap[res] = ruleSet
-	}
-
-	m := make(trafficControllerMap)
-	for res, rules := range newRuleMap {
-		m[res] = make([]TrafficShapingController, 0, len(rules))
 	}
 
 	start := util.CurrentTimeNano()
@@ -179,49 +190,11 @@ func onRuleUpdate(rules []*Rule) (err error) {
 	}
 	tcMux.RUnlock()
 
-	for res, resRules := range newRuleMap {
-		emptyTcList := make([]TrafficShapingController, 0, 0)
-		for _, r := range resRules {
-			oldResTcs := tcMapClone[res]
-			if oldResTcs == nil {
-				oldResTcs = emptyTcList
-			}
-
-			equalIdx, reuseStatIdx := calculateReuseIndexFor(r, oldResTcs)
-			// there is equivalent rule in old traffic shaping controller slice
-			if equalIdx >= 0 {
-				equalOldTC := oldResTcs[equalIdx]
-				insertTcToTcMap(equalOldTC, res, m)
-				// remove old tc from old resTcs
-				tcMapClone[res] = append(oldResTcs[:equalIdx], oldResTcs[equalIdx+1:]...)
-				continue
-			}
-
-			// generate new traffic shaping controller
-			generator, supported := tcGenFuncMap[r.ControlBehavior]
-			if !supported {
-				logging.Warn("[HotSpot onRuleUpdate] Ignoring the frequent param flow rule due to unsupported control behavior", "rule", r)
-				continue
-			}
-			var tc TrafficShapingController
-			if reuseStatIdx >= 0 {
-				// generate new traffic shaping controller with reusable statistic metric.
-				tc = generator(r, oldResTcs[reuseStatIdx].BoundMetric())
-			} else {
-				tc = generator(r, nil)
-			}
-			if tc == nil {
-				logging.Debug("[HotSpot onRuleUpdate] Ignoring the frequent param flow rule due to bad generated traffic controller", "rule", r)
-				continue
-			}
-
-			//  remove the reused traffic shaping controller old res tcs
-			if reuseStatIdx >= 0 {
-				tcMapClone[res] = append(oldResTcs[:reuseStatIdx], oldResTcs[reuseStatIdx+1:]...)
-			}
-			insertTcToTcMap(tc, res, m)
-		}
+	m := make(trafficControllerMap, len(validResRulesMap))
+	for res, rules := range validResRulesMap {
+		m[res] = buildResourceTrafficShapingController(res, rules, tcMapClone[res])
 	}
+
 	for res, tcs := range m {
 		if len(tcs) > 0 {
 			// update resource slot chain
@@ -237,14 +210,102 @@ func onRuleUpdate(rules []*Rule) (err error) {
 
 	tcMux.Lock()
 	tcMap = m
-	currentRules = rules
 	tcMux.Unlock()
 
-	logging.Debug("[HotSpot onRuleUpdate] Time statistic(ns) for updating hotSpot rule", "timeCost", util.CurrentTimeNano()-start)
-	logRuleUpdate(newRuleMap)
+	currentRules = rawResRulesMap
+
+	logging.Debug("[HotSpot onRuleUpdate] Time statistic(ns) for updating hotspot param flow rules", "timeCost", util.CurrentTimeNano()-start)
+	logRuleUpdate(validResRulesMap)
+	return nil
+}
+
+func onResourceRuleUpdate(res string, rawResRules []*Rule) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			var ok bool
+			err, ok = r.(error)
+			if !ok {
+				err = fmt.Errorf("%v", r)
+			}
+		}
+	}()
+
+	validResRules := make([]*Rule, 0, len(rawResRules))
+	for _, rule := range rawResRules {
+		if err := IsValidRule(rule); err != nil {
+			logging.Warn("[HotSpot onResourceRuleUpdate] Ignoring invalid hotspot param flow rule", "rule", rule, "reason", err.Error())
+			continue
+		}
+		validResRules = append(validResRules, rule)
+	}
+
+	start := util.CurrentTimeNano()
+	oldResTcs := make([]TrafficShapingController, 0, 8)
+	tcMux.RLock()
+	oldResTcs = append(oldResTcs, tcMap[res]...)
+	tcMux.RUnlock()
+
+	newResTcs := buildResourceTrafficShapingController(res, validResRules, oldResTcs)
+	if len(newResTcs) > 0 {
+		// update resource slot chain
+		misc.RegisterRuleCheckSlotForResource(res, DefaultSlot)
+		for _, tc := range newResTcs {
+			if tc.BoundRule().MetricType == Concurrency {
+				misc.RegisterStatSlotForResource(res, DefaultConcurrencyStatSlot)
+				break
+			}
+		}
+	}
+
+	tcMux.Lock()
+	if len(newResTcs) == 0 {
+		delete(tcMap, res)
+	} else {
+		tcMap[res] = newResTcs
+	}
+	tcMux.Unlock()
+
+	currentRules[res] = rawResRules
+
+	logging.Debug("[HotSpot onResourceRuleUpdate] Time statistic(ns) for updating hotspot param flow rules", "timeCost", util.CurrentTimeNano()-start)
+	logging.Info("[HotSpot] load resource level hotspot param flow rules", "resource", res, "validResRules", validResRules)
 	return nil
 }
 
+// LoadRulesOfResource loads the given resource's hotspot param flow rules to the rule manager,
+// while all previous resource's rules will be replaced. The first returned value indicates whether
+// do real load operation, if the rules is the same with previous resource's rules, return false.
+func LoadRulesOfResource(res string, rules []*Rule) (bool, error) {
+	if len(res) == 0 {
+		return false, errors.New("empty resource")
+	}
+
+	updateRuleMux.Lock()
+	defer updateRuleMux.Unlock()
+
+	// clear resource rules
+	if len(rules) == 0 {
+		// clear resource's currentRules
+		delete(currentRules, res)
+		// clear tcMap
+		tcMux.Lock()
+		delete(tcMap, res)
+		tcMux.Unlock()
+		logging.Info("[HotSpot] clear resource level hotspot param flow rules", "resource", res)
+		return true, nil
+	}
+
+	// load resource level rules
+	isEqual := reflect.DeepEqual(currentRules[res], rules)
+	if isEqual {
+		logging.Info("[HotSpot] Load resource level hotspot param flow rules is the same with current resource level rules, so ignore load operation.")
+		return false, nil
+	}
+
+	err := onResourceRuleUpdate(res, rules)
+	return true, err
+}
+
 func logRuleUpdate(m map[string][]*Rule) {
 	rules := make([]*Rule, 0, 8)
 	for _, rs := range m {
@@ -254,9 +315,9 @@ func logRuleUpdate(m map[string][]*Rule) {
 		rules = append(rules, rs...)
 	}
 	if len(rules) == 0 {
-		logging.Info("[HotspotRuleManager] Hotspot rules were cleared")
+		logging.Info("[HotspotRuleManager] Hotspot param flow rules were cleared")
 	} else {
-		logging.Info("[HotspotRuleManager] Hotspot rules were loaded", "rules", rules)
+		logging.Info("[HotspotRuleManager] Hotspot param flow rules were loaded", "rules", rules)
 	}
 }
 
@@ -304,14 +365,48 @@ func calculateReuseIndexFor(r *Rule, oldResTcs []TrafficShapingController) (equa
 	return equalIdx, reuseStatIdx
 }
 
-func insertTcToTcMap(tc TrafficShapingController, res string, m trafficControllerMap) {
-	tcsOfRes, exists := m[res]
-	if !exists {
-		tcsOfRes = make([]TrafficShapingController, 0, 1)
-		m[res] = append(tcsOfRes, tc)
-	} else {
-		m[res] = append(tcsOfRes, tc)
+// buildResourceTrafficShapingController builds TrafficShapingController slice from rules. the resource of rules must be equals to res.
+func buildResourceTrafficShapingController(res string, resRules []*Rule, oldResTcs []TrafficShapingController) []TrafficShapingController {
+	newTcsOfRes := make([]TrafficShapingController, 0, len(resRules))
+	for _, rule := range resRules {
+		if res != rule.Resource {
+			logging.Error(errors.Errorf("unmatched resource name, expect: %s, actual: %s", res, rule.Resource), "Unmatched resource name in hotspot.buildResourceTrafficShapingController()", "rule", rule)
+			continue
+		}
+
+		equalIdx, reuseStatIdx := calculateReuseIndexFor(rule, oldResTcs)
+		// there is equivalent rule in old traffic shaping controller slice
+		if equalIdx >= 0 {
+			equalOldTC := oldResTcs[equalIdx]
+			newTcsOfRes = append(newTcsOfRes, equalOldTC)
+			// remove old tc from old resTcs
+			oldResTcs = append(oldResTcs[:equalIdx], oldResTcs[equalIdx+1:]...)
+			continue
+		}
+
+		// generate new traffic shaping controller
+		generator, supported := tcGenFuncMap[rule.ControlBehavior]
+		if !supported {
+			logging.Warn("[HotSpot buildResourceTrafficShapingController] Ignoring the hotspot param flow rule due to unsupported control behavior", "rule", rule)
+			continue
+		}
+		var tc TrafficShapingController
+		if reuseStatIdx >= 0 {
+			// generate new traffic shaping controller with reusable statistic metric.
+			tc = generator(rule, oldResTcs[reuseStatIdx].BoundMetric())
+			// remove the reused traffic shaping controller old res tcs
+			oldResTcs = append(oldResTcs[:reuseStatIdx], oldResTcs[reuseStatIdx+1:]...)
+		} else {
+			tc = generator(rule, nil)
+		}
+		if tc == nil {
+			logging.Debug("[HotSpot buildResourceTrafficShapingController] Ignoring the hotspot param flow rule due to bad generated traffic controller", "rule", rule)
+			continue
+		}
+
+		newTcsOfRes = append(newTcsOfRes, tc)
 	}
+	return newTcsOfRes
 }
 
 func IsValidRule(rule *Rule) error {