Add support for prometheus rule query offset

Signed-off-by: Mustafain Ali Khan <[email protected]>

Author: mustafain117

pkg/ruler/compat.go

@@ -358,6 +358,9 @@ func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engi
 			ResendDelay:            cfg.ResendDelay,
 			ConcurrentEvalsEnabled: cfg.ConcurrentEvalsEnabled,
 			MaxConcurrentEvals:     cfg.MaxConcurrentEvals,
+			DefaultRuleQueryOffset: func() time.Duration {
+				return cfg.RuleQueryOffset
+			},
 		})
 	}
 }

pkg/ruler/ruler.go

@@ -121,6 +121,8 @@ type Config struct {
 	// Client configs for interacting with the Alertmanager
 	Notifier NotifierConfig `yaml:"alertmanager_client"`
 
+	//Default offset for all rule evaluation queries.
+	RuleQueryOffset time.Duration `yaml:"rule_query_offset"`
 	// Max time to tolerate outage for restoring "for" state of alert.
 	OutageTolerance time.Duration `yaml:"for_outage_tolerance"`
 	// Minimum duration between alert and restored "for" state. This is maintained only for alerts with configured "for" time greater than grace period.
@@ -194,6 +196,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.Var(&cfg.ExternalURL, "ruler.external.url", "URL of alerts return path.")
 	f.DurationVar(&cfg.EvaluationInterval, "ruler.evaluation-interval", 1*time.Minute, "How frequently to evaluate rules")
 	f.DurationVar(&cfg.PollInterval, "ruler.poll-interval", 1*time.Minute, "How frequently to poll for rule changes")
+	f.DurationVar(&cfg.RuleQueryOffset, "ruler.rule-query-offset", 0*time.Minute, "Default offset for all rule evaluation queries")
 
 	f.StringVar(&cfg.AlertmanagerURL, "ruler.alertmanager-url", "", "Comma-separated list of URL(s) of the Alertmanager(s) to send notifications to. Each Alertmanager URL is treated as a separate group in the configuration. Multiple Alertmanagers in HA per group can be supported by using DNS resolution via -ruler.alertmanager-discovery.")
 	f.BoolVar(&cfg.AlertmanagerDiscovery, "ruler.alertmanager-discovery", false, "Use DNS SRV records to discover Alertmanager hosts.")
@@ -909,11 +912,12 @@ func (r *Ruler) getLocalRules(userID string, rulesRequest RulesRequest, includeB
 
 		groupDesc := &GroupStateDesc{
 			Group: &rulespb.RuleGroupDesc{
-				Name:      group.Name(),
-				Namespace: string(decodedNamespace),
-				Interval:  interval,
-				User:      userID,
-				Limit:     int64(group.Limit()),
+				Name:        group.Name(),
+				Namespace:   string(decodedNamespace),
+				Interval:    interval,
+				User:        userID,
+				Limit:       int64(group.Limit()),
+				QueryOffset: group.QueryOffset(),
 			},
 
 			EvaluationTimestamp: group.GetLastEvaluation(),

pkg/ruler/ruler_test.go

@@ -2397,3 +2397,34 @@ func TestRulerDisablesRuleGroups(t *testing.T) {
 		})
 	}
 }
+
+func TestRuler_QueryOffset(t *testing.T) {
+	store := newMockRuleStore(mockRulesQueryOffset, nil)
+	cfg := defaultRulerConfig(t)
+
+	r := newTestRuler(t, cfg, store, nil)
+	defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck
+
+	ctx := user.InjectOrgID(context.Background(), "user1")
+	rls, err := r.Rules(ctx, &RulesRequest{})
+	require.NoError(t, err)
+	require.Len(t, rls.Groups, 1)
+	rg := rls.Groups[0]
+	expectedRg := mockRulesQueryOffset["user1"][0]
+	compareRuleGroupDescToStateDesc(t, expectedRg, rg)
+
+	// test default query offset=0
+	require.Equal(t, time.Duration(0), rg.GetGroup().QueryOffset)
+
+	ctx = user.InjectOrgID(context.Background(), "user2")
+	rls, err = r.Rules(ctx, &RulesRequest{})
+	require.NoError(t, err)
+	require.Len(t, rls.Groups, 1)
+	rg = rls.Groups[0]
+	expectedRg = mockRules["user2"][0]
+	compareRuleGroupDescToStateDesc(t, expectedRg, rg)
+
+	// test group query offset is set
+	require.Equal(t, time.Minute*2, rg.GetGroup().QueryOffset)
+	rulespb.FromProto(rg.GetGroup())
+}

pkg/ruler/rulespb/compat.go

@@ -14,12 +14,13 @@ import (
 // ToProto transforms a formatted prometheus rulegroup to a rule group protobuf
 func ToProto(user string, namespace string, rl rulefmt.RuleGroup) *RuleGroupDesc {
 	rg := RuleGroupDesc{
-		Name:      rl.Name,
-		Namespace: namespace,
-		Interval:  time.Duration(rl.Interval),
-		Rules:     formattedRuleToProto(rl.Rules),
-		User:      user,
-		Limit:     int64(rl.Limit),
+		Name:        rl.Name,
+		Namespace:   namespace,
+		Interval:    time.Duration(rl.Interval),
+		Rules:       formattedRuleToProto(rl.Rules),
+		User:        user,
+		Limit:       int64(rl.Limit),
+		QueryOffset: time.Duration(*rl.QueryOffset),
 	}
 	return &rg
 }
@@ -43,11 +44,13 @@ func formattedRuleToProto(rls []rulefmt.RuleNode) []*RuleDesc {
 
 // FromProto generates a rulefmt RuleGroup
 func FromProto(rg *RuleGroupDesc) rulefmt.RuleGroup {
+	queryOffset := model.Duration(rg.QueryOffset)
 	formattedRuleGroup := rulefmt.RuleGroup{
-		Name:     rg.GetName(),
-		Interval: model.Duration(rg.Interval),
-		Rules:    make([]rulefmt.RuleNode, len(rg.GetRules())),
-		Limit:    int(rg.Limit),
+		Name:        rg.GetName(),
+		Interval:    model.Duration(rg.Interval),
+		Rules:       make([]rulefmt.RuleNode, len(rg.GetRules())),
+		Limit:       int(rg.Limit),
+		QueryOffset: &queryOffset,
 	}
 
 	for i, rl := range rg.GetRules() {

pkg/ruler/rulespb/compat_test.go

@@ -29,14 +29,18 @@ func TestProto(t *testing.T) {
 
 	rules = append(rules, testRule)
 
+	queryOffset := model.Duration(30 * time.Second)
 	rg := rulefmt.RuleGroup{
-		Name:     "group1",
-		Rules:    rules,
-		Interval: model.Duration(time.Minute),
+		Name:        "group1",
+		Rules:       rules,
+		Interval:    model.Duration(time.Minute),
+		QueryOffset: &queryOffset,
 	}
+
 	desc := ToProto("test", "namespace", rg)
 
 	assert.Equal(t, len(rules), len(desc.Rules))
+	assert.Equal(t, 30*time.Second, desc.QueryOffset)
 
 	ruleDesc := desc.Rules[0]