enhancement: add support for extra query matchers in loki.rules.kubernetes

CHANGELOG.md

@@ -46,6 +46,9 @@ Main (unreleased)
 
 - Pretty print diagnostic errors when using `alloy run` (@kalleep)
 
+- The `loki.rules.kubernetes` component now supports adding extra label matchers
+  to all queries discovered via `PrometheusRule` CRDs. (@QuentinBisson)
+
 -  Add optional `id` field to `foreach` block to generate more meaningful component paths in metrics by using a specific field from collection items. (@harshrai654)
 
 - Fix validation logic in `beyla.ebpf` component to ensure that either metrics or traces are enabled. (@marctc)

docs/sources/reference/components/loki/loki.rules.kubernetes.md

@@ -81,23 +81,27 @@ You can use the following blocks with `loki.rules.kubernetes`:
 
 | Block                                                              | Description                                                | Required |
 | ------------------------------------------------------------------ | ---------------------------------------------------------- | -------- |
-| [`authorization`][authorization]                                   | Configure generic authorization to the endpoint.           | no       |
-| [`basic_auth`][basic_auth]                                         | Configure `basic_auth` for authenticating to the endpoint. | no       |
-| [`rule_namespace_selector`][label_selector]                        | Label selector for `Namespace` resources.                  | no       |
-| `rule_namespace_selector` > [`match_expression`][match_expression] | Label match expression for `Namespace` resources.          | no       |
-| [`rule_selector`][label_selector]                                  | Label selector for `PrometheusRule` resources.             | no       |
-| `rule_selector` > [`match_expression`][match_expression]           | Label match expression for `PrometheusRule` resources.     | no       |
-| [`oauth2`][oauth2]                                                 | Configure OAuth 2.0 for authenticating to the endpoint.    | no       |
-| `oauth2` > [`tls_config`][tls_config]                              | Configure TLS settings for connecting to the endpoint.     | no       |
-| [`tls_config`][tls_config]                                         | Configure TLS settings for connecting to the endpoint.     | no       |
+| [`authorization`][authorization]                                   | Configure generic authorization to the endpoint.           | no |
+| [`basic_auth`][basic_auth]                                         | Configure `basic_auth` for authenticating to the endpoint. | no |
+| [`extra_query_matchers`][extra_query_matchers]                     | Additional label matchers to add to each query.            | no |
+| `extra_query_matchers` > [`matcher`][matcher]                      | A label matcher to add to each query.                      | no |
+| [`rule_namespace_selector`][label_selector]                        | Label selector for `Namespace` resources.                  | no |
+| `rule_namespace_selector` > [`match_expression`][match_expression] | Label match expression for `Namespace` resources.          | no |
+| [`rule_selector`][label_selector]                                  | Label selector for `PrometheusRule` resources.             | no |
+| `rule_selector` > [`match_expression`][match_expression]           | Label match expression for `PrometheusRule` resources.     | no |
+| [`oauth2`][oauth2]                                                 | Configure OAuth 2.0 for authenticating to the endpoint.    | no |
+| `oauth2` > [`tls_config`][tls_config]                              | Configure TLS settings for connecting to the endpoint.     | no |
+| [`tls_config`][tls_config]                                         | Configure TLS settings for connecting to the endpoint.     | no |
 
 The > symbol indicates deeper levels of nesting.
 For example, `oauth2` > `tls_config` refers to a `tls_config` block defined inside an `oauth2` block.
 
 [authorization]: #authorization
 [basic_auth]: #basic_auth
+[extra_query_matchers]: #extra_query_matchers
 [label_selector]: #rule_selector-and-rule_namespace_selector
 [match_expression]: #match_expression
+[matcher]: #matcher
 [oauth2]: #oauth2
 [tls_config]: #tls_config
 
