Add source_tenants fields to RuleGroup

model/rulefmt/rulefmt.go

@@ -120,6 +120,7 @@ type RuleGroup struct {
 	Interval      model.Duration `yaml:"interval,omitempty"`
 	Limit         int            `yaml:"limit,omitempty"`
 	Rules         []RuleNode     `yaml:"rules"`
+	SourceTenants []string       `yaml:"source_tenants,omitempty"`
 }
 
 // Rule describes an alerting or recording rule.

model/rulefmt/testdata/test.yaml

@@ -36,6 +36,7 @@ groups:
 
   - name: my-another-name
     interval: 30s # defaults to global interval
+    source_tenants: [tenant-1]
     rules:
       - alert: HighErrors
         expr: |

rules/manager.go

@@ -246,6 +246,7 @@ type Group struct {
 	interval             time.Duration
 	limit                int
 	rules                []Rule
+	sourceTenants        []string
 	seriesInPreviousEval []map[string]labels.Labels // One per Rule.
 	staleSeries          []labels.Labels
 	opts                 *ManagerOptions
@@ -270,6 +271,7 @@ type GroupOptions struct {
 	Interval      time.Duration
 	Limit         int
 	Rules         []Rule
+	SourceTenants []string
 	ShouldRestore bool
 	Opts          *ManagerOptions
 	done          chan struct{}
@@ -301,6 +303,7 @@ func NewGroup(o GroupOptions) *Group {
 		rules:                o.Rules,
 		shouldRestore:        o.ShouldRestore,
 		opts:                 o.Opts,
+		sourceTenants:        o.SourceTenants,
 		seriesInPreviousEval: make([]map[string]labels.Labels, len(o.Rules)),
 		done:                 make(chan struct{}),
 		managerDone:          o.done,
@@ -325,6 +328,9 @@ func (g *Group) Interval() time.Duration { return g.interval }
 // Limit returns the group's limit.
 func (g *Group) Limit() int { return g.limit }
 
+// SourceTenants returns the source tenants for the group.
+func (g *Group) SourceTenants() []string { return g.sourceTenants }
+
 func (g *Group) run(ctx context.Context) {
 	defer close(g.terminated)
 
@@ -863,6 +869,16 @@ func (g *Group) Equals(ng *Group) bool {
 		return false
 	}
 
+	if len(g.sourceTenants) != len(ng.sourceTenants) {
+		return false
+	}
+
+	for i, tenant := range g.sourceTenants {
+		if ng.sourceTenants[i] != tenant {
+			return false
+		}
+	}
+
 	for i, gr := range g.rules {
 		if gr.String() != ng.rules[i].String() {
 			return false
@@ -887,12 +903,15 @@ type Manager struct {
 // NotifyFunc sends notifications about a set of alerts generated by the given expression.
 type NotifyFunc func(ctx context.Context, expr string, alerts ...*Alert)
 
+type ContextFunc func(g *Group) context.Context
+
 // ManagerOptions bundles options for the Manager.
 type ManagerOptions struct {
 	ExternalURL          *url.URL
 	QueryFunc            QueryFunc
 	NotifyFunc           NotifyFunc
 	Context              context.Context
+	FederatedContextFunc ContextFunc
 	Appendable           storage.Appendable
 	Queryable            storage.Queryable
 	Logger               log.Logger
@@ -992,11 +1011,16 @@ func (m *Manager) Update(interval time.Duration, files []string, externalLabels
 				newg.CopyState(oldg)
 			}
 			wg.Done()
+
+			ctx := m.opts.Context
+			if len(newg.sourceTenants) > 0 {
+				ctx = m.opts.FederatedContextFunc(newg)
+			}
 			// Wait with starting evaluation until the rule manager
 			// is told to run. This is necessary to avoid running
 			// queries against a bootstrapping storage.
 			<-m.block
-			newg.run(m.opts.Context)
+			newg.run(ctx)
 		}(newg)
 	}
 
@@ -1097,6 +1121,7 @@ func (m *Manager) LoadGroups(
 				Interval:      itv,
 				Limit:         rg.Limit,
 				Rules:         rules,
+				SourceTenants: rg.SourceTenants,
 				ShouldRestore: shouldRestore,
 				Opts:          m.opts,
 				done:          m.done,

rules/manager_test.go

@@ -722,6 +722,7 @@ func TestUpdate(t *testing.T) {
 		Queryable:            st,
 		QueryFunc:            EngineQueryFunc(engine, st),
 		Context:              context.Background(),
+		FederatedContextFunc: func(*Group) context.Context { return context.Background() },
 		Logger:               log.NewNopLogger(),
 	})
 	ruleManager.start()
@@ -787,6 +788,12 @@ func TestUpdate(t *testing.T) {
 		}
 	}
 	reloadAndValidate(rgs, t, tmpFile, ruleManager, expected, ogs)
+
+	// Change group source tenants and reload.
+	for i := range rgs.Groups {
+		rgs.Groups[i].SourceTenants = []string{"tenant-2"}
+	}
+	reloadAndValidate(rgs, t, tmpFile, ruleManager, expected, ogs)
 }
 
 // ruleGroupsTest for running tests over rules.
@@ -800,6 +807,7 @@ type ruleGroupTest struct {
 	Interval      model.Duration `yaml:"interval,omitempty"`
 	Limit         int            `yaml:"limit,omitempty"`
 	Rules         []rulefmt.Rule `yaml:"rules"`
+	SourceTenants []string       `yaml:"source_tenants,omitempty"`
 }
 
 func formatRules(r *rulefmt.RuleGroups) ruleGroupsTest {
@@ -822,6 +830,7 @@ func formatRules(r *rulefmt.RuleGroups) ruleGroupsTest {
 			Interval:      g.Interval,
 			Limit:         g.Limit,
 			Rules:         rtmp,
+			SourceTenants: g.SourceTenants,
 		})
 	}
 	return ruleGroupsTest{