diff --git a/CHANGELOG.md b/CHANGELOG.md index a04a1784e99..8815b415c8e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446 * [CHANGE] HA Tracker: Move `-distributor.ha-tracker.failover-timeout` from a global config to a per-tenant runtime config. The flag name and default value (30s) remain the same. #7481 +* [FEATURE] Ruler: Add experimental support for federated rule groups. A rule group listing tenants in its `src_tenants` field is evaluated against those tenants while the resulting series and alerts are written to the tenant owning the rule group. Enabled with `-ruler.enable-federated-rules` (requires `-tenant-federation.enabled`), and restricted to selected tenants with `-ruler.allowed-federated-tenants` and `-ruler.disallowed-federated-tenants`. #7828 * [FEATURE] Parquet: Support sharded parquet file conversion and querying. #7610 * [FEATURE] Parquet Converter: Add experimental `-parquet-converter.max-num-columns` flag to automatically shard parquet files when the number of columns exceeds the configured limit. This prevents failures when a TSDB block has more unique label names than the parquet library's column limit (32767). #7624 * [FEATURE] Distributor: Add experimental `-distributor.num-query-workers` flag to use a goroutine worker pool for query fan-out calls to ingesters. Reuses pre-grown goroutine stacks to eliminate the `runtime.copystack` overhead (~8% CPU) observed on rulers with wide ingester fan-out. Falls back to spawning a new goroutine when no worker is available. #7623 diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index 1321bb42b75..1e8adc7d2bd 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -6243,6 +6243,25 @@ ring: # CLI flag: -ruler.disabled-tenants [disabled_tenants: | default = ""] +# [Experimental] Enable federated rule groups. A federated rule group lists the +# tenants to query in its `src_tenants` field, while the resulting series and +# alerts always belong to the tenant owning the rule group. Requires +# -tenant-federation.enabled=true. +# CLI flag: -ruler.enable-federated-rules +[enable_federated_rules: | default = false] + +# [Experimental] Comma separated list of tenants allowed to create federated +# rule groups. If specified, only these tenants can create federated rule +# groups, otherwise all tenants can. +# CLI flag: -ruler.allowed-federated-tenants +[allowed_federated_tenants: | default = ""] + +# [Experimental] Comma separated list of tenants that cannot create federated +# rule groups. If specified, a tenant that would normally be allowed to create +# federated rule groups is denied instead. +# CLI flag: -ruler.disallowed-federated-tenants +[disallowed_federated_tenants: | default = ""] + # 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 diff --git a/docs/configuration/v1-guarantees.md b/docs/configuration/v1-guarantees.md index 0582882eb59..c704412b812 100644 --- a/docs/configuration/v1-guarantees.md +++ b/docs/configuration/v1-guarantees.md @@ -38,6 +38,7 @@ Currently experimental features are: - Ruler - Evaluate rules to query frontend instead of ingesters (enabled via `-ruler.frontend-address`). - When `-ruler.frontend-address` is specified, the response format can be specified (via `-ruler.query-response-format`). + - Federated rule groups (`-ruler.enable-federated-rules`, `-ruler.allowed-federated-tenants`, `-ruler.disallowed-federated-tenants`). - S3 Server Side Encryption (SSE) using KMS (including per-tenant KMS config overrides). - Alertmanager: - Receiver integrations firewall (configured via `-alertmanager.receivers-firewall.*`) diff --git a/docs/guides/ruler-tenant-federation.md b/docs/guides/ruler-tenant-federation.md new file mode 100644 index 00000000000..9a50db6dd75 --- /dev/null +++ b/docs/guides/ruler-tenant-federation.md @@ -0,0 +1,108 @@ +--- +title: "Ruler tenant federation" +linkTitle: "Ruler tenant federation" +weight: 10 +slug: ruler-tenant-federation +--- + +This guide explains how to configure the Ruler to evaluate federated rule groups, which query data from several tenants while the resulting series and alerts belong to a single tenant. The feature is experimental and implements the [federated ruler proposal](../proposals/federated-ruler.md). + +## How it works + +A federated rule group is a regular rule group with an additional `src_tenants` field listing the tenants to query: + +```yaml +name: cortex-admin +interval: 1m +src_tenants: [team-a, team-b, team-c] +rules: + - record: tenant:prometheus_rule_evaluation_failures:rate5m + expr: sum by (__tenant_id__) (rate(prometheus_rule_evaluation_failures_total[5m])) + - alert: TenantRuleEvaluationFailures + expr: sum by (__tenant_id__) (rate(prometheus_rule_evaluation_failures_total[5m])) > 0 + for: 10m + labels: + severity: warning + annotations: + summary: "Rule evaluations are failing in tenant {{ $labels.__tenant_id__ }}" +``` + +- The rule group is owned by the tenant that created it through the Ruler API, `infra` for example. +- Every rule in the group is evaluated with the `X-Scope-OrgID` set to `team-a|team-b|team-c`, so the query runs across the source tenants exactly like a federated query, and each series carries the `__tenant_id__` label. As for any federated query, the label is not added when `src_tenants` lists a single tenant. +- The resulting series, the `ALERTS` and `ALERTS_FOR_STATE` series and the notifications are written to `infra` only. The source tenants are never modified and cannot see the rule group. +- Rule groups without `src_tenants` are not affected. + +Alerting rules work the same way: the `TenantRuleEvaluationFailures` alert above fires once per source tenant with failing rule evaluations, and every alert is sent to the Alertmanager configuration of `infra`. The `__tenant_id__` label is kept on the alert as long as the expression does not aggregate it away, so the Alertmanager configuration of the owning tenant can route the alerts per source tenant: + +```yaml +route: + receiver: infra-default + routes: + - matchers: ['__tenant_id__="team-a"'] + receiver: team-a-slack + - matchers: ['__tenant_id__="team-b"'] + receiver: team-b-slack +``` + +### Chaining rules within a federated rule group + +In Prometheus, the rules of a group are evaluated in order, so a rule can use the series recorded by a previous rule of the same group. In a federated rule group this does not work out of the box, because the results of a recording rule are stored in the owning tenant while the following rules still query the source tenants. In the following group owned by `infra`, the alert never fires: `job:requests:rate5m` is written to `infra`, but the alert looks for it in `team-a` and `team-b`. + +```yaml +name: traffic +src_tenants: [team-a, team-b] +rules: + - record: job:requests:rate5m + expr: sum by (job) (rate(http_requests_total[5m])) + - alert: HighTraffic + expr: job:requests:rate5m > 1000 +``` + +To reuse the output of a previous rule, add the owning tenant to `src_tenants`. The recorded series is then found in `infra`, carrying the `__tenant_id__="infra"` label: + +```yaml +name: traffic +src_tenants: [infra, team-a, team-b] +rules: + - record: job:requests:rate5m + expr: sum by (job) (rate(http_requests_total[5m])) + - alert: HighTraffic + expr: job:requests:rate5m > 1000 +``` + +## Configuration + +Federated rule groups require multi-tenant query federation and the ruler flag: + +``` +-tenant-federation.enabled=true +-ruler.enable-federated-rules=true +``` + +`-tenant-federation.enabled` must be set on all Cortex services. When the ruler evaluates rules through the query frontend (`-ruler.frontend-address`), the query frontend and the queriers perform the federated query; otherwise the ruler merges the results of the source tenants itself. + +When the feature is disabled, the Ruler API rejects rule groups with `src_tenants` and any stored federated rule group is skipped with a warning log. + +### Restricting the tenants allowed to create federated rule groups + +By default, every tenant can create federated rule groups querying any tenant. The following flags restrict which tenants may own federated rule groups: + +``` +-ruler.allowed-federated-tenants=infra,platform +-ruler.disallowed-federated-tenants=untrusted +``` + +- If `-ruler.allowed-federated-tenants` is set, only the listed tenants can create federated rule groups. +- If `-ruler.disallowed-federated-tenants` is set, the listed tenants cannot create federated rule groups even if they are allowed otherwise. + +The checks apply when a rule group is created and again when the ruler loads the rule groups, so changing the flags (and restarting the ruler) also disables the federated rule groups already stored for a tenant. Note that these flags do not restrict which tenants can be listed in `src_tenants`. + +### Limits + +- `-tenant-federation.max-tenant` also limits the number of tenants listed in `src_tenants`. +- When `-tenant-federation.regex-matcher-enabled` is set, the joined tenant IDs are resolved as a regular expression against the tenants discovered in the blocks storage. Tenant IDs containing regex metacharacters (`.`, `*`, `(`, `)`) are therefore rejected in `src_tenants`, and a source tenant that has not uploaded any block yet is silently ignored. + +## Deployment notes + +- Rulers running a version without this feature ignore the `src_tenants` field and evaluate such rule groups against the owning tenant only. Enable the feature and create federated rule groups only once every ruler has been upgraded, and delete them before downgrading. +- The `local` and `configdb` rule stores load Prometheus rule files, which cannot contain `src_tenants`. Federated rule groups require a rule store backed by an object store. diff --git a/docs/proposals/federated-ruler.md b/docs/proposals/federated-ruler.md index d2a9609f4a4..eabbc646670 100644 --- a/docs/proposals/federated-ruler.md +++ b/docs/proposals/federated-ruler.md @@ -71,7 +71,7 @@ To support this we suggest an additional field `src_tenants` on the rule group c | Challenge | Status | |--------------------------------------------------------------------------|---------------------------------------| -| Allow federated rules behind feature flag | Planned but not yet implemented | -| Allow federated rules only for select tenants | Planned but not yet implemented | -| Where to store resulting series of federated rules | Planned but not yet implemented | -| Which tenants to query from for federated rules | Planned but not yet implemented | +| Allow federated rules behind feature flag | Implemented | +| Allow federated rules only for select tenants | Implemented | +| Where to store resulting series of federated rules | Implemented | +| Which tenants to query from for federated rules | Implemented | diff --git a/integration/e2ecortex/client.go b/integration/e2ecortex/client.go index 92352031bed..c871bd77553 100644 --- a/integration/e2ecortex/client.go +++ b/integration/e2ecortex/client.go @@ -818,12 +818,15 @@ func (c *Client) GetRuleGroups() (map[string][]rulefmt.RuleGroup, error) { // SetRuleGroup configures the provided rulegroup to the ruler. func (c *Client) SetRuleGroup(rulegroup rulefmt.RuleGroup, namespace string) error { - // Create write request data, err := yaml.Marshal(rulegroup) if err != nil { return err } + return c.SetRuleGroupYAML(data, namespace) +} +// SetRuleGroupYAML configures the provided YAML encoded rulegroup to the ruler. +func (c *Client) SetRuleGroupYAML(data []byte, namespace string) error { // Create HTTP request req, err := http.NewRequest("POST", fmt.Sprintf("http://%s/api/prom/rules/%s", c.rulerAddress, url.PathEscape(namespace)), bytes.NewReader(data)) if err != nil { diff --git a/integration/ruler_test.go b/integration/ruler_test.go index be36e5bc7c9..8b07bcf2db9 100644 --- a/integration/ruler_test.go +++ b/integration/ruler_test.go @@ -9,6 +9,7 @@ import ( "crypto/x509/pkix" "encoding/json" "fmt" + "io" "math/rand" "net/http" "os" @@ -27,6 +28,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/thanos-io/objstore/providers/s3" + "gopkg.in/yaml.v3" "github.com/cortexproject/cortex/integration/ca" "github.com/cortexproject/cortex/integration/e2e" @@ -1941,3 +1943,203 @@ func TestRulerXFunctionsWithThanosEngine(t *testing.T) { require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_prometheus_rule_evaluation_failures_total"}, e2e.WithLabelMatchers(m), e2e.WaitMissingMetrics)) } + +func TestRulerFederatedRules(t *testing.T) { + s, err := e2e.NewScenario(networkName) + require.NoError(t, err) + defer s.Close() + + // Start dependencies. + consul := e2edb.NewConsul() + minio := e2edb.NewMinio(9000, bucketName, rulestoreBucketName) + require.NoError(t, s.StartAndWaitReady(consul, minio)) + + baseFlags := mergeFlags( + BlocksStorageFlags(), + RulerFlags(), + map[string]string{ + // All the data lives in the ingester, so the store-gateway does not need to be reachable. + "-querier.store-gateway-addresses": "localhost:12345", + // Enable the bucket index so we can skip the initial bucket scan. + "-blocks-storage.bucket-store.bucket-index.enabled": "true", + // Evaluate rules often, so that we don't need to wait for metrics to show up. + "-ruler.evaluation-interval": "2s", + "-ruler.poll-interval": "2s", + // We run single ingester only, no replication. + "-distributor.replication-factor": "1", + // Federated rule groups. + "-tenant-federation.enabled": "true", + "-ruler.enable-federated-rules": "true", + "-ruler.allowed-federated-tenants": "infra", + }, + ) + + const ( + namespace = "test" + owner = "infra" + ) + srcTenants := []string{"team-a", "team-b"} + + distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), baseFlags, "") + ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), baseFlags, "") + require.NoError(t, s.StartAndWaitReady(distributor, ingester)) + require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + + // Push one series per source tenant. + for _, tenant := range srcTenants { + c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", tenant) + require.NoError(t, err) + + series, _ := generateSeries("metric", time.Now(), prompb.Label{Name: "tenant", Value: tenant}) + res, err := c.Push(series) + require.NoError(t, err) + require.Equal(t, 200, res.StatusCode) + } + + for _, tc := range []struct { + name string + viaQueryFrontend bool + }{ + {name: "ruler", viaQueryFrontend: false}, + {name: "query_frontend", viaQueryFrontend: true}, + } { + t.Run(tc.name, func(t *testing.T) { + // Evaluate the rules either directly from the ruler or through the query frontend. + var ( + queryFrontend *e2ecortex.CortexService + querier *e2ecortex.CortexService + rulerFlags = baseFlags + ) + if tc.viaQueryFrontend { + queryFrontend = e2ecortex.NewQueryFrontend("query-frontend", baseFlags, "") + require.NoError(t, s.Start(queryFrontend)) + querier = e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(baseFlags, map[string]string{ + "-querier.frontend-address": queryFrontend.NetworkGRPCEndpoint(), + }), "") + rulerFlags = mergeFlags(baseFlags, map[string]string{ + "-ruler.frontend-address": queryFrontend.NetworkGRPCEndpoint(), + }) + } else { + querier = e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), baseFlags, "") + } + ruler := e2ecortex.NewRuler("ruler", consul.NetworkHTTPEndpoint(), rulerFlags, "") + require.NoError(t, s.StartAndWaitReady(querier, ruler)) + t.Cleanup(func() { + _ = s.Stop(ruler) + _ = s.Stop(querier) + if queryFrontend != nil { + _ = s.Stop(queryFrontend) + } + }) + + queryAddress := querier.HTTPEndpoint() + if tc.viaQueryFrontend { + queryAddress = queryFrontend.HTTPEndpoint() + } + + // Wait until the querier and ruler have updated the ring. + require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + require.NoError(t, ruler.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total")) + + groupName := "federated_" + tc.name + recordName := "federated_metric_" + tc.name + alertName := "FederatedMetricPresent_" + tc.name + + ruleGroup := ruleGroupWithRule(groupName, recordName, "count by (__tenant_id__) (metric)") + ruleGroup.Rules = append(ruleGroup.Rules, rulefmt.Rule{ + Alert: alertName, + Expr: "count by (__tenant_id__) (metric) > 0", + Labels: map[string]string{"severity": "warning"}, + }) + federatedGroup, err := yaml.Marshal(rulespb.RuleGroup{ + RuleGroup: ruleGroup, + SrcTenants: srcTenants, + }) + require.NoError(t, err) + + // A tenant that is not allowed to create federated rule groups is rejected. + teamA, err := e2ecortex.NewClient("", queryAddress, "", ruler.HTTPEndpoint(), srcTenants[0]) + require.NoError(t, err) + require.ErrorContains(t, teamA.SetRuleGroupYAML(federatedGroup, namespace), "403") + + infra, err := e2ecortex.NewClient("", queryAddress, "", ruler.HTTPEndpoint(), owner) + require.NoError(t, err) + require.NoError(t, infra.SetRuleGroupYAML(federatedGroup, namespace)) + + // The stored rule group keeps its source tenants. + res, err := infra.GetRuleGroup(namespace, groupName) + require.NoError(t, err) + body, err := io.ReadAll(res.Body) + require.NoError(t, res.Body.Close()) + require.NoError(t, err) + require.Equal(t, http.StatusOK, res.StatusCode) + require.Contains(t, string(body), "src_tenants:\n - team-a\n - team-b\n") + + // Wait until the ruler has loaded and successfully evaluated the group. + rgMatcher := ruleGroupMatcher(owner, namespace, groupName) + require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(2), []string{"cortex_prometheus_rule_group_rules"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics)) + require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_prometheus_rule_evaluations_total"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics)) + require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(0), []string{"cortex_prometheus_rule_evaluation_failures_total"}, e2e.WithLabelMatchers(rgMatcher), e2e.WaitMissingMetrics)) + + if tc.viaQueryFrontend { + // The federated queries reach the query frontend with the joined source tenants as tenant. + require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_ruler_query_frontend_clients"}, e2e.WaitMissingMetrics)) + require.NoError(t, queryFrontend.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_query_frontend_queries_total"}, e2e.WithLabelMatchers( + labels.MustNewMatcher(labels.MatchEqual, "user", "team-a|team-b"), + labels.MustNewMatcher(labels.MatchEqual, "source", "ruler"), + ), e2e.WaitMissingMetrics)) + } + + // The recording rule result is stored in the owning tenant with one series per source tenant. + var result model.Vector + require.Eventually(t, func() bool { + value, err := infra.Query(recordName, time.Now()) + if err != nil { + return false + } + result = value.(model.Vector) + return len(result) == len(srcTenants) + }, 30*time.Second, time.Second) + + resultTenants := make([]string, 0, len(result)) + for _, sample := range result { + require.Equal(t, model.SampleValue(1), sample.Value) + resultTenants = append(resultTenants, string(sample.Metric["__tenant_id__"])) + } + sort.Strings(resultTenants) + require.Equal(t, srcTenants, resultTenants) + + // The source tenants do not get the result. + value, err := teamA.Query(recordName, time.Now()) + require.NoError(t, err) + require.Empty(t, value.(model.Vector)) + + // The alerting rule fires once per source tenant, keeping the __tenant_id__ label. + var alerts []*Alert + require.Eventually(t, func() bool { + groups, _, err := infra.GetPrometheusRules(e2ecortex.RuleFilter{RuleNames: []string{alertName}}) + if err != nil || len(groups) != 1 || len(groups[0].Rules) != 1 { + return false + } + rule := parseAlertFromRule(t, groups[0].Rules[0]) + alerts = rule.Alerts + return rule.State == "firing" && len(alerts) == len(srcTenants) + }, 30*time.Second, time.Second) + + alertTenants := make([]string, 0, len(alerts)) + for _, alert := range alerts { + require.Equal(t, "firing", alert.State) + require.Equal(t, alertName, alert.Labels.Get(model.AlertNameLabel)) + require.Equal(t, "warning", alert.Labels.Get("severity")) + alertTenants = append(alertTenants, alert.Labels.Get("__tenant_id__")) + } + sort.Strings(alertTenants) + require.Equal(t, srcTenants, alertTenants) + + // The source tenants do not see the rule group nor its alerts. + groups, _, err := teamA.GetPrometheusRules(e2ecortex.RuleFilter{RuleNames: []string{alertName}}) + require.NoError(t, err) + require.Empty(t, groups) + }) + } +} diff --git a/pkg/cortex/cortex.go b/pkg/cortex/cortex.go index 321c7008f04..744a67eabb2 100644 --- a/pkg/cortex/cortex.go +++ b/pkg/cortex/cortex.go @@ -70,8 +70,9 @@ import ( ) var ( - errInvalidHTTPPrefix = errors.New("HTTP prefix should be empty or start with /") - errTimeoutClassificationRequiresQueryStats = errors.New("timeout classification requires query stats to be enabled (frontend.query-stats-enabled)") + errInvalidHTTPPrefix = errors.New("HTTP prefix should be empty or start with /") + errRulerFederatedRulesRequireTenantFederation = errors.New("-ruler.enable-federated-rules requires -tenant-federation.enabled") + errTimeoutClassificationRequiresQueryStats = errors.New("timeout classification requires query stats to be enabled (frontend.query-stats-enabled)") ) // The design pattern for Cortex is a series of config objects, which are @@ -263,6 +264,9 @@ func (c *Config) Validate(log log.Logger) error { if err := c.Ruler.Validate(c.LimitsConfig, log); err != nil { return errors.Wrap(err, "invalid ruler config") } + if c.Ruler.EnableFederatedRules && !c.TenantFederation.Enabled { + return errRulerFederatedRulesRequireTenantFederation + } if err := c.BlocksStorage.Validate(); err != nil { return errors.Wrap(err, "invalid TSDB config") } diff --git a/pkg/cortex/cortex_test.go b/pkg/cortex/cortex_test.go index 3d38a98be2a..6c44feb303d 100644 --- a/pkg/cortex/cortex_test.go +++ b/pkg/cortex/cortex_test.go @@ -174,6 +174,25 @@ func TestConfigValidation(t *testing.T) { }, expectedError: errInvalidHTTPPrefix, }, + { + name: "should fail validation if federated rules are enabled without tenant federation", + getTestConfig: func() *Config { + configuration := newDefaultConfig() + configuration.Ruler.EnableFederatedRules = true + return configuration + }, + expectedError: errRulerFederatedRulesRequireTenantFederation, + }, + { + name: "should pass validation if federated rules are enabled with tenant federation", + getTestConfig: func() *Config { + configuration := newDefaultConfig() + configuration.Ruler.EnableFederatedRules = true + configuration.TenantFederation.Enabled = true + return configuration + }, + expectedError: nil, + }, { name: "should fail validation for invalid resource to monitor", getTestConfig: func() *Config { diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go index 1106a458e38..0684119d065 100644 --- a/pkg/cortex/modules.go +++ b/pkg/cortex/modules.go @@ -695,6 +695,8 @@ func (t *Cortex) initRuler() (serv services.Service, err error) { t.Cfg.Ruler.PrometheusHTTPPrefix = t.Cfg.API.PrometheusHTTPPrefix t.Cfg.Ruler.Ring.ListenPort = t.Cfg.Server.GRPCListenPort t.Cfg.Ruler.NameValidationScheme = t.Cfg.NameValidationScheme + t.Cfg.Ruler.TenantFederationRegexMatcherEnabled = t.Cfg.TenantFederation.RegexMatcherEnabled + t.Cfg.Ruler.TenantFederationMaxTenant = t.Cfg.TenantFederation.MaxTenant metrics := ruler.NewRuleEvalMetrics(t.Cfg.Ruler, prometheus.DefaultRegisterer) rulerRegisterer := prometheus.WrapRegistererWith(prometheus.Labels{"engine": "ruler"}, prometheus.DefaultRegisterer) @@ -731,6 +733,14 @@ func (t *Cortex) initRuler() (serv services.Service, err error) { // TODO: Consider wrapping logger to differentiate from querier module logger queryable, _, queryEngine, _ = querier.New(t.Cfg.Querier, t.OverridesConfig, t.Distributor, t.StoreQueryables, rulerRegisterer, util_log.Logger, t.OverridesConfig.RulesPartialData, nil) } + if t.Cfg.Ruler.EnableFederatedRules { + util_log.WarnExperimentalUse("ruler.enable-federated-rules") + // Federated rule groups are evaluated with a multi-tenant org ID, so the + // queryable has to merge the results of every source tenant. Metrics are + // not registered because the querier registers the same ones when both + // run in a single process. + queryable = tenantfederation.NewQueryable(queryable, t.Cfg.TenantFederation, true, nil) + } managerFactory := ruler.DefaultTenantManagerFactory(t.Cfg.Ruler, pusher, queryable, queryEngine, t.OverridesConfig, metrics, prometheus.DefaultRegisterer) manager, err = ruler.NewDefaultMultiTenantManager(t.Cfg.Ruler, t.OverridesConfig, managerFactory, metrics, prometheus.DefaultRegisterer, util_log.Logger) diff --git a/pkg/ruler/api.go b/pkg/ruler/api.go index 177782e607b..8330e333e74 100644 --- a/pkg/ruler/api.go +++ b/pkg/ruler/api.go @@ -17,7 +17,6 @@ import ( "github.com/pkg/errors" v1 "github.com/prometheus/client_golang/api/prometheus/v1" "github.com/prometheus/prometheus/model/labels" - "github.com/prometheus/prometheus/model/rulefmt" "github.com/weaveworks/common/user" "gopkg.in/yaml.v3" @@ -527,7 +526,7 @@ func (a *API) ListRules(w http.ResponseWriter, req *http.Request) { level.Debug(logger).Log("msg", "retrieved rule groups from rule store", "userID", userID, "num_namespaces", len(rgs)) - formatted := rgs.Formatted() + formatted := rgs.FormattedRuleGroups() marshalAndSend(formatted, w, logger) } @@ -570,7 +569,7 @@ func (a *API) CreateRuleGroup(w http.ResponseWriter, req *http.Request) { level.Debug(logger).Log("msg", "attempting to unmarshal rulegroup", "userID", userID, "group", string(payload)) - rg := rulefmt.RuleGroup{} + rg := rulespb.RuleGroup{} err = yaml.Unmarshal(payload, &rg) if err != nil { level.Error(logger).Log("msg", "unable to unmarshal rule group payload", "err", err.Error()) @@ -578,7 +577,7 @@ func (a *API) CreateRuleGroup(w http.ResponseWriter, req *http.Request) { return } - errs := a.ruler.manager.ValidateRuleGroup(rg) + errs := a.ruler.manager.ValidateRuleGroup(rg.RuleGroup) if len(errs) > 0 { e := []string{} for _, err := range errs { @@ -590,6 +589,20 @@ func (a *API) CreateRuleGroup(w http.ResponseWriter, req *http.Request) { return } + if len(rg.SrcTenants) > 0 { + srcTenants, err := a.ruler.manager.ValidateFederatedRuleGroup(userID, rg.SrcTenants) + if err != nil { + level.Error(logger).Log("msg", "federated rule group validation failure", "err", err.Error(), "user", userID) + status := http.StatusBadRequest + if errors.Is(err, errFederatedRulesNotAllowed) { + status = http.StatusForbidden + } + http.Error(w, err.Error(), status) + return + } + rg.SrcTenants = srcTenants + } + if err := a.ruler.AssertMaxRulesPerRuleGroup(userID, len(rg.Rules)); err != nil { level.Error(logger).Log("msg", "limit validation failure", "err", err.Error(), "user", userID) http.Error(w, err.Error(), http.StatusBadRequest) @@ -615,7 +628,7 @@ func (a *API) CreateRuleGroup(w http.ResponseWriter, req *http.Request) { loadedRg := rulespb.FromProto(rgProto) rgYaml, err := yaml.Marshal(loadedRg) if err == nil { - err = yaml.Unmarshal(rgYaml, &rulefmt.RuleGroup{}) + err = yaml.Unmarshal(rgYaml, &rulespb.RuleGroup{}) } if err != nil { level.Error(logger).Log("msg", "unable to load rule group from proto", "err", err.Error(), "user", userID) diff --git a/pkg/ruler/api_test.go b/pkg/ruler/api_test.go index f55b1e0e314..43ce613f224 100644 --- a/pkg/ruler/api_test.go +++ b/pkg/ruler/api_test.go @@ -786,3 +786,132 @@ func requestFor(t *testing.T, method string, url string, body io.Reader, userID return req.WithContext(ctx) } + +func TestRuler_CreateFederated(t *testing.T) { + const federatedGroup = ` +name: test +interval: 15s +src_tenants: [team-b, team-a, team-b] +rules: +- record: up_rule + expr: sum by (__tenant_id__) (up) +` + const plainGroup = ` +name: test +interval: 15s +rules: +- record: up_rule + expr: up +` + + tc := []struct { + name string + cfg func(cfg *Config) + user string + input string + status int + err string + output string + }{ + { + name: "federated rules disabled", + cfg: func(*Config) {}, + user: "infra", + input: federatedGroup, + status: 400, + err: "federated rules are disabled\n", + }, + { + name: "plain group accepted when federated rules are disabled", + cfg: func(*Config) {}, + user: "infra", + input: plainGroup, + status: 202, + output: "name: test\ninterval: 15s\nrules:\n - record: up_rule\n expr: up\n", + }, + { + name: "tenant not allowed", + cfg: func(cfg *Config) { + cfg.EnableFederatedRules = true + cfg.AllowedFederatedTenants = []string{"infra"} + }, + user: "team-a", + input: federatedGroup, + status: 403, + err: "tenant is not allowed to create federated rule groups: team-a\n", + }, + { + name: "invalid src tenant", + cfg: func(cfg *Config) { + cfg.EnableFederatedRules = true + }, + user: "infra", + input: strings.Replace(federatedGroup, "team-a", "team|a", 1), + status: 400, + err: "invalid src tenant \"team|a\"", + }, + { + name: "too many src tenants", + cfg: func(cfg *Config) { + cfg.EnableFederatedRules = true + cfg.TenantFederationMaxTenant = 1 + }, + user: "infra", + input: federatedGroup, + status: 400, + err: "too many src tenants (limit: 1 actual: 2)\n", + }, + { + name: "stored with normalized src tenants", + cfg: func(cfg *Config) { + cfg.EnableFederatedRules = true + cfg.AllowedFederatedTenants = []string{"infra"} + }, + user: "infra", + input: federatedGroup, + status: 202, + output: "name: test\ninterval: 15s\nrules:\n - record: up_rule\n expr: sum by (__tenant_id__) (up)\nsrc_tenants:\n - team-a\n - team-b\n", + }, + } + + for _, tt := range tc { + t.Run(tt.name, func(t *testing.T) { + store := newMockRuleStore(make(map[string]rulespb.RuleGroupList), nil) + cfg := defaultRulerConfig(t) + tt.cfg(&cfg) + + r := newTestRuler(t, cfg, store, nil) + defer services.StopAndAwaitTerminated(context.Background(), r) //nolint:errcheck + + a := NewAPI(r, r.store, log.NewNopLogger()) + + router := mux.NewRouter() + router.Path("/api/v1/rules").Methods("GET").HandlerFunc(a.ListRules) + router.Path("/api/v1/rules/{namespace}").Methods("POST").HandlerFunc(a.CreateRuleGroup) + router.Path("/api/v1/rules/{namespace}/{groupName}").Methods("GET").HandlerFunc(a.GetRuleGroup) + + req := requestFor(t, http.MethodPost, "https://localhost:8080/api/v1/rules/namespace", strings.NewReader(tt.input), tt.user) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + require.Equal(t, tt.status, w.Code) + + if tt.err != "" { + require.Contains(t, w.Body.String(), tt.err) + return + } + + req = requestFor(t, http.MethodGet, "https://localhost:8080/api/v1/rules/namespace/test", nil, tt.user) + w = httptest.NewRecorder() + router.ServeHTTP(w, req) + require.Equal(t, 200, w.Code) + require.Equal(t, tt.output, w.Body.String()) + + // The rule group listing exposes src_tenants as well. + req = requestFor(t, http.MethodGet, "https://localhost:8080/api/v1/rules", nil, tt.user) + w = httptest.NewRecorder() + router.ServeHTTP(w, req) + require.Equal(t, 200, w.Code) + require.Equal(t, strings.Contains(tt.input, "src_tenants"), strings.Contains(w.Body.String(), "src_tenants")) + }) + } +} diff --git a/pkg/ruler/federation.go b/pkg/ruler/federation.go new file mode 100644 index 00000000000..3211c9048f3 --- /dev/null +++ b/pkg/ruler/federation.go @@ -0,0 +1,71 @@ +package ruler + +import ( + "fmt" + "regexp" + "slices" + + "github.com/pkg/errors" + + "github.com/cortexproject/cortex/pkg/util/users" +) + +var ( + errFederatedRulesDisabled = errors.New("federated rules are disabled") + errFederatedRulesNotAllowed = errors.New("tenant is not allowed to create federated rule groups") +) + +// federatedRulesChecker decides whether a tenant may own federated rule groups +// and validates their source tenants. +type federatedRulesChecker struct { + enabled bool + allowedTenants *users.AllowedTenants + regexMatcherEnabled bool + maxTenant int +} + +func newFederatedRulesChecker(cfg Config) *federatedRulesChecker { + return &federatedRulesChecker{ + enabled: cfg.EnableFederatedRules, + allowedTenants: users.NewAllowedTenants(cfg.AllowedFederatedTenants, cfg.DisallowedFederatedTenants), + regexMatcherEnabled: cfg.TenantFederationRegexMatcherEnabled, + maxTenant: cfg.TenantFederationMaxTenant, + } +} + +// checkOwner returns an error if userID may not own federated rule groups. +func (c *federatedRulesChecker) checkOwner(userID string) error { + if !c.enabled { + return errFederatedRulesDisabled + } + if !c.allowedTenants.IsAllowed(userID) { + return fmt.Errorf("%w: %s", errFederatedRulesNotAllowed, userID) + } + return nil +} + +// validateSrcTenants validates the source tenants of a rule group and returns +// them sorted and de-duplicated. +func (c *federatedRulesChecker) validateSrcTenants(srcTenants []string) ([]string, error) { + for _, id := range srcTenants { + // ValidTenantID accepts the empty string, which would produce an org ID + // that fails at every evaluation. + if id == "" { + return nil, errors.New("src tenant must not be empty") + } + if err := users.ValidTenantID(id); err != nil { + return nil, errors.Wrapf(err, "invalid src tenant %q", id) + } + // The querier interprets the joined org ID as a regex when the regex + // matcher is enabled, so only literal tenant IDs are accepted then. + if c.regexMatcherEnabled && regexp.QuoteMeta(id) != id { + return nil, fmt.Errorf("src tenant %q contains regex metacharacters, which are not supported when -tenant-federation.regex-matcher-enabled is set", id) + } + } + + normalized := users.NormalizeTenantIDs(slices.Clone(srcTenants)) + if c.maxTenant > 0 && len(normalized) > c.maxTenant { + return nil, fmt.Errorf("too many src tenants (limit: %d actual: %d)", c.maxTenant, len(normalized)) + } + return normalized, nil +} diff --git a/pkg/ruler/federation_test.go b/pkg/ruler/federation_test.go new file mode 100644 index 00000000000..18e8f03c38a --- /dev/null +++ b/pkg/ruler/federation_test.go @@ -0,0 +1,115 @@ +package ruler + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestFederatedRulesChecker_CheckOwner(t *testing.T) { + tests := map[string]struct { + cfg Config + userID string + expectedErr error + }{ + "disabled": { + cfg: Config{}, + userID: "infra", + expectedErr: errFederatedRulesDisabled, + }, + "enabled for all tenants": { + cfg: Config{EnableFederatedRules: true}, + userID: "infra", + }, + "allowed tenant": { + cfg: Config{EnableFederatedRules: true, AllowedFederatedTenants: []string{"infra"}}, + userID: "infra", + }, + "tenant not in allowed list": { + cfg: Config{EnableFederatedRules: true, AllowedFederatedTenants: []string{"infra"}}, + userID: "team-a", + expectedErr: errFederatedRulesNotAllowed, + }, + "disallowed tenant": { + cfg: Config{EnableFederatedRules: true, DisallowedFederatedTenants: []string{"team-a"}}, + userID: "team-a", + expectedErr: errFederatedRulesNotAllowed, + }, + "allowed and disallowed": { + cfg: Config{EnableFederatedRules: true, AllowedFederatedTenants: []string{"infra"}, DisallowedFederatedTenants: []string{"infra"}}, + userID: "infra", + expectedErr: errFederatedRulesNotAllowed, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + err := newFederatedRulesChecker(tc.cfg).checkOwner(tc.userID) + if tc.expectedErr == nil { + require.NoError(t, err) + return + } + require.ErrorIs(t, err, tc.expectedErr) + }) + } +} + +func TestFederatedRulesChecker_ValidateSrcTenants(t *testing.T) { + tests := map[string]struct { + cfg Config + srcTenants []string + expected []string + expectedErr string + }{ + "sorted and de-duplicated": { + srcTenants: []string{"team-b", "team-a", "team-b"}, + expected: []string{"team-a", "team-b"}, + }, + "invalid tenant id": { + srcTenants: []string{"team-a", "team|b"}, + expectedErr: `invalid src tenant "team|b"`, + }, + "empty tenant id": { + srcTenants: []string{""}, + expectedErr: "src tenant must not be empty", + }, + "empty tenant id among valid ones": { + srcTenants: []string{"team-a", ""}, + expectedErr: "src tenant must not be empty", + }, + "regex metacharacters allowed without regex matcher": { + srcTenants: []string{"team.a"}, + expected: []string{"team.a"}, + }, + "regex metacharacters rejected with regex matcher": { + cfg: Config{TenantFederationRegexMatcherEnabled: true}, + srcTenants: []string{"team.a"}, + expectedErr: `src tenant "team.a" contains regex metacharacters`, + }, + "max tenant": { + cfg: Config{TenantFederationMaxTenant: 2}, + srcTenants: []string{"team-a", "team-b", "team-c"}, + expectedErr: "too many src tenants (limit: 2 actual: 3)", + }, + "max tenant counts unique tenants": { + cfg: Config{TenantFederationMaxTenant: 2}, + srcTenants: []string{"team-a", "team-b", "team-a"}, + expected: []string{"team-a", "team-b"}, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + input := append([]string(nil), tc.srcTenants...) + actual, err := newFederatedRulesChecker(tc.cfg).validateSrcTenants(input) + if tc.expectedErr != "" { + require.ErrorContains(t, err, tc.expectedErr) + return + } + require.NoError(t, err) + require.Equal(t, tc.expected, actual) + // The input must not be modified. + require.Equal(t, tc.srcTenants, input) + }) + } +} diff --git a/pkg/ruler/manager.go b/pkg/ruler/manager.go index 86611201899..f918a746d39 100644 --- a/pkg/ruler/manager.go +++ b/pkg/ruler/manager.go @@ -28,6 +28,7 @@ import ( "github.com/cortexproject/cortex/pkg/parser" "github.com/cortexproject/cortex/pkg/ring/client" "github.com/cortexproject/cortex/pkg/ruler/rulespb" + "github.com/cortexproject/cortex/pkg/util/users" ) type DefaultMultiTenantManager struct { @@ -71,6 +72,12 @@ type DefaultMultiTenantManager struct { syncRuleMtx sync.Mutex ruleGroupIterationFunc promRules.GroupEvalIterationFunc + + federatedRules *federatedRulesChecker + // Org ID to inject when evaluating a federated rule group, keyed by user + // and then by the prometheus rule group key. + federatedGroupsMtx sync.RWMutex + federatedGroups map[string]map[string]string } func NewDefaultMultiTenantManager(cfg Config, limits RulesLimits, managerFactory ManagerFactory, evalMetrics *RuleEvalMetrics, reg prometheus.Registerer, logger log.Logger) (*DefaultMultiTenantManager, error) { @@ -133,6 +140,8 @@ func NewDefaultMultiTenantManager(cfg Config, limits RulesLimits, managerFactory registry: reg, logger: logger, ruleGroupIterationFunc: defaultRuleGroupIterationFunc, + federatedRules: newFederatedRulesChecker(cfg), + federatedGroups: map[string]map[string]string{}, } if cfg.RulesBackupEnabled() { m.rulesBackupManager = newRulesBackupManager(cfg, logger, reg) @@ -169,6 +178,7 @@ func (r *DefaultMultiTenantManager) SyncRuleGroups(ctx context.Context, ruleGrou r.removeNotifier(userID) r.mapper.cleanupUser(userID) + r.setFederatedGroups(userID, nil) r.userExternalLabels.remove(userID) r.userExternalURL.remove(userID) r.lastReloadSuccessful.DeleteLabelValues(userID) @@ -206,6 +216,9 @@ func (r *DefaultMultiTenantManager) BackUpRuleGroups(ctx context.Context, ruleGr // syncRulesToManager maps the rule files to disk, detects any changes and will create/update the // users Prometheus Rules Manager. func (r *DefaultMultiTenantManager) syncRulesToManager(ctx context.Context, user string, groups rulespb.RuleGroupList) { + groups = r.filterFederatedRuleGroups(user, groups) + r.setFederatedGroups(user, r.federatedOrgIDs(user, groups)) + // Map the files to disk and return the file names to be passed to the users manager if they // have been updated rulesUpdated, files, err := r.mapper.MapRules(user, groups.Formatted()) @@ -234,7 +247,7 @@ func (r *DefaultMultiTenantManager) syncRulesToManager(ctx context.Context, user if (rulesUpdated || externalLabelsUpdated || externalURLUpdated) && existing { r.updateRuleCache(user, manager.RuleGroups()) } - err = manager.Update(r.cfg.EvaluationInterval, files, externalLabels, externalURL, r.ruleGroupIterationFunc) + err = manager.Update(r.cfg.EvaluationInterval, files, externalLabels, externalURL, r.ruleGroupIterationFuncFor(user)) r.deleteRuleCache(user) if err != nil { r.lastReloadSuccessful.WithLabelValues(user).Set(0) @@ -277,6 +290,76 @@ func (r *DefaultMultiTenantManager) createRulesManager(user string, ctx context. return manager } +// ValidateFederatedRuleGroup implements MultiTenantManager. +func (r *DefaultMultiTenantManager) ValidateFederatedRuleGroup(userID string, srcTenants []string) ([]string, error) { + if err := r.federatedRules.checkOwner(userID); err != nil { + return nil, err + } + return r.federatedRules.validateSrcTenants(srcTenants) +} + +// filterFederatedRuleGroups drops the federated rule groups of a user that may +// not own them, e.g. because the feature was disabled after they were stored. +func (r *DefaultMultiTenantManager) filterFederatedRuleGroups(userID string, groups rulespb.RuleGroupList) rulespb.RuleGroupList { + ownerErr := r.federatedRules.checkOwner(userID) + if ownerErr == nil { + return groups + } + + filtered := make(rulespb.RuleGroupList, 0, len(groups)) + for _, g := range groups { + if g.IsFederated() { + level.Warn(r.logger).Log("msg", "skipping federated rule group", "user", userID, "namespace", g.Namespace, "group", g.Name, "err", ownerErr) + continue + } + filtered = append(filtered, g) + } + return filtered +} + +// federatedOrgIDs returns the org ID to query for each federated rule group, +// keyed by the prometheus rule group key. +func (r *DefaultMultiTenantManager) federatedOrgIDs(userID string, groups rulespb.RuleGroupList) map[string]string { + orgIDs := map[string]string{} + for _, g := range groups { + if !g.IsFederated() { + continue + } + key := promRules.GroupKey(r.mapper.ruleFilePath(userID, g.Namespace), g.Name) + orgIDs[key] = users.JoinTenantIDs(g.SrcTenants) + } + return orgIDs +} + +func (r *DefaultMultiTenantManager) setFederatedGroups(userID string, orgIDs map[string]string) { + r.federatedGroupsMtx.Lock() + defer r.federatedGroupsMtx.Unlock() + if len(orgIDs) == 0 { + delete(r.federatedGroups, userID) + return + } + r.federatedGroups[userID] = orgIDs +} + +func (r *DefaultMultiTenantManager) federatedOrgID(userID string, g *promRules.Group) (string, bool) { + r.federatedGroupsMtx.RLock() + defer r.federatedGroupsMtx.RUnlock() + orgID, ok := r.federatedGroups[userID][promRules.GroupKey(g.File(), g.Name())] + return orgID, ok +} + +// ruleGroupIterationFuncFor wraps the iteration func so that federated rule +// groups query their source tenants. The appender re-injects the owner, so the +// resulting series and alerts still belong to the user. +func (r *DefaultMultiTenantManager) ruleGroupIterationFuncFor(userID string) promRules.GroupEvalIterationFunc { + return func(ctx context.Context, g *promRules.Group, evalTimestamp time.Time) { + if orgID, ok := r.federatedOrgID(userID, g); ok { + ctx = user.InjectOrgID(ctx, orgID) + } + r.ruleGroupIterationFunc(ctx, g, evalTimestamp) + } +} + func defaultRuleGroupIterationFunc(ctx context.Context, g *promRules.Group, evalTimestamp time.Time) { logMessage := []any{ "component", "ruler", diff --git a/pkg/ruler/manager_test.go b/pkg/ruler/manager_test.go index 845d0ffc203..6b02de4a9de 100644 --- a/pkg/ruler/manager_test.go +++ b/pkg/ruler/manager_test.go @@ -8,17 +8,21 @@ import ( "github.com/go-kit/log" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/promslog" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/model/rulefmt" "github.com/prometheus/prometheus/notifier" promRules "github.com/prometheus/prometheus/rules" + "github.com/spf13/afero" "github.com/stretchr/testify/require" + "github.com/weaveworks/common/user" "go.uber.org/atomic" "github.com/cortexproject/cortex/pkg/ring/client" "github.com/cortexproject/cortex/pkg/ruler/rulespb" "github.com/cortexproject/cortex/pkg/util" "github.com/cortexproject/cortex/pkg/util/test" + "github.com/cortexproject/cortex/pkg/util/users" ) func TestSyncRuleGroups(t *testing.T) { @@ -452,3 +456,118 @@ func TestValidateRuleGroup_RejectsEmptyGroupName(t *testing.T) { require.NotEmpty(t, errs, "Expected validation errors for empty group name") require.Contains(t, errs[0].Error(), "rule group name must not be empty", "Error should mention empty group name") } + +func TestSyncRuleGroups_FederatedRuleGroups(t *testing.T) { + const owner = "infra" + + newGroup := func(file, name string) *promRules.Group { + return promRules.NewGroup(promRules.GroupOptions{ + Name: name, + File: file, + Opts: &promRules.ManagerOptions{Logger: promslog.NewNopLogger()}, + }) + } + federated := &rulespb.RuleGroupDesc{ + Name: "federated", + Namespace: "ns", + Interval: time.Minute, + User: owner, + SrcTenants: []string{"team-a", "team-b"}, + Rules: []*rulespb.RuleDesc{{Record: "federated_rule", Expr: "up"}}, + } + plain := &rulespb.RuleGroupDesc{ + Name: "plain", + Namespace: "ns", + Interval: time.Minute, + User: owner, + Rules: []*rulespb.RuleDesc{{Record: "plain_rule", Expr: "up"}}, + } + + // Federated rule groups require the multi tenant resolver, like in production. + users.WithDefaultResolver(users.NewMultiResolver()) + t.Cleanup(func() { users.WithDefaultResolver(users.NewSingleResolver()) }) + + newFederatedManager := func(t *testing.T, cfg Config, captured map[string]string) *DefaultMultiTenantManager { + cfg.RulePath = t.TempDir() + iterFunc := func(ctx context.Context, g *promRules.Group, _ time.Time) { + tenantIDs, err := users.TenantIDs(ctx) + require.NoError(t, err) + captured[g.Name()] = users.JoinTenantIDs(tenantIDs) + } + waitDurations := []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond} + m, err := NewDefaultMultiTenantManagerWithIterationFunc(iterFunc, cfg, &ruleLimits{}, RuleManagerFactory([][]*promRules.Group{{}, {}, {}}, waitDurations), nil, prometheus.NewRegistry(), log.NewNopLogger()) + require.NoError(t, err) + t.Cleanup(m.Stop) + return m + } + + evaluate := func(m *DefaultMultiTenantManager, name string) { + file := m.mapper.ruleFilePath(owner, "ns") + m.ruleGroupIterationFuncFor(owner)(user.InjectOrgID(context.Background(), owner), newGroup(file, name), time.Now()) + } + + readRuleFile := func(t *testing.T, m *DefaultMultiTenantManager) string { + content, err := afero.ReadFile(m.mapper.FS, m.mapper.ruleFilePath(owner, "ns")) + require.NoError(t, err) + return string(content) + } + + t.Run("federated groups query their src tenants", func(t *testing.T) { + captured := map[string]string{} + m := newFederatedManager(t, Config{EnableFederatedRules: true}, captured) + + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{owner: {federated, plain}}) + require.NotNil(t, getManager(m, owner)) + require.Contains(t, readRuleFile(t, m), "federated_rule") + + evaluate(m, "federated") + evaluate(m, "plain") + require.Equal(t, "team-a|team-b", captured["federated"]) + require.Equal(t, owner, captured["plain"]) + + // Storing the same group without src tenants stops the injection at the next + // evaluation, even though the prometheus manager keeps the existing group. + noSrcTenants := *federated + noSrcTenants.SrcTenants = nil + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{owner: {&noSrcTenants, plain}}) + require.Contains(t, readRuleFile(t, m), "federated_rule") + + evaluate(m, "federated") + require.Equal(t, owner, captured["federated"]) + }) + + t.Run("federated groups are skipped when disabled", func(t *testing.T) { + captured := map[string]string{} + m := newFederatedManager(t, Config{}, captured) + + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{owner: {federated, plain}}) + require.NotNil(t, getManager(m, owner)) + content := readRuleFile(t, m) + require.Contains(t, content, "plain_rule") + require.NotContains(t, content, "federated_rule") + + evaluate(m, "federated") + require.Equal(t, owner, captured["federated"]) + }) + + t.Run("federated groups are skipped when the owner is not allowed", func(t *testing.T) { + captured := map[string]string{} + m := newFederatedManager(t, Config{EnableFederatedRules: true, AllowedFederatedTenants: []string{"other"}}, captured) + + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{owner: {federated, plain}}) + content := readRuleFile(t, m) + require.Contains(t, content, "plain_rule") + require.NotContains(t, content, "federated_rule") + }) + + t.Run("federated groups are forgotten when the user is removed", func(t *testing.T) { + captured := map[string]string{} + m := newFederatedManager(t, Config{EnableFederatedRules: true}, captured) + + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{owner: {federated}}) + m.SyncRuleGroups(context.Background(), map[string]rulespb.RuleGroupList{}) + m.federatedGroupsMtx.RLock() + defer m.federatedGroupsMtx.RUnlock() + require.Empty(t, m.federatedGroups) + }) +} diff --git a/pkg/ruler/mapper.go b/pkg/ruler/mapper.go index fb14daa5a89..e65c08997a3 100644 --- a/pkg/ruler/mapper.go +++ b/pkg/ruler/mapper.go @@ -82,9 +82,7 @@ func (m *mapper) MapRules(user string, ruleConfigs map[string][]rulefmt.RuleGrou // write all rule configs to disk for filename, groups := range ruleConfigs { - // Store the encoded file name to better handle `/` characters - encodedFileName := url.PathEscape(filename) - fullFileName := filepath.Join(path, encodedFileName) + fullFileName := m.ruleFilePath(user, filename) fileUpdated, err := m.writeRuleGroupsIfNewer(groups, fullFileName) if err != nil { @@ -124,6 +122,12 @@ func (m *mapper) MapRules(user string, ruleConfigs map[string][]rulefmt.RuleGrou return anyUpdated, filenames, nil } +// ruleFilePath returns the on-disk file holding the namespace rules of a user. +// The namespace is path-escaped to better handle `/` characters. +func (m *mapper) ruleFilePath(user, namespace string) string { + return filepath.Join(m.Path, user, url.PathEscape(namespace)) +} + func (m *mapper) writeRuleGroupsIfNewer(groups []rulefmt.RuleGroup, filename string) (bool, error) { sort.Slice(groups, func(i, j int) bool { return groups[i].Name > groups[j].Name diff --git a/pkg/ruler/ruler.go b/pkg/ruler/ruler.go index 82f7c57fb0d..df495cd042d 100644 --- a/pkg/ruler/ruler.go +++ b/pkg/ruler/ruler.go @@ -168,11 +168,19 @@ type Config struct { EnabledTenants flagext.StringSliceCSV `yaml:"enabled_tenants"` DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"` + // Federated rule groups query data from the tenants listed in `src_tenants`. + EnableFederatedRules bool `yaml:"enable_federated_rules"` + AllowedFederatedTenants flagext.StringSliceCSV `yaml:"allowed_federated_tenants"` + DisallowedFederatedTenants flagext.StringSliceCSV `yaml:"disallowed_federated_tenants"` + RingCheckPeriod time.Duration `yaml:"-"` // Field will be populated during runtime. LookbackDelta time.Duration `yaml:"-"` PrometheusHTTPPrefix string `yaml:"-"` + // Populated from the tenant federation config. + TenantFederationRegexMatcherEnabled bool `yaml:"-"` + TenantFederationMaxTenant int `yaml:"-"` EnableQueryStats bool `yaml:"query_stats_enabled"` DisableRuleGroupLabel bool `yaml:"disable_rule_group_label"` @@ -268,6 +276,10 @@ 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.BoolVar(&cfg.EnableFederatedRules, "ruler.enable-federated-rules", false, "[Experimental] Enable federated rule groups. A federated rule group lists the tenants to query in its `src_tenants` field, while the resulting series and alerts always belong to the tenant owning the rule group. Requires -tenant-federation.enabled=true.") + f.Var(&cfg.AllowedFederatedTenants, "ruler.allowed-federated-tenants", "[Experimental] Comma separated list of tenants allowed to create federated rule groups. If specified, only these tenants can create federated rule groups, otherwise all tenants can.") + f.Var(&cfg.DisallowedFederatedTenants, "ruler.disallowed-federated-tenants", "[Experimental] Comma separated list of tenants that cannot create federated rule groups. If specified, a tenant that would normally be allowed to create federated rule groups is denied instead.") + 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") @@ -297,6 +309,9 @@ type MultiTenantManager interface { Stop() // ValidateRuleGroup validates a rulegroup ValidateRuleGroup(rulefmt.RuleGroup) []error + // ValidateFederatedRuleGroup checks that userID may own a federated rule group + // and returns its source tenants sorted and de-duplicated. + ValidateFederatedRuleGroup(userID string, srcTenants []string) ([]string, error) } // Ruler evaluates rules. @@ -1712,7 +1727,7 @@ func (r *Ruler) ListAllRules(w http.ResponseWriter, req *http.Request) { if userRules, err = r.store.LoadRuleGroups(ctx, userRules); err != nil { return errors.Wrapf(err, "failed to load ruler config for user %s", userID) } - data := map[string]map[string][]rulefmt.RuleGroup{userID: userRules[userID].Formatted()} + data := map[string]map[string][]rulespb.RuleGroup{userID: userRules[userID].FormattedRuleGroups()} select { case iter <- data: diff --git a/pkg/ruler/ruler_test.go b/pkg/ruler/ruler_test.go index cf42a970677..62a25d44eb2 100644 --- a/pkg/ruler/ruler_test.go +++ b/pkg/ruler/ruler_test.go @@ -2643,7 +2643,7 @@ func setupRuleGroupsStore(t *testing.T, ruleGroups []ruleGroupKey) (*objstore.In // "upload" rule groups for _, key := range ruleGroups { - desc := rulespb.ToProto(key.user, key.namespace, rulefmt.RuleGroup{Name: key.group}) + desc := rulespb.ToProto(key.user, key.namespace, rulespb.RuleGroup{Name: key.group}) require.NoError(t, rs.SetRuleGroup(context.Background(), key.user, key.namespace, desc)) } diff --git a/pkg/ruler/rulespb/compat.go b/pkg/ruler/rulespb/compat.go index a6a44736128..60a964bd0b4 100644 --- a/pkg/ruler/rulespb/compat.go +++ b/pkg/ruler/rulespb/compat.go @@ -11,8 +11,8 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" //lint:ignore faillint allowed to import other protobuf ) -// ToProto transforms a formatted prometheus rulegroup to a rule group protobuf -func ToProto(user string, namespace string, rl rulefmt.RuleGroup) *RuleGroupDesc { +// ToProto transforms a formatted rule group to a rule group protobuf +func ToProto(user string, namespace string, rl RuleGroup) *RuleGroupDesc { var queryOffset *time.Duration if rl.QueryOffset != nil { offset := time.Duration(*rl.QueryOffset) @@ -27,6 +27,7 @@ func ToProto(user string, namespace string, rl rulefmt.RuleGroup) *RuleGroupDesc Limit: int64(rl.Limit), QueryOffset: queryOffset, Labels: cortexpb.FromLabelsToLabelAdapters(labels.FromMap(rl.Labels)), + SrcTenants: rl.SrcTenants, } return &rg } @@ -48,8 +49,8 @@ func formattedRuleToProto(rls []rulefmt.Rule) []*RuleDesc { return rules } -// FromProto generates a rulefmt RuleGroup -func FromProto(rg *RuleGroupDesc) rulefmt.RuleGroup { +// FromProto generates a formatted rule group +func FromProto(rg *RuleGroupDesc) RuleGroup { var queryOffset *model.Duration if rg.QueryOffset != nil { offset := model.Duration(*rg.QueryOffset) @@ -85,5 +86,8 @@ func FromProto(rg *RuleGroupDesc) rulefmt.RuleGroup { formattedRuleGroup.Rules[i] = newRule } - return formattedRuleGroup + return RuleGroup{ + RuleGroup: formattedRuleGroup, + SrcTenants: rg.GetSrcTenants(), + } } diff --git a/pkg/ruler/rulespb/compat_test.go b/pkg/ruler/rulespb/compat_test.go index 414f6e84569..121e10c2c72 100644 --- a/pkg/ruler/rulespb/compat_test.go +++ b/pkg/ruler/rulespb/compat_test.go @@ -7,6 +7,8 @@ import ( "github.com/prometheus/common/model" "github.com/prometheus/prometheus/model/rulefmt" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" ) func TestProto(t *testing.T) { @@ -32,7 +34,7 @@ func TestProto(t *testing.T) { Labels: map[string]string{}, } - desc := ToProto("test", "namespace", rg) + desc := ToProto("test", "namespace", RuleGroup{RuleGroup: rg}) assert.Equal(t, len(rules), len(desc.Rules)) assert.Equal(t, 30*time.Second, *desc.QueryOffset) @@ -45,5 +47,45 @@ func TestProto(t *testing.T) { assert.Equal(t, time.Hour, ruleDesc.KeepFiringFor) formatted := FromProto(desc) - assert.Equal(t, rg, formatted) + assert.Equal(t, rg, formatted.RuleGroup) + assert.Empty(t, formatted.SrcTenants) +} + +func TestProtoRuleGroup(t *testing.T) { + rg := RuleGroup{ + Name: "group1", + Interval: model.Duration(time.Minute), + Rules: []rulefmt.Rule{{Record: "test_record", Expr: "test_expr", Labels: map[string]string{}, Annotations: map[string]string{}}}, + Labels: map[string]string{}, + SrcTenants: []string{"team-a", "team-b"}, + } + + desc := ToProto("test", "namespace", rg) + assert.Equal(t, []string{"team-a", "team-b"}, desc.SrcTenants) + assert.True(t, desc.IsFederated()) + + assert.Equal(t, rg, FromProto(desc)) + + // Groups without src tenants are not federated. + plain := ToProto("test", "namespace", RuleGroup{RuleGroup: rg.RuleGroup}) + assert.False(t, plain.IsFederated()) + assert.Empty(t, FromProto(plain).SrcTenants) +} + +func TestRuleGroupYAML(t *testing.T) { + in := "name: group1\ninterval: 1m\nsrc_tenants:\n - team-a\n - team-b\nrules:\n - record: test_record\n expr: test_expr\n" + + rg := RuleGroup{} + require.NoError(t, yaml.Unmarshal([]byte(in), &rg)) + assert.Equal(t, "group1", rg.Name) + assert.Equal(t, []string{"team-a", "team-b"}, rg.SrcTenants) + + out, err := yaml.Marshal(rg) + require.NoError(t, err) + assert.Equal(t, "name: group1\ninterval: 1m\nrules:\n - record: test_record\n expr: test_expr\nsrc_tenants:\n - team-a\n - team-b\n", string(out)) + + // src_tenants is omitted when empty, keeping the output of plain groups unchanged. + out, err = yaml.Marshal(RuleGroup{RuleGroup: rg.RuleGroup}) + require.NoError(t, err) + assert.NotContains(t, string(out), "src_tenants") } diff --git a/pkg/ruler/rulespb/custom.go b/pkg/ruler/rulespb/custom.go index d896afa1499..0b9389d2978 100644 --- a/pkg/ruler/rulespb/custom.go +++ b/pkg/ruler/rulespb/custom.go @@ -5,17 +5,36 @@ import "github.com/prometheus/prometheus/model/rulefmt" // RuleGroupList contains a set of rule groups type RuleGroupList []*RuleGroupDesc -// Formatted returns the rule group list as a set of formatted rule groups mapped -// by namespace +// RuleGroup is a rulefmt.RuleGroup extended with the Cortex-specific fields +// exposed through the ruler API. +type RuleGroup struct { + rulefmt.RuleGroup `yaml:",inline"` + // Tenants queried when evaluating the group. Empty means the owning tenant only. + SrcTenants []string `yaml:"src_tenants,omitempty"` +} + +// IsFederated returns true if the group queries data from explicitly listed tenants. +func (m *RuleGroupDesc) IsFederated() bool { + return len(m.GetSrcTenants()) > 0 +} + +// Formatted returns the rule group list as prometheus rule groups mapped by +// namespace, without the Cortex-specific fields. It is meant for the rule files +// loaded by the prometheus rules manager. func (l RuleGroupList) Formatted() map[string][]rulefmt.RuleGroup { ruleMap := map[string][]rulefmt.RuleGroup{} for _, g := range l { - if _, exists := ruleMap[g.Namespace]; !exists { - ruleMap[g.Namespace] = []rulefmt.RuleGroup{FromProto(g)} - continue - } - ruleMap[g.Namespace] = append(ruleMap[g.Namespace], FromProto(g)) + ruleMap[g.Namespace] = append(ruleMap[g.Namespace], FromProto(g).RuleGroup) + } + return ruleMap +} +// FormattedRuleGroups returns the rule group list as formatted rule groups +// mapped by namespace, keeping the Cortex-specific fields. +func (l RuleGroupList) FormattedRuleGroups() map[string][]RuleGroup { + ruleMap := map[string][]RuleGroup{} + for _, g := range l { + ruleMap[g.Namespace] = append(ruleMap[g.Namespace], FromProto(g)) } return ruleMap } diff --git a/pkg/ruler/rulespb/rules.pb.go b/pkg/ruler/rulespb/rules.pb.go index 6a7aef6327c..ce14f08f7d1 100644 --- a/pkg/ruler/rulespb/rules.pb.go +++ b/pkg/ruler/rulespb/rules.pb.go @@ -47,6 +47,9 @@ type RuleGroupDesc struct { Limit int64 `protobuf:"varint,10,opt,name=limit,proto3" json:"limit,omitempty"` QueryOffset *time.Duration `protobuf:"bytes,11,opt,name=queryOffset,proto3,stdduration" json:"queryOffset,omitempty"` Labels []github_com_cortexproject_cortex_pkg_cortexpb.LabelAdapter `protobuf:"bytes,12,rep,name=labels,proto3,customtype=github.com/cortexproject/cortex/pkg/cortexpb.LabelAdapter" json:"labels"` + // Tenants whose data is queried when evaluating this group. Empty means the + // owning tenant only. Non-empty makes this a federated rule group. + SrcTenants []string `protobuf:"bytes,13,rep,name=src_tenants,json=srcTenants,proto3" json:"src_tenants,omitempty"` } func (m *RuleGroupDesc) Reset() { *m = RuleGroupDesc{} } @@ -137,6 +140,13 @@ func (m *RuleGroupDesc) GetQueryOffset() *time.Duration { return nil } +func (m *RuleGroupDesc) GetSrcTenants() []string { + if m != nil { + return m.SrcTenants + } + return nil +} + // RuleDesc is a proto representation of a Prometheus Rule type RuleDesc struct { Expr string `protobuf:"bytes,1,opt,name=expr,proto3" json:"expr,omitempty"` @@ -223,42 +233,43 @@ func init() { func init() { proto.RegisterFile("rules.proto", fileDescriptor_8e722d3e922f0937) } var fileDescriptor_8e722d3e922f0937 = []byte{ - // 551 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x53, 0x31, 0x6f, 0xd3, 0x40, - 0x18, 0xf5, 0x61, 0xc7, 0x75, 0xce, 0x44, 0xad, 0x8e, 0x08, 0xb9, 0x05, 0x5d, 0xa2, 0x4a, 0x48, - 0x99, 0x1c, 0xa9, 0x88, 0x81, 0x01, 0xa1, 0x44, 0xa5, 0x48, 0x11, 0x12, 0xc8, 0x23, 0x42, 0xaa, - 0xce, 0xce, 0xd9, 0x98, 0x3a, 0x3e, 0x73, 0x3e, 0xa3, 0x66, 0xe3, 0x27, 0x30, 0xf2, 0x13, 0xf8, - 0x29, 0x1d, 0x18, 0xc2, 0x56, 0x31, 0x04, 0xe2, 0x2c, 0x88, 0xa9, 0x3f, 0x01, 0xdd, 0xd9, 0x86, - 0x00, 0x03, 0x65, 0x80, 0x29, 0xdf, 0xbb, 0x77, 0xef, 0xbe, 0xf7, 0xbd, 0x2f, 0x86, 0x36, 0x2f, - 0x12, 0x9a, 0xbb, 0x19, 0x67, 0x82, 0xa1, 0x96, 0x02, 0x7b, 0xdd, 0x88, 0x45, 0x4c, 0x9d, 0x0c, - 0x65, 0x55, 0x91, 0x7b, 0x38, 0x62, 0x2c, 0x4a, 0xe8, 0x50, 0x21, 0xbf, 0x08, 0x87, 0xd3, 0x82, - 0x13, 0x11, 0xb3, 0xb4, 0xe6, 0x77, 0x7f, 0xe5, 0x49, 0x3a, 0xaf, 0xa9, 0xbb, 0x51, 0x2c, 0x9e, - 0x17, 0xbe, 0x1b, 0xb0, 0xd9, 0x30, 0x60, 0x5c, 0xd0, 0xd3, 0x8c, 0xb3, 0x17, 0x34, 0x10, 0x35, - 0x1a, 0x66, 0x27, 0x51, 0x43, 0xf8, 0x75, 0x51, 0x49, 0xf7, 0xdf, 0xeb, 0xb0, 0xe3, 0x15, 0x09, - 0x7d, 0xc8, 0x59, 0x91, 0x1d, 0xd2, 0x3c, 0x40, 0x08, 0x1a, 0x29, 0x99, 0x51, 0x07, 0xf4, 0xc1, - 0xa0, 0xed, 0xa9, 0x1a, 0xdd, 0x84, 0x6d, 0xf9, 0x9b, 0x67, 0x24, 0xa0, 0xce, 0x15, 0x45, 0xfc, - 0x38, 0x40, 0xf7, 0xa1, 0x15, 0xa7, 0x82, 0xf2, 0x57, 0x24, 0x71, 0xf4, 0x3e, 0x18, 0xd8, 0x07, - 0xbb, 0x6e, 0x65, 0xd6, 0x6d, 0xcc, 0xba, 0x87, 0xf5, 0x30, 0x63, 0xeb, 0x6c, 0xd9, 0xd3, 0xde, - 0x7e, 0xea, 0x01, 0xef, 0xbb, 0x08, 0xdd, 0x82, 0x55, 0x32, 0x8e, 0xd1, 0xd7, 0x07, 0xf6, 0xc1, - 0xb6, 0xab, 0x90, 0x2b, 0x7d, 0x49, 0x4b, 0x5e, 0xc5, 0x4a, 0x67, 0x45, 0x4e, 0xb9, 0x63, 0x56, - 0xce, 0x64, 0x8d, 0x5c, 0xb8, 0xc5, 0x32, 0xf9, 0x70, 0xee, 0xb4, 0x95, 0xb8, 0xfb, 0x5b, 0xeb, - 0x51, 0x3a, 0xf7, 0x9a, 0x4b, 0xa8, 0x0b, 0x5b, 0x49, 0x3c, 0x8b, 0x85, 0x03, 0xfb, 0x60, 0xa0, - 0x7b, 0x15, 0x40, 0x0f, 0xa0, 0xfd, 0xb2, 0xa0, 0x7c, 0xfe, 0x38, 0x0c, 0x73, 0x2a, 0x1c, 0xfb, - 0x32, 0x43, 0x00, 0x35, 0xc4, 0xa6, 0x0e, 0xa5, 0xd0, 0x4c, 0x88, 0x4f, 0x93, 0xdc, 0xb9, 0xaa, - 0xbc, 0x5c, 0x73, 0x9b, 0xd0, 0xdd, 0x47, 0xf2, 0xfc, 0x09, 0x89, 0xf9, 0x78, 0x24, 0x03, 0xf8, - 0xb8, 0xec, 0xfd, 0xd5, 0xd2, 0x2a, 0xfd, 0x68, 0x4a, 0x32, 0x41, 0xb9, 0x57, 0x77, 0x99, 0x18, - 0x56, 0x6b, 0xc7, 0x9c, 0x18, 0xd6, 0xd6, 0x8e, 0x35, 0x31, 0x2c, 0x6b, 0xa7, 0xbd, 0xff, 0x41, - 0x87, 0x56, 0x13, 0x9b, 0xcc, 0x4b, 0x3e, 0xda, 0x6c, 0x52, 0xd6, 0xe8, 0x3a, 0x34, 0x39, 0x0d, - 0x18, 0x9f, 0xd6, 0x6b, 0xac, 0x91, 0xcc, 0x85, 0x24, 0x94, 0x0b, 0xb5, 0xc0, 0xb6, 0x57, 0x01, - 0x74, 0x07, 0xea, 0x21, 0xe3, 0x8e, 0x71, 0xf9, 0xa5, 0xca, 0xfb, 0x1b, 0x39, 0xb4, 0xfe, 0x47, - 0x0e, 0xe8, 0x14, 0xda, 0x24, 0x4d, 0x99, 0x20, 0xd5, 0x1f, 0xc1, 0xfc, 0xa7, 0x4d, 0x37, 0x5b, - 0xa1, 0x67, 0xb0, 0x73, 0x42, 0x69, 0x76, 0x14, 0xf3, 0x38, 0x8d, 0x8e, 0x18, 0x77, 0x3a, 0x7f, - 0x8a, 0xea, 0x86, 0x74, 0xf0, 0x75, 0xd9, 0xdb, 0x96, 0xba, 0xe3, 0x50, 0x09, 0x8f, 0x43, 0xc6, - 0x55, 0x7a, 0x3f, 0x3f, 0xa6, 0x36, 0xdb, 0x19, 0xdf, 0x5b, 0xac, 0xb0, 0x76, 0xbe, 0xc2, 0xda, - 0xc5, 0x0a, 0x83, 0xd7, 0x25, 0x06, 0xef, 0x4a, 0x0c, 0xce, 0x4a, 0x0c, 0x16, 0x25, 0x06, 0x9f, - 0x4b, 0x0c, 0xbe, 0x94, 0x58, 0xbb, 0x28, 0x31, 0x78, 0xb3, 0xc6, 0xda, 0x62, 0x8d, 0xb5, 0xf3, - 0x35, 0xd6, 0x9e, 0x6e, 0xa9, 0x6f, 0x26, 0xf3, 0x7d, 0x53, 0x79, 0xb8, 0xfd, 0x6d, 0x00, 0x39, - 0x21, 0xc0, 0xef, 0x8a, 0x04, 0x00, 0x00, + // 574 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xbc, 0x53, 0x41, 0x6b, 0xd4, 0x4c, + 0x18, 0xce, 0x7c, 0xc9, 0xa6, 0xc9, 0xe4, 0x5b, 0x5a, 0xc6, 0x22, 0xd3, 0x2a, 0xb3, 0x4b, 0x41, + 0xd8, 0x53, 0x16, 0x2a, 0x1e, 0x3c, 0x88, 0xb4, 0xd4, 0x0a, 0x45, 0x50, 0x82, 0x27, 0x11, 0xca, + 0x24, 0x9d, 0xc4, 0xd8, 0x34, 0x13, 0x27, 0x13, 0xe9, 0xde, 0xfc, 0x09, 0x1e, 0xfd, 0x09, 0xfe, + 0x94, 0x1e, 0xd7, 0x5b, 0xf1, 0xb0, 0xba, 0xd9, 0x8b, 0x08, 0x42, 0x7f, 0x82, 0xcc, 0x24, 0xd1, + 0x55, 0x0f, 0xd6, 0x83, 0x9e, 0xf6, 0x7d, 0xde, 0x67, 0x9e, 0x79, 0x9f, 0x79, 0xde, 0x0d, 0xf4, + 0x44, 0x95, 0xb1, 0xd2, 0x2f, 0x04, 0x97, 0x1c, 0xf5, 0x34, 0xd8, 0x5c, 0x4f, 0x78, 0xc2, 0x75, + 0x67, 0xac, 0xaa, 0x86, 0xdc, 0x24, 0x09, 0xe7, 0x49, 0xc6, 0xc6, 0x1a, 0x85, 0x55, 0x3c, 0x3e, + 0xaa, 0x04, 0x95, 0x29, 0xcf, 0x5b, 0x7e, 0xe3, 0x67, 0x9e, 0xe6, 0x93, 0x96, 0xba, 0x9d, 0xa4, + 0xf2, 0x59, 0x15, 0xfa, 0x11, 0x3f, 0x19, 0x47, 0x5c, 0x48, 0x76, 0x5a, 0x08, 0xfe, 0x9c, 0x45, + 0xb2, 0x45, 0xe3, 0xe2, 0x38, 0xe9, 0x88, 0xb0, 0x2d, 0x1a, 0xe9, 0xd6, 0x17, 0x13, 0xf6, 0x83, + 0x2a, 0x63, 0xf7, 0x05, 0xaf, 0x8a, 0x3d, 0x56, 0x46, 0x08, 0x41, 0x2b, 0xa7, 0x27, 0x0c, 0x83, + 0x21, 0x18, 0xb9, 0x81, 0xae, 0xd1, 0x75, 0xe8, 0xaa, 0xdf, 0xb2, 0xa0, 0x11, 0xc3, 0xff, 0x69, + 0xe2, 0x7b, 0x03, 0xdd, 0x85, 0x4e, 0x9a, 0x4b, 0x26, 0x5e, 0xd2, 0x0c, 0x9b, 0x43, 0x30, 0xf2, + 0xb6, 0x37, 0xfc, 0xc6, 0xac, 0xdf, 0x99, 0xf5, 0xf7, 0xda, 0xc7, 0xec, 0x3a, 0x67, 0xb3, 0x81, + 0xf1, 0xe6, 0xc3, 0x00, 0x04, 0xdf, 0x44, 0xe8, 0x06, 0x6c, 0x92, 0xc1, 0xd6, 0xd0, 0x1c, 0x79, + 0xdb, 0xab, 0xbe, 0x46, 0xbe, 0xf2, 0xa5, 0x2c, 0x05, 0x0d, 0xab, 0x9c, 0x55, 0x25, 0x13, 0xd8, + 0x6e, 0x9c, 0xa9, 0x1a, 0xf9, 0x70, 0x85, 0x17, 0xea, 0xe2, 0x12, 0xbb, 0x5a, 0xbc, 0xfe, 0xcb, + 0xe8, 0x9d, 0x7c, 0x12, 0x74, 0x87, 0xd0, 0x3a, 0xec, 0x65, 0xe9, 0x49, 0x2a, 0x31, 0x1c, 0x82, + 0x91, 0x19, 0x34, 0x00, 0xdd, 0x83, 0xde, 0x8b, 0x8a, 0x89, 0xc9, 0xc3, 0x38, 0x2e, 0x99, 0xc4, + 0xde, 0x65, 0x1e, 0x01, 0xf4, 0x23, 0x96, 0x75, 0x28, 0x87, 0x76, 0x46, 0x43, 0x96, 0x95, 0xf8, + 0x7f, 0xed, 0xe5, 0x8a, 0xdf, 0x85, 0xee, 0x3f, 0x50, 0xfd, 0x47, 0x34, 0x15, 0xbb, 0x3b, 0x2a, + 0x80, 0xf7, 0xb3, 0xc1, 0x1f, 0x2d, 0xad, 0xd1, 0xef, 0x1c, 0xd1, 0x42, 0x32, 0x11, 0xb4, 0x53, + 0xd0, 0x00, 0x7a, 0xa5, 0x88, 0x0e, 0x25, 0xcb, 0x69, 0x2e, 0x4b, 0xdc, 0x1f, 0x9a, 0x23, 0x37, + 0x80, 0xa5, 0x88, 0x1e, 0x37, 0x9d, 0x03, 0xcb, 0xe9, 0xad, 0xd9, 0x07, 0x96, 0xb3, 0xb2, 0xe6, + 0x1c, 0x58, 0x8e, 0xb3, 0xe6, 0x6e, 0xbd, 0x33, 0xa1, 0xd3, 0xe5, 0xaa, 0x02, 0x55, 0x53, 0xbb, + 0x55, 0xab, 0x1a, 0x5d, 0x85, 0xb6, 0x60, 0x11, 0x17, 0x47, 0xed, 0x9e, 0x5b, 0xa4, 0x82, 0xa3, + 0x19, 0x13, 0x52, 0x6f, 0xd8, 0x0d, 0x1a, 0x80, 0x6e, 0x41, 0x33, 0xe6, 0x02, 0x5b, 0x97, 0xdf, + 0xba, 0x3a, 0xbf, 0x14, 0x54, 0xef, 0x9f, 0x04, 0x75, 0x0a, 0x3d, 0x9a, 0xe7, 0x5c, 0xd2, 0xe6, + 0x9f, 0x62, 0xff, 0xd5, 0xa1, 0xcb, 0xa3, 0xd0, 0x53, 0xd8, 0x3f, 0x66, 0xac, 0xd8, 0x4f, 0x45, + 0x9a, 0x27, 0xfb, 0x5c, 0xe0, 0xfe, 0xef, 0xa2, 0xba, 0xa6, 0x1c, 0x7c, 0x9e, 0x0d, 0x56, 0x95, + 0xee, 0x30, 0xd6, 0xc2, 0xc3, 0x98, 0x0b, 0x9d, 0xde, 0x8f, 0x97, 0xe9, 0xcd, 0xf6, 0x77, 0xef, + 0x4c, 0xe7, 0xc4, 0x38, 0x9f, 0x13, 0xe3, 0x62, 0x4e, 0xc0, 0xab, 0x9a, 0x80, 0xb7, 0x35, 0x01, + 0x67, 0x35, 0x01, 0xd3, 0x9a, 0x80, 0x8f, 0x35, 0x01, 0x9f, 0x6a, 0x62, 0x5c, 0xd4, 0x04, 0xbc, + 0x5e, 0x10, 0x63, 0xba, 0x20, 0xc6, 0xf9, 0x82, 0x18, 0x4f, 0x56, 0xf4, 0x47, 0x55, 0x84, 0xa1, + 0xad, 0x3d, 0xdc, 0xfc, 0x3a, 0x00, 0x82, 0x8a, 0xee, 0x75, 0xab, 0x04, 0x00, 0x00, } func (this *RuleGroupDesc) Equal(that interface{}) bool { @@ -328,6 +339,14 @@ func (this *RuleGroupDesc) Equal(that interface{}) bool { return false } } + if len(this.SrcTenants) != len(that1.SrcTenants) { + return false + } + for i := range this.SrcTenants { + if this.SrcTenants[i] != that1.SrcTenants[i] { + return false + } + } return true } func (this *RuleDesc) Equal(that interface{}) bool { @@ -386,7 +405,7 @@ func (this *RuleGroupDesc) GoString() string { if this == nil { return "nil" } - s := make([]string, 0, 13) + s := make([]string, 0, 14) s = append(s, "&rulespb.RuleGroupDesc{") s = append(s, "Name: "+fmt.Sprintf("%#v", this.Name)+",\n") s = append(s, "Namespace: "+fmt.Sprintf("%#v", this.Namespace)+",\n") @@ -401,6 +420,7 @@ func (this *RuleGroupDesc) GoString() string { s = append(s, "Limit: "+fmt.Sprintf("%#v", this.Limit)+",\n") s = append(s, "QueryOffset: "+fmt.Sprintf("%#v", this.QueryOffset)+",\n") s = append(s, "Labels: "+fmt.Sprintf("%#v", this.Labels)+",\n") + s = append(s, "SrcTenants: "+fmt.Sprintf("%#v", this.SrcTenants)+",\n") s = append(s, "}") return strings.Join(s, "") } @@ -448,6 +468,15 @@ func (m *RuleGroupDesc) MarshalToSizedBuffer(dAtA []byte) (int, error) { _ = i var l int _ = l + if len(m.SrcTenants) > 0 { + for iNdEx := len(m.SrcTenants) - 1; iNdEx >= 0; iNdEx-- { + i -= len(m.SrcTenants[iNdEx]) + copy(dAtA[i:], m.SrcTenants[iNdEx]) + i = encodeVarintRules(dAtA, i, uint64(len(m.SrcTenants[iNdEx]))) + i-- + dAtA[i] = 0x6a + } + } if len(m.Labels) > 0 { for iNdEx := len(m.Labels) - 1; iNdEx >= 0; iNdEx-- { { @@ -681,6 +710,12 @@ func (m *RuleGroupDesc) Size() (n int) { n += 1 + l + sovRules(uint64(l)) } } + if len(m.SrcTenants) > 0 { + for _, s := range m.SrcTenants { + l = len(s) + n += 1 + l + sovRules(uint64(l)) + } + } return n } @@ -751,6 +786,7 @@ func (this *RuleGroupDesc) String() string { `Limit:` + fmt.Sprintf("%v", this.Limit) + `,`, `QueryOffset:` + strings.Replace(fmt.Sprintf("%v", this.QueryOffset), "Duration", "durationpb.Duration", 1) + `,`, `Labels:` + fmt.Sprintf("%v", this.Labels) + `,`, + `SrcTenants:` + fmt.Sprintf("%v", this.SrcTenants) + `,`, `}`, }, "") return s @@ -1094,6 +1130,38 @@ func (m *RuleGroupDesc) Unmarshal(dAtA []byte) error { return err } iNdEx = postIndex + case 13: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field SrcTenants", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowRules + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthRules + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return ErrInvalidLengthRules + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.SrcTenants = append(m.SrcTenants, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipRules(dAtA[iNdEx:]) diff --git a/pkg/ruler/rulespb/rules.proto b/pkg/ruler/rulespb/rules.proto index f60a0a00fbe..41c35937ea3 100644 --- a/pkg/ruler/rulespb/rules.proto +++ b/pkg/ruler/rulespb/rules.proto @@ -34,6 +34,9 @@ message RuleGroupDesc { (gogoproto.nullable) = false, (gogoproto.customtype) = "github.com/cortexproject/cortex/pkg/cortexpb.LabelAdapter" ]; + // Tenants whose data is queried when evaluating this group. Empty means the + // owning tenant only. Non-empty makes this a federated rule group. + repeated string src_tenants = 13; } // RuleDesc is a proto representation of a Prometheus Rule diff --git a/pkg/ruler/rulestore/bucketclient/bucket_client_test.go b/pkg/ruler/rulestore/bucketclient/bucket_client_test.go index 343a3354d7f..33c0dc92d9c 100644 --- a/pkg/ruler/rulestore/bucketclient/bucket_client_test.go +++ b/pkg/ruler/rulestore/bucketclient/bucket_client_test.go @@ -39,7 +39,7 @@ func TestListRules(t *testing.T) { } for _, g := range groups { - desc := rulespb.ToProto(g.user, g.namespace, g.ruleGroup) + desc := rulespb.ToProto(g.user, g.namespace, rulespb.RuleGroup{RuleGroup: g.ruleGroup}) require.NoError(t, rs.SetRuleGroup(context.Background(), g.user, g.namespace, desc)) } @@ -119,7 +119,7 @@ func TestLoadPartialRules(t *testing.T) { } for _, g := range groups { - desc := rulespb.ToProto(g.user, g.namespace, g.ruleGroup) + desc := rulespb.ToProto(g.user, g.namespace, rulespb.RuleGroup{RuleGroup: g.ruleGroup}) require.NoError(t, bucketStore.SetRuleGroup(context.Background(), g.user, g.namespace, desc)) } allGroups, err := bucketStore.ListAllRuleGroups(context.Background()) @@ -149,7 +149,7 @@ func TestLoadRules(t *testing.T) { } for _, g := range groups { - desc := rulespb.ToProto(g.user, g.namespace, g.ruleGroup) + desc := rulespb.ToProto(g.user, g.namespace, rulespb.RuleGroup{RuleGroup: g.ruleGroup}) require.NoError(t, rs.SetRuleGroup(context.Background(), g.user, g.namespace, desc)) } @@ -217,7 +217,7 @@ func TestDelete(t *testing.T) { } for _, g := range groups { - desc := rulespb.ToProto(g.user, g.namespace, g.ruleGroup) + desc := rulespb.ToProto(g.user, g.namespace, rulespb.RuleGroup{RuleGroup: g.ruleGroup}) require.NoError(t, rs.SetRuleGroup(context.Background(), g.user, g.namespace, desc)) } diff --git a/pkg/ruler/rulestore/configdb/store.go b/pkg/ruler/rulestore/configdb/store.go index 3ef63478e0f..d2918cbb365 100644 --- a/pkg/ruler/rulestore/configdb/store.go +++ b/pkg/ruler/rulestore/configdb/store.go @@ -70,7 +70,7 @@ func (c *ConfigRuleStore) ListAllRuleGroups(ctx context.Context) (map[string]rul } for file, rgs := range rMap { for _, rg := range rgs.Groups { - userRules = append(userRules, rulespb.ToProto(user, file, rg)) + userRules = append(userRules, rulespb.ToProto(user, file, rulespb.RuleGroup{RuleGroup: rg})) } } c.ruleGroupList[user] = userRules diff --git a/pkg/ruler/rulestore/local/local.go b/pkg/ruler/rulestore/local/local.go index b196ad6b3ea..a3e739399c7 100644 --- a/pkg/ruler/rulestore/local/local.go +++ b/pkg/ruler/rulestore/local/local.go @@ -186,7 +186,7 @@ func (l *Client) loadAllRulesGroupsForUserAndNamespace(_ context.Context, userID var list rulespb.RuleGroupList for _, group := range rulegroups.Groups { - desc := rulespb.ToProto(userID, namespace, group) + desc := rulespb.ToProto(userID, namespace, rulespb.RuleGroup{RuleGroup: group}) list = append(list, desc) } diff --git a/pkg/ruler/rulestore/local/local_test.go b/pkg/ruler/rulestore/local/local_test.go index 8cd5345d9b6..6ad1dc2ba9c 100644 --- a/pkg/ruler/rulestore/local/local_test.go +++ b/pkg/ruler/rulestore/local/local_test.go @@ -79,7 +79,7 @@ func TestClient_LoadAllRuleGroups(t *testing.T) { require.Equal(t, 2, len(actual)) // We rely on the fact that files are parsed in alphabetical order, and our namespace1 < namespace2. - require.Equal(t, rulespb.ToProto(u, namespace1, ruleGroups.Groups[0]), actual[0]) - require.Equal(t, rulespb.ToProto(u, namespace2, ruleGroups.Groups[0]), actual[1]) + require.Equal(t, rulespb.ToProto(u, namespace1, rulespb.RuleGroup{RuleGroup: ruleGroups.Groups[0]}), actual[0]) + require.Equal(t, rulespb.ToProto(u, namespace2, rulespb.RuleGroup{RuleGroup: ruleGroups.Groups[0]}), actual[1]) } } diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index 2fa4d4eaf97..78a79fade0e 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -7351,6 +7351,11 @@ "type": "string", "x-cli-flag": "ruler.alertmanager-url" }, + "allowed_federated_tenants": { + "description": "[Experimental] Comma separated list of tenants allowed to create federated rule groups. If specified, only these tenants can create federated rule groups, otherwise all tenants can.", + "type": "string", + "x-cli-flag": "ruler.allowed-federated-tenants" + }, "api_deduplicate_rules": { "default": false, "description": "EXPERIMENTAL: Remove duplicate rules in the prometheus rules and alerts API response. If there are duplicate rules the rule with the latest evaluation timestamp will be kept.", @@ -7374,6 +7379,11 @@ "type": "string", "x-cli-flag": "ruler.disabled-tenants" }, + "disallowed_federated_tenants": { + "description": "[Experimental] Comma separated list of tenants that cannot create federated rule groups. If specified, a tenant that would normally be allowed to create federated rule groups is denied instead.", + "type": "string", + "x-cli-flag": "ruler.disallowed-federated-tenants" + }, "enable_alertmanager_discovery": { "default": false, "description": "Use DNS SRV records to discover Alertmanager hosts.", @@ -7386,6 +7396,12 @@ "type": "boolean", "x-cli-flag": "ruler.enable-api" }, + "enable_federated_rules": { + "default": false, + "description": "[Experimental] Enable federated rule groups. A federated rule group lists the tenants to query in its `src_tenants` field, while the resulting series and alerts always belong to the tenant owning the rule group. Requires -tenant-federation.enabled=true.", + "type": "boolean", + "x-cli-flag": "ruler.enable-federated-rules" + }, "enable_ha_evaluation": { "default": false, "description": "Enable high availability",