Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@
* [ENHANCEMENT] Upgrade Thanos and promql-engine to latest. #7740
* [ENHANCEMENT] Ruler: Adjust ruler frontend decoder to not wrap query error messages with execution prefix, this makes error responses consistent between internal and external ruler paths. #7741
* [ENHANCEMENT] Distributor: Deduplicate metric metadata when converting PRW 2.0 requests. PRW 2.0 attaches metadata to every series, so a metric family was previously expanded into one `MetricMetadata` per series. #7760
* [ENHANCEMENT] Ruler: Add new limit `-ruler.list-rules-max-rules` on the total number of rules returned by the Prometheus ListRules API. Responses exceeding the limit are truncated on a rule group boundary and return a `groupNextToken` for retrieving the remaining groups. A rule group is never split, so a single group larger than the limit is still returned whole. Defaults to 0, which is unlimited. #7785
* [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
* [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
* [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389
Expand Down
8 changes: 8 additions & 0 deletions docs/configuration/config-file-reference.md
Original file line number Diff line number Diff line change
Expand Up @@ -6242,6 +6242,14 @@ ring:
# CLI flag: -ruler.disabled-tenants
[disabled_tenants: <string> | default = ""]

# Maximum number of rules returned by the Prometheus ListRules API. If there are
# more rulegroups, the response will include a pagination token which can be
# used to fetch the next set. The API will always return at least one rulegroup,
# even if it contains more rules than the limit. Defaults to 0, which is
# unlimited
# CLI flag: -ruler.list-rules-max-rules
[list_rules_max_rules: <int> | default = 0]

# Report query statistics for ruler queries to complete as a per user metric and
# as an info level log message.
# CLI flag: -ruler.query-stats-enabled
Expand Down
4 changes: 2 additions & 2 deletions pkg/ruler/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,7 @@ func (a *API) PrometheusRules(w http.ResponseWriter, req *http.Request) {
}

w.Header().Set("Content-Type", "application/json")
response, err := a.ruler.GetRules(req.Context(), rulesRequest)
response, err := a.ruler.GetRules(req.Context(), rulesRequest, a.ruler.cfg.ListRulesMaxRules)

if err != nil {
util_api.RespondError(logger, w, v1.ErrServer, err.Error(), http.StatusInternalServerError)
Expand Down Expand Up @@ -346,7 +346,7 @@ func (a *API) PrometheusAlerts(w http.ResponseWriter, req *http.Request) {
Type: alertingRuleFilter,
MaxRuleGroups: -1,
}
rulesResponse, err := a.ruler.GetRules(req.Context(), rulesRequest)
rulesResponse, err := a.ruler.GetRules(req.Context(), rulesRequest, 0)

if err != nil {
util_api.RespondError(logger, w, v1.ErrServer, err.Error(), http.StatusInternalServerError)
Expand Down
130 changes: 130 additions & 0 deletions pkg/ruler/api_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import (

"github.com/go-kit/log"
"github.com/gorilla/mux"
v1 "github.com/prometheus/client_golang/api/prometheus/v1"
"github.com/stretchr/testify/require"
"github.com/weaveworks/common/user"

Expand Down Expand Up @@ -426,6 +427,135 @@ func TestRuler_rules_limit(t *testing.T) {
require.JSONEq(t, string(expectedResponse), string(actual))
}

func TestRuler_rules_max_rules(t *testing.T) {
// mockRulesNamespaces gives user1 two groups of two rules each: namespace2/fail
// sorts before namespace1/group1 in rule group token order, so it is the group
// that survives truncation. Filters and pagination vary how many rules a single
// request returns.
for _, tc := range []struct {
name string
rules map[string]rulespb.RuleGroupList
userID string
maxRules uint
queryParams string
expectedGroups int
expectedRules int
expectedNextToken string
}{
{
name: "limit disabled returns every rule",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 0,
expectedGroups: 2,
expectedRules: 4,
},
{
name: "count below the limit",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 10,
expectedGroups: 2,
expectedRules: 4,
},
{
name: "count exactly at the limit",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 4,
expectedGroups: 2,
expectedRules: 4,
},
{
name: "count over the limit is truncated, summed across groups",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 3,
expectedGroups: 1,
expectedRules: 2,
expectedNextToken: GetRuleGroupNextToken("namespace2", "fail"),
},
{
name: "only the rules matching the filters are counted",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 3,
queryParams: "?type=alert",
expectedGroups: 2,
expectedRules: 2,
},
{
name: "paginating below the limit succeeds",
rules: mockRulesNamespaces,
userID: "user1",
maxRules: 3,
queryParams: "?group_limit=1",
expectedGroups: 1,
expectedRules: 2,
expectedNextToken: GetRuleGroupNextToken("namespace2", "fail"),
},
{
// A rule group is indivisible, so the response is allowed to exceed the
// limit rather than return an unusable empty page. It is the only group, so
// there is nothing left to page to and no token is returned.
name: "a lone rule group over the limit is returned whole",
rules: mockRules,
userID: "user1",
maxRules: 1,
expectedGroups: 1,
expectedRules: 2,
},
{
name: "limit applies per request: tenant under the limit",
rules: mockRules,
userID: "user2",
maxRules: 1,
expectedGroups: 1,
expectedRules: 1,
},
} {
t.Run(tc.name, func(t *testing.T) {
cfg := defaultRulerConfig(t)
cfg.ListRulesMaxRules = tc.maxRules

r := newTestRuler(t, cfg, newMockRuleStore(tc.rules, nil), nil)
defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck

a := NewAPI(r, r.store, log.NewNopLogger())

req := requestFor(t, http.MethodGet, "https://localhost:8080/api/prom/api/v1/rules"+tc.queryParams, nil, tc.userID)
w := httptest.NewRecorder()
a.PrometheusRules(w, req)

resp := w.Result()
body, err := io.ReadAll(resp.Body)
require.NoError(t, err)

// util_api.Response holds Data as an any, so decode into a shape that
// exposes the rule groups directly.
parsed := struct {
Status string `json:"status"`
Data RuleDiscovery `json:"data"`
ErrorType v1.ErrorType `json:"errorType"`
Error string `json:"error"`
}{}
require.NoError(t, json.Unmarshal(body, &parsed))

require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, "success", parsed.Status)
require.Empty(t, parsed.Error)
require.Len(t, parsed.Data.RuleGroups, tc.expectedGroups)

rulesCount := 0
for _, g := range parsed.Data.RuleGroups {
rulesCount += len(g.Rules)
}
require.Equal(t, tc.expectedRules, rulesCount)
require.Equal(t, tc.expectedNextToken, parsed.Data.GroupNextToken)
})
}
}

func TestRuler_alerts(t *testing.T) {
store := newMockRuleStore(mockRules, nil)
cfg := defaultRulerConfig(t)
Expand Down
8 changes: 4 additions & 4 deletions pkg/ruler/merger.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@ import (
// mergeGroupStateDesc removes duplicates from the provided []*GroupStateDesc by keeping the GroupStateDesc with the
// latest information. It uses the EvaluationTimestamp of the GroupStateDesc and the EvaluationTimestamp of the
// ActiveRules in a GroupStateDesc to determine the which GroupStateDesc has the latest information.
// It also truncates rule groups if maxRuleGroups > 0
func mergeGroupStateDesc(ruleResponses []*RulesResponse, maxRuleGroups int32, dedup bool) *RulesResponse {
// It also truncates rule groups if maxRuleGroups > 0 or maxRules > 0
func mergeGroupStateDesc(ruleResponses []*RulesResponse, maxRuleGroups int32, maxRules uint, dedup bool) *RulesResponse {

var groupsStateDescs []*GroupStateDesc

Expand Down Expand Up @@ -44,10 +44,10 @@ func mergeGroupStateDesc(ruleResponses []*RulesResponse, maxRuleGroups int32, de
groups = groupsStateDescs
}

if maxRuleGroups > 0 {
if maxRuleGroups > 0 || maxRules > 0 {
//Need to sort here before we truncate
sort.Sort(PaginatedGroupStates(groups))
result, nextToken := generatePage(groups, int(maxRuleGroups))
result, nextToken := generatePage(groups, int(maxRuleGroups), maxRules)
return &RulesResponse{
Groups: result,
NextToken: nextToken,
Expand Down
45 changes: 43 additions & 2 deletions pkg/ruler/merger_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ func TestMergeGroupStateDesc(t *testing.T) {
input []*RulesResponse
expectedOutput *RulesResponse
maxRuleGroups int32
maxRules uint
}

testCases := map[string]testCase{
Expand Down Expand Up @@ -189,20 +190,60 @@ func TestMergeGroupStateDesc(t *testing.T) {
},
maxRuleGroups: 1,
},
// gs1 (ns1/g1) sorts before gs2 (ns1/g2) in token order, and both hold two
// active rules, so a maxRules of 2 keeps gs1 and pages at gs1.
"maxRules truncates the merged page without maxRuleGroups": {
input: []*RulesResponse{
{
Groups: []*GroupStateDesc{&gs1, &gs2},
NextToken: "",
},
},
expectedOutput: &RulesResponse{
Groups: []*GroupStateDesc{&gs1},
NextToken: GetRuleGroupNextToken(gs1.Group.Namespace, gs1.Group.Name),
},
maxRules: 2,
},
"maxRules leaves the merged page intact when the total fits": {
input: []*RulesResponse{
{
Groups: []*GroupStateDesc{&gs1, &gs2},
NextToken: "",
},
},
expectedOutput: &RulesResponse{
Groups: []*GroupStateDesc{&gs1, &gs2},
NextToken: "",
},
maxRules: 4,
},
"both limits disabled returns everything untruncated": {
input: []*RulesResponse{
{
Groups: []*GroupStateDesc{&gs1, &gs2},
NextToken: "",
},
},
expectedOutput: &RulesResponse{
Groups: []*GroupStateDesc{&gs1, &gs2},
NextToken: "",
},
},
}

for name, tc := range testCases {
t.Run(name, func(t *testing.T) {

out := mergeGroupStateDesc(tc.input, tc.maxRuleGroups, true)
out := mergeGroupStateDesc(tc.input, tc.maxRuleGroups, tc.maxRules, true)
slices.SortFunc(out.Groups, func(a, b *GroupStateDesc) int {
fileCompare := strings.Compare(a.Group.Namespace, b.Group.Namespace)
if fileCompare != 0 {
return fileCompare
}
return strings.Compare(a.Group.Name, b.Group.Name)
})
require.Equal(t, int(tc.maxRuleGroups), len(out.Groups))
require.Len(t, out.Groups, len(tc.expectedOutput.Groups))
t.Log(tc.expectedOutput)
t.Log(out)
require.True(t, reflect.DeepEqual(tc.expectedOutput, out))
Expand Down
36 changes: 19 additions & 17 deletions pkg/ruler/ruler.go
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,8 @@ type Config struct {
EnabledTenants flagext.StringSliceCSV `yaml:"enabled_tenants"`
DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`

ListRulesMaxRules uint `yaml:"list_rules_max_rules"`

RingCheckPeriod time.Duration `yaml:"-"`

// Field will be populated during runtime.
Expand Down Expand Up @@ -268,6 +270,8 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
f.Var(&cfg.EnabledTenants, "ruler.enabled-tenants", "Comma separated list of tenants whose rules this ruler can evaluate. If specified, only these tenants will be handled by ruler, otherwise this ruler can process rules from all tenants. Subject to sharding.")
f.Var(&cfg.DisabledTenants, "ruler.disabled-tenants", "Comma separated list of tenants whose rules this ruler cannot evaluate. If specified, a ruler that would normally pick the specified tenant(s) for processing will ignore them instead. Subject to sharding.")

f.UintVar(&cfg.ListRulesMaxRules, "ruler.list-rules-max-rules", 0, "Maximum number of rules returned by the Prometheus ListRules API. If there are more rulegroups, the response will include a pagination token which can be used to fetch the next set. The API will always return at least one rulegroup, even if it contains more rules than the limit. Defaults to 0, which is unlimited")

f.BoolVar(&cfg.EnableQueryStats, "ruler.query-stats-enabled", false, "Report query statistics for ruler queries to complete as a per user metric and as an info level log message.")
f.BoolVar(&cfg.DisableRuleGroupLabel, "ruler.disable-rule-group-label", false, "Disable the rule_group label on exported metrics")

Expand Down Expand Up @@ -1192,14 +1196,14 @@ func (r *Ruler) filterBackupRuleGroups(userID string, ruleGroups []*rulespb.Rule

// GetRules retrieves the running rules from this ruler and all running rulers in the ring if
// sharding is enabled
func (r *Ruler) GetRules(ctx context.Context, rulesRequest RulesRequest) (*RulesResponse, error) {
func (r *Ruler) GetRules(ctx context.Context, rulesRequest RulesRequest, maxRules uint) (*RulesResponse, error) {
userID, err := users.TenantID(ctx)
if err != nil {
return nil, fmt.Errorf("no user id found in context")
}

if r.cfg.EnableSharding {
resp, err := r.getShardedRules(ctx, userID, rulesRequest)
resp, err := r.getShardedRules(ctx, userID, rulesRequest, maxRules)
if resp == nil {
return &RulesResponse{
Groups: make([]*GroupStateDesc, 0),
Expand All @@ -1209,11 +1213,11 @@ func (r *Ruler) GetRules(ctx context.Context, rulesRequest RulesRequest) (*Rules
return resp, err
}

response, err := r.getLocalRules(userID, rulesRequest, false)
response, err := r.getLocalRules(userID, rulesRequest, maxRules, false)
return &response, err
}

func (r *Ruler) getLocalRules(userID string, rulesRequest RulesRequest, includeBackups bool) (RulesResponse, error) {
func (r *Ruler) getLocalRules(userID string, rulesRequest RulesRequest, maxRules uint, includeBackups bool) (RulesResponse, error) {
groups := r.manager.GetRules(userID)

groupDescs := make([]*GroupStateDesc, 0, len(groups))
Expand Down Expand Up @@ -1375,7 +1379,7 @@ func (r *Ruler) getLocalRules(userID string, rulesRequest RulesRequest, includeB
combinedRuleStateDescs = append(combinedRuleStateDescs, backupGroupDescs...)
}

if rulesRequest.MaxRuleGroups <= 0 {
if rulesRequest.MaxRuleGroups <= 0 && maxRules == 0 {
return RulesResponse{
Groups: combinedRuleStateDescs,
NextToken: "",
Expand All @@ -1398,7 +1402,7 @@ func (r *Ruler) getLocalRules(userID string, rulesRequest RulesRequest, includeB
}
}

resultingGroupDescs, nextToken := generatePage(resultingGroupDescs, int(rulesRequest.MaxRuleGroups))
resultingGroupDescs, nextToken := generatePage(resultingGroupDescs, int(rulesRequest.MaxRuleGroups), maxRules)
return RulesResponse{
Groups: resultingGroupDescs,
NextToken: nextToken,
Expand Down Expand Up @@ -1519,7 +1523,7 @@ func (r *Ruler) getShardSizeForUser(userID string) int {
return max(newShardSize, r.cfg.Ring.ReplicationFactor)
}

func (r *Ruler) getShardedRules(ctx context.Context, userID string, rulesRequest RulesRequest) (*RulesResponse, error) {
func (r *Ruler) getShardedRules(ctx context.Context, userID string, rulesRequest RulesRequest, maxRules uint) (*RulesResponse, error) {
ring := ring.ReadRing(r.ring)

if shardSize := r.limits.RulerTenantShardSize(userID); shardSize > 0 && r.cfg.ShardingStrategy == util.ShardingStrategyShuffle {
Expand Down Expand Up @@ -1598,17 +1602,15 @@ func (r *Ruler) getShardedRules(ctx context.Context, userID string, rulesRequest
return nil
})

if err == nil {
if r.cfg.RulesBackupEnabled() || r.cfg.APIDeduplicateRules {
return mergeGroupStateDesc(merged, rulesRequest.MaxRuleGroups, true), nil
}
return mergeGroupStateDesc(merged, rulesRequest.MaxRuleGroups, false), nil
if err != nil {
return &RulesResponse{
Groups: make([]*GroupStateDesc, 0),
NextToken: "",
}, err
}

return &RulesResponse{
Groups: make([]*GroupStateDesc, 0),
NextToken: "",
}, err
backup := r.cfg.RulesBackupEnabled() || r.cfg.APIDeduplicateRules
return mergeGroupStateDesc(merged, rulesRequest.MaxRuleGroups, maxRules, backup), nil
}

// Rules implements the rules service
Expand All @@ -1619,7 +1621,7 @@ func (r *Ruler) Rules(ctx context.Context, in *RulesRequest) (*RulesResponse, er
return nil, fmt.Errorf("no user id found in context")
}

response, err := r.getLocalRules(userID, *in, r.cfg.RulesBackupEnabled())
response, err := r.getLocalRules(userID, *in, 0, r.cfg.RulesBackupEnabled())
if err != nil {
return nil, err
}
Expand Down
Loading