Register xfunctions in global parser for rule file loading

Signed-off-by: Paurush Garg <paurushg@amazon.com>

Author: PaurushGarg

CHANGELOG.md

@@ -55,8 +55,9 @@
 * [BUGFIX] Ring: Fix ring token conflict resolution only applied to updated instance and make constantly token conflict check during instance observe period.
 * [BUGFIX] Distributor: Fix a panic (`slice bounds out of range`) in the stream push path when the context deadline expires while the worker goroutine is still marshalling a `WriteRequest`. #7541
 * [BUGFIX] Query Frontend: Fix native histogram responses not being handled correctly in `minTime()` sort ordering for split_by_interval merge. #7555
-* [BUGFIX] Distributor: Release the push worker pool goroutines on shutdown by stopping the async executor during the stopping phase when `-distributor.num-push-workers` is set. #7602
 * [BUGFIX] Querier: Fix unbounded resource leak in the bucket-scan blocks finder (used when the bucket index is disabled). Per-tenant metadata fetchers, their Prometheus registries, and on-disk meta caches are now evicted once a tenant is no longer active, instead of being retained for the lifetime of the process. #7573
+* [BUGFIX] Distributor: Release the push worker pool goroutines on shutdown by stopping the async executor during the stopping phase when `-distributor.num-push-workers` is set. #7602
+* [BUGFIX] Ruler: Register xfunctions (xincrease, xrate, xdelta) in the global parser before loading rule files. #7621
 
 ## 1.21.0 2026-04-24
 

integration/ruler_test.go

@@ -1887,3 +1887,57 @@ func createTestRuleGroup(t *testing.T) rulefmt.RuleGroup {
 		},
 	}
 }
+
+func TestRulerXFunctionsWithThanosEngine(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, bucketName, rulestoreBucketName)
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	flags := mergeFlags(
+		BlocksStorageFlags(),
+		RulerFlags(),
+		map[string]string{
+			"-querier.thanos-engine":           "true",
+			"-querier.enable-x-functions":      "true",
+			"-ruler.evaluation-interval":       "2s",
+			"-ruler.poll-interval":             "2s",
+			"-ruler.evaluation-delay-duration": "0",
+			"-distributor.replication-factor":  "1",
+		},
+	)
+
+	const namespace = "test"
+	const user = "user-1"
+
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ruler := e2ecortex.NewRuler("ruler", consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, ruler))
+
+	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, ruler.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", ruler.HTTPEndpoint(), user)
+	require.NoError(t, err)
+
+	series, _ := generateSeries("metric_total", time.Now(), prompb.Label{Name: "job", Value: "test"})
+	res, err := c.Push(series)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	ruleGroup := ruleGroupWithRule("xfunctions_group", "xincrease_rule", `xincrease(metric_total{job="test"}[5m])`)
+	require.NoError(t, c.SetRuleGroup(ruleGroup, namespace))
+
+	m := ruleGroupMatcher(user, namespace, "xfunctions_group")
+
+	// Wait until ruler has loaded the rule group.
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_prometheus_rule_group_rules"}, e2e.WithLabelMatchers(m), e2e.WaitMissingMetrics))
+
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_prometheus_rule_evaluations_total"}, e2e.WithLabelMatchers(m), e2e.WaitMissingMetrics))
+
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_prometheus_rule_evaluation_failures_total"}, e2e.WithLabelMatchers(m), e2e.WaitMissingMetrics))
+}

pkg/cortex/modules.go

