mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2024-11-09 23:24:05 -08:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Rules: Refactor concurrency controller interface (#14491)

Browse source 
* Rules: Refactor concurrency controller interface

Even though the main purpose of this refactor is to modify the interface of the concurrency controller to accept a Context. I did two drive-by modifications that I think are sensible:

1. I have moved the check for dependencies on rules to the controller itself - this aligns with how the controller should behave as it is a deciding factor on wether we should run concurrently or not.
2. I cleaned up some unused methods from the days of the old interface before #13527 changed it.

Signed-off-by: gotjosh <josue.abreu@gmail.com>
---------

Signed-off-by: gotjosh <josue.abreu@gmail.com>
This commit is contained in:
gotjosh 2024-07-22 14:11:18 +01:00 committed by GitHub
parent 398504e080
commit 465891cc56
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 20 additions and 46 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
rules/group.go
View file

@ -621,14 +621,12 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
} }
} }
// If the rule has no dependencies, it can run concurrently because no other rules in this group depend on its output. if ctrl := g.concurrencyController; ctrl.Allow(ctx, g, rule) {
// Try run concurrently if there are slots available.
if ctrl := g.concurrencyController; isRuleEligibleForConcurrentExecution(rule) && ctrl.Allow() {
wg.Add(1) wg.Add(1)
go eval(i, rule, func() { go eval(i, rule, func() {
wg.Done() wg.Done()
ctrl.Done() ctrl.Done(ctx)
}) })
} else { } else {
eval(i, rule, nil) eval(i, rule, nil)
@ -1094,7 +1092,3 @@ func buildDependencyMap(rules []Rule) dependencyMap {
return dependencies return dependencies
} }
func isRuleEligibleForConcurrentExecution(rule Rule) bool {
return rule.NoDependentRules() && rule.NoDependencyRules()
}

56
rules/manager.go
View file

@ -457,67 +457,47 @@ func (c ruleDependencyController) AnalyseRules(rules []Rule) {
// Its purpose is to bound the amount of concurrency in rule evaluations to avoid overwhelming the Prometheus // Its purpose is to bound the amount of concurrency in rule evaluations to avoid overwhelming the Prometheus
// server with additional query load. Concurrency is controlled globally, not on a per-group basis. // server with additional query load. Concurrency is controlled globally, not on a per-group basis.
type RuleConcurrencyController interface { type RuleConcurrencyController interface {
// Allow determines whether any concurrent evaluation slots are available. // Allow determines if the given rule is allowed to be evaluated concurrently.
// If Allow() returns true, then Done() must be called to release the acquired slot. // If Allow() returns true, then Done() must be called to release the acquired slot and corresponding cleanup is done.
Allow() bool // It is important that both *Group and Rule are not retained and only be used for the duration of the call.
Allow(ctx context.Context, group *Group, rule Rule) bool
// Done releases a concurrent evaluation slot. // Done releases a concurrent evaluation slot.
Done() Done(ctx context.Context)
} }
// concurrentRuleEvalController holds a weighted semaphore which controls the concurrent evaluation of rules. // concurrentRuleEvalController holds a weighted semaphore which controls the concurrent evaluation of rules.
type concurrentRuleEvalController struct { type concurrentRuleEvalController struct {
sema *semaphore.Weighted sema *semaphore.Weighted
depMapsMu sync.Mutex
depMaps map[*Group]dependencyMap
} }
func newRuleConcurrencyController(maxConcurrency int64) RuleConcurrencyController { func newRuleConcurrencyController(maxConcurrency int64) RuleConcurrencyController {
return &concurrentRuleEvalController{ return &concurrentRuleEvalController{
sema: semaphore.NewWeighted(maxConcurrency), sema: semaphore.NewWeighted(maxConcurrency),
depMaps: map[*Group]dependencyMap{},
} }
} }
func (c *concurrentRuleEvalController) RuleEligible(g *Group, r Rule) bool { func (c *concurrentRuleEvalController) Allow(_ context.Context, _ *Group, rule Rule) bool {
c.depMapsMu.Lock() // To allow a rule to be executed concurrently, we need 3 conditions:
defer c.depMapsMu.Unlock() // 1. The rule must not have any rules that depend on it.
// 2. The rule itself must not depend on any other rules.
depMap, found := c.depMaps[g] // 3. If 1 & 2 are true, then and only then we should try to acquire the concurrency slot.
if !found { if rule.NoDependentRules() && rule.NoDependencyRules() {
depMap = buildDependencyMap(g.rules) return c.sema.TryAcquire(1)
c.depMaps[g] = depMap
} }
return depMap.isIndependent(r) return false
} }
func (c *concurrentRuleEvalController) Allow() bool { func (c *concurrentRuleEvalController) Done(_ context.Context) {
return c.sema.TryAcquire(1)
}
func (c *concurrentRuleEvalController) Done() {
c.sema.Release(1) c.sema.Release(1)
} }
func (c *concurrentRuleEvalController) Invalidate() {
c.depMapsMu.Lock()
defer c.depMapsMu.Unlock()
// Clear out the memoized dependency maps because some or all groups may have been updated.
c.depMaps = map[*Group]dependencyMap{}
}
// sequentialRuleEvalController is a RuleConcurrencyController that runs every rule sequentially. // sequentialRuleEvalController is a RuleConcurrencyController that runs every rule sequentially.
type sequentialRuleEvalController struct{} type sequentialRuleEvalController struct{}
func (c sequentialRuleEvalController) RuleEligible(_ *Group, _ Rule) bool { func (c sequentialRuleEvalController) Allow(_ context.Context, _ *Group, _ Rule) bool {
return false return false
} }
func (c sequentialRuleEvalController) Allow() bool { func (c sequentialRuleEvalController) Done(_ context.Context) {}
return false
}
func (c sequentialRuleEvalController) Done() {}
func (c sequentialRuleEvalController) Invalidate() {}