feat(pyroscope.receive_http): Support pushv1.Push in receive_http (#2431)

* feat(pyroscope.receive_http): Support pushv1.Push in receive_http

/push.v1.PusherService/Push which is a connect API used by profilecli
and pyroscope.write with pyroscope.ebpf and pyroscope.alloy.

This is in addtion to the /ingest API the component already supports.

* Update changelog and docs

Author: simonswine

CHANGELOG.md

@@ -36,6 +36,8 @@ Main (unreleased)
 
 - Bump snmp_exporter and embedded modules to 0.27.0. Add support for multi-module handling by comma separation and expose argument to increase SNMP polling concurrency for `prometheus.exporter.snmp`. (@v-zhuravlev)
 
+- Add support for pushv1.PusherService Connect API in `pyroscope.receive_http`. (@simonswine)
+
 v1.6.1
 -----------------
 

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

@@ -12,7 +12,7 @@ title: pyroscope.receive_http
 
 `pyroscope.receive_http` receives profiles over HTTP and forwards them to `pyroscope.*` components capable of receiving profiles.
 
-The HTTP API exposed is compatible with the Pyroscope [HTTP ingest API](https://grafana.com/docs/pyroscope/latest/configure-server/about-server-api/).
+The HTTP API exposed is compatible with both the Pyroscope [HTTP ingest API](https://grafana.com/docs/pyroscope/latest/configure-server/about-server-api/) and the [pushv1.PusherService](https://github.com/grafana/pyroscope/blob/main/api/push/v1/push.proto) Connect API.
 This allows `pyroscope.receive_http` to act as a proxy for Pyroscope profiles, enabling flexible routing and distribution of profile data.
 
 ## Usage
@@ -30,6 +30,7 @@ pyroscope.receive_http "LABEL" {
 The component will start an HTTP server supporting the following endpoint.
 
 * `POST /ingest` - send profiles to the component, which will be forwarded to the receivers as configured in the `forward_to argument`. The request format must match the format of the Pyroscope ingest API.
+* `POST /push.v1.PusherService/Push` - send profiles to the component, which will be forwarded to the receivers as configured in the `forward_to argument`. The request format must match the format of the Pyroscope pushv1.PusherService Connect API.
 
 ## Arguments
 

internal/component/pyroscope/receive_http/receive_http.go

@@ -9,8 +9,10 @@ import (
 	"reflect"
 	"sync"
 
+	"connectrpc.com/connect"
 	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/grafana/alloy/internal/component"
@@ -20,6 +22,9 @@ import (
 	"github.com/grafana/alloy/internal/featuregate"
 	"github.com/grafana/alloy/internal/runtime/logging/level"
 	"github.com/grafana/alloy/internal/util"
+	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"
 )
 
 const (
@@ -137,14 +142,81 @@ func (c *Component) Update(args component.Arguments) error {
 	c.server = srv
 
 	return c.server.MountAndRun(func(router *mux.Router) {
+		// this mounts the og pyroscope ingest API, mostly used by SDKs
 		router.HandleFunc("/ingest", c.handleIngest).Methods(http.MethodPost)
+
+		// mount connect go pushv1
+		pathPush, handlePush := pushv1connect.NewPusherServiceHandler(c)
+		router.PathPrefix(pathPush).Handler(handlePush).Methods(http.MethodPost)
 	})
 }
 
-func (c *Component) handleIngest(w http.ResponseWriter, r *http.Request) {
+func setLabelBuilderFromAPI(lb *labels.Builder, api []*typesv1.LabelPair) {
+	for i := range api {
+		lb.Set(api[i].Name, api[i].Value)
+	}
+}
+
+func apiToAlloySamples(api []*pushv1.RawSample) []*pyroscope.RawSample {
+	var (
+		alloy = make([]*pyroscope.RawSample, len(api))
+	)
+	for i := range alloy {
+		alloy[i] = &pyroscope.RawSample{
+			RawProfile: api[i].RawProfile,
+		}
+	}
+	return alloy
+}
+
+func (c *Component) Push(ctx context.Context, req *connect.Request[pushv1.PushRequest],
+) (*connect.Response[pushv1.PushResponse], error) {
+	appendables := c.getAppendables()
+
+	// Create an errgroup with the timeout context
+	g, ctx := errgroup.WithContext(ctx)
+
+	// Start copying the request body to all pipes
+	for i := range appendables {
+		appendable := appendables[i].Appender()
+		g.Go(func() error {
+			var (
+				errs error
+				lb   = labels.NewBuilder(nil)
+			)
+
+			for idx := range req.Msg.Series {
+				lb.Reset(nil)
+				setLabelBuilderFromAPI(lb, req.Msg.Series[idx].Labels)
+				err := appendable.Append(ctx, lb.Labels(), apiToAlloySamples(req.Msg.Series[idx].Samples))
+				if err != nil {
+					errs = errors.Join(
+						errs,
+						fmt.Errorf("unable to append series %s to appendable %d: %w", lb.Labels().String(), i, err),
+					)
+				}
+			}
+			return errs
+		})
+	}
+	if err := g.Wait(); err != nil {
+		level.Error(c.opts.Logger).Log("msg", "Failed to forward profiles requests", "err", err)
+		return nil, connect.NewError(connect.CodeInternal, err)
+	}
+
+	level.Debug(c.opts.Logger).Log("msg", "Profiles successfully forwarded")
+	return connect.NewResponse(&pushv1.PushResponse{}), nil
+}
+
+func (c *Component) getAppendables() []pyroscope.Appendable {
 	c.mut.Lock()
+	defer c.mut.Unlock()
 	appendables := c.appendables
-	c.mut.Unlock()
+	return appendables
+}
+
+func (c *Component) handleIngest(w http.ResponseWriter, r *http.Request) {
+	appendables := c.getAppendables()
 
 	// Create a pipe for each appendable
 	pipeWriters := make([]io.Writer, len(appendables))