diff --git a/cmd/pyroscope/help-all.txt.tmpl b/cmd/pyroscope/help-all.txt.tmpl index 2edfb0d484..40186b63b6 100644 --- a/cmd/pyroscope/help-all.txt.tmpl +++ b/cmd/pyroscope/help-all.txt.tmpl @@ -68,11 +68,11 @@ Usage of ./pyroscope: -compaction-worker.metadata-source string Source of the compaction job source block metadata. Supported values: metastore, object-storage. (default "metastore") -compaction-worker.metrics-exporter.enabled - This parameter specifies whether the metrics exporter is enabled. + Deprecated: use recording-rules.enabled. Specifies whether the metrics exporter is enabled. -compaction-worker.metrics-exporter.remote-write-address string The address to use for metrics tenant. -compaction-worker.metrics-exporter.rules-source.client-address string - The address to use for the recording rules client connection. + Deprecated: use recording-rules.client-address. The address to use for the recording rules client connection. -compaction-worker.metrics-exporter.rules-source.static value [experimental] List of static recording rules of the type settingsv1.RecordingRule. Will only be use in the absence of a recording rules client. (default []) -compaction-worker.request-timeout duration @@ -1221,6 +1221,10 @@ Usage of ./pyroscope: Backend storage to use for the ring. Supported values are: consul, etcd, inmemory, memberlist, multi. (default "memberlist") -query-scheduler.service-discovery-mode string [experimental] Service discovery mode that query-frontends and queriers use to find query-scheduler instances. When query-scheduler ring-based service discovery is enabled, this option needs be set on query-schedulers, query-frontends and queriers. Supported values are: dns, ring. (default "ring") + -recording-rules.client-address string + Address of the recording rules source (the tenant-settings service). When empty, static rules from per-tenant overrides are used. Read by the distributor and compaction-worker. + -recording-rules.enabled + Enable generating metrics from profiles using recording rules. Read by the query-frontend, distributor and compaction-worker. -recording-rules.max-rules-per-tenant int Maximum number of recording rules a tenant can create. 0 to disable. (default 25) -retention-period duration diff --git a/cmd/pyroscope/help.txt.tmpl b/cmd/pyroscope/help.txt.tmpl index b55d5900b6..f9bf661b05 100644 --- a/cmd/pyroscope/help.txt.tmpl +++ b/cmd/pyroscope/help.txt.tmpl @@ -215,6 +215,8 @@ Usage of ./pyroscope: List of network interface names to look up when finding the instance IP address. (default []) -query-scheduler.ring.store string Backend storage to use for the ring. Supported values are: consul, etcd, inmemory, memberlist, multi. (default "memberlist") + -recording-rules.enabled + Enable generating metrics from profiles using recording rules. Read by the query-frontend, distributor and compaction-worker. -recording-rules.max-rules-per-tenant int Maximum number of recording rules a tenant can create. 0 to disable. (default 25) -retention-period duration diff --git a/cmd/pyroscope/pyroscope.yaml b/cmd/pyroscope/pyroscope.yaml index 0a34e3c61e..59922e98bb 100644 --- a/cmd/pyroscope/pyroscope.yaml +++ b/cmd/pyroscope/pyroscope.yaml @@ -614,6 +614,11 @@ runtime_config: # right. # file: "" +recording_rules: + # Enable generating metrics from profiles using recording rules. Read by the + # query-frontend, distributor and compaction-worker. + # enabled: false + storage: # Backend storage to use. Supported backends are: s3, gcs, azure, swift, # filesystem, cos. diff --git a/docs/sources/configure-server/reference-configuration-parameters/index.md b/docs/sources/configure-server/reference-configuration-parameters/index.md index 6cc0050f3b..c502efe949 100644 --- a/docs/sources/configure-server/reference-configuration-parameters/index.md +++ b/docs/sources/configure-server/reference-configuration-parameters/index.md @@ -185,6 +185,18 @@ tenant_settings: # CLI flag: -tenant-settings.recording-rules.enabled [enabled: | default = false] +recording_rules: + # Enable generating metrics from profiles using recording rules. Read by the + # query-frontend, distributor and compaction-worker. + # CLI flag: -recording-rules.enabled + [enabled: | default = false] + + # (advanced) Address of the recording rules source (the tenant-settings + # service). When empty, static rules from per-tenant overrides are used. Read + # by the distributor and compaction-worker. + # CLI flag: -recording-rules.client-address + [client_address: | default = ""] + storage: # Backend storage to use. Supported backends are: s3, gcs, azure, swift, # filesystem, cos. @@ -1412,12 +1424,14 @@ The `compaction_worker` block configures the compaction-worker (V2). [metadata_fetch_timeout: | default = 30s] metrics_exporter: - # (advanced) This parameter specifies whether the metrics exporter is enabled. + # (advanced) Deprecated: use recording-rules.enabled. Specifies whether the + # metrics exporter is enabled. # CLI flag: -compaction-worker.metrics-exporter.enabled [enabled: | default = false] rules_source: - # (advanced) The address to use for the recording rules client connection. + # (advanced) Deprecated: use recording-rules.client-address. The address to + # use for the recording rules client connection. # CLI flag: -compaction-worker.metrics-exporter.rules-source.client-address [client_address: | default = ""] diff --git a/pkg/block/compaction_test.go b/pkg/block/compaction_test.go index 6655458e6e..749962227c 100644 --- a/pkg/block/compaction_test.go +++ b/pkg/block/compaction_test.go @@ -192,7 +192,7 @@ func Test_CompactBlocks_recordingRules(t *testing.T) { ExternalLabels: labels.New(labels.Label{Name: "__name__", Value: "profiles_recorded_cpu_usage_function_total_servehttp_samples"}), FunctionName: "net/http.HandlerFunc.ServeHTTP", }, - }) + }, nil) sampleObserver := metrics.NewSampleObserver(0, exporter, ruler, labels.EmptyLabels()) compactedBlocks, err := block.Compact(ctx, resp.Blocks, bucket, @@ -298,7 +298,7 @@ func Test_CompactBlocks_recordingRules_shadowedSymbols(t *testing.T) { ExternalLabels: labels.New(labels.Label{Name: "__name__", Value: "profiles_recorded_mem_alloc_total_query_Serve_bytes"}), FunctionName: "net/http.HandlerFunc.ServeHTTP", }, - }) + }, nil) sampleObserver := metrics.NewSampleObserver(0, exporter, ruler, labels.EmptyLabels()) compactedBlocks, err := block.Compact(ctx, resp.Blocks, bucket, diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go index 4478f57fec..1ffe3ed761 100644 --- a/pkg/distributor/distributor.go +++ b/pkg/distributor/distributor.go @@ -176,6 +176,7 @@ func New( reg prometheus.Registerer, logger log.Logger, segmentWriter SegmentWriterClient, + ruler sampling.Ruler, ingesterClientsOptions ...connect.ClientOption, ) (*Distributor, error) { ingesterClientsOptions = append( @@ -205,7 +206,7 @@ func New( profileReceivedStats: usagestats.NewMultiCounter("distributor_profiles_received", "lang"), profileScopeStats: usagestats.NewMultiCounter("distributor_profiles_received_by_scope", "scope"), profileSizeStats: usagestats.NewMultiStatistics("distributor_profile_sizes", "lang"), - stripper: sampling.NewProfileStripper(), + stripper: sampling.NewProfileStripper(ruler), inflight: inflight.NewLimiter(config.MaxInflightBytes, m.inflightBytesHighWatermark), } m.inflightBytesLimit.Set(float64(config.MaxInflightBytes)) @@ -658,12 +659,15 @@ func (d *Distributor) pushSeries(ctx context.Context, req *distributormodel.Prof groups.CountDiscardedBytes(string(validation.SkippedBySamplingRules), req.TotalBytesUncompressed) return nil } - finalLog.msg = "stripping profile stacktraces, keeping totals" + finalLog.msg = "stripping profile stacktraces, keeping totals and recording-rule targeted samples" // Language detection reads the string table, which is about to be // stripped; the result is cached in the request. d.GetProfileLanguage(req) - d.stripper.StripToTotals(req.Profile.Profile) + if d.stripper.Strip(tenantID, req.Labels, req.Profile.Profile) { + d.metrics.strippedWithoutRules.WithLabelValues(tenantID).Add(float64(req.TotalProfiles)) + finalLog.addFields("recording_rules_unavailable", true) + } req.Labels = phlaremodel.Labels(req.Labels).InsertSorted(phlaremodel.LabelNameSampled, "true") // The stripped part of the profile is discarded, and from here on diff --git a/pkg/distributor/distributor_inflight_test.go b/pkg/distributor/distributor_inflight_test.go index d71e36ab47..31a53e9c15 100644 --- a/pkg/distributor/distributor_inflight_test.go +++ b/pkg/distributor/distributor_inflight_test.go @@ -68,7 +68,7 @@ func newInflightDistributor(t *testing.T, reg prometheus.Registerer, maxInflight Config{DistributorRing: ringConfig, MaxInflightBytes: maxInflightBytes}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "foo"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil }}, - overrides, reg, log.NewNopLogger(), sw, + overrides, reg, log.NewNopLogger(), sw, nil, ) require.NoError(t, err) return d @@ -266,7 +266,7 @@ func TestPushBatch_MaxInflightBytes_IngesterQuorum(t *testing.T) { Config{DistributorRing: ringConfig, PushTimeout: time.Minute}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "1"}, {Addr: "2"}, {Addr: "3"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ingesters[addr], nil }}, - overrides, nil, log.NewNopLogger(), nil, + overrides, nil, log.NewNopLogger(), nil, nil, ) require.NoError(t, err) @@ -325,7 +325,7 @@ func TestPushBatch_MaxInflightBytes_Aggregation(t *testing.T) { Config{DistributorRing: ringConfig, PushTimeout: time.Minute}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "foo"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil }}, - overrides, nil, log.NewNopLogger(), sw, + overrides, nil, log.NewNopLogger(), sw, nil, ) require.NoError(t, err) sw.limiter = d.inflight @@ -372,7 +372,7 @@ func TestPushBatch_MaxInflightBytes_PendingAggregates(t *testing.T) { Config{DistributorRing: ringConfig, PushTimeout: time.Minute}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "foo"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil }}, - overrides, nil, log.NewNopLogger(), sw, + overrides, nil, log.NewNopLogger(), sw, nil, ) require.NoError(t, err) sw.limiter = d.inflight @@ -427,7 +427,7 @@ func TestPushBatch_MaxInflightBytes_AggregateGrowth(t *testing.T) { Config{DistributorRing: ringConfig, PushTimeout: time.Minute}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "foo"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil }}, - overrides, nil, log.NewNopLogger(), sw, + overrides, nil, log.NewNopLogger(), sw, nil, ) require.NoError(t, err) sw.limiter = d.inflight diff --git a/pkg/distributor/distributor_pushbatch_test.go b/pkg/distributor/distributor_pushbatch_test.go index c4f07b8c13..c55f671b4d 100644 --- a/pkg/distributor/distributor_pushbatch_test.go +++ b/pkg/distributor/distributor_pushbatch_test.go @@ -100,7 +100,7 @@ func newProbeDistributor(t *testing.T, sw SegmentWriterClient, maxConcurrency in Config{DistributorRing: ringConfig}, testhelper.NewMockRing([]ring.InstanceDesc{{Addr: "foo"}}, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil }}, - overrides, nil, log.NewNopLogger(), sw, + overrides, nil, log.NewNopLogger(), sw, nil, ) require.NoError(t, err) return d diff --git a/pkg/distributor/distributor_readiness_test.go b/pkg/distributor/distributor_readiness_test.go index 824804250b..0f42f9ee44 100644 --- a/pkg/distributor/distributor_readiness_test.go +++ b/pkg/distributor/distributor_readiness_test.go @@ -63,6 +63,7 @@ func newReadinessDistributor( nil, logger, swClient, + nil, ) require.NoError(t, err) require.NoError(t, services.StartAndAwaitRunning(ctx, d)) diff --git a/pkg/distributor/distributor_recvmetric_test.go b/pkg/distributor/distributor_recvmetric_test.go index d6a3e0c929..6c7aa7512c 100644 --- a/pkg/distributor/distributor_recvmetric_test.go +++ b/pkg/distributor/distributor_recvmetric_test.go @@ -131,6 +131,7 @@ func TestDistributorPushWithDifferentTenantStages(t *testing.T) { reg, log.NewLogfmtLogger(os.Stdout), nil, + nil, ) require.NoError(t, err) diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go index 32365736ce..2ad3b81e94 100644 --- a/pkg/distributor/distributor_test.go +++ b/pkg/distributor/distributor_test.go @@ -85,7 +85,7 @@ func newTestDistributor(t testing.TB, logger log.Logger, overrides *validation.O {Addr: "foo"}, }, 3), &poolFactory{func(addr string) (client.PoolClient, error) { return ing, nil - }}, overrides, nil, logger, nil) + }}, overrides, nil, logger, nil, nil) return d, ing, err } @@ -157,7 +157,7 @@ func Test_Replication(t *testing.T) { {Addr: "3"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ingesters[addr], nil - }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) // only 1 ingester failing should be fine. resp, err := d.Push(ctx, req) @@ -291,7 +291,7 @@ func Test_Subservices(t *testing.T) { {Addr: "foo"}, }, 1), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) require.NoError(t, d.StartAsync(context.Background())) @@ -454,7 +454,7 @@ func Test_Limits(t *testing.T) { {Addr: "foo"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, tc.overrides, nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, tc.overrides, nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) @@ -546,7 +546,7 @@ func Test_Sessions_Limit(t *testing.T) { l := validation.MockDefaultLimits() l.MaxSessionsPerSeries = tc.maxSessions tenantLimits["user-1"] = l - }), nil, log.NewLogfmtLogger(os.Stdout), nil) + }), nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) limit := d.limits.MaxSessionsPerSeries("user-1") @@ -814,7 +814,7 @@ func Test_IngestLimits(t *testing.T) { {Addr: "foo"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, tc.overrides, nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, tc.overrides, nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) err = d.PushBatch(tenant.InjectTenantID(context.Background(), "user-1"), tc.pushReq) @@ -1354,7 +1354,7 @@ func Test_SampleLabels_Ingester(t *testing.T) { {Addr: "foo"}, }, 3), &poolFactory{func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil - }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) var series []*distributormodel.ProfileSeries series, err = d.visitSampleSeries(tc.pushReq, visitSampleSeriesForIngester) @@ -1911,7 +1911,7 @@ func Test_SampleLabels_SegmentWriter(t *testing.T) { {Addr: "foo"}, }, 3), &poolFactory{func(addr string) (client.PoolClient, error) { return newFakeIngester(t, false), nil - }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) var series []*distributormodel.ProfileSeries @@ -1947,7 +1947,7 @@ func TestBadPushRequest(t *testing.T) { {Addr: "foo"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) mux.Handle(pushv1connect.NewPusherServiceHandler(d, handlerOptions...)) @@ -2028,6 +2028,7 @@ func TestPush_ShuffleSharding(t *testing.T) { nil, log.NewLogfmtLogger(os.Stdout), nil, + nil, ) require.NoError(t, err) @@ -2140,7 +2141,7 @@ func TestPush_Aggregation(t *testing.T) { } tenantLimits["user-1"] = l }), - nil, log.NewLogfmtLogger(os.Stdout), nil, + nil, log.NewLogfmtLogger(os.Stdout), nil, nil, ) require.NoError(t, err) ctx := tenant.InjectTenantID(context.Background(), "user-1") @@ -2229,6 +2230,7 @@ func TestPushBatch_SeriesHistogram(t *testing.T) { reg, log.NewLogfmtLogger(os.Stdout), nil, + nil, ) require.NoError(t, err) @@ -2466,7 +2468,7 @@ func TestPush_LabelRewrites(t *testing.T) { {Addr: "mock"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, newOverrides(t), nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) ctx := tenant.InjectTenantID(context.Background(), "user-1") @@ -2835,7 +2837,7 @@ func newStripTestDistributor(t *testing.T, keepStrippedProfiles bool) (*Distribu {Addr: "foo"}, }, 3), &poolFactory{f: func(addr string) (client.PoolClient, error) { return ing, nil - }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil) + }}, overrides, nil, log.NewLogfmtLogger(os.Stdout), nil, nil) require.NoError(t, err) return d, ing } @@ -2890,6 +2892,101 @@ func TestPushSeries_KeepStrippedProfiles(t *testing.T) { assert.Equal(t, keptSize, received.SizeVT()) } +type fakeRuler struct { + rules []*phlaremodel.RecordingRule +} + +func (r fakeRuler) RecordingRules(string) ([]*phlaremodel.RecordingRule, error) { + return r.rules, nil +} + +type failingRuler struct{} + +func (failingRuler) RecordingRules(string) ([]*phlaremodel.RecordingRule, error) { + return nil, errors.New("rules unavailable") +} + +func TestPushSeries_KeepStrippedProfiles_RulesUnavailable(t *testing.T) { + d, _ := newStripTestDistributor(t, true) + d.stripper = sampling.NewProfileStripper(failingRuler{}) + + counter := d.metrics.strippedWithoutRules.WithLabelValues("user-1") + before := testutil.ToFloat64(counter) + + req := &distributormodel.ProfileSeries{ + Labels: []*typesv1.LabelPair{ + {Name: "__name__", Value: "cpu"}, + {Name: phlaremodel.LabelNameServiceName, Value: "svc"}, + }, + Profile: pprof2.RawFromProto(stripTestProfile()), + TotalProfiles: 1, + } + require.NoError(t, d.pushSeries(context.Background(), req, distributormodel.RawProfileTypePPROF, "user-1", 0)) + + assert.Equal(t, before+1, testutil.ToFloat64(counter)) +} + +// A total rule that groups by a sample label makes the stripped totals carry +// that label, which the pprof split then promotes to series labels. This drives +// a labeled totals sample through the full push -> split path: the label slices +// must not alias source samples (which keep dropped labels past len()), or the +// split panics indexing the rebuilt string table. +func TestPushSeries_KeepStrippedProfiles_SampleSeries(t *testing.T) { + d, ing := newStripTestDistributor(t, true) + d.stripper = sampling.NewProfileStripper(fakeRuler{rules: []*phlaremodel.RecordingRule{ + {GroupBy: []string{"user_id"}}, // total rule (no FunctionName) + }}) + + prof := stripTestProfile() + // user_id=9, u1=10, u2=11, bytes=12. bytes is unreferenced by the rule and + // gets dropped, leaving it past len() in the sample's label backing array. + prof.StringTable = append(prof.StringTable, "user_id", "u1", "u2", "bytes") + prof.Sample = []*profilev1.Sample{ + {LocationId: []uint64{1}, Value: []int64{10, 1000}, Label: []*profilev1.Label{ + {Key: 9, Str: 10}, + {Key: 12, Num: 42}, + }}, + {LocationId: []uint64{2}, Value: []int64{5, 500}, Label: []*profilev1.Label{ + {Key: 9, Str: 11}, + }}, + } + + req := &distributormodel.ProfileSeries{ + Labels: []*typesv1.LabelPair{ + {Name: "__name__", Value: "cpu"}, + {Name: phlaremodel.LabelNameServiceName, Value: "svc"}, + }, + Profile: pprof2.RawFromProto(prof), + } + + err := d.pushSeries(context.Background(), req, distributormodel.RawProfileTypePPROF, "user-1", 0) + require.NoError(t, err) + + ing.mtx.Lock() + defer ing.mtx.Unlock() + byUser := make(map[string]*pushv1.RawProfileSeries) + for _, r := range ing.requests { + for _, s := range r.Series { + byUser[phlaremodel.Labels(s.Labels).Get("user_id")] = s + } + } + require.Len(t, byUser, 2) + + for user, want := range map[string][]int64{"u1": {10, 1000}, "u2": {5, 500}} { + series := byUser[user] + require.NotNil(t, series, "missing series for user_id=%s", user) + assert.Equal(t, "true", phlaremodel.Labels(series.Labels).Get(phlaremodel.LabelNameSampled)) + + received, err := pprof2.RawFromBytes(series.Samples[0].RawProfile) + require.NoError(t, err) + require.Len(t, received.Sample, 1) + assert.Equal(t, want, received.Sample[0].Value) + assert.Empty(t, received.Sample[0].LocationId) + assert.Empty(t, received.Sample[0].Label) + assert.NotContains(t, received.StringTable, "bytes") + } +} + func TestPushSeries_DropSampledOutProfiles(t *testing.T) { d, ing := newStripTestDistributor(t, false) diff --git a/pkg/distributor/metrics.go b/pkg/distributor/metrics.go index 8f0326d018..5f2d77fefe 100644 --- a/pkg/distributor/metrics.go +++ b/pkg/distributor/metrics.go @@ -46,6 +46,7 @@ type metrics struct { inflightBytesHighWatermark prometheus.Summary rejectedRequests *prometheus.CounterVec profileIDGeneration *prometheus.CounterVec + strippedWithoutRules *prometheus.CounterVec } func newMetrics(reg prometheus.Registerer) *metrics { @@ -193,6 +194,14 @@ func newMetrics(reg prometheus.Registerer) *metrics { }, []string{"source"}, ), + strippedWithoutRules: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "pyroscope", + Name: "distributor_stripped_profiles_without_rules_total", + Help: "Number of sampled-out profiles stripped to totals because the tenant's recording rules could not be loaded.", + }, + []string{"tenant"}, + ), } if reg != nil { reg.MustRegister( @@ -210,6 +219,7 @@ func newMetrics(reg prometheus.Registerer) *metrics { m.inflightBytesHighWatermark, m.rejectedRequests, m.profileIDGeneration, + m.strippedWithoutRules, ) } // Initialize expected rejected request labels, so the series exists before diff --git a/pkg/distributor/sampling/profile_stripper.go b/pkg/distributor/sampling/profile_stripper.go index 79e8550461..fe830b18fd 100644 --- a/pkg/distributor/sampling/profile_stripper.go +++ b/pkg/distributor/sampling/profile_stripper.go @@ -1,43 +1,48 @@ package sampling import ( + "sort" + + "github.com/prometheus/prometheus/model/labels" + profilev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1" + phlaremodel "github.com/grafana/pyroscope/v2/pkg/model" + "github.com/grafana/pyroscope/v2/pkg/pprof" ) // ProfileStripper reduces profiles that were sampled out but should be kept // in a minimal form. -type ProfileStripper struct{} +type ProfileStripper struct { + ruler Ruler +} -func NewProfileStripper() *ProfileStripper { - return &ProfileStripper{} +func NewProfileStripper(ruler Ruler) *ProfileStripper { + return &ProfileStripper{ruler: ruler} } -// StripToTotals reduces the profile to a single sample holding the -// sum of all sample values: stacktraces, symbols, and sample labels are -// dropped, only the series totals are kept. -func (*ProfileStripper) StripToTotals(p *profilev1.Profile) { - var total *profilev1.Sample - for _, s := range p.Sample { - // Samples that Normalize would drop (value length mismatch, - // negative values) must not contribute to the totals. - if len(s.Value) != len(p.SampleType) || hasNegativeValue(s) { - continue - } - if total == nil { - total = &profilev1.Sample{Value: make([]int64, len(p.SampleType))} - } - for i, v := range s.Value { - total.Value[i] += v - } - } - p.Sample = nil - if total != nil { - p.Sample = []*profilev1.Sample{total} - } - p.Location = nil - p.Function = nil - p.Mapping = nil +// Strip reduces the profile to a single sample holding the sum of all sample +// values: stacktraces, symbols, and sample labels are dropped, only the series +// totals are kept. Stacktraces targeted by the tenant's recording rules are the +// exception: they are kept, trimmed to the frames of the targeted functions, +// together with the labels those rules need to match and group. +// +// It returns true when the tenant's rules could not be loaded. The profile is +// then stripped to plain totals, so the stacktraces the rules need are lost. +func (s *ProfileStripper) Strip(tenantID string, seriesLabels phlaremodel.Labels, p *profilev1.Profile) (rulesUnavailable bool) { + exc := s.computeExceptions(tenantID, seriesLabels, p) + + stripSamples(p, exc) + + pruneUnreferencedSymbols(p) + compactStringTable(p) + + return exc.rulesUnavailable +} + +// compactStringTable rebuilds p.StringTable so it holds only the strings still +// referenced after stripping, remapping every reference to the new indices. +func compactStringTable(p *profilev1.Profile) { oldStrings := p.StringTable newStrings := []string{""} remap := map[int64]int64{0: 0} @@ -64,9 +69,210 @@ func (*ProfileStripper) StripToTotals(p *profilev1.Profile) { for i, c := range p.Comment { p.Comment[i] = intern(c) } + for _, fn := range p.Function { + fn.Name = intern(fn.Name) + fn.SystemName = intern(fn.SystemName) + fn.Filename = intern(fn.Filename) + } + for _, m := range p.Mapping { + m.Filename = intern(m.Filename) + m.BuildId = intern(m.BuildId) + } + for _, sample := range p.Sample { + for _, l := range sample.Label { + l.Key = intern(l.Key) + l.Str = intern(l.Str) + l.NumUnit = intern(l.NumUnit) + } + } p.StringTable = newStrings } +// stripSamples reduces p.Sample to the samples that hit an exception location +// (trimmed to those locations and to the labels the matching rules need) plus +// one totals sample per distinct label set holding the summed values of +// everything else. +func stripSamples(p *profilev1.Profile, exc exceptions) { + var kept, folded []*profilev1.Sample + // Reused per sample to avoid allocating a keep-set for every sample. + totalKeep := make(map[string]struct{}) + funcKeep := make(map[string]struct{}) + for _, sample := range p.Sample { + // Samples that Normalize would drop (value length mismatch, + // negative values) must not contribute to the totals. + if len(sample.Value) != len(p.SampleType) || hasNegativeValue(sample) { + continue + } + // Labels the sample must keep for the total rules. These apply whether + // the sample is kept (it still counts toward those totals) or folded. + clear(totalKeep) + for _, rule := range exc.totalRules { + keepRuleLabels(totalKeep, rule, sampleFulfillsMatchers(sample, p.StringTable, rule.deferredMatchers)) + } + + // Trim to the locations kept by function rules the sample matches. Every + // rule targeting a kept location re-matches this sample's series at + // compaction, so all of them contribute the labels they need — including + // the ones the sample does not match (via keepRuleLabels). + clear(funcKeep) + var keptLocations int + for _, l := range sample.LocationId { + rules, ok := exc.functionLocations[l] + if !ok { + continue + } + var fulfilled bool + for _, rule := range rules { + matched := sampleFulfillsMatchers(sample, p.StringTable, rule.deferredMatchers) + keepRuleLabels(funcKeep, rule, matched) + if matched { + fulfilled = true + } + } + if fulfilled { + sample.LocationId[keptLocations] = l + keptLocations++ + } + } + if keptLocations > 0 { + sample.LocationId = sample.LocationId[:keptLocations] + sample.Label = retainLabels(sample.Label, p.StringTable, funcKeep, totalKeep) + kept = append(kept, sample) + continue + } + // Folds into a total: drop the stacktrace, keep only the total labels. + sample.LocationId = nil + sample.Label = retainLabels(sample.Label, p.StringTable, totalKeep) + folded = append(folded, sample) + } + p.Sample = append(kept, foldToTotals(folded, len(p.SampleType))...) +} + +// foldToTotals groups the label-reduced folded samples by their labels and sums +// each group into a single totals sample. Group labels are copied into a fresh +// slice: GroupSamplesByLabels aliases a source sample's label slice, whose +// backing array can hold labels past len() that the pprof split later restores +// and indexes against the rebuilt (smaller) string table. +func foldToTotals(folded []*profilev1.Sample, valueLen int) []*profilev1.Sample { + if len(folded) == 0 { + return nil + } + for _, s := range folded { + sort.Sort(pprof.LabelsByKeyValue(s.Label)) + } + sort.Sort(pprof.SamplesByLabels(folded)) + groups := pprof.GroupSamplesByLabels(&profilev1.Profile{Sample: folded}) + totals := make([]*profilev1.Sample, len(groups)) + for i, g := range groups { + labels := make([]*profilev1.Label, len(g.Labels)) + copy(labels, g.Labels) + total := &profilev1.Sample{Value: make([]int64, valueLen), Label: labels} + for _, s := range g.Samples { + for j, v := range s.Value { + total.Value[j] += v + } + } + totals[i] = total + } + return totals +} + +// keepRuleLabels records the labels a sample must retain for one rule. When the +// sample matches the rule, those are the rule's matcher + group_by labels. When +// it doesn't, only the matchers that also match "" are kept — otherwise dropping +// their label (which reads as "" downstream) would flip them into a spurious +// match. Shared by the function and total paths so they can't drift. +func keepRuleLabels(dst map[string]struct{}, rule deferredRule, matched bool) { + if matched { + for _, name := range rule.keepLabels { + dst[name] = struct{}{} + } + return + } + for _, m := range rule.deferredMatchers { + if m.Matches("") { + dst[m.Name] = struct{}{} + } + } +} + +func sampleFulfillsMatchers(s *profilev1.Sample, stringTable []string, matchers []*labels.Matcher) bool { + for _, m := range matchers { + if !m.Matches(sampleLabelValue(s, stringTable, m.Name)) { + return false + } + } + return true +} + +func sampleLabelValue(s *profilev1.Sample, stringTable []string, name string) string { + for _, l := range s.Label { + if l.Str > 0 && stringTable[l.Key] == name { + return stringTable[l.Str] + } + } + return "" +} + +func retainLabels(sampleLabels []*profilev1.Label, stringTable []string, keepSets ...map[string]struct{}) []*profilev1.Label { + retained := sampleLabels[:0] + for _, l := range sampleLabels { + name := stringTable[l.Key] + for _, keep := range keepSets { + if _, ok := keep[name]; ok { + retained = append(retained, l) + break + } + } + } + return retained[:len(retained):len(retained)] +} + +func pruneUnreferencedSymbols(p *profilev1.Profile) { + keepLoc := make(map[uint64]struct{}) + for _, s := range p.Sample { + for _, locID := range s.LocationId { + keepLoc[locID] = struct{}{} + } + } + if len(keepLoc) == 0 { + p.Location = nil + p.Function = nil + p.Mapping = nil + return + } + keepFunc := make(map[uint64]struct{}) + keepMapping := make(map[uint64]struct{}) + locations := p.Location[:0] + for _, loc := range p.Location { + if _, ok := keepLoc[loc.Id]; !ok { + continue + } + locations = append(locations, loc) + if loc.MappingId != 0 { + keepMapping[loc.MappingId] = struct{}{} + } + for _, line := range loc.Line { + keepFunc[line.FunctionId] = struct{}{} + } + } + p.Location = locations + functions := p.Function[:0] + for _, fn := range p.Function { + if _, ok := keepFunc[fn.Id]; ok { + functions = append(functions, fn) + } + } + p.Function = functions + mappings := p.Mapping[:0] + for _, m := range p.Mapping { + if _, ok := keepMapping[m.Id]; ok { + mappings = append(mappings, m) + } + } + p.Mapping = mappings +} + func hasNegativeValue(s *profilev1.Sample) bool { for _, v := range s.Value { if v < 0 { diff --git a/pkg/distributor/sampling/profile_stripper_test.go b/pkg/distributor/sampling/profile_stripper_test.go index 4cd010e8fe..d1a311b226 100644 --- a/pkg/distributor/sampling/profile_stripper_test.go +++ b/pkg/distributor/sampling/profile_stripper_test.go @@ -1,14 +1,33 @@ package sampling import ( + "errors" "testing" + "github.com/prometheus/prometheus/model/labels" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" profilev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1" + phlaremodel "github.com/grafana/pyroscope/v2/pkg/model" ) +const regionLabel = "region" + +type fakeRuler struct { + rules map[string][]*phlaremodel.RecordingRule +} + +func (r fakeRuler) RecordingRules(tenant string) ([]*phlaremodel.RecordingRule, error) { + return r.rules[tenant], nil +} + +type failingRuler struct{} + +func (failingRuler) RecordingRules(string) ([]*phlaremodel.RecordingRule, error) { + return nil, errors.New("rules unavailable") +} + // stripTestProfile returns a deterministic profile without sample labels, // so it maps to a single series. func stripTestProfile() *profilev1.Profile { @@ -70,9 +89,190 @@ func strippedTestProfile() *profilev1.Profile { } } -func TestProfileStripper_StripToTotals(t *testing.T) { +// sampleView is a human-readable view of a sample: its stacktrace as function +// names (leaf -> root) and its sample labels. +type sampleView struct { + stack []string + labels map[string]string + value int64 +} + +// buildProfile assembles a profile from sampleViews, interning function names +// (one location/function per name) and label strings. +func buildProfile(views []sampleView) *profilev1.Profile { + stringTable := []string{"", "samples", "count"} + intern := func(s string) int64 { + for i, existing := range stringTable { + if existing == s { + return int64(i) + } + } + stringTable = append(stringTable, s) + return int64(len(stringTable) - 1) + } + locByFunc := map[string]uint64{} + var functions []*profilev1.Function + var locations []*profilev1.Location + location := func(fn string) uint64 { + if id, ok := locByFunc[fn]; ok { + return id + } + id := uint64(len(locations) + 1) + functions = append(functions, &profilev1.Function{Id: id, Name: intern(fn)}) + locations = append(locations, &profilev1.Location{Id: id, Line: []*profilev1.Line{{FunctionId: id}}}) + locByFunc[fn] = id + return id + } + samples := make([]*profilev1.Sample, 0, len(views)) + for _, v := range views { + var locs []uint64 + for _, fn := range v.stack { + locs = append(locs, location(fn)) + } + var lbls []*profilev1.Label + for k, val := range v.labels { + lbls = append(lbls, &profilev1.Label{Key: intern(k), Str: intern(val)}) + } + samples = append(samples, &profilev1.Sample{LocationId: locs, Value: []int64{v.value}, Label: lbls}) + } + return &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: stringTable, + Function: functions, + Location: locations, + Sample: samples, + } +} + +// readProfile is the inverse of buildProfile: it resolves each sample back to +// function names and label values (nil stack for a stackless total). +func readProfile(p *profilev1.Profile) []sampleView { + funcName := map[uint64]string{} + for _, f := range p.Function { + funcName[f.Id] = p.StringTable[f.Name] + } + locFunc := map[uint64]string{} + for _, l := range p.Location { + locFunc[l.Id] = funcName[l.Line[0].FunctionId] + } + views := make([]sampleView, 0, len(p.Sample)) + for _, s := range p.Sample { + var stack []string + for _, id := range s.LocationId { + stack = append(stack, locFunc[id]) + } + labels := map[string]string{} + for _, l := range s.Label { + labels[p.StringTable[l.Key]] = p.StringTable[l.Str] + } + views = append(views, sampleView{stack: stack, labels: labels, value: s.Value[0]}) + } + return views +} + +// TestProfileStripper_Strip_Overview is a readable, end-to-end reference for Strip. It +// runs a whole profile of the series service_name="checkout" against a mix of +// function and total rules that together exercise every knob of the feature. +// +// Recording rules for the tenant (a rule is a "function rule" when it sets +// function_name, otherwise a "total rule"): +// +// FA function_name="db.query" matchers=[service_name="checkout"] group_by=[route] +// base matcher matches this series -> active. Keeps db.query stacks; a kept +// sample carries its route (the group_by label). +// FB function_name="encode" matchers=[format="proto"] +// format is a sample label, not a series label -> the matcher is DEFERRED to +// each sample. Keeps encode stacks only for samples with format="proto", +// which then carry format. +// FC function_name="db.query" matchers=[service_name="billing"] +// base matcher does NOT match this series -> the rule is dropped entirely and +// never affects the output. +// TA (total) group_by=[tenant] +// every surviving sample carries tenant; the leftovers split into per-tenant +// totals. +// TB (total) matchers=[region!="eu"] +// region is a sample label -> DEFERRED. Because != also matches a missing +// label (""), region is kept on every sample that has it (even region="eu"), +// so the totals split by region too. +// +// A kept sample keeps only the frames of the functions it hits, plus the labels +// the matching rules need. A folded sample drops its stack and keeps only the +// total-rule labels. Samples Normalize would reject (negative values) are dropped. +// +// INPUT +// # stack (leaf -> root) labels value +// I1 db.query -> handler tenant=a route=/x region=us 10 +// I2 db.query -> auth -> handler tenant=b route=/y region=eu 20 +// I3 encode -> handler tenant=a format=proto 5 +// I4 encode -> handler tenant=a format=json 8 +// I5 db.query -> encode -> handler tenant=b route=/z format=proto region=us 12 +// I6 db.query -> cache -> handler tenant=a route=/x region=us 7 +// I7 db.query -> parse -> db.query -> handler tenant=b route=/w region=eu 9 +// I8 gzip -> handler tenant=a region=us 4 +// I9 gzip -> handler tenant=a region=us 6 +// I10 read -> handler tenant=b region=eu 7 +// I11 read -> handler tenant=a 3 +// I12 read -> handler tenant=a region=us -1 +// +// OUTPUT. Strip's output is not minimal: it does not merge identical kept +// samples (I1 and I6 below) nor collapse repeated recursion frames (I7). +// Downstream Normalize does both. +// from stack labels value +// I1 -> [db.query] tenant=a route=/x region=us 10 <-+ same stack AND labels; +// I6 -> [db.query] tenant=a route=/x region=us 7 <-+ Normalize merges to 17 +// I2 -> [db.query] tenant=b route=/y region=eu 20 trimmed to db.query +// I3 -> [encode] tenant=a format=proto 5 kept: format=proto matches FB +// I5 -> [db.query,encode] tenant=b route=/z format=proto region=us 12 hits both FA and FB +// I7 -> [db.query,db.query] tenant=b route=/w region=eu 9 recursion: both frames kept +// I8+I9 -> (total) tenant=a region=us 10 folded and summed +// I10 -> (total) tenant=b region=eu 7 folded +// I4+I11 -> (total) tenant=a 11 I4 folds: format=json fails FB +// I12 -> dropped negative value +func TestProfileStripper_Strip_Overview(t *testing.T) { + in := []sampleView{ + {stack: []string{"db.query", "handler"}, labels: map[string]string{"tenant": "a", "route": "/x", "region": "us"}, value: 10}, + {stack: []string{"db.query", "auth", "handler"}, labels: map[string]string{"tenant": "b", "route": "/y", "region": "eu"}, value: 20}, + {stack: []string{"encode", "handler"}, labels: map[string]string{"tenant": "a", "format": "proto"}, value: 5}, + {stack: []string{"encode", "handler"}, labels: map[string]string{"tenant": "a", "format": "json"}, value: 8}, + {stack: []string{"db.query", "encode", "handler"}, labels: map[string]string{"tenant": "b", "route": "/z", "format": "proto", "region": "us"}, value: 12}, + {stack: []string{"db.query", "cache", "handler"}, labels: map[string]string{"tenant": "a", "route": "/x", "region": "us"}, value: 7}, + {stack: []string{"db.query", "parse", "db.query", "handler"}, labels: map[string]string{"tenant": "b", "route": "/w", "region": "eu"}, value: 9}, + {stack: []string{"gzip", "handler"}, labels: map[string]string{"tenant": "a", "region": "us"}, value: 4}, + {stack: []string{"gzip", "handler"}, labels: map[string]string{"tenant": "a", "region": "us"}, value: 6}, + {stack: []string{"read", "handler"}, labels: map[string]string{"tenant": "b", "region": "eu"}, value: 7}, + {stack: []string{"read", "handler"}, labels: map[string]string{"tenant": "a"}, value: 3}, + {stack: []string{"read", "handler"}, labels: map[string]string{"tenant": "a", "region": "us"}, value: -1}, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": { + {FunctionName: "db.query", Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, phlaremodel.LabelNameServiceName, "checkout")}, GroupBy: []string{"route"}}, + {FunctionName: "encode", Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "format", "proto")}}, + {FunctionName: "db.query", Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, phlaremodel.LabelNameServiceName, "billing")}}, + {GroupBy: []string{"tenant"}}, + {Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchNotEqual, regionLabel, "eu")}}, + }, + }} + + p := buildProfile(in) + NewProfileStripper(ruler).Strip("tenant-a", phlaremodel.LabelsFromStrings(phlaremodel.LabelNameServiceName, "checkout"), p) + + want := []sampleView{ + {stack: []string{"db.query"}, labels: map[string]string{"tenant": "a", "route": "/x", "region": "us"}, value: 10}, + {stack: []string{"db.query"}, labels: map[string]string{"tenant": "a", "route": "/x", "region": "us"}, value: 7}, + {stack: []string{"db.query"}, labels: map[string]string{"tenant": "b", "route": "/y", "region": "eu"}, value: 20}, + {stack: []string{"encode"}, labels: map[string]string{"tenant": "a", "format": "proto"}, value: 5}, + {stack: []string{"db.query", "encode"}, labels: map[string]string{"tenant": "b", "route": "/z", "format": "proto", "region": "us"}, value: 12}, + {stack: []string{"db.query", "db.query"}, labels: map[string]string{"tenant": "b", "route": "/w", "region": "eu"}, value: 9}, + {labels: map[string]string{"tenant": "a", "region": "us"}, value: 10}, + {labels: map[string]string{"tenant": "b", "region": "eu"}, value: 7}, + {labels: map[string]string{"tenant": "a"}, value: 11}, + } + assert.ElementsMatch(t, want, readProfile(p)) +} + +func TestProfileStripper_Strip(t *testing.T) { p := stripTestProfile() - NewProfileStripper().StripToTotals(p) + NewProfileStripper(nil).Strip("", nil, p) want := strippedTestProfile() require.Len(t, p.Sample, 1) @@ -88,26 +288,26 @@ func TestProfileStripper_StripToTotals(t *testing.T) { assert.Equal(t, want.SizeVT(), p.SizeVT()) } -func TestProfileStripper_StripToTotals_NoSamples(t *testing.T) { +func TestProfileStripper_Strip_NoSamples(t *testing.T) { p := stripTestProfile() p.Sample = nil - NewProfileStripper().StripToTotals(p) + NewProfileStripper(nil).Strip("", nil, p) assert.Empty(t, p.Sample) assert.Empty(t, p.Location) } -func TestProfileStripper_StripToTotals_AllSamplesInvalid(t *testing.T) { +func TestProfileStripper_Strip_AllSamplesInvalid(t *testing.T) { p := stripTestProfile() p.Sample = []*profilev1.Sample{ {LocationId: []uint64{1}, Value: []int64{-1, 100}}, {LocationId: []uint64{1}, Value: []int64{7}}, } - NewProfileStripper().StripToTotals(p) + NewProfileStripper(nil).Strip("", nil, p) assert.Empty(t, p.Sample) assert.Empty(t, p.Location) } -func TestProfileStripper_StripToTotals_SampleLabels(t *testing.T) { +func TestProfileStripper_Strip_SampleLabels(t *testing.T) { p := stripTestProfile() // Indices into the extended string table: span_id=9, abc=10, def=11. p.StringTable = append(p.StringTable, "span_id", "abc", "def") @@ -119,7 +319,7 @@ func TestProfileStripper_StripToTotals_SampleLabels(t *testing.T) { {LocationId: []uint64{1}, Value: []int64{16, 160}, Label: []*profilev1.Label{{Key: 9, Num: 42}}}, } - NewProfileStripper().StripToTotals(p) + NewProfileStripper(nil).Strip("", nil, p) require.Len(t, p.Sample, 1) assert.Equal(t, []int64{31, 310}, p.Sample[0].Value) @@ -130,3 +330,336 @@ func TestProfileStripper_StripToTotals_SampleLabels(t *testing.T) { assert.Empty(t, p.Mapping) assert.Equal(t, []string{"", "samples", "count", "cpu", "nanoseconds"}, p.StringTable) } + +// A rule targeting func-a keeps the stacktraces that reach it (trimmed to the +// func-a frame), folds the rest into the totals, and leaves only the referenced +// symbols, re-interned. +func TestProfileStripper_Strip_KeepsTargetedFunction(t *testing.T) { + p := stripTestProfile() + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{FunctionName: "func-a"}}, + }} + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + require.Len(t, p.Sample, 2) + // Kept targeted sample first, then the totals of the untargeted samples. + kept, total := p.Sample[0], p.Sample[1] + assert.Equal(t, []int64{10, 1000}, kept.Value) + assert.Equal(t, []uint64{1}, kept.LocationId) + assert.Empty(t, kept.Label) + assert.Equal(t, []int64{5, 500}, total.Value) + assert.Empty(t, total.LocationId) + + require.Len(t, p.Location, 1) + require.Len(t, p.Function, 1) + require.Len(t, p.Mapping, 1) + assert.Equal(t, "func-a", p.StringTable[p.Function[0].Name]) + assert.Equal(t, "app.py", p.StringTable[p.Function[0].Filename]) + assert.Equal(t, []string{"", "samples", "count", "cpu", "nanoseconds", "func-a", "app.py"}, p.StringTable) +} + +// Rules that don't target any function present in the profile leave the strip +// identical to a plain strip-to-totals. +func TestProfileStripper_Strip_RulesNoMatch(t *testing.T) { + p := stripTestProfile() + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{FunctionName: "not-present"}}, + }} + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + want := strippedTestProfile() + require.Len(t, p.Sample, 1) + assert.Equal(t, want.Sample[0].Value, p.Sample[0].Value) + assert.Empty(t, p.Location) + assert.Empty(t, p.Function) + assert.Empty(t, p.Mapping) + assert.Equal(t, want.StringTable, p.StringTable) +} + +// A matcher on a label absent from the series labels is deferred to the sample +// labels: only samples that satisfy it are kept, and only the labels the rule +// references (matchers + group_by) are retained. +func TestProfileStripper_Strip_DeferredSampleLabelMatcher(t *testing.T) { + // String table: func-a=3, region=4, eu=5, us=6, env=7, prod=8, extra=9, x=10. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", "func-a", regionLabel, "eu", "us", "env", "prod", "extra", "x"}, + Function: []*profilev1.Function{{Id: 1, Name: 3}}, + Location: []*profilev1.Location{{Id: 1, Line: []*profilev1.Line{{FunctionId: 1}}}}, + Sample: []*profilev1.Sample{ + {LocationId: []uint64{1}, Value: []int64{100}, Label: []*profilev1.Label{{Key: 4, Str: 5}, {Key: 7, Str: 8}, {Key: 9, Str: 10}}}, + {LocationId: []uint64{1}, Value: []int64{50}, Label: []*profilev1.Label{{Key: 4, Str: 6}}}, + {LocationId: []uint64{1}, Value: []int64{10}}, + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{ + FunctionName: "func-a", + GroupBy: []string{"env"}, + Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, regionLabel, "eu")}, + }}, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + require.Len(t, p.Sample, 2) + kept, total := p.Sample[0], p.Sample[1] + // region=eu is kept; region=us and the label-less sample fold into the total. + assert.Equal(t, []int64{100}, kept.Value) + require.Len(t, kept.LocationId, 1) + // region (matcher) and env (group_by) retained, extra dropped, and their + // string indices still resolve after the table was compacted. + require.Len(t, kept.Label, 2) + labelValues := map[string]string{} + for _, l := range kept.Label { + labelValues[p.StringTable[l.Key]] = p.StringTable[l.Str] + } + assert.Equal(t, map[string]string{regionLabel: "eu", "env": "prod"}, labelValues) + assert.Equal(t, []int64{60}, total.Value) + assert.Empty(t, total.LocationId) +} + +// A non-function ("total") rule that groups by a sample label splits the totals +// into one sample per distinct value of that label, each carrying it, so the +// rule can still match and group after stripping. +func TestProfileStripper_Strip_TotalsSplitByNonFunctionRuleLabel(t *testing.T) { + // String table: region=3, eu=4, us=5. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", regionLabel, "eu", "us"}, + Sample: []*profilev1.Sample{ + {Value: []int64{10}, Label: []*profilev1.Label{{Key: 3, Str: 4}}}, + {Value: []int64{20}, Label: []*profilev1.Label{{Key: 3, Str: 4}}}, + {Value: []int64{5}, Label: []*profilev1.Label{{Key: 3, Str: 5}}}, + {Value: []int64{3}}, + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{GroupBy: []string{regionLabel}}}, // no FunctionName -> total rule + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + // One total per region value (eu=30, us=5) plus the region-less samples (3), + // and the region label still resolves through the compacted string table. + require.Len(t, p.Sample, 3) + byRegion := map[string]int64{} + for _, s := range p.Sample { + assert.Empty(t, s.LocationId) + region := "" + for _, l := range s.Label { + if p.StringTable[l.Key] == regionLabel { + region = p.StringTable[l.Str] + } + } + byRegion[region] = s.Value[0] + } + assert.Equal(t, map[string]int64{"eu": 30, "us": 5, "": 3}, byRegion) + assert.Empty(t, p.Location) +} + +// A non-function rule with a sample-label matcher only affects the samples that +// satisfy it: those keep the label (so the rule matches them downstream); the +// rest fold together into a single label-less total, not split by the label. +func TestProfileStripper_Strip_TotalsMatcherGatesLabelKeep(t *testing.T) { + // String table: region=3, eu=4, us=5. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", regionLabel, "eu", "us"}, + Sample: []*profilev1.Sample{ + {Value: []int64{10}, Label: []*profilev1.Label{{Key: 3, Str: 4}}}, // region=eu + {Value: []int64{20}, Label: []*profilev1.Label{{Key: 3, Str: 4}}}, // region=eu + {Value: []int64{5}, Label: []*profilev1.Label{{Key: 3, Str: 5}}}, // region=us + {Value: []int64{3}}, // no region + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, regionLabel, "eu")}}}, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + // region=eu samples keep region and merge (30); region=us and region-less + // samples don't match, so they fold together into a label-less total (8). + require.Len(t, p.Sample, 2) + byRegion := map[string]int64{} + for _, s := range p.Sample { + region := "" + for _, l := range s.Label { + if p.StringTable[l.Key] == regionLabel { + region = p.StringTable[l.Str] + } + } + byRegion[region] = s.Value[0] + } + assert.Equal(t, map[string]int64{"eu": 30, "": 8}, byRegion) +} + +// A "!=" matcher also matches "", so a non-matching sample must keep its label +// (its real value) rather than fold into the label-less total — otherwise the +// dropped label would read as "" downstream and wrongly satisfy the matcher. +func TestProfileStripper_Strip_TotalsNotEqualMatcherKeepsLabel(t *testing.T) { + // String table: region=3, eu=4, us=5. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", regionLabel, "eu", "us"}, + Sample: []*profilev1.Sample{ + {Value: []int64{10}, Label: []*profilev1.Label{{Key: 3, Str: 4}}}, // region=eu (excluded by !=eu) + {Value: []int64{20}, Label: []*profilev1.Label{{Key: 3, Str: 5}}}, // region=us + {Value: []int64{5}}, // no region + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchNotEqual, regionLabel, "eu")}}}, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + // region=eu stays in its own total (region=eu) so the observer's region!=eu + // excludes it; it must NOT fold into the label-less total with the 5. + require.Len(t, p.Sample, 3) + byRegion := map[string]int64{} + for _, s := range p.Sample { + region := "" + for _, l := range s.Label { + if p.StringTable[l.Key] == regionLabel { + region = p.StringTable[l.Str] + } + } + byRegion[region] = s.Value[0] + } + assert.Equal(t, map[string]int64{"eu": 10, "us": 20, "": 5}, byRegion) +} + +// A sample kept for a function rule must also carry the labels of the total +// rules it matches, so it still counts toward their per-series totals. +func TestProfileStripper_Strip_KeptSampleCarriesTotalRuleLabels(t *testing.T) { + // String table: func-a=3, func-b=4, region=5, eu=6. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", "func-a", "func-b", regionLabel, "eu"}, + Function: []*profilev1.Function{{Id: 1, Name: 3}, {Id: 2, Name: 4}}, + Location: []*profilev1.Location{ + {Id: 1, Line: []*profilev1.Line{{FunctionId: 1}}}, // func-a + {Id: 2, Line: []*profilev1.Line{{FunctionId: 2}}}, // func-b + }, + Sample: []*profilev1.Sample{ + {LocationId: []uint64{1}, Value: []int64{10}, Label: []*profilev1.Label{{Key: 5, Str: 6}}}, // func-a, region=eu + {LocationId: []uint64{2}, Value: []int64{5}, Label: []*profilev1.Label{{Key: 5, Str: 6}}}, // func-b, region=eu + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": { + {FunctionName: "func-a"}, // function rule + {GroupBy: []string{regionLabel}}, // total rule + }, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + require.Len(t, p.Sample, 2) + regionOf := func(s *profilev1.Sample) string { + for _, l := range s.Label { + if p.StringTable[l.Key] == regionLabel { + return p.StringTable[l.Str] + } + } + return "" + } + // Kept func-a sample keeps its stack AND region, so it joins the region total. + kept, total := p.Sample[0], p.Sample[1] + require.NotEmpty(t, kept.LocationId) + assert.Equal(t, []int64{10}, kept.Value) + assert.Equal(t, "eu", regionOf(kept)) + // The func-b sample folds into the region=eu total. + assert.Empty(t, total.LocationId) + assert.Equal(t, []int64{5}, total.Value) + assert.Equal(t, "eu", regionOf(total)) +} + +// After stripping, no sample may keep live spare label capacity: retainLabels +// filters in place, and if the dropped labels lingered past len() the pprof +// split's restoreRemovedLabels (labels[len:cap]) would resurrect them with +// their pre-compaction string indices and index the rebuilt table out of range. +func TestProfileStripper_Strip_NoAliasedLabelCapacity(t *testing.T) { + // String table: func-a=3, func-b=4, env=5, prod=6, extra=7, x=8. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", "func-a", "func-b", "env", "prod", "extra", "x"}, + Function: []*profilev1.Function{{Id: 1, Name: 3}, {Id: 2, Name: 4}}, + Location: []*profilev1.Location{ + {Id: 1, Line: []*profilev1.Line{{FunctionId: 1}}}, // func-a + {Id: 2, Line: []*profilev1.Line{{FunctionId: 2}}}, // func-b + }, + Sample: []*profilev1.Sample{ + // func-a kept: keeps env (group_by), drops extra -> extra must not + // linger in the spare capacity. + {LocationId: []uint64{1}, Value: []int64{10}, Label: []*profilev1.Label{{Key: 5, Str: 6}, {Key: 7, Str: 8}}}, + // func-b folds into the label-less total; its dropped label likewise. + {LocationId: []uint64{2}, Value: []int64{5}, Label: []*profilev1.Label{{Key: 7, Str: 8}}}, + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": {{FunctionName: "func-a", GroupBy: []string{"env"}}}, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + require.NotEmpty(t, p.Sample) + for i, s := range p.Sample { + assert.Equalf(t, len(s.Label), cap(s.Label), "sample %d retains aliased spare label capacity", i) + } +} + +// Two rules target the same function; one has a "!=" matcher on a sample label. +// A sample kept for the first rule must retain that label (its real value) so +// the "!=" rule still excludes it downstream instead of matching a dropped "". +func TestProfileStripper_Strip_FunctionRuleNotEqualKeepsLabel(t *testing.T) { + // String table: func-a=3, region=4, us=5. + p := &profilev1.Profile{ + SampleType: []*profilev1.ValueType{{Type: 1, Unit: 2}}, + StringTable: []string{"", "samples", "count", "func-a", regionLabel, "us"}, + Function: []*profilev1.Function{{Id: 1, Name: 3}}, + Location: []*profilev1.Location{{Id: 1, Line: []*profilev1.Line{{FunctionId: 1}}}}, + Sample: []*profilev1.Sample{ + {LocationId: []uint64{1}, Value: []int64{10}, Label: []*profilev1.Label{{Key: 4, Str: 5}}}, // region=us + }, + } + ruler := fakeRuler{rules: map[string][]*phlaremodel.RecordingRule{ + "tenant-a": { + {FunctionName: "func-a"}, // keeps the stacktrace + {FunctionName: "func-a", Matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchNotEqual, regionLabel, "us")}}, + }, + }} + + NewProfileStripper(ruler).Strip("tenant-a", nil, p) + + require.Len(t, p.Sample, 1) + kept := p.Sample[0] + require.NotEmpty(t, kept.LocationId) + region := "" + for _, l := range kept.Label { + if p.StringTable[l.Key] == regionLabel { + region = p.StringTable[l.Str] + } + } + assert.Equal(t, "us", region) +} + +func TestProfileStripper_Strip_RulesUnavailable(t *testing.T) { + t.Run("ruler error strips to plain totals and reports it", func(t *testing.T) { + p := stripTestProfile() + unavailable := NewProfileStripper(failingRuler{}).Strip("tenant-a", nil, p) + + assert.True(t, unavailable) + require.Len(t, p.Sample, 1) + assert.Empty(t, p.Sample[0].LocationId) + assert.Equal(t, []int64{15, 1500}, p.Sample[0].Value) + }) + + t.Run("working and nil rulers are available", func(t *testing.T) { + rules := map[string][]*phlaremodel.RecordingRule{"tenant-a": {{FunctionName: "func-a"}}} + assert.False(t, NewProfileStripper(fakeRuler{rules}).Strip("tenant-a", nil, stripTestProfile())) + assert.False(t, NewProfileStripper(nil).Strip("tenant-a", nil, stripTestProfile())) + }) +} diff --git a/pkg/distributor/sampling/rules.go b/pkg/distributor/sampling/rules.go new file mode 100644 index 0000000000..90503a23fd --- /dev/null +++ b/pkg/distributor/sampling/rules.go @@ -0,0 +1,132 @@ +package sampling + +import ( + "github.com/prometheus/prometheus/model/labels" + + profilev1 "github.com/grafana/pyroscope/api/gen/proto/go/google/v1" + phlaremodel "github.com/grafana/pyroscope/v2/pkg/model" +) + +// Ruler provides a tenant's recording rules, used to decide which stacktraces +// of a sampled-out profile must be preserved during stripping. +type Ruler interface { + RecordingRules(tenant string) ([]*phlaremodel.RecordingRule, error) +} + +type deferredRule struct { + deferredMatchers []*labels.Matcher + keepLabels []string +} + +// exceptions captures what recording rules require to survive stripping. Each +// rule contributes a deferredRule whose deferredMatchers must be satisfied by a +// sample before its keepLabels apply: +// - functionLocations: per location id, the function-filter rules that keep a +// stacktrace reaching that location. +// - totalRules: the non-function ("total") rules; a sample they match keeps +// their labels so the totals can be split and matched/grouped downstream. +type exceptions struct { + functionLocations map[uint64][]deferredRule + totalRules []deferredRule + rulesUnavailable bool +} + +func (s *ProfileStripper) computeExceptions(tenantID string, seriesLabels phlaremodel.Labels, p *profilev1.Profile) exceptions { + if s.ruler == nil { + return exceptions{} + } + rules, err := s.ruler.RecordingRules(tenantID) + if err != nil { + return exceptions{rulesUnavailable: true} + } + profileTypes := profileTypeNames(p, seriesLabels.Get(phlaremodel.LabelNameProfileName)) + rulesByFunctionName := map[string][]deferredRule{} + var totalRules []deferredRule + for _, rule := range rules { + if rule == nil { + continue + } + matched, deferredMatchers := recordingRuleMatchesSeries(rule, seriesLabels, profileTypes) + if !matched { + continue + } + keepLabels := make([]string, 0, len(deferredMatchers)+len(rule.GroupBy)) + for _, m := range deferredMatchers { + keepLabels = append(keepLabels, m.Name) + } + keepLabels = append(keepLabels, rule.GroupBy...) + r := deferredRule{deferredMatchers: deferredMatchers, keepLabels: keepLabels} + + if rule.FunctionName == "" { + // Non-function rule: samples that match it keep its labels. + totalRules = append(totalRules, r) + continue + } + // Function rule: keep the stacktraces reaching FunctionName. + rulesByFunctionName[rule.FunctionName] = append(rulesByFunctionName[rule.FunctionName], r) + } + + functionLocations := map[uint64][]deferredRule{} + if len(rulesByFunctionName) > 0 { + rulesByFunctionID := make(map[uint64][]deferredRule, len(p.Function)) + for _, fn := range p.Function { + if rules, ok := rulesByFunctionName[p.StringTable[fn.Name]]; ok { + rulesByFunctionID[fn.Id] = rules + } + } + for _, loc := range p.Location { + var rules []deferredRule + for _, line := range loc.Line { + rules = append(rules, rulesByFunctionID[line.FunctionId]...) + } + if len(rules) > 0 { + functionLocations[loc.Id] = rules + } + } + } + + return exceptions{ + functionLocations: functionLocations, + totalRules: totalRules, + } +} + +func recordingRuleMatchesSeries(rule *phlaremodel.RecordingRule, seriesLabels phlaremodel.Labels, profileTypes []string) (bool, []*labels.Matcher) { + var deferredMatchers []*labels.Matcher + for _, m := range rule.Matchers { + if m.Name == phlaremodel.LabelNameProfileType { + var matched bool + for _, pt := range profileTypes { + if m.Matches(pt) { + matched = true + break + } + } + if !matched { + return false, nil + } + continue + } + if v, ok := seriesLabels.GetLabel(m.Name); ok { + if !m.Matches(v.Value) { + return false, nil + } + continue + } + deferredMatchers = append(deferredMatchers, m) + } + return true, deferredMatchers +} + +func profileTypeNames(p *profilev1.Profile, metricName string) []string { + var periodType, periodUnit string + if p.PeriodType != nil { + periodType = p.StringTable[p.PeriodType.Type] + periodUnit = p.StringTable[p.PeriodType.Unit] + } + types := make([]string, 0, len(p.SampleType)) + for _, st := range p.SampleType { + types = append(types, metricName+":"+p.StringTable[st.Type]+":"+p.StringTable[st.Unit]+":"+periodType+":"+periodUnit) + } + return types +} diff --git a/pkg/metrics/config.go b/pkg/metrics/config.go index 828576d85a..605746bcc4 100644 --- a/pkg/metrics/config.go +++ b/pkg/metrics/config.go @@ -3,6 +3,9 @@ package metrics import ( "errors" "flag" + + "github.com/go-kit/log" + "github.com/go-kit/log/level" ) type Config struct { @@ -26,7 +29,43 @@ func (c *Config) Validate() error { func (c *Config) RegisterFlags(f *flag.FlagSet) { const prefix = "compaction-worker.metrics-exporter." - f.BoolVar(&c.Enabled, prefix+"enabled", false, "This parameter specifies whether the metrics exporter is enabled.") - f.StringVar(&c.RulesSource.ClientAddress, prefix+"rules-source.client-address", "", "The address to use for the recording rules client connection.") + f.BoolVar(&c.Enabled, prefix+"enabled", false, "Deprecated: use recording-rules.enabled. Specifies whether the metrics exporter is enabled.") + f.StringVar(&c.RulesSource.ClientAddress, prefix+"rules-source.client-address", "", "Deprecated: use recording-rules.client-address. The address to use for the recording rules client connection.") f.StringVar(&c.RemoteWriteAddress, prefix+"remote-write-address", "", "The address to use for metrics tenant.") } + +// RecordingRulesConfig is the shared configuration for generating metrics from +// profiles via recording rules. It is read by the query-frontend (feature +// flag), the distributor (rule-aware stripping of sampled-out profiles) and the +// compaction-worker (metrics generation), so they agree on whether the feature +// is enabled and where the rules come from. +type RecordingRulesConfig struct { + Enabled bool `yaml:"enabled"` + ClientAddress string `yaml:"client_address" category:"advanced"` +} + +func (c *RecordingRulesConfig) RegisterFlags(f *flag.FlagSet) { + f.BoolVar(&c.Enabled, "recording-rules.enabled", false, + "Enable generating metrics from profiles using recording rules. Read by the query-frontend, distributor and compaction-worker.") + f.StringVar(&c.ClientAddress, "recording-rules.client-address", "", + "Address of the recording rules source (the tenant-settings service). When empty, static rules from per-tenant overrides are used. Read by the distributor and compaction-worker.") +} + +// ReconcileWithExporter merges the deprecated compaction-worker metrics-exporter +// settings into the shared recording-rules config. The shared values take +// precedence; the deprecated ones are honored as a fallback. Effective values +// are written back to both, so components reading either stay consistent. +func (c *RecordingRulesConfig) ReconcileWithExporter(exporter *Config, logger log.Logger) { + if exporter.Enabled || exporter.RulesSource.ClientAddress != "" { + level.Warn(logger).Log("msg", "compaction-worker.metrics-exporter.enabled and compaction-worker.metrics-exporter.rules-source.client-address are deprecated, use recording-rules.enabled and recording-rules.client-address") + } + if exporter.Enabled { + c.Enabled = true + } + exporter.Enabled = c.Enabled + + if c.ClientAddress == "" { + c.ClientAddress = exporter.RulesSource.ClientAddress + } + exporter.RulesSource.ClientAddress = c.ClientAddress +} diff --git a/pkg/metrics/observer.go b/pkg/metrics/observer.go index a133f4db96..00a213bb17 100644 --- a/pkg/metrics/observer.go +++ b/pkg/metrics/observer.go @@ -56,7 +56,7 @@ type Ruler interface { // with `profiles_recorded_` // - a "profiles_rule_id" label is present among ExternalLabels. It identifies the rule. // - a matcher with name "__profile__type__" is present in Matchers - RecordingRules(tenant string) []*phlaremodel.RecordingRule + RecordingRules(tenant string) ([]*phlaremodel.RecordingRule, error) } type Exporter interface { @@ -82,7 +82,7 @@ func NewSampleObserver(recordingTime int64, exporter Exporter, ruler Ruler, labe func (o *SampleObserver) initTenantState(tenant string) { o.state.tenant = tenant - recordingRules := o.ruler.RecordingRules(tenant) + recordingRules, _ := o.ruler.RecordingRules(tenant) for _, rule := range recordingRules { o.state.recordings = append(o.state.recordings, &recording{ diff --git a/pkg/metrics/observer_test.go b/pkg/metrics/observer_test.go index f59669ae07..98af0accdc 100644 --- a/pkg/metrics/observer_test.go +++ b/pkg/metrics/observer_test.go @@ -42,7 +42,7 @@ func Test_Observer_observe(t *testing.T) { }, GroupBy: []string{}, }, - }) + }, nil) observer := NewSampleObserver(blockTime, exporter, ruler, labels.New(labels.Label{Name: "external2", Value: "external2"})) entries := entriesOf([][]any{ {"tenant1", [][]string{{"a", "0"}, {"b", "0"}, {"c", "0"}, {"d", "0"}}, int64(1) << 0}, diff --git a/pkg/metrics/recording_rules_test.go b/pkg/metrics/recording_rules_test.go new file mode 100644 index 0000000000..53d0114b77 --- /dev/null +++ b/pkg/metrics/recording_rules_test.go @@ -0,0 +1,77 @@ +package metrics + +import ( + "bytes" + "testing" + + "github.com/go-kit/log" + "github.com/stretchr/testify/assert" +) + +func TestRecordingRulesConfig_ReconcileWithExporter(t *testing.T) { + const addr = "settings.svc:4040" + const legacyAddr = "legacy.svc:4040" + + for _, tc := range []struct { + name string + shared RecordingRulesConfig + legacyOn bool + legacyAddr string + wantEnabled bool + wantAddress string + }{ + { + name: "unset -> disabled, static", + wantEnabled: false, + wantAddress: "", + }, + { + name: "shared flags only", + shared: RecordingRulesConfig{Enabled: true, ClientAddress: addr}, + wantEnabled: true, + wantAddress: addr, + }, + { + name: "deprecated flags only are honored", + legacyOn: true, + legacyAddr: legacyAddr, + wantEnabled: true, + wantAddress: legacyAddr, + }, + { + name: "shared address wins over deprecated", + shared: RecordingRulesConfig{Enabled: true, ClientAddress: addr}, + legacyOn: true, + legacyAddr: legacyAddr, + wantEnabled: true, + wantAddress: addr, + }, + { + name: "shared enabled, static (no address)", + shared: RecordingRulesConfig{Enabled: true}, + wantEnabled: true, + wantAddress: "", + }, + } { + t.Run(tc.name, func(t *testing.T) { + exporter := Config{Enabled: tc.legacyOn} + exporter.RulesSource.ClientAddress = tc.legacyAddr + shared := tc.shared + + var logs bytes.Buffer + shared.ReconcileWithExporter(&exporter, log.NewLogfmtLogger(&logs)) + + if tc.legacyOn || tc.legacyAddr != "" { + assert.Contains(t, logs.String(), "deprecated") + } else { + assert.Empty(t, logs.String()) + } + + // Shared and deprecated fields both hold the effective values. + assert.Equal(t, tc.wantEnabled, shared.Enabled) + assert.Equal(t, tc.wantEnabled, exporter.Enabled) + assert.Equal(t, tc.wantAddress, shared.ClientAddress) + assert.Equal(t, tc.wantAddress, exporter.RulesSource.ClientAddress) + }) + } +} diff --git a/pkg/metrics/ruler.go b/pkg/metrics/ruler.go index d001ada37a..e17c3eb1a8 100644 --- a/pkg/metrics/ruler.go +++ b/pkg/metrics/ruler.go @@ -13,6 +13,7 @@ import ( const ( rulesExpiryTime = time.Minute + rulesRetryTime = 10 * time.Second ) type StaticRuler struct { @@ -25,7 +26,7 @@ func NewStaticRulerFromOverrides(overrides *validation.Overrides) Ruler { } } -func (ruler StaticRuler) RecordingRules(tenant string) []*model.RecordingRule { +func (ruler StaticRuler) RecordingRules(tenant string) ([]*model.RecordingRule, error) { rules := ruler.overrides.RecordingRules(tenant) rs := make([]*model.RecordingRule, 0, len(rules)) for _, rule := range rules { @@ -33,7 +34,7 @@ func (ruler StaticRuler) RecordingRules(tenant string) []*model.RecordingRule { r, _ := model.NewRecordingRule(rule) rs = append(rs, r) } - return rs + return rs, nil } // CachedRemoteRuler is a thread-safe ruler that retrieves rules from an external service. @@ -44,6 +45,9 @@ type CachedRemoteRuler struct { client RecordingRulesClient + expiry time.Duration + retry time.Duration + logger log.Logger } @@ -55,11 +59,13 @@ func NewCachedRemoteRuler(client RecordingRulesClient, logger log.Logger) (Ruler return &CachedRemoteRuler{ rulesPerTenant: make(map[string]*tenantCache), client: client, + expiry: rulesExpiryTime, + retry: rulesRetryTime, logger: logger, }, nil } -func (r *CachedRemoteRuler) RecordingRules(tenant string) []*model.RecordingRule { +func (r *CachedRemoteRuler) RecordingRules(tenant string) ([]*model.RecordingRule, error) { // get the per-tenant cache r.mu.RLock() cache, ok := r.rulesPerTenant[tenant] @@ -77,6 +83,8 @@ func (r *CachedRemoteRuler) RecordingRules(tenant string) []*model.RecordingRule initFunc: func() ([]*model.RecordingRule, error) { return r.client.RecordingRules(tenant) }, + expiry: r.expiry, + retry: r.retry, logger: r.logger, } r.rulesPerTenant[tenant] = cache @@ -89,38 +97,70 @@ func (r *CachedRemoteRuler) RecordingRules(tenant string) []*model.RecordingRule // tenantCache is a thread-safe cache that holds an expirable array of rules. type tenantCache struct { - value []*model.RecordingRule - ttl time.Time - initFunc func() ([]*model.RecordingRule, error) - mu sync.RWMutex - logger log.Logger + value []*model.RecordingRule + loaded bool + err error + refreshing bool + ttl time.Time + expiry time.Duration + retry time.Duration + initFunc func() ([]*model.RecordingRule, error) + mu sync.RWMutex + logger log.Logger } -// get returns the stored value if present and not expired. -// Otherwise, a single call to initFunc will be performed to retrieve the value and hold it for future calls within -// the ttl. -func (c *tenantCache) get() []*model.RecordingRule { +// get returns the cached value while it is fresh. Once it expires, the stale +// value keeps being served while a single background refresh runs. Only the +// very first load blocks callers, and it is bounded by the client's timeout. +// A failed fetch is not retried until the retry interval has passed. +func (c *tenantCache) get() ([]*model.RecordingRule, error) { c.mu.RLock() - if c.value != nil && time.Now().Before(c.ttl) { + if time.Now().Before(c.ttl) { defer c.mu.RUnlock() - // value exists and didn't expired - return c.value + return c.result() } c.mu.RUnlock() c.mu.Lock() defer c.mu.Unlock() - // only race-winner will fetch the data - if c.value == nil || time.Now().After(c.ttl) { - value, err := c.initFunc() - if err != nil { - // keep old value and ttl, just log an error - level.Error(c.logger).Log("msg", "failed to fetch recording rules", "err", err) - } else { - c.value = value - c.ttl = time.Now().Add(rulesExpiryTime) - } + if time.Now().Before(c.ttl) { + return c.result() + } + if !c.loaded { + c.apply(c.initFunc()) + return c.result() + } + if !c.refreshing { + c.refreshing = true + go c.refresh() + } + return c.result() +} + +func (c *tenantCache) result() ([]*model.RecordingRule, error) { + if !c.loaded { + return nil, c.err + } + return c.value, nil +} + +func (c *tenantCache) refresh() { + value, err := c.initFunc() + c.mu.Lock() + defer c.mu.Unlock() + c.refreshing = false + c.apply(value, err) +} + +func (c *tenantCache) apply(value []*model.RecordingRule, err error) { + if err != nil { + level.Error(c.logger).Log("msg", "failed to fetch recording rules", "err", err) + c.err = err + c.ttl = time.Now().Add(c.retry) + return } - return c.value + c.value = value + c.loaded = true + c.ttl = time.Now().Add(c.expiry) } diff --git a/pkg/metrics/ruler_cache_test.go b/pkg/metrics/ruler_cache_test.go new file mode 100644 index 0000000000..cda14b9971 --- /dev/null +++ b/pkg/metrics/ruler_cache_test.go @@ -0,0 +1,147 @@ +package metrics + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-kit/log" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/grafana/pyroscope/v2/pkg/model" +) + +type fakeRulesClient struct { + calls atomic.Int32 + fn func(call int32) ([]*model.RecordingRule, error) +} + +func (f *fakeRulesClient) RecordingRules(string) ([]*model.RecordingRule, error) { + return f.fn(f.calls.Add(1)) +} + +func newTestRuler(c RecordingRulesClient, expiry, retry time.Duration) *CachedRemoteRuler { + r, _ := NewCachedRemoteRuler(c, log.NewNopLogger()) + cr := r.(*CachedRemoteRuler) + cr.expiry = expiry + cr.retry = retry + return cr +} + +func TestCachedRemoteRuler_FirstLoadIsSingleFlight(t *testing.T) { + release := make(chan struct{}) + client := &fakeRulesClient{fn: func(int32) ([]*model.RecordingRule, error) { + <-release + return defaultRecordingRules, nil + }} + ruler := newTestRuler(client, time.Hour, time.Hour) + + var wg sync.WaitGroup + results := make([][]*model.RecordingRule, 8) + for i := range results { + wg.Add(1) + go func() { + defer wg.Done() + results[i] = rulesOf(t, ruler) + }() + } + time.Sleep(50 * time.Millisecond) + close(release) + wg.Wait() + + assert.EqualValues(t, 1, client.calls.Load()) + for _, r := range results { + assert.Equal(t, defaultRecordingRules, r) + } +} + +func TestCachedRemoteRuler_ServesStaleWhileRefreshing(t *testing.T) { + release := make(chan struct{}) + client := &fakeRulesClient{fn: func(call int32) ([]*model.RecordingRule, error) { + if call == 1 { + return defaultRecordingRules, nil + } + <-release + return overriddenRecordingRules, nil + }} + ruler := newTestRuler(client, 10*time.Millisecond, time.Hour) + + require.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) + time.Sleep(20 * time.Millisecond) + + for range 5 { + assert.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) + } + require.Eventually(t, func() bool { return client.calls.Load() == 2 }, time.Second, 5*time.Millisecond) + + close(release) + require.Eventually(t, func() bool { + rules, err := ruler.RecordingRules("t") + return err == nil && assert.ObjectsAreEqual(overriddenRecordingRules, rules) + }, time.Second, 5*time.Millisecond) + assert.EqualValues(t, 2, client.calls.Load()) +} + +func TestCachedRemoteRuler_FailureBacksOff(t *testing.T) { + client := &fakeRulesClient{fn: func(int32) ([]*model.RecordingRule, error) { + return nil, errors.New("unavailable") + }} + ruler := newTestRuler(client, time.Hour, time.Hour) + + for range 5 { + rules, err := ruler.RecordingRules("t") + assert.Nil(t, rules) + assert.Error(t, err) + } + assert.EqualValues(t, 1, client.calls.Load()) +} + +func TestCachedRemoteRuler_RefreshFailureKeepsStaleValue(t *testing.T) { + client := &fakeRulesClient{fn: func(call int32) ([]*model.RecordingRule, error) { + if call == 1 { + return defaultRecordingRules, nil + } + return nil, errors.New("unavailable") + }} + ruler := newTestRuler(client, 10*time.Millisecond, time.Hour) + + require.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) + time.Sleep(20 * time.Millisecond) + + assert.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) + require.Eventually(t, func() bool { return client.calls.Load() == 2 }, time.Second, 5*time.Millisecond) + time.Sleep(20 * time.Millisecond) + for range 5 { + assert.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) + } + assert.EqualValues(t, 2, client.calls.Load()) +} + +func TestCachedRemoteRuler_RecoversAfterFailure(t *testing.T) { + fail := atomic.Bool{} + fail.Store(true) + client := &fakeRulesClient{fn: func(int32) ([]*model.RecordingRule, error) { + if fail.Load() { + return nil, errors.New("unavailable") + } + return defaultRecordingRules, nil + }} + ruler := newTestRuler(client, time.Hour, 10*time.Millisecond) + + _, err := ruler.RecordingRules("t") + assert.Error(t, err) + + fail.Store(false) + time.Sleep(20 * time.Millisecond) + assert.Equal(t, defaultRecordingRules, rulesOf(t, ruler)) +} + +func rulesOf(t *testing.T, r Ruler) []*model.RecordingRule { + t.Helper() + rules, err := r.RecordingRules("t") + require.NoError(t, err) + return rules +} diff --git a/pkg/metrics/ruler_test.go b/pkg/metrics/ruler_test.go index 48aaf88f24..67fa1b689c 100644 --- a/pkg/metrics/ruler_test.go +++ b/pkg/metrics/ruler_test.go @@ -61,10 +61,12 @@ func Test_Ruler_happyPath(t *testing.T) { ruler := NewStaticRulerFromOverrides(overrides) - rules := ruler.RecordingRules("non-configured-tenant") + rules, err := ruler.RecordingRules("non-configured-tenant") + assert.NoError(t, err) assert.Equal(t, defaultRecordingRules, rules) - rules = ruler.RecordingRules("tenant-override") + rules, err = ruler.RecordingRules("tenant-override") + assert.NoError(t, err) assert.Equal(t, overriddenRecordingRules, rules) } diff --git a/pkg/pyroscope/feature_flags.go b/pkg/pyroscope/feature_flags.go index ac7cfa9313..c1309e087b 100644 --- a/pkg/pyroscope/feature_flags.go +++ b/pkg/pyroscope/feature_flags.go @@ -7,7 +7,7 @@ import ( ) func (c *Config) getFeatureFlags() map[string]bool { - rulerEnabled := c.CompactionWorker.MetricsExporter.Enabled + rulerEnabled := c.RecordingRules.Enabled return map[string]bool{ featureflags.V2StorageLayer: c.ArchitectureStorage == V1V2Dual || c.ArchitectureStorage == V2, featureflags.PyroscopeRuler: rulerEnabled, diff --git a/pkg/pyroscope/modules.go b/pkg/pyroscope/modules.go index d907c19943..eed5e3b657 100644 --- a/pkg/pyroscope/modules.go +++ b/pkg/pyroscope/modules.go @@ -43,6 +43,7 @@ import ( "github.com/grafana/pyroscope/v2/pkg/embedded/grafana" "github.com/grafana/pyroscope/v2/pkg/featureflags" "github.com/grafana/pyroscope/v2/pkg/ingester" + "github.com/grafana/pyroscope/v2/pkg/metrics" objstoreclient "github.com/grafana/pyroscope/v2/pkg/objstore/client" "github.com/grafana/pyroscope/v2/pkg/objstore/providers/filesystem" "github.com/grafana/pyroscope/v2/pkg/operations" @@ -330,7 +331,21 @@ func (f *Pyroscope) initDistributor() (services.Service, error) { if f.segmentWriterClient != nil { swClient = f.segmentWriterClient } - d, err := distributor.New(f.Cfg.Distributor, f.ingesterRing, nil, f.Overrides, f.reg, logger, swClient, f.auth) + + var ruler metrics.Ruler + if f.Cfg.RecordingRules.Enabled { + if f.recordingRulesClient != nil { + r, err := metrics.NewCachedRemoteRuler(f.recordingRulesClient, f.logger) + if err != nil { + return nil, err + } + ruler = r + } else { + ruler = metrics.NewStaticRulerFromOverrides(f.Overrides) + } + } + + d, err := distributor.New(f.Cfg.Distributor, f.ingesterRing, nil, f.Overrides, f.reg, logger, swClient, ruler, f.auth) if err != nil { return nil, err } diff --git a/pkg/pyroscope/modules_experimental.go b/pkg/pyroscope/modules_experimental.go index 78efd5c6d1..ff596deac5 100644 --- a/pkg/pyroscope/modules_experimental.go +++ b/pkg/pyroscope/modules_experimental.go @@ -292,7 +292,11 @@ func (f *Pyroscope) initCompactionWorker() (svc services.Service, err error) { var ruler metrics.Ruler var exporter metrics.Exporter - if f.Cfg.CompactionWorker.MetricsExporter.Enabled { + if f.Cfg.RecordingRules.Enabled { + // remote-write-address is mandatory when the exporter runs. + if err = f.Cfg.CompactionWorker.MetricsExporter.Validate(); err != nil { + return nil, err + } if f.recordingRulesClient != nil { ruler, err = metrics.NewCachedRemoteRuler(f.recordingRulesClient, f.logger) if err != nil { @@ -427,14 +431,10 @@ func (f *Pyroscope) initQueryBackendClient() (services.Service, error) { } func (f *Pyroscope) initRecordingRulesClient() (services.Service, error) { - if err := f.Cfg.CompactionWorker.MetricsExporter.Validate(); err != nil { - return nil, err - } - if !f.Cfg.CompactionWorker.MetricsExporter.Enabled || - f.Cfg.CompactionWorker.MetricsExporter.RulesSource.ClientAddress == "" { + if !f.Cfg.RecordingRules.Enabled || f.Cfg.RecordingRules.ClientAddress == "" { return nil, nil } - c, err := recordingrulesclient.NewClient(f.Cfg.CompactionWorker.MetricsExporter.RulesSource.ClientAddress, f.logger, f.auth) + c, err := recordingrulesclient.NewClient(f.Cfg.RecordingRules.ClientAddress, f.logger, f.auth) if err != nil { return nil, err } diff --git a/pkg/pyroscope/pyroscope.go b/pkg/pyroscope/pyroscope.go index 897f482df3..3e4f613d87 100644 --- a/pkg/pyroscope/pyroscope.go +++ b/pkg/pyroscope/pyroscope.go @@ -55,6 +55,7 @@ import ( "github.com/grafana/pyroscope/v2/pkg/metastore" metastoreadmin "github.com/grafana/pyroscope/v2/pkg/metastore/admin" metastoreclient "github.com/grafana/pyroscope/v2/pkg/metastore/client" + "github.com/grafana/pyroscope/v2/pkg/metrics" phlareobj "github.com/grafana/pyroscope/v2/pkg/objstore" objstoreclient "github.com/grafana/pyroscope/v2/pkg/objstore/client" "github.com/grafana/pyroscope/v2/pkg/operations/v2/querydiagnostics" @@ -83,25 +84,26 @@ import ( ) type Config struct { - Target flagext.StringSliceCSV `yaml:"target,omitempty"` - API api.Config `yaml:"api"` - Server server.Config `yaml:"server,omitempty"` - Metastore metastore.Config `yaml:"metastore"` - Distributor distributor.Config `yaml:"distributor,omitempty"` - Frontend frontend.Config `yaml:"frontend,omitempty"` - QueryBackend querybackend.Config `yaml:"query_backend"` - AdaptivePlacement placement.Config `yaml:"adaptive_placement"` - Symbolizer symbolizer.Config `yaml:"symbolizer"` - Worker worker.Config `yaml:"frontend_worker"` - LimitsConfig validation.Limits `yaml:"limits"` - SegmentWriter segmentwriter.Config `yaml:"segment_writer"` - MemberlistKV memberlist.KVConfig `yaml:"memberlist"` - PhlareDB phlaredb.Config `yaml:"pyroscopedb,omitempty"` - Tracing tracing.Config `yaml:"tracing"` - OverridesExporter exporter.Config `yaml:"overrides_exporter"` - RuntimeConfig runtimeconfig.Config `yaml:"runtime_config"` - CompactionWorker compactionworker.Config `yaml:"compaction_worker"` - TenantSettings settings.Config `yaml:"tenant_settings"` + Target flagext.StringSliceCSV `yaml:"target,omitempty"` + API api.Config `yaml:"api"` + Server server.Config `yaml:"server,omitempty"` + Metastore metastore.Config `yaml:"metastore"` + Distributor distributor.Config `yaml:"distributor,omitempty"` + Frontend frontend.Config `yaml:"frontend,omitempty"` + QueryBackend querybackend.Config `yaml:"query_backend"` + AdaptivePlacement placement.Config `yaml:"adaptive_placement"` + Symbolizer symbolizer.Config `yaml:"symbolizer"` + Worker worker.Config `yaml:"frontend_worker"` + LimitsConfig validation.Limits `yaml:"limits"` + SegmentWriter segmentwriter.Config `yaml:"segment_writer"` + MemberlistKV memberlist.KVConfig `yaml:"memberlist"` + PhlareDB phlaredb.Config `yaml:"pyroscopedb,omitempty"` + Tracing tracing.Config `yaml:"tracing"` + OverridesExporter exporter.Config `yaml:"overrides_exporter"` + RuntimeConfig runtimeconfig.Config `yaml:"runtime_config"` + CompactionWorker compactionworker.Config `yaml:"compaction_worker"` + TenantSettings settings.Config `yaml:"tenant_settings"` + RecordingRules metrics.RecordingRulesConfig `yaml:"recording_rules"` Storage StorageConfig `yaml:"storage"` SelfProfiling SelfProfilingConfig `yaml:"self_profiling,omitempty"` @@ -267,6 +269,7 @@ func (c *Config) RegisterFlagsWithContext(f *flag.FlagSet) { c.AdminServer.RegisterFlags(f) c.EmbeddedGrafana.RegisterFlags(f) c.TenantSettings.RegisterFlags(f) + c.RecordingRules.RegisterFlags(f) // Legacy flags c.StoreGateway.RegisterFlags(f, util.Logger) @@ -402,6 +405,8 @@ func (c *Config) Validate() error { return errors.New("no modules specified") } + c.RecordingRules.ReconcileWithExporter(&c.CompactionWorker.MetricsExporter, util.Logger) + if err := c.Frontend.Validate(); err != nil { return err } @@ -657,7 +662,7 @@ func (f *Pyroscope) setupModuleManager() error { API: {Server, AdminServer}, Metastore: {Overrides, API, MetastoreClient, Storage, PlacementManager}, MetastoreAdmin: {API, MetastoreClient}, - Distributor: {Overrides, SegmentWriterClient, API, UsageReport, Storage, IngesterRing}, + Distributor: {Overrides, SegmentWriterClient, API, UsageReport, Storage, IngesterRing, RecordingRulesClient}, PlacementAgent: {Overrides, API, Storage}, PlacementManager: {Overrides, API, Storage}, SegmentWriter: {Overrides, API, MemberlistKV, Storage, UsageReport, MetastoreClient}, diff --git a/pkg/settings/recording/client/client.go b/pkg/settings/recording/client/client.go index 61e470f931..cad5d3e921 100644 --- a/pkg/settings/recording/client/client.go +++ b/pkg/settings/recording/client/client.go @@ -2,6 +2,7 @@ package recording import ( "context" + "time" "connectrpc.com/connect" "github.com/go-kit/log" @@ -22,6 +23,8 @@ type Client struct { logger log.Logger } +const requestTimeout = 5 * time.Second + func NewClient(address string, logger log.Logger, auth connect.Option) (*Client, error) { httpClient := httputil.InstrumentedDefaultHTTPClient() opts := connectapi.DefaultClientOptions() @@ -35,7 +38,8 @@ func NewClient(address string, logger log.Logger, auth connect.Option) (*Client, } func (b *Client) RecordingRules(tenantId string) ([]*phlaremodel.RecordingRule, error) { - ctx := tenant.InjectTenantID(context.Background(), tenantId) + ctx, cancel := context.WithTimeout(tenant.InjectTenantID(context.Background(), tenantId), requestTimeout) + defer cancel() resp, err := b.client.ListRecordingRules(ctx, connect.NewRequest(&settingsv1.ListRecordingRulesRequest{})) if err != nil { return nil, err diff --git a/pkg/test/mocks/mockmetrics/mock_ruler.go b/pkg/test/mocks/mockmetrics/mock_ruler.go index 9ed75a77a3..7c1a91a1cd 100644 --- a/pkg/test/mocks/mockmetrics/mock_ruler.go +++ b/pkg/test/mocks/mockmetrics/mock_ruler.go @@ -21,7 +21,7 @@ func (_m *MockRuler) EXPECT() *MockRuler_Expecter { } // RecordingRules provides a mock function with given fields: tenant -func (_m *MockRuler) RecordingRules(tenant string) []*model.RecordingRule { +func (_m *MockRuler) RecordingRules(tenant string) ([]*model.RecordingRule, error) { ret := _m.Called(tenant) if len(ret) == 0 { @@ -29,6 +29,10 @@ func (_m *MockRuler) RecordingRules(tenant string) []*model.RecordingRule { } var r0 []*model.RecordingRule + var r1 error + if rf, ok := ret.Get(0).(func(string) ([]*model.RecordingRule, error)); ok { + return rf(tenant) + } if rf, ok := ret.Get(0).(func(string) []*model.RecordingRule); ok { r0 = rf(tenant) } else { @@ -37,7 +41,13 @@ func (_m *MockRuler) RecordingRules(tenant string) []*model.RecordingRule { } } - return r0 + if rf, ok := ret.Get(1).(func(string) error); ok { + r1 = rf(tenant) + } else { + r1 = ret.Error(1) + } + + return r0, r1 } // MockRuler_RecordingRules_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RecordingRules' @@ -58,12 +68,12 @@ func (_c *MockRuler_RecordingRules_Call) Run(run func(tenant string)) *MockRuler return _c } -func (_c *MockRuler_RecordingRules_Call) Return(_a0 []*model.RecordingRule) *MockRuler_RecordingRules_Call { - _c.Call.Return(_a0) +func (_c *MockRuler_RecordingRules_Call) Return(_a0 []*model.RecordingRule, _a1 error) *MockRuler_RecordingRules_Call { + _c.Call.Return(_a0, _a1) return _c } -func (_c *MockRuler_RecordingRules_Call) RunAndReturn(run func(string) []*model.RecordingRule) *MockRuler_RecordingRules_Call { +func (_c *MockRuler_RecordingRules_Call) RunAndReturn(run func(string) ([]*model.RecordingRule, error)) *MockRuler_RecordingRules_Call { _c.Call.Return(run) return _c }