rule: Add support for query offset (#8158)

Support Prometheus rule manager upstream "query offset" feature.
* Add support for a default rule query offset via command flag.
* Add per rule group query_offset support.

Fixes: #7596

Signed-off-by: SuperQ <[email protected]>

Author: SuperQ

CHANGELOG.md

@@ -25,6 +25,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#8017](https://github.com/thanos-io/thanos/pull/8017) Store Gateway: Use native histogram for binary reader load and download duration and fixed download duration metric. #8017
 - [#8131](https://github.com/thanos-io/thanos/pull/8131) Store Gateway: Optimize regex matchers for .* and .+. #8131
 - [#7808](https://github.com/thanos-io/thanos/pull/7808) Query: Support chain deduplication algorithm.
+- [#8158](https://github.com/thanos-io/thanos/pull/8158) Rule: Add support for query offset.
 
 ### Changed
 

cmd/thanos/rule.go

@@ -99,6 +99,7 @@ type ruleConfig struct {
 
 	resendDelay        time.Duration
 	evalInterval       time.Duration
+	queryOffset        time.Duration
 	outageTolerance    time.Duration
 	forGracePeriod     time.Duration
 	ruleFiles          []string
@@ -150,6 +151,8 @@ func registerRule(app *extkingpin.App) {
 		Default("1m").DurationVar(&conf.resendDelay)
 	cmd.Flag("eval-interval", "The default evaluation interval to use.").
 		Default("1m").DurationVar(&conf.evalInterval)
+	cmd.Flag("rule-query-offset", "The default rule group query_offset duration to use.").
+		Default("0s").DurationVar(&conf.queryOffset)
 	cmd.Flag("for-outage-tolerance", "Max time to tolerate prometheus outage for restoring \"for\" state of alert.").
 		Default("1h").DurationVar(&conf.outageTolerance)
 	cmd.Flag("for-grace-period", "Minimum duration between alert and restored \"for\" state. This is maintained only for alerts with configured \"for\" time greater than grace period.").
@@ -607,14 +610,15 @@ func runRule(
 		}
 
 		managerOpts := rules.ManagerOptions{
-			NotifyFunc:      notifyFunc,
-			Logger:          logutil.GoKitLogToSlog(logger),
-			Appendable:      appendable,
-			ExternalURL:     nil,
-			Queryable:       queryable,
-			ResendDelay:     conf.resendDelay,
-			OutageTolerance: conf.outageTolerance,
-			ForGracePeriod:  conf.forGracePeriod,
+			NotifyFunc:             notifyFunc,
+			Logger:                 logutil.GoKitLogToSlog(logger),
+			Appendable:             appendable,
+			ExternalURL:            nil,
+			Queryable:              queryable,
+			ResendDelay:            conf.resendDelay,
+			OutageTolerance:        conf.outageTolerance,
+			ForGracePeriod:         conf.forGracePeriod,
+			DefaultRuleQueryOffset: func() time.Duration { return conf.queryOffset },
 		}
 		if conf.ruleConcurrentEval > 1 {
 			managerOpts.MaxConcurrentEvals = conf.ruleConcurrentEval

docs/components/rule.md

@@ -18,6 +18,7 @@ The data of each Rule node can be labeled to satisfy the clusters labeling schem
 thanos rule \
     --data-dir             "/path/to/data" \
     --eval-interval        "30s" \
+    --rule-query-offset    "10s" \
     --rule-file            "/path/to/rules/*.rules.yaml" \
     --alert.query-url      "http://0.0.0.0:9090" \ # This tells what query URL to link to in UI.
     --alertmanagers.url    "http://alert.thanos.io" \
@@ -64,6 +65,9 @@ name: <string>
 # How often rules in the group are evaluated.
 [ interval: <duration> | default = global.evaluation_interval ]
 
+# Offset the rule evaluation timestamp of this particular group by the specified duration into the past.
+[ query_offset: <duration> | default = --rule-query-offset flag ]
+
 rules:
   [ - <rule> ... ]
 ```
@@ -471,6 +475,8 @@ Flags:
                                  Note that rules are not automatically detected,
                                  use SIGHUP or do HTTP POST /-/reload to re-read
                                  them.
+      --rule-query-offset=0s     The default rule group query_offset duration to
+                                 use.
       --shipper.meta-file-name="thanos.shipper.json"
                                  the file to store shipper metadata in
       --shipper.upload-compacted

test/e2e/rule_test.go

@@ -162,6 +162,16 @@ groups:
     expr: 1
 `
 
+	testRuleQueryOffset = `
+groups:
+- name: recording_query_offset
+  interval: 1s
+  query_offset: 1s
+  rules:
+  - record: metric_query_offset
+    expr: 1
+`
+
 	amTimeout = model.Duration(10 * time.Second)
 )
 
@@ -616,6 +626,25 @@ func TestRule_KeepFiringFor(t *testing.T) {
 		})
 	})
 
+	t.Run("rule query_offset", func(t *testing.T) {
+		// Create a recording rule that will add the missing metric
+		createRuleFile(t, filepath.Join(rulesPath, "record_metric_query_offset.yaml"), testRuleQueryOffset)
+		reloadRulesHTTP(t, ctx, r.Endpoint("http"))
+
+		// Wait for metric to pop up
+		queryWaitAndAssert(t, ctx, q.Endpoint("http"), func() string { return "metric_query_offset" }, time.Now, promclient.QueryOptions{
+			Deduplicate: false,
+		}, model.Vector{
+			&model.Sample{
+				Metric: model.Metric{
+					"__name__": "metric_query_offset",
+					"replica":  "1",
+				},
+				Value: model.SampleValue(1),
+			},
+		})
+	})
+
 	t.Run("keep_firing_for should continue triggering", func(t *testing.T) {
 		// Alert should still be firing
 		queryAndAssertSeries(t, ctx, q.Endpoint("http"), func() string { return "ALERTS" }, time.Now, promclient.QueryOptions{