chore: Recording rules overrides (#3973)

* Recording rules overrides

* use settingsv1.RecordingRule for limits config

* tests

* moving flags to v2 scope

Author: alsoba13

pkg/experiment/metrics/ruler.go

@@ -1,17 +1,14 @@
 package metrics
 
 import (
-	"encoding/json"
-	"fmt"
-	"os"
 	"sync"
 	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 
-	settingsv1 "github.com/grafana/pyroscope/api/gen/proto/go/settings/v1"
 	"github.com/grafana/pyroscope/pkg/model"
+	"github.com/grafana/pyroscope/pkg/validation"
 )
 
 const (
@@ -20,39 +17,24 @@ const (
 )
 
 type StaticRuler struct {
-	rules  map[string][]*model.RecordingRule
-	logger log.Logger
+	overrides *validation.Overrides
 }
 
-func NewStaticRulerFromEnvVars(logger log.Logger) (Ruler, error) {
-	jsonRules := os.Getenv(envVarRecordingRules)
-
-	var rulesByTenant map[string][]*settingsv1.RecordingRule
-	if err := json.Unmarshal([]byte(jsonRules), &rulesByTenant); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal recording rules: %w", err)
-	}
-
-	ruler := &StaticRuler{
-		rules:  make(map[string][]*model.RecordingRule, len(rulesByTenant)),
-		logger: logger,
+func NewStaticRulerFromOverrides(overrides *validation.Overrides) Ruler {
+	return &StaticRuler{
+		overrides: overrides,
 	}
-	for tenant, rules := range rulesByTenant {
-		rs := make([]*model.RecordingRule, 0, len(rules))
-		for _, rule := range rules {
-			r, err := model.NewRecordingRule(rule)
-			if err == nil {
-				rs = append(rs, r)
-			} else {
-				level.Error(logger).Log("msg", "failed to parse recording rule", "rule", rule, "err", err)
-			}
-		}
-		ruler.rules[tenant] = rs
-	}
-	return ruler, nil
 }
 
-func (r StaticRuler) RecordingRules(tenant string) []*model.RecordingRule {
-	return r.rules[tenant]
+func (ruler StaticRuler) RecordingRules(tenant string) []*model.RecordingRule {
+	rules := ruler.overrides.RecordingRules(tenant)
+	rs := make([]*model.RecordingRule, 0, len(rules))
+	for _, rule := range rules {
+		// should never fail, overrides already validated
+		r, _ := model.NewRecordingRule(rule)
+		rs = append(rs, r)
+	}
+	return rs
 }
 
 // CachedRemoteRuler is a thread-safe ruler that retrieves rules from an external service.

pkg/experiment/metrics/ruler_test.go

@@ -1,95 +1,73 @@
 package metrics
 
 import (
-	"bytes"
-	"os"
-	"strings"
 	"testing"
 
-	"github.com/go-kit/log"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/stretchr/testify/assert"
 
+	settingsv1 "github.com/grafana/pyroscope/api/gen/proto/go/settings/v1"
+	typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1"
 	"github.com/grafana/pyroscope/pkg/model"
+	"github.com/grafana/pyroscope/pkg/validation"
 )
 
-func Test_Ruler_missconfigured(t *testing.T) {
-	_, err := NewStaticRulerFromEnvVars(log.NewNopLogger())
-	assert.EqualError(t, err, "failed to unmarshal recording rules: unexpected end of JSON input", "Empty env var should result in error at creating ruler")
-}
-
-func Test_Ruler_happyPath(t *testing.T) {
-	jsonRecordingRules :=
-		`{
-					"tenant1": [{
-            "metric_name": "metric1",
-            "matchers": ["{__profile_type__=\"profile_type\", pod=\"foo\"}"],
-            "group_by": ["bar"],
-            "external_labels": [{"name":"__name__", "value":"metric1"}]
-          }],
-          "tenant2": [{
-            "metric_name": "metric2",
-            "matchers": ["{__profile_type__=\"profile_type\"}"],
-            "group_by": [],
-            "external_labels": [{"name":"__name__", "value":"should be ignored"}]
-          },
-          {
-            "metric_name": "no_profile_type",
-            "matchers": ["{}"],
-            "group_by": [],
-            "external_labels": []
-          },
-          {
-            "metric_name": "Wrong metric name",
-            "matchers": ["{__profile_type__=\"profile_type\"}"],
-            "group_by": [],
-            "external_labels": []
-          },
-          {
-            "metric_name": "wrong_matcher",
-            "matchers": ["{foo==\"bar\"}"],
-            "group_by": [],
-            "external_labels": []
-          }]
-	      }`
-	_ = os.Setenv(envVarRecordingRules, jsonRecordingRules)
-	loggerBuffer := &bytes.Buffer{}
-	logger := log.NewJSONLogger(loggerBuffer)
-	ruler, err := NewStaticRulerFromEnvVars(logger)
-	assert.NoError(t, err)
+var (
+	defaultRecordingRulesProto = []*settingsv1.RecordingRule{{
+		MetricName: "default_recording_rule",
+		Matchers:   []string{"{__profile_type__=\"any-profile-type\"}"},
+	}}
 
-	rules := ruler.RecordingRules("tenant1")
-	assert.Equal(t, []*model.RecordingRule{{
-		Matchers: []*labels.Matcher{{
-			Type:  labels.MatchEqual,
-			Name:  "__profile_type__",
-			Value: "profile_type",
-		}, {
-			Type:  labels.MatchEqual,
-			Name:  "pod",
-			Value: "foo",
+	defaultRecordingRules = []*model.RecordingRule{{
+		ExternalLabels: labels.Labels{{
+			Name:  "__name__",
+			Value: "default_recording_rule",
 		}},
-		GroupBy: []string{"bar"},
-		ExternalLabels: labels.Labels{
-			{Name: "__name__", Value: "metric1"},
-		},
-	}}, rules)
-
-	rules = ruler.RecordingRules("tenant2")
-	assert.Equal(t, []*model.RecordingRule{{
 		Matchers: []*labels.Matcher{{
 			Type:  labels.MatchEqual,
 			Name:  "__profile_type__",
-			Value: "profile_type",
+			Value: "any-profile-type",
 		}},
-		GroupBy: []string{},
+	}}
+
+	overriddenRecordingRulesProto = []*settingsv1.RecordingRule{{
+		MetricName:     "rule",
+		Matchers:       []string{"{__profile_type__=\"any-profile-type\", matcher1!=\"value\"}"},
+		GroupBy:        []string{"group_by_label"},
+		ExternalLabels: []*typesv1.LabelPair{{Name: "foo", Value: "bar"}},
+	}}
+
+	overriddenRecordingRules = []*model.RecordingRule{{
+		Matchers: []*labels.Matcher{
+			{Type: labels.MatchEqual, Name: "__profile_type__", Value: "any-profile-type"},
+			{Type: labels.MatchNotEqual, Name: "matcher1", Value: "value"},
+		},
+		GroupBy: []string{"group_by_label"},
 		ExternalLabels: labels.Labels{
-			{Name: "__name__", Value: "metric2"},
+			{Name: "foo", Value: "bar"},
+			{Name: "__name__", Value: "rule"},
 		},
-	}}, rules)
+	}}
+)
+
+func Test_Ruler_happyPath(t *testing.T) {
+	overrides := newOverrides(t)
+
+	ruler := NewStaticRulerFromOverrides(overrides)
+
+	rules := ruler.RecordingRules("non-configured-tenant")
+	assert.Equal(t, defaultRecordingRules, rules)
+
+	rules = ruler.RecordingRules("tenant-override")
+	assert.Equal(t, overriddenRecordingRules, rules)
+}
 
-	loggedLines := strings.Split(loggerBuffer.String(), "\n")
-	assert.True(t, strings.Contains(loggedLines[0], "no __profile_type__ matcher present"))
-	assert.True(t, strings.Contains(loggedLines[1], "invalid metric name: Wrong metric name"))
-	assert.True(t, strings.Contains(loggedLines[2], "failed to parse matchers: 1:6: parse error: unexpected \\\"=\\\" in label matching, expected string"))
+func newOverrides(t *testing.T) *validation.Overrides {
+	t.Helper()
+	return validation.MockOverrides(func(defaults *validation.Limits, tenantLimits map[string]*validation.Limits) {
+		defaults.RecordingRules = defaultRecordingRulesProto
+		l := validation.MockDefaultLimits()
+		l.RecordingRules = overriddenRecordingRulesProto
+		tenantLimits["tenant-override"] = l
+	})
 }

pkg/phlare/modules_experimental.go

@@ -105,11 +105,11 @@ func (f *Phlare) initCompactionWorker() (svc services.Service, err error) {
 	if f.Cfg.CompactionWorker.MetricsExporter.Enabled {
 		if f.recordingRulesClient != nil {
 			ruler, err = metrics.NewCachedRemoteRuler(f.recordingRulesClient, f.logger)
+			if err != nil {
+				return nil, err
+			}
 		} else {
-			ruler, err = metrics.NewStaticRulerFromEnvVars(f.logger)
-		}
-		if err != nil {
-			return nil, err
+			ruler = metrics.NewStaticRulerFromOverrides(f.Overrides)
 		}
 
 		exporter, err = metrics.NewStaticExporterFromEnvVars(f.logger, f.reg)

pkg/phlare/phlare.go

@@ -225,14 +225,15 @@ func (c *Config) registerServerFlagsWithChangedDefaultValues(fs *flag.FlagSet) {
 	c.v2Experiment = os.Getenv("PYROSCOPE_V2_EXPERIMENT") != ""
 	if c.v2Experiment {
 		for k, v := range map[string]string{
-			"server.grpc-max-recv-msg-size-bytes":               "104857600",
-			"server.grpc-max-send-msg-size-bytes":               "104857600",
-			"server.grpc.keepalive.min-time-between-pings":      "1s",
-			"segment-writer.grpc-client-config.connect-timeout": "1s",
-			"segment-writer.num-tokens":                         "4",
-			"segment-writer.heartbeat-timeout":                  "1m",
-			"segment-writer.unregister-on-shutdown":             "false",
-			"segment-writer.min-ready-duration":                 "30s",
+			"server.grpc-max-recv-msg-size-bytes":                    "104857600",
+			"server.grpc-max-send-msg-size-bytes":                    "104857600",
+			"server.grpc.keepalive.min-time-between-pings":           "1s",
+			"segment-writer.grpc-client-config.connect-timeout":      "1s",
+			"segment-writer.num-tokens":                              "4",
+			"segment-writer.heartbeat-timeout":                       "1m",
+			"segment-writer.unregister-on-shutdown":                  "false",
+			"segment-writer.min-ready-duration":                      "30s",
+			"compaction-worker.metrics-exporter.rules-source.static": "[]",
 		} {
 			overrides[k] = v
 		}
@@ -245,6 +246,7 @@ func (c *Config) registerServerFlagsWithChangedDefaultValues(fs *flag.FlagSet) {
 		c.LimitsConfig.WritePathOverrides.RegisterFlags(throwaway)
 		c.LimitsConfig.ReadPathOverrides.RegisterFlags(throwaway)
 		c.LimitsConfig.AdaptivePlacementLimits.RegisterFlags(throwaway)
+		c.LimitsConfig.RecordingRules.RegisterFlags(throwaway)
 	}
 
 	throwaway.VisitAll(func(f *flag.Flag) {

pkg/validation/limits.go

@@ -14,6 +14,7 @@ import (
 	writepath "github.com/grafana/pyroscope/pkg/distributor/write_path"
 	"github.com/grafana/pyroscope/pkg/experiment/distributor/placement/adaptive_placement"
 	readpath "github.com/grafana/pyroscope/pkg/frontend/read_path"
+	phlaremodel "github.com/grafana/pyroscope/pkg/model"
 	"github.com/grafana/pyroscope/pkg/phlaredb/block"
 )
 
@@ -113,6 +114,10 @@ type Limits struct {
 	// Distributors use these limits to determine how many shards to allocate
 	// to a tenant dataset by default, if no placement rules defined.
 	AdaptivePlacementLimits adaptive_placement.PlacementLimits `yaml:",inline" json:",inline"`
+
+	// RecordingRules allow to specify static recording rules. This is not compatible with recording rules
+	// coming from a RecordingRulesClient, that will replace any static rules defined.
+	RecordingRules RecordingRules `yaml:"recording_rules" json:"recording_rules" category:"experimental" doc:"hidden"`
 }
 
 // LimitError are errors that do not comply with the limits specified.
@@ -209,13 +214,19 @@ func (l *Limits) UnmarshalYAML(unmarshal func(interface{}) error) error {
 
 // Validate validates that this limits config is valid.
 func (l *Limits) Validate() error {
-
 	if l.IngestionRelabelingDefaultRulesPosition != "" {
 		if err := l.IngestionRelabelingDefaultRulesPosition.Set(string(l.IngestionRelabelingDefaultRulesPosition)); err != nil {
 			return err
 		}
 	}
 
+	for idx, rule := range l.RecordingRules {
+		_, err := phlaremodel.NewRecordingRule(rule)
+		if err != nil {
+			return fmt.Errorf("rule at pos %d is not valid: %v", idx, err)
+		}
+	}
+
 	return nil
 }
 

pkg/validation/limits_test.go

@@ -75,6 +75,13 @@ shard_streams:
 blocked_queries:
   - pattern: ".*foo.*"
     regex: true
+recording_rules:
+  - metric_name: 'any-name'
+    matchers: ['any-matcher']
+    group_by: ['any-group-by']
+    external_labels:
+      - name: 'any-label-name'
+        value: 'any-label-value'
 `
 	inputJSON := `
  {
@@ -120,10 +127,23 @@ blocked_queries:
     "logging_enabled": true
   },
   "blocked_queries": [
-	{
-		"pattern": ".*foo.*",
-		"regex": true
-	}
+  {
+    "pattern": ".*foo.*",
+    "regex": true
+  }
+  ],
+  "recording_rules": [
+  {
+    "metric_name": "any-name",
+    "matchers": ["any-matcher"],
+    "group_by": ["any-group-by"],
+    "external_labels": [
+    {
+      "name" : "any-label-name",
+      "value" : "any-label-value"
+    }
+    ]
+  }
   ]
  }
 `