pyroscope: Add relabel component for modifying and filtering profiles

CHANGELOG.md

@@ -14,6 +14,8 @@ Main (unreleased)
 
 - Add the possibility to export span events as logs in `otelcol.connector.spanlogs`. (@steve-hb)
 
+- Add `pyroscope.relabel` component to modify or filter profiles using Prometheus relabeling rules. (@marcsanmi)
+
 ### Enhancements
 
 - (_Experimental_) Log instance label key in `database_observability.mysql` (@cristiangreco)

docs/sources/reference/compatibility/_index.md

@@ -392,6 +392,7 @@ The following components, grouped by namespace, _export_ Pyroscope `ProfilesRece
 <!-- START GENERATED SECTION: EXPORTERS OF Pyroscope `ProfilesReceiver` -->
 
 {{< collapse title="pyroscope" >}}
+- [pyroscope.relabel](../components/pyroscope/pyroscope.relabel)
 - [pyroscope.write](../components/pyroscope/pyroscope.write)
 {{< /collapse >}}
 
@@ -408,6 +409,7 @@ The following components, grouped by namespace, _consume_ Pyroscope `ProfilesRec
 - [pyroscope.ebpf](../components/pyroscope/pyroscope.ebpf)
 - [pyroscope.java](../components/pyroscope/pyroscope.java)
 - [pyroscope.receive_http](../components/pyroscope/pyroscope.receive_http)
+- [pyroscope.relabel](../components/pyroscope/pyroscope.relabel)
 - [pyroscope.scrape](../components/pyroscope/pyroscope.scrape)
 {{< /collapse >}}
 

docs/sources/reference/components/pyroscope/pyroscope.relabel.md