@@ -5,6 +5,7 @@ import (
 	"flag"
 	"fmt"
 	"log/slog"
+	"maps"
 	"net/http"
 	"runtime"
 	"runtime/debug"
@@ -18,9 +19,11 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/promql"
+	"github.com/prometheus/prometheus/promql/parser"
 	"github.com/prometheus/prometheus/rules"
 	prom_storage "github.com/prometheus/prometheus/storage"
 	"github.com/thanos-io/objstore"
+	"github.com/thanos-io/promql-engine/execution/parse"
 	"github.com/thanos-io/thanos/pkg/discovery/dns"
 	"github.com/thanos-io/thanos/pkg/querysharding"
 	httpgrpc_server "github.com/weaveworks/common/httpgrpc/server"
@@ -661,6 +664,11 @@ func (t *Cortex) initRulerStorage() (serv services.Service, err error) {
 		return
 	}
 
+	// Register xfunctions (xincrease, xrate, xdelta) in the global parser
+	if t.Cfg.Querier.ThanosEngine.Enabled && t.Cfg.Querier.ThanosEngine.EnableXFunctions {
+		maps.Copy(parser.Functions, parse.XFunctions)
+	}
+
 	t.RulerStorage, err = ruler.NewRuleStore(context.Background(), t.Cfg.RulerStorage, t.OverridesConfig, rules.FileLoader{}, util_log.Logger, prometheus.DefaultRegisterer, t.Cfg.NameValidationScheme)
 	return
 }

pkg/cortex/modules_test.go

@@ -2,6 +2,7 @@ package cortex
 
 import (
 	"context"
+	"maps"
 	"net/http/httptest"
 	"os"
 	"reflect"
@@ -10,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/gorilla/mux"
+	"github.com/prometheus/prometheus/promql/parser"
 	prom_storage "github.com/prometheus/prometheus/storage"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -169,6 +171,75 @@ func (p *myPusher) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 	return nil, nil
 }
 
+func TestCortex_InitRulerStorage_RegistersXFunctions(t *testing.T) {
+	tests := map[string]struct {
+		enabled          bool
+		enableXFunctions bool
+		expectRegistered bool
+	}{
+		"should register xfunctions when thanos engine is enabled and EnableXFunctions is true": {
+			enabled:          true,
+			enableXFunctions: true,
+			expectRegistered: true,
+		},
+		"should not register xfunctions when EnableXFunctions is false": {
+			enabled:          true,
+			enableXFunctions: false,
+			expectRegistered: false,
+		},
+		"should not register xfunctions when thanos engine is disabled": {
+			enabled:          false,
+			enableXFunctions: true,
+			expectRegistered: false,
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			// Clean up global state after each test
+			originalFunctions := make(map[string]*parser.Function, len(parser.Functions))
+			maps.Copy(originalFunctions, parser.Functions)
+			defer func() {
+				// Restore original parser.Functions
+				for k := range parser.Functions {
+					if _, ok := originalFunctions[k]; !ok {
+						delete(parser.Functions, k)
+					}
+				}
+			}()
+
+			cfg := newDefaultConfig()
+			cfg.Target = []string{"ruler"}
+			cfg.RulerStorage.Backend = "local"
+			cfg.RulerStorage.Local.Directory = os.TempDir()
+			cfg.Querier.ThanosEngine.Enabled = testData.enabled
+			cfg.Querier.ThanosEngine.EnableXFunctions = testData.enableXFunctions
+
+			cortex := &Cortex{
+				Server: &server.Server{},
+				Cfg:    *cfg,
+			}
+
+			_, err := cortex.initRulerStorage()
+			require.NoError(t, err)
+
+			_, hasXincrease := parser.Functions["xincrease"]
+			_, hasXrate := parser.Functions["xrate"]
+			_, hasXdelta := parser.Functions["xdelta"]
+
+			if testData.expectRegistered {
+				assert.True(t, hasXincrease, "xincrease should be registered")
+				assert.True(t, hasXrate, "xrate should be registered")
+				assert.True(t, hasXdelta, "xdelta should be registered")
+			} else {
+				assert.False(t, hasXincrease, "xincrease should not be registered")
+				assert.False(t, hasXrate, "xrate should not be registered")
+				assert.False(t, hasXdelta, "xdelta should not be registered")
+			}
+		})
+	}
+}
+
 type myQueryable struct{}
 
 func (q *myQueryable) Querier(mint, maxt int64) (prom_storage.Querier, error) {