From 81abd84ab77d5c8fcfc5080a52c52c6b9ec138cf Mon Sep 17 00:00:00 2001 From: Yuchen Wang Date: Thu, 15 Jan 2026 17:14:15 -0800 Subject: [PATCH] add legacy tenant attribution in router --- cmd/thanos/receive.go | 33 + pkg/receive/handler.go | 39 +- pkg/receive/handler_test.go | 7 +- pkg/receive/tenant_attribution.go | 938 +++++++++++++++++++++++++ pkg/receive/tenant_attribution_test.go | 460 ++++++++++++ 5 files changed, 1468 insertions(+), 9 deletions(-) create mode 100644 pkg/receive/tenant_attribution.go create mode 100644 pkg/receive/tenant_attribution_test.go diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go index fc95d7f4667..b2a580e8a6d 100644 --- a/cmd/thanos/receive.go +++ b/cmd/thanos/receive.go @@ -302,6 +302,23 @@ func runReceive( return errors.Wrap(err, "creating limiter") } + // Initialize tenant attributor if configured. + var tenantAttributor *receive.TenantAttributor + if conf.tenantRulesPath != "" { + tenantAttributor, err = receive.NewTenantAttributor( + conf.tenantRulesPath, + conf.defaultTenantID, + conf.verifyTenantAttribution, + reg, + log.With(logger, "component", "tenant-attributor"), + ) + if err != nil { + return errors.Wrap(err, "creating tenant attributor") + } + } else if conf.verifyTenantAttribution { + level.Warn(logger).Log("msg", "receive.verify-tenant-attribution is set but receive.tenant-rules is not configured, ignoring") + } + webHandler := receive.NewHandler(log.With(logger, "component", "receive-handler"), &receive.Options{ Writer: writer, ListenAddress: conf.rwAddress, @@ -324,6 +341,7 @@ func runReceive( Limiter: limiter, AsyncForwardWorkerCount: conf.asyncForwardWorkerCount, ReplicationProtocol: receive.ReplicationProtocol(conf.replicationProtocol), + TenantAttributor: tenantAttributor, }) { @@ -1004,6 +1022,9 @@ type receiveConfig struct { writerInterning bool splitTenantLabelName string + tenantRulesPath string + verifyTenantAttribution bool + hashFunc string ignoreBlockSize bool @@ -1089,6 +1110,18 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) { cmd.Flag("receive.split-tenant-label-name", "Label name through which the request will be split into multiple tenants. This takes precedence over the HTTP header.").Default("").StringVar(&rc.splitTenantLabelName) + cmd.Flag("receive.tenant-rules", + "Path to YAML file containing tenant attribution rules. "+ + "Rules use M3-style filter syntax (e.g., 'job:prometheus namespace:prod*'). "+ + "First matching rule wins. Falls back to --receive.default-tenant-id if no rule matches."). + PlaceHolder("").StringVar(&rc.tenantRulesPath) + + cmd.Flag("receive.verify-tenant-attribution", + "When set with --receive.tenant-rules, compares attributed tenant with HTTP header tenant "+ + "and exposes metrics (thanos_receive_tenant_attribution_matches_total and "+ + "thanos_receive_tenant_attribution_mismatches_total). Does not change routing behavior."). + Default("false").BoolVar(&rc.verifyTenantAttribution) + cmd.Flag("receive.tenant-label-name", "Label name through which the tenant will be announced.").Default(tenancy.DefaultTenantLabel).StringVar(&rc.tenantLabelName) cmd.Flag("receive.replica-header", "HTTP header specifying the replica number of a write request.").Default(receive.DefaultReplicaHeader).StringVar(&rc.replicaHeader) diff --git a/pkg/receive/handler.go b/pkg/receive/handler.go index cfedb288ae5..e6f69c092a8 100644 --- a/pkg/receive/handler.go +++ b/pkg/receive/handler.go @@ -117,6 +117,7 @@ type Options struct { Limiter *Limiter AsyncForwardWorkerCount uint ReplicationProtocol ReplicationProtocol + TenantAttributor *TenantAttributor } // Handler serves a Prometheus remote write receiving HTTP endpoint. @@ -529,6 +530,9 @@ func (h *Handler) receiveHTTP(w http.ResponseWriter, r *http.Request) { span.SetTag("receiver.mode", string(h.receiverMode)) defer span.Finish() + // Check if tenant header is present before calling GetTenantFromHTTP + httpHeaderPresent := r.Header.Get(h.options.TenantHeader) != "" || r.Header.Get(tenancy.DefaultTenantHeader) != "" + tenantHTTP, err := tenancy.GetTenantFromHTTP(r, h.options.TenantHeader, h.options.DefaultTenantID, h.options.TenantField) if err != nil { level.Error(h.logger).Log("msg", "error getting tenant from HTTP", "err", err) @@ -644,7 +648,7 @@ func (h *Handler) receiveHTTP(w http.ResponseWriter, r *http.Request) { } responseStatusCode := http.StatusOK - tenantStats, err := h.handleRequest(ctx, rep, tenantHTTP, &wreq) + tenantStats, err := h.handleRequest(ctx, rep, tenantHTTP, httpHeaderPresent, &wreq) if err != nil { level.Debug(tLogger).Log("msg", "failed to handle request", "err", err.Error()) switch errors.Cause(err) { @@ -703,7 +707,7 @@ type requestStats struct { type tenantRequestStats map[string]requestStats -func (h *Handler) handleRequest(ctx context.Context, rep uint64, tenantHTTP string, wreq *prompb.WriteRequest) (tenantRequestStats, error) { +func (h *Handler) handleRequest(ctx context.Context, rep uint64, tenantHTTP string, httpHeaderPresent bool, wreq *prompb.WriteRequest) (tenantRequestStats, error) { tLogger := log.With(h.logger, "tenantHTTP", tenantHTTP) // This replica value is used to detect cycles in cyclic topologies. @@ -732,7 +736,7 @@ func (h *Handler) handleRequest(ctx context.Context, rep uint64, tenantHTTP stri // Forward any time series as necessary. All time series // destined for the local node will be written to the receiver. // Time series will be replicated as necessary. - return h.forward(ctx, tenantHTTP, r, wreq) + return h.forward(ctx, tenantHTTP, httpHeaderPresent, r, wreq) } // forward accepts a write request, batches its time series by @@ -743,7 +747,7 @@ func (h *Handler) handleRequest(ctx context.Context, rep uint64, tenantHTTP stri // unless the request needs to be replicated. // The function only returns when all requests have finished // or the context is canceled. -func (h *Handler) forward(ctx context.Context, tenantHTTP string, r replica, wreq *prompb.WriteRequest) (tenantRequestStats, error) { +func (h *Handler) forward(ctx context.Context, tenantHTTP string, httpHeaderPresent bool, r replica, wreq *prompb.WriteRequest) (tenantRequestStats, error) { span, ctx := tracing.StartSpan(ctx, "receive_fanout_forward") defer span.Finish() @@ -761,6 +765,7 @@ func (h *Handler) forward(ctx context.Context, tenantHTTP string, r replica, wre writeRequest: wreq, replicas: replicas, alreadyReplicated: r.replicated, + httpHeaderPresent: httpHeaderPresent, } return h.fanoutForward(ctx, params) @@ -771,6 +776,7 @@ type remoteWriteParams struct { writeRequest *prompb.WriteRequest replicas []uint64 alreadyReplicated bool + httpHeaderPresent bool // true if tenant came from HTTP header (not default) } func (h *Handler) gatherWriteStats(rf int, writes ...map[endpointReplica]map[string]trackedSeries) tenantRequestStats { @@ -831,7 +837,7 @@ func (h *Handler) fanoutForward(ctx context.Context, params remoteWriteParams) ( } requestLogger := log.With(h.logger, logTags...) - localWrites, remoteWrites, err := h.distributeTimeseriesToReplicas(params.tenant, params.replicas, params.writeRequest.Timeseries) + localWrites, remoteWrites, err := h.distributeTimeseriesToReplicas(params.tenant, params.httpHeaderPresent, params.replicas, params.writeRequest.Timeseries) if err != nil { level.Error(requestLogger).Log("msg", "failed to distribute timeseries to replicas", "err", err) return stats, err @@ -914,6 +920,7 @@ func (h *Handler) fanoutForward(ctx context.Context, params remoteWriteParams) ( // series that should be written to remote nodes. func (h *Handler) distributeTimeseriesToReplicas( tenantHTTP string, + httpHeaderPresent bool, replicas []uint64, timeseries []prompb.TimeSeries, ) (map[endpointReplica]map[string]trackedSeries, map[endpointReplica]map[string]trackedSeries, error) { @@ -935,6 +942,25 @@ func (h *Handler) distributeTimeseriesToReplicas( } } + // Tenant attribution logic + if h.options.TenantAttributor != nil { + lbls := labelpb.ZLabelsToPromLabels(ts.Labels) + attributedTenant := h.options.TenantAttributor.GetTenantFromLabels(lbls) + + if h.options.TenantAttributor.IsVerifyMode() { + // VERIFICATION MODE: compute attribution for metrics only + // Compare attributed tenant with HTTP tenant + h.options.TenantAttributor.RecordVerification(attributedTenant, tenantHTTP) + // DO NOT modify tenant - keep using HTTP tenant for routing + } else { + // ATTRIBUTION MODE: only attribute if no HTTP header was present + if !httpHeaderPresent { + tenant = attributedTenant + } + // If HTTP header present, use that (tenant already set correctly) + } + } + for _, rn := range replicas { endpoint, err := h.hashring.GetN(tenant, &ts, rn) if err != nil { @@ -1125,7 +1151,8 @@ func (h *Handler) RemoteWrite(ctx context.Context, r *storepb.WriteRequest) (*st span, ctx := tracing.StartSpan(ctx, "receive_grpc") defer span.Finish() - _, err := h.handleRequest(ctx, uint64(r.Replica), r.Tenant, &prompb.WriteRequest{Timeseries: r.Timeseries}) + // For gRPC requests, tenant is explicitly provided, treat as if HTTP header was present + _, err := h.handleRequest(ctx, uint64(r.Replica), r.Tenant, true, &prompb.WriteRequest{Timeseries: r.Timeseries}) if err != nil { level.Debug(h.logger).Log("msg", "failed to handle request", "err", err) } diff --git a/pkg/receive/handler_test.go b/pkg/receive/handler_test.go index 3be7f476de6..55257639931 100644 --- a/pkg/receive/handler_test.go +++ b/pkg/receive/handler_test.go @@ -1862,6 +1862,7 @@ func TestDistributeSeries(t *testing.T) { _, remote, err := h.distributeTimeseriesToReplicas( "foo", + true, []uint64{0}, []prompb.TimeSeries{ { @@ -1920,7 +1921,7 @@ func TestHandlerFlippingHashrings(t *testing.T) { return } - _, err := h.handleRequest(ctx, 0, "test", &prompb.WriteRequest{ + _, err := h.handleRequest(ctx, 0, "test", true, &prompb.WriteRequest{ Timeseries: []prompb.TimeSeries{ { Labels: labelpb.ZLabelsFromPromLabels(labels.FromStrings("foo", "bar")), @@ -2004,7 +2005,7 @@ func TestIngestorRestart(t *testing.T) { }, } - stats, err := client.handleRequest(ctx, 0, "test", data) + stats, err := client.handleRequest(ctx, 0, "test", true, data) require.NoError(t, err) require.Equal(t, tenantRequestStats{ "test": requestStats{timeseries: 1, totalSamples: 1}, @@ -2019,7 +2020,7 @@ func TestIngestorRestart(t *testing.T) { iter, errs := 10, 0 for i := 0; i < iter; i++ { - _, err = client.handleRequest(ctx, 0, "test", data) + _, err = client.handleRequest(ctx, 0, "test", true, data) if err != nil { require.Error(t, errUnavailable, err) errs++ diff --git a/pkg/receive/tenant_attribution.go b/pkg/receive/tenant_attribution.go new file mode 100644 index 00000000000..b84b4452192 --- /dev/null +++ b/pkg/receive/tenant_attribution.go @@ -0,0 +1,938 @@ +// Copyright (c) The Thanos Authors. +// Licensed under the Apache License 2.0. + +package receive + +import ( + "bytes" + "errors" + "fmt" + "os" + "sort" + "strings" + + "github.com/go-kit/log" + "github.com/go-kit/log/level" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/prometheus/model/labels" + "gopkg.in/yaml.v2" +) + +// TenantRuleConfig represents a single tenant rule in the configuration file. +type TenantRuleConfig struct { + Filter string `yaml:"filter"` + Tenant string `yaml:"tenant"` +} + +// TenantRule is a parsed tenant rule with a compiled filter. +type TenantRule struct { + Filter TagsFilter + Tenant string +} + +// TenantAttributor handles tenant attribution based on time series labels. +type TenantAttributor struct { + rules []TenantRule + defaultTenant string + logger log.Logger + + // Metrics (only registered in verify mode) + verifyMode bool + attributionMatches prometheus.Counter + attributionMismatches prometheus.Counter +} + +// NewTenantAttributor creates a new TenantAttributor from a config file. +func NewTenantAttributor( + configPath string, + defaultTenant string, + verifyMode bool, + reg prometheus.Registerer, + logger log.Logger, +) (*TenantAttributor, error) { + ta := &TenantAttributor{ + defaultTenant: defaultTenant, + logger: logger, + verifyMode: verifyMode, + } + + // Register metrics only in verify mode + if verifyMode && reg != nil { + ta.attributionMatches = promauto.With(reg).NewCounter(prometheus.CounterOpts{ + Namespace: "thanos", + Subsystem: "receive", + Name: "tenant_attribution_matches_total", + Help: "Total number of time series where attributed tenant matches HTTP header tenant.", + }) + ta.attributionMismatches = promauto.With(reg).NewCounter(prometheus.CounterOpts{ + Namespace: "thanos", + Subsystem: "receive", + Name: "tenant_attribution_mismatches_total", + Help: "Total number of time series where attributed tenant differs from HTTP header tenant.", + }) + } + + // Load and parse config file + if configPath == "" { + return nil, errors.New("tenant rules config path is required") + } + + configData, err := os.ReadFile(configPath) + if err != nil { + return nil, fmt.Errorf("reading tenant rules config: %w", err) + } + + var ruleConfigs []TenantRuleConfig + if err := yaml.UnmarshalStrict(configData, &ruleConfigs); err != nil { + return nil, fmt.Errorf("parsing tenant rules config: %w", err) + } + + // Parse and compile each rule + for i, rc := range ruleConfigs { + if rc.Filter == "" { + return nil, fmt.Errorf("rule %d: filter is required", i) + } + if rc.Tenant == "" { + return nil, fmt.Errorf("rule %d: tenant is required", i) + } + + filterValues, err := ParseTagFilterValueMap(rc.Filter) + if err != nil { + return nil, fmt.Errorf("rule %d: parsing filter %q: %w", i, rc.Filter, err) + } + + filter, err := NewTagsFilter(filterValues, Conjunction, TagsFilterOptions{}) + if err != nil { + return nil, fmt.Errorf("rule %d: creating filter %q: %w", i, rc.Filter, err) + } + + ta.rules = append(ta.rules, TenantRule{ + Filter: filter, + Tenant: rc.Tenant, + }) + + level.Info(logger).Log("msg", "loaded tenant rule", "index", i, "filter", rc.Filter, "tenant", rc.Tenant) + } + + level.Info(logger).Log("msg", "tenant attributor initialized", "rules", len(ta.rules), "default_tenant", defaultTenant, "verify_mode", verifyMode) + return ta, nil +} + +// GetTenantFromLabels returns the tenant for the given labels based on rules. +// First matching rule wins. Returns defaultTenant if no rule matches. +func (ta *TenantAttributor) GetTenantFromLabels(lbls labels.Labels) string { + for _, rule := range ta.rules { + if rule.Filter.MatchLabels(lbls) { + return rule.Tenant + } + } + return ta.defaultTenant +} + +// RecordVerification records match/mismatch metrics for verification mode. +// httpTenant is the tenant from HTTP header (or default if no header). +func (ta *TenantAttributor) RecordVerification(attributedTenant, httpTenant string) { + if !ta.verifyMode { + return + } + if attributedTenant == httpTenant { + if ta.attributionMatches != nil { + ta.attributionMatches.Inc() + } + } else { + if ta.attributionMismatches != nil { + ta.attributionMismatches.Inc() + } + } +} + +// IsVerifyMode returns true if the attributor is in verification mode. +func (ta *TenantAttributor) IsVerifyMode() bool { + return ta.verifyMode +} + +// ============================================================================= +// Filter Implementation (Ported from M3) +// ============================================================================= + +var ( + errInvalidFilterPattern = errors.New("invalid filter pattern defined") +) + +// LogicalOp is a logical operator. +type LogicalOp string + +// chainSegment is the part of the pattern that the chain represents. +type chainSegment int + +const ( + // Conjunction is logical AND. + Conjunction LogicalOp = "&&" + // Disjunction is logical OR. + Disjunction LogicalOp = "||" + + middle chainSegment = iota + start + end + + wildcardChar = '*' + negationChar = '!' + singleAnyChar = '?' + singleRangeStartChar = '[' + singleRangeEndChar = ']' + rangeChar = '-' + multiRangeStartChar = '{' + multiRangeEndChar = '}' + invalidNestedChars = "?[{" + + // tagFilterListSeparator splits key:value pairs in a tag filter list. + tagFilterListSeparator = " " +) + +var ( + multiRangeSplit = []byte(",") + + // defaultFilterSeparator represents the default filter separator with no negation. + defaultFilterSeparator = tagFilterSeparator{str: ":", negate: false} + // databricksFilterSeparator is an alternative separator. + databricksFilterSeparator = tagFilterSeparator{str: "#", negate: false} + + // validFilterSeparators represent a list of valid filter separators. + // NB: the separators here are tried in order during parsing. + validFilterSeparators = []tagFilterSeparator{ + databricksFilterSeparator, + defaultFilterSeparator, + } +) + +type tagFilterSeparator struct { + str string + negate bool +} + +// FilterValue contains the filter pattern and a boolean flag indicating +// whether the filter should be negated. +type FilterValue struct { + Pattern string + Negate bool +} + +// TagFilterValueMapKey represents a key in the tag filter value map. +type TagFilterValueMapKey struct { + Name string + Exclude bool +} + +// TagFilterValueMap is a map containing mappings from tag names to filter values. +type TagFilterValueMap map[TagFilterValueMapKey]FilterValue + +// ParseTagFilterValueMap parses the input string and creates a tag filter value map. +func ParseTagFilterValueMap(str string) (TagFilterValueMap, error) { + trimmed := strings.TrimSpace(str) + tagPairs := strings.Split(trimmed, tagFilterListSeparator) + res := make(map[TagFilterValueMapKey]FilterValue, len(tagPairs)) + for _, p := range tagPairs { + sanitized := strings.TrimSpace(p) + if sanitized == "" { + continue + } + parts, separator, err := parseTagFilter(sanitized) + if err != nil { + return nil, err + } + // NB: we do not allow duplicate tags at the moment. + exclude := parts[0][0] == negationChar + name := parts[0] + if exclude { + name = parts[0][1:] + } + key := TagFilterValueMapKey{Name: name, Exclude: exclude} + _, exists := res[key] + if exists { + return nil, fmt.Errorf("invalid filter %s: duplicate tag %s found", str, parts[0]) + } + if key.Exclude { + // For exclude rules, it doesn't make sense to have a value filter pattern since we are simply checking for the absence of the tag. + // To avoid confusion, we enforce that the value filter pattern is a wildcard. + if parts[1] != string(wildcardChar) { + return nil, fmt.Errorf("invalid filter %s: negation only supported for wildcard patterns", str) + } + } + res[key] = FilterValue{Pattern: parts[1], Negate: separator.negate} + } + return res, nil +} + +func parseTagFilter(str string) ([]string, tagFilterSeparator, error) { + unknownSeparator := tagFilterSeparator{} + parseByOneSeparator := func(separator tagFilterSeparator) ([]string, tagFilterSeparator, error) { + items := strings.Split(str, separator.str) + if len(items) == 2 { + if items[0] == "" { + return nil, unknownSeparator, fmt.Errorf("invalid filter %s: empty tag name", str) + } + if items[1] == "" { + return nil, unknownSeparator, fmt.Errorf("invalid filter %s: empty filter pattern", str) + } + return items, separator, nil + } + return nil, unknownSeparator, fmt.Errorf("invalid filter %s: expecting tag pattern pairs", str) + } + var returnedErr error + for _, separator := range validFilterSeparators { + parts, _, err := parseByOneSeparator(separator) + if err == nil { + return parts, separator, nil + } + returnedErr = err + } + return nil, unknownSeparator, returnedErr +} + +// Filter matches a string against certain conditions. +type Filter interface { + fmt.Stringer + Matches(val []byte) bool +} + +// TagsFilter matches labels against certain conditions. +type TagsFilter interface { + fmt.Stringer + MatchLabels(lbls labels.Labels) bool +} + +// TagsFilterOptions provide a set of tag filter options. +type TagsFilterOptions struct { + // NameTagKey is the name of the name tag (not used in this simplified version). + NameTagKey []byte +} + +// tagFilter is a filter associated with a given tag. +type tagFilter struct { + name []byte + valueFilter Filter + exclude bool +} + +func (f *tagFilter) String() string { + excludePrefix := "" + if f.exclude { + excludePrefix = string(negationChar) + } + return fmt.Sprintf("%s%s:%s", excludePrefix, string(f.name), f.valueFilter.String()) +} + +type tagFiltersByNameAsc []tagFilter + +func (tn tagFiltersByNameAsc) Len() int { return len(tn) } +func (tn tagFiltersByNameAsc) Swap(i, j int) { tn[i], tn[j] = tn[j], tn[i] } +func (tn tagFiltersByNameAsc) Less(i, j int) bool { return bytes.Compare(tn[i].name, tn[j].name) < 0 } + +// tagsFilter contains a list of tag filters. +type tagsFilter struct { + tagFilters []tagFilter + op LogicalOp + opts TagsFilterOptions +} + +// NewTagsFilter creates a new tags filter. +func NewTagsFilter( + filters TagFilterValueMap, + op LogicalOp, + opts TagsFilterOptions, +) (TagsFilter, error) { + tagFilters := make([]tagFilter, 0, len(filters)) + for entry, value := range filters { + // We disallow OR support for exclude rules for simplicity. + if entry.Exclude && op == Disjunction { + return nil, fmt.Errorf("invalid filter %s: exclude not supported for disjunction", entry.Name) + } + valFilter, err := NewFilterFromFilterValue(value) + if err != nil { + return nil, err + } + tagFilters = append(tagFilters, tagFilter{ + name: []byte(entry.Name), + valueFilter: valFilter, + exclude: entry.Exclude, + }) + } + sort.Sort(tagFiltersByNameAsc(tagFilters)) + return &tagsFilter{ + tagFilters: tagFilters, + op: op, + opts: opts, + }, nil +} + +func (f *tagsFilter) String() string { + separator := " " + string(f.op) + " " + var buf bytes.Buffer + numTagFilters := len(f.tagFilters) + for i := 0; i < numTagFilters; i++ { + buf.WriteString(f.tagFilters[i].String()) + if i < numTagFilters-1 { + buf.WriteString(separator) + } + } + return buf.String() +} + +// MatchLabels checks if the labels match the filter. +func (f *tagsFilter) MatchLabels(lbls labels.Labels) bool { + return f.matchLabelsSimple(lbls) +} + +func (f *tagsFilter) matchLabelsSimple(lbls labels.Labels) bool { + if len(f.tagFilters) == 0 { + return true + } + + for _, tf := range f.tagFilters { + labelValue := lbls.Get(string(tf.name)) + hasLabel := labelValue != "" + + if tf.exclude { + // For exclude rules, we want the tag to NOT exist + if hasLabel { + return false + } + continue + } + + if !hasLabel { + // Tag doesn't exist + if f.op == Conjunction { + return false + } + continue + } + + match := tf.valueFilter.Matches([]byte(labelValue)) + if match && f.op == Disjunction { + return true + } + if !match && f.op == Conjunction { + return false + } + } + + return f.op != Disjunction +} + +// NewFilterFromFilterValue creates a filter from the given filter value. +func NewFilterFromFilterValue(fv FilterValue) (Filter, error) { + f, err := NewFilter([]byte(fv.Pattern)) + if err != nil { + return nil, err + } + if !fv.Negate { + return f, nil + } + return &negationFilter{filter: f}, nil +} + +// NewFilter supports startsWith, endsWith, contains and a single wildcard +// along with negation and glob matching support. +func NewFilter(pattern []byte) (Filter, error) { + if len(pattern) == 0 { + return &equalityFilter{pattern: pattern}, nil + } + + if pattern[0] != negationChar { + return newWildcardFilter(pattern) + } + + if len(pattern) == 1 { + // Only negation symbol. + return nil, errInvalidFilterPattern + } + + filter, err := newWildcardFilter(pattern[1:]) + if err != nil { + return nil, err + } + + return &negationFilter{filter: filter}, nil +} + +// newWildcardFilter creates a filter that segments the pattern based +// on wildcards, creating a rangeFilter for each segment. +func newWildcardFilter(pattern []byte) (Filter, error) { + wIdx := bytes.IndexRune(pattern, wildcardChar) + + if wIdx == -1 { + // No wildcards. + return newRangeFilter(pattern, false, middle) + } + + if len(pattern) == 1 { + // Whole thing is wildcard. + return &allowFilter{}, nil + } + + if wIdx == len(pattern)-1 { + // Single wildcard at end. + return newRangeFilter(pattern[:len(pattern)-1], false, start) + } + + secondWIdx := bytes.IndexRune(pattern[wIdx+1:], wildcardChar) + if secondWIdx == -1 { + if wIdx == 0 { + // Single wildcard at start. + return newRangeFilter(pattern[1:], true, end) + } + + // Single wildcard in the middle. + first, err := newRangeFilter(pattern[:wIdx], false, start) + if err != nil { + return nil, err + } + + second, err := newRangeFilter(pattern[wIdx+1:], true, end) + if err != nil { + return nil, err + } + + return &multiFilter{filters: []Filter{first, second}, op: Conjunction}, nil + } + + if wIdx == 0 && secondWIdx == len(pattern)-2 && len(pattern) > 2 { + // Wildcard at beginning and end. + return newContainsFilter(pattern[1 : len(pattern)-1]) + } + + return nil, errInvalidFilterPattern +} + +// newRangeFilter creates a filter that checks for ranges (? or [] or {}) and segments +// the pattern into a multiple chain filters based on ranges found. +func newRangeFilter(pattern []byte, backwards bool, seg chainSegment) (Filter, error) { + var filters []chainFilter + eqIdx := -1 + for i := 0; i < len(pattern); i++ { + if pattern[i] == singleRangeStartChar { + // Found '[', create an equality filter for the chars before this one if any + if eqIdx != -1 { + filters = append(filters, &equalityChainFilter{pattern: pattern[eqIdx:i], backwards: backwards}) + eqIdx = -1 + } + + endIdx := bytes.IndexRune(pattern[i+1:], singleRangeEndChar) + if endIdx == -1 { + return nil, errInvalidFilterPattern + } + + f, err := newSingleRangeFilter(pattern[i+1:i+1+endIdx], backwards) + if err != nil { + return nil, errInvalidFilterPattern + } + + filters = append(filters, f) + i += endIdx + 1 + } else if pattern[i] == multiRangeStartChar { + // Found '{', create equality filter for chars before this if any + if eqIdx != -1 { + filters = append(filters, &equalityChainFilter{pattern: pattern[eqIdx:i], backwards: backwards}) + eqIdx = -1 + } + + endIdx := bytes.IndexRune(pattern[i+1:], multiRangeEndChar) + if endIdx == -1 { + return nil, errInvalidFilterPattern + } + + f, err := newMultiCharSequenceFilter(pattern[i+1:i+1+endIdx], backwards) + if err != nil { + return nil, errInvalidFilterPattern + } + + filters = append(filters, f) + i += endIdx + 1 + } else if pattern[i] == singleAnyChar { + // Found '?', create equality filter for chars before this one if any + if eqIdx != -1 { + filters = append(filters, &equalityChainFilter{pattern: pattern[eqIdx:i], backwards: backwards}) + eqIdx = -1 + } + + filters = append(filters, &singleAnyCharFilter{backwards: backwards}) + } else if eqIdx == -1 { + // Normal char, need to mark index to start next equality filter. + eqIdx = i + } + } + + if eqIdx != -1 { + filters = append(filters, &equalityChainFilter{pattern: pattern[eqIdx:], backwards: backwards}) + } + + return &multiChainFilter{filters: filters, seg: seg, backwards: backwards}, nil +} + +// allowFilter is a filter that allows all. +type allowFilter struct{} + +func (f *allowFilter) String() string { return "All" } +func (f *allowFilter) Matches(val []byte) bool { return true } + +// equalityFilter is a filter that matches exact values. +type equalityFilter struct { + pattern []byte +} + +func (f *equalityFilter) String() string { + return "Equals(\"" + string(f.pattern) + "\")" +} + +func (f *equalityFilter) Matches(val []byte) bool { + return bytes.Equal(f.pattern, val) +} + +// containsFilter is a filter that performs contains matches. +type containsFilter struct { + pattern []byte +} + +func newContainsFilter(pattern []byte) (Filter, error) { + if bytes.ContainsAny(pattern, invalidNestedChars) { + return nil, errInvalidFilterPattern + } + return &containsFilter{pattern: pattern}, nil +} + +func (f *containsFilter) String() string { + return "Contains(\"" + string(f.pattern) + "\")" +} + +func (f *containsFilter) Matches(val []byte) bool { + return bytes.Contains(val, f.pattern) +} + +// negationFilter is a filter that matches the opposite of the provided filter. +type negationFilter struct { + filter Filter +} + +func (f *negationFilter) String() string { + return "Not(" + f.filter.String() + ")" +} + +func (f *negationFilter) Matches(val []byte) bool { + return !f.filter.Matches(val) +} + +// multiFilter chains multiple filters together with a logicalOp. +type multiFilter struct { + filters []Filter + op LogicalOp +} + +func (f *multiFilter) String() string { + separator := " " + string(f.op) + " " + var buf bytes.Buffer + numFilters := len(f.filters) + for i := 0; i < numFilters; i++ { + buf.WriteString(f.filters[i].String()) + if i < numFilters-1 { + buf.WriteString(separator) + } + } + return buf.String() +} + +func (f *multiFilter) Matches(val []byte) bool { + if len(f.filters) == 0 { + return true + } + + for _, filter := range f.filters { + match := filter.Matches(val) + if f.op == Conjunction && !match { + return false + } + + if f.op == Disjunction && match { + return true + } + } + + return f.op == Conjunction +} + +// chainFilter matches an input string against certain conditions +// while returning the unmatched part of the input if there is a match. +type chainFilter interface { + fmt.Stringer + matches(val []byte) ([]byte, bool) +} + +// equalityChainFilter is a filter that performs equality string matches +// from either the front or back of the string. +type equalityChainFilter struct { + pattern []byte + backwards bool +} + +func (f *equalityChainFilter) String() string { + return "Equals(\"" + string(f.pattern) + "\")" +} + +func (f *equalityChainFilter) matches(val []byte) ([]byte, bool) { + if f.backwards && bytes.HasSuffix(val, f.pattern) { + return val[:len(val)-len(f.pattern)], true + } + + if !f.backwards && bytes.HasPrefix(val, f.pattern) { + return val[len(f.pattern):], true + } + + return nil, false +} + +// singleAnyCharFilter is a filter that allows any one char. +type singleAnyCharFilter struct { + backwards bool +} + +func (f *singleAnyCharFilter) String() string { return "AnyChar" } + +func (f *singleAnyCharFilter) matches(val []byte) ([]byte, bool) { + if len(val) == 0 { + return nil, false + } + + if f.backwards { + return val[:len(val)-1], true + } + + return val[1:], true +} + +// newSingleRangeFilter creates a filter that performs range matching +// on a single char. +func newSingleRangeFilter(pattern []byte, backwards bool) (chainFilter, error) { + if len(pattern) == 0 { + return nil, errInvalidFilterPattern + } + + negate := false + if pattern[0] == negationChar { + negate = true + pattern = pattern[1:] + } + + if len(pattern) > 1 && pattern[1] == rangeChar { + // If there is a '-' char at position 2, look for repeated instances of a-z. + if len(pattern)%3 != 0 { + return nil, errInvalidFilterPattern + } + + patterns := make([][]byte, 0, len(pattern)%3) + for i := 0; i < len(pattern); i += 3 { + if pattern[i+1] != rangeChar || pattern[i] > pattern[i+2] { + return nil, errInvalidFilterPattern + } + patterns = append(patterns, pattern[i:i+3]) + } + + return &singleRangeFilter{patterns: patterns, backwards: backwards, negate: negate}, nil + } + + return &singleCharSetFilter{pattern: pattern, backwards: backwards, negate: negate}, nil +} + +// singleRangeFilter is a filter that performs a single character match against +// a range of chars given in a range format eg. [a-z]. +type singleRangeFilter struct { + patterns [][]byte + backwards bool + negate bool +} + +func (f *singleRangeFilter) String() string { + var negatePrefix, negateSuffix string + if f.negate { + negatePrefix = "Not(" + negateSuffix = ")" + } + return negatePrefix + "Range(\"" + string(bytes.Join(f.patterns, []byte(fmt.Sprintf(" %s ", Disjunction)))) + "\")" + negateSuffix +} + +func (f *singleRangeFilter) matches(val []byte) ([]byte, bool) { + if len(val) == 0 { + return nil, false + } + + match := false + idx := 0 + remainder := val[1:] + if f.backwards { + idx = len(val) - 1 + remainder = val[:idx] + } + + for _, pattern := range f.patterns { + if val[idx] >= pattern[0] && val[idx] <= pattern[2] { + match = true + break + } + } + + if f.negate { + match = !match + } + + return remainder, match +} + +// singleCharSetFilter is a filter that performs a single character match against +// a set of chars given explicitly eg. [abcdefg]. +type singleCharSetFilter struct { + pattern []byte + backwards bool + negate bool +} + +func (f *singleCharSetFilter) String() string { + var negatePrefix, negateSuffix string + if f.negate { + negatePrefix = "Not(" + negateSuffix = ")" + } + return negatePrefix + "Range(\"" + string(f.pattern) + "\")" + negateSuffix +} + +func (f *singleCharSetFilter) matches(val []byte) ([]byte, bool) { + if len(val) == 0 { + return nil, false + } + + match := false + for i := 0; i < len(f.pattern); i++ { + if f.backwards && val[len(val)-1] == f.pattern[i] { + match = true + break + } + + if !f.backwards && val[0] == f.pattern[i] { + match = true + break + } + } + + if f.negate { + match = !match + } + + if f.backwards { + return val[:len(val)-1], match + } + + return val[1:], match +} + +// multiCharSequenceFilter is a filter that performs matches against multiple sets of chars +// eg. {abc,defg}. +type multiCharSequenceFilter struct { + patterns [][]byte + backwards bool +} + +func newMultiCharSequenceFilter(patterns []byte, backwards bool) (chainFilter, error) { + if len(patterns) == 0 { + return nil, errInvalidFilterPattern + } + + return &multiCharSequenceFilter{ + patterns: bytes.Split(patterns, multiRangeSplit), + backwards: backwards, + }, nil +} + +func (f *multiCharSequenceFilter) String() string { + return "Range(\"" + string(bytes.Join(f.patterns, multiRangeSplit)) + "\")" +} + +func (f *multiCharSequenceFilter) matches(val []byte) ([]byte, bool) { + if len(val) == 0 { + return nil, false + } + + for _, pattern := range f.patterns { + if f.backwards && bytes.HasSuffix(val, pattern) { + return val[:len(val)-len(pattern)], true + } + + if !f.backwards && bytes.HasPrefix(val, pattern) { + return val[len(pattern):], true + } + } + + return nil, false +} + +// multiChainFilter chains multiple chainFilters together with &&. +type multiChainFilter struct { + filters []chainFilter + seg chainSegment + backwards bool +} + +func (f *multiChainFilter) String() string { + separator := " then " + var buf bytes.Buffer + switch f.seg { + case start: + buf.WriteString("StartsWith(") + case end: + buf.WriteString("EndsWith(") + } + + numFilters := len(f.filters) + for i := 0; i < numFilters; i++ { + buf.WriteString(f.filters[i].String()) + if i < numFilters-1 { + buf.WriteString(separator) + } + } + + switch f.seg { + case start, end: + buf.WriteString(")") + } + + return buf.String() +} + +func (f *multiChainFilter) Matches(val []byte) bool { + if len(f.filters) == 0 { + return true + } + + var match bool + + if f.backwards { + for i := len(f.filters) - 1; i >= 0; i-- { + val, match = f.filters[i].matches(val) + if !match { + return false + } + } + } else { + for i := 0; i < len(f.filters); i++ { + val, match = f.filters[i].matches(val) + if !match { + return false + } + } + } + + if f.seg == middle && len(val) != 0 { + // chain was middle segment and some value was left over at end of chain. + return false + } + + return true +} diff --git a/pkg/receive/tenant_attribution_test.go b/pkg/receive/tenant_attribution_test.go new file mode 100644 index 00000000000..ced3c1495ef --- /dev/null +++ b/pkg/receive/tenant_attribution_test.go @@ -0,0 +1,460 @@ +// Copyright (c) The Thanos Authors. +// Licensed under the Apache License 2.0. + +package receive + +import ( + "os" + "path/filepath" + "testing" + + "github.com/go-kit/log" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/prometheus/model/labels" + "github.com/stretchr/testify/require" +) + +func TestParseTagFilterValueMap(t *testing.T) { + tests := []struct { + name string + input string + wantErr bool + wantCount int + }{ + { + name: "single tag exact match", + input: "job:prometheus", + wantErr: false, + wantCount: 1, + }, + { + name: "single tag with wildcard", + input: "job:prom*", + wantErr: false, + wantCount: 1, + }, + { + name: "multiple tags AND", + input: "job:prometheus namespace:monitoring", + wantErr: false, + wantCount: 2, + }, + { + name: "tag with hash separator", + input: "job#prometheus", + wantErr: false, + wantCount: 1, + }, + { + name: "negation tag", + input: "!internal:*", + wantErr: false, + wantCount: 1, + }, + { + name: "negation with non-wildcard pattern should error", + input: "!internal:value", + wantErr: true, + }, + { + name: "empty filter", + input: "", + wantErr: false, + }, + { + name: "invalid filter no separator", + input: "jobprometheus", + wantErr: true, + }, + { + name: "invalid filter empty tag name", + input: ":prometheus", + wantErr: true, + }, + { + name: "invalid filter empty pattern", + input: "job:", + wantErr: true, + }, + { + name: "duplicate tag should error", + input: "job:prometheus job:alertmanager", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result, err := ParseTagFilterValueMap(tt.input) + if tt.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + require.Len(t, result, tt.wantCount) + } + }) + } +} + +func TestFilterPatterns(t *testing.T) { + tests := []struct { + name string + pattern string + input string + want bool + }{ + // Exact match + {"exact match", "prometheus", "prometheus", true}, + {"exact match fail", "prometheus", "alertmanager", false}, + {"exact match empty", "", "", true}, + + // Wildcard patterns + {"wildcard all", "*", "anything", true}, + {"prefix wildcard", "prom*", "prometheus", true}, + {"prefix wildcard fail", "prom*", "alertmanager", false}, + {"suffix wildcard", "*eus", "prometheus", true}, + {"suffix wildcard fail", "*eus", "alertmanager", false}, + {"contains wildcard", "*met*", "prometheus", true}, + {"contains wildcard fail", "*xyz*", "prometheus", false}, + {"middle wildcard", "prom*eus", "prometheus", true}, + {"middle wildcard fail", "prom*eus", "prometheusX", false}, + + // Single char wildcard + {"single char ?", "pro?", "prod", true}, + {"single char ? fail", "pro?", "production", false}, + + // Character set + {"char set [abc]", "[abc]bc", "abc", true}, + {"char set [abc] fail", "[abc]bc", "dbc", false}, + + // Character range + {"char range [a-z]", "[a-z]bc", "abc", true}, + {"char range [a-z] fail", "[a-z]bc", "1bc", false}, + + // Multi-char sequence + {"multi char {a,b,c}", "{prod,dev,staging}", "prod", true}, + {"multi char {a,b,c} staging", "{prod,dev,staging}", "staging", true}, + {"multi char {a,b,c} fail", "{prod,dev,staging}", "test", false}, + + // Negation + {"negation", "!prometheus", "alertmanager", true}, + {"negation fail", "!prometheus", "prometheus", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + filter, err := NewFilter([]byte(tt.pattern)) + require.NoError(t, err) + result := filter.Matches([]byte(tt.input)) + require.Equal(t, tt.want, result, "pattern=%q input=%q", tt.pattern, tt.input) + }) + } +} + +func TestTagsFilter(t *testing.T) { + tests := []struct { + name string + filter string + labels labels.Labels + want bool + }{ + { + name: "single tag match", + filter: "job:prometheus", + labels: labels.FromStrings("job", "prometheus", "namespace", "monitoring"), + want: true, + }, + { + name: "single tag no match", + filter: "job:prometheus", + labels: labels.FromStrings("job", "alertmanager", "namespace", "monitoring"), + want: false, + }, + { + name: "multiple tags AND match", + filter: "job:prometheus namespace:monitoring", + labels: labels.FromStrings("job", "prometheus", "namespace", "monitoring"), + want: true, + }, + { + name: "multiple tags AND partial match fails", + filter: "job:prometheus namespace:monitoring", + labels: labels.FromStrings("job", "prometheus", "namespace", "default"), + want: false, + }, + { + name: "wildcard prefix match", + filter: "job:prom*", + labels: labels.FromStrings("job", "prometheus"), + want: true, + }, + { + name: "wildcard suffix match", + filter: "namespace:*ing", + labels: labels.FromStrings("namespace", "monitoring"), + want: true, + }, + { + name: "wildcard contains match", + filter: "namespace:*nitor*", + labels: labels.FromStrings("namespace", "monitoring"), + want: true, + }, + { + name: "multi-char sequence match", + filter: "env:{prod,staging,dev}", + labels: labels.FromStrings("env", "prod"), + want: true, + }, + { + name: "multi-char sequence match staging", + filter: "env:{prod,staging,dev}", + labels: labels.FromStrings("env", "staging"), + want: true, + }, + { + name: "multi-char sequence no match", + filter: "env:{prod,staging,dev}", + labels: labels.FromStrings("env", "test"), + want: false, + }, + { + name: "tag not present fails for conjunction", + filter: "job:prometheus", + labels: labels.FromStrings("namespace", "monitoring"), + want: false, + }, + { + name: "exclude tag match (tag absent)", + filter: "!internal:*", + labels: labels.FromStrings("job", "prometheus"), + want: true, + }, + { + name: "exclude tag no match (tag present)", + filter: "!internal:*", + labels: labels.FromStrings("job", "prometheus", "internal", "true"), + want: false, + }, + { + name: "empty labels with filter fails", + filter: "job:prometheus", + labels: labels.EmptyLabels(), + want: false, + }, + { + name: "empty filter matches all", + filter: "", + labels: labels.FromStrings("job", "prometheus"), + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + filterValues, err := ParseTagFilterValueMap(tt.filter) + require.NoError(t, err) + + tagsFilter, err := NewTagsFilter(filterValues, Conjunction, TagsFilterOptions{}) + require.NoError(t, err) + + result := tagsFilter.MatchLabels(tt.labels) + require.Equal(t, tt.want, result, "filter=%q labels=%v", tt.filter, tt.labels) + }) + } +} + +func TestTenantAttributor(t *testing.T) { + // Create a temporary config file + configContent := ` +- filter: "job:prometheus" + tenant: "prometheus-tenant" +- filter: "namespace:prod*" + tenant: "production" +- filter: "job:alertmanager namespace:monitoring" + tenant: "alerting-tenant" +- filter: "env:{staging,dev}" + tenant: "non-prod" +` + tmpDir := t.TempDir() + configPath := filepath.Join(tmpDir, "tenant-rules.yaml") + err := os.WriteFile(configPath, []byte(configContent), 0644) + require.NoError(t, err) + + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + + // Test without verify mode + ta, err := NewTenantAttributor(configPath, "default-tenant", false, reg, logger) + require.NoError(t, err) + require.NotNil(t, ta) + require.Len(t, ta.rules, 4) + require.False(t, ta.IsVerifyMode()) + + tests := []struct { + name string + labels labels.Labels + wantTenant string + }{ + { + name: "first rule match - prometheus", + labels: labels.FromStrings("job", "prometheus"), + wantTenant: "prometheus-tenant", + }, + { + name: "second rule match - production namespace", + labels: labels.FromStrings("namespace", "production"), + wantTenant: "production", + }, + { + name: "second rule match - prod prefix", + labels: labels.FromStrings("namespace", "prod-us-east"), + wantTenant: "production", + }, + { + name: "third rule match - alertmanager in monitoring", + labels: labels.FromStrings("job", "alertmanager", "namespace", "monitoring"), + wantTenant: "alerting-tenant", + }, + { + name: "fourth rule match - staging env", + labels: labels.FromStrings("env", "staging"), + wantTenant: "non-prod", + }, + { + name: "fourth rule match - dev env", + labels: labels.FromStrings("env", "dev"), + wantTenant: "non-prod", + }, + { + name: "no rule match - default tenant", + labels: labels.FromStrings("job", "unknown"), + wantTenant: "default-tenant", + }, + { + name: "empty labels - default tenant", + labels: labels.EmptyLabels(), + wantTenant: "default-tenant", + }, + { + name: "first match wins - prometheus also has prod namespace", + labels: labels.FromStrings("job", "prometheus", "namespace", "production"), + wantTenant: "prometheus-tenant", // First rule wins + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tenant := ta.GetTenantFromLabels(tt.labels) + require.Equal(t, tt.wantTenant, tenant) + }) + } +} + +func TestTenantAttributorVerifyMode(t *testing.T) { + configContent := ` +- filter: "job:prometheus" + tenant: "prometheus-tenant" +` + tmpDir := t.TempDir() + configPath := filepath.Join(tmpDir, "tenant-rules.yaml") + err := os.WriteFile(configPath, []byte(configContent), 0644) + require.NoError(t, err) + + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + + ta, err := NewTenantAttributor(configPath, "default-tenant", true, reg, logger) + require.NoError(t, err) + require.True(t, ta.IsVerifyMode()) + require.NotNil(t, ta.attributionMatches) + require.NotNil(t, ta.attributionMismatches) + + // Test recording verification + lbls := labels.FromStrings("job", "prometheus") + attributedTenant := ta.GetTenantFromLabels(lbls) + require.Equal(t, "prometheus-tenant", attributedTenant) + + // Record match + ta.RecordVerification(attributedTenant, "prometheus-tenant") + + // Record mismatch + ta.RecordVerification(attributedTenant, "other-tenant") + + // Verify metrics + matchCount, err := getCounterValue(ta.attributionMatches) + require.NoError(t, err) + require.Equal(t, float64(1), matchCount) + + mismatchCount, err := getCounterValue(ta.attributionMismatches) + require.NoError(t, err) + require.Equal(t, float64(1), mismatchCount) +} + +func TestTenantAttributorConfigErrors(t *testing.T) { + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + tmpDir := t.TempDir() + + tests := []struct { + name string + config string + wantErrMsg string + emptyConfig bool + }{ + { + name: "empty config path", + config: "", + wantErrMsg: "tenant rules config path is required", + }, + { + name: "missing filter", + config: "- tenant: my-tenant\n", + wantErrMsg: "filter is required", + }, + { + name: "missing tenant", + config: "- filter: \"job:prometheus\"\n", + wantErrMsg: "tenant is required", + }, + { + name: "invalid filter syntax", + config: "- filter: \"invalid\"\n tenant: my-tenant\n", + wantErrMsg: "parsing filter", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + configPath := "" + if tt.config != "" { + configPath = filepath.Join(tmpDir, tt.name+".yaml") + err := os.WriteFile(configPath, []byte(tt.config), 0644) + require.NoError(t, err) + } + + _, err := NewTenantAttributor(configPath, "default", false, reg, logger) + require.Error(t, err) + require.Contains(t, err.Error(), tt.wantErrMsg) + }) + } +} + +func TestTenantAttributorFileNotFound(t *testing.T) { + logger := log.NewNopLogger() + reg := prometheus.NewRegistry() + + _, err := NewTenantAttributor("/nonexistent/path/config.yaml", "default", false, reg, logger) + require.Error(t, err) + require.Contains(t, err.Error(), "reading tenant rules config") +} + +// Helper function to get counter value. +func getCounterValue(c prometheus.Counter) (float64, error) { + var m dto.Metric + if err := c.Write(&m); err != nil { + return 0, err + } + return m.GetCounter().GetValue(), nil +}