@@ -0,0 +1,115 @@
+---
+canonical: https://grafana.com/docs/alloy/latest/reference/components/pyroscope/pyroscope.relabel/
+aliases:
+  - ../pyroscope.relabel/ # /docs/alloy/latest/reference/components/pyroscope.relabel/
+description: Learn about pyroscope.relabel
+title: pyroscope.relabel
+---
+
+# pyroscope.relabel
+
+The `pyroscope.relabel` component rewrites the label set of each profile passed to its receiver by applying one or more relabeling `rule`s and forwards the results to the list of receivers in the component's arguments.
+
+If no rules are defined or applicable to some profiles, then those profiles are forwarded as-is to each receiver passed in the component's arguments. If no labels remain after the relabeling rules are applied, then the profile is dropped.
+
+The most common use of `pyroscope.relabel` is to filter profiles or standardize the label set that is passed to one or more downstream receivers. The `rule` blocks are applied to the label set of each profile in order of their appearance in the configuration file.
+
+## Usage
+
+```alloy
+pyroscope.relabel "process" {
+    forward_to = RECEIVER_LIST
+
+    rule {
+        ...
+    }
+
+    ...
+}
+```
+
+## Arguments
+
+The following arguments are supported:
+
+| Name | Type | Description | Default | Required |
+| ---- | ---- | ----------- | ------- | -------- |
+| `forward_to` | `list(pyroscope.Appendable)` | List of receivers to forward profiles to after relabeling | | yes |
+| `max_cache_size` | `number` | Maximum number of entries in the label cache | 10000 | no |
+
+## Blocks
+
+The following blocks are supported inside the definition of `pyroscope.relabel`:
+
+Hierarchy | Name     | Description                                        | Required
+----------|----------|----------------------------------------------------|---------
+rule      | [rule][] | Relabeling rules to apply to received profile entries. | no
+
+[rule]: #rule-block
+
+### rule block
+
+{{< docs/shared lookup="reference/components/rule-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+## Exported fields
+
+The following fields are exported and can be referenced by other components:
+
+Name | Type | Description
+-----|------|------------
+`receiver` | `ProfilesReceiver` | A receiver that accepts profiles for relabeling.
+`rules` | `[]relabel.Config` | The list of relabeling rules.
+
+## Component health
+
+`pyroscope.relabel` is reported as unhealthy if it is given an invalid configuration.
+
+## Debug metrics
+
+* `pyroscope_relabel_profiles_dropped` (counter): Total number of profiles dropped by relabeling rules.
+* `pyroscope_relabel_profiles_processed` (counter): Total number of profiles processed.
+* `pyroscope_relabel_profiles_written` (counter): Total number of profiles forwarded.
+* `pyroscope_relabel_cache_misses` (counter): Total number of cache misses.
+* `pyroscope_relabel_cache_hits` (counter): Total number of cache hits.
+* `pyroscope_relabel_cache_size` (gauge): Total size of relabel cache.
+
+## Example
+
+```alloy
+pyroscope.relabel "process" {
+    forward_to = [pyroscope.write.backend.receiver]
+
+    # Rules are applied in order
+    rule {
+        source_labels = ["env"]
+        target_label = "environment"
+        action = "replace"
+        regex = "(.)"
+        replacement = "$1"
+    }
+
+    rule {
+        action = "labeldrop"
+        regex = "env"
+    }
+}
+```
+
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`pyroscope.relabel` can accept arguments from the following components:
+
+- Components that export [Pyroscope `ProfilesReceiver`](../../../compatibility/#pyroscope-profilesreceiver-exporters)
+
+`pyroscope.relabel` has exports that can be consumed by the following components:
+
+- Components that consume [Pyroscope `ProfilesReceiver`](../../../compatibility/#pyroscope-profilesreceiver-consumers)
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

internal/component/all/all.go

@@ -146,6 +146,7 @@ import (
 	_ "github.com/grafana/alloy/internal/component/pyroscope/ebpf"                           // Import pyroscope.ebpf
 	_ "github.com/grafana/alloy/internal/component/pyroscope/java"                           // Import pyroscope.java
 	_ "github.com/grafana/alloy/internal/component/pyroscope/receive_http"                   // Import pyroscope.receive_http
+	_ "github.com/grafana/alloy/internal/component/pyroscope/relabel"                        // Import pyroscope.relabel
 	_ "github.com/grafana/alloy/internal/component/pyroscope/scrape"                         // Import pyroscope.scrape
 	_ "github.com/grafana/alloy/internal/component/pyroscope/write"                          // Import pyroscope.write
 	_ "github.com/grafana/alloy/internal/component/remote/http"                              // Import remote.http

internal/component/pyroscope/appender.go

@@ -37,6 +37,7 @@ type IncomingProfile struct {
 	Body    io.ReadCloser
 	Headers http.Header
 	URL     *url.URL
+	Labels  labels.Labels
 }
 
 var _ Appendable = (*Fanout)(nil)

internal/component/pyroscope/receive_http/receive_http.go

@@ -25,6 +25,7 @@ import (
 	pushv1 "github.com/grafana/pyroscope/api/gen/proto/go/push/v1"
 	"github.com/grafana/pyroscope/api/gen/proto/go/push/v1/pushv1connect"
 	typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1"
+	"github.com/grafana/pyroscope/api/model/labelset"
 )
 
 const (
@@ -218,6 +219,26 @@ func (c *Component) getAppendables() []pyroscope.Appendable {
 func (c *Component) handleIngest(w http.ResponseWriter, r *http.Request) {
 	appendables := c.getAppendables()
 
+	// Parse labels early
+	var lbls labels.Labels
+	if nameParam := r.URL.Query().Get("name"); nameParam != "" {
+		ls, err := labelset.Parse(nameParam)
+		if err != nil {
+			level.Warn(c.opts.Logger).Log(
+				"msg", "Failed to parse labels from name parameter",
+				"name", nameParam,
+				"err", err,
+			)
+			// Continue with empty labels instead of returning an error
+		} else {
+			var labelPairs []labels.Label
+			for k, v := range ls.Labels() {
+				labelPairs = append(labelPairs, labels.Label{Name: k, Value: v})
+			}
+			lbls = labels.New(labelPairs...)
+		}
+	}
+
 	// Create a pipe for each appendable
 	pipeWriters := make([]io.Writer, len(appendables))
 	pipeReaders := make([]io.Reader, len(appendables))
@@ -251,6 +272,7 @@ func (c *Component) handleIngest(w http.ResponseWriter, r *http.Request) {
 				Body:    io.NopCloser(pipeReaders[i]),
 				Headers: r.Header.Clone(),
 				URL:     r.URL,
+				Labels:  lbls,
 			}
 
 			err := appendable.Appender().AppendIngest(ctx, profile)

internal/component/pyroscope/receive_http/receive_http_test.go

@@ -25,6 +25,7 @@ import (
 	pushv1 "github.com/grafana/pyroscope/api/gen/proto/go/push/v1"
 	"github.com/grafana/pyroscope/api/gen/proto/go/push/v1/pushv1connect"
 	typesv1 "github.com/grafana/pyroscope/api/gen/proto/go/types/v1"
+	"github.com/grafana/pyroscope/api/model/labelset"
 )
 
 // TestForwardsProfilesIngest verifies the behavior of the
@@ -44,13 +45,14 @@ func TestForwardsProfilesIngest(t *testing.T) {
 		appendableErrors []error
 		expectedStatus   int
 		expectedForwards int
+		expectedLabels   map[string]string
 	}{
 		{
 			name:             "Small profile",
 			profileSize:      1024, // 1KB
 			method:           "POST",
 			path:             "/ingest",
-			queryParams:      "name=test_app_1&from=1234567890&until=1234567900",
+			queryParams:      "name=test_app&from=1234567890&until=1234567900",
 			headers:          map[string]string{"Content-Type": "application/octet-stream"},
 			appendableErrors: []error{nil, nil},
 			expectedStatus:   http.StatusOK,
@@ -61,7 +63,7 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			profileSize:      1024 * 1024, // 1MB
 			method:           "POST",
 			path:             "/ingest",
-			queryParams:      "name=test_app_2&from=1234567891&until=1234567901&custom=param1",
+			queryParams:      "name=test_app&from=1234567891&until=1234567901&custom=param1",
 			headers:          map[string]string{"X-Scope-OrgID": "1234"},
 			appendableErrors: []error{nil},
 			expectedStatus:   http.StatusOK,
@@ -72,7 +74,7 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			profileSize:      1024,
 			method:           "GET",
 			path:             "/ingest",
-			queryParams:      "name=test_app_3&from=1234567892&until=1234567902",
+			queryParams:      "name=test_app&from=1234567892&until=1234567902",
 			headers:          map[string]string{},
 			appendableErrors: []error{nil, nil},
 			expectedStatus:   http.StatusMethodNotAllowed,
@@ -94,7 +96,7 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			profileSize:      1024,
 			method:           "POST",
 			path:             "/invalid",
-			queryParams:      "name=test_app_4&from=1234567893&until=1234567903",
+			queryParams:      "name=test_app&from=1234567893&until=1234567903",
 			headers:          map[string]string{"Content-Type": "application/octet-stream"},
 			appendableErrors: []error{nil, nil},
 			expectedStatus:   http.StatusNotFound,
@@ -105,7 +107,7 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			profileSize:      2048,
 			method:           "POST",
 			path:             "/ingest",
-			queryParams:      "name=test_app_5&from=1234567894&until=1234567904&scenario=all_fail",
+			queryParams:      "name=test_app&from=1234567894&until=1234567904&scenario=all_fail",
 			headers:          map[string]string{"Content-Type": "application/octet-stream", "X-Test": "fail-all"},
 			appendableErrors: []error{fmt.Errorf("error1"), fmt.Errorf("error2")},
 			expectedStatus:   http.StatusInternalServerError,
@@ -116,12 +118,42 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			profileSize:      4096,
 			method:           "POST",
 			path:             "/ingest",
-			queryParams:      "name=test_app_6&from=1234567895&until=1234567905&scenario=partial_failure",
+			queryParams:      "name=test_app&from=1234567895&until=1234567905&scenario=partial_failure",
 			headers:          map[string]string{"X-Custom-ID": "test-6"},
 			appendableErrors: []error{fmt.Errorf("error"), nil},
 			expectedStatus:   http.StatusInternalServerError,
 			expectedForwards: 2,
 		},
+		{
+			name:             "Valid labels are parsed and forwarded",
+			profileSize:      1024,
+			method:           "POST",
+			path:             "/ingest",
+			queryParams:      "name=test.app{env=prod,region=us-east}",
+			headers:          map[string]string{"Content-Type": "application/octet-stream"},
+			appendableErrors: []error{nil, nil},
+			expectedStatus:   http.StatusOK,
+			expectedForwards: 2,
+			expectedLabels: map[string]string{
+				"__name__": "test.app",
+				"env":      "prod",
+				"region":   "us-east",
+			},
+		},
+		{
+			name:             "Invalid labels still forward profile",
+			profileSize:      1024,
+			method:           "POST",
+			path:             "/ingest",
+			queryParams:      "name=test.app{invalid-label-syntax}",
+			headers:          map[string]string{"Content-Type": "application/octet-stream"},
+			appendableErrors: []error{nil, nil},
+			expectedStatus:   http.StatusOK,
+			expectedForwards: 2,
+			expectedLabels: map[string]string{
+				"__name__": "test.app", // Only __name__ is preserved
+			},
+		},
 	}
 
 	for _, tt := range tests {
@@ -136,7 +168,7 @@ func TestForwardsProfilesIngest(t *testing.T) {
 			require.Equal(t, tt.expectedForwards, forwardedCount, "Unexpected number of forwards")
 
 			if tt.expectedForwards > 0 {
-				verifyForwardedProfiles(t, appendables, testProfile, tt.headers, tt.queryParams)
+				verifyForwardedProfiles(t, appendables, testProfile, tt.headers, tt.queryParams, tt.expectedLabels)
 			}
 		})
 	}
@@ -296,11 +328,26 @@ func verifyForwardedProfiles(
 	expectedProfile []byte,
 	expectedHeaders map[string]string,
 	expectedQueryParams string,
+	expectedLabels map[string]string,
 ) {
 	for i, app := range appendables {
 		testApp, ok := app.(*testAppender)
 		require.True(t, ok, "Appendable is not a testAppender")
 
+		// Verify labels if name parameter exists and is valid
+		if nameParam := testApp.lastProfile.URL.Query().Get("name"); nameParam != "" {
+			ls, err := labelset.Parse(nameParam)
+			if err == nil {
+				require.Equal(t, ls.Labels(), testApp.lastProfile.Labels.Map(),
+					"Labels mismatch for appendable %d", i)
+			}
+		}
+
+		if expectedLabels != nil {
+			require.Equal(t, expectedLabels, testApp.lastProfile.Labels.Map(),
+				"Labels mismatch for appendable %d", i)
+		}
+
 		if testApp.lastProfile != nil {
 			// Verify profile body
 			body, err := io.ReadAll(testApp.lastProfile.Body)
@@ -446,6 +493,7 @@ func (a *testAppender) AppendIngest(_ context.Context, profile *pyroscope.Incomi
 		Body:    io.NopCloser(&buf),
 		Headers: profile.Headers,
 		URL:     profile.URL,
+		Labels:  profile.Labels,
 	}
 	a.lastProfile = newProfile