vmalert: support extra_filter_labels setting per-group (#1319)

The new setting `extra_filter_labels` may be assigned to group.
If it is, then all rules within a group will automatically filter
for configured labels. The feature is well-described here
https://docs.victoriametrics.com#prometheus-querying-api-enhancements

New setting is compatible only with VM datasource.

app/vmalert/README.md

@@ -85,6 +85,12 @@ name: <string>
 # By default "prometheus" rule type is used.
 [ type: <string> ]
 
+# Optional list of label filters applied to every rule's
+# request withing a group. Is compatible only with VM datasource.
+# See more details at https://docs.victoriametrics.com#prometheus-querying-api-enhancements
+extra_filter_labels:
+  [ <labelname>: <labelvalue> ... ]
+
 rules:
   [ - <rule> ... ]
 ```

app/vmalert/alerting.go

@@ -65,6 +65,7 @@ func newAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 		q: qb.BuildWithParams(datasource.QuerierParams{
 			DataSourceType:     &cfg.Type,
 			EvaluationInterval: group.Interval,
+			ExtraLabels:        group.ExtraFilterLabels,
 		}),
 		alerts:  make(map[uint64]*notifier.Alert),
 		metrics: &alertingRuleMetrics{},
@@ -250,6 +251,7 @@ func (ar *AlertingRule) UpdateWith(r Rule) error {
 	ar.For = nr.For
 	ar.Labels = nr.Labels
 	ar.Annotations = nr.Annotations
+	ar.q = nr.q
 	return nil
 }
 

app/vmalert/config/config.go

@@ -29,6 +29,10 @@ type Group struct {
 	Interval    time.Duration `yaml:"interval,omitempty"`
 	Rules       []Rule        `yaml:"rules"`
 	Concurrency int           `yaml:"concurrency"`
+	// ExtraFilterLabels is a list label filters applied to every rule
+	// request withing a group. Is compatible only with VM datasources.
+	// See https://docs.victoriametrics.com#prometheus-querying-api-enhancements
+	ExtraFilterLabels map[string]string `yaml:"extra_filter_labels"`
 	// Checksum stores the hash of yaml definition for this group.
 	// May be used to detect any changes like rules re-ordering etc.
 	Checksum string

app/vmalert/config/testdata/rules2-good.rules

@@ -2,6 +2,8 @@ groups:
   - name: TestGroup
     interval: 2s
     concurrency: 2
+    extra_filter_labels:
+        job: victoriametrics
     rules:
       - alert: Conns
         expr: sum(vm_tcplistener_conns) by(instance) > 1

app/vmalert/datasource/vm.go

@@ -85,6 +85,7 @@ type VMStorage struct {
 
 	dataSourceType     Type
 	evaluationInterval time.Duration
+	extraLabels        []string
 }
 
 const queryPath = "/api/v1/query"
@@ -97,6 +98,8 @@ const graphitePrefix = "/graphite"
 type QuerierParams struct {
 	DataSourceType     *Type
 	EvaluationInterval time.Duration
+	// see https://docs.victoriametrics.com/#prometheus-querying-api-enhancements
+	ExtraLabels map[string]string
 }
 
 // Clone makes clone of VMStorage, shares http client.
@@ -119,6 +122,9 @@ func (s *VMStorage) ApplyParams(params QuerierParams) *VMStorage {
 		s.dataSourceType = *params.DataSourceType
 	}
 	s.evaluationInterval = params.EvaluationInterval
+	for k, v := range params.ExtraLabels {
+		s.extraLabels = append(s.extraLabels, fmt.Sprintf("%s=%s", k, v))
+	}
 	return s
 }
 
@@ -222,6 +228,9 @@ func (s *VMStorage) setPrometheusReqParams(r *http.Request, query string, timest
 	if s.roundDigits != "" {
 		q.Set("round_digits", s.roundDigits)
 	}
+	for _, l := range s.extraLabels {
+		q.Add("extra_label", l)
+	}
 	r.URL.RawQuery = q.Encode()
 }
 

app/vmalert/datasource/vm_test.go

@@ -253,6 +253,19 @@ func TestPrepareReq(t *testing.T) {
 				checkEqualString(t, exp, r.URL.RawQuery)
 			},
 		},
+		{
+			"extra labels",
+			&VMStorage{
+				extraLabels: []string{
+					"env=prod",
+					"query=es=cape",
+				},
+			},
+			func(t *testing.T, r *http.Request) {
+				exp := fmt.Sprintf("extra_label=env%%3Dprod&extra_label=query%%3Des%%3Dcape&query=%s&time=%d", query, timestamp.Unix())
+				checkEqualString(t, exp, r.URL.RawQuery)
+			},
+		},
 	}
 
 	for _, tc := range testCases {

app/vmalert/group.go

@@ -18,14 +18,15 @@ import (
 
 // Group is an entity for grouping rules
 type Group struct {
-	mu          sync.RWMutex
-	Name        string
-	File        string
-	Rules       []Rule
-	Type        datasource.Type
-	Interval    time.Duration
-	Concurrency int
-	Checksum    string
+	mu                sync.RWMutex
+	Name              string
+	File              string
+	Rules             []Rule
+	Type              datasource.Type
+	Interval          time.Duration
+	Concurrency       int
+	Checksum          string
+	ExtraFilterLabels map[string]string
 
 	doneCh     chan struct{}
 	finishedCh chan struct{}
@@ -51,15 +52,17 @@ func newGroupMetrics(name, file string) *groupMetrics {
 
 func newGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval time.Duration, labels map[string]string) *Group {
 	g := &Group{
-		Type:        cfg.Type,
-		Name:        cfg.Name,
-		File:        cfg.File,
-		Interval:    cfg.Interval,
-		Concurrency: cfg.Concurrency,
-		Checksum:    cfg.Checksum,
-		doneCh:      make(chan struct{}),
-		finishedCh:  make(chan struct{}),
-		updateCh:    make(chan *Group),
+		Type:              cfg.Type,
+		Name:              cfg.Name,
+		File:              cfg.File,
+		Interval:          cfg.Interval,
+		Concurrency:       cfg.Concurrency,
+		Checksum:          cfg.Checksum,
+		ExtraFilterLabels: cfg.ExtraFilterLabels,
+
+		doneCh:     make(chan struct{}),
+		finishedCh: make(chan struct{}),
+		updateCh:   make(chan *Group),
 	}
 	g.metrics = newGroupMetrics(g.Name, g.File)
 	if g.Interval == 0 {
@@ -115,6 +118,8 @@ func (g *Group) Restore(ctx context.Context, qb datasource.QuerierBuilder, lookb
 		if rr.For < 1 {
 			continue
 		}
+		// ignore g.ExtraFilterLabels on purpose, so it
+		// won't affect the restore procedure.
 		q := qb.BuildWithParams(datasource.QuerierParams{})
 		if err := rr.Restore(ctx, q, lookback, labels); err != nil {
 			return fmt.Errorf("error while restoring rule %q: %w", rule, err)
@@ -163,6 +168,7 @@ func (g *Group) updateWith(newGroup *Group) error {
 	}
 	g.Type = newGroup.Type
 	g.Concurrency = newGroup.Concurrency
+	g.ExtraFilterLabels = newGroup.ExtraFilterLabels
 	g.Checksum = newGroup.Checksum
 	g.Rules = newRules
 	return nil

app/vmalert/manager.go

@@ -147,12 +147,14 @@ func (g *Group) toAPI() APIGroup {
 
 	ag := APIGroup{
 		// encode as string to avoid rounding
-		ID:          fmt.Sprintf("%d", g.ID()),
-		Name:        g.Name,
-		Type:        g.Type.String(),
-		File:        g.File,
-		Interval:    g.Interval.String(),
-		Concurrency: g.Concurrency,
+		ID: fmt.Sprintf("%d", g.ID()),
+
+		Name:              g.Name,
+		Type:              g.Type.String(),
+		File:              g.File,
+		Interval:          g.Interval.String(),
+		Concurrency:       g.Concurrency,
+		ExtraFilterLabels: g.ExtraFilterLabels,
 	}
 	for _, r := range g.Rules {
 		switch v := r.(type) {

app/vmalert/recording.go

@@ -66,6 +66,7 @@ func newRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
 		q: qb.BuildWithParams(datasource.QuerierParams{
 			DataSourceType:     &cfg.Type,
 			EvaluationInterval: group.Interval,
+			ExtraLabels:        group.ExtraFilterLabels,
 		}),
 	}
 
@@ -151,8 +152,6 @@ func (rr *RecordingRule) toTimeSeries(m datasource.Metric, timestamp time.Time)
 }
 
 // UpdateWith copies all significant fields.
-// alerts state isn't copied since
-// it should be updated in next 2 Execs
 func (rr *RecordingRule) UpdateWith(r Rule) error {
 	nr, ok := r.(*RecordingRule)
 	if !ok {
@@ -160,6 +159,7 @@ func (rr *RecordingRule) UpdateWith(r Rule) error {
 	}
 	rr.Expr = nr.Expr
 	rr.Labels = nr.Labels
+	rr.q = nr.q
 	return nil
 }
 

app/vmalert/web_types.go

@@ -20,14 +20,15 @@ type APIAlert struct {
 
 // APIGroup represents Group for WEB view
 type APIGroup struct {
-	Name           string             `json:"name"`
-	Type           string             `json:"type"`
-	ID             string             `json:"id"`
-	File           string             `json:"file"`
-	Interval       string             `json:"interval"`
-	Concurrency    int                `json:"concurrency"`
-	AlertingRules  []APIAlertingRule  `json:"alerting_rules"`
-	RecordingRules []APIRecordingRule `json:"recording_rules"`
+	Name              string             `json:"name"`
+	Type              string             `json:"type"`
+	ID                string             `json:"id"`
+	File              string             `json:"file"`
+	Interval          string             `json:"interval"`
+	Concurrency       int                `json:"concurrency"`
+	ExtraFilterLabels map[string]string  `json:"extra_filter_labels"`
+	AlertingRules     []APIAlertingRule  `json:"alerting_rules"`
+	RecordingRules    []APIRecordingRule `json:"recording_rules"`
 }
 
 // APIAlertingRule represents AlertingRule for WEB view