@@ -109,6 +113,26 @@ For example, `oauth2` > `tls_config` refers to a `tls_config` block defined insi
 
 {{< docs/shared lookup="reference/components/basic-auth-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
 
+### `extra_query_matchers`
+
+The `extra_query_matchers` block has no attributes.
+It contains zero or more [matcher][] blocks.
+These blocks allow you to add extra label matchers to all queries that are discovered by the `loki.rules.kubernetes` component.
+The algorithm for adding the label matchers to queries is the same as the one used by the [`promtool promql label-matchers set` command](https://prometheus.io/docs/prometheus/latest/command-line/promtool/#promtool-promql).
+It's adapted to work with the LogQL parser.
+
+### `matcher`
+
+The `matcher` block describes a label matcher that's added to each query found in `PrometheusRule` CRDs.
+
+The following arguments are supported:
+
+| Name         | Type     | Description                                        | Default | Required |
+| ------------ | -------- | -------------------------------------------------- | ------- | -------- |
+| `match_type` | `string` | The type of match. One of `=`, `!=`, `=~` or `!~`. |         | yes      |
+| `name`       | `string` | Name of the label to match.                        |         | yes      |
+| `value`      | `string` | Value of the label to match.                       |         | yes      |
+
 ### `rule_selector` and `rule_namespace_selector`
 
 The `rule_selector` and `rule_namespace_selector` blocks describe a Kubernetes label selector for rule or namespace discovery.
@@ -228,6 +252,23 @@ Replace the following:
 * _`<GRAFANA_CLOUD_API_KEY>`_: Your Grafana Cloud API key.
 * _`<GRAFANA_CLOUD_API_KEY_PATH>`_: The path to the Grafana Cloud API key.
 
+This example adds label matcher `{cluster=~"prod-.*"}` to all the queries discovered by `loki.rules.kubernetes`.
+
+```alloy
+loki.rules.kubernetes "default" {
+    address = "loki:3100"
+    extra_query_matchers {
+        matcher {
+            name = "cluster"
+            match_type = "=~"
+            value = "prod-.*"
+        }
+    }
+}
+```
+
+If a query in the form of `{app="my-app"}` is found in `PrometheusRule` CRDs, it will be modified to `{app="my-app", cluster=~"prod-.*"}` before sending it to Loki.
+
 The following example is an RBAC configuration for Kubernetes. It authorizes {{< param "PRODUCT_NAME" >}} to query the Kubernetes REST API:
 
 ```yaml

internal/component/loki/rules/kubernetes/events.go

@@ -6,12 +6,16 @@ import (
 	"regexp"
 	"time"
 
-	"github.com/grafana/alloy/internal/component/common/kubernetes"
-	"github.com/grafana/alloy/internal/runtime/logging/level"
+	"github.com/grafana/loki/v3/pkg/logql/syntax"
 	"github.com/hashicorp/go-multierror"
 	promv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
+	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/rulefmt"
+	"github.com/prometheus/prometheus/promql/parser"
 	"sigs.k8s.io/yaml" // Used for CRD compatibility instead of gopkg.in/yaml.v2
+
+	"github.com/grafana/alloy/internal/component/common/kubernetes"
+	"github.com/grafana/alloy/internal/runtime/logging/level"
 )
 
 const eventTypeSyncLoki kubernetes.EventType = "sync-loki"
@@ -102,16 +106,16 @@ func (c *Component) reconcileState(ctx context.Context) error {
 	}
 
 	diffs := kubernetes.DiffRuleState(desiredState, c.currentState)
-	var result error
+	var errs error
 	for ns, diff := range diffs {
 		err = c.applyChanges(ctx, ns, diff)
 		if err != nil {
-			result = multierror.Append(result, err)
+			errs = multierror.Append(errs, err)
 			continue
 		}
 	}
 
-	return result
+	return errs
 }
 
 func (c *Component) loadStateFromK8s() (kubernetes.RuleGroupsByNamespace, error) {
@@ -127,33 +131,111 @@ func (c *Component) loadStateFromK8s() (kubernetes.RuleGroupsByNamespace, error)
 			return nil, fmt.Errorf("failed to list rules: %w", err)
 		}
 
-		for _, pr := range crdState {
-			lokiNs := lokiNamespaceForRuleCRD(c.args.LokiNameSpacePrefix, pr)
-
-			groups, err := convertCRDRuleGroupToRuleGroup(pr.Spec)
+		for _, rule := range crdState {
+			lokiNs := lokiNamespaceForRuleCRD(c.args.LokiNameSpacePrefix, rule)
+			groups, err := convertCRDRuleGroupToRuleGroup(rule.Spec)
 			if err != nil {
 				return nil, fmt.Errorf("failed to convert rule group: %w", err)
 			}
 
+			if c.args.ExtraQueryMatchers != nil {
+				for _, ruleGroup := range groups {
+					for i := range ruleGroup.Rules {
+						query := ruleGroup.Rules[i].Expr.Value
+						newQuery, err := addMatchersToQuery(query, c.args.ExtraQueryMatchers.Matchers)
+						if err != nil {
+							level.Error(c.log).Log("msg", "failed to add labels to PrometheusRule query", "query", query, "err", err)
+						}
+						ruleGroup.Rules[i].Expr.Value = newQuery
+					}
+				}
+			}
+
 			desiredState[lokiNs] = groups
 		}
 	}
 
 	return desiredState, nil
 }
 
+func addMatchersToQuery(query string, matchers []Matcher) (string, error) {
+	var err error
+	for _, s := range matchers {
+		query, err = labelsSetLogQL(query, s.MatchType, s.Name, s.Value)
+		if err != nil {
+			return "", err
+		}
+	}
+	return query, nil
+}
+
+// Inspired from the labelsSetPromQL function from the mimir.rules.kubernetes component
+// this function was modified to use the logql parser instead
+func labelsSetLogQL(query, labelMatchType, name, value string) (string, error) {
+	expr, err := syntax.ParseExpr(query)
+	if err != nil {
+		return query, err
+	}
+
+	var matchType labels.MatchType
+	switch labelMatchType {
+	case parser.ItemType(parser.EQL).String():
+		matchType = labels.MatchEqual
+	case parser.ItemType(parser.NEQ).String():
+		matchType = labels.MatchNotEqual
+	case parser.ItemType(parser.EQL_REGEX).String():
+		matchType = labels.MatchRegexp
+	case parser.ItemType(parser.NEQ_REGEX).String():
+		matchType = labels.MatchNotRegexp
+	default:
+		return query, fmt.Errorf("invalid label match type: %s", labelMatchType)
+	}
+	expr.Walk(func(e syntax.Expr) {
+		switch concrete := e.(type) {
+		case *syntax.MatchersExpr:
+			var found bool
+			for _, l := range concrete.Mts {
+				if l.Name == name {
+					l.Type = matchType
+					l.Value = value
+					found = true
+				}
+			}
+			if !found {
+				concrete.Mts = append(concrete.Mts, &labels.Matcher{
+					Type:  matchType,
+					Name:  name,
+					Value: value,
+				})
+			}
+		}
+	})
+
+	return expr.String(), nil
+}
+
 func convertCRDRuleGroupToRuleGroup(crd promv1.PrometheusRuleSpec) ([]rulefmt.RuleGroup, error) {
 	buf, err := yaml.Marshal(crd)
 	if err != nil {
 		return nil, err
 	}
 
+	var errs error
 	groups, _ := rulefmt.Parse(buf)
-
-	// Disable looking for errors, loki queries won't be valid prometheus queries, but still want the similar information
-	//if len(errs) > 0 {
-	//	return nil, multierror.Append(nil, errs...)
-	//}
+	for _, group := range groups.Groups {
+		for _, rule := range group.Rules {
+			if _, err := syntax.ParseExpr(rule.Expr.Value); err != nil {
+				if rule.Record.Value != "" {
+					errs = multierror.Append(errs, fmt.Errorf("could not parse expression for record '%s' in group '%s': %w", rule.Record.Value, group.Name, err))
+				} else {
+					errs = multierror.Append(errs, fmt.Errorf("could not parse expression for alert '%s' in group '%s': %w", rule.Alert.Value, group.Name, err))
+				}
+			}
+		}
+	}
+	if errs != nil {
+		return nil, errs
+	}
 
 	return groups.Groups, nil
 }