feat(distributor): make profile stripping recording-rules aware

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

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 [<private network interfaces>])
   -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

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.

docs/sources/configure-server/reference-configuration-parameters/index.md

@@ -185,6 +185,18 @@ tenant_settings:
     # CLI flag: -tenant-settings.recording-rules.enabled
     [enabled: <boolean> | 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: <boolean> | 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: <string> | 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: <duration> | 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: <boolean> | 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: <string> | default = ""]
 

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,

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

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

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

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))

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)