Add support for rule group labels

Signed-off-by: Mustafain Ali Khan <mustalik@amazon.com>

Author: mustafain117

CHANGELOG.md

@@ -6,6 +6,7 @@
 * [FEATURE] Querier/Ruler: Add `query_partial_data` and `rules_partial_data` limits to allow queries/rules to be evaluated with data from a single zone, if other zones are not available. #6526
 * [FEATURE] Update prometheus alertmanager version to v0.28.0 and add new integration msteamsv2, jira, and rocketchat. #6590
 * [FEATURE] Ingester: Support out-of-order native histogram ingestion. It automatically enabled when `-ingester.out-of-order-time-window > 0` and `-blocks-storage.tsdb.enable-native-histograms=true`. #6626 #6663
+* [FEATURE] Ruler: Add support for group labels. #6665
 * [ENHANCEMENT] Query Frontend: Add new limit `-frontend.max-query-response-size` for total query response size after decompression in query frontend. #6607
 * [ENHANCEMENT] Alertmanager: Add nflog and silences maintenance metrics. #6659
 * [ENHANCEMENT] Querier: limit label APIs to query only ingesters if `start` param is not been specified. #6618

integration/ruler_test.go

@@ -1739,6 +1739,106 @@ func TestRulerEvalWithQueryFrontend(t *testing.T) {
 	}
 }
 
+func TestRulerGroupLabels(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))
+
+	// Configure the ruler.
+	flags := mergeFlags(
+		BlocksStorageFlags(),
+		RulerFlags(),
+		map[string]string{
+			// Since we're not going to run any rule (our only rule is invalid), we don't need the
+			// store-gateway to be configured to a valid address.
+			"-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",
+
+			"-blocks-storage.tsdb.block-ranges-period":   "1h",
+			"-blocks-storage.bucket-store.sync-interval": "1s",
+			"-blocks-storage.tsdb.retention-period":      "2h",
+
+			// We run single ingester only, no replication.
+			"-distributor.replication-factor": "1",
+		},
+	)
+
+	const namespace = "test"
+	const user = "user"
+
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	ruler := e2ecortex.NewRuler("ruler", consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester, ruler))
+
+	// Wait until both the distributor and ruler have updated the ring. The querier will also watch
+	// the store-gateway ring if blocks sharding is enabled.
+	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+	require.NoError(t, ruler.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", ruler.HTTPEndpoint(), user)
+	require.NoError(t, err)
+
+	groupLabels := map[string]string{
+		"group_label_1":   "val1",
+		"group_label_2":   "val2",
+		"duplicate_label": "group_val",
+	}
+	ruleLabels := map[string]string{
+		"rule_label_1":    "val1",
+		"rule_label_2":    "val2",
+		"duplicate_label": "rule_val",
+	}
+	var recordNode = yaml.Node{}
+	var exprNode = yaml.Node{}
+	recordNode.SetString("ruleName")
+	exprNode.SetString("vector(1) > 0")
+	groupName := "rulegrouptest"
+
+	rg := rulefmt.RuleGroup{
+		Name:     groupName,
+		Labels:   groupLabels,
+		Interval: 10,
+		Rules: []rulefmt.RuleNode{{
+			Alert:  recordNode,
+			Expr:   exprNode,
+			Labels: ruleLabels,
+		}},
+	}
+	require.NoError(t, c.SetRuleGroup(rg, namespace))
+
+	m := ruleGroupMatcher(user, namespace, groupName)
+
+	// Wait until ruler has loaded the group.
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_prometheus_rule_group_rules"}, e2e.WithLabelMatchers(m), e2e.WaitMissingMetrics))
+
+	groups, _, err := c.GetPrometheusRules(e2ecortex.RuleFilter{
+		RuleGroupNames: []string{groupName},
+	})
+
+	require.NoError(t, err)
+	require.NotEmpty(t, groups)
+	require.Equal(t, 1, len(groups))
+	require.Equal(t, 1, len(groups[0].Rules))
+	ar := parseAlertFromRule(t, groups[0].Rules[0])
+	require.Equal(t, 5, len(ar.Labels))
+	for _, label := range ar.Labels {
+		if label.Name == "duplicate_label" {
+			// rule label should override group label
+			require.Equal(t, ruleLabels["duplicate_label"], label.Value)
+		}
+	}
+}
+
 func parseAlertFromRule(t *testing.T, rules interface{}) *alertingRule {
 	responseJson, err := json.Marshal(rules)
 	require.NoError(t, err)

pkg/ruler/rulespb/compat.go

@@ -26,6 +26,7 @@ func ToProto(user string, namespace string, rl rulefmt.RuleGroup) *RuleGroupDesc
 		User:        user,
 		Limit:       int64(rl.Limit),
 		QueryOffset: queryOffset,
+		Labels:      cortexpb.FromLabelsToLabelAdapters(labels.FromMap(rl.Labels)),
 	}
 	return &rg
 }
@@ -60,6 +61,7 @@ func FromProto(rg *RuleGroupDesc) rulefmt.RuleGroup {
 		Rules:       make([]rulefmt.RuleNode, len(rg.GetRules())),
 		Limit:       int(rg.Limit),
 		QueryOffset: queryOffset,
+		Labels:      cortexpb.FromLabelAdaptersToLabels(rg.Labels).Map(),
 	}
 
 	for i, rl := range rg.GetRules() {

pkg/ruler/rulespb/rules.pb.go

@@ -30,6 +30,10 @@ message RuleGroupDesc {
   int64 limit =10;
   google.protobuf.Duration queryOffset = 11
       [(gogoproto.nullable) = true, (gogoproto.stdduration) = true];
+  repeated cortexpb.LabelPair labels = 12 [
+    (gogoproto.nullable) = false,
+    (gogoproto.customtype) = "github.com/cortexproject/cortex/pkg/cortexpb.LabelAdapter"
+  ];
 }
 
 // RuleDesc is a proto representation of a Prometheus Rule